Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,12 @@ and versions are tracked in the repo-root `VERSION` file.

## [Unreleased]

No unreleased changes yet.
### Fixed

- Make history persistence best-effort so secondary failures cannot mask the
command outcome or skip cleanup, context reset, and logger shutdown.
- Allow finished history records to omit `log_path` when file logging is
disabled.

## [0.3.0] - 2026-08-01

Expand Down
38 changes: 28 additions & 10 deletions lib/python/base_cli/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,14 @@ def _default_log_file(layout: Any, configured_log_file: Path | None) -> Path:
return configured_log_file or layout.log_dir / "primary.log"


def _warn_lifecycle_failure(context: Context, message: str, exc: Exception) -> None:
"""Report a secondary lifecycle failure without breaking teardown."""
try:
context.log.warning("%s: %s", message, exc)
except Exception: # pylint: disable=broad-exception-caught
pass


def _require_click():
try:
import click
Expand Down Expand Up @@ -163,16 +171,26 @@ def wrapper(**kwargs: Any):
exit_code = ExitCode.FAILURE
raise
finally:
if self.profile.history_writer is not None:
self.profile.history_writer(
context,
invocation_argv,
sensitive_options,
started_at,
exit_code,
)
reset_current_context(token)
context.cleanup()
try:
if self.profile.history_writer is not None:
try:
self.profile.history_writer(
context,
invocation_argv,
sensitive_options,
started_at,
exit_code,
)
except Exception as exc: # pylint: disable=broad-exception-caught
_warn_lifecycle_failure(context, "History finalization failed", exc)
finally:
try:
try:
context.cleanup()
except Exception as exc: # pylint: disable=broad-exception-caught
_warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc)
finally:
reset_current_context(token)

for kind, param_decls, attrs in getattr(func, "__base_cli_param_specs__", []):
if kind == "option":
Expand Down
14 changes: 10 additions & 4 deletions lib/python/base_cli/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,12 +59,18 @@ def bind_project(self, project_name: str | None, project_root: Path, manifest_pa
self.project_root = project_root.resolve()
self.manifest_path = manifest_path.resolve() if manifest_path is not None else None

def _warn_cleanup_failure(self, message: str, *args: object) -> None:
try:
self.log.warning(message, *args)
except Exception: # pylint: disable=broad-exception-caught
pass

def cleanup(self) -> None:
for hook in self.cleanup_hooks:
try:
hook()
except Exception as exc: # pylint: disable=broad-exception-caught
self.log.warning("Cleanup hook failed: %s", exc)
self._warn_cleanup_failure("Cleanup hook failed: %s", exc)
if not self.keep_temp and self.temp_dir.exists():
try:
shutil.rmtree(self.temp_dir)
Expand All @@ -74,16 +80,16 @@ def cleanup(self) -> None:
except OSError:
break
except OSError as exc:
self.log.warning("Temp directory cleanup failed for '%s': %s", self.temp_dir, exc)
self._warn_cleanup_failure("Temp directory cleanup failed for '%s': %s", self.temp_dir, exc)
for handler in list(self.log.handlers):
try:
handler.flush()
except Exception as exc: # pylint: disable=broad-exception-caught
self.log.warning("Log handler flush failed: %s", exc)
self._warn_cleanup_failure("Log handler flush failed: %s", exc)
try:
handler.close()
except Exception as exc: # pylint: disable=broad-exception-caught
self.log.warning("Log handler close failed: %s", exc)
self._warn_cleanup_failure("Log handler close failed: %s", exc)
self.log.removeHandler(handler)


Expand Down
2 changes: 1 addition & 1 deletion lib/python/base_cli/history.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,6 @@ def build_finished_record(
"duration_ms": duration_ms(started_at, ended_at),
"exit_code": exit_code,
"status": "ok" if exit_code == 0 else "error",
"log_path": compact_path(context.log_file),
"owner": context.runtime_owner,
"bundle_path": compact_path(context.run_root or context.state_dir),
"os": normalized_os(),
Expand All @@ -84,6 +83,7 @@ def build_finished_record(
"project_root": compact_optional_path(context.project_root),
"manifest": compact_optional_path(context.manifest_path),
"workspace_root": compact_optional_path(context.workspace_root),
"log_path": compact_optional_path(context.log_file),
"shell": current_shell(),
"scope": context.history_scope,
"parent_run_id": context.history_parent_run_id,
Expand Down
133 changes: 133 additions & 0 deletions tests/test_app_lifecycle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
from __future__ import annotations

import importlib.util
import logging
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path

import base_cli
from base_cli.testing import invoke


class _BrokenHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
del record

def flush(self) -> None:
raise OSError("flush unavailable")

def close(self) -> None:
raise OSError("close unavailable")


class _CommandFailure(RuntimeError):
pass


@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
class AppLifecycleTests(unittest.TestCase):
def test_history_failure_does_not_change_success_or_skip_teardown(self) -> None:
def fail_history(*_args: object) -> None:
raise RuntimeError("history unavailable")

profile = replace(base_cli.CliProfile.generic(), history_writer=fail_history)
app = base_cli.App(name="lifecycle-success", profile=profile)
seen: dict[str, object] = {}

@app.command()
def main(ctx: base_cli.Context) -> None:
seen["context"] = ctx
seen["temp_dir"] = ctx.temp_dir
seen["logger"] = ctx.log
seen["cleanup_context"] = None

def fail_cleanup() -> None:
raise RuntimeError("cleanup unavailable")

def record_cleanup_context() -> None:
seen["cleanup_context"] = base_cli.get_current_context()

ctx.on_cleanup(fail_cleanup)
ctx.on_cleanup(record_cleanup_context)
ctx.log.handlers.insert(0, _BrokenHandler())

with tempfile.TemporaryDirectory() as tmpdir:
result = invoke(app, [], home=Path(tmpdir))

self.assertEqual(result.exit_code, 0, result.output)
self.assertIsNone(result.exception)
self.assertIs(seen["cleanup_context"], seen["context"])
self.assertFalse(Path(seen["temp_dir"]).exists())
self.assertEqual(seen["logger"].handlers, [])

with self.assertRaisesRegex(RuntimeError, "context is not active"):
base_cli.get_current_context()
self.assertIn("History finalization failed: history unavailable", result.stderr)
self.assertIn("Cleanup hook failed: cleanup unavailable", result.stderr)
self.assertIn("Log handler flush failed: flush unavailable", result.stderr)
self.assertIn("Log handler close failed: close unavailable", result.stderr)

def test_cleanup_failure_does_not_change_success_or_leak_context(self) -> None:
app = base_cli.App(name="cleanup-failure")
seen: dict[str, object] = {}
cleanup_calls: list[None] = []

@app.command()
def main(ctx: base_cli.Context) -> None:
seen["context"] = ctx
seen["original_cleanup"] = ctx.cleanup

def fail_cleanup() -> None:
cleanup_calls.append(None)
raise RuntimeError("cleanup unavailable")

ctx.cleanup = fail_cleanup # type: ignore[method-assign]

with tempfile.TemporaryDirectory() as tmpdir:
result = invoke(app, [], home=Path(tmpdir))

self.assertEqual(result.exit_code, 0, result.output)
self.assertIsNone(result.exception)
self.assertEqual(cleanup_calls, [None])
self.assertIn("Lifecycle cleanup failed: cleanup unavailable", result.stderr)
with self.assertRaisesRegex(RuntimeError, "context is not active"):
base_cli.get_current_context()

original_cleanup = seen["original_cleanup"]
self.assertTrue(callable(original_cleanup))
original_cleanup()

def test_history_failure_does_not_mask_command_failure(self) -> None:
def fail_history(*_args: object) -> None:
raise OSError("history unavailable")

profile = replace(base_cli.CliProfile.generic(), history_writer=fail_history)
app = base_cli.App(name="lifecycle-failure", profile=profile)
primary_failure = _CommandFailure("command failed")
seen: dict[str, object] = {}

@app.command()
def main(ctx: base_cli.Context) -> None:
seen["temp_dir"] = ctx.temp_dir
seen["logger"] = ctx.log
ctx.on_cleanup(lambda: seen.update(cleanup_called=True))
raise primary_failure

with tempfile.TemporaryDirectory() as tmpdir:
result = invoke(app, [], home=Path(tmpdir))

self.assertEqual(result.exit_code, 1)
self.assertIs(result.exception, primary_failure)
self.assertTrue(seen["cleanup_called"])
self.assertFalse(Path(seen["temp_dir"]).exists())
self.assertEqual(seen["logger"].handlers, [])

with self.assertRaisesRegex(RuntimeError, "context is not active"):
base_cli.get_current_context()
self.assertIn("History finalization failed: history unavailable", result.stderr)


if __name__ == "__main__":
unittest.main()
28 changes: 28 additions & 0 deletions tests/test_generic_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,34 @@ def test_finished_record_has_no_product_version_field(self) -> None:
self.assertEqual(record["command"], "demo-tool")
self.assertNotIn("base_version", record)

def test_finished_record_omits_log_path_when_file_logging_is_disabled(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
context = Context(
cli_name="demo_tool",
run_id="run-1",
state_dir=root / "state",
log_dir=root / "logs",
cache_dir=root / "cache",
temp_dir=root / "tmp",
log_file=None,
config={},
environment="dev",
debug=False,
keep_temp=False,
log=logging.getLogger("generic-core-no-log-test"),
)

record = history.build_finished_record(
context,
["demo_tool"],
set(),
history.utc_now() - timedelta(seconds=1),
0,
)

self.assertNotIn("log_path", record)

def test_base_specific_path_helpers_are_not_in_generic_module(self) -> None:
import base_cli.paths as paths

Expand Down