fix: rotate bounded retention size scans (#367)

## Summary
- Persist an advisory byte-scan cursor in the atomic diagnostic run
index.
- Rotate bounded size walks across invocations and process restarts,
advancing past unreadable entries.
- Keep deletion based on live filesystem discovery and metadata/liveness
revalidation, never the cursor/index.
- Cover 1,025 bundles over three fresh passes with protected early
bundles and an oversized tail bundle.

Closes #355

## Dependency
This PR is intentionally stacked on #366 (issue #354) so retention lease
hardening merges first.

## Validation
- `uv run --extra dev --extra typer --extra quality python -m pytest
tests/test_run_bundle_retention.py -q`
- `uv run --extra dev --extra typer --extra quality python -m mypy
--strict lib/python/base_cli/_runtime.py`
- Ruff check/format and `git diff --check`

Author: codeforester

CHANGELOG.md

@@ -23,6 +23,8 @@ 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.
 - Resolve lifecycle values through the active Typer/Click context for attached
   commands, including renamed options, defaults, and environment variables.
 - Reject non-finite numbers in JSON and NDJSON output so emitted records remain
@@ -47,7 +49,6 @@ and versions are tracked in the repo-root `VERSION` file.
 - Detect JSON capture without running Click callbacks, callable defaults, type
   converters, or close hooks a second time; respect option-value arity so a
   payload equal to `--json` remains human output.
-
 - 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

@@ -906,7 +906,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

@@ -105,8 +105,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
@@ -460,7 +461,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,
@@ -482,13 +483,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
@@ -524,18 +526,6 @@ def _discover_run_bundles(
         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
@@ -548,14 +538,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(
@@ -705,6 +741,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():
@@ -734,6 +771,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_platform_edge_paths.py

@@ -150,13 +150,14 @@ def test_retention_scan_and_apply_are_portable_without_recursive_sizes(self) ->
             for index in range(3):
                 bundle = root / f"run-{index}"
                 bundle.mkdir()
+                (bundle / ".base-cli-run-lease").write_bytes(b"0")
                 (bundle / "run.json").write_text(
                     f'{{"run_id": "run-{index}", "status": "ok", '
                     '"started_at": "2020-01-01T00:00:00Z", "preserve": false}',
                     encoding="utf-8",
                 )
             with mock.patch.object(runtime, "_bundle_size", side_effect=AssertionError("unexpected size walk")):
-                bundles = runtime._discover_run_bundles(  # pylint: disable=protected-access
+                bundles, _size_scan_cursor = runtime._discover_run_bundles(  # pylint: disable=protected-access
                     root,
                     protected=set(),
                     max_age_seconds=None,

tests/test_run_bundle_retention.py

@@ -94,6 +94,40 @@ def test_byte_retention_bounds_recursive_size_work(self) -> None:
             self.assertLessEqual(bundle_size.call_count, 512)
             self.assertTrue(any("size walk(s)" in str(call) for call in logger.warning.call_args_list))
 
+    def test_byte_retention_cursor_advances_across_restarted_bounded_passes(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir) / "runs"
+            root.mkdir()
+            expected_names = {f"run-{index:05d}" for index in range(1_025)}
+            for index in range(1_025):
+                _bundle(
+                    root,
+                    f"run-{index:05d}",
+                    preserve=index < 1_024,
+                    size=1_048_576 if index == 1_024 else 1,
+                )
+
+            scanned: set[str] = set()
+            policy = RetentionPolicy(max_total_bytes=400_000)
+            for pass_number in range(1, 4):
+                with mock.patch.object(runtime, "_bundle_size", wraps=runtime._bundle_size) as bundle_size:
+                    prune_run_bundles(
+                        root,
+                        policy=policy,
+                        logger=logging.getLogger(__name__),
+                        now=1_600_000_000,
+                    )
+                scanned.update(Path(call.args[0]).name for call in bundle_size.call_args_list)
+                self.assertLessEqual(bundle_size.call_count, 512)
+                index = json.loads((root / ".base-cli-run-index.json").read_text(encoding="utf-8"))
+                self.assertIn("byte_scan_cursor", index)
+                if pass_number < 3:
+                    self.assertTrue((root / "run-01024").exists())
+
+            self.assertTrue(expected_names <= scanned)
+            self.assertFalse((root / "run-01024").exists())
+            self.assertTrue(all((root / name).exists() for name in expected_names if name != "run-01024"))
+
     def test_corrupt_index_is_reconciled_without_trusting_paths(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             root = Path(tmpdir) / "runs"