diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 73966d4bf4..a2ce483d3d 100644
--- a/docs/reference/required-work-context.md
+++ b/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
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 92339b96f0..741e45b729 100644
--- a/loopx/cli_commands/quota.py
+++ b/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):
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 4e77e76faf..4bf1e7449a 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/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):
diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 5cb3f8f59b..72abed39c0 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/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,
diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py
index 8eb671980a..2c11461295 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/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
diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py
index 8aa1326f6b..59f425c41e 100644
--- a/tests/architecture/test_content_digest_single_owner.py
+++ b/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",
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index abe20b78e2..979698860f 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -70,6 +70,8 @@ def _run(
     leases: list[dict[str, object]],
     *guard_args: str,
     todos: list[dict[str, object]] | None = None,
+    context_delivery: str = "off",
+    goal_suffix: str = "",
     write_scopes: list[str] | None = None,
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
@@ -79,7 +81,7 @@ def _run(
     state.write_text(
         "---\nstatus: active\n---\n# Goal\n\n## Objective\n"
         "Deliver one validated synthetic change.\n\n## Next Action\n"
-        "Run the selected Todo.\n",
+        "Run the selected Todo.\n" + goal_suffix,
         encoding="utf-8",
     )
     write_fixture_registry(
@@ -109,7 +111,7 @@ def _run(
     )
     code, packet = run_json_cli_result(
         "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT,
-        "--context-delivery", "off",
+        "--context-delivery", context_delivery,
         *guard_args,
         registry_path=registry, runtime_root=runtime,
     )
@@ -117,6 +119,20 @@ def _run(
     return packet
 
 
+def _present_saved_guard(tmp_path, native, capture, scenario, turn):
+    source = capture / "decision.json"
+    digest = hashlib.sha256(source.read_bytes()).hexdigest()
+    code, view = run_json_cli_result(
+        "quota", "context-project", "--runtime-profile", scenario,
+        "--context-delivery", "on", "--codex-context-dir", str(native),
+        "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", turn,
+        "--capture-file", str(source), "--capture-sha256", digest,
+        registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime",
+    )
+    assert code == 0 and view["ok"] is True, view
+    return digest
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize(
     "leases,reason_code",
@@ -205,16 +221,15 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     assert codex_context.codex_context_hook(str(native), event) == {}
     original = {**_todo(), "text": "Deliver the complete synthetic request: 保留原文" + "。不可截断的验收条件" * 100,
         "note": "Preserve the original stop and once-only settlement."}
+    capture = tmp_path / "before-claim"
     before = _run(tmp_path, monkeypatch, provider, [], "--runtime-profile", scenario,
-        "--turn-instance-id", "preparation-turn", "--todo-id", TODO, todos=[original])
+        "--turn-instance-id", "preparation-turn", "--todo-id", TODO,
+        "--defer-context-presentation", "--codex-context-dir", str(native), "--decision-output-dir", str(capture),
+        todos=[original], context_delivery="on",
+        goal_suffix="\n\n## Constraints\n\n" + "Preserve the complete Goal and its stop. 保留停止条件。\n" * 100)
     assert before["execution_preparation"]["state"] == "claim_required"
-    goal_file = tmp_path / "state.md"
-    goal_text = goal_file.read_text() + "\n\n## Constraints\n\n" + "Preserve the complete Goal and its stop. 保留停止条件。\n" * 100
-    goal_file.write_text(goal_text)
-    capture = tmp_path / "before-claim"
-    capture.mkdir(mode=0o700)
-    codex_context.prepare_codex_context(before, directory=str(native), capture_directory=capture,
-        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="before-claim").delivery
+    goal_text = (tmp_path / "state.md").read_text()
+    digest = _present_saved_guard(tmp_path, native, capture, scenario, "preparation-turn")
     hook = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(tmp_path / "registry.json"),
         "--runtime-root", str(tmp_path / "runtime"), "--format", "json", "quota", "context-hook",
         "--context-delivery", "on", "--runtime-profile", scenario,
@@ -240,7 +255,7 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
     assert receipt["todo_id"] == TODO and receipt["turn_instance_id"] == "preparation-turn"
     assert receipt["execution_preparation"]["state"] == "ready"
     claim_reference = receipt["claim_operation_id"]
-    assert claim_reference.startswith("todo-claim:")
+    assert claim_reference == f"todo-claim:codex-capture:{digest}"
     ready = receipt["context"]
     assert ready["ok"], ready.get("reason")
     if location == "root":
@@ -307,7 +322,6 @@ def test_native_hook_returns_readable_ready_source_and_original_claim(
 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
@@ -320,13 +334,10 @@ def test_prepared_hook_refuses_a_replaced_ready_capture_without_reclaim(
     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.prepare_codex_context(before, directory=str(native), capture_directory=original,
-        goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="claim-required").delivery
+    _run(tmp_path, monkeypatch, provider, [], "--codex-app",
+        "--turn-instance-id", "prepared-turn", "--todo-id", TODO, "--codex-context-dir", str(native),
+        "--decision-output-dir", str(original), context_delivery="on")
     from loopx.control_plane.turn_driver import codex_execution_preparation
     record = codex_context.record_native_execution_preparation
     run = codex_execution_preparation._default_runner
@@ -387,6 +398,124 @@ def hook():
     assert log.read_bytes() == after_claim
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("case", [
+    "same-execution", "released-execution", "legacy-released-execution",
+    "reencoded-released-execution", "legacy-marked-released-execution",
+])
+def test_native_hook_replays_the_original_claim_after_guard_timeout(
+    tmp_path, monkeypatch, capsys, provider, case,
+):
+    import io
+    from loopx.cli import main
+    from loopx.control_plane.turn_driver import codex_context, codex_execution_preparation
+
+    tmp_path = tmp_path.resolve()
+    monkeypatch.chdir(tmp_path)
+    session, turn = "claim-replay-session", "claim-replay-turn"
+    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) == {}
+    capture = tmp_path / "before-claim"
+    legacy = case.startswith("legacy-")
+    released = case != "same-execution"
+    native_args = () if legacy else ("--defer-context-presentation", "--codex-context-dir", str(native))
+    before = _run(tmp_path, monkeypatch, provider, [], "--codex-app",
+        "--turn-instance-id", turn, "--todo-id", TODO, "--decision-output-dir", str(capture),
+        *native_args, context_delivery="off" if legacy else "on")
+    assert before["execution_preparation"]["state"] == "claim_required"
+    source = capture / "decision.json"
+    original_bytes = source.read_bytes()
+    digest = _present_saved_guard(tmp_path, native, capture, "codex_app_heartbeat", turn)
+
+    run = codex_execution_preparation._default_runner
+    claims, guards = [], []
+
+    def interrupt_first_guard(command, **kwargs):
+        if "--prepared-capture-reference" in command:
+            guards.append(command[command.index("--turn-instance-id") + 1])
+            if len(guards) == 1:
+                raise subprocess.TimeoutExpired(command, kwargs["timeout"])
+        result = run(command, **kwargs)
+        if "claim" in command:
+            claims.append(json.loads(result.stdout))
+        return result
+
+    monkeypatch.setattr(codex_execution_preparation, "_default_runner", interrupt_first_guard)
+    registry, runtime = tmp_path / "registry.json", tmp_path / "runtime"
+    command = ["--registry", str(registry), "--runtime-root", str(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)
+
+    def read(*args):
+        code, payload = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime, cwd=tmp_path)
+        assert code == 0 and payload["ok"], payload
+        return payload
+
+    if legacy:
+        # Before the upgrade the Host issued the ordinary claim without a
+        # persisted operation ID, then could lose the ready result.
+        legacy_claim = read(*before["execution_preparation"]["claim"]["cli_args"])
+        original_id = legacy_claim["claim_operation_id"]
+    else:
+        assert hook() == {}
+        original_id = claims[0]["claim_operation_id"]
+        assert original_id == f"todo-claim:codex-capture:{digest}"
+    original = read("todo", "receipt", "--goal-id", GOAL, "--operation-id", original_id)
+    lease = original["receipts"][0]["result"]["lease"]
+    if released:
+        read("task-lease", "release", "--goal-id", GOAL, "--todo-id", TODO, "--owner", AGENT,
+            "--expected-version", str(lease["version"]), "--idempotency-key", lease["idempotency_key"])
+
+    presented_capture = capture
+    unverified_copy = case in ("reencoded-released-execution", "legacy-marked-released-execution")
+    if unverified_copy:
+        presented_capture = tmp_path / "unverified"
+        presented_capture.mkdir(mode=0o700)
+        copy = presented_capture / "decision.json"
+        if legacy:
+            copy.write_text(json.dumps({**json.loads(original_bytes), "native_claim_replay_bound": True}))
+        else:
+            copy.write_bytes(original_bytes + b" ")
+        copy.chmod(0o600)
+    presented_digest = _present_saved_guard(tmp_path, native, presented_capture, "codex_app_heartbeat", turn)
+    assert (presented_digest != digest) == unverified_copy
+    result = hook()
+    current = read("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
+    assert source.read_bytes() == original_bytes
+    if legacy or unverified_copy:
+        guidance = result["hookSpecificOutput"]["additionalContext"]
+        assert "no verified live guard binding" in guidance and "fresh capture" in guidance
+        assert "host_execution_prepared" not in guidance
+        assert len(claims) == len(guards) == (0 if legacy else 1)
+        assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"]
+    elif released:
+        assert len(claims) == 2
+        assert result == {} and not claims[1]["ok"]
+        assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"]
+        assert guards == [turn]
+    else:
+        assert len(claims) == 2
+        context = result["hookSpecificOutput"]["additionalContext"]
+        receipt = json.loads(context.rsplit("Receipt: ", 1)[1])
+        assert receipt["claim_operation_id"] == claims[1]["claim_operation_id"] == original_id
+        assert receipt["turn_instance_id"] == turn and receipt["execution_preparation"]["state"] == "ready"
+        assert current["active"] and current["lease"] == lease
+        assert guards == [turn, turn]
+    readback = read("todo", "receipt", "--goal-id", GOAL, "--operation-id", original_id)
+    assert readback["cursor"] == original["cursor"] and readback["receipts"] == original["receipts"]
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("multiple_candidates", [False, True])
 def test_projected_claim_prepares_the_same_receipt_bound_turn(
@@ -555,47 +684,44 @@ def test_preparation_does_not_replace_explicit_action_selection(monkeypatch):
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("lease_state", ["missing", "expired"])
+@pytest.mark.parametrize("startup", ["before-guard", "after-guard"])
 def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
-    tmp_path, monkeypatch, provider, lease_state,
+    tmp_path, monkeypatch, provider, lease_state, startup,
 ):
     from loopx.control_plane.turn_driver.codex_context import (
         codex_context_hook,
-        prepare_codex_context,
     )
 
-    before = _run(
-        tmp_path,
-        monkeypatch,
-        provider,
-        [] if lease_state == "missing" else [_lease(expired=True)],
-        "--runtime-profile", "codex_app_ssh_goal",
-        "--begin-turn",
-    )
-    turn = before["heartbeat_receipt"]["turn_instance_id"]
+    tmp_path = tmp_path.resolve()
     context_root = tmp_path / "context"
     capture_root = tmp_path / "capture"
     context_root.mkdir(mode=0o700)
-    capture_root.mkdir(mode=0o700)
+    monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
     foreign_workspace = tmp_path / "project-b"
     foreign_workspace.mkdir()
-    monkeypatch.setenv("CODEX_THREAD_ID", "codex-hook-session")
-    monkeypatch.chdir(foreign_workspace)
     event = {
         "session_id": "codex-hook-session",
         "cwd": str(foreign_workspace.resolve()),
         "hook_event_name": "SessionStart",
         "source": "startup",
     }
-    assert codex_context_hook(str(context_root), event) == {}
-    prepare_codex_context(
-        before,
-        directory=str(context_root),
-        capture_directory=capture_root,
-        goal_id=GOAL,
-        agent_id=AGENT,
-        surface="quota_packet",
-        capture_id="claim-required",
-    ).delivery
+    if startup == "before-guard":
+        assert codex_context_hook(str(context_root), event) == {}
+    before = _run(
+        tmp_path,
+        monkeypatch,
+        provider,
+        [] if lease_state == "missing" else [_lease(expired=True)],
+        "--runtime-profile", "codex_app_ssh_goal",
+        "--begin-turn", "--defer-context-presentation", "--codex-context-dir", str(context_root),
+        "--decision-output-dir", str(capture_root), context_delivery="on",
+    )
+    turn = before["heartbeat_receipt"]["turn_instance_id"]
+    monkeypatch.setenv("CODEX_THREAD_ID", "codex-hook-session")
+    monkeypatch.chdir(foreign_workspace)
+    if startup == "after-guard":
+        assert codex_context_hook(str(context_root), event) == {}
+    _present_saved_guard(tmp_path, context_root, capture_root, "codex_app_ssh_goal", turn)
 
     command = [
         sys.executable, "-m", "loopx.entrypoint",