fix(codex): return the captured ready context and original claim reference

docs/architecture/rfcs/controller-executor-unified-contract-v0.md

@@ -121,6 +121,18 @@ intentionally have no LoopX Core and must remain uncontaminated controls.
 
 ### C1 source audit: shared owners and the remaining split
 
+Bound native execution preparation retains the ordinary managed claim reference
+and captures its same-Turn ready decision before CLI compaction. The Host hands
+that source to the existing saved-capture presentation/reader owner, rather than
+discarding it and making the model evaluate another guard to obtain its current
+contract. Goal, Agent, Todo and Turn must match before publication. Required
+material reads, including an explicit complete Goal read, remain required; a
+ready receipt or saved file does not fulfill them. Optional presentation failure
+retains the original digest-bound recovery. Explicit off is inert, and an
+unbound request keeps its original readiness route without inventing a Turn.
+This is a bounded Host handoff repair, not C1 cutover, installed adoption or
+qualification of model cost and long-horizon benefit.
+
 The source already shares more than the output names suggest. `_emit_quota_result`
 captures the complete decision **before** `_project_quota_cli_payload` chooses
 ordinary CLI compaction or `buildTurnEnvelope`. The latter delegates to the

docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md

@@ -96,6 +96,14 @@
 
 ### C1 源码核对：已有共享 owner 与剩余分叉
 
+绑定 Turn 的原生执行准备保留普通 managed claim 原引用，并在 CLI 压缩前捕获同一
+Turn 的 ready 决策。Host 将该原件交给既有 saved-capture 展示/reader owner，不丢弃
+它再要求模型重算 guard 才能取得当前契约。发布前须核对 Goal、Agent、Todo 和 Turn。
+必需材料读取（包括显式的完整 Goal 读取）仍须执行，ready 回执或已保存文件不能代替
+读取。可选展示失败保留原摘要绑定的恢复引用；显式 off 不产生动作，未绑定请求沿原
+readiness 路径，不补造 Turn。这只是有界的 Host 交接修复，不代表 C1 切换、已安装
+采用，或模型费用与长程收益已取得资格。
+
 源码实际共享的部分比输出名称体现的更多。`_emit_quota_result` 在
 `_project_quota_cli_payload` 选择普通 CLI 压缩或 `buildTurnEnvelope` **之前**，
 已经捕获完整决策。后者委托 TypeScript owner，Python 模块是传输 adapter。

loopx/cli_commands/quota_capture.py

@@ -1154,17 +1154,37 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                     if request is not None:
                         workspace = Path(event.get("cwd") or "")
                         if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve():
+                            capture = (
+                                Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready"
+                                if request.get("turn_instance_id") else None
+                            )
                             receipt = execute_codex_execution_preparation(
                                 request,
                                 cwd=workspace.resolve(),
                                 scenario=activation["scenario"],
+                                capture_directory=capture,
                             )
                             record_native_execution_preparation(
                                 directory,
                                 event,
                                 source_sha256=request["source_sha256"],
                                 receipt=receipt,
                             )
+                            # Publish the decision that the Host just verified;
+                            # a model must not need a second guard to obtain its
+                            # Goal, task, requirements and settlement contract.
+                            if capture is not None:
+                                source = capture / "decision.json"
+                                # The existing saved-capture owner preserves a
+                                # pinned full recovery even if native projection
+                                # fails. It never rereads authority or admits work.
+                                receipt["context"] = project_saved_context_capture(argparse.Namespace(**{
+                                    **vars(args), "quota_command": "context-project",
+                                    "capture_file": str(source),
+                                    "capture_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
+                                    "goal_id": request["goal_id"], "agent_id": request["agent_id"],
+                                    "turn_instance_id": request["turn_instance_id"],
+                                }))
                             existing = payload.get("hookSpecificOutput")
                             existing_context = (
                                 str(existing.get("additionalContext") or "")

loopx/control_plane/turn_driver/codex_execution_preparation.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+import hashlib
 import json
 import subprocess
 from collections.abc import Callable, Mapping, Sequence
@@ -124,6 +125,7 @@ def execute_codex_execution_preparation(
     *,
     cwd: Path,
     scenario: str,
+    capture_directory: Path | None = None,
     timeout: float = 8,
     runner: CommandRunner | None = None,
 ) -> dict[str, Any]:
@@ -173,7 +175,7 @@ def execute_codex_execution_preparation(
     ):
         raise ValueError("Codex execution preparation requires a physical workspace")
     invoke = runner or _default_runner
-    _run_json(
+    claimed = _run_json(
         expected_claim,
         registry_path=registry_path,
         runtime_root=runtime_root,
@@ -190,6 +192,7 @@ def execute_codex_execution_preparation(
         "--context-delivery", "off",
         "--todo-id", todo_id,
         *(["--turn-instance-id", turn_instance_id] if turn_instance_id else []),
+        *(["--decision-output-dir", str(capture_directory)] if capture_directory is not None else []),
     ]
     after = _run_json(
         guard,
@@ -199,6 +202,13 @@ def execute_codex_execution_preparation(
         timeout=timeout / 2,
         runner=invoke,
     )
+    if capture_directory is not None:
+        from .decision_capture import read_context_capture_file
+
+        source = capture_directory / "decision.json"
+        _, _, after = read_context_capture_file(
+            str(source), hashlib.sha256(source.read_bytes()).hexdigest(), private=True,
+        )
     preparation = after.get("execution_preparation")
     selected = after.get("selected_todo")
     receipt = after.get("heartbeat_receipt")
@@ -207,20 +217,24 @@ def execute_codex_execution_preparation(
         and receipt.get("turn_instance_id") == turn_instance_id
     )
     ready = (
-        after.get("should_run") is True
+        after.get("ok") is True
+        and after.get("goal_id") == goal_id
+        and after.get("should_run") is True
         and after.get("normal_delivery_allowed") is True
         and isinstance(selected, Mapping)
         and selected.get("todo_id") == todo_id
         and isinstance(preparation, Mapping)
         and preparation.get("state") == "ready"
+        and preparation.get("todo_id") == todo_id
+        and preparation.get("agent_id") == agent_id
         and preparation.get("delivery_allowed") is True
         and same_turn
     )
     if not ready:
         raise RuntimeError(
             "LoopX Host claim completed without a same-Todo ready postcondition"
         )
-    return {
+    receipt = {
         "event": "host_execution_prepared",
         "goal_id": goal_id,
         "agent_id": agent_id,
@@ -229,6 +243,12 @@ def execute_codex_execution_preparation(
         "execution_preparation": dict(preparation),
         "normal_delivery_allowed": True,
     }
+    claim_reference = claimed.get("claim_operation_id")
+    if isinstance(claim_reference, str) and claim_reference:
+        receipt["claim_operation_id"] = claim_reference
+    # The Host already paid for the captured complete decision. Its caller
+    # presents that source through the normal reader instead of admitting again.
+    return receipt
 
 
 def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str:
@@ -239,6 +259,11 @@ def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str:
         "rechecked the same Todo/Turn before this model step. Treat the earlier "
         "claim_required state as superseded by this ready receipt; do not repeat "
         "its claim or guard solely for that lease state. Current gates and later "
-        "settlement checks still apply. Receipt: "
+        "settlement checks still apply. "
+        + ("Read the ready context in this receipt for current work and settlement, "
+           "replacing the earlier pre-claim view. Retain its original claim reference "
+           "for managed progress or completion. A saved context is not proof of "
+           "model consumption. " if "context" in receipt else "")
+        + "Receipt: "
         + json.dumps(dict(receipt), ensure_ascii=False, separators=(",", ":"))
     )

tests/control_plane/test_codex_execution_preparation.py

@@ -78,15 +78,18 @@ def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None:
     assert compact_execution_preparation_request(malformed) is None
 
 
-def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path) -> None:
-    request = compact_execution_preparation_request(_packet(tmp_path))
+@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"])
+@pytest.mark.parametrize("turn", ["turn-example", None])
+def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path, scenario: str, turn: str | None) -> None:
+    request = compact_execution_preparation_request(_packet(tmp_path, turn=turn))
     assert request is not None
     calls: list[list[str]] = []
 
     def runner(command, *, cwd, timeout):
         calls.append(list(command))
         if "should-run" not in command:
-            return _completed({"ok": True, "claimed": True})
+            return _completed({"ok": True, "claimed": True,
+                "claim_operation_id": "todo-claim:goal-example:todo_example:original"})
         return _completed({
             "ok": True,
             "goal_id": GOAL,
@@ -100,31 +103,37 @@ def runner(command, *, cwd, timeout):
                 "agent_id": AGENT,
                 "delivery_allowed": True,
             },
-            "heartbeat_receipt": {"turn_instance_id": "turn-example"},
+            **({"heartbeat_receipt": {"turn_instance_id": turn}} if turn is not None else {}),
         })
 
     receipt = execute_codex_execution_preparation(
         request,
         cwd=tmp_path.resolve(),
-        scenario="codex_app_ssh_goal",
+        scenario=scenario,
         runner=runner,
     )
 
     assert receipt["event"] == "host_execution_prepared"
     assert receipt["normal_delivery_allowed"] is True
+    assert receipt["claim_operation_id"] == "todo-claim:goal-example:todo_example:original"
     assert len(calls) == 2
     for call in calls:
         assert call[call.index("--registry") + 1] == (
             str((tmp_path / "loopx-project-a" / ".loopx" / "registry.json").resolve())
         )
         assert call[call.index("--runtime-root") + 1] == str((tmp_path / "loopx-runtime-a").resolve())
     assert calls[0][-len(request["claim_args"]):] == request["claim_args"]
-    assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", "turn-example"]
+    if turn is not None:
+        assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", turn]
+    else:
+        assert calls[1][-2:] == ["--todo-id", TODO]
+        assert "--turn-instance-id" not in calls[1] and "turn_instance_id" not in receipt
     context = codex_execution_preparation_context(receipt)
     assert "do not repeat" in context and '"state":"ready"' in context
 
 
-def test_executor_rejects_changed_todo_or_turn(tmp_path: Path) -> None:
+@pytest.mark.parametrize("changed", ["goal", "agent", "todo", "turn"])
+def test_executor_rejects_changed_todo_or_turn(tmp_path: Path, changed: str) -> None:
     request = compact_execution_preparation_request(_packet(tmp_path))
     assert request is not None
 
@@ -133,15 +142,18 @@ def runner(command, *, cwd, timeout):
             return _completed({"ok": True})
         return _completed({
             "ok": True,
+            "goal_id": "goal-other" if changed == "goal" else GOAL,
             "should_run": True,
             "normal_delivery_allowed": True,
-            "selected_todo": {"todo_id": "todo-other"},
+            "selected_todo": {"todo_id": "todo-other" if changed == "todo" else TODO},
             "execution_preparation": {
                 "schema_version": "execution_preparation_v0",
                 "state": "ready",
+                "todo_id": TODO,
+                "agent_id": "agent-other" if changed == "agent" else AGENT,
                 "delivery_allowed": True,
             },
-            "heartbeat_receipt": {"turn_instance_id": "turn-other"},
+            "heartbeat_receipt": {"turn_instance_id": "turn-other" if changed == "turn" else "turn-example"},
         })
 
     with pytest.raises(RuntimeError, match="same-Todo ready"):

tests/control_plane/test_quota_execution_preparation.py

@@ -4,6 +4,7 @@
 import json
 import os
 from pathlib import Path
+import shlex
 import subprocess
 import sys
 
@@ -174,6 +175,106 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard(
     assert after["normal_delivery_allowed"] is True
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"])
+def test_native_hook_returns_readable_ready_source_and_original_claim(
+    tmp_path, monkeypatch, provider, scenario,
+):
+    from loopx.control_plane.turn_driver import codex_context
+
+    tmp_path = tmp_path.resolve()
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("CODEX_THREAD_ID", "preparation-session")
+    native = tmp_path / "native"
+    native.mkdir(mode=0o700)
+    event = {"session_id": "preparation-session", "cwd": str(tmp_path),
+        "hook_event_name": "SessionStart", "source": "startup"}
+    assert codex_context.codex_context_hook(str(native), event) == {}
+    original = {**_todo(), "text": "Deliver the complete synthetic request: 保留原文" + "。不可截断的验收条件" * 100,
+        "note": "Preserve the original stop and once-only settlement."}
+    before = _run(tmp_path, monkeypatch, provider, [], "--runtime-profile", scenario,
+        "--turn-instance-id", "preparation-turn", "--todo-id", TODO, todos=[original])
+    assert before["execution_preparation"]["state"] == "claim_required"
+    goal_file = tmp_path / "state.md"
+    goal_text = goal_file.read_text() + "\n\n## Constraints\n\n" + "Preserve the complete Goal and its stop. 保留停止条件。\n" * 100
+    goal_file.write_text(goal_text)
+    capture = tmp_path / "before-claim"
+    capture.mkdir(mode=0o700)
+    codex_context.project_codex_context(before, directory=str(native), capture_directory=capture,
+        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="before-claim")
+    hook = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(tmp_path / "registry.json"),
+        "--runtime-root", str(tmp_path / "runtime"), "--format", "json", "quota", "context-hook",
+        "--context-delivery", "on", "--runtime-profile", scenario, "--codex-context-dir", str(native)]
+    hook_event = {**event, "hook_event_name": "PostToolUse"}
+    log = tmp_path / "runtime" / "goals" / GOAL / "rollout-event-log.jsonl"
+    unchanged = log.read_bytes()
+    disabled = subprocess.run(["off" if item == "on" else item for item in hook],
+        input=json.dumps(hook_event), cwd=tmp_path, capture_output=True, text=True, timeout=30)
+    assert disabled.returncode == 0 and json.loads(disabled.stdout) == {}
+    assert log.read_bytes() == unchanged
+    completed = subprocess.run(hook, input=json.dumps(hook_event), cwd=tmp_path,
+        capture_output=True, text=True, timeout=30)
+    assert completed.returncode == 0, completed.stderr
+    additional = json.loads(completed.stdout)["hookSpecificOutput"]["additionalContext"]
+    receipt = json.loads(additional.rsplit("Receipt: ", 1)[1])
+    assert receipt["event"] == "host_execution_prepared"
+    assert receipt["todo_id"] == TODO and receipt["turn_instance_id"] == "preparation-turn"
+    assert receipt["execution_preparation"]["state"] == "ready"
+    claim_reference = receipt["claim_operation_id"]
+    assert claim_reference.startswith("todo-claim:")
+    ready = receipt["context"]
+    read = shlex.split(ready["recovery"]["command"])
+    pointers = ["/execution_preparation", "/selected_todo", "/heartbeat_receipt", "/interaction_contract/cli_channel",
+        "/interaction_contract/agent_channel/work_context", "/interaction_contract/agent_channel/required_reads"]
+    for pointer in pointers:
+        read.extend(("--select", pointer))
+    completed = subprocess.run(read, cwd=tmp_path, capture_output=True, text=True, timeout=30)
+    assert completed.returncode == 0, completed.stderr
+    page = json.loads(completed.stdout)
+    assert page["eof"] is True and page["offset"] == 0
+    selected = {item["pointer"]: item["value"] for item in json.loads(page["content"])["selections"]}
+    assert selected["/execution_preparation"]["state"] == "ready"
+    assert selected["/heartbeat_receipt"]["turn_instance_id"] == "preparation-turn"
+    plan = selected["/interaction_contract/cli_channel"]["settlement_plan"]
+    assert plan["identity"]["todo_id"] == TODO and plan["identity"]["turn_instance_id"] == "preparation-turn"
+    sources = selected["/interaction_contract/agent_channel/work_context"]["sources"]
+    task = next(item["content"]["todo"] for item in sources if item["source"] == "selected_todo")
+    # The v0 owner may deduplicate short text already in selected_todo.
+    # The complete request must survive either original placement, without
+    # replacing it with the selected summary when that summary is shorter.
+    assert task.get("text", selected["/selected_todo"]["text"]) == original["text"]
+    assert task["note"] == original["note"]
+    goal_read = next(item for item in selected["/interaction_contract/agent_channel/required_reads"]
+        if item["source"] == "goal_state")
+    goal = subprocess.run(shlex.split(goal_read["command"]), capture_output=True, text=True, timeout=30)
+    assert goal.returncode == 0 and goal.stdout == goal_text
+    code, current = run_json_cli_result("todo", "list", "--goal-id", GOAL, "--todo-id", TODO,
+        registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", cwd=tmp_path)
+    assert code == 0
+    code, progress = run_json_cli_result("todo", "update", "--goal-id", GOAL, "--todo-id", TODO,
+        "--agent-id", AGENT, "--claim-operation-id", claim_reference,
+        "--update-operation-id", "prepared-progress",
+        "--update-expected-provider-revision", current["authority_read"]["provider_revision"],
+        "--note", original["note"] + " Checkpoint verified.",
+        registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", cwd=tmp_path)
+    assert code == 0 and progress["ok"], progress
+    events = log.read_bytes()
+    repeat = subprocess.run(hook, input=json.dumps(hook_event), cwd=tmp_path,
+        capture_output=True, text=True, timeout=30)
+    assert repeat.returncode == 0 and "host_execution_prepared" not in repeat.stdout
+    assert log.read_bytes() == events
+    guards = [json.loads(line) for line in events.splitlines()
+        if json.loads(line)["event_kind"] == "quota_should_run"]
+    assert len(guards) == 2
+    assert [item["details"]["delivery_allowed"] for item in guards] == [False, True]
+    source, _ = codex_context.read_codex_context_reference(str(native), goal_id=GOAL, agent_id=AGENT)
+    source = Path(source)
+    source.write_bytes(source.read_bytes() + b" ")
+    corrupted = subprocess.run(read, cwd=tmp_path, capture_output=True, text=True, timeout=30)
+    assert corrupted.returncode != 0 and json.loads(corrupted.stdout)["ok"] is False
+    assert log.read_bytes() == events
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("multiple_candidates", [False, True])
 def test_projected_claim_prepares_the_same_receipt_bound_turn(