Reuse the original Codex claim when readiness is interrupted

docs/reference/required-work-context.md

@@ -415,7 +415,21 @@ After a Host claim, the ready guard uses the private
 returns its writer digest with the fields needed to verify the same Todo and
 Turn. The hook verifies that digest when it opens the original file. If the
 file changed, the receipt preserves the claim and pins recovery to the original
-bytes before work can continue.
+bytes before work can continue. If readiness is interrupted, retrying the
+same original decision capture reuses its managed claim operation. The
+authority owner checks the original lease on replay. The native session
+keeps the latest eligible live guard digest separately from the capture.
+Presenting the exact saved decision preserves 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.
+
+中文：Host 重试绑定原始决策 capture 的领取操作，由权威 owner 验证原租约。
+原生会话独立保存最新合格准入的摘要。重新展示逐字节一致的快照会保留绑定。
+旧快照和字节不同的副本仍可读，保留已有领取回执。
+仍需准备时，用原 Goal、Agent、Todo 和 Turn 运行 `quota should-run`，
+生成新的 capture，再按当前准入结果准备。已退役的租约需要新的合格 capture。
 
 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.py

@@ -386,8 +386,9 @@ def _emit_quota_result(
         _export("attach_host_poll_receipt")(
             context.status_payload, args, payload, registry_path=registry_path,
         )
+    native_context = context_delivery_enabled and bool(getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None))
     try:
-        captured_digest = capture_decision(capture_directory, payload)
+        captured_digest = capture_decision(capture_directory, payload, args=args, native_context=native_context)
         if getattr(args, "prepared_capture_reference", False) and payload.get("ok") is True:
             assert captured_digest is not None  # Preflight requires an explicit capture directory.
             from ..control_plane.turn_driver.codex_execution_preparation import prepared_capture_reference
@@ -432,7 +433,6 @@ def _emit_quota_result(
     # Handoff releases the immutable original before any presentation. Native
     # actual-input observation owns its projection when explicitly selected;
     # never wrap one context-delivery wire in a second projection.
-    native_context = context_delivery_enabled and bool(getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None))
     deferred_context = getattr(args, "defer_context_presentation", False)
     if (not deferred_context and args.format == "json" and succeeded and delivery_request is not None and not native_context
             and context_delivery_enabled):

loopx/cli_commands/quota_capture.py

@@ -1045,13 +1045,20 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None:
     return directory
 
 
-def capture_decision(directory: Path | None, payload: dict[str, object]) -> str | None:
+def capture_decision(directory: Path | None, payload: dict[str, object], *,
+                     args: argparse.Namespace | None = None, native_context: bool = False) -> str | None:
     if directory is None:
         return
     from ..control_plane.turn_driver.decision_capture import write_decision_capture
 
     try:
-        return write_decision_capture(directory / "decision.json", payload, publish=os.replace)
+        digest = write_decision_capture(directory / "decision.json", payload, publish=os.replace)
+        if native_context and args is not None and args.quota_command == "should-run" and payload.get("ok") is True:
+            from ..control_plane.turn_driver.codex_context import record_native_claim_source
+
+            record_native_claim_source(payload, digest, directory=args.codex_context_dir,
+                context_root=getattr(args, "codex_context_root", None))
+        return digest
     except (OSError, ValueError) as exc:
         raise QuotaDecisionTransportError(
             "Decision capture failed after guard evaluation; the Turn receipt may "
@@ -1373,6 +1380,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
             record_native_execution_preparation,
         )
         from ..control_plane.turn_driver.codex_execution_preparation import (
+            LegacyNativePreparationCapture,
             codex_execution_preparation_context,
             execute_codex_execution_preparation,
         )
@@ -1391,11 +1399,14 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                 directory = codex_context_directory(directory=args.codex_context_dir,
                     root=getattr(args, "codex_context_root", None), session=event.get("session_id"), create=True)
                 payload = codex_context_hook(directory, event)
+                preparation_context = ""
                 try:
                     request = native_execution_preparation_request(directory, event)
                     if request is not None:
                         workspace = Path(event.get("cwd") or "")
                         if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve():
+                            if "claim_source_sha256" not in request:
+                                raise LegacyNativePreparationCapture()
                             capture = (
                                 Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready"
                                 if request.get("turn_instance_id") else None
@@ -1434,24 +1445,21 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
 
                                     receipt["context"]["recovery_command"] = context_capture_read_command(
                                         source, digest, reader_argv=current_loopx_cli_argv())
-                            existing = payload.get("hookSpecificOutput")
-                            existing_context = (
-                                str(existing.get("additionalContext") or "")
-                                if isinstance(existing, dict)
-                                else ""
-                            )
-                            payload["hookSpecificOutput"] = {
-                                "hookEventName": event["hook_event_name"],
-                                "additionalContext": " ".join(filter(None, [
-                                    existing_context,
-                                    codex_execution_preparation_context(receipt),
-                                ])),
-                            }
+                            preparation_context = codex_execution_preparation_context(receipt)
+                except LegacyNativePreparationCapture as error:
+                    preparation_context = str(error)
                 except (OSError, RuntimeError, subprocess.SubprocessError, ValueError, TypeError, RecursionError):
                     # Preserve the original explicit action and any context
                     # recovery output when Host-side preparation cannot prove
                     # its postcondition.
                     pass
+                if preparation_context:
+                    existing = payload.get("hookSpecificOutput")
+                    existing_context = str(existing.get("additionalContext") or "") if isinstance(existing, dict) else ""
+                    payload["hookSpecificOutput"] = {
+                        "hookEventName": event["hook_event_name"],
+                        "additionalContext": " ".join(filter(None, [existing_context, preparation_context])),
+                    }
             else:
                 payload = {}
         except (OSError, RuntimeError, subprocess.SubprocessError, ValueError, TypeError, RecursionError):

loopx/control_plane/turn_driver/codex_context.py

@@ -210,6 +210,46 @@ def _save(root: Path, state: dict[str, Any]) -> None:
     write_decision_capture(root / "session.json", state, publish=os.replace)
 
 
+def record_native_claim_source(
+    payload: Mapping[str, Any], digest: str, *, directory: str | None, context_root: str | None,
+) -> None:
+    """Remember the live guard's latest eligible capture outside its JSON."""
+    preparation = compact_execution_preparation_request(payload)
+    session = os.environ.get("CODEX_THREAD_ID") or None
+    if not preparation or not preparation.get("turn_instance_id") or (not session and directory is None):
+        return
+    location = codex_context_directory(directory=directory, root=context_root, session=session, create=True)
+    if location is None:
+        return
+    with _locked(location) as root:
+        active_session = _state(root).get("session_id")
+        if session and active_session not in (None, session):
+            return
+        # A caller outside the Host can explicitly select its private directory.
+        # The journal identifies that session even without an ambient thread ID.
+        write_decision_capture(root / "claim-source.json",
+            {"session_id": session or active_session, "sha256": digest}, publish=os.replace)
+
+
+def _original_native_claim_source(
+    root: Path, session: str, digest: object, *, bind_unattached: bool = False,
+) -> bool:
+    if not isinstance(digest, str):
+        return False
+    try:
+        path = root / "claim-source.json"
+        saved = _read(path)
+    except (OSError, ValueError):
+        return False
+    bound = {"session_id": session, "sha256": digest}
+    if bind_unattached and saved == {**bound, "session_id": None}:
+        # An explicit private directory can receive a guard before startup.
+        # Adopt it once from the live session while holding its journal lock.
+        write_decision_capture(path, bound, publish=os.replace)
+        return True
+    return saved == bound
+
+
 def read_codex_context_reference(directory: str, *, goal_id: str | None, agent_id: str | None) -> tuple[str, str]:
     """Select an original capture after a lost hook; never assert retention.
 
@@ -421,7 +461,10 @@ def native_execution_preparation_request(
                 and previous.get("state") == "ready"
             ):
                 return None
-            return {**request, "source_sha256": source_sha256}
+            action = {**request, "source_sha256": source_sha256}
+            if not _original_native_claim_source(root, session, action.get("claim_source_sha256")):
+                action.pop("claim_source_sha256", None)
+            return action
     except (OSError, ValueError, TypeError, RecursionError):
         return None
 
@@ -673,11 +716,13 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 state["source_recovery"] = {key: recovery[key] for key in ("path", "sha256")}
                 state.pop("section_reads", None)  # partial reads of an older source are not current coverage
                 page_bytes, output_bytes = native_context_page_budget()
+                preparation = compact_execution_preparation_request(payload)
+                if preparation and _original_native_claim_source(root, session, capture_id, bind_unattached=True):
+                    preparation = compact_execution_preparation_request(payload, claim_source_sha256=capture_id)
 
                 def publish_recovery(reference: dict[str, Any]) -> None:
                     reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"],
                         page_bytes=page_bytes, max_page_output_bytes=output_bytes)
-                    preparation = compact_execution_preparation_request(payload)
                     state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")},
                         "goal_id": goal_id, "agent_id": agent_id,
                         "total_bytes": Path(reference["path"]).stat().st_size,

loopx/control_plane/turn_driver/codex_execution_preparation.py

@@ -13,13 +13,25 @@
 from typing import Any
 
 from ...command_invocation import current_loopx_cli_argv
+from ..content_digest import BARE_SHA256_PATTERN
 
 
 CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
 
 
+class LegacyNativePreparationCapture(ValueError):
+    """The capture cannot prove a live guard bound its claim."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            "This saved capture has no verified live guard binding for Host claims. Preserve any existing claim receipt. "
+            "If preparation is still needed, run the current quota should-run for the original Goal, "
+            "Agent, Todo and Turn with a fresh capture and follow its preparation only if eligible."
+        )
+
+
 def compact_execution_preparation_request(
-    payload: Mapping[str, Any],
+    payload: Mapping[str, Any], *, claim_source_sha256: str | None = None,
 ) -> dict[str, Any] | None:
     """Retain only the exact Host-executable action beside a private capture."""
 
@@ -62,6 +74,11 @@ def compact_execution_preparation_request(
         if isinstance(receipt, Mapping)
         else ""
     )
+    claim_binding = {}
+    if turn_instance_id and claim_source_sha256 is not None:
+        operation_id = f"todo-claim:codex-capture:{claim_source_sha256}"
+        expected_claim += ["--claim-operation-id", operation_id]
+        claim_binding["claim_source_sha256"] = claim_source_sha256
     return {
         "goal_id": goal_id,
         "agent_id": agent_id,
@@ -75,6 +92,7 @@ def compact_execution_preparation_request(
             "runtime_root": str(runtime_root),
         },
         **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}),
+        **claim_binding,
     }
 
 
@@ -173,7 +191,6 @@ def execute_codex_execution_preparation(
         not goal_id
         or not agent_id
         or not todo_id
-        or request.get("claim_args") != expected_claim
         or not registry_path.is_absolute()
         or registry_path.resolve() != registry_path
         or not runtime_root.is_absolute()
@@ -189,8 +206,17 @@ def execute_codex_execution_preparation(
     ):
         raise ValueError("Codex execution preparation requires a physical workspace")
     invoke = runner or _default_runner
-    if context_directory is not None and (capture_directory is None or not turn_instance_id):
-        raise ValueError("Native execution preparation requires a bound captured Turn")
+    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]
+    if request.get("claim_args") != expected_claim:
+        raise ValueError("invalid Codex execution-preparation claim arguments")
     claimed = _run_json(
         expected_claim,
         registry_path=registry_path,
@@ -199,6 +225,9 @@ def execute_codex_execution_preparation(
         timeout=timeout / 2,
         runner=invoke,
     )
+    claim_reference = claimed.get("claim_operation_id")
+    if claim_operation_id is not None and 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,
@@ -259,7 +288,6 @@ def execute_codex_execution_preparation(
         if not isinstance(digest, str) or len(digest) != 64:
             raise ValueError("The guard did not return its published capture digest")
         receipt["capture_sha256"] = digest
-    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

tests/architecture/test_content_digest_single_owner.py

@@ -164,6 +164,7 @@
     "loopx.capabilities.benchmark_toolkit.trajectory_assessment_store",
     "loopx.capabilities.content_ops.item_lifecycle",
     "loopx.capabilities.deep_research.runtime",
+    "loopx.capabilities.issue_fix.cli",
     "loopx.capabilities.issue_fix.outcome_projection",
     "loopx.capabilities.issue_fix.reviewer_notification",
     "loopx.capabilities.machine_configuration.store",
@@ -207,6 +208,7 @@
     "loopx.control_plane.todos.completion_validation",
     "loopx.control_plane.todos.completion_validation_store",
     "loopx.control_plane.todos.machine_section_projection",
+    "loopx.control_plane.turn_driver.codex_execution_preparation",
     "loopx.control_plane.turn_driver.codex_sessions",
     "loopx.control_plane.turn_driver.decision_capture",
     "loopx.control_plane.turn_driver.operation_requests",