diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 337a2bb461..3fe0c5993d 100644
--- a/docs/reference/required-work-context.md
+++ b/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 仍须渐进式全文读取。显式
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index bce2192865..0cb147ab61 100644
--- a/loopx/cli_commands/quota.py
+++ b/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:
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index dbdabeacb9..682146520f 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/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)
diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py
index f88a0f6435..b404b91c29 100644
--- a/loopx/cli_commands/quota_registration.py
+++ b/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,
diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py
index 34251eef4a..8eb671980a 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/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=(",", ":"))
     )
diff --git a/loopx/control_plane/turn_driver/codex_input_trace.py b/loopx/control_plane/turn_driver/codex_input_trace.py
index 5f211f7541..83d4378014 100644
--- a/loopx/control_plane/turn_driver/codex_input_trace.py
+++ b/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)
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index 7a55c6f84c..1521219608 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -44,6 +44,10 @@ def private_capture_uid() -> int:
     return cast(int, getattr(os, "getuid")())
 
 
+def _decision_json(payload: Mapping[str, Any]) -> str:
+    return json.dumps(dict(payload), ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n"
+
+
 def write_decision_capture(
     path: Path, payload: Mapping[str, Any], *, publish: Callable[[str, Path], None] = os.link,
 ) -> str:
@@ -52,7 +56,7 @@ def write_decision_capture(
     Digest the bytes we publish, rather than reopening the file solely to hash
     it. Independent readers still verify the original bytes on every read.
     """
-    raw = (json.dumps(dict(payload), ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8")
+    raw = _decision_json(payload).encode("utf-8")
     descriptor, temporary = tempfile.mkstemp(prefix=".decision-", dir=path.parent)
     try:
         with os.fdopen(descriptor, "wb") as stream:
@@ -65,6 +69,37 @@ def write_decision_capture(
         Path(temporary).unlink(missing_ok=True)
 
 
+def open_regular_capture(path: Path) -> int:
+    """Bind a regular file to its directory entry before reading.
+
+    Nonblocking open avoids waiting on FIFOs under the journal lock. Checking
+    descriptor identity also rejects substituted entries on hosts without
+    O_NOFOLLOW. Callers retain their own size and privacy checks.
+    """
+    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 not stat.S_ISREG(opened.st_mode):
+            raise ValueError("capture must be a regular file")
+        if stat.S_ISLNK(entry.st_mode):
+            raise ValueError("capture must not be a symlink")
+        if (entry.st_dev, entry.st_ino) != (opened.st_dev, opened.st_ino):
+            raise ValueError("capture changed while it was opened")
+    except BaseException:
+        os.close(descriptor)
+        raise
+    return descriptor
+
+
+def read_turn_observation_text(path: Path) -> str:
+    with os.fdopen(open_regular_capture(path), encoding="utf-8") as source:
+        return source.read()
+
+
 @dataclass(frozen=True)
 class StagedTurnDecisionCapture:
     """Validated private IO, published only after the native start is admitted."""
@@ -80,10 +115,15 @@ def publish(self) -> dict[str, Any]:
         if self.path.exists():
             # Python considers True == 1; original JSON evidence must retain
             # its value types as well as its content. Object key order is IO.
-            current = json.loads(self.path.read_text(encoding="utf-8"))
-            if (json.dumps(current, sort_keys=True, allow_nan=False)
-                    != json.dumps(self.observation, sort_keys=True, allow_nan=False)):
-                raise ValueError("Turn decision capture changed before publication")
+            text = read_turn_observation_text(self.path)
+            # Compact objects start with a quoted key (or are empty). Pretty
+            # captures can go straight to their order-independent comparison.
+            compact = text.startswith('{"') or text == "{}\n"
+            if not compact or text != _decision_json(self.observation):
+                current = json.loads(text)
+                if (json.dumps(current, sort_keys=True, allow_nan=False)
+                        != json.dumps(self.observation, sort_keys=True, allow_nan=False)):
+                    raise ValueError("Turn decision capture changed before publication")
         elif self.existed:
             raise ValueError("Original Turn decision capture is missing; restore the original observation")
         else:
@@ -113,7 +153,7 @@ def stage_turn_decision_capture(
         raise ValueError("Turn decision capture must not be a symlink")
     existed = path.exists()
     if existed:
-        observation = json.loads(path.read_text(encoding="utf-8"))
+        observation = json.loads(read_turn_observation_text(path))
     elif existing_journal or capture:
         raise ValueError("Original Turn decision capture is missing; restore the original "
                          "observation, do not rerun the guard to replace it")
@@ -291,10 +331,10 @@ def read_context_read_reference(value: str) -> tuple[Path, dict[str, Any]]:
     path = path.parent.resolve(strict=True) / path.name
     if path.is_symlink():
         raise ValueError("reader reference must be a physical file, not a symlink")
-    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
+    descriptor = open_regular_capture(path)
     with os.fdopen(descriptor, "rb") as stream:
         info = os.fstat(stream.fileno())
-        if (not stat.S_ISREG(info.st_mode) or info.st_uid != uid or info.st_mode & 0o077
+        if (info.st_uid != uid or info.st_mode & 0o077
                 or info.st_size > MAX_REQUEST_BYTES):
             raise ValueError("reader reference must be a private regular file owned by this user")
         raw = stream.read(MAX_REQUEST_BYTES + 1)
@@ -322,10 +362,10 @@ def read_context_capture_file(value: str, digest: str, *, private: bool = False)
     path = path.parent.resolve(strict=True) / path.name
     if path.is_symlink():
         raise ValueError("context-read requires the physical original capture path, not a symlink")
-    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
+    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:
+        if metadata.st_size > MAX_REQUEST_BYTES:
             raise ValueError("capture must be a regular JSON file within the existing transport size limit")
         if private and (metadata.st_uid != uid or metadata.st_mode & 0o077):
             raise ValueError("native context requires a private capture owned by this user")
@@ -364,8 +404,7 @@ def capture_context_projection(payload: dict[str, Any], *, directory: Path,
     if filename not in ("projection.json", "required-context.json"):
         raise ValueError("unsupported captured context filename")
     projection = directory / filename
-    write_decision_capture(projection, payload)
-    digest = hashlib.sha256(projection.read_bytes()).hexdigest()
+    digest = write_decision_capture(projection, payload)
     return {
         "path": str(projection), "sha256": digest,
         "instruction": "Complete JSON for this captured view. Use it when required content is missing or uncertain; "
diff --git a/tests/control_plane/test_quota_decision_capture.py b/tests/control_plane/test_quota_decision_capture.py
index 5432a77666..0cbf4b7cc4 100644
--- a/tests/control_plane/test_quota_decision_capture.py
+++ b/tests/control_plane/test_quota_decision_capture.py
@@ -25,6 +25,21 @@ def request(path, **overrides):
          'turn_instance_id': TURN_ID}, **overrides))
 
 
+@pytest.mark.parametrize("overrides", [
+    {"turn_instance_id": None}, {"quota_command": "plan"}, {"format": "markdown"},
+    {"turn_envelope": True}, {"decision_handoff_socket": "/unused/socket"},
+    {"context_delivery": "on", "codex_app": True},
+])
+def test_prepared_reference_rejects_incompatible_delivery_before_guard(tmp_path, overrides):
+    from loopx.cli_commands.quota_capture import prepare_quota_capture
+
+    values = {"format": "json", "context_delivery": "off", **overrides}
+    args = request(tmp_path / "capture", prepared_capture_reference=True, **values)
+    with pytest.raises(QuotaCommandValidationError, match="prepared capture reference"):
+        prepare_quota_capture(args, None)
+    assert list(tmp_path.iterdir()) == []
+
+
 def test_disabled_capture_does_not_touch_disk(tmp_path):
     assert prepare_decision_capture(request(None)) is None
     capture_decision(None, {'not_serializable': object()})
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 95fcf3cdf6..7570fd3f26 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -256,6 +256,7 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     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"
+    assert plan["schema_version"] == "quota_settlement_plan_v1"
     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.
@@ -263,10 +264,14 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     # 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
+    goal_reads = [item for item in selected["/interaction_contract/agent_channel/required_reads"]
+        if item["source"] == "goal_state"]
+    if goal_reads:
+        goal = subprocess.run(shlex.split(goal_reads[0]["command"]), capture_output=True, text=True, timeout=30)
+        assert goal.returncode == 0 and goal.stdout == goal_text
+    else:
+        goal = next(item["content"] for item in sources if item["source"] == "goal_state")
+        assert goal["text"] == 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
@@ -294,6 +299,91 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     assert log.read_bytes() == events
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("replace_stage", ["guard", "record"])
+def test_prepared_hook_refuses_a_replaced_ready_capture_without_reclaim(
+    tmp_path, monkeypatch, capsys, provider, replace_stage,
+):
+    import hashlib
+    import io
+    from loopx.cli import main
+    from loopx.control_plane.turn_driver import codex_context
+
+    tmp_path = tmp_path.resolve()
+    monkeypatch.chdir(tmp_path)
+    session = "bound-preparation"
+    monkeypatch.setenv("CODEX_THREAD_ID", session)
+    native = tmp_path / "native"
+    native.mkdir(mode=0o700)
+    event = {"session_id": session, "cwd": str(tmp_path), "hook_event_name": "SessionStart", "source": "startup"}
+    assert codex_context.codex_context_hook(str(native), event) == {}
+    before = _run(tmp_path, monkeypatch, provider, [], "--codex-app",
+        "--turn-instance-id", "prepared-turn", "--todo-id", TODO)
+    assert before["execution_preparation"]["state"] == "claim_required"
+    original = tmp_path / "preparation"
+    original.mkdir(mode=0o700)
+    codex_context.project_codex_context(before, directory=str(native), capture_directory=original,
+        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="claim-required")
+    from loopx.control_plane.turn_driver import codex_execution_preparation
+    record = codex_context.record_native_execution_preparation
+    run = codex_execution_preparation._default_runner
+    verified = {}
+
+    def replace_capture():
+        source, = native.glob("prepared-*/ready/decision.json")
+        verified["path"], verified["bytes"] = source, source.read_bytes()
+        source.write_bytes(verified["bytes"] + b" ")
+
+    def observe_guard(command, **kwargs):
+        result = run(command, **kwargs)
+        if "--prepared-capture-reference" in command:
+            verified["writer"] = json.loads(result.stdout)
+            if replace_stage == "guard":
+                replace_capture()
+        return result
+
+    monkeypatch.setattr(codex_execution_preparation, "_default_runner", observe_guard)
+
+    def replace_after_ready(*args, **kwargs):
+        record(*args, **kwargs)
+        if replace_stage == "record":
+            replace_capture()
+
+    monkeypatch.setattr(codex_context, "record_native_execution_preparation", replace_after_ready)
+    command = ["--registry", str(tmp_path / "registry.json"), "--runtime-root", str(tmp_path / "runtime"),
+               "--format", "json", "quota", "context-hook", "--codex-app", "--context-delivery", "on",
+               "--codex-context-dir", str(native)]
+    def hook():
+        data = json.dumps({**event, "hook_event_name": "PostToolUse"}).encode()
+        monkeypatch.setattr(sys, "stdin", io.TextIOWrapper(io.BytesIO(data)))
+        capsys.readouterr()
+        assert main(command) == 0
+        return json.loads(capsys.readouterr().out)
+
+    result = hook()["hookSpecificOutput"]["additionalContext"]
+    receipt = json.loads(result.rsplit("Receipt: ", 1)[1])
+    assert receipt["claim_operation_id"].startswith("todo-claim:")
+    context = receipt["context"]
+    assert context["ok"] is False
+    assert "digest mismatch" in context["reason"]
+    recovery = shlex.split(context["recovery_command"])
+    digest = hashlib.sha256(verified["bytes"]).hexdigest()
+    writer = verified["writer"]
+    assert writer["capture_sha256"] == digest and writer["execution_preparation"] == receipt["execution_preparation"]
+    assert writer["selected_todo"] == {"todo_id": TODO}
+    assert writer["heartbeat_receipt"] == {"turn_instance_id": "prepared-turn"}
+    assert "interaction_contract" not in writer and "work_context" not in writer
+    assert recovery[recovery.index("--capture-sha256") + 1] == digest
+    log = tmp_path / "runtime" / "goals" / GOAL / "rollout-event-log.jsonl"
+    after_claim = log.read_bytes()
+    assert hook() == {}
+    verified["path"].write_bytes(verified["bytes"])
+    read = subprocess.run([*recovery, "--select", "/execution_preparation/state"], cwd=tmp_path,
+                          check=True, text=True, capture_output=True)
+    assert json.loads(json.loads(read.stdout)["content"])["selections"][0]["value"] == "ready"
+    assert log.read_bytes() == after_claim
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("multiple_candidates", [False, True])
 def test_projected_claim_prepares_the_same_receipt_bound_turn(
diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py
index c5da60a849..323a8d0fa0 100644
--- a/tests/control_plane/test_replan_staged_read.py
+++ b/tests/control_plane/test_replan_staged_read.py
@@ -701,3 +701,31 @@ def test_explore_uses_its_own_incremental_budget_and_real_result_log(tmp_path):
     assert example["explore_review"]["revision"] == basis["revision"]
     assert example["explore_review"]["considered_refs"] == []
     assert json_bytes(writer) <= 16 * 1024
+
+
+def test_replan_reader_rejects_replaced_entries_and_recovers_original(tmp_path):
+    source = _quota_payload()
+    original = tmp_path / "decision.json"
+    original.write_text(json.dumps(source), encoding="utf-8")
+    digest = build_turn_envelope(source)["action_signature"]["source_decision_hash"]
+    command = [str(Path(sys.executable).with_name("loopx")), "--format", "json", "quota", "replan-view",
+               "--decision-file", str(original), "--source-hash", digest, "--replan-stage", "review"]
+
+    def read():
+        result = subprocess.run(command, capture_output=True, text=True, timeout=15)
+        return result.returncode, json.loads(result.stdout)
+
+    code, review = read()
+    assert code == 0 and review["ok"]
+    saved = original.rename(tmp_path / "saved.json")
+    original.symlink_to(saved)
+    code, rejected = read()
+    assert code == 1 and rejected["ok"] is False
+    original.unlink()
+    if hasattr(os, "mkfifo"):
+        os.mkfifo(original)
+        code, rejected = read()
+        assert code == 1 and rejected["ok"] is False
+        original.unlink()
+    saved.rename(original)
+    assert read() == (0, review)
diff --git a/tests/control_plane/test_turn_host_decision_capture.py b/tests/control_plane/test_turn_host_decision_capture.py
index 804ee99515..a0073bb774 100644
--- a/tests/control_plane/test_turn_host_decision_capture.py
+++ b/tests/control_plane/test_turn_host_decision_capture.py
@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 import importlib.util
+import hashlib
 import json
+import os
 import shlex
 import subprocess
 import sys
@@ -125,6 +127,22 @@ def test_real_host_reads_original_capture_after_source_change(
     if first_attempt_fails:
         retry = [*replay, "--resume-turn-key", request["turn_key"], "--retry-failed-turn", "--execute"]
         received_before = (workspace / "received.txt").read_bytes()
+        if hasattr(os, "mkfifo"):
+            journal_before_fifo = journal_path.read_bytes()
+            path.unlink()
+            os.mkfifo(path, 0o600)
+            # Bound the real recovery process: opening a pipe in blocking mode
+            # would wait for an absent writer under the Turn journal lock.
+            rejected_pipe = subprocess.run(
+                [sys.executable, "-m", "loopx.entrypoint", *retry], capture_output=True, timeout=15,
+            )
+            assert rejected_pipe.returncode == 1
+            assert "regular file" in json.loads(rejected_pipe.stdout)["error"]
+            assert journal_path.read_bytes() == journal_before_fifo
+            assert (workspace / "received.txt").read_bytes() == received_before
+            assert f._quota_spend_count(runtime) == 0
+            path.unlink()
+            path.write_bytes(original)
         for corrupt in (True, False):
             if corrupt:
                 path.write_text(json.dumps({**source_a, "reason": "replaced evidence"}))
@@ -205,3 +223,19 @@ def test_capture_compacts_layout_without_changing_values(tmp_path, caller):
     assert "保留".encode() in raw
     assert b"two  spaces" in raw and b"\\nnew line\\t" in raw
     assert len(raw) < len((json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode())
+
+
+def test_publication_digest_covers_exact_bytes_without_a_file_readback(tmp_path, monkeypatch):
+    from loopx.control_plane.turn_driver.decision_capture import capture_context_projection
+
+    payload = {"requirement": "保留每一条约束。", "counter": 2**80 + 1,
+               "values": [True, 1, -0.0, 1.25, None]}
+    expected = (json.dumps(payload, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode()
+    with monkeypatch.context() as writing:
+        def no_readback(*args, **kwargs):
+            pytest.fail("Publishing this new capture must not reread it merely to hash its bytes")
+        writing.setattr(Path, "read_bytes", no_readback)
+        reference = capture_context_projection(payload, directory=tmp_path)
+    assert Path(reference["path"]).read_bytes() == expected
+    assert reference["sha256"] == hashlib.sha256(expected).hexdigest()
+    assert Path(reference["path"]).stat().st_mode & 0o077 == 0