Classify interrupted runs as aborted in history

README.md

@@ -714,6 +714,10 @@ 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.
 
+Terminal lifecycle status is `ok` for exit code `0`, `aborted` for the
+interrupt exit code `130`, and `error` for other nonzero exit codes. The
+technical outcome remains precise: Ctrl+C is recorded as `interrupted`.
+
 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

docs/cache-ownership-and-layout.md

@@ -29,8 +29,9 @@ 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`,
+`duration_ms`. Terminal status is `ok` for exit code zero, `aborted` for the
+interrupt exit code 130, and `error` for other nonzero exit codes. The outcome
+discriminator is one of `success`,
 `usage_error`, `nonzero_return`, `click_error`, `aborted`, `interrupted`,
 `system_exit`, or `unexpected_error`.
 

lib/python/base_cli/_lifecycle.py

@@ -9,7 +9,7 @@
 from ._runtime import refresh_run_bundle_index
 from .context import Context
 from .exit_codes import ExitCode
-from .history import compact_optional_path, format_timestamp
+from .history import compact_optional_path, format_timestamp, status_for_exit_code
 
 
 @dataclass(frozen=True)
@@ -116,17 +116,17 @@ def outcome_from_exit_code(exit_code: int) -> InvocationOutcome:
         return InvocationOutcome("success", "ok", exit_code)
     if exit_code == ExitCode.USAGE_ERROR:
         return InvocationOutcome("usage_error", "error", exit_code)
-    return InvocationOutcome("nonzero_return", "error", exit_code)
+    return InvocationOutcome("nonzero_return", status_for_exit_code(exit_code), exit_code)
 
 
 def outcome_from_exception(click: Any, exc: BaseException) -> InvocationOutcome:
     if isinstance(exc, KeyboardInterrupt):
-        return InvocationOutcome("interrupted", "error", ExitCode.INTERRUPTED)
+        return InvocationOutcome("interrupted", "aborted", 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("interrupted", "aborted", ExitCode.INTERRUPTED)
         return InvocationOutcome("aborted", "error", ExitCode.FAILURE)
     if isinstance(exc, click.exceptions.Exit):
         exit_code = _click_exception_exit_code(exc)
@@ -137,15 +137,15 @@ def outcome_from_exception(click: Any, exc: BaseException) -> InvocationOutcome:
         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)
+        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)
+        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("system_exit", status_for_exit_code(exit_code), exit_code)
     return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE)
 
 
@@ -157,10 +157,6 @@ def system_exit_code(exc: SystemExit) -> int:
     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)

lib/python/base_cli/_runtime.py

@@ -385,7 +385,7 @@ def _discover_run_bundles(
         stale_running = running and max_age_seconds is not None and age >= max_age_seconds
         if running and not stale_running:
             continue
-        if status not in {"running", "ok", "error"}:
+        if status not in {"running", "ok", "aborted", "error"}:
             continue
         resolved = _safe_resolved_path(child)
         try:

lib/python/base_cli/history.py

@@ -18,6 +18,7 @@
     _msvcrt = None  # type: ignore[assignment]
 
 from ._private_files import restrict_file, write_private_json
+from .exit_codes import ExitCode
 from .redaction import REDACTED, is_secret_key, option_name_to_parameter, redact_argv, redact_text_value
 
 if TYPE_CHECKING:
@@ -40,6 +41,7 @@
     "parse_positive_int",
     "redact_history_argv",
     "redact_history_text",
+    "status_for_exit_code",
     "utc_now",
     "write_history_record",
     "write_primary_record",
@@ -56,6 +58,15 @@ def utc_now() -> datetime:
     return datetime.now(timezone.utc)
 
 
+def status_for_exit_code(exit_code: int) -> str:
+    """Map a process result to its terminal lifecycle status."""
+    if exit_code == ExitCode.SUCCESS:
+        return "ok"
+    if exit_code == ExitCode.INTERRUPTED:
+        return "aborted"
+    return "error"
+
+
 def build_finished_record(
     context: Context[Any, Any, Any],
     argv: list[str],
@@ -77,7 +88,7 @@ def build_finished_record(
         "ended_at": format_timestamp(ended_at),
         "duration_ms": duration_ms(started_at, ended_at),
         "exit_code": exit_code,
-        "status": "ok" if exit_code == 0 else "error",
+        "status": status_for_exit_code(exit_code),
         "owner": context.runtime_owner,
         "bundle_path": compact_path(context.run_root or context.state_dir),
         "os": normalized_os(),
@@ -127,7 +138,7 @@ def write_primary_record(
         "ended_at": format_timestamp(ended_at),
         "duration_ms": duration_ms(started_at, ended_at),
         "exit_code": exit_code,
-        "status": "ok" if exit_code == 0 else "error",
+        "status": status_for_exit_code(exit_code),
         "os": normalized_os(),
         "scope": scope,
     }

tests/test_adversarial_regressions.py

@@ -287,7 +287,7 @@ def main(ctx: base_cli.Context) -> None:
         self.assertIn("Interrupted.", log_text)
         self.assertEqual(len(metadata_paths), 1)
         self.assertEqual(payload["outcome"], "interrupted")
-        self.assertEqual(payload["status"], "error")
+        self.assertEqual(payload["status"], "aborted")
         self.assertEqual(temp_contents, ())
         self.assertEqual(logger_handlers, [])
         with self.assertRaisesRegex(RuntimeError, "context is not active"):
@@ -385,7 +385,7 @@ def main(ctx: base_cli.Context) -> None:
         self.assertIn("Interrupted.", log_text)
         self.assertEqual(len(metadata_paths), 1)
         self.assertEqual(payload["outcome"], "interrupted")
-        self.assertEqual(payload["status"], "error")
+        self.assertEqual(payload["status"], "aborted")
         self.assertEqual(temp_contents, ())
 
 

tests/test_app_run_metadata.py

@@ -84,7 +84,7 @@ def test_normal_returns_finalize_core_owned_metadata(self) -> None:
             ("zero", 0, 0, "ok", "success"),
             ("usage", 2, 2, "error", "usage_error"),
             ("nonzero", 7, 7, "error", "nonzero_return"),
-            ("returned-interrupted", 130, 130, "error", "nonzero_return"),
+            ("returned-interrupted", 130, 130, "aborted", "nonzero_return"),
         )
         for name, returned, expected_code, expected_status, expected_outcome in cases:
             with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
@@ -203,10 +203,10 @@ 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."),
+            ("abort", click.Abort(), 1, "error", "aborted", "Aborted!"),
+            ("interrupt", KeyboardInterrupt(), 130, "aborted", "interrupted", "Interrupted."),
         )
-        for name, raised, expected_code, expected_outcome, expected_message in cases:
+        for name, raised, expected_code, expected_status, expected_outcome, expected_message in cases:
             with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir:
                 app = base_cli.App(name=f"metadata-{name}")
 
@@ -231,7 +231,7 @@ def main(
                 _assert_terminal_metadata(
                     self,
                     metadata,
-                    status="error",
+                    status=expected_status,
                     outcome=expected_outcome,
                     exit_code=expected_code,
                 )
@@ -241,7 +241,7 @@ def test_explicit_click_and_system_exits_are_normalized(self) -> None:
 
         cases = (
             ("click", click.exceptions.Exit(9), 9, "error", "nonzero_return", ""),
-            ("click-interrupted-code", click.exceptions.Exit(130), 130, "error", "nonzero_return", ""),
+            ("click-interrupted-code", click.exceptions.Exit(130), 130, "aborted", "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", ""),
@@ -768,7 +768,7 @@ def main(ctx: base_cli.Context) -> None:
         self.assertEqual(status, 130)
         self.assertEqual(called, [])
         self.assertIn("Interrupted.", stderr)
-        _assert_terminal_metadata(self, metadata, status="error", outcome="interrupted", exit_code=130)
+        _assert_terminal_metadata(self, metadata, status="aborted", outcome="interrupted", exit_code=130)
         with self.assertRaisesRegex(RuntimeError, "context is not active"):
             base_cli.get_current_context()
 

tests/test_generic_core.py

@@ -93,6 +93,33 @@ def test_finished_record_omits_log_path_when_file_logging_is_disabled(self) -> N
 
         self.assertNotIn("log_path", record)
 
+    def test_finished_record_classifies_interrupt_exit_as_aborted(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            context = Context(
+                cli_name="demo_tool",
+                run_id="run-1",
+                state_dir=root / "state",
+                log_dir=root / "logs",
+                cache_dir=root / "cache",
+                temp_dir=root / "tmp",
+                log_file=root / "logs" / "run.log",
+                config={},
+                environment="dev",
+                debug=False,
+                keep_temp=False,
+                log=logging.getLogger("generic-core-aborted-test"),
+            )
+            record = history.build_finished_record(
+                context,
+                ["demo_tool"],
+                set(),
+                history.utc_now() - timedelta(seconds=1),
+                base_cli.ExitCode.INTERRUPTED,
+            )
+
+        self.assertEqual(record["status"], "aborted")
+
     def test_base_specific_path_helpers_are_not_in_generic_module(self) -> None:
         import base_cli.paths as paths
 

tests/test_run_bundle_retention.py

@@ -93,6 +93,22 @@ def test_stale_running_bundle_is_recoverable_with_age_bound(self) -> None:
 
             self.assertFalse(stale.exists())
 
+    def test_aborted_bundle_is_indexed_as_terminal(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir) / "runs"
+            root.mkdir()
+            _bundle(root, "aborted", status="aborted")
+
+            prune_run_bundles(
+                root,
+                policy=RetentionPolicy(max_bundles=1),
+                logger=logging.getLogger(__name__),
+            )
+
+            payload = json.loads((root / ".base-cli-run-index.json").read_text(encoding="utf-8"))
+
+        self.assertEqual([bundle["status"] for bundle in payload["bundles"]], ["aborted"])
+
     def test_symlink_bundle_is_not_followed_or_removed(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             root = Path(tmpdir) / "runs"