Preserve the original ready decision across native claims and retries

docs/reference/required-work-context.md

@@ -355,6 +355,13 @@ cannot be combined with socket handoff. Ordinary callers keep their existing
 presentation, and context delivery off keeps its original pending Goal reads.
 Unbound staged replans retain their existing review-owned original Goal read.
 
+After a Host claim, the ready guard uses the private
+`--prepared-capture-reference` route. It publishes the complete decision and
+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.
+
 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
 internal context-projection bridge, preserving the complete progressive read
@@ -881,6 +888,10 @@ Checkpoint Host 预检在显式开启 native JSON `should-run` 捕获时使用
 与 socket handoff 组合。普通调用保持既有呈现；关闭差分时保留原来的待完成 Goal
 读取。未绑定任务的分阶段重规划继续沿用 review 所属的原始 Goal 读取。
 
+Host 领取 Todo 后，就绪检查使用内部 `--prepared-capture-reference` 路径。它保存完整决策，
+只返回写入时的摘要和核对同一 Todo、Turn 所需的字段。hook 打开原文件时校验该摘要；
+若文件已被替换，回执保留原领取记录，恢复原始内容后才能继续工作。
+
 Goal adapter 仍完整读取文件、校验 UTF-8，并以全部字节计算修订摘要；关闭 native
 context delivery 时，未标记或混合归属仅在内部桥接传路径与修订，避免未被消费的正文
 副本，Agent 仍须渐进式全文读取。显式

loopx/cli_commands/quota.py

@@ -387,7 +387,13 @@ def _emit_quota_result(
             context.status_payload, args, payload, registry_path=registry_path,
         )
     try:
-        capture_decision(capture_directory, payload)
+        captured_digest = capture_decision(capture_directory, payload)
+        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
+
+            print_payload(prepared_capture_reference(payload, captured_digest), args.format, _quota_renderer(args))
+            return 0
         if (payload.get("ok") is False and context_delivery_enabled and capture_directory
                 and args.format == "json" and not getattr(args, "defer_context_presentation", False)):
             print_payload(_export("captured_quota_failure_view")(payload, capture_directory=capture_directory,
@@ -442,7 +448,8 @@ def _emit_quota_result(
     if (not deferred_context and args.format == "json" and succeeded and capture_directory
             and native_context):
         try:
-            payload = project_native_quota_context(payload, args, capture_directory=capture_directory,
+            assert captured_digest is not None  # Native publication requires the captured original.
+            payload = project_native_quota_context(payload, args, capture_directory=capture_directory, capture_id=captured_digest,
                 decision_identity=decision_identity, progressive_enabled=activation["progressive_enabled"],
                 reader_argv=reader_argv)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:

loopx/cli_commands/quota_capture.py

@@ -23,7 +23,7 @@
     validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command,
     CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
     read_context_capture_file as _read_capture,
-    read_context_read_reference, write_context_read_reference, private_capture_uid,
+    read_context_read_reference, write_context_read_reference, private_capture_uid, read_turn_observation_text,
 )
 from ..control_plane.turn_driver.native_context_profile import (
     NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, native_context_page_budget,
@@ -551,13 +551,12 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
 
 
 def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespace, *,
-                                 capture_directory: Path, decision_identity: tuple[str, str],
+                                 capture_directory: Path, capture_id: str, decision_identity: tuple[str, str],
                                  progressive_enabled: bool, reader_argv: list[str] | None) -> dict[str, Any]:
     """Prepare one already-decided native view; callers retain receipt failures."""
     from ..control_plane.turn_driver.codex_context import (
         NativeContextSessionUnavailable, codex_context_directory, project_codex_context,
     )
-    capture_id = hashlib.sha256((capture_directory / "decision.json").read_bytes()).hexdigest()
     try:
         directory = codex_context_directory(directory=args.codex_context_dir,
             root=getattr(args, "codex_context_root", None), create=True)
@@ -629,7 +628,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
     reader_argv = current_loopx_cli_argv()
     try:
         activation = native_context_activation(args)
-        path, raw, payload = _read_capture(args.capture_file, args.capture_sha256)
+        path, raw, payload = _read_capture(args.capture_file, args.capture_sha256, private=hook_session is not None)
         identity = payload.get("agent_identity")
         receipt = payload.get("heartbeat_receipt")
         if (payload.get("ok") is not True or not isinstance(identity, dict) or not isinstance(receipt, dict)
@@ -763,6 +762,13 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
 
 def prepare_quota_capture(args: argparse.Namespace, directory: Path | None) -> Path | None:
     """Validate transport options before allocating a new guard capture."""
+    if getattr(args, "prepared_capture_reference", False) and (
+        args.quota_command != "should-run" or args.format != "json"
+        or not getattr(args, "decision_output_dir", None) or not getattr(args, "turn_instance_id", None)
+        or getattr(args, "decision_handoff_socket", None) or getattr(args, "turn_envelope", False)
+        or (native_context_activation(args)["enabled"] and not getattr(args, "defer_context_presentation", False))
+    ):
+        raise QuotaCommandValidationError("prepared capture reference requires an explicit Turn and raw JSON capture")
     if getattr(args, "defer_context_presentation", False) and (
         args.quota_command != "should-run" or args.format != "json"
         or not native_context_activation(args)["enabled"]
@@ -834,13 +840,13 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None:
     return directory
 
 
-def capture_decision(directory: Path | None, payload: dict[str, object]) -> None:
+def capture_decision(directory: Path | None, payload: dict[str, object]) -> str | None:
     if directory is None:
         return
     from ..control_plane.turn_driver.decision_capture import write_decision_capture
 
     try:
-        write_decision_capture(directory / "decision.json", payload, publish=os.replace)
+        return write_decision_capture(directory / "decision.json", payload, publish=os.replace)
     except (OSError, ValueError) as exc:
         raise QuotaDecisionTransportError(
             "Decision capture failed after guard evaluation; the Turn receipt may "
@@ -1167,6 +1173,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                                 cwd=workspace.resolve(),
                                 scenario=activation["scenario"],
                                 capture_directory=capture,
+                                context_directory=directory if capture is not None else None,
                             )
                             record_native_execution_preparation(
                                 directory,
@@ -1182,13 +1189,19 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
                                 # 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["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(),
+                                    "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())
                             existing = payload.get("hookSpecificOutput")
                             existing_context = (
                                 str(existing.get("additionalContext") or "")
@@ -1236,8 +1249,7 @@ def read_replan_view(args: argparse.Namespace) -> dict[str, Any]:
     if not args.decision_file or not args.source_hash:
         raise ValueError("replan-view requires --decision-file and --source-hash")
     path = Path(args.decision_file).expanduser()
-    with path.open(encoding="utf-8") as stream:
-        payload = json.load(stream)
+    payload = json.loads(read_turn_observation_text(path))
     return _turn_envelope_result("replan_view", payload=payload, source_hash=args.source_hash,
                                  captured_decision_path=str(path), reader_argv=current_loopx_cli_argv(),
                                  stage=args.replan_stage, outcome_ref=args.outcome_ref)

loopx/cli_commands/quota_registration.py

@@ -217,6 +217,7 @@ def register_quota_command(
     # Host preflight captures the complete native input, then presents that same
     # immutable view after the model's lifecycle hooks establish continuity.
     quota_parser.add_argument("--defer-context-presentation", action="store_true", help=argparse.SUPPRESS)
+    quota_parser.add_argument("--prepared-capture-reference", action="store_true", help=argparse.SUPPRESS)
     from ..control_plane.turn_driver.codex_context import CONTEXT_DELIVERY_SETTINGS, DELTA_DELIVERY_SETTINGS
 
     quota_parser.add_argument("--context-delivery", choices=CONTEXT_DELIVERY_SETTINGS,

loopx/control_plane/turn_driver/codex_execution_preparation.py

@@ -6,7 +6,6 @@
 """
 from __future__ import annotations
 
-import hashlib
 import json
 import subprocess
 from collections.abc import Callable, Mapping, Sequence
@@ -120,12 +119,27 @@ def _run_json(
     return payload
 
 
+def prepared_capture_reference(payload: Mapping[str, Any], digest: str) -> dict[str, Any]:
+    """Carry only the guard fields the Host uses to verify its claimed Turn."""
+    selected = payload.get("selected_todo")
+    heartbeat = payload.get("heartbeat_receipt")
+    return {
+        **{key: payload.get(key) for key in (
+            "ok", "goal_id", "should_run", "normal_delivery_allowed", "execution_preparation",
+        )},
+        "selected_todo": {"todo_id": selected.get("todo_id")} if isinstance(selected, Mapping) else None,
+        "heartbeat_receipt": {"turn_instance_id": heartbeat.get("turn_instance_id")} if isinstance(heartbeat, Mapping) else None,
+        "capture_sha256": digest,
+    }
+
+
 def execute_codex_execution_preparation(
     request: Mapping[str, Any],
     *,
     cwd: Path,
     scenario: str,
     capture_directory: Path | None = None,
+    context_directory: Path | None = None,
     timeout: float = 8,
     runner: CommandRunner | None = None,
 ) -> dict[str, Any]:
@@ -175,6 +189,8 @@ 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")
     claimed = _run_json(
         expected_claim,
         registry_path=registry_path,
@@ -189,10 +205,12 @@ def execute_codex_execution_preparation(
         "--agent-id", agent_id,
         *( ["--codex-app"] if scenario == "codex_app_heartbeat" else
            ["--runtime-profile", "codex_app_ssh_goal"] ),
-        "--context-delivery", "off",
+        *(["--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)] if capture_directory is not None else []),
+        *(["--decision-output-dir", str(capture_directory), "--prepared-capture-reference"] if capture_directory is not None else []),
     ]
     after = _run_json(
         guard,
@@ -202,13 +220,6 @@ 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")
@@ -243,6 +254,11 @@ def execute_codex_execution_preparation(
         "execution_preparation": dict(preparation),
         "normal_delivery_allowed": True,
     }
+    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
     claim_reference = claimed.get("claim_operation_id")
     if isinstance(claim_reference, str) and claim_reference:
         receipt["claim_operation_id"] = claim_reference
@@ -254,16 +270,24 @@ def execute_codex_execution_preparation(
 def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str:
     """Render the small Host receipt that supersedes an earlier stale guard."""
 
+    context = receipt.get("context")
+    guidance = ""
+    if isinstance(context, Mapping):
+        if context.get("ok") is False:
+            guidance = ("Ready context delivery failed. Restore the private original capture and use its "
+                        "recovery command before work. Retain the original claim reference. ")
+        else:
+            guidance = ("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. ")
     return (
         "LoopX Host automatically restored the exact selected Todo lease and "
         "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. "
-        + ("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 "")
+        + guidance
         + "Receipt: "
         + json.dumps(dict(receipt), ensure_ascii=False, separators=(",", ":"))
     )

loopx/control_plane/turn_driver/codex_input_trace.py

@@ -4,53 +4,21 @@
 import hashlib
 import json
 import os
-import stat
 from collections.abc import Iterator
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
-from .decision_capture import capture_page_range as _capture_page_range
-
-
-def _open_verified(path: Path) -> int:
-    """Open the entry at `path` without following a link, on any host.
-
-    `O_NOFOLLOW` and `O_NONBLOCK` are POSIX-only attributes, so a bare reference
-    raises AttributeError on a host that does not define them -- here, before the
-    bounded-regular-file check below can refuse anything.
-
-    Where the host has no no-follow flag, the open cannot refuse a link for us,
-    and checking the path first would only move the race: the entry can be
-    replaced between the check and the open. Bind the descriptor to the entry
-    instead, and do it before the caller reads one byte: a link opens to its
-    target, whose device and inode differ from the link's own, so a swapped or
-    linked entry is refused rather than followed.
-    """
-    descriptor = os.open(
-        path,
-        os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0),
-    )
-    try:
-        opened = os.fstat(descriptor)
-        entry = os.lstat(path)
-        if stat.S_ISLNK(entry.st_mode):
-            raise ValueError("native observation must not be a symlink")
-        if (entry.st_dev, entry.st_ino) != (opened.st_dev, opened.st_ino):
-            raise ValueError("native observation changed while it was opened")
-    except BaseException:
-        os.close(descriptor)
-        raise
-    return descriptor
+from .decision_capture import capture_page_range as _capture_page_range, open_regular_capture
 
 
 def read_native_trace_file(path: Path, *, tail: bool = False) -> bytes:
     from ..effect_runtime import MAX_REQUEST_BYTES
 
-    descriptor = _open_verified(path)
+    descriptor = open_regular_capture(path)
     with os.fdopen(descriptor, "rb") as stream:
         metadata = os.fstat(stream.fileno())
-        if not stat.S_ISREG(metadata.st_mode) or (metadata.st_size > MAX_REQUEST_BYTES and not tail):
+        if metadata.st_size > MAX_REQUEST_BYTES and not tail:
             raise ValueError("native observation is not a bounded regular file")
         offset = max(0, metadata.st_size - MAX_REQUEST_BYTES) if tail else 0
         stream.seek(offset)