fix(benchmark): capture full checkpoints after settled work

benchmark/README.md

@@ -215,7 +215,8 @@ name is already declared; source-path deduplication and byte witnesses still app
 The destination must be a new direct child of the trial log owner's
 `checkpoints` directory. Capture runs after a successful execute wake's original
 settlement, before the next wake. It uses the remaining phase budget with a
-30-second capture ceiling, preserves SQLite's committed WAL through the existing
+30-second capture ceiling (90 seconds for the complete controller handoff below),
+preserves SQLite's committed WAL through the existing
 backup owner, rejects observed source changes/symlinks and never overwrites a
 destination. Capture failures are recorded separately in the wake receipt;
 they do not replay work or change its settled outcome. The ceiling bounds
@@ -244,6 +245,34 @@ requesting a new point. Omit the checkpoint declaration on a new worker to
 retain ordinary launch behavior. This is file/session reconstruction, not
 process-memory recovery or permission to inspect crash memory.
 
+For an explicitly requested **complete capture after settled work**, omit
+`trigger`, use a fresh direct-child `destination` as above, and include all four
+flags: `"native_session": true`, `"loopx_state": true`,
+`"evaluation_opportunity": true`, and `"controller_handoff": true`. Pair the
+request with the existing controller file and native feedback. The native tool
+must finish its original Turn settlement before returning; the worker then
+captures the completed native cursor, canonical research context and declared
+workspace, and the external controller freezes their writers while capturing
+native opportunity, incumbent and the original phase clock. This does not admit
+a replan, interrupt the source or spend another work slot. An unsettled wake
+refuses capture. The preparation and writer barrier share the original
+90-second window, capped by the original phase; post-release sealing uses the
+existing bounded completion allowance within that phase. Failure keeps the
+source's settled outcome and records a missed point. Each destination is attempted
+once, including failures or uncertain replies: a later wake cannot replace that
+point. Its original wake retains an immutable declaration witness before the
+artifact-side attempt index is written, so index permission recovery cannot
+erase the missed point. A separately declared new point needs a fresh destination. An exhausted
+original phase rejects before reading the declaration or any source. Omitting the declaration
+keeps ordinary execution unchanged; no new operator trigger is accepted.
+
+The joint qualification command below accepts `--capture-boundary settled_work`
+to exercise this source through the actual worker and controller, fresh ordinary
+and stride receivers, and another full capture/receiver. Its default remains
+`fresh_replan`. Both use deterministic local inference and synthetic native
+feedback: they qualify mechanical reconstruction and adoption, not scientific
+benefit or authorization to stop a live campaign.
+
 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

@@ -9,6 +9,7 @@
 import hashlib
 import json
 import math
+import os
 import subprocess
 import tempfile
 import time
@@ -55,9 +56,9 @@ def parse_checkpoint_request(raw: bytes) -> dict[str, Any]:
     if "controller_handoff" in request and type(request["controller_handoff"]) is not bool:
         raise ValueError("checkpoint controller_handoff must be an explicit boolean")
     if request.get("controller_handoff") and (
-            request.get("trigger") != "fresh_replan" or not request.get("native_session") or
+            not request.get("native_session") or
             not request.get("loopx_state") or not request.get("evaluation_opportunity")):
-        raise ValueError("controller handoff requires fresh_replan and all three state participants")
+        raise ValueError("controller handoff requires all three state participants")
     if "trigger" in request and request["trigger"] != "fresh_replan":
         raise ValueError("checkpoint trigger must be fresh_replan or omitted")
     if not isinstance(request["components"], list) or not request["components"]:
@@ -117,6 +118,66 @@ def _prior_replan_guard_seen(env: dict[str, str], current: Path, obligation_id:
     return False
 
 
+def _write_checkpoint_attempt(path: Path, value: dict[str, str]) -> bool:
+    """Retain a private first-attempt record, including interrupted writes."""
+    try:
+        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    except FileExistsError:
+        if path.is_symlink():
+            raise ValueError("completed capture attempt must not be a symlink")
+        return False
+    with os.fdopen(descriptor, "w") as stream:
+        json.dump(value, stream)
+        stream.flush()
+        os.fsync(stream.fileno())
+    for parent in (path.parent, path.parent.parent, path.parent.parent.parent):
+        directory = os.open(parent, os.O_RDONLY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    return True
+
+
+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.
+
+    A controller export lives outside this guest. The declared destination names
+    one point, including a missed or uncertain capture; later work cannot replace
+    it. These private Host records grant no work or restore authority.
+    """
+    if (not wake_root.is_absolute() or wake_root.resolve() != wake_root or
+            not turn_id or Path(turn_id).name != turn_id):
+        raise ValueError("completed capture requires its physical owned wake")
+    for prior in sorted(wake_root.glob("*/checkpoint-completed-request.json")):
+        if prior.resolve(strict=True) != prior:
+            raise ValueError("completed capture wake witness must be physical")
+        value = json.loads(prior.read_bytes())
+        if (not isinstance(value, dict) or
+                set(value) != {"destination", "settlement_effect_id", "request_sha256"} or
+                any(not isinstance(item, str) or not item for item in value.values())):
+            raise ValueError("completed capture wake witness is incomplete")
+        if prior.parent.name == turn_id or value["destination"] == str(destination):
+            return False
+    wake = wake_root / turn_id
+    wake.mkdir(parents=True, exist_ok=True, mode=0o700)
+    if wake.resolve() != wake:
+        raise ValueError("completed capture wake must be physical")
+    value = {"destination": str(destination), "settlement_effect_id": effect_id,
+             "request_sha256": request_sha256}
+    # This existing wake log owner is independent of the optional artifact
+    # directory. Index permission recovery cannot erase the original attempt.
+    if not _write_checkpoint_attempt(wake / "checkpoint-completed-request.json", value):
+        return False
+    attempts = destination.parent / ".completed-attempts"
+    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")
+    return _write_checkpoint_attempt(marker, value)
+
+
 def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: str) -> str:
     """Preload the original guard once, then capture before invoking the model.
 
@@ -274,22 +335,28 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
         receipt["checkpoint"]["reason"] = "wake_not_successful_work"
         return
     started = time.monotonic()
+    capture_started = time.time()
     try:
         phase_deadline = env.get("LOOPX_PHASE_DEADLINE_EPOCH")
-        window = CHECKPOINT_CAPTURE_SECONDS if supplied_request and supplied_request[0].get("controller_handoff") else 30
-        deadline = time.time() + window
+        original_deadline = None
         if phase_deadline is not None:
             original_deadline = float(phase_deadline)
             if not math.isfinite(original_deadline):
                 raise ValueError("checkpoint requires a finite original phase deadline")
-            deadline = min(deadline, original_deadline)
-        if deadline <= time.time():
-            receipt["checkpoint"]["reason"] = "capture_window_closed"
-            return
+            if original_deadline <= capture_started:
+                receipt["checkpoint"]["reason"] = "capture_window_closed"
+                return
         if supplied_request is None:
             request, raw = read_checkpoint_request(request_path)
         else:
             request, raw = supplied_request
+        window = CHECKPOINT_CAPTURE_SECONDS if request.get("controller_handoff") else 30
+        deadline = capture_started + window
+        if original_deadline is not None:
+            deadline = min(deadline, original_deadline)
+        if deadline <= time.time():
+            receipt["checkpoint"]["reason"] = "capture_window_closed"
+            return
         if supplied_request is None:
             from .checkpoint_continuation import restored_capture_gate
             restore_gate = restored_capture_gate(env, full_history=not request.get("native_session", False))
@@ -319,14 +386,20 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
         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 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"
+                return
         # Only this returned typed owner decision admits the capture boundary.
         # Point metadata remains evidence, never authority or scientific validity.
         metadata = {
             "operator_metadata": request["point_metadata"],
             "wake_turn_id": receipt["turn_id"],
             "settlement_effect_id": identity.effect_id,
             "session": receipt.get("session"),
-            "request_sha256": hashlib.sha256(raw).hexdigest(),
+            "request_sha256": request_sha256,
         }
         components = request["components"]
         if pending_guard is not None:

benchmark/tests/qualify_checkpoint_joint.py

@@ -93,6 +93,8 @@ 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",
+                        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")
     args = parser.parse_args()
@@ -188,7 +190,8 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                 "action": "verify", "root": str(point), "expected_sha256": packet["sha256"]}, large_local_snapshot=True)
             assert verified["integrity_verified"]
             metadata = packet["point_metadata"]
-            assert metadata["held_decision_component_id"] and metadata["worker_file_snapshot"]["prepared_source_verified"]
+            assert metadata["worker_file_snapshot"]["prepared_source_verified"]
+            assert bool(metadata.get("held_decision_component_id")) == (args.capture_boundary == "fresh_replan")
             config = output / "native-config"
             config.write_bytes(config_source.read_bytes())
             config.chmod(0o600)
@@ -201,7 +204,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                       f"(p/'phase-deadline').write_text(str(int(time.time())+{SYNTHETIC_PHASE_SECONDS})); "
                       "code=pytest.main(["
                       "'/source/benchmark/tests/test_native_checkpoint_handoff.py::"
-                      f"test_actual_worker_captures_pending_replan_before_native_model[{args.context_delivery}-planning]',"
+                      f"test_actual_worker_captures_pending_replan_before_native_model[{args.context_delivery}-planning-{args.capture_boundary}]',"
                       "'-q','-rs','-p','no:cacheprovider','--basetemp=/qualification/source-tmp',"
                       "'--junitxml=/qualification/source.xml']); ")
             if args.context_delivery == "on":
@@ -243,7 +246,12 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
             point = points[0]
             packet = json.loads((point / "checkpoint.json").read_bytes())
             metadata = packet["point_metadata"]
-            assert metadata["held_decision_component_id"] and metadata["decision_sha256"]
+            if args.capture_boundary == "fresh_replan":
+                assert metadata["held_decision_component_id"] and metadata["decision_sha256"]
+            else:
+                assert "held_decision_component_id" not in metadata and "replan" not in metadata
+                assert metadata["settlement_effect_id"] and metadata["wake_turn_id"]
+                assert metadata["native_session"]["completed_turn_id"] == metadata["native_session"]["turn_ids"][-1]
             git_object = point / "components/task_workspace/.git/objects/ab/retained-object"
             assert git_object.read_text() == "Readonly source evidence" and git_object.stat().st_mode & 0o777 == 0o444
             config = _stage_source_configuration(container, metadata["native_session"], output / "native-config")
@@ -329,7 +337,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
         require_source_checkout(source, source_revision)
         receipt = {"passed": True, "controller_uid": os.getuid(), "provider_revision": args.provider_revision,
                    "source_revision": source_revision, "source_tree_dirty": False, "source_imports_verified": True,
-                   "context_delivery": args.context_delivery,
+                   "context_delivery": args.context_delivery, "capture_boundary": args.capture_boundary,
                    "native_context_reads": context_results,
                    "qualification_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in
                        [Path(__file__), source / "benchmark/tests/test_native_checkpoint_handoff.py",