Retire unbound native execution preparation

docs/reference/required-work-context.md

@@ -442,6 +442,9 @@ that binding. Older captures and copies with different bytes remain readable.
 Preserve any original claim receipt and use `quota should-run` with the same
 Goal, Agent, Todo and Turn and a fresh capture if preparation is still
 required. A retired lease requires a fresh eligible preparation capture.
+The native hook executes only a preparation bound to a captured Turn. An
+ordinary claim recommendation remains available to its caller; reading one
+or reading an original receipt does not trigger an automatic claim.
 
 中文：私有原生日志在对绑定 capture 发起领取前先登记自动准备，并在投影前记录已校验的
 ready capture。若就绪检查或 Hook 送达中断，Hook 可重复提供原领取回执、当前租约
@@ -452,6 +455,8 @@ ready capture。若就绪检查或 Hook 送达中断，Hook 可重复提供原
 快照会保留绑定。旧快照和字节不同的副本仍可读，保留已有领取回执。仍需准备时，用原
 Goal、Agent、Todo 和 Turn 运行 `quota should-run`，生成新的 capture，再按当前准入
 结果准备。已退役的租约需要新的合格 capture。
+原生 Hook 只执行与已捕获 Turn 绑定的准备。调用方仍可读取普通领取建议；读取建议或原
+领取回执都不会触发自动领取。
 
 The Goal availability adapter still reads and UTF-8-validates the complete file
 and hashes all bytes for its revision. With native context delivery off, only the path and revision of unmarked or mixed sources cross the

loopx/cli_commands/quota_capture.py

@@ -844,21 +844,18 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                     if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve():
                         request = native_execution_preparation_request(directory, event)
                     if request is not None:
-                        if "claim_source_sha256" not in request:
+                        if "claim_source_sha256" not in request or not request.get("turn_instance_id"):
                             raise LegacyNativePreparationCapture()
                         if request.get("readback_only") is True:
                             preparation_context = codex_execution_preparation_readback_context(request) or ""
                         else:
-                            capture = (
-                                Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready"
-                                if request.get("turn_instance_id") else None
-                            )
+                            capture = Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready"
                             receipt = execute_codex_execution_preparation(
                                 request,
                                 cwd=workspace.resolve(),
                                 scenario=activation["scenario"],
                                 capture_directory=capture,
-                                context_directory=directory if capture is not None else None,
+                                context_directory=Path(directory),
                             )
                             record_native_execution_preparation(
                                 directory,
@@ -869,25 +866,22 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                             # 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.
-                                digest = receipt.pop("capture_sha256")
-                                receipt.pop("capture_path", None)
-                                receipt["context"] = project_saved_context_capture(argparse.Namespace(**{
-                                    **vars(args), "quota_command": "context-project",
-                                    "capture_file": str(source),
-                                    "capture_sha256": digest,
-                                    "goal_id": request["goal_id"], "agent_id": request["agent_id"],
-                                    "turn_instance_id": request["turn_instance_id"],
-                                }), hook_session=event["session_id"])
-                                if receipt["context"].get("ok") is False:
-                                    from ..command_invocation import current_loopx_cli_argv
-
-                                    receipt["context"]["recovery_command"] = context_capture_read_command(
-                                        source, digest, reader_argv=current_loopx_cli_argv())
+                            source = capture / "decision.json"
+                            # The saved-capture reader never rereads authority or admits work.
+                            digest = receipt.pop("capture_sha256")
+                            receipt.pop("capture_path")
+                            receipt["context"] = project_saved_context_capture(argparse.Namespace(**{
+                                **vars(args), "quota_command": "context-project",
+                                "capture_file": str(source),
+                                "capture_sha256": digest,
+                                "goal_id": request["goal_id"], "agent_id": request["agent_id"],
+                                "turn_instance_id": request["turn_instance_id"],
+                            }), hook_session=event["session_id"])
+                            if receipt["context"].get("ok") is False:
+                                from ..command_invocation import current_loopx_cli_argv
+
+                                receipt["context"]["recovery_command"] = context_capture_read_command(
+                                    source, digest, reader_argv=current_loopx_cli_argv())
                             preparation_context = codex_execution_preparation_context(receipt)
                 except LegacyNativePreparationCapture as error:
                     preparation_context = str(error)

loopx/control_plane/turn_driver/codex_execution_preparation.py

@@ -2,7 +2,7 @@
 
 The TypeScript control plane remains the decision owner.  This adapter accepts
 only its exact ordinary claim, executes it through the public CLI, and proves
-the same Todo (and Turn when present) is ready before returning a Host receipt.
+the captured Todo and Turn are ready before returning a Host receipt.
 """
 from __future__ import annotations
 
@@ -223,12 +223,12 @@ def execute_codex_execution_preparation(
     *,
     cwd: Path,
     scenario: str,
-    capture_directory: Path | None = None,
-    context_directory: Path | None = None,
+    capture_directory: Path,
+    context_directory: Path,
     timeout: float = 8,
     runner: CommandRunner | None = None,
 ) -> dict[str, Any]:
-    """Execute one projected claim and prove same-identity readiness.
+    """Execute one journal-bound claim and publish its ready capture.
 
     The hook never selects work.  Unknown/malformed requests fail before a
     subprocess.  The ordinary claim remains the authority and rechecks all
@@ -272,18 +272,18 @@ def execute_codex_execution_preparation(
         or workspace.resolve() != workspace
     ):
         raise ValueError("Codex execution preparation requires a physical workspace")
-    invoke = runner or _default_runner
-    claim_operation_id = None
-    if context_directory is not None or "source_sha256" in request:
-        if context_directory is None or capture_directory is None or not turn_instance_id:
-            raise ValueError("Native execution preparation requires a bound captured Turn")
-        source_sha256 = request.get("claim_source_sha256")
-        if not isinstance(source_sha256, str) or BARE_SHA256_PATTERN.fullmatch(source_sha256) is None:
-            raise ValueError("Native execution preparation requires its original capture digest")
-        claim_operation_id = f"todo-claim:codex-capture:{source_sha256}"
-        expected_claim += ["--claim-operation-id", claim_operation_id]
+    source_sha256 = request.get("source_sha256")
+    if (
+        context_directory is None or capture_directory is None or not turn_instance_id
+        or request.get("readback_only") is True
+        or not isinstance(source_sha256, str) or BARE_SHA256_PATTERN.fullmatch(source_sha256) is None
+    ):
+        raise ValueError("Native execution preparation requires a bound captured Turn")
+    claim_operation_id = _native_claim_operation_id(request)
+    expected_claim += ["--claim-operation-id", claim_operation_id]
     if request.get("claim_args") != expected_claim:
         raise ValueError("invalid Codex execution-preparation claim arguments")
+    invoke = runner or _default_runner
     claimed = _run_json(
         expected_claim,
         registry_path=registry_path,
@@ -293,20 +293,17 @@ def execute_codex_execution_preparation(
         runner=invoke,
     )
     claim_reference = claimed.get("claim_operation_id")
-    if claim_operation_id is not None and claim_reference != claim_operation_id:
+    if claim_reference != claim_operation_id:
         raise RuntimeError("Native execution preparation did not recover its original managed claim")
     guard = [
         "quota", "should-run",
         "--goal-id", goal_id,
         "--agent-id", agent_id,
         *( ["--codex-app"] if scenario == "codex_app_heartbeat" else
            ["--runtime-profile", "codex_app_ssh_goal"] ),
-        *(["--context-delivery", "on", "--codex-context-dir", str(context_directory),
-           "--defer-context-presentation"] if context_directory is not None else
-          ["--context-delivery", "off"]),
-        "--todo-id", todo_id,
-        *(["--turn-instance-id", turn_instance_id] if turn_instance_id else []),
-        *(["--decision-output-dir", str(capture_directory), "--prepared-capture-reference"] if capture_directory is not None else []),
+        "--context-delivery", "on", "--codex-context-dir", str(context_directory), "--defer-context-presentation",
+        "--todo-id", todo_id, "--turn-instance-id", turn_instance_id,
+        "--decision-output-dir", str(capture_directory), "--prepared-capture-reference",
     ]
     after = _run_json(
         guard,
@@ -319,10 +316,6 @@ def execute_codex_execution_preparation(
     preparation = after.get("execution_preparation")
     selected = after.get("selected_todo")
     receipt = after.get("heartbeat_receipt")
-    same_turn = not turn_instance_id or (
-        isinstance(receipt, Mapping)
-        and receipt.get("turn_instance_id") == turn_instance_id
-    )
     ready = (
         after.get("ok") is True
         and after.get("goal_id") == goal_id
@@ -335,32 +328,29 @@ def execute_codex_execution_preparation(
         and preparation.get("todo_id") == todo_id
         and preparation.get("agent_id") == agent_id
         and preparation.get("delivery_allowed") is True
-        and same_turn
+        and isinstance(receipt, Mapping)
+        and receipt.get("turn_instance_id") == turn_instance_id
     )
     if not ready:
         raise RuntimeError(
             "LoopX Host claim completed without a same-Todo ready postcondition"
         )
-    receipt = {
+    digest = after.get("capture_sha256")
+    if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None:
+        raise ValueError("The guard did not return its published capture digest")
+    # The caller presents this same complete decision without admitting again.
+    return {
         "event": "host_execution_prepared",
         "goal_id": goal_id,
         "agent_id": agent_id,
         "todo_id": todo_id,
-        **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}),
+        "turn_instance_id": turn_instance_id,
         "execution_preparation": dict(preparation),
         "normal_delivery_allowed": True,
+        "capture_sha256": digest,
+        "capture_path": str(capture_directory / "decision.json"),
+        "claim_operation_id": claim_operation_id,
     }
-    if capture_directory is not None:
-        digest = after.get("capture_sha256")
-        if not isinstance(digest, str) or len(digest) != 64:
-            raise ValueError("The guard did not return its published capture digest")
-        receipt["capture_sha256"] = digest
-        receipt["capture_path"] = str(capture_directory / "decision.json")
-    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:

tests/control_plane/test_codex_execution_preparation.py

@@ -16,6 +16,10 @@
 GOAL = "goal-example"
 TODO = "todo_example"
 AGENT = "agent-a"
+ORIGINAL_DIGEST = "a" * 64
+OBSERVED_DIGEST = "b" * 64
+READY_DIGEST = "c" * 64
+CLAIM = f"todo-claim:codex-capture:{ORIGINAL_DIGEST}"
 
 
 def _packet(tmp_path: Path, *, turn: str | None = "turn-example") -> dict[str, object]:
@@ -55,14 +59,15 @@ def _completed(payload: dict[str, object], *, code: int = 0) -> subprocess.Compl
     )
 
 
-def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None:
-    packet = _packet(tmp_path)
+@pytest.mark.parametrize("turn", ["turn-example", None])
+def test_compact_request_accepts_only_exact_host_action(tmp_path: Path, turn: str | None) -> None:
+    packet = _packet(tmp_path, turn=turn)
     request = compact_execution_preparation_request(packet)
     assert request == {
         "goal_id": GOAL,
         "agent_id": AGENT,
         "todo_id": TODO,
-        "turn_instance_id": "turn-example",
+        **({"turn_instance_id": turn} if turn else {}),
         "claim_args": [
             "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO,
             "--agent-id", AGENT, "--claimed-by", AGENT,
@@ -78,18 +83,23 @@ def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None:
     assert compact_execution_preparation_request(malformed) is None
 
 
-@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))
+def _bound_request(tmp_path: Path) -> dict:
+    request = compact_execution_preparation_request(_packet(tmp_path), claim_source_sha256=ORIGINAL_DIGEST)
     assert request is not None
+    return {**request, "source_sha256": OBSERVED_DIGEST}
+
+
+@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"])
+def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path, scenario: str) -> None:
+    request = _bound_request(tmp_path)
+    capture, native = tmp_path / "ready", tmp_path / "native"
     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,
-                "claim_operation_id": "todo-claim:goal-example:todo_example:original"})
+                "claim_operation_id": CLAIM})
         return _completed({
             "ok": True,
             "goal_id": GOAL,
@@ -103,43 +113,49 @@ def runner(command, *, cwd, timeout):
                 "agent_id": AGENT,
                 "delivery_allowed": True,
             },
-            **({"heartbeat_receipt": {"turn_instance_id": turn}} if turn is not None else {}),
+            "heartbeat_receipt": {"turn_instance_id": "turn-example"},
+            "capture_sha256": READY_DIGEST,
         })
 
     receipt = execute_codex_execution_preparation(
         request,
         cwd=tmp_path.resolve(),
         scenario=scenario,
+        capture_directory=capture,
+        context_directory=native,
         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 receipt["claim_operation_id"] == CLAIM
+    assert receipt["turn_instance_id"] == "turn-example"
+    assert receipt["capture_sha256"] == READY_DIGEST
+    assert receipt["capture_path"] == str(capture / "decision.json")
     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"]
-    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
+    assert calls[0][-2:] == ["--claim-operation-id", CLAIM]
+    assert calls[1][-12:] == [
+        "--context-delivery", "on", "--codex-context-dir", str(native), "--defer-context-presentation",
+        "--todo-id", TODO, "--turn-instance-id", "turn-example",
+        "--decision-output-dir", str(capture), "--prepared-capture-reference",
+    ]
     context = codex_execution_preparation_context(receipt)
     assert "do not repeat" in context and '"state":"ready"' in context
 
 
 @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
+    request = _bound_request(tmp_path)
 
     def runner(command, *, cwd, timeout):
         if "should-run" not in command:
-            return _completed({"ok": True})
+            return _completed({"ok": True, "claim_operation_id": CLAIM})
         return _completed({
             "ok": True,
             "goal_id": "goal-other" if changed == "goal" else GOAL,
@@ -161,5 +177,30 @@ def runner(command, *, cwd, timeout):
             request,
             cwd=tmp_path.resolve(),
             scenario="codex_app_ssh_goal",
+            capture_directory=tmp_path / "ready",
+            context_directory=tmp_path / "native",
             runner=runner,
         )
+
+
+@pytest.mark.parametrize("fault", ["unbound", "missing-source", "missing-turn", "readback-only"])
+def test_executor_does_not_run_a_claim_without_live_binding(tmp_path: Path, fault: str) -> None:
+    request = _bound_request(tmp_path)
+    capture, native = tmp_path / "ready", tmp_path / "native"
+    if fault == "unbound":
+        request = compact_execution_preparation_request(_packet(tmp_path, turn=None))
+        assert request is not None
+        capture = native = None
+    elif fault == "missing-source":
+        request.pop("source_sha256")
+    elif fault == "missing-turn":
+        request.pop("turn_instance_id")
+    else:
+        request["readback_only"] = True
+
+    with pytest.raises(ValueError, match="bound captured Turn"):
+        execute_codex_execution_preparation(
+            request, cwd=tmp_path.resolve(), scenario="codex_app_heartbeat",
+            capture_directory=capture, context_directory=native,
+            runner=lambda *_args, **_kwargs: pytest.fail("unbound preparation must not run a subprocess"),
+        )