diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md
index c418c6e875..e1fcb4d356 100644
--- a/benchmark/edgebench/README.md
+++ b/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
diff --git a/benchmark/edgebench/checkpoint_controller.py b/benchmark/edgebench/checkpoint_controller.py
index d9aa99d1ea..6c2708a5ea 100644
--- a/benchmark/edgebench/checkpoint_controller.py
+++ b/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.
diff --git a/benchmark/edgebench/checkpoint_supervisor.py b/benchmark/edgebench/checkpoint_supervisor.py
index 26356af1cd..d770c1b760 100644
--- a/benchmark/edgebench/checkpoint_supervisor.py
+++ b/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__}))
diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py
index e16e2a0e9e..f8ef59bf85 100644
--- a/benchmark/edgebench/run.py
+++ b/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,
diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py
index 53af2c87af..04bafb9463 100644
--- a/benchmark/runtime/sforge_warm_start.py
+++ b/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
diff --git a/benchmark/tests/test_checkpoint_controller.py b/benchmark/tests/test_checkpoint_controller.py
index 4dcccb933a..a5e3f239d0 100644
--- a/benchmark/tests/test_checkpoint_controller.py
+++ b/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]
diff --git a/benchmark/tests/test_checkpoint_worker.py b/benchmark/tests/test_checkpoint_worker.py
index d28bc16558..b575d0f342 100644
--- a/benchmark/tests/test_checkpoint_worker.py
+++ b/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":
diff --git a/benchmark/tests/test_sforge_runtime.py b/benchmark/tests/test_sforge_runtime.py
index 74666c9bbc..e0cce406b9 100644
--- a/benchmark/tests/test_sforge_runtime.py
+++ b/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')
diff --git a/docs/reference/research-checkpoint.md b/docs/reference/research-checkpoint.md
index 04a68425aa..ddd5450f0a 100644
--- a/docs/reference/research-checkpoint.md
+++ b/docs/reference/research-checkpoint.md
@@ -132,10 +132,125 @@ admission and inspect the next accepted replan. A useful readback cites retained
 evidence, explains the next complete experiment and later checks its outcome.
 Restoring files or accepting a structured review alone proves no research gain.
 
+## Default bank rotation
+
+**Default change:** new controller-enabled benchmark checkpoint banks rotate
+automatically after a successful capture reply, keeping only the newest complete
+publication and explicitly pinned points. Capture itself still requires its
+existing explicit enablement. Use `--no-checkpoint-retention` to retain every
+point, or `--checkpoint-retention-policy RETENTION_POLICY` to supply a custom
+policy. The default policy is saved in the bank's private
+`.checkpoint-retention/policy.json`; it can be inspected and edited for pins.
+An explicit policy file remains the active configuration when supplied.
+
+Ordinary single-point capture accepts an arbitrary destination and does not
+implicitly sweep its parent. Select a bank explicitly through `rotate` or the
+storage watcher below. Existing running processes are not hot-patched.
+Retention is local storage maintenance through the same artifact owner; it does
+not select scientific results, qualify a point, change feedback opportunities,
+or delete the native evaluator's separate bank. It remains part of this CLI/Host
+stage; a packaged frontend retention editor is not delivered here.
+
+Use one bank per source lineage. The typed default is:
+
+```json
+{
+  "strategy": "uniform",
+  "keep_latest": 1,
+  "keep_first": false,
+  "minimum_age_seconds": 0,
+  "history_slots": 0,
+  "protect_since": null,
+  "pins": []
+}
+```
+
+This compact policy retains only the newest complete publication plus explicit
+pins. It has no origin anchor, sampled history or age grace. To preserve more
+history, enable `keep_first`, increase `keep_latest`, set
+`minimum_age_seconds`, or set positive `history_slots` for equal-time history.
+For nonuniform history, replace `strategy` with `tiered` and `history_slots`
+with `bucket_seconds` (for example, `14400`). That retains the newest
+representative in each age band measured from the newest publication: under
+4 hours, 4–8 hours, 8–16 hours, and so on. Idle time alone does not shift bands.
+Uniform slots include the historical end;
+repeating a policy on an unchanged bank does not progressively thin it.
+Uniform history requires `keep_first=true` to preserve its time anchors.
+`keep_latest` must be at least one. Protection is additive, so pinned or recent
+points can exceed the sampling count. There is no hard byte cap that overrides
+these protections.
+
+Before cleanup, put every selected study point, source still needed by a
+receiver, and referenced dependency in `pins` as an exact
+`{"id":"POINT_DIRECTORY_NAME","sha256":"CAPTURE_DIGEST"}`. References in
+arbitrary plans, sessions or conversations are not automatically discovered.
+For example, a source used by two continuation arms remains pinned until both
+arms no longer need it. The ID is the immediate point directory name and the
+digest is the `sha256` field in that point's `checkpoint.json`. Remove the entry
+to make an older point eligible on the next rotation. Stop the watcher or
+quiesce the capture supervisor before changing pins, preview the new policy,
+then resume maintenance; editing a policy does not recover already deleted data.
+`protect_since` optionally preserves every publication at or after a Unix epoch,
+useful while a chronological selection is unresolved. This is filesystem seal
+time, not scientific elapsed age or source qualification. A missing/changed pin
+holds cleanup, including pending quarantine recovery.
+
+Preview the complete retained/deletion lists, logical byte totals and plan digest.
+Without `--request-file`, the CLI reads the bank's policy file when present,
+otherwise the typed latest-only default. Supplying the file explicitly overrides
+that selection for this operation; it does not rewrite another worker's configuration:
+
+```sh
+loopx --format json research-checkpoint rotate --root CHECKPOINT_BANK
+loopx --format json research-checkpoint rotate --root CHECKPOINT_BANK \
+  --request-file RETENTION_POLICY --expected-plan-sha256 PLAN_DIGEST --execute
+```
+
+Execution requires a private, operator-owned bank on the qualified POSIX
+transport. Preview does not change it. Before any deletion or quarantine recovery,
+the complete retained points must pass member-integrity verification. A corrupt
+newest point holds pruning of older recovery points. This storage RPC allowance
+does not extend a research capture, native deadline or continuation budget.
+Execution excludes supported readers, rechecks the inventory and atomically
+moves only the selected whole point directories into a private quarantine.
+It preserves response/error sidecars, unknown paths, incomplete captures and
+staging directories. Unlinking runs outside the reader lock, with a separate
+single-flight lock preventing concurrent recovery. Compact digest-bound records
+under `.checkpoint-retention/receipts/` distinguish planned, quarantined and
+deleted points. A failed unlink resumes from its original quarantine on the next
+explicit rotation; it never recaptures newer state into an old identity.
+Missing/ambiguous files or lock failures hold the affected operation.
+
+The shipped verify/admit/adopt CLI and native benchmark restore/upload adapters
+use the bank's shared read lock once retention has been activated, so same-point
+arms can read concurrently while rotation waits. Older readers and
+direct Effect clients must use an explicitly pinned point or quiesce before
+cleanup. At first activation also quiesce or pin any reads that already began
+before the marker existed. New supervisor-enabled banks install the marker before
+capture starts. Pinning is still required for references that outlive one read.
+Preview inventory is metadata-only. The execution-time member check still does
+not qualify scientific dependencies, session adoption or research admission.
+
+For an existing donor, run an independently managed storage watcher without
+editing its frozen runtime:
+
+```sh
+loopx --format json research-checkpoint rotate --root CHECKPOINT_BANK \
+  --request-file RETENTION_POLICY --execute --watch --interval-seconds 300
+```
+
+It reacts to changed local publication/policy generations and due grace expiry. It makes no
+model or Goal calls. Errors stop the watcher with a failure result for operator
+reconciliation. Stop this process to disable automatic rotation. For a new
+controller-enabled benchmark run, pass `--no-checkpoint-retention`; removing a
+custom policy option selects the default and does not disable rotation.
+Previously deleted bytes are not
+restored by disabling; recover them only from an independently retained copy.
+
 ## Disable and qualification
 
-Omit `--execute` or stop issuing the explicit operations to disable this entry;
-there is no automatic activation switch. Published points stay inert. Change
+Omit `--execute` or stop issuing the explicit operations to disable capture and
+adoption; stop a retention watcher separately. Published points stay inert. Change
 adopted Todo notes through their normal owner; dispose of an isolated receiver
 through its normal lifecycle. Do not delete or rewrite source history to roll
 back a continuation.
diff --git a/loopx/cli_commands/research_checkpoint.py b/loopx/cli_commands/research_checkpoint.py
index 87c40d5dc7..e5305dd0ae 100644
--- a/loopx/cli_commands/research_checkpoint.py
+++ b/loopx/cli_commands/research_checkpoint.py
@@ -5,6 +5,8 @@
 import json
 import os
 from pathlib import Path
+import sys
+import time
 
 
 def register_research_checkpoint(subparsers, add_subcommand_format) -> None:
@@ -13,13 +15,16 @@ def register_research_checkpoint(subparsers, add_subcommand_format) -> None:
     add_subcommand_format(parser)
     # Local CLI operations compose existing artifact/session/continuation APIs;
     # they are not a new persisted lifecycle or authority vocabulary.
-    parser.add_argument("operation", choices=("capture", "verify", "admit-session", "adopt-context"))
+    parser.add_argument("operation", choices=("capture", "verify", "admit-session", "adopt-context", "rotate"))
     parser.add_argument("--goal-id")
     parser.add_argument("--agent-id")
     parser.add_argument("--project", default=".")
     parser.add_argument("--request-file", type=Path)
     parser.add_argument("--root", type=Path)
     parser.add_argument("--expected-sha256")
+    parser.add_argument("--expected-plan-sha256", help="Rotate only the unchanged previewed retention plan")
+    parser.add_argument("--watch", action="store_true", help="Explicit rotate --execute loop; stop the process to disable")
+    parser.add_argument("--interval-seconds", type=float, default=300)
     parser.add_argument("--turn-instance-id")
     parser.add_argument("--config-path", type=Path)
     from ..control_plane.turn_driver.codex_cli import CODEX_CLI_SANDBOXES
@@ -29,7 +34,7 @@ def register_research_checkpoint(subparsers, add_subcommand_format) -> None:
         parser.add_argument("--" + name)
     parser.add_argument("--lease-version", type=int)
     parser.add_argument("--execute", action="store_true",
-        help="Explicitly perform capture, session admission or context adoption; otherwise preview only.")
+        help="Explicitly perform capture, adoption or retention deletion; otherwise preview only.")
 
 
 def handle_research_checkpoint(args: argparse.Namespace, *, registry_path: Path,
@@ -42,6 +47,39 @@ def required(*names):
         if any(getattr(args, name) is None or getattr(args, name) == "" for name in names):
             raise ValueError("Missing explicit " + ", ".join("--" + name.replace("_", "-") for name in names))
     try:
+        if args.watch and (args.operation != "rotate" or not args.execute or args.expected_plan_sha256):
+            raise ValueError("--watch requires rotate --execute without a frozen preview digest")
+        if args.expected_plan_sha256 and args.operation != "rotate":
+            raise ValueError("--expected-plan-sha256 requires rotate")
+        if args.operation == "rotate":
+            required("root")
+            if not 10 <= args.interval_seconds <= 86400:
+                raise ValueError("Retention interval must be between 10 and 86400 seconds")
+            from ..hosts.checkpoint_retention import bank_generation, retention_policy_bytes, rotate_bank
+            previous = None
+            next_due = None
+            while True:
+                # A supervisor bank publishes its editable default policy.
+                # An unmanaged, explicitly selected bank uses the typed default
+                # without creating any state during preview.
+                raw = retention_policy_bytes(args.root, args.request_file)
+                generation = bank_generation(args.root, raw)
+                if generation != previous or (next_due is not None and time.time() >= next_due):
+                    result = rotate_bank(args.root, json.loads(raw), execute=args.execute,
+                                         expected_plan_sha256=args.expected_plan_sha256)
+                    payload = {"ok": True, **result}
+                    print_payload(payload, output_format(args), lambda value: json.dumps(value, ensure_ascii=False, indent=2))
+                    sys.stdout.flush()
+                    # Remember the post-cleanup bank; do not wake just because
+                    # this operation removed its own selected directories.
+                    after = bank_generation(args.root, raw)
+                    previous = after if {p[0] for p in after[1]} == {p["id"] for p in result["retained"]} else None
+                    next_due = result["grace_expires_at"]
+                if not args.watch:
+                    return 0
+                # No model/Goal/control-plane reads. Only the selected local
+                # bank and explicit policy are revisited at this bounded rate.
+                time.sleep(args.interval_seconds)
         if args.operation == "capture":
             required("request_file", "turn_instance_id", "goal_id", "agent_id", "sandbox")
         else:
@@ -52,8 +90,10 @@ def required(*names):
             required("goal_id", "agent_id", "sandbox", "source_todo_id", "todo_id",
                      "operation_id", "expected_revision", "lease_key", "lease_version")
         if args.operation == "verify":
-            result = effect_runtime_result("runtime.checkpoint_artifacts", {
-                "action": "verify", "root": str(args.root), "expected_sha256": args.expected_sha256}, large_local_snapshot=True)
+            from ..hosts.checkpoint_retention import checkpoint_bank_use
+            with checkpoint_bank_use(args.root):
+                result = effect_runtime_result("runtime.checkpoint_artifacts", {
+                    "action": "verify", "root": str(args.root), "expected_sha256": args.expected_sha256}, large_local_snapshot=True)
         elif not args.execute:
             result = {"dry_run": True, "operation": args.operation, "execution_authority_granted": False,
                 "next_action": "Review the explicit source/destination and use --execute for this operation."}
@@ -71,12 +111,16 @@ def required(*names):
                     turn_instance_id=args.turn_instance_id, timeout=args.timeout_seconds)
             elif args.operation == "admit-session":
                 from ..hosts.codex_checkpoint import admit_native_checkpoint_session
-                result = admit_native_checkpoint_session(env, args.root, args.expected_sha256,
-                    args.config_path, sandbox=args.sandbox, timeout=args.timeout_seconds)
+                from ..hosts.checkpoint_retention import checkpoint_bank_use
+                with checkpoint_bank_use(args.root):
+                    result = admit_native_checkpoint_session(env, args.root, args.expected_sha256,
+                        args.config_path, sandbox=args.sandbox, timeout=args.timeout_seconds)
             else:
-                result = adopt_todo_context(env, args.root, args.expected_sha256,
-                    source_todo_id=args.source_todo_id, todo_id=args.todo_id, operation_id=args.operation_id,
-                    expected_revision=args.expected_revision, lease_key=args.lease_key, lease_version=args.lease_version)
+                from ..hosts.checkpoint_retention import checkpoint_bank_use
+                with checkpoint_bank_use(args.root):
+                    result = adopt_todo_context(env, args.root, args.expected_sha256,
+                        source_todo_id=args.source_todo_id, todo_id=args.todo_id, operation_id=args.operation_id,
+                        expected_revision=args.expected_revision, lease_key=args.lease_key, lease_version=args.lease_version)
         payload = {"ok": True, **result}
     except (ValueError, OSError, RuntimeError) as error:
         payload = {"ok": False, "error": str(error), "execution_authority_granted": False,
diff --git a/loopx/control_plane/runtime/checkpoint_artifacts.ts b/loopx/control_plane/runtime/checkpoint_artifacts.ts
index cf66252267..345ad044f6 100644
--- a/loopx/control_plane/runtime/checkpoint_artifacts.ts
+++ b/loopx/control_plane/runtime/checkpoint_artifacts.ts
@@ -16,6 +16,7 @@ import {normalizeHeartbeatReplanObligationId, normalizeHeartbeatTodoId} from "..
 import {EffectRuntimeConflictError, EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {EffectiveAction} from "../quota/effective_action.generated.ts";
 import {BARE_SHA256_PATTERN} from "../content_digest.ts";
+import {checkpointRetentionPlan} from "./checkpoint_retention.ts";
 
 const SCHEMA = "loopx_checkpoint_artifact_set_v0";
 type Component = {id: string; source_path: string};
@@ -228,6 +229,52 @@ async function verify(root: string, expected: unknown): Promise<JsonObject> {
 const ADMITTED_RECEIPT_STATUSES: ReadonlySet<string> = new Set(["committed", "replayed", "upgraded"]);
 export async function checkpointArtifacts(value: unknown): Promise<JsonObject> {
   const request = requireJsonObject(value, "checkpoint artifacts request");
+  if (request.action === "retention_plan") {
+    if (!hasExactAuthorityKeys(request, ["action", "input"])) throw new Error("invalid checkpoint retention request");
+    return checkpointRetentionPlan(request.input);
+  }
+  if (request.action === "retention_inventory") {
+    if (!hasExactAuthorityKeys(request, ["action", "bank"])) throw new Error("invalid checkpoint inventory request");
+    const bank = absolute(request.bank, "checkpoint bank");
+    await physical(bank);
+    const points: JsonObject[] = [], incomplete: string[] = [];
+    for (const name of (await readdir(bank)).sort()) {
+      if (!/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(name)) continue;
+      const root = join(bank, name), info = await lstat(root);
+      if (info.isSymbolicLink()) throw new Error("checkpoint bank contains a symlink");
+      if (!info.isDirectory()) continue;
+      const path = join(root, "checkpoint.json");
+      try {await lstat(path);} catch (error) {
+        if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
+        incomplete.push(name); continue;
+      }
+      await physical(path);
+      const text = await readFile(path, "utf8"), manifest = manifestBody(JSON.parse(text));
+      if ((await readdir(root)).sort().join(",") !== "checkpoint.json,components") {
+        throw new Error("checkpoint retention found unexpected point members");
+      }
+      await physical(join(root, "components"));
+      const declared = components(manifest.components).map(c => c.id).sort();
+      if ((await readdir(join(root, "components"))).sort().join(",") !== declared.join(",")) {
+        throw new Error("checkpoint retention found undeclared components");
+      }
+      const witness = requireJsonObject(manifest.artifact_witness, "checkpoint artifact witness");
+      let bytes = Buffer.byteLength(text);
+      for (const members of Object.values(witness)) {
+        if (!Array.isArray(members)) throw new Error("invalid checkpoint artifact inventory");
+        for (const member of members) {
+          const row = requireJsonObject(member, "checkpoint member");
+          if (row.kind === "file") {
+            if (typeof row.size !== "number" || !Number.isSafeInteger(row.size) || row.size < 0) throw new Error("invalid checkpoint member size");
+            bytes += row.size;
+          }
+        }
+      }
+      points.push({id: name, sha256: manifest.sha256, sealed_at: (await lstat(path)).mtimeMs / 1000, bytes});
+    }
+    // Metadata-only inventory: this is not member verification or admission.
+    return {points, incomplete, integrity_verified: false, execution_authority_granted: false};
+  }
   if (request.action === "restore_capture") {
     if (!hasExactAuthorityKeys(request, ["action", "checkpoint_sha256", "goal_id", "agent_id", "current_obligation_id", "review",
         ...(request.include_review_basis === undefined ? [] : ["include_review_basis"])]) ||
diff --git a/loopx/control_plane/runtime/checkpoint_retention.ts b/loopx/control_plane/runtime/checkpoint_retention.ts
new file mode 100644
index 0000000000..22a5a98fd4
--- /dev/null
+++ b/loopx/control_plane/runtime/checkpoint_retention.ts
@@ -0,0 +1,110 @@
+/** Selection only. The explicit local-bank transport owns locks and deletion.
+ * Pins and recent points are hard protections, never a storage-budget hint.
+ */
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject} from "../runtime_decode.ts";
+import {authorityUnicodeCompare, canonicalAuthoritySha256, hasExactAuthorityKeys} from "../coordination/authority_store_codec.ts";
+import {BARE_SHA256_PATTERN} from "../content_digest.ts";
+
+type Point = {id: string; sha256: string; sealed_at: number; bytes: number};
+type Reason = "latest" | "first" | "pin" | "grace" | "since" | "history";
+
+function finite(value: unknown, label: string, minimum = 0): number {
+  if (typeof value !== "number" || !Number.isFinite(value) || value < minimum) throw new Error(`invalid retention ${label}`);
+  return value;
+}
+function integer(value: unknown, label: string, minimum: number): number {
+  const number = finite(value, label, minimum);
+  if (!Number.isSafeInteger(number)) throw new Error(`invalid retention ${label}`);
+  return number;
+}
+function identity(raw: unknown): {id: string; sha256: string} {
+  const row = requireJsonObject(raw, "retention point");
+  if (typeof row.id !== "string" || !/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(row.id) ||
+      typeof row.sha256 !== "string" || !BARE_SHA256_PATTERN.test(row.sha256)) throw new Error("invalid retention point identity");
+  return {id: row.id, sha256: row.sha256};
+}
+
+export function checkpointRetentionPlan(raw: unknown): JsonObject {
+  const input = requireJsonObject(raw, "retention input");
+  if (!hasExactAuthorityKeys(input, ["policy", "points", "now"])) throw new Error("invalid retention input fields");
+  // The one default owner. Capture enablement and local execution remain with
+  // their callers; selecting this policy grants neither.
+  const policy = input.policy === null ? {strategy: "uniform", keep_latest: 1, keep_first: false,
+    minimum_age_seconds: 0, history_slots: 0, protect_since: null, pins: []} :
+    requireJsonObject(input.policy, "retention policy");
+  const common = ["strategy", "keep_latest", "keep_first", "minimum_age_seconds", "pins", "protect_since"];
+  const strategy = policy.strategy;
+  const extra = strategy === "tiered" ? "bucket_seconds" : strategy === "uniform" ? "history_slots" : null;
+  if (!extra || !hasExactAuthorityKeys(policy, [...common, extra]) || typeof policy.keep_first !== "boolean") {
+    throw new Error("invalid retention policy fields");
+  }
+  const latest = integer(policy.keep_latest, "keep_latest", 1);
+  const grace = finite(policy.minimum_age_seconds, "minimum_age_seconds");
+  const since = policy.protect_since === null ? null : finite(policy.protect_since, "protect_since");
+  const spacing = integer(policy[extra], extra, strategy === "tiered" ? 1 : 0);
+  if (strategy === "uniform" && spacing > 0 && policy.keep_first !== true) {
+    throw new Error("uniform retention history requires keep_first for stable time anchors");
+  }
+  const now = finite(input.now, "now");
+  if (!Array.isArray(input.points) || !Array.isArray(policy.pins)) throw new Error("retention points and pins must be arrays");
+  const seen = new Set<string>();
+  const points: Point[] = input.points.map(rawPoint => {
+    const p = requireJsonObject(rawPoint, "retention point");
+    if (!hasExactAuthorityKeys(p, ["id", "sha256", "sealed_at", "bytes"])) throw new Error("invalid retention point fields");
+    const ref = identity(p);
+    if (seen.has(ref.id)) throw new Error("duplicate retention point identity");
+    seen.add(ref.id);
+    return {...ref, sealed_at: finite(p.sealed_at, "sealed_at"), bytes: integer(p.bytes, "bytes", 0)};
+  }).sort((a, b) => a.sealed_at - b.sealed_at || authorityUnicodeCompare(a.id, b.id));
+  const reasons = new Map<string, Set<Reason>>();
+  const keep = (p: Point, reason: Reason) => {
+    if (!reasons.has(p.id)) reasons.set(p.id, new Set());
+    reasons.get(p.id)!.add(reason);
+  };
+  for (const rawPin of policy.pins) {
+    const pin = requireJsonObject(rawPin, "retention pin");
+    if (!hasExactAuthorityKeys(pin, ["id", "sha256"])) throw new Error("invalid retention pin fields");
+    const ref = identity(pin), found = points.find(p => p.id === ref.id);
+    if (!found || found.sha256 !== ref.sha256) throw new Error("pinned checkpoint missing or changed; reconcile before pruning");
+    keep(found, "pin");
+  }
+  points.slice(-latest).forEach(p => keep(p, "latest"));
+  if (policy.keep_first && points.length) keep(points[0], "first");
+  for (const p of points) {
+    // Future timestamps are protected too; clock skew cannot evict a point.
+    if (now - p.sealed_at < grace) keep(p, "grace");
+    if (since !== null && p.sealed_at >= since) keep(p, "since");
+  }
+  // Choose history independently of pins, so adding a pin can never evict an
+  // existing history representative. No score, filenames or model judgments.
+  const history = points.slice(0, Math.max(0, points.length - latest));
+  if (strategy === "tiered" && points.length) {
+    const newest = points[points.length - 1].sealed_at, buckets = new Set<number>();
+    for (const p of history.slice().reverse()) {
+      const age = Math.max(0, newest - p.sealed_at);
+      const bucket = age < spacing ? 0 : 1 + Math.floor(Math.log2(age / spacing));
+      if (!buckets.has(bucket)) {buckets.add(bucket); keep(p, "history");}
+    }
+  } else if (history.length && spacing) {
+    // Equal-time targets across the historical range (not equal row indices).
+    const start = history[0].sealed_at, end = history[history.length - 1].sealed_at;
+    const remaining = new Set(history);
+    for (let n = 1; n <= Math.min(spacing, history.length); n++) {
+      // Include the historical end: keeping both temporal anchors makes a
+      // repeated application on an unchanged bank idempotent.
+      const target = start + (end - start) * n / Math.min(spacing, history.length);
+      const nearest = [...remaining].sort((a, b) => Math.abs(a.sealed_at - target) - Math.abs(b.sealed_at - target) ||
+        b.sealed_at - a.sealed_at || authorityUnicodeCompare(a.id, b.id))[0];
+      remaining.delete(nearest); keep(nearest, "history");
+    }
+  }
+  const retained = points.filter(p => reasons.has(p.id)).map(p => ({...p, reasons: [...reasons.get(p.id)!].sort()}));
+  const removed = points.filter(p => !reasons.has(p.id));
+  const expiries = points.filter(p => reasons.get(p.id)?.has("grace")).map(p => p.sealed_at + grace);
+  const body: JsonObject = {policy, points, retained, removed,
+    retained_bytes: retained.reduce((n, p) => n + p.bytes, 0), removable_bytes: removed.reduce((n, p) => n + p.bytes, 0),
+    grace_expires_at: expiries.length ? Math.min(...expiries) : null,
+    execution_authority_granted: false};
+  return {...body, plan_sha256: canonicalAuthoritySha256(body)};
+}
diff --git a/loopx/hosts/checkpoint_retention.py b/loopx/hosts/checkpoint_retention.py
new file mode 100644
index 0000000000..170ceefe52
--- /dev/null
+++ b/loopx/hosts/checkpoint_retention.py
@@ -0,0 +1,244 @@
+"""Explicit local checkpoint-bank transport; selection belongs to TypeScript.
+
+No store/Goal or benchmark scoring authority. This only removes self-contained
+artifact directories from an operator-selected bank. Native banks are separate.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import time
+from time import monotonic, sleep
+
+from ..control_plane.effect_runtime import effect_runtime_result
+from ..file_lock import exclusive_file_lock, write_json_atomic
+
+_STATE = ".checkpoint-retention"
+_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}\Z")
+_SHA = re.compile(r"[0-9a-f]{64}\Z")
+
+
+def _physical(path: Path) -> Path:
+    path = Path(path)
+    if not path.is_absolute() or path.resolve(strict=True) != path:
+        raise ValueError("Checkpoint retention requires physical absolute paths")
+    return path
+
+
+def _state(bank: Path) -> Path:
+    _physical(bank)
+    if os.name != "posix" or not shutil.rmtree.avoids_symlink_attacks:
+        raise ValueError("Checkpoint rotation requires a qualified POSIX descriptor-relative unlink transport")
+    bank_info = bank.stat()
+    if not stat.S_ISDIR(bank_info.st_mode) or bank_info.st_uid != os.getuid() or bank_info.st_mode & 0o077:
+        raise ValueError("Checkpoint bank must be private and operator-owned before rotation")
+    root = bank / _STATE
+    root.mkdir(mode=0o700, exist_ok=True)
+    info = _physical(root).stat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
+        raise ValueError("Checkpoint retention state must be private and operator-owned")
+    for name in ("receipts", "trash"):
+        (root / name).mkdir(mode=0o700, exist_ok=True)
+        member = _physical(root / name).stat()
+        if not stat.S_ISDIR(member.st_mode) or member.st_uid != os.getuid() or member.st_mode & 0o077:
+            raise ValueError("Checkpoint retention directories must be private and operator-owned")
+    return root
+
+
+@contextmanager
+def checkpoint_bank_use(root: Path):
+    """Serialize a supported reader against rotation only in a managed bank.
+
+    Off-state is a no-op: no lock files or directory creation. Standalone Effect
+    clients and older readers must pin the point before using a managed bank.
+    """
+    bank = next((p for p in Path(root).parents if (p / _STATE).exists()), None)
+    if bank is None:
+        yield
+        return
+    state = _state(bank)
+    # Reuse the kernel-lock owner's descriptor/path checks and inode. Shared
+    # reads permit same-point arms to restore concurrently; only the rotator
+    # needs exclusivity. The POSIX-only transport was checked by _state.
+    import fcntl
+    from ..file_lock import _lock_path, _open_lock_descriptor
+    descriptor = _open_lock_descriptor(_lock_path(state / "bank"), flags=os.O_CREAT | os.O_RDWR)
+    try:
+        deadline = monotonic() + 30
+        while True:
+            try:
+                fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
+                break
+            except BlockingIOError:
+                if monotonic() >= deadline:
+                    raise TimeoutError("Checkpoint reader is waiting for storage rotation") from None
+                sleep(0.05)
+        yield
+    finally:
+        os.close(descriptor)
+
+
+def _inventory(bank: Path) -> dict:
+    return effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "retention_inventory", "bank": str(bank)}, large_local_snapshot=True)
+
+
+def retention_policy_bytes(bank: Path, request_file: Path | None = None) -> bytes:
+    """Read explicit/default bank configuration without creating preview state."""
+    path = request_file or _physical(bank) / _STATE / "policy.json"
+    if request_file is not None or path.exists() or path.is_symlink():
+        return _physical(path).read_bytes()
+    return b"null"  # The typed artifact owner supplies the default.
+
+
+def retention_plan(bank: Path, policy: dict | None, *, now: float | None = None) -> dict:
+    if policy is None:
+        policy = json.loads(retention_policy_bytes(bank))
+    inventory = _inventory(_physical(bank))
+    plan = effect_runtime_result("runtime.checkpoint_artifacts", {
+        "action": "retention_plan", "input": {"policy": policy, "points": inventory["points"],
+        "now": time.time() if now is None else now}}, large_local_snapshot=True)
+    return {**plan, "incomplete_preserved": inventory["incomplete"]}
+
+
+def _sync(path: Path) -> None:
+    if os.name == "posix":
+        fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+
+def _receipt(path: Path, record: dict) -> None:
+    if path.is_symlink():
+        raise ValueError("Checkpoint retention receipt cannot be a symlink")
+    write_json_atomic(path, record, mode=0o600, fsync_directory=True)
+
+
+def _purge(trash: Path) -> None:
+    # Never follow links, including chmod on read-only captured directories.
+    _physical(trash)
+    for directory, children, _ in os.walk(trash, followlinks=False):
+        path = Path(directory)
+        path.chmod(stat.S_IMODE(path.lstat().st_mode) | 0o700, follow_symlinks=False)
+        children[:] = [name for name in children if not (path / name).is_symlink()]
+    shutil.rmtree(trash)
+    _sync(trash.parent)
+
+
+def _recover_trash(bank: Path, state: Path) -> list[dict]:
+    recovered = []
+    for path in sorted((state / "receipts").glob("*.json")):
+        record = json.loads(_physical(path).read_bytes())
+        point = record["point"]
+        if (not _NAME.fullmatch(point["id"]) or not _SHA.fullmatch(point["sha256"]) or
+                path.name != point["id"] + "-" + point["sha256"] + ".json" or
+                record["status"] not in {"planned", "quarantined", "deleted"}):
+            raise ValueError("Malformed retention recovery identity")
+        trash = state / "trash" / (point["id"] + "-" + point["sha256"])
+        if trash.is_symlink():
+            raise ValueError("Checkpoint quarantine cannot be a symlink")
+        if record["status"] == "deleted":
+            continue
+        if trash.exists():
+            if (bank / point["id"]).exists():
+                raise ValueError("Retention source and quarantine both exist; reconcile original deletion")
+            _purge(trash)
+        elif record["status"] == "planned":
+            # No committed move: recompute selection, never replay an old plan.
+            continue
+        elif record["status"] != "quarantined":
+            raise ValueError("Unknown checkpoint retention recovery status")
+        record = {**record, "status": "deleted"}
+        _receipt(path, record)
+        recovered.append(record)
+    return recovered
+
+
+def rotate_bank(bank: Path, policy: dict | None = None, *, execute: bool = False,
+                expected_plan_sha256: str | None = None) -> dict:
+    """Plan, then recheck and quarantine exact unprotected points under a lock.
+
+    Preserve response/error sidecars, incomplete captures and unrecognized paths.
+    A cleanup failure leaves an owned receipt and quarantine for the next call.
+    """
+    bank = _physical(bank)
+    if not execute:
+        return {**retention_plan(bank, policy), "dry_run": True}
+    state = _state(bank)
+    # Keep rotation/recovery single-flight while unlink runs outside the brief
+    # reader lock. Never race another recovery against an in-progress purge.
+    with exclusive_file_lock(state / "rotation", operation="checkpoint-rotation", timeout_seconds=1):
+        return _rotate_locked(bank, state, policy, expected_plan_sha256)
+
+
+def _rotate_locked(bank: Path, state: Path, policy: dict | None, expected_plan_sha256: str | None) -> dict:
+    # One inode, never unlink the lock; existing reader/rotator waiters agree.
+    with exclusive_file_lock(state / "bank", operation="checkpoint-rotate", timeout_seconds=1):
+        # A newly declared missing/changed pin holds even recovery of an older
+        # quarantine. Never finish unlinking before validating current pins.
+        configured = retention_policy_bytes(bank) if policy is None else None
+        plan = retention_plan(bank, policy)
+        if expected_plan_sha256 is not None and plan["plan_sha256"] != expected_plan_sha256:
+            raise ValueError("Checkpoint retention plan changed; review a fresh plan")
+        verified = []
+        if plan["removed"] or any((state / "trash").iterdir()):
+            if not plan["retained"]:
+                raise ValueError("No retained recovery point; reconcile quarantine before unlink")
+            for point in plan["retained"]:
+                effect_runtime_result("runtime.checkpoint_artifacts", {"action": "verify",
+                    "root": str(bank / point["id"]), "expected_sha256": point["sha256"]},
+                    timeout=120, large_local_snapshot=True)
+                verified.append(point["id"])
+        recovered = _recover_trash(bank, state)
+        # A new publication or changed manifest invalidates this exact plan.
+        current = sorted(_inventory(bank)["points"], key=lambda p: (p["sealed_at"], p["id"]))
+        if current != plan["points"]:
+            raise ValueError("Checkpoint bank moved before retention; retry with a fresh plan")
+        if configured is not None and retention_policy_bytes(bank) != configured:
+            raise ValueError("Checkpoint bank policy moved before retention; review a fresh plan")
+        deleted = []
+        for point in plan["removed"]:
+            source = _physical(bank / point["id"])
+            # Stale/replaced roots cannot be renamed under a previous identity.
+            observed = json.loads(_physical(source / "checkpoint.json").read_bytes())
+            if observed["sha256"] != point["sha256"]:
+                raise ValueError("Checkpoint changed before quarantine")
+            target = state / "trash" / (point["id"] + "-" + point["sha256"])
+            path = state / "receipts" / (point["id"] + "-" + point["sha256"] + ".json")
+            if target.exists() or target.is_symlink():
+                raise ValueError("Checkpoint quarantine already exists; recover original deletion")
+            record = {"point": point, "plan_sha256": plan["plan_sha256"], "status": "planned"}
+            _receipt(path, record)
+            source.rename(target)
+            _sync(bank)
+            _sync(target.parent)
+            _receipt(path, {**record, "status": "quarantined"})
+            # Release read serialization before the potentially long unlink.
+            deleted.append((target, path, record))
+    for target, path, record in deleted:
+        _purge(target)
+        _receipt(path, {**record, "status": "deleted"})
+    return {**plan, "dry_run": False, "deleted": [p for p in plan["removed"]], "recovered": recovered,
+            "retained_integrity_verified": verified}
+
+
+def bank_generation(bank: Path, policy_bytes: bytes) -> tuple:
+    """Cheap watch trigger; unchanged banks do not redo selection or write logs."""
+    rows = []
+    for root in _physical(bank).iterdir():
+        if not _NAME.fullmatch(root.name):
+            continue
+        if root.is_symlink():
+            raise ValueError("Checkpoint bank contains a symlink")
+        manifest = root / "checkpoint.json"
+        if manifest.exists():
+            info = manifest.lstat()
+            rows.append((root.name, info.st_ino, info.st_size, info.st_mtime_ns))
+    return policy_bytes, tuple(sorted(rows))
diff --git a/tests/control_plane/test_checkpoint_retention.py b/tests/control_plane/test_checkpoint_retention.py
new file mode 100644
index 0000000000..4e9b510234
--- /dev/null
+++ b/tests/control_plane/test_checkpoint_retention.py
@@ -0,0 +1,252 @@
+"""Real local artifact owner, retention CLI and filesystem recovery negatives."""
+import json
+import os
+from pathlib import Path
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from loopx.cli import main
+from loopx.control_plane.effect_runtime import effect_runtime_result
+from loopx.hosts import checkpoint_retention as transport
+
+pytestmark = pytest.mark.skipif(os.name != "posix", reason="Local unlink transport is qualified on POSIX")
+
+
+@pytest.fixture
+def bank(tmp_path):
+    root = (tmp_path / "bank").resolve()
+    root.mkdir(mode=0o700)
+    source = tmp_path / "code"
+    source.mkdir()
+    (source / "negative.txt").write_text("retained counterexample")
+    source.chmod(0o500)
+    components = [{"id": "research", "source_path": str(source)}]
+    probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components})
+    for n in range(8):
+        point = root / f"p{n}"
+        effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components,
+            "destination": str(point), "point_metadata": {}, "expected_source_sha256": probe["source_sha256"]})
+        os.utime(point / "checkpoint.json", (1000 + n * 100, 1000 + n * 100))
+    source.chmod(0o700)
+    return root
+
+
+@pytest.fixture
+def policy():
+    return dict(strategy="uniform", keep_latest=2, keep_first=True, minimum_age_seconds=0,
+                pins=[], protect_since=None, history_slots=1)
+
+
+def test_preview_is_read_only_and_execute_preserves_pins_latest_and_unsealed(bank, policy):
+    (bank / "unsealed").mkdir()
+    (bank / "unsealed/work-in-progress").write_text("held writer export")
+    (bank / "p2.json").write_text('{"ok":false,"error_kind":"TimeoutError"}')
+    pin = json.loads((bank / "p2/checkpoint.json").read_bytes())["sha256"]
+    policy["pins"] = [dict(id="p2", sha256=pin)]
+    plan = transport.rotate_bank(bank, policy)
+    assert plan["dry_run"] and not (bank / ".checkpoint-retention").exists()
+    assert {p["id"] for p in plan["retained"]} == {"p0", "p2", "p5", "p6", "p7"}
+    result = transport.rotate_bank(bank, policy, execute=True, expected_plan_sha256=plan["plan_sha256"])
+    assert {p["id"] for p in result["deleted"]} == {"p1", "p3", "p4"}
+    assert (bank / "p2.json").read_text().startswith('{"ok":false')
+    assert (bank / "unsealed/work-in-progress").read_text() == "held writer export"
+    assert not list((bank / ".checkpoint-retention/trash").iterdir())
+    for p in result["retained"]:
+        observed = effect_runtime_result("runtime.checkpoint_artifacts", {
+            "action": "verify", "root": str(bank / p["id"]), "expected_sha256": p["sha256"]})
+        assert observed["integrity_verified"]
+    assert transport.rotate_bank(bank, policy, execute=True)["deleted"] == []
+
+
+def test_stale_preview_and_changed_pin_cannot_delete(bank, policy):
+    preview = transport.rotate_bank(bank, policy)
+    os.utime(bank / "p2/checkpoint.json", (9000, 9000))
+    with pytest.raises(ValueError, match="plan changed"):
+        transport.rotate_bank(bank, policy, execute=True, expected_plan_sha256=preview["plan_sha256"])
+    assert len(list(bank.glob("*/checkpoint.json"))) == 8
+    policy["pins"] = [dict(id="missing", sha256="a" * 64)]
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert len(list(bank.glob("*/checkpoint.json"))) == 8
+
+
+def test_interrupted_unlink_recovers_original_quarantine_without_recapturing(bank, policy, monkeypatch):
+    purge = transport._purge
+    monkeypatch.setattr(transport, "_purge", lambda path: (_ for _ in ()).throw(OSError("injected unlink failure")))
+    with pytest.raises(OSError, match="injected"):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert list((bank / ".checkpoint-retention/trash").iterdir())
+    monkeypatch.setattr(transport, "_purge", purge)
+    result = transport.rotate_bank(bank, policy, execute=True)
+    assert len(result["recovered"]) == 4 and result["deleted"] == []
+    receipts = [json.loads(p.read_bytes()) for p in (bank / ".checkpoint-retention/receipts").glob("*.json")]
+    assert {p["status"] for p in receipts} == {"deleted"}
+    assert (bank / "p7/checkpoint.json").exists()
+
+
+def test_new_missing_pin_holds_prior_quarantine_recovery(bank, policy, monkeypatch):
+    purge = transport._purge
+    monkeypatch.setattr(transport, "_purge", lambda path: (_ for _ in ()).throw(OSError("injected")))
+    with pytest.raises(OSError):
+        transport.rotate_bank(bank, policy, execute=True)
+    pending = next((bank / ".checkpoint-retention/receipts").glob("*.json"))
+    ref = json.loads(pending.read_bytes())["point"]
+    policy["pins"] = [dict(id=ref["id"], sha256=ref["sha256"])]
+    monkeypatch.setattr(transport, "_purge", purge)
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert (bank / ".checkpoint-retention/trash" / (ref["id"] + "-" + ref["sha256"])).exists()
+
+
+def test_symlink_bank_point_or_manifest_never_follows_outside(bank, policy, tmp_path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    sentinel = outside / "valuable"
+    sentinel.write_text("keep")
+    (bank / "foreign").symlink_to(outside, target_is_directory=True)
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert sentinel.read_text() == "keep" and (bank / "p0").exists()
+    (bank / "foreign").unlink()
+    manifest = bank / "p1/checkpoint.json"
+    manifest.unlink()
+    manifest.symlink_to(sentinel)
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert sentinel.read_text() == "keep"
+
+
+def test_unknown_point_members_hold_whole_bank_without_erasing_new_data(bank, policy):
+    member = bank / "p2/new-unsealed-evidence"
+    member.write_text("not declared in the immutable capture")
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert member.read_text() == "not declared in the immutable capture"
+    assert len(list(bank.glob("*/checkpoint.json"))) == 8
+
+
+def test_corrupt_newest_cannot_evict_the_older_recovery_points(bank, policy):
+    (bank / "p7/components/research/negative.txt").write_text("changed after publication")
+    with pytest.raises(RuntimeError):
+        transport.rotate_bank(bank, policy, execute=True)
+    assert len(list(bank.glob("*/checkpoint.json"))) == 8
+
+
+def test_managed_reader_serializes_cleanup_and_off_state_creates_nothing(bank, policy):
+    with transport.checkpoint_bank_use(bank / "p7"):
+        assert not (bank / ".checkpoint-retention").exists()
+    transport._state(bank)
+    entered, release = [threading.Event(), threading.Event()], threading.Event()
+    def reader(index):
+        with transport.checkpoint_bank_use(bank / "p7/components/research"):
+            entered[index].set()
+            assert release.wait(10)
+    workers = [threading.Thread(target=reader, args=(index,)) for index in range(2)]
+    for worker in workers:
+        worker.start()
+    assert all(event.wait(5) for event in entered), "same-point readers must not serialize each other"
+    try:
+        with pytest.raises(TimeoutError):
+            transport.rotate_bank(bank, policy, execute=True)
+        assert len(list(bank.glob("*/checkpoint.json"))) == 8
+    finally:
+        release.set()
+        for worker in workers:
+            worker.join(10)
+    assert not any(worker.is_alive() for worker in workers)
+    assert transport.rotate_bank(bank, policy, execute=True)["deleted"]
+
+
+def test_real_cli_preview_execute_and_newest_restore(bank, policy, tmp_path, capsys):
+    request = tmp_path / "retention.json"
+    request.write_text(json.dumps(policy))
+    args = ["--format", "json", "research-checkpoint", "rotate", "--root", str(bank), "--request-file", str(request)]
+    assert main(args) == 0
+    plan = json.loads(capsys.readouterr().out)
+    assert plan["dry_run"] and not (bank / ".checkpoint-retention").exists()
+    assert main([*args, "--execute", "--expected-plan-sha256", plan["plan_sha256"]]) == 0
+    applied = json.loads(capsys.readouterr().out)
+    assert len(applied["deleted"]) == 4
+    newest = next(p for p in applied["retained"] if p["id"] == "p7")
+    assert main(["--format", "json", "research-checkpoint", "verify", "--root", str(bank / "p7"),
+                 "--expected-sha256", newest["sha256"]]) == 0
+    assert json.loads(capsys.readouterr().out)["integrity_verified"]
+    result = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "materialize", "root": str(bank / "p7"),
+        "expected_sha256": newest["sha256"], "destination": str(tmp_path / "inert")})
+    assert result["materialized"] and not result["activation_performed"]
+    assert (tmp_path / "inert/components/research/negative.txt").read_text() == "retained counterexample"
+    assert main([*args, "--watch"]) == 1
+    assert "requires rotate" in json.loads(capsys.readouterr().out)["error"]
+
+
+def test_cli_default_is_latest_only_and_managed_bank_policy_adds_explicit_pins(bank, capsys):
+    args = ["--format", "json", "research-checkpoint", "rotate", "--root", str(bank)]
+    assert main(args) == 0
+    plan = json.loads(capsys.readouterr().out)
+    assert [p["id"] for p in plan["retained"]] == ["p7"]
+    assert not (bank / ".checkpoint-retention").exists()
+    state = transport._state(bank)
+    pin = json.loads((bank / "p2/checkpoint.json").read_bytes())["sha256"]
+    selected = {**plan["policy"], "pins": [dict(id="p2", sha256=pin)]}
+    (state / "policy.json").write_text(json.dumps(selected))
+    assert main(args) == 0
+    pinned = json.loads(capsys.readouterr().out)
+    assert {p["id"] for p in pinned["retained"]} == {"p2", "p7"}
+    assert {p["id"] for p in transport.rotate_bank(bank)["retained"]} == {"p2", "p7"}
+    assert main([*args, "--execute", "--expected-plan-sha256", pinned["plan_sha256"]]) == 0
+    applied = json.loads(capsys.readouterr().out)
+    assert len(applied["deleted"]) == 6 and (bank / "p2/checkpoint.json").exists()
+    (state / "policy.json").write_text("{}")
+    assert main([*args, "--execute"]) == 1
+    assert not json.loads(capsys.readouterr().out)["ok"]
+    assert (bank / "p2/checkpoint.json").exists()
+
+
+def test_cli_watcher_rotates_new_publication_and_silences_unchanged_generation(bank, policy, tmp_path, capsys, monkeypatch):
+    from loopx.cli_commands import research_checkpoint as cli
+    policy.update(keep_first=False, keep_latest=1, history_slots=0)
+    request = tmp_path / "policy.json"
+    request.write_text(json.dumps(policy))
+    sleeps = []
+    def wait(_):
+        sleeps.append(True)
+        if len(sleeps) == 1:
+            components = [dict(id="research", source_path=str(bank / "p7/components/research"))]
+            probe = effect_runtime_result("runtime.checkpoint_artifacts", {"action": "probe", "components": components})
+            effect_runtime_result("runtime.checkpoint_artifacts", {"action": "capture", "components": components,
+                "destination": str(bank / "p8"), "point_metadata": {}, "expected_source_sha256": probe["source_sha256"]})
+        elif len(sleeps) == 3:
+            raise KeyboardInterrupt
+    import time
+    monkeypatch.setattr(cli, "time", SimpleNamespace(time=time.time, sleep=wait))
+    with pytest.raises(KeyboardInterrupt):
+        main(["--format", "json", "research-checkpoint", "rotate", "--root", str(bank),
+              "--request-file", str(request), "--execute", "--watch", "--interval-seconds", "10"])
+    packets = [json.loads(line) for line in capsys.readouterr().out.splitlines()]
+    assert len(packets) == 2
+    assert [p["id"] for p in packets[1]["deleted"]] == ["p7"]
+    assert (bank / "p8/checkpoint.json").exists()
+
+
+def test_cli_watcher_revisits_grace_expiry_without_a_new_capture(bank, policy, tmp_path, capsys, monkeypatch):
+    from loopx.cli_commands import research_checkpoint as cli
+    clock, sleeps = [1800.0], []
+    policy.update(history_slots=0, minimum_age_seconds=500)
+    request = tmp_path / "policy.json"
+    request.write_text(json.dumps(policy))
+    def wait(_):
+        sleeps.append(True)
+        if len(sleeps) > 1:
+            raise KeyboardInterrupt
+        clock[0] += 600
+    monkeypatch.setattr(cli, "time", SimpleNamespace(time=lambda: clock[0], sleep=wait))
+    monkeypatch.setattr(transport, "time", SimpleNamespace(time=lambda: clock[0]))
+    with pytest.raises(KeyboardInterrupt):
+        main(["--format", "json", "research-checkpoint", "rotate", "--root", str(bank),
+              "--request-file", str(request), "--execute", "--watch", "--interval-seconds", "10"])
+    packets = [json.loads(line) for line in capsys.readouterr().out.splitlines()]
+    assert len(packets) == 2
+    assert {p["id"] for p in packets[1]["deleted"]} == {"p4", "p5"}
+    assert [p.name for p in sorted(bank.glob("p*"))] == ["p0", "p6", "p7"]
diff --git a/tests/control_plane_ts/checkpoint_retention.test.ts b/tests/control_plane_ts/checkpoint_retention.test.ts
new file mode 100644
index 0000000000..73ea4493b4
--- /dev/null
+++ b/tests/control_plane_ts/checkpoint_retention.test.ts
@@ -0,0 +1,77 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {checkpointArtifacts} from "../../loopx/control_plane/runtime/checkpoint_artifacts.ts";
+import {checkpointRetentionPlan} from "../../loopx/control_plane/runtime/checkpoint_retention.ts";
+
+const points = Array.from({length: 21}, (_, n) => ({id: `p${String(n).padStart(2, "0")}`,
+  sha256: n.toString(16).padStart(64, "0"), sealed_at: n * 100, bytes: 10}));
+const policy = {strategy: "tiered", keep_latest: 2, keep_first: true, minimum_age_seconds: 0,
+  pins: [], protect_since: null, bucket_seconds: 100};
+const plan = (overrides = {}, rows = points) => checkpointRetentionPlan({policy: {...policy, ...overrides}, points: rows, now: 3000});
+const ids = (rows: unknown) => (rows as {id: string}[]).map(p => p.id);
+
+test("default retention keeps only the latest complete publication without origin or sampled history", () => {
+  const result = checkpointRetentionPlan({policy: null, points, now: 3000});
+  assert.deepEqual(ids(result.retained), ["p20"]);
+  assert.equal((result.removed as unknown[]).length, 20);
+  assert.deepEqual(result.policy, {strategy: "uniform", keep_latest: 1, keep_first: false,
+    minimum_age_seconds: 0, history_slots: 0, protect_since: null, pins: []});
+  assert.equal(result.execution_authority_granted, false);
+  assert.deepEqual(checkpointRetentionPlan({policy: null, points: [], now: 3000}).removed, []);
+});
+
+test("tiered retention preserves latest, origin and newest representative per exponentially wider age band", () => {
+  const result = plan();
+  assert.deepEqual(ids(result.retained), ["p00", "p04", "p12", "p16", "p18", "p19", "p20"]);
+  assert.equal(result.retained_bytes, 70);
+  assert.equal(result.removable_bytes, 140);
+  assert.deepEqual(plan({}, points.slice().reverse()), result);
+  assert.equal(result.execution_authority_granted, false);
+});
+
+test("uniform history samples elapsed time, including irregular capture spacing", () => {
+  const {bucket_seconds: _, ...common} = policy;
+  const input = {policy: {...common, strategy: "uniform", history_slots: 2}, now: 3000, points};
+  assert.deepEqual(ids(checkpointRetentionPlan(input).retained), ["p00", "p09", "p18", "p19", "p20"]);
+  const irregular = [0, 1, 2, 3, 4, 50, 98, 99, 100].map((at, n) => ({...points[n], sealed_at: at}));
+  const uneven = checkpointRetentionPlan({...input, points: irregular});
+  assert.deepEqual(ids(uneven.retained), ["p00", "p05", "p06", "p07", "p08"]);
+});
+
+test("pins, grace and pending-selection cutoff add hard protections without evicting history", () => {
+  const baseline = new Set(ids(plan().retained));
+  const protectedResult = plan({pins: [{id: points[5].id, sha256: points[5].sha256}], protect_since: 1700});
+  const kept = new Set(ids(protectedResult.retained));
+  for (const id of baseline) assert(kept.has(id));
+  assert(kept.has("p05") && kept.has("p17"));
+  const grace = plan({minimum_age_seconds: 1500});
+  assert.equal(grace.grace_expires_at, 3100);
+  for (let n = 16; n <= 20; n++) assert(ids(grace.retained).includes(points[n].id));
+  assert(ids(plan({minimum_age_seconds: 1}, [...points, {...points[0], id: "future", sealed_at: 9000}]).retained).includes("future"));
+});
+
+test("both sampling strategies are idempotent without a new publication or expired protection", () => {
+  for (const selected of [policy, {strategy: "uniform", keep_latest: 2, keep_first: true, minimum_age_seconds: 0,
+    pins: [], protect_since: null, history_slots: 3}]) {
+    const first = checkpointRetentionPlan({policy: selected, points, now: 3000});
+    const retainedIds = new Set(ids(first.retained));
+    const again = checkpointRetentionPlan({policy: selected, points: points.filter(p => retainedIds.has(p.id)), now: 3000});
+    assert.deepEqual(again.removed, []);
+  }
+});
+
+test("no policy can disable newest protection; malformed, missing and changed pins fail closed", () => {
+  for (const overrides of [{keep_latest: 0}, {keep_latest: 1.5}, {bucket_seconds: 0}, {minimum_age_seconds: -1},
+    {pins: [{id: "missing", sha256: points[0].sha256}]}, {pins: [{id: points[0].id, sha256: points[1].sha256}]},
+    {pins: [{id: "../escape", sha256: points[0].sha256}]}, {strategy: "other"}, {unknown: true}]) {
+    assert.throws(() => plan(overrides));
+  }
+  assert.throws(() => plan({}, [...points, points[0]]), /duplicate/);
+  assert.throws(() => plan({}, [{...points[0], bytes: NaN}]), /bytes/);
+  assert.equal((plan({}, []).removed as unknown[]).length, 0);
+});
+
+test("existing artifact Effect entry delegates to the sole typed retention selection owner", async () => {
+  assert.deepEqual(await checkpointArtifacts({action: "retention_plan", input: {policy, points, now: 3000}}), plan());
+  await assert.rejects(checkpointArtifacts({action: "retention_plan", input: {}, execute: true}), /request/);
+});