From b0877759194210b67c014ad398848530f80fce16 Mon Sep 17 00:00:00 2001
From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com>
Date: Sun, 2 Aug 2026 07:30:09 -0700
Subject: [PATCH] fix: finalize run outcomes and metadata
---
CHANGELOG.md | 21 +
README.md | 55 +-
docs/cache-ownership-and-layout.md | 23 +
docs/consumer-profiles.md | 14 +
lib/python/base_cli/__init__.py | 2 +
lib/python/base_cli/_lifecycle.py | 165 +++++
lib/python/base_cli/_runtime.py | 6 +-
lib/python/base_cli/app.py | 512 +++++++++++---
lib/python/base_cli/config.py | 11 +-
lib/python/base_cli/context.py | 30 +-
lib/python/base_cli/errors.py | 5 +
lib/python/base_cli/exit_codes.py | 1 +
lib/python/base_cli/logging.py | 6 +
tests/test_app_lifecycle.py | 90 +++
tests/test_app_run.py | 106 ++-
tests/test_app_run_metadata.py | 931 ++++++++++++++++++++++++++
tests/test_app_runtime_errors.py | 28 +
tests/test_app_startup_transaction.py | 243 +++++++
tests/test_logging.py | 14 +
tests/test_public_api.py | 2 +
20 files changed, 2149 insertions(+), 116 deletions(-)
create mode 100644 lib/python/base_cli/_lifecycle.py
create mode 100644 lib/python/base_cli/errors.py
create mode 100644 tests/test_app_run_metadata.py
create mode 100644 tests/test_app_startup_transaction.py
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cdc7352..8e2cbe6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,8 +7,29 @@ and versions are tracked in the repo-root `VERSION` file.
## [Unreleased]
+### Added
+
+- Add `ConfigurationError` so consumer profiles can explicitly mark
+ user-correctable configuration messages as safe usage errors.
+
+### Changed
+
+- Normalize command returns, Click errors, aborts, interrupts, `SystemExit`, and
+ unexpected exceptions through one core outcome model and clean `run_app()`
+ process boundary.
+- Treat plain profile-callback exceptions as private internal failures. Profiles
+ that used `ValueError` for expected configuration problems should raise
+ `ConfigurationError` instead.
+
### Fixed
+- Finalize core-owned run metadata for successful, failed, aborted, interrupted,
+ and unexpected command outcomes without letting secondary persistence
+ failures replace the command result.
+- Roll back partially constructed command contexts without leaking handlers,
+ temporary directories, or incomplete run bundles.
+- Preserve exception tracebacks in persistent logs and show them on stderr only
+ when debug output is enabled.
- 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
diff --git a/README.md b/README.md
index 6f7b2a0..7e0aa2a 100644
--- a/README.md
+++ b/README.md
@@ -275,6 +275,8 @@ command result meanings:
prevented successful completion.
- `ExitCode.USAGE_ERROR` (`2`): the command could not proceed because user
input, configuration, or environment setup was invalid or incomplete.
+- `ExitCode.INTERRUPTED` (`130`): the user interrupted the command with
+ Ctrl+C.
Existing commands can keep returning integers. New code should prefer the named
constants when it makes intent clearer:
@@ -285,6 +287,37 @@ if ctx.project_root is None:
return base_cli.ExitCode.USAGE_ERROR
```
+`run_app()` is the process boundary for production entry points. It preserves
+Click's messages and exit codes for usage and application errors, reports an
+explicit abort as `1`, and reports Ctrl+C during startup or
+command execution as `130` without a traceback. After the command outcome has
+settled, history, metadata, and cleanup are best-effort teardown: even a second
+interrupt there cannot replace the primary result.
+
+An unexpected exception returns `1` with a stable, detail-free message. The run
+ID and diagnostic-log path are included when context and file logging are
+available. The traceback is kept in the persistent log when enabled and is
+shown on stderr with an effective `--debug` setting. A failure before option
+parsing can provide a traceback only when `--debug` is an unambiguous leading
+flag; otherwise the message says that diagnostic context was unavailable.
+Tests or embedding code that need the original exception can pass
+`reraise_unexpected=True`.
+
+| Command result or exception | `outcome` | Exit code | Default message |
+| --- | --- | ---: | --- |
+| `None` or returned `0` | `success` | 0 | none |
+| returned `2` | `usage_error` | 2 | none |
+| another returned nonzero integer | `nonzero_return` | returned value | none |
+| `click.UsageError` | `usage_error` | exception code | Click usage error |
+| another `click.ClickException` | `click_error` | exception code | Click error |
+| `click.Abort` | `aborted` | 1 | `Aborted!` |
+| Ctrl+C | `interrupted` | 130 | `Interrupted.` |
+| `SystemExit` | `system_exit` | normalized payload | string payload, if any |
+| another unexpected exception | `unexpected_error` | 1 | stable internal-error message |
+
+For `SystemExit`, a missing payload becomes `0`, an integer payload is preserved,
+and any other payload is printed and normalized to `1`.
+
## Context
`Context` is the object command code should pass around instead of rediscovering
@@ -454,8 +487,26 @@ Windows uses `%LOCALAPPDATA%` (falling back to `~/AppData/Local`). Set
`BASE_CLI_CACHE_DIR` to override the default on any platform. The generic
profile does not prescribe a product-wide cache name or cleanup command.
-Each invocation is a run bundle containing a private `run.json`, `logs/`, and
-`tmp/`, while persistent component caches live in the bundle's cache directory.
+Each lifecycle-owned invocation is a run bundle containing a private
+`run.json`, `logs/`, and `tmp/`, while persistent component caches live in the
+owner's cache directory. When persistence succeeds, `run.json` begins with
+`status: "running"` and is finalized with `status`, `outcome`, `exit_code`,
+`ended_at`, and `duration_ms`, including command failures and interruptions.
+The stable outcome values are `success`, `usage_error`, `nonzero_return`,
+`click_error`, `aborted`, `interrupted`, `system_exit`, and
+`unexpected_error`. Terminal-write failures are warnings and cannot change the
+process result; base-cli then removes a matching or corrupt owned record on a
+best-effort basis so history data cannot masquerade as authoritative core data.
+
+Parsing errors, help, and version requests occur before the command lifecycle
+owns a bundle and therefore do not create one. Neither do inherited runtimes,
+`log_to_file=False`, or dry-run invocations; an explicit log path can still
+receive diagnostics in the latter two modes. Context startup is transactional:
+if directory creation, logger setup, or retention fails, base-cli closes
+partially installed handlers and removes new bundle-local temp/log artifacts
+and empty directories. Pre-existing content, persistent component caches, and
+parent-runtime data are preserved.
+
On POSIX, base-cli enforces owner-only `0600`/`0700` modes. On Windows, the
default user-local cache root relies on inherited user-profile ACLs; consumers
using a custom cache root must provide the appropriate ACL themselves.
diff --git a/docs/cache-ownership-and-layout.md b/docs/cache-ownership-and-layout.md
index b9c1ebb..320a6be 100644
--- a/docs/cache-ownership-and-layout.md
+++ b/docs/cache-ownership-and-layout.md
@@ -25,6 +25,29 @@ Each invocation has a private run bundle containing:
- `logs/` for diagnostic logs; and
- `tmp/` for temporary command data.
+The core lifecycle, rather than an optional history adapter, owns `run.json`.
+Once command context construction succeeds, the file is written with
+`status: "running"`. When persistence succeeds, the core writes a terminal
+snapshot containing `status`, `outcome`, `exit_code`, `ended_at`, and
+`duration_ms`. Terminal status is `ok` only for exit code zero; all other exit
+codes use `error`. The outcome discriminator is one of `success`,
+`usage_error`, `nonzero_return`, `click_error`, `aborted`, `interrupted`,
+`system_exit`, or `unexpected_error`.
+
+History may enrich a matching record with consumer fields, but the core writes
+the canonical lifecycle fields last. If terminal persistence fails, the
+process keeps its primary result and the framework best-effort removes its
+matching or corrupt record rather than leave history data or `running` state
+looking authoritative. Writes are not yet promised to be atomic.
+
+The ownership boundary intentionally excludes parser failures, help and version
+requests, inherited runtime bindings, `log_to_file=False`, and dry-run mode.
+Those invocations do not create or finalize a bundle. If context construction
+fails after creating artifacts, rollback closes partial logging handlers and
+removes new bundle-local temp/log artifacts and empty directories. It does not
+delete pre-existing content, persistent component caches, paths outside the
+selected run root, or a parent runtime's metadata.
+
Persistent component caches live under the owner's `cache/components/` path.
On POSIX systems, runtime directories are owner-only (`0700`) and runtime files
are owner-only (`0600`). On Windows, the default `%LOCALAPPDATA%` root relies
diff --git a/docs/consumer-profiles.md b/docs/consumer-profiles.md
index eff306f..8936b10 100644
--- a/docs/consumer-profiles.md
+++ b/docs/consumer-profiles.md
@@ -80,6 +80,20 @@ translate internal entry-point names into user-facing labels. The generic
default only replaces underscores with hyphens; it does not know any product's
command aliases.
+## Safe profile errors
+
+Plain exceptions from profile callbacks are treated as unexpected internal
+errors: production output hides their details, while `--debug` exposes the
+traceback after option parsing. This prevents a programming error or a private
+value in a callback from becoming user-facing output by accident.
+
+For a user-correctable configuration problem whose message is safe to show,
+raise `base_cli.ConfigurationError`; `run_app()` renders it as a Click usage
+error with exit code `2`. A callback may raise `click.UsageError` or another
+`click.ClickException` when it needs Click's standard rendering or a custom
+exit code. Consumers that previously raised plain `ValueError` for expected
+configuration failures should migrate those sites to `ConfigurationError`.
+
## Consumer-owned adapters
`App()` uses `CliProfile.generic()` when no profile is supplied. This keeps the
diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py
index de2669d..76e6dbb 100644
--- a/lib/python/base_cli/__init__.py
+++ b/lib/python/base_cli/__init__.py
@@ -45,6 +45,7 @@ def _resolve_version() -> str:
register_record_schema,
)
from .context import Context, get_current_context
+from .errors import ConfigurationError
from .exit_codes import ExitCode
from .inspection import inspection_envelope, render_inspection_json
from .logging import configure_logger, log_critical, log_debug, log_error, log_info, log_warning
@@ -66,6 +67,7 @@ def _resolve_version() -> str:
"CliProfile",
"CommandFilterNormalizer",
"CommandProtocolError",
+ "ConfigurationError",
"Context",
"ExitCode",
"FieldSpec",
diff --git a/lib/python/base_cli/_lifecycle.py b/lib/python/base_cli/_lifecycle.py
new file mode 100644
index 0000000..e6a6ffe
--- /dev/null
+++ b/lib/python/base_cli/_lifecycle.py
@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from ._private_files import write_private_json
+from .context import Context
+from .exit_codes import ExitCode
+from .history import format_timestamp
+
+
+@dataclass(frozen=True)
+class InvocationOutcome:
+ """Private normalized result of one lifecycle-owned invocation."""
+
+ kind: str
+ status: str
+ exit_code: int
+
+
+@dataclass(frozen=True)
+class RunRecorder:
+ """Write core-owned lifecycle snapshots for one Context."""
+
+ context: Context
+ started_at: datetime
+ started_monotonic_ns: int
+
+ def start(self) -> None:
+ if self.context._run_metadata_path is None:
+ return
+ write_private_json(
+ self.context._run_metadata_path,
+ self._metadata(status="running"),
+ )
+
+ def finish(
+ self,
+ outcome: InvocationOutcome,
+ *,
+ ended_at: datetime,
+ ended_monotonic_ns: int,
+ ) -> None:
+ if self.context._run_metadata_path is None:
+ return
+ elapsed_ns = max(0, ended_monotonic_ns - self.started_monotonic_ns)
+ metadata = self._existing_metadata()
+ metadata.update(self._metadata(status=outcome.status))
+ metadata.update(
+ {
+ "outcome": outcome.kind,
+ "exit_code": outcome.exit_code,
+ "ended_at": format_timestamp(ended_at),
+ "duration_ms": round(elapsed_ns / 1_000_000),
+ }
+ )
+ write_private_json(self.context._run_metadata_path, metadata)
+
+ def _existing_metadata(self) -> dict[str, Any]:
+ path = self.context._run_metadata_path
+ if path is None:
+ return {}
+ try:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+ return {}
+ if not isinstance(payload, dict) or payload.get("run_id") != self.context.run_id:
+ return {}
+ return dict(payload)
+
+ def discard_owned_record(self) -> None:
+ """Remove our misleading record after terminal persistence fails."""
+ path = self.context._run_metadata_path
+ if path is None:
+ return
+ try:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ if (
+ isinstance(payload, dict)
+ and payload.get("run_id") == self.context.run_id
+ ):
+ path.unlink()
+ except (UnicodeDecodeError, json.JSONDecodeError):
+ try:
+ path.unlink()
+ except OSError:
+ pass
+ except OSError:
+ pass
+
+ def _metadata(self, *, status: str) -> dict[str, Any]:
+ context = self.context
+ return {
+ "schema_version": 1,
+ "run_id": context.run_id,
+ "owner": context.runtime_owner,
+ "cli": context.cli_name,
+ "status": status,
+ "started_at": format_timestamp(self.started_at),
+ "project": context.project_name,
+ "project_root": str(context.project_root) if context.project_root else None,
+ "manifest": str(context.manifest_path) if context.manifest_path else None,
+ "workspace_root": str(context.workspace_root) if context.workspace_root else None,
+ }
+
+
+def outcome_from_exit_code(exit_code: int) -> InvocationOutcome:
+ if exit_code == ExitCode.SUCCESS:
+ return InvocationOutcome("success", "ok", exit_code)
+ if exit_code == ExitCode.USAGE_ERROR:
+ return InvocationOutcome("usage_error", "error", exit_code)
+ if exit_code == ExitCode.INTERRUPTED:
+ return InvocationOutcome("interrupted", "error", exit_code)
+ return InvocationOutcome("nonzero_return", "error", exit_code)
+
+
+def outcome_from_exception(click: Any, exc: BaseException) -> InvocationOutcome:
+ if isinstance(exc, KeyboardInterrupt):
+ return InvocationOutcome("interrupted", "error", ExitCode.INTERRUPTED)
+ if isinstance(exc, EOFError):
+ return InvocationOutcome("aborted", "error", ExitCode.FAILURE)
+ if isinstance(exc, click.Abort):
+ if isinstance(exc.__cause__, KeyboardInterrupt):
+ return InvocationOutcome("interrupted", "error", ExitCode.INTERRUPTED)
+ return InvocationOutcome("aborted", "error", ExitCode.FAILURE)
+ if isinstance(exc, click.exceptions.Exit):
+ exit_code = _click_exception_exit_code(exc)
+ if exit_code is None:
+ return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE)
+ return outcome_from_exit_code(exit_code)
+ if isinstance(exc, click.UsageError):
+ exit_code = _click_exception_exit_code(exc)
+ if exit_code is None:
+ return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE)
+ return InvocationOutcome("usage_error", _status_for_exit_code(exit_code), exit_code)
+ if isinstance(exc, click.ClickException):
+ exit_code = _click_exception_exit_code(exc)
+ if exit_code is None:
+ return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE)
+ return InvocationOutcome("click_error", _status_for_exit_code(exit_code), exit_code)
+ if isinstance(exc, SystemExit):
+ exit_code = system_exit_code(exc)
+ return InvocationOutcome("system_exit", _status_for_exit_code(exit_code), exit_code)
+ return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE)
+
+
+def system_exit_code(exc: SystemExit) -> int:
+ if exc.code is None:
+ return ExitCode.SUCCESS
+ if isinstance(exc.code, int):
+ return int(exc.code)
+ return ExitCode.FAILURE
+
+
+def _status_for_exit_code(exit_code: int) -> str:
+ return "ok" if exit_code == ExitCode.SUCCESS else "error"
+
+
+def _click_exception_exit_code(exc: Any) -> int | None:
+ try:
+ return int(exc.exit_code)
+ except BaseException: # pylint: disable=broad-exception-caught
+ return None
diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py
index 52f46a4..1c94b9f 100644
--- a/lib/python/base_cli/_runtime.py
+++ b/lib/python/base_cli/_runtime.py
@@ -23,6 +23,10 @@ class RuntimeLayout:
_LOG_INDEX_NAME = ".base-cli-log-index.json"
+class RuntimeDirectoryError(RuntimeError):
+ """An actionable failure to create a framework-owned runtime directory."""
+
+
# pylint: disable=too-many-arguments
def runtime_layout(
cache_root: Path,
@@ -62,7 +66,7 @@ def create_runtime_directory(path: Path, cache_root: Path) -> None:
for directory in [path, *missing]:
restrict_directory(directory)
except OSError as exc:
- raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc
+ raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc
def runtime_namespace_root(cache_root: Path, namespace: str) -> Path:
diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py
index a5fb842..6d88da0 100644
--- a/lib/python/base_cli/app.py
+++ b/lib/python/base_cli/app.py
@@ -1,15 +1,29 @@
from __future__ import annotations
import functools
+import logging
import os
+import shutil
import sys
-from contextvars import ContextVar
+import time
+import traceback
+from contextvars import ContextVar, Token
+from dataclasses import dataclass
+from datetime import datetime
from pathlib import Path
from typing import Any, Callable
-from ._runtime import create_runtime_directory, prune_log_files
+from ._lifecycle import (
+ InvocationOutcome,
+ RunRecorder,
+ outcome_from_exception,
+ outcome_from_exit_code,
+ system_exit_code,
+)
from ._private_files import write_private_json
-from .context import Context, reset_current_context, set_current_context
+from ._runtime import RuntimeDirectoryError, create_runtime_directory, prune_log_files
+from .context import Context, recover_current_context, reset_current_context, set_current_context
+from .errors import ConfigurationError
from .exit_codes import ExitCode
from .history import utc_now
from .logging import configure_logger, log_invocation
@@ -26,18 +40,130 @@
_INVOCATION_ARGV: ContextVar[list[str] | None] = ContextVar("base_cli_invocation_argv", default=None)
+@dataclass
+class _InvocationState:
+ run_id: str | None = None
+ log_file: Path | None = None
+ debug: bool = False
+ quiet: bool = False
+ options_parsed: bool = False
+
+
+_INVOCATION_STATE: ContextVar[_InvocationState | None] = ContextVar("base_cli_invocation_state", default=None)
+
+
+def _reset_context_var(variable: ContextVar[Any], token: Any) -> None:
+ try:
+ variable.reset(token)
+ except BaseException: # pylint: disable=broad-exception-caught
+ try:
+ previous = token.old_value
+ variable.set(None if previous is Token.MISSING else previous)
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+
+
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:
+def _warn_lifecycle_failure(context: Context, message: str, exc: BaseException) -> None:
"""Report a secondary lifecycle failure without breaking teardown."""
try:
- context.log.warning("%s: %s", message, exc)
- except Exception: # pylint: disable=broad-exception-caught
+ detail = str(exc) or type(exc).__name__
+ context.log.warning("%s: %s", message, detail)
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+
+
+def _capture_invocation_context(context: Context) -> None:
+ state = _INVOCATION_STATE.get()
+ if state is None:
+ return
+ state.run_id = context.run_id
+ state.log_file = context.log_file
+ state.debug = context.debug
+ state.quiet = context.quiet
+
+
+def _capture_standard_options(standard: dict[str, Any]) -> None:
+ state = _INVOCATION_STATE.get()
+ if state is None:
+ return
+ state.debug = bool(standard.get("debug"))
+ state.quiet = bool(standard.get("quiet"))
+ state.options_parsed = True
+
+
+def _capture_effective_output_options(*, debug: bool, quiet: bool) -> None:
+ state = _INVOCATION_STATE.get()
+ if state is None:
+ return
+ state.debug = debug
+ state.quiet = quiet
+
+
+def _record_unexpected_traceback(context: Context, outcome: InvocationOutcome) -> None:
+ if outcome.kind != "unexpected_error":
+ return
+ try:
+ context.log.debug("Unexpected command exception", exc_info=True)
+ except BaseException: # pylint: disable=broad-exception-caught
pass
+def _start_run_recorder(recorder: RunRecorder) -> None:
+ try:
+ recorder.start()
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(recorder.context, "Run metadata start failed", exc)
+
+
+def _finish_run_recorder(
+ recorder: RunRecorder,
+ outcome: InvocationOutcome,
+ *,
+ ended_at: datetime,
+ ended_monotonic_ns: int,
+) -> None:
+ try:
+ recorder.finish(
+ outcome,
+ ended_at=ended_at,
+ ended_monotonic_ns=ended_monotonic_ns,
+ )
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ path = recorder.context._run_metadata_path
+ _warn_lifecycle_failure(
+ recorder.context,
+ f"Run metadata finalization failed for '{path}'",
+ exc,
+ )
+ _discard_owned_run_record(recorder)
+
+
+def _discard_owned_run_record(recorder: RunRecorder) -> None:
+ try:
+ recorder.discard_owned_record()
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(
+ recorder.context,
+ f"Run metadata recovery failed for '{recorder.context._run_metadata_path}'",
+ exc,
+ )
+
+
+def _reset_active_context(context: Context, token: Any) -> None:
+ try:
+ reset_current_context(token)
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(context, "Active context reset failed", exc)
+ try:
+ recover_current_context(token)
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+
+
def _require_click():
try:
import click
@@ -68,8 +194,8 @@ def __init__(
self.log_to_file = log_to_file
self.max_log_files = max_log_files
# Standalone applications must not inherit a consumer's product
- # conventions. Consumers with an existing integration should pass an
- # Consumers with product-specific policies should pass an explicit profile.
+ # conventions. Consumers with product-specific policies should pass an
+ # explicit profile.
self.profile = profile or CliProfile.generic()
self._click_command = None
self._command_func: Callable[..., Any] | None = None
@@ -136,7 +262,12 @@ def _build_click_command(self) -> Any:
group.add_command(click.command(*command_args, **command_kwargs)(wrapper))
return group
- def _build_command_wrapper(self, click: Any, func: Callable[..., Any], include_version: bool) -> Callable[..., Any]:
+ def _build_command_wrapper(
+ self,
+ click: Any,
+ func: Callable[..., Any],
+ include_version: bool,
+ ) -> Callable[..., Any]:
sensitive_options = set(getattr(func, "__base_cli_sensitive_options__", set()))
dry_run_parameter = getattr(func, "__base_cli_dry_run_parameter__", "dry_run")
@@ -147,15 +278,31 @@ def wrapper(**kwargs: Any):
_pop_standard_options(kwargs),
)
_validate_standard_options(click, standard)
- try:
- context = self._create_context(standard, sensitive_options, dry_run=bool(kwargs.get(dry_run_parameter)))
- except (RuntimeError, ValueError) as exc:
- raise click.ClickException(str(exc)) from exc
- token = set_current_context(context)
+ _capture_standard_options(standard)
started_at = utc_now()
- exit_code = ExitCode.SUCCESS
- invocation_argv = _current_invocation_argv()
+ started_monotonic_ns = time.monotonic_ns()
+ context: Context | None = None
+ recorder: RunRecorder | None = None
+ outcome = outcome_from_exit_code(ExitCode.SUCCESS)
+ invocation_argv: list[str] = []
+ token = None
try:
+ try:
+ context = self._create_context(
+ standard,
+ sensitive_options,
+ dry_run=bool(kwargs.get(dry_run_parameter)),
+ )
+ except ConfigurationError as exc:
+ raise click.UsageError(str(exc)) from exc
+ except RuntimeDirectoryError as exc:
+ raise click.ClickException(str(exc)) from exc
+
+ recorder = RunRecorder(context, started_at, started_monotonic_ns)
+ token = set_current_context(context)
+ _capture_invocation_context(context)
+ invocation_argv = _current_invocation_argv()
+ _start_run_recorder(recorder)
log_invocation(context.log, invocation_argv, sensitive_options)
if context.project_root is not None:
context.log.debug("project_root=%s", context.project_root)
@@ -166,31 +313,55 @@ def wrapper(**kwargs: Any):
exit_code = _normalize_command_result(result)
except TypeError as exc:
raise click.ClickException(str(exc)) from exc
+ outcome = outcome_from_exit_code(exit_code)
return result
- except Exception:
- exit_code = ExitCode.FAILURE
+ except BaseException as exc:
+ if context is not None:
+ outcome = outcome_from_exception(click, exc)
+ _record_unexpected_traceback(context, outcome)
raise
finally:
- try:
- if self.profile.history_writer is not None:
- try:
+ if context is not None:
+ try:
+ ended_at = utc_now()
+ ended_monotonic_ns = time.monotonic_ns()
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ ended_at = started_at
+ ended_monotonic_ns = started_monotonic_ns
+ _warn_lifecycle_failure(context, "Terminal clock capture failed", exc)
+
+ try:
+ if self.profile.history_writer is not None:
self.profile.history_writer(
context,
invocation_argv,
sensitive_options,
started_at,
- exit_code,
+ outcome.exit_code,
)
- except Exception as exc: # pylint: disable=broad-exception-caught
- _warn_lifecycle_failure(context, "History finalization failed", exc)
- finally:
- try:
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(context, "History finalization failed", exc)
+
+ if recorder is None:
try:
- context.cleanup()
- except Exception as exc: # pylint: disable=broad-exception-caught
- _warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc)
+ recorder = RunRecorder(context, started_at, started_monotonic_ns)
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(context, "Run recorder construction failed", exc)
+ if recorder is not None:
+ _finish_run_recorder(
+ recorder,
+ outcome,
+ ended_at=ended_at,
+ ended_monotonic_ns=ended_monotonic_ns,
+ )
+
+ try:
+ context.cleanup()
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ _warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc)
finally:
- reset_current_context(token)
+ if token is not None:
+ _reset_active_context(context, token)
for kind, param_decls, attrs in getattr(func, "__base_cli_param_specs__", []):
if kind == "option":
@@ -213,6 +384,7 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
debug = bool(standard.get("debug") or str(config.get("log_level", "")).lower() == "debug")
quiet = bool(standard.get("quiet"))
keep_temp = bool(standard.get("keep_temp") or config.get("keep_temp"))
+ _capture_effective_output_options(debug=debug, quiet=quiet)
runtime = self.profile.resolve_runtime(self.name, project)
cache_root = runtime.cache_root
@@ -225,57 +397,26 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
log_file = Path(standard["log_file"]).expanduser() if standard.get("log_file") else None
uses_default_log_file = log_file is None
- if dry_run or not self.log_to_file:
- if log_file is not None:
- create_runtime_directory(log_file.parent, cache_root)
- else:
- for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir):
- create_runtime_directory(directory, cache_root)
- if log_file is None:
- log_file = _default_log_file(layout, runtime.primary_log_file)
- create_runtime_directory(log_file.parent, cache_root)
- if inherited_path is None and not dry_run and self.log_to_file:
- create_runtime_directory(layout.owner_root, cache_root)
- create_runtime_directory(layout.run_root, cache_root)
- try:
- run_metadata = {
- "run_id": run_id,
- "owner": runtime_owner,
- "cli": self.name,
- "status": "running",
- "started_at": utc_now().isoformat(timespec="seconds").replace("+00:00", "Z"),
- "project": selected_project_name,
- "project_root": str(selected_project_root) if selected_project_root else None,
- "manifest": str(manifest_path) if manifest_path else None,
- "workspace_root": str(workspace_root) if workspace_root else None,
- }
- run_metadata_path = layout.run_root / "run.json"
- write_private_json(run_metadata_path, run_metadata)
- except OSError:
- pass
- if runtime.write_identity and selected_project_root is not None and not dry_run and self.log_to_file:
- try:
- create_runtime_directory(layout.owner_root, cache_root)
- identity_path = layout.owner_root / "identity.json"
- if not identity_path.exists():
- write_private_json(
- identity_path,
- {
- "schema_version": 1,
- "project": selected_project_name,
- "project_root": str(selected_project_root),
- "manifest": str(manifest_path) if manifest_path is not None else None,
- "checkout_id": layout.owner_root.name,
- },
- )
- except OSError:
- pass
- logger = configure_logger(self.name, log_file, debug, quiet=quiet)
- logger.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment)
- if self.max_log_files is not None and uses_default_log_file and log_file is not None:
- prune_log_files(layout.owner_root / "runs", log_file, self.max_log_files, logger)
-
- return Context(
+ if not dry_run and self.log_to_file and log_file is None:
+ log_file = _default_log_file(layout, runtime.primary_log_file)
+
+ owns_run_metadata = inherited_path is None and not dry_run and self.log_to_file
+ run_metadata_path = layout.run_root / "run.json" if owns_run_metadata else None
+ run_root_was_new = not layout.run_root.exists()
+ temp_dir_was_new = not layout.temp_dir.exists()
+ rollback_empty_directories = tuple(
+ directory
+ for directory in (
+ layout.temp_dir.parent,
+ layout.temp_dir.parent.parent,
+ layout.log_dir,
+ layout.run_root,
+ )
+ if not directory.exists()
+ )
+ log_file_existed = log_file.exists() if log_file is not None else False
+ logger = logging.getLogger(f"base_cli.{self.name}")
+ context = Context(
cli_name=self.name,
run_id=run_id,
runtime_owner=runtime_owner,
@@ -303,9 +444,121 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str],
history_scope=runtime.history_scope,
history_parent_run_id=runtime.history_parent_run_id,
)
+ context._run_metadata_path = run_metadata_path
+
+ logger_activation_started = False
+ try:
+ if owns_run_metadata:
+ create_runtime_directory(layout.run_root, cache_root)
+ if dry_run or not self.log_to_file:
+ if log_file is not None:
+ create_runtime_directory(log_file.parent, cache_root)
+ else:
+ for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir):
+ create_runtime_directory(directory, cache_root)
+ if log_file is not None:
+ create_runtime_directory(log_file.parent, cache_root)
+
+ logger_activation_started = True
+ try:
+ context.log = configure_logger(self.name, log_file, debug, quiet=quiet)
+ except OSError as exc:
+ target = f"persistent log file '{log_file}'" if log_file is not None else "stderr logging"
+ raise RuntimeDirectoryError(f"Unable to configure {target}: {exc}") from exc
+ context.log.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment)
+ if self.max_log_files is not None and uses_default_log_file and log_file is not None:
+ prune_log_files(layout.owner_root / "runs", log_file, self.max_log_files, context.log)
+
+ if runtime.write_identity and selected_project_root is not None and not dry_run and self.log_to_file:
+ try:
+ create_runtime_directory(layout.owner_root, cache_root)
+ identity_path = layout.owner_root / "identity.json"
+ if not identity_path.exists():
+ write_private_json(
+ identity_path,
+ {
+ "schema_version": 1,
+ "project": selected_project_name,
+ "project_root": str(selected_project_root),
+ "manifest": str(manifest_path) if manifest_path is not None else None,
+ "checkout_id": layout.owner_root.name,
+ },
+ )
+ except OSError:
+ pass
+ return context
+ except BaseException:
+ remove_owned_log = bool(
+ uses_default_log_file
+ and log_file is not None
+ and not log_file_existed
+ and run_root_was_new
+ and _path_is_within(log_file, layout.run_root)
+ )
+ _rollback_context_creation(
+ context,
+ logger_activation_started=logger_activation_started,
+ remove_owned_log=remove_owned_log,
+ remove_new_temp=(
+ temp_dir_was_new
+ and _path_is_within(layout.temp_dir, layout.run_root, strict=True)
+ ),
+ empty_directories=rollback_empty_directories,
+ )
+ raise
+
+
+def _rollback_context_creation(
+ context: Context,
+ *,
+ logger_activation_started: bool,
+ remove_owned_log: bool,
+ remove_new_temp: bool,
+ empty_directories: tuple[Path, ...],
+) -> None:
+ if logger_activation_started:
+ keep_temp = context.keep_temp
+ context.keep_temp = True
+ try:
+ try:
+ context.cleanup()
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+ finally:
+ context.keep_temp = keep_temp
+ if remove_new_temp:
+ _remove_new_temp_directory(context.temp_dir)
-def run_app(app: App, argv: list[str] | None = None) -> int:
+ if remove_owned_log and context.log_file is not None:
+ try:
+ context.log_file.unlink()
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+ for directory in sorted(set(empty_directories), key=lambda path: len(path.parts), reverse=True):
+ try:
+ directory.rmdir()
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+
+
+def _remove_new_temp_directory(temp_dir: Path) -> None:
+ try:
+ if temp_dir.exists():
+ shutil.rmtree(temp_dir)
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
+
+
+def _path_is_within(path: Path, root: Path, *, strict: bool = False) -> bool:
+ try:
+ relative = path.resolve().relative_to(root.resolve())
+ except BaseException: # pylint: disable=broad-exception-caught
+ return False
+ return not strict or relative != Path(".")
+
+
+def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool = False) -> int:
"""Run an :class:`App` and return its normalized process exit code."""
try:
@@ -316,26 +569,74 @@ def run_app(app: App, argv: list[str] | None = None) -> int:
explicit_argv = argv is not None
args = list(sys.argv[1:] if argv is None else argv)
+ leading_debug, leading_quiet = _leading_output_flags(args)
+ state = _InvocationState(debug=leading_debug, quiet=leading_quiet)
+ state_token = _INVOCATION_STATE.set(state)
try:
- _reject_equals_option_values(click, args)
- display_command = app.profile.display_command()
- invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command)
- invocation_token = _INVOCATION_ARGV.set(invocation_argv)
try:
- if display_command:
- result = app.click_command.main(args=args, prog_name=display_command, standalone_mode=False)
+ _reject_equals_option_values(click, args)
+ display_command = app.profile.display_command()
+ invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command)
+ invocation_token = _INVOCATION_ARGV.set(invocation_argv)
+ try:
+ if display_command:
+ result = app.click_command.main(args=args, prog_name=display_command, standalone_mode=False)
+ else:
+ result = app.click_command.main(args=args, standalone_mode=False)
+ finally:
+ _reset_context_var(_INVOCATION_ARGV, invocation_token)
+ except click.Abort as exc:
+ outcome = outcome_from_exception(click, exc)
+ if outcome.kind == "interrupted":
+ print("Interrupted.", file=sys.stderr)
else:
- result = app.click_command.main(args=args, standalone_mode=False)
- finally:
- _INVOCATION_ARGV.reset(invocation_token)
- except click.ClickException as exc:
- exc.show()
- return int(exc.exit_code)
- try:
- return _normalize_command_result(result)
- except TypeError as exc:
- print(f"ERROR: {exc}", file=sys.stderr)
- return ExitCode.FAILURE
+ print("Aborted!", file=sys.stderr)
+ return outcome.exit_code
+ except click.ClickException as exc:
+ outcome = outcome_from_exception(click, exc)
+ if outcome.kind == "unexpected_error":
+ if reraise_unexpected:
+ raise
+ _show_unexpected_error(state, exc)
+ return outcome.exit_code
+ exc.show()
+ return outcome.exit_code
+ except KeyboardInterrupt:
+ print("Interrupted.", file=sys.stderr)
+ return ExitCode.INTERRUPTED
+ except SystemExit as exc:
+ if exc.code is not None and not isinstance(exc.code, int):
+ print(str(exc.code), file=sys.stderr)
+ return system_exit_code(exc)
+ except Exception as exc:
+ if reraise_unexpected:
+ raise
+ _show_unexpected_error(state, exc)
+ return ExitCode.FAILURE
+
+ try:
+ return _normalize_command_result(result)
+ except TypeError as exc:
+ print(f"ERROR: {exc}", file=sys.stderr)
+ return ExitCode.FAILURE
+ finally:
+ _reset_context_var(_INVOCATION_STATE, state_token)
+
+
+def _show_unexpected_error(state: _InvocationState, exc: Exception) -> None:
+ print("Error: Unexpected internal error.", file=sys.stderr)
+ if state.run_id is not None:
+ print(f"Run ID: {state.run_id}", file=sys.stderr)
+ if state.log_file is not None:
+ print(f"Diagnostic log: {state.log_file}", file=sys.stderr)
+ traceback_visible = state.debug and not state.quiet
+ if traceback_visible and state.run_id is None:
+ traceback.print_exception(type(exc), exc, exc.__traceback__, file=sys.stderr)
+ elif not traceback_visible:
+ if state.options_parsed:
+ print("Re-run with --debug for a traceback.", file=sys.stderr)
+ else:
+ print("Diagnostic context was unavailable before option parsing completed.", file=sys.stderr)
def _normalize_command_result(result: Any) -> int:
@@ -349,6 +650,19 @@ def _normalize_command_result(result: Any) -> int:
)
+def _leading_output_flags(argv: list[str]) -> tuple[bool, bool]:
+ debug = False
+ quiet = False
+ for token in argv:
+ if token == "--debug":
+ debug = True
+ elif token in ("--quiet", "-q"):
+ quiet = True
+ else:
+ break
+ return debug, quiet
+
+
def _effective_invocation_argv(
app: App,
args: list[str],
diff --git a/lib/python/base_cli/config.py b/lib/python/base_cli/config.py
index 058a701..2b51394 100644
--- a/lib/python/base_cli/config.py
+++ b/lib/python/base_cli/config.py
@@ -4,6 +4,7 @@
from typing import Any
from ._dependencies import require_yaml
+from .errors import ConfigurationError
__all__ = [
@@ -18,11 +19,15 @@ def load_yaml_file(path: Path) -> dict[str, Any]:
yaml = require_yaml("PyYAML is required to load the explicit CLI configuration file.")
try:
- data = yaml.safe_load(path.read_text(encoding="utf-8"))
+ contents = path.read_text(encoding="utf-8")
+ except OSError as exc:
+ raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc
+ try:
+ data = yaml.safe_load(contents)
except yaml.YAMLError as exc:
- raise ValueError(f"Config file '{path}' contains invalid YAML: {exc}") from exc
+ raise ConfigurationError(f"Config file '{path}' contains invalid YAML: {exc}") from exc
if data is None:
return {}
if not isinstance(data, dict):
- raise ValueError(f"Config file '{path}' must contain a YAML mapping.")
+ raise ConfigurationError(f"Config file '{path}' must contain a YAML mapping.")
return data
diff --git a/lib/python/base_cli/context.py b/lib/python/base_cli/context.py
index 20d169c..1efdd7f 100644
--- a/lib/python/base_cli/context.py
+++ b/lib/python/base_cli/context.py
@@ -49,6 +49,7 @@ class Context:
runtime_owner: str = "default"
owner_root: Path | None = None
run_root: Path | None = None
+ _run_metadata_path: Path | None = field(default=None, init=False, repr=False, compare=False)
def on_cleanup(self, hook: Callable[[], None]) -> None:
self.cleanup_hooks.append(hook)
@@ -62,14 +63,14 @@ def bind_project(self, project_name: str | None, project_root: Path, manifest_pa
def _warn_cleanup_failure(self, message: str, *args: object) -> None:
try:
self.log.warning(message, *args)
- except Exception: # pylint: disable=broad-exception-caught
+ except BaseException: # 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
+ except BaseException as exc: # pylint: disable=broad-exception-caught
self._warn_cleanup_failure("Cleanup hook failed: %s", exc)
if not self.keep_temp and self.temp_dir.exists():
try:
@@ -79,18 +80,25 @@ def cleanup(self) -> None:
parent.rmdir()
except OSError:
break
- except OSError as exc:
+ except BaseException as exc: # pylint: disable=broad-exception-caught
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
+ except BaseException as exc: # pylint: disable=broad-exception-caught
self._warn_cleanup_failure("Log handler flush failed: %s", exc)
try:
handler.close()
- except Exception as exc: # pylint: disable=broad-exception-caught
+ except BaseException as exc: # pylint: disable=broad-exception-caught
self._warn_cleanup_failure("Log handler close failed: %s", exc)
- self.log.removeHandler(handler)
+ try:
+ self.log.removeHandler(handler)
+ except BaseException as exc: # pylint: disable=broad-exception-caught
+ self._warn_cleanup_failure("Log handler removal failed: %s", exc)
+ try:
+ self.log.handlers.remove(handler)
+ except BaseException: # pylint: disable=broad-exception-caught
+ pass
def set_current_context(context: Context | None) -> contextvars.Token[Context | None]:
@@ -98,7 +106,15 @@ def set_current_context(context: Context | None) -> contextvars.Token[Context |
def reset_current_context(token: contextvars.Token[Context | None]) -> None:
- _current_context.reset(token)
+ try:
+ _current_context.reset(token)
+ except BaseException: # pylint: disable=broad-exception-caught
+ recover_current_context(token)
+
+
+def recover_current_context(token: contextvars.Token[Context | None]) -> None:
+ previous = token.old_value
+ _current_context.set(None if previous is contextvars.Token.MISSING else previous)
def get_current_context() -> Context:
diff --git a/lib/python/base_cli/errors.py b/lib/python/base_cli/errors.py
new file mode 100644
index 0000000..119f3d5
--- /dev/null
+++ b/lib/python/base_cli/errors.py
@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+
+class ConfigurationError(ValueError):
+ """A user-correctable configuration error that is safe to show."""
diff --git a/lib/python/base_cli/exit_codes.py b/lib/python/base_cli/exit_codes.py
index 75d989b..9ffe9ae 100644
--- a/lib/python/base_cli/exit_codes.py
+++ b/lib/python/base_cli/exit_codes.py
@@ -7,3 +7,4 @@ class ExitCode:
SUCCESS = 0
FAILURE = 1
USAGE_ERROR = 2
+ INTERRUPTED = 130
diff --git a/lib/python/base_cli/logging.py b/lib/python/base_cli/logging.py
index 416e8ca..52b7105 100644
--- a/lib/python/base_cli/logging.py
+++ b/lib/python/base_cli/logging.py
@@ -116,6 +116,12 @@ def format(self, record: logging.LogRecord) -> str:
source = _source_path(record)
level = _level_name(record)
line = f"{timestamp} {level:<7} {source}:{record.lineno} {record.getMessage()}"
+ if record.exc_info:
+ if not record.exc_text:
+ record.exc_text = self.formatException(record.exc_info)
+ line = f"{line}\n{record.exc_text}"
+ if record.stack_info:
+ line = f"{line}\n{self.formatStack(record.stack_info)}"
if not self.use_color:
return line
color = _LEVEL_COLORS.get(record.levelno)
diff --git a/tests/test_app_lifecycle.py b/tests/test_app_lifecycle.py
index fc34839..d86e1e5 100644
--- a/tests/test_app_lifecycle.py
+++ b/tests/test_app_lifecycle.py
@@ -6,8 +6,11 @@
import unittest
from dataclasses import replace
from pathlib import Path
+from unittest import mock
import base_cli
+import base_cli.app as app_module
+import base_cli.context as context_module
from base_cli.testing import invoke
@@ -128,6 +131,93 @@ def main(ctx: base_cli.Context) -> None:
base_cli.get_current_context()
self.assertIn("History finalization failed: history unavailable", result.stderr)
+ def test_non_os_temp_cleanup_failure_still_closes_handlers(self) -> None:
+ app = base_cli.App(name="cleanup-runtime-failure")
+ seen: dict[str, object] = {}
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ seen["logger"] = ctx.log
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ with mock.patch.object(
+ context_module.shutil,
+ "rmtree",
+ side_effect=RuntimeError("cleanup implementation failed"),
+ ):
+ result = invoke(app, [], home=Path(tmpdir))
+
+ self.assertEqual(result.exit_code, 0, result.output)
+ self.assertIn("Temp directory cleanup failed", result.stderr)
+ self.assertEqual(seen["logger"].handlers, [])
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_active_context_reset_interruption_uses_direct_recovery(self) -> None:
+ app = base_cli.App(name="context-reset-interrupt")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir, mock.patch.object(
+ app_module,
+ "reset_current_context",
+ side_effect=KeyboardInterrupt(),
+ ):
+ result = invoke(app, [], home=Path(tmpdir))
+
+ self.assertEqual(result.exit_code, 0, result.output)
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_handler_removal_interruption_uses_direct_detach_fallback(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ logger = logging.Logger("isolated-handler-removal")
+ logger.addHandler(logging.NullHandler())
+ logger.addHandler(logging.NullHandler())
+ logger.removeHandler = mock.Mock(side_effect=KeyboardInterrupt())
+ context = base_cli.Context(
+ cli_name="handler-removal-interrupt",
+ 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=logger,
+ )
+
+ context.cleanup()
+
+ self.assertEqual(logger.handlers, [])
+
+ def test_context_var_reset_helper_restores_previous_value_after_interrupt(self) -> None:
+ class Token:
+ MISSING = object()
+ old_value = "parent"
+
+ class InterruptedVariable:
+ def __init__(self) -> None:
+ self.restored: object | None = None
+
+ def reset(self, _token: object) -> None:
+ raise KeyboardInterrupt()
+
+ def set(self, value: object) -> None:
+ self.restored = value
+
+ variable = InterruptedVariable()
+
+ app_module._reset_context_var(variable, Token())
+
+ self.assertEqual(variable.restored, "parent")
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_app_run.py b/tests/test_app_run.py
index e01a142..b92d485 100644
--- a/tests/test_app_run.py
+++ b/tests/test_app_run.py
@@ -18,11 +18,85 @@ def generic_app(**kwargs: object) -> base_cli.App:
class RunAppTests(unittest.TestCase):
+ @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+ def test_malformed_click_exit_code_before_context_is_an_unexpected_error(self) -> None:
+ import click
+
+ class MalformedExit(click.ClickException):
+ exit_code = object()
+
+ profile = replace(
+ base_cli.CliProfile.generic(),
+ display_command=lambda: (_ for _ in ()).throw(
+ MalformedExit("private pre-context detail")
+ ),
+ )
+ app = base_cli.App(name="malformed-pre-context", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ stderr = io.StringIO()
+ with redirect_stderr(stderr):
+ status = base_cli.run_app(app, [])
+
+ output = stderr.getvalue()
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", output)
+ self.assertIn("Diagnostic context was unavailable", output)
+ self.assertNotIn("private pre-context detail", output)
+
+ @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+ def test_profile_programming_errors_use_the_unexpected_error_boundary(self) -> None:
+ callbacks = (
+ ("discover_project", RuntimeError),
+ ("load_user_config", ValueError),
+ ("resolve_workspace_root", RuntimeError),
+ ("load_config", ValueError),
+ ("resolve_runtime", RuntimeError),
+ ("display_command", ValueError),
+ )
+ for field_name, error_type in callbacks:
+ with self.subTest(field=field_name), tempfile.TemporaryDirectory() as tmpdir:
+ detail = f"private {field_name} detail"
+
+ def fail_callback(*_args: object) -> object:
+ raise error_type(detail)
+
+ profile = replace(base_cli.CliProfile.generic(), **{field_name: fail_callback})
+ app = base_cli.App(name=f"profile-error-{field_name}", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ home = Path(tmpdir)
+ stderr = io.StringIO()
+ with mock.patch.dict(
+ os.environ,
+ {
+ "HOME": str(home),
+ "BASE_CLI_CACHE_DIR": str(home / ".cache"),
+ },
+ ), redirect_stderr(stderr):
+ status = base_cli.run_app(app, [])
+
+ output = stderr.getvalue()
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", output)
+ if field_name == "display_command":
+ self.assertIn("Diagnostic context was unavailable", output)
+ else:
+ self.assertIn("Re-run with --debug for a traceback.", output)
+ self.assertNotIn(detail, output)
+ self.assertNotIn("Traceback", output)
+
@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
def test_run_app_reports_config_errors_without_traceback(self) -> None:
profile = base_cli.CliProfile.generic(
load_config=lambda _project, _explicit: (_ for _ in ()).throw(
- ValueError("workspace must be a mapping when provided.")
+ base_cli.ConfigurationError("workspace must be a mapping when provided.")
)
)
app = base_cli.App(profile=profile, name="bad-config", log_to_file=False)
@@ -45,13 +119,37 @@ def main(ctx: base_cli.Context) -> None:
), redirect_stderr(stderr):
status = base_cli.run_app(app, [])
- self.assertEqual(status, 1)
+ self.assertEqual(status, 2)
self.assertEqual(seen, {})
self.assertIn("workspace must be a mapping", stderr.getvalue())
self.assertNotIn("Traceback", stderr.getvalue())
@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
- def test_run_app_preserves_unexpected_command_exceptions(self) -> None:
+ def test_generic_invalid_yaml_is_a_safe_usage_error(self) -> None:
+ app = base_cli.App(name="invalid-yaml", log_to_file=False)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ config = home / "invalid.yml"
+ config.write_text("broken: [", encoding="utf-8")
+ stderr = io.StringIO()
+ with mock.patch.dict(
+ os.environ,
+ {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(home / ".cache")},
+ ), redirect_stderr(stderr):
+ status = base_cli.run_app(app, ["--config", str(config)])
+
+ self.assertEqual(status, 2)
+ self.assertIn("contains invalid YAML", stderr.getvalue())
+ self.assertNotIn("Traceback", stderr.getvalue())
+
+ @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+ def test_run_app_can_reraise_unexpected_command_exceptions(self) -> None:
app = base_cli.App(name="boom", log_to_file=False)
@app.command()
@@ -69,7 +167,7 @@ def main(ctx: base_cli.Context) -> None:
},
):
with self.assertRaisesRegex(RuntimeError, "boom"):
- base_cli.run_app(app, [])
+ base_cli.run_app(app, [], reraise_unexpected=True)
@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
def test_run_app_reports_invalid_command_return_values(self) -> None:
diff --git a/tests/test_app_run_metadata.py b/tests/test_app_run_metadata.py
new file mode 100644
index 0000000..421d00b
--- /dev/null
+++ b/tests/test_app_run_metadata.py
@@ -0,0 +1,931 @@
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+import logging
+import os
+import tempfile
+import unittest
+from contextlib import redirect_stderr
+from dataclasses import replace
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest import mock
+
+import base_cli
+import base_cli._lifecycle as lifecycle_module
+import base_cli.app as app_module
+from base_cli._lifecycle import RunRecorder
+from base_cli._runtime import runtime_layout
+
+
+def _run(app: base_cli.App, home: Path, args: list[str] | None = None) -> tuple[int, str]:
+ stderr = io.StringIO()
+ with mock.patch.dict(
+ os.environ,
+ {
+ "HOME": str(home),
+ "BASE_CLI_CACHE_DIR": str(home / "cache"),
+ },
+ ), redirect_stderr(stderr):
+ status = base_cli.run_app(app, args or [])
+ return status, stderr.getvalue()
+
+
+def _metadata_files(home: Path) -> list[Path]:
+ return sorted((home / "cache").glob("**/run.json"))
+
+
+def _load_only_metadata(test: unittest.TestCase, home: Path) -> tuple[Path, dict[str, object]]:
+ paths = _metadata_files(home)
+ test.assertEqual(len(paths), 1, paths)
+ payload = json.loads(paths[0].read_text(encoding="utf-8"))
+ test.assertIsInstance(payload, dict)
+ return paths[0], payload
+
+
+def _assert_terminal_metadata(
+ test: unittest.TestCase,
+ payload: dict[str, object],
+ *,
+ status: str,
+ outcome: str,
+ exit_code: int,
+) -> None:
+ test.assertEqual(payload["schema_version"], 1)
+ test.assertEqual(payload["status"], status)
+ test.assertEqual(payload["outcome"], outcome)
+ test.assertEqual(payload["exit_code"], exit_code)
+ test.assertIsInstance(payload["run_id"], str)
+ test.assertIsInstance(payload["owner"], str)
+ test.assertIsInstance(payload["cli"], str)
+ started_text = str(payload["started_at"])
+ ended_text = str(payload["ended_at"])
+ test.assertTrue(started_text.endswith("Z"), started_text)
+ test.assertTrue(ended_text.endswith("Z"), ended_text)
+ started_at = datetime.fromisoformat(started_text.replace("Z", "+00:00"))
+ ended_at = datetime.fromisoformat(ended_text.replace("Z", "+00:00"))
+ test.assertEqual(started_at.utcoffset(), timezone.utc.utcoffset(started_at))
+ test.assertEqual(ended_at.utcoffset(), timezone.utc.utcoffset(ended_at))
+ test.assertGreaterEqual(ended_at, started_at)
+ test.assertIs(type(payload["duration_ms"]), int)
+ test.assertGreaterEqual(payload["duration_ms"], 0)
+
+
+@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+class AppRunMetadataTests(unittest.TestCase):
+ def test_normal_returns_finalize_core_owned_metadata(self) -> None:
+ cases = (
+ ("none", None, 0, "ok", "success"),
+ ("zero", 0, 0, "ok", "success"),
+ ("usage", 2, 2, "error", "usage_error"),
+ ("nonzero", 7, 7, "error", "nonzero_return"),
+ )
+ for name, returned, expected_code, expected_status, expected_outcome in cases:
+ with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
+ app = base_cli.App(name=f"metadata-{name}")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> int | None:
+ del ctx
+ return returned
+
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, expected_code)
+ self.assertEqual(stderr, "")
+ _assert_terminal_metadata(
+ self,
+ metadata,
+ status=expected_status,
+ outcome=expected_outcome,
+ exit_code=expected_code,
+ )
+
+ def test_command_usage_error_preserves_click_rendering_and_exit_code(self) -> None:
+ import click
+
+ app = base_cli.App(name="metadata-usage-error")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise click.UsageError("choose a valid target")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 2)
+ self.assertIn("Usage:", stderr)
+ self.assertIn("Error: choose a valid target", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="usage_error", exit_code=2)
+
+ def test_click_exception_preserves_custom_exit_code(self) -> None:
+ import click
+
+ class Unavailable(click.ClickException):
+ exit_code = 78
+
+ app = base_cli.App(name="metadata-click-error")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise Unavailable("service unavailable")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 78)
+ self.assertIn("Error: service unavailable", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="click_error", exit_code=78)
+
+ def test_zero_code_click_exception_keeps_code_and_status_consistent(self) -> None:
+ import click
+
+ class InformationalExit(click.ClickException):
+ exit_code = 0
+
+ app = base_cli.App(name="metadata-zero-click-error")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise InformationalExit("informational stop")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("Error: informational stop", stderr)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="click_error", exit_code=0)
+
+ def test_malformed_click_exit_code_is_an_unexpected_error(self) -> None:
+ import click
+
+ class MalformedExit(click.ClickException):
+ exit_code = "not-an-exit-code"
+
+ app = base_cli.App(name="metadata-malformed-click-error")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise MalformedExit("private malformed exception detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertNotIn("private malformed exception detail", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1)
+
+ def test_abort_and_keyboard_interrupt_have_distinct_outcomes(self) -> None:
+ import click
+
+ cases = (
+ ("abort", click.Abort(), 1, "aborted", "Aborted!"),
+ ("interrupt", KeyboardInterrupt(), 130, "interrupted", "Interrupted."),
+ )
+ for name, raised, expected_code, expected_outcome, expected_message in cases:
+ with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
+ app = base_cli.App(name=f"metadata-{name}")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise raised
+
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, expected_code)
+ self.assertIn(expected_message, stderr)
+ self.assertNotIn("Traceback", stderr)
+ _assert_terminal_metadata(
+ self,
+ metadata,
+ status="error",
+ outcome=expected_outcome,
+ exit_code=expected_code,
+ )
+
+ def test_explicit_click_and_system_exits_are_normalized(self) -> None:
+ import click
+
+ cases = (
+ ("click", click.exceptions.Exit(9), 9, "error", "nonzero_return", ""),
+ ("system-none", SystemExit(None), 0, "ok", "system_exit", ""),
+ ("system-success", SystemExit(0), 0, "ok", "system_exit", ""),
+ ("system-failure", SystemExit(5), 5, "error", "system_exit", ""),
+ ("system-message", SystemExit("exit detail"), 1, "error", "system_exit", "exit detail\n"),
+ )
+ for name, raised, expected_code, expected_status, expected_outcome, expected_stderr in cases:
+ with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
+ app = base_cli.App(name=f"metadata-{name}")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise raised
+
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, expected_code)
+ self.assertEqual(stderr, expected_stderr)
+ _assert_terminal_metadata(
+ self,
+ metadata,
+ status=expected_status,
+ outcome=expected_outcome,
+ exit_code=expected_code,
+ )
+
+ def test_unexpected_error_is_clean_but_persists_traceback(self) -> None:
+ app = base_cli.App(name="metadata-unexpected")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise RuntimeError("private failure detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ metadata_path, metadata = _load_only_metadata(self, home)
+ log_text = (metadata_path.parent / "logs" / "primary.log").read_text(encoding="utf-8")
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertIn(f"Run ID: {metadata['run_id']}", stderr)
+ self.assertIn("Diagnostic log:", stderr)
+ self.assertIn("Re-run with --debug for a traceback.", stderr)
+ self.assertNotIn("private failure detail", stderr)
+ self.assertNotIn("Traceback", stderr)
+ self.assertIn("Traceback", log_text)
+ self.assertIn("RuntimeError: private failure detail", log_text)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1)
+
+ def test_debug_mirrors_unexpected_traceback_to_stderr(self) -> None:
+ app = base_cli.App(name="metadata-debug")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise RuntimeError("debug failure detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--debug"])
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 1)
+ self.assertIn("Traceback", stderr)
+ self.assertIn("RuntimeError: debug failure detail", stderr)
+ self.assertNotIn("Re-run with --debug", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1)
+
+ def test_unexpected_error_without_file_logging_reports_only_available_diagnostics(self) -> None:
+ app = base_cli.App(name="metadata-no-file-error", log_to_file=False)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise RuntimeError("no-file private detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertIn("Run ID:", stderr)
+ self.assertNotIn("Diagnostic log:", stderr)
+ self.assertNotIn("no-file private detail", stderr)
+ self.assertIn("Re-run with --debug for a traceback.", stderr)
+ self.assertEqual(_metadata_files(home), [])
+
+ def test_debug_shows_traceback_for_failure_before_context_activation(self) -> None:
+ def fail_discovery(_cwd: Path) -> base_cli.ProjectInfo | None:
+ raise RuntimeError("pre-context private detail")
+
+ profile = base_cli.CliProfile.generic(discover_project=fail_discovery)
+ app = base_cli.App(name="metadata-pre-context", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--debug"])
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertIn("Traceback", stderr)
+ self.assertIn("RuntimeError: pre-context private detail", stderr)
+ self.assertNotIn("Re-run with --debug", stderr)
+ self.assertNotIn("Run ID:", stderr)
+ self.assertNotIn("Diagnostic log:", stderr)
+ self.assertEqual(_metadata_files(home), [])
+
+ def test_leading_debug_shows_traceback_for_failure_before_click_parsing(self) -> None:
+ def fail_display_command() -> str | None:
+ raise RuntimeError("pre-parser private detail")
+
+ profile = replace(base_cli.CliProfile.generic(), display_command=fail_display_command)
+ app = base_cli.App(name="metadata-pre-parser", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--debug"])
+
+ self.assertEqual(status, 1)
+ self.assertIn("Traceback", stderr)
+ self.assertIn("RuntimeError: pre-parser private detail", stderr)
+ self.assertNotIn("Diagnostic context was unavailable", stderr)
+
+ def test_debug_in_an_option_value_position_never_exposes_pre_parser_failure(self) -> None:
+ def fail_display_command() -> str | None:
+ raise RuntimeError("value-position private detail")
+
+ profile = replace(base_cli.CliProfile.generic(), display_command=fail_display_command)
+ app = base_cli.App(name="metadata-value-position", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--config", "--debug"])
+
+ self.assertEqual(status, 1)
+ self.assertNotIn("Traceback", stderr)
+ self.assertNotIn("value-position private detail", stderr)
+ self.assertIn("Diagnostic context was unavailable", stderr)
+
+ def test_debug_literal_after_option_terminator_does_not_expose_traceback(self) -> None:
+ def fail_discovery(_cwd: Path) -> base_cli.ProjectInfo | None:
+ raise KeyError("literal-debug private detail")
+
+ profile = base_cli.CliProfile.generic(discover_project=fail_discovery)
+ app = base_cli.App(name="metadata-literal-debug", profile=profile)
+
+ @app.command()
+ @base_cli.argument("value")
+ def main(ctx: base_cli.Context, value: str) -> None:
+ del ctx, value
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--", "--debug"])
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertNotIn("Traceback", stderr)
+ self.assertNotIn("literal-debug private detail", stderr)
+ self.assertIn("Re-run with --debug for a traceback.", stderr)
+
+ def test_config_derived_debug_applies_to_later_startup_failure(self) -> None:
+ def fail_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ raise KeyError("configured-debug private detail")
+
+ profile = replace(
+ base_cli.CliProfile.generic(
+ load_config=lambda _project, _explicit: {"log_level": "debug"},
+ ),
+ resolve_runtime=fail_runtime,
+ )
+ app = base_cli.App(name="metadata-configured-debug-startup", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertIn("Traceback", stderr)
+ self.assertIn("KeyError: 'configured-debug private detail'", stderr)
+ self.assertNotIn("Re-run with --debug", stderr)
+
+ def test_config_debug_with_quiet_keeps_traceback_out_of_stderr_and_shows_hint(self) -> None:
+ profile = base_cli.CliProfile.generic(
+ load_config=lambda _project, _explicit: {"log_level": "debug"},
+ )
+ app = base_cli.App(name="metadata-config-debug-quiet", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ raise RuntimeError("quiet private detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home, ["--quiet"])
+ metadata_path, metadata = _load_only_metadata(self, home)
+ log_text = (metadata_path.parent / "logs" / "primary.log").read_text(encoding="utf-8")
+
+ self.assertEqual(status, 1)
+ self.assertNotIn("Traceback", stderr)
+ self.assertNotIn("quiet private detail", stderr)
+ self.assertIn("Re-run with --debug for a traceback.", stderr)
+ self.assertIn("RuntimeError: quiet private detail", log_text)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1)
+
+ def test_traceback_logging_interruption_cannot_replace_primary_exception(self) -> None:
+ app = base_cli.App(name="metadata-traceback-interrupt")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ ctx.log.debug = mock.Mock(side_effect=KeyboardInterrupt())
+ raise RuntimeError("primary private detail")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 1)
+ self.assertIn("Error: Unexpected internal error.", stderr)
+ self.assertNotIn("Interrupted.", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1)
+
+ def test_history_writer_cannot_override_core_terminal_outcome(self) -> None:
+ observed_codes: list[int] = []
+
+ def history_writer(
+ ctx: base_cli.Context,
+ _argv: list[str],
+ _sensitive: set[str],
+ _started: object,
+ exit_code: int,
+ ) -> None:
+ observed_codes.append(exit_code)
+ assert ctx.run_root is not None
+ (ctx.run_root / "run.json").write_text(
+ json.dumps(
+ {
+ "run_id": ctx.run_id,
+ "status": "error",
+ "exit_code": 99,
+ "command": "metadata-history-authority",
+ "custom_history_field": "preserved",
+ }
+ ),
+ encoding="utf-8",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer)
+ app = base_cli.App(name="metadata-history-authority", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, _ = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(observed_codes, [0])
+ self.assertEqual(metadata["command"], "metadata-history-authority")
+ self.assertEqual(metadata["custom_history_field"], "preserved")
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+
+ def test_history_failure_does_not_block_core_finalization(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="metadata-history-failure", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("History finalization failed: history unavailable", stderr)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+
+ def test_command_duration_excludes_history_writer_latency(self) -> None:
+ observed: list[None] = []
+
+ def history_writer(*_args: object) -> None:
+ observed.append(None)
+
+ profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer)
+ app = base_cli.App(name="metadata-duration", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(
+ app_module.time,
+ "monotonic_ns",
+ side_effect=(1_000_000_000, 1_025_600_000),
+ ):
+ status, _ = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(observed, [None])
+ self.assertEqual(metadata["duration_ms"], 26)
+
+ def test_terminal_write_failure_preserves_primary_outcome_and_discards_running_marker(self) -> None:
+ cases = (("success", False, 0), ("failure", True, 1))
+ for name, fail_command, expected_code in cases:
+ with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
+ app = base_cli.App(name=f"metadata-finalize-{name}")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ if fail_command:
+ raise RuntimeError("command failure")
+
+ home = Path(tmpdir)
+ with mock.patch.object(RunRecorder, "finish", side_effect=RuntimeError("metadata unavailable")):
+ status, stderr = _run(app, home)
+
+ self.assertEqual(status, expected_code)
+ self.assertIn("Run metadata finalization failed", stderr)
+ self.assertEqual(_metadata_files(home), [])
+ self.assertEqual(logging.getLogger(f"base_cli.metadata-finalize-{name}").handlers, [])
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_metadata_recovery_failure_cannot_replace_command_outcome(self) -> None:
+ app = base_cli.App(name="metadata-recovery-failure")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with (
+ mock.patch.object(RunRecorder, "finish", side_effect=RuntimeError("finish unavailable")),
+ mock.patch.object(
+ RunRecorder,
+ "discard_owned_record",
+ side_effect=RuntimeError("recovery unavailable"),
+ ),
+ ):
+ status, stderr = _run(app, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("Run metadata finalization failed", stderr)
+ self.assertIn("Run metadata recovery failed", stderr)
+ self.assertEqual(logging.getLogger("base_cli.metadata-recovery-failure").handlers, [])
+
+ def test_finish_failure_removes_matching_history_owned_terminal_fields(self) -> None:
+ def history_writer(
+ ctx: base_cli.Context,
+ _argv: list[str],
+ _sensitive: set[str],
+ _started: object,
+ _exit_code: int,
+ ) -> None:
+ assert ctx.run_root is not None
+ (ctx.run_root / "run.json").write_text(
+ json.dumps({"run_id": ctx.run_id, "status": "error", "exit_code": 99}),
+ encoding="utf-8",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer)
+ app = base_cli.App(name="metadata-history-finish-failure", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(RunRecorder, "finish", side_effect=OSError("finish unavailable")):
+ status, stderr = _run(app, home)
+ metadata_files = _metadata_files(home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("Run metadata finalization failed", stderr)
+ self.assertEqual(metadata_files, [])
+
+ def test_partial_terminal_write_is_removed_without_masking_success(self) -> None:
+ app = base_cli.App(name="metadata-partial-write")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ real_write = lifecycle_module.write_private_json
+ calls = 0
+
+ def fail_second_write(path: Path, value: dict[str, object]) -> None:
+ nonlocal calls
+ calls += 1
+ if calls == 2:
+ path.write_text("{", encoding="utf-8")
+ raise OSError("partial terminal write")
+ real_write(path, value)
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(lifecycle_module, "write_private_json", side_effect=fail_second_write):
+ status, stderr = _run(app, home)
+ metadata_files = _metadata_files(home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(calls, 2)
+ self.assertIn("Run metadata finalization failed", stderr)
+ self.assertEqual(metadata_files, [])
+
+ def test_terminal_merge_rejects_metadata_from_a_different_run(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ cache_root = home / "cache"
+ layout = runtime_layout(cache_root, "metadata-stale-merge", "new-run")
+ layout.run_root.mkdir(parents=True)
+ metadata_path = layout.run_root / "run.json"
+ metadata_path.write_text(
+ json.dumps({"run_id": "old-run", "status": "ok", "stale_field": "do not copy"}),
+ encoding="utf-8",
+ )
+
+ def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ return base_cli.RuntimeBinding(
+ cache_root=cache_root,
+ layout=layout,
+ application_home=None,
+ runtime_owner="custom-owner",
+ project_root=None,
+ project_name=None,
+ inherited_path=None,
+ history_parent_run_id=None,
+ run_id="new-run",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime)
+ app = base_cli.App(name="metadata-stale-merge", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with mock.patch.object(RunRecorder, "start", return_value=None):
+ status, _ = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(metadata["run_id"], "new-run")
+ self.assertEqual(metadata["owner"], "custom-owner")
+ self.assertNotIn("stale_field", metadata)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+
+ def test_start_write_failure_can_recover_with_terminal_snapshot(self) -> None:
+ app = base_cli.App(name="metadata-start-failure")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(RunRecorder, "start", side_effect=RuntimeError("metadata unavailable")):
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("Run metadata start failed: metadata unavailable", stderr)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+
+ def test_interrupt_during_metadata_start_still_finalizes_and_tears_down(self) -> None:
+ app = base_cli.App(name="metadata-start-interrupt")
+ called: list[None] = []
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ called.append(None)
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(RunRecorder, "start", side_effect=KeyboardInterrupt()):
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 130)
+ self.assertEqual(called, [])
+ self.assertIn("Interrupted.", stderr)
+ _assert_terminal_metadata(self, metadata, status="error", outcome="interrupted", exit_code=130)
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_interrupt_during_history_cannot_replace_settled_command_outcome(self) -> None:
+ def interrupt_history(*_args: object) -> None:
+ raise KeyboardInterrupt()
+
+ profile = replace(base_cli.CliProfile.generic(), history_writer=interrupt_history)
+ app = base_cli.App(name="metadata-history-interrupt", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("History finalization failed: KeyboardInterrupt", stderr)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+ self.assertEqual(logging.getLogger("base_cli.metadata-history-interrupt").handlers, [])
+
+ def test_interrupt_during_terminal_write_cannot_replace_settled_command_outcome(self) -> None:
+ app = base_cli.App(name="metadata-finish-interrupt")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ calls = 0
+
+ def interrupt_finish(recorder: RunRecorder, *args: object, **kwargs: object) -> None:
+ nonlocal calls
+ del recorder, args, kwargs
+ calls += 1
+ raise KeyboardInterrupt()
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(RunRecorder, "finish", new=interrupt_finish):
+ status, stderr = _run(app, home)
+ metadata_files = _metadata_files(home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(calls, 1)
+ self.assertIn("Run metadata finalization failed", stderr)
+ self.assertEqual(metadata_files, [])
+ self.assertEqual(logging.getLogger("base_cli.metadata-finish-interrupt").handlers, [])
+
+ def test_interrupt_during_cleanup_cannot_replace_settled_command_outcome(self) -> None:
+ app = base_cli.App(name="metadata-cleanup-interrupt")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ ctx.on_cleanup(lambda: (_ for _ in ()).throw(KeyboardInterrupt()))
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, stderr = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertIn("Cleanup hook failed", stderr)
+ _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0)
+ self.assertEqual(logging.getLogger("base_cli.metadata-cleanup-interrupt").handlers, [])
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_terminal_metadata_refreshes_project_binding(self) -> None:
+ app = base_cli.App(name="metadata-project-binding")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ ctx.bind_project("bound-project", Path("/tmp/bound-project"), Path("/tmp/bound-project/project.yml"))
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ status, _ = _run(app, home)
+ _, metadata = _load_only_metadata(self, home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(metadata["project"], "bound-project")
+ self.assertEqual(metadata["project_root"], str(Path("/tmp/bound-project").resolve()))
+ self.assertEqual(metadata["manifest"], str(Path("/tmp/bound-project/project.yml").resolve()))
+
+ def test_parse_error_and_no_file_modes_do_not_own_run_metadata(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ parse_app = base_cli.App(name="metadata-parse-error")
+
+ @parse_app.command()
+ @base_cli.option("--name", required=True)
+ def parse_main(ctx: base_cli.Context, name: str) -> None:
+ del ctx, name
+
+ status, _ = _run(parse_app, home)
+ self.assertEqual(status, 2)
+ self.assertEqual(_metadata_files(home), [])
+
+ informational_app = base_cli.App(name="metadata-informational", version="1.2.3")
+
+ @informational_app.command()
+ def informational_main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ for args in (["--help"], ["--version"]):
+ status, _ = _run(informational_app, home, args)
+ self.assertEqual(status, 0)
+ self.assertEqual(_metadata_files(home), [])
+
+ no_file_app = base_cli.App(name="metadata-no-file", log_to_file=False)
+
+ @no_file_app.command()
+ def no_file_main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ status, _ = _run(no_file_app, home)
+ self.assertEqual(status, 0)
+ self.assertEqual(_metadata_files(home), [])
+
+ dry_run_app = base_cli.App(name="metadata-dry-run")
+
+ @dry_run_app.command()
+ @base_cli.option("--dry-run", is_flag=True, dry_run=True)
+ def dry_run_main(ctx: base_cli.Context, dry_run: bool) -> None:
+ self.assertTrue(ctx.dry_run)
+ self.assertTrue(dry_run)
+
+ status, _ = _run(dry_run_app, home, ["--dry-run"])
+ self.assertEqual(status, 0)
+ self.assertEqual(_metadata_files(home), [])
+
+ def test_inherited_runtime_does_not_mutate_parent_metadata(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ cache_root = home / "cache"
+ parent_run = cache_root / "parent" / "runs" / "parent-run"
+ parent_run.mkdir(parents=True)
+ parent_metadata = parent_run / "run.json"
+ parent_payload = {"run_id": "parent-run", "status": "running", "custom": "parent-owned"}
+ parent_metadata.write_text(json.dumps(parent_payload), encoding="utf-8")
+
+ def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ return base_cli.RuntimeBinding(
+ cache_root=cache_root,
+ layout=runtime_layout(
+ cache_root,
+ "metadata-inherited",
+ "child-run",
+ namespace="parent",
+ inherited_run_root=parent_run,
+ ),
+ application_home=None,
+ runtime_owner="parent",
+ project_root=None,
+ project_name=None,
+ inherited_path=parent_run,
+ history_parent_run_id="parent-run",
+ run_id="child-run",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime)
+ app = base_cli.App(name="metadata-inherited", profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+
+ status, _ = _run(app, home)
+
+ self.assertEqual(status, 0)
+ self.assertEqual(json.loads(parent_metadata.read_text(encoding="utf-8")), parent_payload)
+ self.assertEqual(_metadata_files(home), [parent_metadata])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_app_runtime_errors.py b/tests/test_app_runtime_errors.py
index 2e2354d..5321ae5 100644
--- a/tests/test_app_runtime_errors.py
+++ b/tests/test_app_runtime_errors.py
@@ -73,3 +73,31 @@ def main(ctx: base_cli.Context) -> None:
self.assertIn("Unable to create runtime directory", error)
self.assertIn(str(cache_root / "cache-failure" / "runs"), error)
self.assertNotIn("Traceback", error)
+
+ @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+ def test_run_app_reports_framework_log_path_failure_without_traceback(self) -> None:
+ app = generic_app(name="log-open-failure")
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command body should not run when logging setup fails")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ not_a_directory = root / "not-a-directory"
+ not_a_directory.write_text("file", encoding="utf-8")
+ log_file = not_a_directory / "primary.log"
+ stderr = io.StringIO()
+ with mock.patch.dict(
+ os.environ,
+ {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(root / "cache")},
+ ), redirect_stderr(stderr):
+ exit_code = base_cli.run_app(app, ["--log-file", str(log_file)])
+
+ error = stderr.getvalue()
+ self.assertEqual(exit_code, 1)
+ self.assertIn(f"Unable to create runtime directory '{not_a_directory}'", error)
+ self.assertNotIn("Unexpected internal error", error)
+ self.assertNotIn("Traceback", error)
diff --git a/tests/test_app_startup_transaction.py b/tests/test_app_startup_transaction.py
new file mode 100644
index 0000000..b22ae74
--- /dev/null
+++ b/tests/test_app_startup_transaction.py
@@ -0,0 +1,243 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import logging
+import os
+import tempfile
+import unittest
+from dataclasses import replace
+from pathlib import Path
+from unittest import mock
+
+import base_cli
+import base_cli.app as app_module
+from base_cli._runtime import runtime_layout
+
+
+def _run(app: base_cli.App, home: Path) -> int:
+ with mock.patch.dict(
+ os.environ,
+ {
+ "HOME": str(home),
+ "BASE_CLI_CACHE_DIR": str(home / "cache"),
+ },
+ ):
+ return base_cli.run_app(app, [])
+
+
+@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
+class AppStartupTransactionTests(unittest.TestCase):
+ def test_retention_failure_rolls_back_new_bundle_and_logger(self) -> None:
+ app = base_cli.App(name="startup-retention", max_log_files=1)
+ called: list[None] = []
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ called.append(None)
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ preserved = home / "cache" / "startup-retention" / "cache" / "existing.txt"
+ preserved.parent.mkdir(parents=True)
+ preserved.write_text("keep", encoding="utf-8")
+
+ with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")):
+ status = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertEqual(called, [])
+ self.assertEqual(list((home / "cache").glob("**/run.json")), [])
+ self.assertEqual(list((home / "cache").glob("**/primary.log")), [])
+ self.assertEqual(logging.getLogger("base_cli.startup-retention").handlers, [])
+ self.assertEqual(preserved.read_text(encoding="utf-8"), "keep")
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_partial_logger_failure_closes_handlers_and_removes_new_bundle(self) -> None:
+ app = base_cli.App(name="startup-logger")
+ original_configure_logger = app_module.configure_logger
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ def fail_after_logger_setup(*args: object, **kwargs: object) -> logging.Logger:
+ original_configure_logger(*args, **kwargs)
+ raise OSError("logger unavailable")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(app_module, "configure_logger", side_effect=fail_after_logger_setup):
+ status = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertEqual(list((home / "cache").glob("**/run.json")), [])
+ self.assertEqual(list((home / "cache").glob("**/primary.log")), [])
+ self.assertEqual(logging.getLogger("base_cli.startup-logger").handlers, [])
+ with self.assertRaisesRegex(RuntimeError, "context is not active"):
+ base_cli.get_current_context()
+
+ def test_inherited_startup_failure_never_finalizes_or_deletes_parent_bundle(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ cache_root = home / "cache"
+ parent_run = cache_root / "parent" / "runs" / "parent-run"
+ parent_run.mkdir(parents=True)
+ parent_metadata = parent_run / "run.json"
+ parent_payload = {"run_id": "parent-run", "status": "running"}
+ parent_metadata.write_text(json.dumps(parent_payload), encoding="utf-8")
+
+ def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ return base_cli.RuntimeBinding(
+ cache_root=cache_root,
+ layout=runtime_layout(
+ cache_root,
+ "startup-inherited",
+ "child-run",
+ namespace="parent",
+ inherited_run_root=parent_run,
+ ),
+ application_home=None,
+ runtime_owner="parent",
+ project_root=None,
+ project_name=None,
+ inherited_path=parent_run,
+ history_parent_run_id="parent-run",
+ run_id="child-run",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime)
+ app = base_cli.App(name="startup-inherited", max_log_files=1, profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")):
+ status = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertEqual(json.loads(parent_metadata.read_text(encoding="utf-8")), parent_payload)
+ self.assertTrue(parent_run.is_dir())
+ self.assertFalse((parent_run / "tmp" / "startup-inherited" / "child-run").exists())
+ self.assertEqual(logging.getLogger("base_cli.startup-inherited").handlers, [])
+
+ def test_startup_rollback_preserves_preexisting_temp_content(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ cache_root = home / "cache"
+ layout = runtime_layout(cache_root, "startup-preexisting", "fixed-run")
+ layout.temp_dir.mkdir(parents=True)
+ marker = layout.temp_dir / "existing.txt"
+ marker.write_text("keep", encoding="utf-8")
+
+ def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ return base_cli.RuntimeBinding(
+ cache_root=cache_root,
+ layout=layout,
+ application_home=None,
+ runtime_owner="startup-preexisting",
+ project_root=None,
+ project_name=None,
+ inherited_path=None,
+ history_parent_run_id=None,
+ run_id="fixed-run",
+ )
+
+ profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime)
+ app = base_cli.App(name="startup-preexisting", max_log_files=1, profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")):
+ status = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertEqual(marker.read_text(encoding="utf-8"), "keep")
+ self.assertEqual(logging.getLogger("base_cli.startup-preexisting").handlers, [])
+
+ def test_startup_rollback_never_recursively_deletes_temp_outside_run_root(self) -> None:
+ with tempfile.TemporaryDirectory() as tmpdir:
+ root = Path(tmpdir)
+ home = root / "home"
+ cache_root = home / "cache"
+ external_temp = root / "profile-selected-external-temp"
+ layout = replace(
+ runtime_layout(cache_root, "startup-contained", "fixed-run"),
+ temp_dir=external_temp,
+ )
+
+ def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding:
+ return base_cli.RuntimeBinding(
+ cache_root=cache_root,
+ layout=layout,
+ application_home=None,
+ runtime_owner="startup-contained",
+ project_root=None,
+ project_name=None,
+ inherited_path=None,
+ history_parent_run_id=None,
+ run_id="fixed-run",
+ )
+
+ marker = external_temp / "created-during-startup.txt"
+
+ def fail_retention(*_args: object) -> None:
+ marker.write_text("preserve", encoding="utf-8")
+ raise RuntimeError("retention unavailable")
+
+ profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime)
+ app = base_cli.App(name="startup-contained", max_log_files=1, profile=profile)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ with mock.patch.object(app_module, "prune_log_files", side_effect=fail_retention):
+ status = _run(app, home)
+
+ self.assertEqual(status, 1)
+ self.assertEqual(marker.read_text(encoding="utf-8"), "preserve")
+ self.assertEqual(logging.getLogger("base_cli.startup-contained").handlers, [])
+
+ def test_rollback_cleanup_runtime_error_cannot_mask_startup_failure(self) -> None:
+ app = base_cli.App(name="startup-rollback-runtime", max_log_files=1)
+
+ @app.command()
+ def main(ctx: base_cli.Context) -> None:
+ del ctx
+ self.fail("command should not run")
+
+ with tempfile.TemporaryDirectory() as tmpdir:
+ home = Path(tmpdir)
+ with mock.patch.object(
+ app_module,
+ "prune_log_files",
+ side_effect=RuntimeError("primary startup failure"),
+ ), mock.patch.object(
+ app_module.shutil,
+ "rmtree",
+ side_effect=RuntimeError("secondary rollback failure"),
+ ):
+ with self.assertRaisesRegex(RuntimeError, "primary startup failure"):
+ with mock.patch.dict(
+ os.environ,
+ {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(home / "cache")},
+ ):
+ base_cli.run_app(app, [], reraise_unexpected=True)
+
+ self.assertEqual(logging.getLogger("base_cli.startup-rollback-runtime").handlers, [])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_logging.py b/tests/test_logging.py
index 3a7d52f..e097b2c 100644
--- a/tests/test_logging.py
+++ b/tests/test_logging.py
@@ -37,6 +37,20 @@ def test_configure_logger_accepts_custom_formatter(self) -> None:
self.assertEqual(stream.getvalue().strip(), "INFO:hello formatter")
+ def test_base_formatter_includes_exception_tracebacks(self) -> None:
+ stream = io.StringIO()
+ logger = base_cli.configure_logger("exception-traceback", None, debug=True, stream=stream)
+
+ try:
+ raise RuntimeError("diagnostic detail")
+ except RuntimeError:
+ logger.debug("unexpected command exception", exc_info=True)
+
+ output = stream.getvalue()
+ self.assertIn("unexpected command exception", output)
+ self.assertIn("Traceback", output)
+ self.assertIn("RuntimeError: diagnostic detail", output)
+
def test_configure_logger_defaults_to_stderr_and_base_formatter(self) -> None:
stream = io.StringIO()
diff --git a/tests/test_public_api.py b/tests/test_public_api.py
index e52394c..a00e1f7 100644
--- a/tests/test_public_api.py
+++ b/tests/test_public_api.py
@@ -32,6 +32,7 @@ def test_version_resolution_ignores_unrelated_ancestor_version_files(self) -> No
def test_facade_exports_supported_modules_functions_and_types(self) -> None:
expected = {
"CommandProtocolError",
+ "ConfigurationError",
"command_filters",
"command_matches",
"command_protocol",
@@ -47,6 +48,7 @@ def test_facade_exports_supported_modules_functions_and_types(self) -> None:
self.assertFalse(hasattr(base_cli, "UserConfig"))
self.assertIs(base_cli.command_filters, command_filters)
self.assertIs(base_cli.command_protocol, command_protocol)
+ self.assertTrue(issubclass(base_cli.ConfigurationError, ValueError))
def test_module_all_surfaces_are_explicit(self) -> None:
self.assertEqual(