feat(benchmark): add one settled-work handoff without changing capture cadence

benchmark/README.md

@@ -276,6 +276,28 @@ and stride receivers, and another full capture/receiver. Its default remains
 feedback: they qualify mechanical reconstruction and adoption, not scientific
 benefit or authorization to stop a live campaign.
 
+To request **one explicit settled-work handoff alongside normal replan capture**,
+keep `trigger: "fresh_replan"` and the checkpoint-root `destination`, and add:
+
+```json
+"settled_work_capture_once": {
+  "destination": "/logs/agent/checkpoints/upgrade-handoff",
+  "purpose": "Explicit upgrade handoff"
+}
+```
+
+The handoff inherits the declared components and participant flags. It waits for
+a successful execute wake with its original settled Turn and unchanged request;
+restored-history gates still apply. It records its own `checkpoint_handoff`
+result without replacing the pre-replan result. A missed or uncertain attempt
+is consumed, including when its artifact index is unavailable. Later work cannot
+backfill it. Normal distinct replan captures continue at the configured cadence.
+Omit this field to retain the existing behavior. Use it only for a necessary
+operator-declared handoff, account for its one additional capture within the
+original phase, and do not turn it into per-wake captures or a preparation Todo.
+The joint qualifier's `--capture-boundary settled_once` exercises this path with
+the same real native session, SQLite state, controller and fresh receivers.
+
 For pre-decision capture at a **fresh replan projected at the start of a heartbeat
 wake**, add `"trigger": "fresh_replan"` and set `destination` to the trial's
 `checkpoints` directory itself. The Host preloads that wake's original quota guard

benchmark/runtime/checkpoint.py

@@ -44,7 +44,7 @@ def parse_checkpoint_request(raw: bytes) -> dict[str, Any]:
     """Validate the same explicit declaration before staging and before capture."""
     request = json.loads(raw)
     if (not isinstance(request, dict) or
-            set(request) - {"destination", "components", "point_metadata", "native_session", "evaluation_opportunity", "loopx_state", "trigger", "controller_handoff"} or
+            set(request) - {"destination", "components", "point_metadata", "native_session", "evaluation_opportunity", "loopx_state", "trigger", "controller_handoff", "settled_work_capture_once"} or
             not {"destination", "components", "point_metadata"}.issubset(request)):
         raise ValueError("checkpoint request requires destination, components and point_metadata")
     if "native_session" in request and type(request["native_session"]) is not bool:
@@ -65,6 +65,18 @@ def parse_checkpoint_request(raw: bytes) -> dict[str, Any]:
         raise ValueError("checkpoint requires explicit components")
     if not isinstance(request["point_metadata"], dict):
         raise ValueError("checkpoint point_metadata must be an object")
+    handoff = request.get("settled_work_capture_once")
+    if "settled_work_capture_once" in request:
+        if (request.get("trigger") != "fresh_replan" or not isinstance(handoff, dict)
+                or set(handoff) != {"destination", "purpose"}
+                or not isinstance(handoff["purpose"], str) or not handoff["purpose"].strip()):
+            raise ValueError("settled-work handoff requires fresh_replan plus destination and explicit purpose")
+        root = normalize_checkpoint_request_path(request["destination"])
+        target = normalize_checkpoint_request_path(handoff["destination"])
+        if PurePosixPath(target).parent != PurePosixPath(root):
+            raise ValueError("settled-work handoff must be a direct child of this trial's checkpoint root")
+        if "settled_work_handoff" in request["point_metadata"]:
+            raise ValueError("settled-work handoff metadata must not replace an operator field")
     return request
 
 
@@ -139,6 +151,10 @@ def _write_checkpoint_attempt(path: Path, value: dict[str, str]) -> bool:
     return True
 
 
+def _completed_attempt_path(destination: Path) -> Path:
+    return destination.parent / ".completed-attempts" / (hashlib.sha256(destination.name.encode()).hexdigest() + ".json")
+
+
 def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_sha256: str,
                                     wake_root: Path, turn_id: str) -> bool:
     """Use the original wake witness when the artifact-side index was unavailable.
@@ -174,7 +190,7 @@ def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_s
     attempts.mkdir(parents=True, exist_ok=True, mode=0o700)
     if attempts.resolve() != attempts:
         raise ValueError("completed capture attempt directory must be physical")
-    marker = attempts / (hashlib.sha256(destination.name.encode()).hexdigest() + ".json")
+    marker = _completed_attempt_path(destination)
     return _write_checkpoint_attempt(marker, value)
 
 
@@ -196,6 +212,10 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
         request, raw = {}, None
     if declaration_error is None and request.get("trigger") != "fresh_replan":
         return body
+    if declaration_error is None and request.get("settled_work_capture_once") is not None:
+        receipt["checkpoint_handoff"] = {"pending": True, "turn_id": receipt["turn_id"],
+            "destination": request["settled_work_capture_once"]["destination"],
+            "request_sha256": hashlib.sha256(raw).hexdigest()}
     if receipt.get("mode") != "heartbeat" or receipt.get("stage") != "execute":
         raise ValueError("fresh_replan checkpoint trigger requires a heartbeat execute wake")
     started = time.monotonic()
@@ -320,13 +340,37 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
     return preloaded_body
 
 
+def _capture_declared_settled_handoff(env: dict[str, str], receipt: dict[str, Any]) -> None:
+    """One explicit handoff, not a periodic capture or a substitute replan point."""
+    pending = receipt.get("checkpoint_handoff")
+    if not isinstance(pending, dict) or pending.get("pending") is not True:
+        return
+    if not receipt.get("ok") or receipt.get("stage") != "execute" or receipt.get("budget_exhausted"):
+        return  # No eligible completed boundary; keep the request unconsumed.
+    result = {"captured": False, "research_point_qualified": False, "phase": "after_settled_work"}
+    completed = {k: v for k, v in receipt.items() if k not in {"checkpoint", "checkpoint_handoff"}}
+    capture_after_wake(env, completed, settled_handoff=pending)
+    result.update(completed["checkpoint"])
+    receipt["checkpoint_handoff"] = {"pending": False, **result}
+
+
+def _capture_destination(root: Path, value: str) -> Path:
+    destination = Path(value)
+    if not destination.is_absolute() or destination != destination.resolve() or destination.parent != root:
+        raise ValueError("checkpoint destination must be a normalized direct child of this trial's checkpoints directory")
+    return destination
+
+
 def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
                        supplied_request: tuple[dict[str, Any], bytes] | None = None,
                        pending_guard: Path | None = None,
-                       restore_gate: dict[str, Any] | None = None) -> None:
+                       restore_gate: dict[str, Any] | None = None,
+                       settled_handoff: dict[str, Any] | None = None) -> None:
     request_path = env.get("LOOPX_CHECKPOINT_REQUEST")
     if not request_path:
         return  # Feature-off path performs no additional read or write.
+    if supplied_request is None and settled_handoff is None:
+        _capture_declared_settled_handoff(env, receipt)
     if supplied_request is None and (receipt.get("checkpoint") or {}).get("phase") == "before_replan":
         return  # The after-wake hook cannot replace the pre-decision point.
     receipt["checkpoint"] = {"captured": False, "research_point_qualified": False}
@@ -346,10 +390,51 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             if original_deadline <= capture_started:
                 receipt["checkpoint"]["reason"] = "capture_window_closed"
                 return
+        root = Path(env["LOOPX_WAKE_LOG_DIR"]).parent / "checkpoints"
+        observed = None
+        if settled_handoff is not None:
+            expected_hash = settled_handoff.get("request_sha256")
+            if (settled_handoff.get("turn_id") != receipt.get("turn_id")
+                    or not isinstance(expected_hash, str) or len(expected_hash) != 64):
+                raise ValueError("settled-work handoff request or original Turn changed")
+            destination = _capture_destination(root, settled_handoff["destination"])
+            if _completed_attempt_path(destination).exists():
+                receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay"
+                return
+            observed = read_heartbeat_settlement(
+                Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"],
+                agent_id=env["LOOPX_AGENT_ID"], todo_id=None,
+                turn_instance_id=receipt["turn_id"], resolve_original_binding=True,
+                registry_path=Path(env["LOOPX_REGISTRY"]),
+            )
+            if observed is None or (observed.progress or {}).get("state") != "settled":
+                receipt["checkpoint"]["reason"] = "owned_turn_not_settled"
+                return
+            identity = observed.identity.value
+            if identity is None:
+                raise ValueError("checkpoint settlement identity is unavailable")
+            # Reserve the preflight's exact point before rereading a declaration
+            # that may have disappeared. The existing wake witness survives an
+            # unavailable artifact index; later work must not backfill this point.
+            if not _mark_completed_capture_attempt(destination, identity.effect_id, expected_hash,
+                    Path(env["LOOPX_WAKE_LOG_DIR"]), receipt["turn_id"]):
+                receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay"
+                return
+        if original_deadline is not None and original_deadline <= time.time():
+            receipt["checkpoint"]["reason"] = "capture_window_closed"
+            return
         if supplied_request is None:
             request, raw = read_checkpoint_request(request_path)
         else:
             request, raw = supplied_request
+        if settled_handoff is not None:
+            handoff = request["settled_work_capture_once"]
+            if (hashlib.sha256(raw).hexdigest() != expected_hash
+                    or handoff["destination"] != str(destination)):
+                raise ValueError("settled-work handoff request or original Turn changed")
+            request = {k: v for k, v in request.items() if k not in {"trigger", "settled_work_capture_once"}}
+            request.update(destination=handoff["destination"], point_metadata={**request["point_metadata"],
+                "settled_work_handoff": {"purpose": handoff["purpose"]}})
         window = CHECKPOINT_CAPTURE_SECONDS if request.get("controller_handoff") else 30
         deadline = capture_started + window
         if original_deadline is not None:
@@ -367,27 +452,26 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             receipt["checkpoint"]["reason"] = "pre_replan_hook_required"
             return
         # This is a private evidence artifact below the runner's log owner.
-        destination = Path(request["destination"])
-        root = Path(env["LOOPX_WAKE_LOG_DIR"]).parent / "checkpoints"
-        if not destination.is_absolute() or destination != destination.resolve() or destination.parent != root:
-            raise ValueError("checkpoint destination must be a normalized direct child of this trial's checkpoints directory")
+        destination = _capture_destination(root, request["destination"])
         if destination.exists():
             receipt["checkpoint"]["reason"] = "destination_already_present_no_replay"
             return
-        observed = read_heartbeat_settlement(
-            Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"],
-            agent_id=env["LOOPX_AGENT_ID"], todo_id=None,
-            turn_instance_id=receipt["turn_id"], resolve_original_binding=True,
-            registry_path=Path(env["LOOPX_REGISTRY"]),
-        )
+        if observed is None:
+            observed = read_heartbeat_settlement(
+                Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"],
+                agent_id=env["LOOPX_AGENT_ID"], todo_id=None,
+                turn_instance_id=receipt["turn_id"], resolve_original_binding=True,
+                registry_path=Path(env["LOOPX_REGISTRY"]),
+            )
         if observed is None or (observed.progress or {}).get("state") != "settled":
             receipt["checkpoint"]["reason"] = "owned_turn_not_settled"
             return
         identity = observed.identity.value
         if identity is None:
             raise ValueError("checkpoint settlement identity is unavailable")
         request_sha256 = hashlib.sha256(raw).hexdigest()
-        if request.get("controller_handoff") and supplied_request is None:
+        if (settled_handoff is None and pending_guard is None
+                and (request.get("controller_handoff") or supplied_request is not None)):
             if not _mark_completed_capture_attempt(destination, identity.effect_id, request_sha256,
                     Path(env["LOOPX_WAKE_LOG_DIR"]), receipt["turn_id"]):
                 receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay"

benchmark/tests/qualify_checkpoint_joint.py

@@ -93,7 +93,7 @@ def main():
     parser.add_argument("--image", default="python:3.12-slim")
     parser.add_argument("--context-delivery", choices=("off", "on"), default="off",
                         help="Qualify the existing opt-in context transport through source and fresh receivers")
-    parser.add_argument("--capture-boundary", choices=("fresh_replan", "settled_work"), default="fresh_replan",
+    parser.add_argument("--capture-boundary", choices=("fresh_replan", "settled_work", "settled_once"), default="fresh_replan",
                         help="Qualify the existing automatic pre-replan or explicit settled-wake full capture")
     parser.add_argument("--reuse-point", type=Path, help="Reuse an already sealed synthetic source; never count its source test as passed")
     parser.add_argument("--captured-config", type=Path, help="The reused point's matching native fixture configuration")

benchmark/tests/test_checkpoint_joint_receiver.py

@@ -40,6 +40,7 @@ def test_captured_point_continues_with_fresh_authority(
     from benchmark.runtime.sessions import BenchmarkSessionWake
     from benchmark.runtime.worker import native_command
     from loopx.cli import main
+    from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
     from loopx.control_plane.effect_runtime import (
         effect_runtime_result,
         EffectRuntimeRemoteError,
@@ -123,7 +124,7 @@ def cli(*args):
         code = main([*prefix, *args])
         output = capsys.readouterr().out
         assert code == 0, output
-        return json.loads(output)
+        return read_cli_policy_payload(output, returncode=code)
 
     def receiver_guard(*args):
         # The receiver checks machine fields. run_native_work presents this