fix: rotate bounded retention size scans

CHANGELOG.md

@@ -13,6 +13,10 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Rotate bounded byte-retention size walks across invocations with a persisted
+  advisory cursor so bundles beyond the first scan budget are eventually seen.
+- 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 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
@@ -896,7 +897,9 @@ Recovery work is bounded on the foreground command path. Count- and age-only
 policies inspect metadata without recursively sizing bundle contents. A byte
 policy performs at most 512 recursive size walks and removes at most 256
 bundles per pass; any remaining policy debt is retained safely and reported as
-a warning for a later invocation. The diagnostic index records at most 512
+a warning for a later invocation. An atomic advisory cursor rotates the size
+walk across invocations, so repeated passes eventually inspect the full set;
+the cursor never authorizes deletion. The diagnostic index records at most 512
 entries and sets `complete: false` plus `omitted_bundles` when a cache is
 larger, so a stale, corrupt, or missing index is always reconciled from the
 filesystem rather than trusted for deletion.

docs/performance.md

@@ -54,8 +54,9 @@ foreground pass (protected bundles and unreadable entries are retained):
 When a bound prevents a complete reconciliation, base-cli leaves the
 unprocessed bundles intact, writes a partial index with `complete: false`, and
 emits a warning describing the remaining policy debt. A later invocation
-continues from the filesystem; the index is an observation aid, never an
-authorization to delete a path. The retention regression suite covers count,
+continues from the filesystem. An atomic advisory cursor rotates the bounded
+byte-size walk across invocations, including after process restart; the index
+is an observation aid, never an authorization to delete a path. The retention regression suite covers count,
 age, byte limits, deep trees, corrupt metadata/index files, unreadable files,
 concurrent invocations, and live-run lease protection.
 

lib/python/base_cli/_runtime.py

@@ -410,20 +410,20 @@ def prune_run_bundles(
         protected.add(_safe_resolved_path(current_run_root))
     clock = time.time() if now is None else now
 
-    # Filesystem discovery and recursive size accounting are deliberately
-    # outside the lock.  The destructive phase revalidates each candidate
-    # under the lock so another invocation can never turn a live bundle into a
-    # deletion candidate while discovery is in progress.
-    bundles = _discover_run_bundles(
-        runs_root,
-        protected=protected,
-        max_age_seconds=effective.max_age_seconds,
-        now=clock,
-        measure_sizes=effective.max_total_bytes is not None,
-        size_budget=_RETENTION_SIZE_MEASUREMENT_BUDGET,
-    )
     try:
         with _retention_lock(runs_root):
+            # Keep cursor read, size walk, and index update in one critical
+            # section so concurrent pruners cannot overwrite scan progress.
+            size_scan_cursor = _read_size_scan_cursor(runs_root)
+            bundles, size_scan_cursor = _discover_run_bundles(
+                runs_root,
+                protected=protected,
+                max_age_seconds=effective.max_age_seconds,
+                now=clock,
+                measure_sizes=effective.max_total_bytes is not None,
+                size_budget=_RETENTION_SIZE_MEASUREMENT_BUDGET,
+                size_scan_cursor=size_scan_cursor,
+            )
             _apply_bundle_retention(
                 runs_root,
                 bundles,
@@ -439,6 +439,7 @@ def prune_run_bundles(
                 log,
                 current_run_root=current_run_root,
                 now=clock,
+                size_scan_cursor=size_scan_cursor,
             )
     except (OSError, RuntimeError) as exc:
         # Retention is maintenance.  An unavailable lock or a transient
@@ -459,7 +460,7 @@ def refresh_run_bundle_index(
     if not runs_root.exists() or runs_root.is_symlink():
         return
     try:
-        bundles = _discover_run_bundles(
+        bundles, _size_scan_cursor = _discover_run_bundles(
             runs_root,
             protected=set(),
             max_age_seconds=None,
@@ -481,13 +482,14 @@ def _discover_run_bundles(
     now: float,
     measure_sizes: bool,
     size_budget: int,
-) -> list[dict[str, Any]]:
+    size_scan_cursor: str | None = None,
+) -> tuple[list[dict[str, Any]], str | None]:
     bundles: list[dict[str, Any]] = []
-    measured_sizes = 0
+    scan_order: list[dict[str, Any]] = []
     try:
         children = sorted(runs_root.iterdir(), key=lambda path: path.name)
     except OSError:
-        return bundles
+        return bundles, size_scan_cursor
     for child in children:
         if child.name.startswith(".") or child.is_symlink() or not child.is_dir():
             continue
@@ -506,28 +508,17 @@ 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.
+        if _run_lease_state(child) != "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"}:
             continue
         resolved = _safe_resolved_path(child)
-        size = 0
-        size_known = False
-        if measure_sizes and measured_sizes < size_budget:
-            try:
-                size = _bundle_size(child)
-                size_known = True
-                measured_sizes += 1
-            except OSError:
-                # A file that disappears or becomes unreadable remains a
-                # retention candidate for count/age policy, but its byte
-                # contribution is unknown and must be reported below.
-                pass
         retention_metadata = metadata.get("retention")
         preserve = bool(metadata.get("preserve")) or (
             isinstance(retention_metadata, dict) and retention_metadata.get("preserve") is True
@@ -540,14 +531,60 @@ def _discover_run_bundles(
                 "status": status,
                 "started_at": started_at,
                 "age": age,
-                "size": size,
-                "size_known": size_known,
+                "size": 0,
+                "size_known": False,
                 "preserve": preserve,
                 "protected": resolved in protected,
             }
         )
+        if measure_sizes:
+            scan_order.append(bundles[-1])
     bundles.sort(key=lambda bundle: (float(bundle["started_at"]), str(bundle["path"])))
-    return bundles
+    if measure_sizes and bundles and size_budget > 0:
+        # The run index's cursor affects only which discovered bundles receive
+        # an expensive size walk. It never authorizes deletion; every candidate
+        # is re-read and revalidated before the destructive phase.
+        if size_scan_cursor is not None:
+            start_index = next(
+                (index for index, bundle in enumerate(scan_order) if bundle["path"].name > size_scan_cursor),
+                0,
+            )
+            scan_order = scan_order[start_index:] + scan_order[:start_index]
+        attempted = 0
+        for bundle in scan_order:
+            if attempted >= size_budget:
+                break
+            attempted += 1
+            path = bundle["path"]
+            size_scan_cursor = path.name
+            try:
+                bundle["size"] = _bundle_size(path)
+                bundle["size_known"] = True
+            except OSError:
+                # A file that disappears or becomes unreadable remains a
+                # retention candidate for count/age policy, but its byte
+                # contribution is unknown and reported below. Advancing the
+                # cursor prevents one unreadable entry from starving others.
+                pass
+    return bundles, size_scan_cursor
+
+
+def _read_size_scan_cursor(runs_root: Path) -> str | None:
+    """Read the advisory byte-scan cursor; never use it to select deletions."""
+
+    index_path = runs_root / _RUN_INDEX_NAME
+    try:
+        if index_path.is_symlink() or not index_path.is_file() or index_path.stat().st_size > 1_048_576:
+            return None
+        payload = json.loads(index_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+    cursor = payload.get("byte_scan_cursor")
+    if not isinstance(cursor, str) or not cursor or len(cursor) > 1024 or "/" in cursor or "\\" in cursor:
+        return None
+    return cursor
 
 
 def _apply_bundle_retention(
@@ -652,9 +689,9 @@ def _bundle_is_still_removable(path: Path, *, policy: RetentionPolicy, now: floa
     if metadata is None:
         return False
     status = str(metadata.get("status", ""))
+    if _run_lease_state(path) != "inactive":
+        return False
     if status == "running":
-        if _run_lease_state(path) != "inactive":
-            return False
         if policy.max_age_seconds is None:
             return False
         started_at = _timestamp_to_epoch(metadata.get("started_at"))
@@ -682,6 +719,7 @@ def _write_run_index(
     *,
     current_run_root: Path | None = None,
     now: float | None = None,
+    size_scan_cursor: str | None = None,
 ) -> None:
     indexed = list(bundles)
     if current_run_root is not None and current_run_root.exists():
@@ -707,6 +745,7 @@ def _write_run_index(
         "version": 1,
         "complete": omitted_bundles == 0,
         "omitted_bundles": omitted_bundles,
+        "byte_scan_cursor": size_scan_cursor if size_scan_cursor is not None else _read_size_scan_cursor(runs_root),
         "bundles": [
             {
                 "path": str(bundle["path"]),

tests/test_adversarial_regressions.py

@@ -87,7 +87,7 @@ def _write_log_worker(path_text: str, seed: int, count: int) -> None:
 def _prune_worker(runs_root_text: str) -> None:
     prune_run_bundles(
         Path(runs_root_text),
-        policy=base_cli.RetentionPolicy(max_bundles=2),
+        policy=base_cli.RetentionPolicy(max_bundles=2, max_total_bytes=2),
     )
 
 
@@ -250,6 +250,9 @@ def test_run_bundle_retention_remains_bounded_across_processes(self) -> None:
                         "preserve": False,
                     },
                 )
+                # Retention now fails closed if a bundle has no lease record,
+                # because missing liveness cannot prove that it is inactive.
+                (bundle / ".base-cli-run-lease").write_bytes(b"0")
 
             _run_processes(_prune_worker, [(str(runs_root),) for _seed in SEEDS])
 

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,118 @@ 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()
+
+                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_normal_returns_finalize_core_owned_metadata(self) -> None:
         cases = (
             ("none", None, 0, "ok", "success"),