fix: retain terminal bundles with active leases

CHANGELOG.md

@@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Preserve run bundles while their lease is held through terminal metadata
+  writing and cleanup; recheck lease state before deletion.
 - Preserve explicit application identities losslessly while using
   collision-resistant, path-safe runtime namespace components.
 - Give `BatteriesIncludedConfigLoader.cli_name` a documented identity role by

README.md

@@ -883,8 +883,9 @@ app = base_cli.App(
 Retention runs during startup after the current run's default log file is
 resolved. The active invocation, inherited parent bundle, and bundles marked
 `preserve` (including `--keep-temp`) are never removed. Each lifecycle-owned
-running bundle also holds an advisory `.base-cli-run-lease` for its lifetime;
-retention never removes a bundle whose lease is active. A stale `running`
+bundle holds an advisory `.base-cli-run-lease` through final cleanup, including
+the brief period after metadata becomes terminal; retention never removes a
+bundle whose lease is active or whose lease file is present but liveness cannot be established. A stale `running`
 bundle is eligible for crash recovery only when an age bound is configured and
 its lease can be acquired, proving that the original process has exited.
 Missing, unreadable, or unsupported leases fail closed and remain retained for

lib/python/base_cli/_lifecycle.py

@@ -61,7 +61,11 @@ def finish(
         write_private_json(self.context._run_metadata_path, metadata)
         owner_root = self.context.owner_root
         if owner_root is not None:
-            refresh_run_bundle_index(owner_root / "runs", logger=self.context.log)
+            refresh_run_bundle_index(
+                owner_root / "runs",
+                current_run_root=self.context.run_root,
+                logger=self.context.log,
+            )
 
     def _existing_metadata(self) -> dict[str, Any]:
         path = self.context._run_metadata_path

lib/python/base_cli/_runtime.py

@@ -450,6 +450,7 @@ def prune_run_bundles(
 def refresh_run_bundle_index(
     runs_root: Path,
     *,
+    current_run_root: Path | None = None,
     logger: logging.Logger | None = None,
 ) -> None:
     """Refresh the diagnostic bundle index after a run becomes terminal."""
@@ -468,7 +469,7 @@ def refresh_run_bundle_index(
             size_budget=0,
         )
         with _retention_lock(runs_root):
-            _write_run_index(runs_root, bundles, log)
+            _write_run_index(runs_root, bundles, log, current_run_root=current_run_root)
     except (OSError, RuntimeError) as exc:
         log.debug("Could not refresh run bundle index under '%s': %s", runs_root, exc)
 
@@ -506,11 +507,18 @@ def _discover_run_bundles(
                 continue
         age = max(0.0, now - started_at)
         running = status == "running"
+        # The owner keeps its lease through cleanup, which occurs after the
+        # run metadata has been made terminal. Liveness therefore protects
+        # every state, not only the transient "running" state. Legacy bundle
+        # fixtures may have no lease file at all; terminal bundles without a
+        # lease are eligible, while unknown liveness remains fail-closed for
+        # running records or a present but unreadable lease.
+        lease_state = _run_lease_state(child)
+        if _lease_blocks_removal(child, lease_state):
+            continue
+        if running and lease_state != "inactive":
+            continue
         if running:
-            # A running record is removable only when the lease proves that
-            # its owner has exited. Missing or unreadable leases fail closed.
-            if _run_lease_state(child) != "inactive":
-                continue
             if max_age_seconds is None or age < max_age_seconds:
                 continue
         if status not in {"running", "ok", "aborted", "error"}:
@@ -652,8 +660,11 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa
     if metadata is None:
         return False
     status = str(metadata.get("status", ""))
+    lease_state = _run_lease_state(path)
+    if _lease_blocks_removal(path, lease_state):
+        return False
     if status == "running":
-        if _run_lease_state(path) != "inactive":
+        if lease_state != "inactive":
             return False
         if policy.max_age_seconds is None:
             return False
@@ -675,6 +686,18 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa
     )
 
 
+def _lease_blocks_removal(path: Path, lease_state: str | None = None) -> bool:
+    """Return whether a bundle's lease proves it must be retained."""
+
+    state = _run_lease_state(path) if lease_state is None else lease_state
+    if state == "active":
+        return True
+    if state != "unknown":
+        return False
+    lease_path = path / _RUN_LEASE_NAME
+    return lease_path.exists() or lease_path.is_symlink()
+
+
 def _write_run_index(
     runs_root: Path,
     bundles: list[dict[str, Any]],
@@ -687,15 +710,19 @@ def _write_run_index(
     if current_run_root is not None and current_run_root.exists():
         current_resolved = _safe_resolved_path(current_run_root)
         if not any(_safe_resolved_path(Path(bundle["path"])) == current_resolved for bundle in indexed):
+            metadata = _read_bundle_metadata(current_run_root) or {}
+            started_at = _timestamp_to_epoch(metadata.get("started_at"))
+            if started_at is None:
+                started_at = time.time() if now is None else now
             indexed.append(
                 {
                     "path": current_run_root,
                     "run_id": current_run_root.name,
-                    "status": "running",
-                    "started_at": time.time() if now is None else now,
+                    "status": str(metadata.get("status", "running")),
+                    "started_at": started_at,
                     "size": 0,
                     "size_known": False,
-                    "preserve": False,
+                    "preserve": bool(metadata.get("preserve")),
                 }
             )
     indexed.sort(key=lambda bundle: (float(bundle.get("started_at", 0)), str(bundle["path"])))

tests/test_app_run_metadata.py

@@ -5,7 +5,10 @@
 import json
 import logging
 import os
+import subprocess
+import sys
 import tempfile
+import time
 import unittest
 from contextlib import redirect_stderr
 from dataclasses import replace
@@ -16,8 +19,9 @@
 import base_cli
 import base_cli._lifecycle as lifecycle_module
 import base_cli.app as app_module
+from base_cli import RetentionPolicy
 from base_cli._lifecycle import RunRecorder
-from base_cli._runtime import runtime_layout
+from base_cli._runtime import prune_run_bundles, runtime_layout
 
 
 def _run(app: base_cli.App, home: Path, args: list[str] | None = None) -> tuple[int, str]:
@@ -78,6 +82,143 @@ def _assert_terminal_metadata(
 
 @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
 class AppRunMetadataTests(unittest.TestCase):
+    def test_terminal_run_lease_survives_a_concurrent_invocation_during_cleanup(self) -> None:
+        import click
+
+        child_program = "\n".join(
+            (
+                "import json, sys, time",
+                "from pathlib import Path",
+                "import click, base_cli",
+                "from base_cli import RetentionPolicy",
+                "mode, app_name, cache_text, ready_text, release_text = sys.argv[1:]",
+                "cache, ready, release = Path(cache_text), Path(ready_text), Path(release_text)",
+                "profile = base_cli.CliProfile.generic(cache_root=cache)",
+                "app = base_cli.App(name=app_name, profile=profile, retention=RetentionPolicy(max_bundles=1))",
+                "def block_cleanup(ctx):",
+                "    metadata = json.loads((ctx.run_root / 'run.json').read_text(encoding='utf-8'))",
+                "    (ctx.run_root / 'cleanup-marker').write_text('held', encoding='utf-8')",
+                "    ready.write_text(json.dumps({'run_root': str(ctx.run_root), 'status': metadata['status']}), encoding='utf-8')",
+                "    while not release.exists(): time.sleep(0.01)",
+                "if mode == 'native':",
+                "    @app.command()",
+                "    def main(ctx: base_cli.Context): ctx.on_cleanup(lambda: block_cleanup(ctx))",
+                "    target = app",
+                "else:",
+                "    @click.command(name=app_name)",
+                "    def command(): base_cli.get_current_context().on_cleanup(lambda: block_cleanup(base_cli.get_current_context()))",
+                "    target = app.attach(command)",
+                "base_cli.run_app(target, [])",
+            )
+        )
+
+        for mode in ("native", "attached"):
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmpdir:
+                root = Path(tmpdir)
+                cache = root / "cache"
+                home = root / "home"
+                home.mkdir()
+                ready = root / "ready.json"
+                release = root / "release"
+                app_name = f"terminal-lease-{mode}"
+                child_env = {
+                    key: value for key, value in os.environ.items() if not key.startswith(("COV_CORE_", "COVERAGE_"))
+                }
+                child_env.update(HOME=str(home), BASE_CLI_CACHE_DIR=str(cache))
+                child = subprocess.Popen(
+                    [sys.executable, "-c", child_program, mode, app_name, str(cache), str(ready), str(release)],
+                    cwd=Path(__file__).resolve().parents[1],
+                    env=child_env,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                try:
+                    deadline = time.monotonic() + 10
+                    while not ready.exists() and child.poll() is None and time.monotonic() < deadline:
+                        time.sleep(0.01)
+                    if not ready.exists():
+                        stdout, stderr = child.communicate(timeout=5)
+                        self.fail(f"cleanup hook did not become ready (exit={child.returncode}): {stdout}\n{stderr}")
+
+                    ready_payload = json.loads(ready.read_text(encoding="utf-8"))
+                    run_root = Path(ready_payload["run_root"])
+                    self.assertEqual(ready_payload["status"], "ok")
+                    self.assertEqual(json.loads((run_root / "run.json").read_text())["status"], "ok")
+                    self.assertTrue((run_root / "cleanup-marker").is_file())
+
+                    profile = base_cli.CliProfile.generic(cache_root=cache)
+                    concurrent_app = base_cli.App(
+                        name=app_name,
+                        profile=profile,
+                        retention=RetentionPolicy(max_bundles=1),
+                    )
+                    if mode == "native":
+
+                        @concurrent_app.command()
+                        def concurrent_main(ctx: base_cli.Context) -> None:
+                            del ctx
+
+                        target = concurrent_app
+                    else:
+
+                        @click.command(name=app_name)
+                        def concurrent_command() -> None:
+                            pass
+
+                        target = concurrent_app.attach(concurrent_command)
+
+                    result = base_cli.testing.invoke(target, [], home=home)
+                    self.assertEqual(result.exit_code, 0, result.output)
+                    self.assertTrue(
+                        run_root.is_dir(), "retention deleted a run whose cleanup hook still held its lease"
+                    )
+                    self.assertTrue((run_root / "cleanup-marker").is_file())
+                finally:
+                    release.touch()
+                    try:
+                        child.wait(timeout=10)
+                    except subprocess.TimeoutExpired:
+                        child.kill()
+                        child.wait(timeout=5)
+                    if child.stdout is not None:
+                        child.stdout.close()
+                    if child.stderr is not None:
+                        child.stderr.close()
+
+                index_path = run_root.parent / ".base-cli-run-index.json"
+                index = json.loads(index_path.read_text(encoding="utf-8"))
+                indexed = {bundle["path"]: bundle for bundle in index["bundles"]}
+                self.assertTrue(indexed)
+                self.assertTrue(all(Path(path).is_dir() for path in indexed))
+
+                prune_run_bundles(
+                    run_root.parent,
+                    policy=RetentionPolicy(max_age_seconds=60),
+                    logger=logging.getLogger(__name__),
+                    now=time.time() + 3_600,
+                )
+                self.assertFalse(run_root.exists(), "eligible terminal run was not pruned after its lease was released")
+
+    def test_terminal_run_is_indexed_while_its_lease_is_held(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            home = Path(tmpdir)
+            app = base_cli.App(name="terminal-index")
+
+            @app.command()
+            def main(ctx: base_cli.Context) -> None:
+                del ctx
+
+            status, stderr = _run(app, home)
+            self.assertEqual(status, 0, stderr)
+            run_path, metadata = _load_only_metadata(self, home)
+            index_path = run_path.parent.parent / ".base-cli-run-index.json"
+            index = json.loads(index_path.read_text(encoding="utf-8"))
+            indexed = {bundle["path"]: bundle for bundle in index["bundles"]}
+            run_root = str(run_path.parent.resolve())
+            self.assertIn(run_root, indexed)
+            self.assertEqual(indexed[run_root]["status"], metadata["status"])
+
     def test_normal_returns_finalize_core_owned_metadata(self) -> None:
         cases = (
             ("none", None, 0, "ok", "success"),

tests/test_run_bundle_retention.py

@@ -31,6 +31,7 @@ def _bundle(
     path = root / name
     (path / "logs").mkdir(parents=True)
     (path / "logs" / "primary.log").write_bytes(b"x" * size)
+    (path / ".base-cli-run-lease").write_bytes(b"0")
     write_private_json(
         path / "run.json",
         {
@@ -230,6 +231,64 @@ def test_live_running_bundle_lease_survives_from_another_process(self) -> None:
                     child.kill()
                     child.wait(timeout=5)
 
+    def test_live_terminal_bundle_lease_survives_from_another_process(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir) / "runs"
+            root.mkdir()
+            live = _bundle(root, "live-terminal", status="ok", started_at="2020-01-01T00:00:00Z")
+            ready = live / "ready"
+            release = live / "release"
+            child = subprocess.Popen(
+                [
+                    sys.executable,
+                    "-c",
+                    "\n".join(
+                        (
+                            "import sys, time",
+                            "from pathlib import Path",
+                            "from base_cli._runtime import acquire_run_lease, close_run_lease",
+                            "run_root, ready_path, release_path = map(Path, sys.argv[1:])",
+                            "lease = acquire_run_lease(run_root)",
+                            "ready_path.touch()",
+                            "while not release_path.exists(): time.sleep(0.01)",
+                            "close_run_lease(lease)",
+                        )
+                    ),
+                    str(live),
+                    str(ready),
+                    str(release),
+                ],
+                env={key: value for key, value in os.environ.items() if not key.startswith(("COV_CORE_", "COVERAGE_"))},
+                stdin=subprocess.DEVNULL,
+            )
+            try:
+                deadline = time.monotonic() + 5
+                while not ready.exists() and time.monotonic() < deadline:
+                    time.sleep(0.01)
+                self.assertTrue(ready.exists(), "lease holder did not start")
+                prune_run_bundles(
+                    root,
+                    policy=RetentionPolicy(max_age_seconds=60),
+                    logger=logging.getLogger(__name__),
+                    now=1_600_000_000,
+                )
+                self.assertTrue(live.exists())
+            finally:
+                release.touch()
+                try:
+                    child.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    child.kill()
+                    child.wait(timeout=5)
+
+            prune_run_bundles(
+                root,
+                policy=RetentionPolicy(max_age_seconds=60),
+                logger=logging.getLogger(__name__),
+                now=1_600_000_000,
+            )
+            self.assertFalse(live.exists(), "finished bundles remain eligible after the lease is released")
+
     def test_aborted_bundle_is_indexed_as_terminal(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             root = Path(tmpdir) / "runs"