From d69329a724f6a90680fec6d290f3c06b0c85e7af Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 4 Nov 2026 01:38:18 +0800
Subject: [PATCH 1/2] fix(benchmark): capture full checkpoints after settled
 work

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/README.md                           |  29 ++++-
 benchmark/runtime/checkpoint.py               |  62 ++++++++--
 benchmark/tests/qualify_checkpoint_joint.py   |  16 ++-
 benchmark/tests/test_checkpoint_handoff.py    |  70 ++++++++++-
 .../tests/test_checkpoint_joint_receiver.py   |  11 +-
 .../tests/test_native_checkpoint_handoff.py   | 113 +++++++++++++-----
 .../hierarchical-agent-stride-control-v0.md   |  21 +++-
 ...rarchical-agent-stride-control-v0.zh-CN.md |  14 ++-
 .../test_research_checkpoint_capture.py       |  17 +++
 9 files changed, 299 insertions(+), 54 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index 22341c4a38..54d050d8f5 100644
--- a/benchmark/README.md
+++ b/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,32 @@ 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. 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
diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py
index 0f5d37c5ea..d8e5a7ca32 100644
--- a/benchmark/runtime/checkpoint.py
+++ b/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,38 @@ def _prior_replan_guard_seen(env: dict[str, str], current: Path, obligation_id:
     return False
 
 
+def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_sha256: str) -> bool:
+    """A controller export lives outside this guest; retain its first attempt here.
+
+    The declared destination names one point, including a missed or uncertain
+    capture. A new wake or repaired source cannot replace it with later state.
+    This private Host transport index grants no work or restore authority.
+    """
+    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")
+    try:
+        descriptor = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    except FileExistsError:
+        if marker.is_symlink():
+            raise ValueError("completed capture attempt must not be a symlink")
+        return False
+    with os.fdopen(descriptor, "w") as stream:
+        json.dump({"destination": destination.name, "settlement_effect_id": effect_id,
+                   "request_sha256": request_sha256}, stream)
+        stream.flush()
+        os.fsync(stream.fileno())
+    for parent in (attempts, attempts.parent, attempts.parent.parent):
+        directory = os.open(parent, os.O_RDONLY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    return True
+
+
 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 +307,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,6 +358,11 @@ 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):
+                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 = {
@@ -326,7 +370,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             "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:
diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py
index 3b186bb8c1..7767415e26 100644
--- a/benchmark/tests/qualify_checkpoint_joint.py
+++ b/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",
diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py
index 2f4d89ed95..e958e974f2 100644
--- a/benchmark/tests/test_checkpoint_handoff.py
+++ b/benchmark/tests/test_checkpoint_handoff.py
@@ -11,6 +11,71 @@
 from benchmark.runtime.checkpoint_handoff import await_controller_capture
 
 
+@pytest.mark.parametrize("trigger", [None, "fresh_replan"])
+@pytest.mark.parametrize("missing", [None, "native_session", "loopx_state", "evaluation_opportunity"])
+def test_full_controller_capture_requires_all_participants_at_either_boundary(trigger, missing):
+    from benchmark.runtime.checkpoint import parse_checkpoint_request
+
+    request = {"destination": "/logs/agent/checkpoints/point", "point_metadata": {},
+        "components": [{"id": "workspace", "source_path": "/task"}],
+        "controller_handoff": True, "native_session": True,
+        "loopx_state": True, "evaluation_opportunity": True}
+    if trigger is not None:
+        request["trigger"] = trigger
+    if missing:
+        del request[missing]
+        with pytest.raises(ValueError, match="all three state participants"):
+            parse_checkpoint_request(json.dumps(request).encode())
+    else:
+        assert parse_checkpoint_request(json.dumps(request).encode()) == request
+
+
+def test_full_controller_capture_does_not_invent_an_operator_trigger():
+    from benchmark.runtime.checkpoint import parse_checkpoint_request
+
+    request = {"destination": "/logs/agent/checkpoints/point", "point_metadata": {},
+        "components": [{"id": "workspace", "source_path": "/task"}],
+        "controller_handoff": True, "native_session": True, "loopx_state": True,
+        "evaluation_opportunity": True, "trigger": "operator_handoff"}
+    with pytest.raises(ValueError, match="trigger must be fresh_replan or omitted"):
+        parse_checkpoint_request(json.dumps(request).encode())
+
+
+def test_completed_controller_point_cannot_retry_a_failed_boundary_with_later_work(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from benchmark.runtime import checkpoint
+
+    declaration = tmp_path / "request.json"
+    destination = tmp_path / "checkpoints" / "point"
+    declaration.write_text(json.dumps({"destination": str(destination), "point_metadata": {},
+        "components": [{"id": "workspace", "source_path": str(tmp_path / "workspace")}],
+        "controller_handoff": True, "native_session": True, "loopx_state": True, "evaluation_opportunity": True}))
+    env = {"LOOPX_CHECKPOINT_REQUEST": str(declaration), "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"),
+        "LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), "LOOPX_REGISTRY": str(tmp_path / "registry"),
+        "LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent"}
+    monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_gate",
+                        lambda *a, **k: {"capture_allowed": True})
+    monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", lambda *a, **k:
+        SimpleNamespace(progress={"state": "settled"},
+                        identity=SimpleNamespace(value=SimpleNamespace(effect_id=k["turn_instance_id"]))))
+    attempts = []
+    def unavailable(*args):
+        attempts.append(True)
+        raise ValueError("synthetic source unavailable before sealing")
+    monkeypatch.setattr(checkpoint, "research_participants", unavailable)
+    first = {"ok": True, "stage": "execute", "mode": "heartbeat", "turn_id": "original"}
+    checkpoint.capture_after_wake(env, first)
+    assert first["checkpoint"]["reason"] == "capture_failed" and not destination.exists()
+    marker = next((destination.parent / ".completed-attempts").iterdir())
+    original = marker.read_bytes()
+    for turn in ("original", "later-work"):
+        later = {**first, "turn_id": turn}
+        checkpoint.capture_after_wake(env, later)
+        assert later["checkpoint"]["reason"] == "destination_already_attempted_no_replay"
+        assert later["ok"] and not later["checkpoint"]["captured"]
+    assert attempts == [True] and marker.read_bytes() == original
+
+
 @pytest.mark.parametrize("response", ["captured", "failed", "foreign", "timeout"])
 def test_original_request_only_and_cleanup_after_missed_point(tmp_path, response):
     mailbox = tmp_path / "checkpoint-controller"
@@ -160,7 +225,10 @@ def controller(root, components, metadata, probe, deadline, **kwargs):
         observed = SimpleNamespace(progress={"state": "settled"},
             identity=SimpleNamespace(value=SimpleNamespace(effect_id="original-effect")))
         monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", lambda *a, **k: observed)
-        checkpoint.capture_after_wake(env, receipt, supplied_request=(request, b"declared"))
+        declaration = tmp_path / "capture-request.json"
+        declaration.write_text(json.dumps(request))
+        env["LOOPX_CHECKPOINT_REQUEST"] = str(declaration)
+        checkpoint.capture_after_wake(env, receipt)
         assert receipt["checkpoint"]["captured"] is (phase_seconds != 20)
         assert receipt["ok"] is True  # An optional miss cannot replay successful work.
     else:
diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py
index 0bbb16d49a..83ee843925 100644
--- a/benchmark/tests/test_checkpoint_joint_receiver.py
+++ b/benchmark/tests/test_checkpoint_joint_receiver.py
@@ -652,7 +652,9 @@ def run_native_work(current_guard, logs, purpose):
             pending_body = capture_before_replan(env, pending_receipt, "Current destination contract.")
             assert pending_receipt["checkpoint"]["reason"] == "restored_replan_pending", pending_receipt
             assert not pending_receipt["checkpoint"]["captured"]
-            assert "Original pending packet" in pending_body and "checkpoint:" + digest in pending_body
+            has_pending = "held_decision_component_id" in metadata
+            assert ("Original pending packet" in pending_body) is has_pending
+            assert "checkpoint:" + digest in pending_body
             # Both stop locations share the same gate. Repeated/restarted
             # capture attempts neither interrupt the fork nor publish a point.
             for _ in range(2):
@@ -793,8 +795,11 @@ def run_native_work(current_guard, logs, purpose):
             # owes this original Turn's settlement, so recapture stays closed.
             unconsumed = restored_capture_gate(env)
             assert not unconsumed["capture_allowed"]
-            original_packet = unit / "components" / metadata["held_decision_component_id"]
-            assert str(original_packet) in unconsumed["instructions"]
+            if has_pending:
+                original_packet = unit / "components" / metadata["held_decision_component_id"]
+                assert str(original_packet) in unconsumed["instructions"]
+            else:
+                assert "no retained pending replan packet" in unconsumed["instructions"]
             assert str(wakes / review_turn / "checkpoint-start-guard.json") not in unconsumed["instructions"]
             cli(
                 "quota",
diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py
index 31f5799577..8f007fe3f9 100644
--- a/benchmark/tests/test_native_checkpoint_handoff.py
+++ b/benchmark/tests/test_native_checkpoint_handoff.py
@@ -369,10 +369,11 @@ def test_midwake_native_profile_refusal_precedes_interruption(tmp_path, monkeypa
 @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN") or
                    os.environ.get("LOOPX_CHECKPOINT_EXTERNAL_CONTROLLER") != "1",
                    reason="Requires the real native CLI and explicitly attached checkpoint supervisor")
+@pytest.mark.parametrize("capture_boundary", ["fresh_replan", "settled_work"])
 @pytest.mark.parametrize("explore_mode", ["off", "planning"])
 @pytest.mark.parametrize("context_delivery", ["off", "on"])
-def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, explore_mode, context_delivery):
-    """Six real settled work wakes -> pre-decision capture -> stride -> work.
+def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, explore_mode, context_delivery, capture_boundary):
+    """Six native work wakes -> explicit settled capture or pre-decision capture.
 
     Inference is deterministic; native history/tools, canonical SQLite, quota,
     periodic cadence and the Host writer barrier use their production paths.
@@ -425,6 +426,33 @@ def run(*args):
         "--explore-mode", explore_mode, "--execute")
     assert configured["after"]["execution_profile"]["replan_after_effective_turns"] == 6
 
+    def settlement_commands(expected):
+        binding = ("--goal-id", cli.GOAL_ID, "--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID)
+        return [
+            ("refresh-state", *binding, "--classification", "synthetic_candidate_verified",
+             "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+             "--delivery-boundary", "in_flight_continuation", "--progress-result-class", "advanced",
+             "--progress-surface-id", "fixture:candidate", "--progress-evidence-id", f"evidence:candidate-{expected}",
+             "--no-global-sync", "--suppress-external-sinks"),
+            ("quota", "spend-slot", *binding, "--source", "heartbeat", "--execute"),
+        ]
+
+    def record_counterexample():
+        if explore_mode == "planning":
+            from loopx.capabilities.explore.result_log import (
+                append_explore_result_events, build_explore_node_event, build_explore_finding_event,
+                explore_result_log_path,
+            )
+            assert candidate.read_text() == "6" and json.loads((project / "evidence-1.json").read_text())["value"] == 1
+            append_explore_result_events(explore_result_log_path(runtime, cli.GOAL_ID), [
+                build_explore_node_event(goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID,
+                    node_id="restart-at-one", title="Does restarting at one retain local progress?", status="dead_end"),
+                build_explore_finding_event(goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID,
+                    node_id="restart-at-one", finding_id="restart-counterexample", title="Restart loses retained local progress",
+                    status="refuted", tags=["writeback-result"], evidence_refs=["evidence-1.json", "evidence-6.json"],
+                    summary="In this synthetic counter fixture, one is below the retained six. Continue from six; this says nothing about benchmark quality.")],
+                expected_goal_id=cli.GOAL_ID)
+
     def response(payload, index, message):
         requests.append(payload)
         if continuation := native_process_continuation(payload, f"checkpoint_{index}"):
@@ -472,6 +500,16 @@ def response(payload, index, message):
                     "[(Path('candidate.txt').write_text(str(v)),"
                     "Path('evidence-'+str(v)+'.json').write_text(json.dumps({'value':v,'synthetic':True}))) for v in values]; "
                     "assert Path('candidate.txt').read_text()==str(values[-1]); print('CANDIDATE_VERIFIED')")
+            if capture_boundary == "settled_work" and phase["values"] == [6]:
+                # The actual native tool settles its original work before the
+                # completed-wake hook. No operator-side settlement is backdated.
+                prefix = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
+                          "--runtime-root", str(runtime), "--format", "json"]
+                code += ("; import os,subprocess; "
+                    f"commands={settlement_commands(6)!r}; "
+                    f"results=[json.loads(subprocess.run({prefix!r}+list(args)+"
+                    "['--turn-instance-id',os.environ['LOOPX_TURN']],capture_output=True,text=True,check=True).stdout) "
+                    "for args in commands]; assert results[-1]['settlement_progress']['state']=='settled'")
             return {"type": "function_call", "id": f"fc_work_{index}", "call_id": f"call_work_{index}",
                     "name": "exec_command", "arguments": json.dumps({
                         "cmd": shlex.quote(sys.executable) + " -c " + shlex.quote(code),
@@ -503,6 +541,8 @@ def response(payload, index, message):
             return {"type": "function_call", "id": "fc_packet", "call_id": "call_packet",
                     "name": "exec_command", "arguments": json.dumps({
                         "cmd": "cat " + shlex.quote(str(paths[0])), "max_output_tokens": 30000})}
+        if capture_boundary == "settled_work" and phase["kind"] == "work" and phase["values"] == [6]:
+            record_counterexample()
         message["content"][0]["text"] = "Original packet read." if phase["kind"] == "review" else "Synthetic work verified."
         return message
 
@@ -511,18 +551,18 @@ def settle_work(result, expected):
         assert candidate.read_text() == str(expected)
         guard = json.loads((project / "work-guard.json").read_text())
         assert guard["heartbeat_receipt"]["turn_instance_id"] == result["turn_id"]
-        binding = ("--goal-id", cli.GOAL_ID, "--agent-id", cli.AGENT_ID,
-                   "--todo-id", cli.TODO_ID, "--turn-instance-id", result["turn_id"])
-        run("refresh-state", *binding, "--classification", "synthetic_candidate_verified",
-            "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
-            "--delivery-boundary", "in_flight_continuation", "--progress-result-class", "advanced",
-            "--progress-surface-id", "fixture:candidate", "--progress-evidence-id", f"evidence:candidate-{expected}",
-            "--no-global-sync", "--suppress-external-sinks")
-        spent = run("quota", "spend-slot", *binding, "--source", "heartbeat", "--execute")
-        assert spent["settlement_progress"]["state"] == "settled"
+        if capture_boundary == "settled_work" and expected == 6:
+            from loopx.control_plane.quota.settlement import read_heartbeat_settlement
+            observed = read_heartbeat_settlement(runtime, goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID,
+                todo_id=cli.TODO_ID, turn_instance_id=result["turn_id"], registry_path=registry)
+            assert observed.progress["state"] == "settled"
+        else:
+            _, spent = [run(*args, "--turn-instance-id", result["turn_id"])
+                                for args in settlement_commands(expected)]
+            assert spent["settlement_progress"]["state"] == "settled"
         work_turns.append(result["turn_id"])
         if expected == 3:
-            replay = run("quota", "spend-slot", *binding, "--source", "heartbeat", "--execute")
+            replay = run(*settlement_commands(expected)[-1], "--turn-instance-id", result["turn_id"])
             assert replay["idempotent_replay"] and not replay["appended"]
 
     with real_native_source(tmp_path / "provider", os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"],
@@ -536,9 +576,19 @@ def settle_work(result, expected):
             REASONING_EFFORT="high", OPENAI_BASE_URL=base["model_providers"]["fixture"]["base_url"],
             OPENAI_API_KEY="synthetic-local-only", LOOPX_CODEX_TURN_TIMEOUT_SEC="60")
         env["LOOPX_CONTEXT_DELIVERY"] = context_delivery
+        request = tmp_path / "request.json"
+        declaration = {"destination": str(logs.parent / "checkpoints"),
+            "components": [{"id": "task_workspace", "source_path": str(project)}], "point_metadata": {},
+            "native_session": True, "loopx_state": True, "evaluation_opportunity": True, "controller_handoff": True}
+        if capture_boundary == "fresh_replan":
+            declaration["trigger"] = "fresh_replan"
+        else:
+            declaration["destination"] += "/settled-work"
+        request.write_text(json.dumps(declaration))
         for value in range(1, 7):
             phase.update(kind="work", values=[value], tool_sent=False, context_started=False)
-            prior = run_once(env)
+            actual = {**env, "LOOPX_CHECKPOINT_REQUEST": str(request)} if capture_boundary == "settled_work" and value == 6 else env
+            prior = run_once(actual)
             settle_work(prior, value)
         assert len(set(work_turns)) == 6
         if context_delivery == "off":
@@ -547,26 +597,23 @@ def settle_work(result, expected):
             assert reader.deliveries[0]["mode"] == "full"
             assert any(row["mode"] == "delta" for row in reader.deliveries[1:])
         review_request_index = len(requests)
-        if explore_mode == "planning":
-            from loopx.capabilities.explore.result_log import (
-                append_explore_result_events, build_explore_node_event, build_explore_finding_event,
-                explore_result_log_path,
-            )
-            assert candidate.read_text() == "6" and json.loads((project / "evidence-1.json").read_text())["value"] == 1
-            append_explore_result_events(explore_result_log_path(runtime, cli.GOAL_ID), [
-                build_explore_node_event(goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID,
-                    node_id="restart-at-one", title="Does restarting at one retain local progress?", status="dead_end"),
-                build_explore_finding_event(goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID,
-                    node_id="restart-at-one", finding_id="restart-counterexample", title="Restart loses retained local progress",
-                    status="refuted", tags=["writeback-result"], evidence_refs=["evidence-1.json", "evidence-6.json"],
-                    summary="In this synthetic counter fixture, one is below the retained six. Continue from six; this says nothing about benchmark quality.")],
-                expected_goal_id=cli.GOAL_ID)
-        request = tmp_path / "request.json"
-        request.write_text(json.dumps({"destination": str(logs.parent / "checkpoints"),
-            "components": [{"id": "task_workspace", "source_path": str(project)}],
-            "point_metadata": {},
-            "trigger": "fresh_replan", "native_session": True, "loopx_state": True,
-            "evaluation_opportunity": True, "controller_handoff": True}))
+        if capture_boundary == "settled_work":
+            assert prior["checkpoint"]["captured"], prior
+            assert not prior["checkpoint"]["research_point_qualified"]
+            assert cli._heartbeat_receipt_count(runtime, prior["turn_id"]) == 1
+            assert not guards and not list(logs.glob("*/checkpoint-start-guard.json"))
+            assert not prior.get("checkpoint_handoffs") and "turn_envelope" not in prior
+            from benchmark.runtime.checkpoint import capture_after_wake
+            replay = dict(prior)
+            call_count = len(calls)
+            capture_after_wake(actual, replay)
+            assert replay["checkpoint"]["reason"] == "destination_already_attempted_no_replay"
+            assert len(calls) == call_count and prior["checkpoint"]["captured"]
+            (tmp_path / "controller-result.json").write_text(json.dumps({**prior["checkpoint"],
+                "natural_work_turns": work_turns, "capture_boundary": capture_boundary,
+                "context_delivery": context_delivery, "context_reads": reader.deliveries}))
+            return
+        record_counterexample()
         phase.update(kind="review", tool_sent=False, context_started=False)
         result = run_once({**env, "LOOPX_CHECKPOINT_REQUEST": str(request)})
         assert result["ok"] and result["checkpoint"]["captured"], result
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
index 12283ce21c..d204e101c3 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
@@ -939,8 +939,8 @@ ten hours is an experiment setting, not a product default or automatic pivot.
 The proposed automatic capture policy is **off by default; when explicitly
 enabled, capture at each fresh replan node before its strategy decision**. Reuse
 the canonical replan obligation identity: repeated context reads or retries of
-that node do not create another point. Ordinary work Turns do not trigger a full
-capture. A successful point records the replan identity, the completed Host
+that node do not create another point. Ordinary work Turns do not automatically
+trigger a full capture. A successful point records the replan identity, the completed Host
 cursor, source revisions and current strategy version; these are evidence
 bindings, not authority. Branches can then compare different replan decisions
 from the same retained research state, including a decision to keep the route.
@@ -960,6 +960,23 @@ Measure capture time and storage per node before choosing compression, shared
 immutable members or retention limits. Retention must preserve declared study
 points and their branches rather than silently selecting favorable outcomes.
 
+An explicitly requested full benchmark capture can also use the existing
+completed-wake boundary: omit the declaration's `trigger` and supply native
+session, canonical research state, evaluation opportunity and controller handoff
+together. The original work must already be settled and its native turn completed;
+no synthetic replan, extra guard or work-slot spend is introduced. The same
+controller writer barrier, participant rereads and fresh destination admission
+apply. File-loaded and preloaded full declarations share the original 90-second
+capture window, capped by the source phase, plus the existing bounded sealing
+allowance after writer release. Ordinary file-only captures keep their 30-second
+window. A missed point continues the source and does not authorize a signal-based
+handoff or relabel a later state. The Host retains the first attempt per declared
+destination, even on failure or a lost reply; replay cannot replace its original
+settled source, and an exhausted phase performs no declaration/source reads.
+The joint native qualification has an explicit
+`settled_work` variant; deterministic inference qualifies mechanical adoption,
+not a scientific advantage or live rollout.
+
 Record the actual boundary and any skew:
 
 - native conversation cursor and a Host-qualified export/fork path, including
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
index ffd6cdd226..7773d40c53 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
@@ -761,7 +761,7 @@ Host 会话 owner 和 artifact provider；本 RFC 定义研究实验的一致性
 
 自动捕获策略提案是**默认关闭；显式开启后，在每个新的 replan 节点、策略决策
 之前捕获**。复用 canonical replan obligation 身份：重复读取 context 或重试
-同一节点不会生成另一研究点。普通工作 Turn 不触发完整捕获。成功研究点绑定
+同一节点不会生成另一研究点。普通工作 Turn 不自动触发完整捕获。成功研究点绑定
 replan 身份、已完成 Host cursor、来源 revision 及当前策略版本；这些是证据
 绑定，不是权限。不同分支可从同一份积累比较不同 replan 决策，包括保留原路线。
 
@@ -772,6 +772,18 @@ replan 身份、已完成 Host cursor、来源 revision 及当前策略版本；
 的捕获时间和存储量，再决定压缩、共享不可变成员或保留上限。保留策略须保护
 已声明研究点及其分支，不能静默挑选有利结果。
 
+显式请求的完整 benchmark 捕获也可使用既有 completed-wake 边界：省略声明的
+`trigger`，同时开启 native session、canonical 研究状态、evaluation opportunity
+和 controller handoff。原工作必须已结算、原生 turn 必须已完成；不虚构 replan、
+增加 guard 或多花工作 slot。继续使用同一 controller writer barrier、参与者
+复读与目标端新准入。文件读取和预加载的完整声明共享原有 90 秒捕获窗口，受
+来源 phase 上限约束，释放 writer 后再使用既有有界 sealing 余量。普通文件捕获
+保持 30 秒窗口。错过研究点时继续来源工作，不授权通过信号交接或把更晚状态
+改称原研究点。Host 保留每个声明 destination 的首次尝试，包括失败或回执丢失；
+重试不能替换原已结算来源，原 phase 耗尽时不读取声明或来源。联合原生资格化
+提供显式 `settled_work` 变体；确定性推理只验证
+机械采用，不证明科学收益或允许在线推广。
+
 记录实际边界与 skew：
 
 - 原生会话 cursor 和已资格化的 Host export/fork 路径，包括 compaction 记录、
diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py
index f7a7b5d3ac..b12c52a679 100644
--- a/tests/control_plane/test_research_checkpoint_capture.py
+++ b/tests/control_plane/test_research_checkpoint_capture.py
@@ -831,6 +831,23 @@ def run(*args):
     data["vision_patch"]["acceptance_summary"] = "The missed original point stays missed; later research boundaries still qualify separately."
     data["path_delta"]["observed_reality"] = "Real permission recovery did not recapture later code as the original research point."
     data["path_delta"].update(schema_version="goal_path_delta_v0", outcome="replan")
+    control = guard["replan_action_packet"].get("stride_control", {})
+    if "minimal_example" in control:
+        # The permission invariant must also settle under the current enabled
+        # stride contract; capture failure does not waive that original review.
+        decision = json.loads(json.dumps(control["minimal_example"]))
+        decision.update(action="verify", axes={
+            "work_slice": {"plan": "Qualify one separately declared later capture boundary.",
+                           "reason": "The original missed point must remain missed."},
+            "route_distance": {"plan": "Retain the source capture method.",
+                               "reason": "This fixture tests permission recovery, not strategy quality."},
+            "validation_scope": {"plan": "Check a new obligation captures the later candidate exactly once.",
+                                 "reason": "Successful old-point refusal alone does not qualify a later point."}},
+            next_probe={"scope": "One new synthetic replan boundary.", "prediction": "Only the new point captures.",
+                "adoption_criterion": "The original stays absent and the new candidate matches.",
+                "cost_bound": "One original admitted wake and local validation.",
+                "minimum_evidence": "Original guard, new obligation and sealed candidate."})
+        data["stride_decision"] = decision
     vision.write_text(json.dumps(data))
     run("refresh-state", "--goal-id", cli.GOAL_ID, *replan_binding,
         "--classification", "missed_original_point_preserved", "--delivery-batch-scale", "multi_surface",

From df2eb06b8d183fbc5c56691a68694b779bd2a258 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 4 Nov 2026 02:17:07 +0800
Subject: [PATCH 2/2] fix(benchmark): retain wake witness before checkpoint
 indexing

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/README.md                           |  4 +-
 benchmark/runtime/checkpoint.py               | 65 ++++++++++++++-----
 benchmark/tests/test_checkpoint_handoff.py    | 48 ++++++++++++++
 .../hierarchical-agent-stride-control-v0.md   |  4 +-
 ...rarchical-agent-stride-control-v0.zh-CN.md |  3 +-
 5 files changed, 103 insertions(+), 21 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index 54d050d8f5..3b51a9b072 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -260,7 +260,9 @@ refuses capture. The preparation and writer barrier share the original
 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. A separately declared new point needs a fresh destination. An exhausted
+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.
 
diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py
index d8e5a7ca32..9d218cdabb 100644
--- a/benchmark/runtime/checkpoint.py
+++ b/benchmark/runtime/checkpoint.py
@@ -118,30 +118,19 @@ def _prior_replan_guard_seen(env: dict[str, str], current: Path, obligation_id:
     return False
 
 
-def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_sha256: str) -> bool:
-    """A controller export lives outside this guest; retain its first attempt here.
-
-    The declared destination names one point, including a missed or uncertain
-    capture. A new wake or repaired source cannot replace it with later state.
-    This private Host transport index grants no work or restore authority.
-    """
-    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")
+def _write_checkpoint_attempt(path: Path, value: dict[str, str]) -> bool:
+    """Retain a private first-attempt record, including interrupted writes."""
     try:
-        descriptor = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     except FileExistsError:
-        if marker.is_symlink():
+        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({"destination": destination.name, "settlement_effect_id": effect_id,
-                   "request_sha256": request_sha256}, stream)
+        json.dump(value, stream)
         stream.flush()
         os.fsync(stream.fileno())
-    for parent in (attempts, attempts.parent, attempts.parent.parent):
+    for parent in (path.parent, path.parent.parent, path.parent.parent.parent):
         directory = os.open(parent, os.O_RDONLY)
         try:
             os.fsync(directory)
@@ -150,6 +139,45 @@ def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_s
     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.
 
@@ -360,7 +388,8 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *,
             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):
+            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.
diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py
index e958e974f2..c86180f2c6 100644
--- a/benchmark/tests/test_checkpoint_handoff.py
+++ b/benchmark/tests/test_checkpoint_handoff.py
@@ -76,6 +76,54 @@ def unavailable(*args):
     assert attempts == [True] and marker.read_bytes() == original
 
 
+@pytest.mark.skipif(os.name != "posix" or (hasattr(os, "geteuid") and os.geteuid() == 0),
+                   reason="Requires real unprivileged directory permission denial")
+def test_completed_attempt_survives_artifact_index_permission_recovery(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from benchmark.runtime import checkpoint
+
+    declaration = tmp_path / "request.json"
+    destination = tmp_path / "checkpoints" / "point"
+    index = destination.parent / ".completed-attempts"
+    index.mkdir(parents=True)
+    declaration.write_text(json.dumps({"destination": str(destination), "point_metadata": {},
+        "components": [{"id": "workspace", "source_path": str(tmp_path / "workspace")}],
+        "controller_handoff": True, "native_session": True, "loopx_state": True, "evaluation_opportunity": True}))
+    env = {"LOOPX_CHECKPOINT_REQUEST": str(declaration), "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"),
+        "LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), "LOOPX_REGISTRY": str(tmp_path / "registry"),
+        "LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent"}
+    monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_gate",
+                        lambda *a, **k: {"capture_allowed": True})
+    monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", lambda *a, **k:
+        SimpleNamespace(progress={"state": "settled"},
+                        identity=SimpleNamespace(value=SimpleNamespace(effect_id=k["turn_instance_id"]))))
+    monkeypatch.setattr(checkpoint, "research_participants",
+                        lambda *a: pytest.fail("A missed original point must not read later source state"))
+    first = {"ok": True, "stage": "execute", "mode": "heartbeat", "turn_id": "original"}
+    index.chmod(0o500)
+    try:
+        checkpoint.capture_after_wake(env, first)
+    finally:
+        index.chmod(0o700)
+    assert first["checkpoint"]["reason"] == "capture_failed"
+    assert first["checkpoint"]["error_kind"] == "PermissionError" and not list(index.iterdir())
+    witness = tmp_path / "wakes/original/checkpoint-completed-request.json"
+    original = witness.read_bytes()
+    for turn, changed_destination in (("original", True), ("later-work", False)):
+        if changed_destination:
+            request = json.loads(declaration.read_text())
+            request["destination"] = str(destination.with_name("another-point"))
+            declaration.write_text(json.dumps(request))
+        else:
+            request["destination"] = str(destination)
+            declaration.write_text(json.dumps(request))
+        later = {**first, "turn_id": turn}
+        checkpoint.capture_after_wake(env, later)
+        assert later["checkpoint"]["reason"] == "destination_already_attempted_no_replay"
+        assert later["ok"] and not later["checkpoint"]["captured"]
+    assert not list(index.iterdir()) and witness.read_bytes() == original
+
+
 @pytest.mark.parametrize("response", ["captured", "failed", "foreign", "timeout"])
 def test_original_request_only_and_cleanup_after_missed_point(tmp_path, response):
     mailbox = tmp_path / "checkpoint-controller"
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
index d204e101c3..4ef8efb9b5 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.md
@@ -972,7 +972,9 @@ allowance after writer release. Ordinary file-only captures keep their 30-second
 window. A missed point continues the source and does not authorize a signal-based
 handoff or relabel a later state. The Host retains the first attempt per declared
 destination, even on failure or a lost reply; replay cannot replace its original
-settled source, and an exhausted phase performs no declaration/source reads.
+settled source. The original wake retains its declaration before the optional
+artifact-side index is written; index permission recovery cannot admit later
+state at the missed point. An exhausted phase performs no declaration/source reads.
 The joint native qualification has an explicit
 `settled_work` variant; deterministic inference qualifies mechanical adoption,
 not a scientific advantage or live rollout.
diff --git a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
index 7773d40c53..a4d73053ad 100644
--- a/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
+++ b/docs/architecture/rfcs/hierarchical-agent-stride-control-v0.zh-CN.md
@@ -780,7 +780,8 @@ replan 身份、已完成 Host cursor、来源 revision 及当前策略版本；
 来源 phase 上限约束，释放 writer 后再使用既有有界 sealing 余量。普通文件捕获
 保持 30 秒窗口。错过研究点时继续来源工作，不授权通过信号交接或把更晚状态
 改称原研究点。Host 保留每个声明 destination 的首次尝试，包括失败或回执丢失；
-重试不能替换原已结算来源，原 phase 耗尽时不读取声明或来源。联合原生资格化
+重试不能替换原已结算来源。原 wake 先保留不可变声明见证，再写可选的 artifact
+侧索引；索引权限恢复不能让更晚状态成为原缺失研究点。原 phase 耗尽时不读取声明或来源。联合原生资格化
 提供显式 `settled_work` 变体；确定性推理只验证
 机械采用，不证明科学收益或允许在线推广。