feat(checkpoint): rotate managed checkpoint banks by default

benchmark/edgebench/README.md

@@ -429,6 +429,35 @@ resume continuity, capability use, evaluator completion and integrity need
 runtime/post-run qualification before any score is countable. Raw trial outputs
 and credentials belong outside the public repository.
 
+## Default research checkpoint retention
+
+**Default change:** new controller-enabled native runs automatically keep only
+the newest complete point plus explicit pins. After publishing a successful
+capture reply, a separate storage worker applies the policy saved at
+`checkpoints/.checkpoint-retention/policy.json`. Pass
+`--checkpoint-retention-policy PRIVATE_POLICY_JSON` to override that policy, or
+`--no-checkpoint-retention` to keep every point without starting storage
+maintenance. These options are mutually exclusive. Capture itself remains
+explicitly enabled through its existing declaration and controller; runs without
+that path do not start retention. A storage-maintenance failure is reported separately in
+`checkpoints/retention-error.json`; it does not turn a successful capture into
+a failure or cancel source research. Inspect that error before assuming space
+was reclaimed. Successful maintenance writes `retention-last.json` and the
+bank's durable deletion receipts.
+
+Use the [checkpoint retention policy and preview](../../docs/reference/research-checkpoint.md#default-bank-rotation)
+to choose uniform or tiered history, protect recent publications and pin every
+selected continuation/dependency. The policy controls the outer inert artifact
+bank only; native submissions, archives, counters and cooldown keep their owners.
+Use separate policy files for separate lineages. Do not import a donor's pins
+into a different receiver's new bank.
+
+For frozen running donors, use the documented standalone storage watcher rather
+than editing or restarting the research runtime. Stop that watcher to disable
+its rotation; use `--no-checkpoint-retention` for new runs. Removing the custom
+policy option reselects the default. The existing CLI/Host
+qualification remains separate from frontend delivery and scientific utility.
+
 ## Default replan cadence
 
 New `heartbeat-resume` and `heartbeat-explore` trials replan after **6 settled

benchmark/edgebench/checkpoint_controller.py

@@ -273,6 +273,17 @@ def _bank_import_lock(target: Path, timeout: float):
 def restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: str,
             timeout: float = 30, phase_deadline_file: Path | None = None, recover: bool = False,
             phase_extension_seconds: int = 0) -> dict:
+    from loopx.hosts.checkpoint_retention import checkpoint_bank_use
+    started = monotonic()
+    with checkpoint_bank_use(root):
+        return _restore(controller_file, root=root, expected_sha256=expected_sha256, run_id=run_id,
+                        timeout=timeout - (monotonic() - started), phase_deadline_file=phase_deadline_file,
+                        recover=recover, phase_extension_seconds=phase_extension_seconds)
+
+
+def _restore(controller_file: Path, *, root: Path, expected_sha256: str, run_id: str,
+             timeout: float, phase_deadline_file: Path | None, recover: bool,
+             phase_extension_seconds: int) -> dict:
     """Import verified inert bytes, then ask fresh native authority to admit them.
 
     An explicitly captured Host clock preserves its remaining phase duration.

benchmark/edgebench/checkpoint_supervisor.py

@@ -16,13 +16,25 @@
 
 
 class CheckpointSupervisor:
-    def __init__(self, controller_file: Path, destination: Path):
+    def __init__(self, controller_file: Path, destination: Path, retention_policy: Path | None = None,
+                 *, retention_enabled: bool = True):
         _controller(controller_file)  # Fail before native registration.
+        if not retention_enabled and retention_policy is not None:
+            raise ValueError("A retention policy cannot be combined with disabled retention")
         self.controller_file, self.destination = controller_file, destination
         self.token = None
         self.stop = threading.Event()
         self.thread = None
         self.failure = None
+        self.retention_enabled = retention_enabled
+        self.retention_policy = retention_policy
+        self.maintenance = None
+        self.maintenance_due = threading.Event()
+        if retention_enabled:
+            from loopx.control_plane.effect_runtime import effect_runtime_result
+            self.initial_retention_policy = effect_runtime_result("runtime.checkpoint_artifacts", {
+                "action": "retention_plan", "input": {"policy": json.loads(retention_policy.read_bytes())
+                    if retention_policy is not None else None, "points": [], "now": time.time()}})["policy"]
 
     def bind(self, environment):
         if self.token is not None or not (environment or {}).get("SFORGE_TOKEN"):
@@ -33,6 +45,16 @@ def start(self, container):
         if self.thread is not None or self.token is None:
             raise ValueError("Checkpoint supervisor requires one bound worker execution")
         self.destination.mkdir(mode=0o700, parents=True, exist_ok=False)
+        if self.retention_enabled:
+            from loopx.hosts.checkpoint_retention import _state
+            state = _state(self.destination.resolve())
+            if self.retention_policy is None:
+                from loopx.file_lock import write_json_atomic
+                self.retention_policy = state / "policy.json"
+                write_json_atomic(self.retention_policy, self.initial_retention_policy,
+                                  mode=0o600, fsync_directory=True)
+            self.maintenance = threading.Thread(target=self._maintain, daemon=True)
+            self.maintenance.start()
         self.container = container
         self.thread = threading.Thread(target=self._run, args=(container,), daemon=True)
         self.thread.start()
@@ -43,6 +65,10 @@ def close(self):
             self.thread.join(60)
             if self.thread.is_alive():
                 raise RuntimeError("Checkpoint writer barrier is still active; preserve the worker for recovery")
+        if self.maintenance is not None:
+            # Storage is a separate worker, never an uncertain writer barrier.
+            # A process exit during unlink leaves its original durable receipt.
+            self.maintenance.join(5)
         if self.failure is not None:
             # A missed optional point does not cancel otherwise useful source
             # work. An uncertain/unreleased writer barrier still needs recovery.
@@ -67,6 +93,27 @@ def _reply(self, container, directory, nonce, response):
         if result.exit_code:
             raise RuntimeError("Checkpoint reply could not be published atomically")
 
+    def _maintain(self):
+        while True:
+            self.maintenance_due.wait(1)
+            if self.maintenance_due.is_set():
+                self.maintenance_due.clear()
+                try:
+                    from loopx.hosts.checkpoint_retention import rotate_bank
+                    from loopx.file_lock import write_json_atomic
+                    receipt = rotate_bank(self.destination.resolve(),
+                        json.loads(self.retention_policy.read_bytes()), execute=True)
+                    write_json_atomic(self.destination / "retention-last.json", receipt,
+                                      mode=0o600, fsync_directory=True)
+                    (self.destination / "retention-error.json").unlink(missing_ok=True)
+                except Exception as error:
+                    from loopx.file_lock import write_json_atomic
+                    write_json_atomic(self.destination / "retention-error.json", {
+                        "error_kind": type(error).__name__}, mode=0o600, fsync_directory=True)
+            if (self.stop.is_set() and self.thread is not None and not self.thread.is_alive()
+                    and not self.maintenance_due.is_set()):
+                return
+
     def _run(self, container):
         from docker.errors import NotFound
 
@@ -126,6 +173,8 @@ def _run(self, container):
                         diagnostic.chmod(0o600)
                     (self.destination / (nonce + ".json")).write_text(json.dumps(response))
                     self._reply(container, mailbox, nonce, response)
+                    if response["ok"] and self.retention_enabled:
+                        self.maintenance_due.set()
         except Exception as error:
             self.failure = error
             (self.destination / "supervisor-failure.json").write_text(json.dumps({"error_kind": type(error).__name__}))

benchmark/edgebench/run.py

@@ -133,6 +133,11 @@ def main(argv=None):
     parser.add_argument("--restore-checkpoint", type=Path,
                         help="Explicit complete checkpoint to continue under fresh native/Goal identities")
     parser.add_argument("--restore-sha256")
+    retention = parser.add_mutually_exclusive_group()
+    retention.add_argument("--checkpoint-retention-policy", type=Path,
+                           help="Override default latest-only rotation in controller-enabled checkpoint banks")
+    retention.add_argument("--no-checkpoint-retention", action="store_true",
+                           help="Keep every captured point; disable default checkpoint rotation")
     parser.add_argument("--restore-config", type=Path)
     parser.add_argument("--restore-goal")
     parser.add_argument("--restore-agent")
@@ -183,6 +188,8 @@ def main(argv=None):
     handoff = checkpoint_request is not None and declaration.get("controller_handoff", False)
     if bool(args.checkpoint_controller_file) != handoff or (handoff and args.feedback != "native"):
         parser.error("controller file and full controller_handoff must be enabled together with native feedback")
+    if (args.checkpoint_retention_policy is not None or args.no_checkpoint_retention) and not handoff:
+        parser.error("Checkpoint retention requires the explicit native controller capture path")
     restore_values = (args.restore_sha256, args.restore_config, args.restore_goal,
                       args.restore_agent, args.restore_source_todo)
     if bool(args.restore_checkpoint) != all(restore_values) or (not args.restore_checkpoint and any(restore_values)):
@@ -313,7 +320,9 @@ def main(argv=None):
     checkpoint = None
     if args.checkpoint_controller_file is not None:
         from benchmark.edgebench.checkpoint_supervisor import CheckpointSupervisor
-        checkpoint = CheckpointSupervisor(args.checkpoint_controller_file, trial / "checkpoints")
+        checkpoint = CheckpointSupervisor(args.checkpoint_controller_file, trial / "checkpoints",
+                                          retention_policy=args.checkpoint_retention_policy,
+                                          retention_enabled=not args.no_checkpoint_retention)
     backend = RecordingDockerBackend(log_dir=trial / "collected", logger=logger,
                                      oauth_proxy=bool(args.api_proxy_url or api_base_url),
                                      blind_api_endpoint=blind_endpoint, feedback=feedback,

benchmark/runtime/sforge_warm_start.py

@@ -47,6 +47,11 @@ class WarmStart:
     fresh_replan: bool = False
 
     def verify(self, workspace: str) -> dict:
+        from loopx.hosts.checkpoint_retention import checkpoint_bank_use
+        with checkpoint_bank_use(self.root):
+            return self._verify(workspace)
+
+    def _verify(self, workspace: str) -> dict:
         if type(self.fresh_replan) is not bool:
             raise ValueError("fresh_replan requires an explicit boolean context treatment")
         if (type(self.normal_cadence) is not int or not 1 <= self.normal_cadence <= 6
@@ -88,7 +93,12 @@ def verify(self, workspace: str) -> dict:
         return metadata
 
     async def stage(self, environment, workspace: str) -> list[str]:
-        metadata = self.verify(workspace)
+        from loopx.hosts.checkpoint_retention import checkpoint_bank_use
+        with checkpoint_bank_use(self.root):
+            return await self._stage(environment, workspace)
+
+    async def _stage(self, environment, workspace: str) -> list[str]:
+        metadata = self._verify(workspace)
         research = _research_components(self.root, metadata, Path(workspace), self.research_component_ids)
         # Install tools first without creating the native home. Stage inert
         # participants outside the workspace so restoring its bytes cannot

benchmark/tests/test_checkpoint_controller.py

@@ -645,6 +645,11 @@ def grade(url, token, score):
                 phase_source.write_text(f"{int(time.time()) + 7200}\n")
             point = transport.capture(files[0], token=token, capture_id="c" * 32, destination=exported,
                                       phase_deadline_file=phase_source if phase_budget else None)
+            if phase_budget and not lose_response:
+                # The real Docker/native restore also traverses a managed-bank
+                # reader lock. Other parameter rows retain off-state coverage.
+                from loopx.hosts.checkpoint_retention import _state
+                _state(tmp_path)
             # Admission clock only: HTTP, transport/fsync, Docker and native
             # grading remain real. Offline/import IO must not spend cooldown.
             now = [time.time() + 7200]

benchmark/tests/test_checkpoint_worker.py

@@ -112,7 +112,8 @@ def test_archive_keeps_private_directory_modes_without_special_bits(writer, tmp_
     assert (copied / "child").stat().st_mode & 0o7777 == 0o700
 
 
-@pytest.mark.parametrize("fault", [None, "source_moved", "expired", "read_race", "late_native"])
+@pytest.mark.parametrize("fault", [None, "source_moved", "expired", "read_race", "late_native",
+    "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"])
 def test_supervisor_captures_only_the_prepared_worker_boundary(controllers, writer, tmp_path, monkeypatch, fault):
     from loopx.control_plane.effect_runtime import effect_runtime_result
     from benchmark.edgebench.checkpoint_worker import _archive
@@ -157,7 +158,33 @@ def late_capture(*args, **kwargs):
             time.sleep(max(0, request["deadline_epoch"] + .1 - time.time()))
             return result
         monkeypatch.setattr(checkpoint_worker, "capture", late_capture)
-    supervisor = CheckpointSupervisor(files[0], tmp_path / "points")
+    policy_file = None
+    if fault in {"retention", "retention_default", "retention_off", "retention_pin", "retention_failure"}:
+        from benchmark.edgebench import checkpoint_supervisor
+        original_capture = checkpoint_supervisor.capture_worker
+        if fault in {"retention", "retention_failure"}:
+            policy_file = tmp_path / "retention.json"
+            policy_file.write_text(json.dumps(dict(strategy="uniform", keep_latest=1, keep_first=False,
+                history_slots=0, minimum_age_seconds=0, pins=[], protect_since=None)))
+        def with_old_point(*args, **kwargs):
+            old = kwargs["destination"].parent / "old"
+            effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture",
+                "components": [{"id": "code", "source_path": str(prepared)}], "destination": str(old),
+                "point_metadata": {}, "expected_source_sha256": probe["source_sha256"]})
+            os.utime(old / "checkpoint.json", (1000, 1000))
+            if fault == "retention_pin":
+                config = old.parent / ".checkpoint-retention/policy.json"
+                policy = json.loads(config.read_bytes())
+                policy["pins"] = [dict(id="old", sha256=json.loads((old / "checkpoint.json").read_bytes())["sha256"])]
+                config.write_text(json.dumps(policy))
+            if fault == "retention":
+                (old.parent / "retention-error.json").write_text('{"error_kind":"old failure"}')
+            if fault == "retention_failure":
+                policy_file.write_text("{}")
+            return original_capture(*args, **kwargs)
+        monkeypatch.setattr(checkpoint_supervisor, "capture_worker", with_old_point)
+    supervisor = CheckpointSupervisor(files[0], tmp_path / "points", retention_policy=policy_file,
+                                      retention_enabled=fault != "retention_off")
     supervisor.bind({"SFORGE_TOKEN": token})
     supervisor.start(container)
     try:
@@ -166,15 +193,31 @@ def late_capture(*args, **kwargs):
     finally:
         supervisor.close()
     response = json.loads((tmp_path / "points" / (nonce + ".json")).read_bytes())
-    assert response["ok"] is (fault in {None, "read_race"})
+    assert response["ok"] is (fault in {None, "read_race", "retention", "retention_default",
+                                      "retention_off", "retention_pin", "retention_failure"})
     unit = tmp_path / "points" / nonce
-    if fault in {None, "read_race"}:
+    if fault in {None, "read_race", "retention", "retention_default", "retention_off", "retention_pin", "retention_failure"}:
         manifest = json.loads((unit / "checkpoint.json").read_bytes())
         metadata = manifest["point_metadata"]
         assert metadata["synthetic_boundary"] is True
         assert metadata["worker_file_snapshot"]["prepared_source_verified"] is True
         assert (unit / "components/code").read_text() == "retained"
         assert response["capture"]["research_point_qualified"] is False
+        if fault in {"retention", "retention_default"}:
+            assert not (tmp_path / "points/old").exists()
+            receipt = json.loads((tmp_path / "points/retention-last.json").read_bytes())
+            assert [p["id"] for p in receipt["deleted"]] == ["old"]
+            assert not (tmp_path / "points/retention-error.json").exists()
+        elif fault == "retention_pin":
+            receipt = json.loads((tmp_path / "points/retention-last.json").read_bytes())
+            assert {p["id"] for p in receipt["retained"]} == {"old", nonce} and receipt["deleted"] == []
+        elif fault == "retention_off":
+            assert (tmp_path / "points/old/checkpoint.json").exists()
+            assert not (tmp_path / "points/.checkpoint-retention").exists()
+            assert supervisor.maintenance is None
+        elif fault == "retention_failure":
+            assert (tmp_path / "points/old/checkpoint.json").exists()
+            assert (tmp_path / "points/retention-error.json").exists()
     else:
         assert not unit.exists()
         if fault != "late_native":

benchmark/tests/test_sforge_runtime.py

@@ -666,6 +666,9 @@ def test_envelope_rejects_incompatible_sforge_worker(profile):
     ("heartbeat-resume", ["--stride-strategy", "future_policy"]),
     ("official", ["--checkpoint-request", "missing.json"]),
     ("native-goal", ["--checkpoint-request", "missing.json"]),
+    ("heartbeat-explore", ["--checkpoint-retention-policy", "policy.json"]),
+    ("heartbeat-explore", ["--no-checkpoint-retention"]),
+    ("heartbeat-explore", ["--no-checkpoint-retention", "--checkpoint-retention-policy", "policy.json"]),
 ])
 def test_edgebench_rejects_invalid_profile_settings_before_creating_trial(tmp_path, profile, options):
     pytest.importorskip('sforge')