From e165096b71969f088185e14f77855ca1a8d46c5a Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 19:31:57 +0800
Subject: [PATCH 1/6] fix(codex): reuse the captured managed claim after lost
 readiness

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       |  5 +-
 .../codex_execution_preparation.py            | 21 ++++-
 .../test_quota_execution_preparation.py       | 83 +++++++++++++++++++
 3 files changed, 104 insertions(+), 5 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index d6b7fac970..d8ef75f7d5 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -373,7 +373,10 @@ 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 signed capture reuses its managed claim operation. The authority owner
+checks the original lease on replay. A retired lease needs a fresh eligible
+preparation 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/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py
index 8eb671980a..dc3cdb4840 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py
@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import json
+import re
 import subprocess
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
@@ -189,16 +190,29 @@ 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_args = expected_claim
+    claim_operation_id = None
+    if context_directory is not None:
+        if capture_directory is None or not turn_instance_id:
+            raise ValueError("Native execution preparation requires a bound captured Turn")
+        source_sha256 = request.get("source_sha256")
+        if not isinstance(source_sha256, str) or re.fullmatch(r"[0-9a-f]{64}", source_sha256) is None:
+            raise ValueError("Native execution preparation requires its original capture digest")
+        # The same capture can reappear after a lost readiness result. Let the
+        # managed claim owner recover the original receipt and current lease.
+        claim_operation_id = f"todo-claim:codex-capture:{source_sha256}"
+        claim_args = [*expected_claim, "--claim-operation-id", claim_operation_id]
     claimed = _run_json(
-        expected_claim,
+        claim_args,
         registry_path=registry_path,
         runtime_root=runtime_root,
         cwd=workspace,
         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 +273,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/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 2473c022d9..4920c2eb38 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -384,6 +384,89 @@ def hook():
     assert log.read_bytes() == after_claim
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("released", [False, True], ids=["same-execution", "released-execution"])
+def test_native_hook_replays_the_original_claim_after_guard_timeout(
+    tmp_path, monkeypatch, capsys, provider, released,
+):
+    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) == {}
+    before = _run(tmp_path, monkeypatch, provider, [], "--codex-app",
+        "--turn-instance-id", turn, "--todo-id", TODO)
+    assert before["execution_preparation"]["state"] == "claim_required"
+    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=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
+
+    assert hook() == {}
+    original_id = claims[0]["claim_operation_id"]
+    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"])
+
+    result = hook()
+    current = read("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
+    assert len(claims) == 2
+    if released:
+        assert result == {} and not claims[1]["ok"]
+        assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"]
+        assert guards == [turn]
+    else:
+        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(

From 8e7c9d6a92cbf2089583d279322d59de987047ad Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 20:44:31 +0800
Subject: [PATCH 2/6] Require a capture-bound claim identity across native Host
 upgrades

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       |  15 ++-
 loopx/cli_commands/quota.py                   |  14 ++-
 loopx/cli_commands/quota_capture.py           |  34 ++---
 .../turn_driver/codex_context.py              |   5 +-
 .../codex_execution_preparation.py            |  40 ++++--
 .../test_quota_execution_preparation.py       | 116 +++++++++++-------
 6 files changed, 147 insertions(+), 77 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index f0178a5df0..72afcd7fba 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -380,9 +380,18 @@ 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. If readiness is interrupted, retrying the
-same signed capture reuses its managed claim operation. The authority owner
-checks the original lease on replay. A retired lease needs a fresh eligible
-preparation capture.
+same original decision capture reuses its managed claim operation. The
+authority owner checks the original lease on replay. Presenting the saved
+decision again preserves its claim binding. Older captures without the
+binding stay 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..f4fb6298a3 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -386,6 +386,17 @@ 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))
+    claim_preparation = payload.get("execution_preparation")
+    bind_native_claim = (args.quota_command == "should-run" and native_context
+            and capture_directory is not None and payload.get("ok") is True
+            and isinstance(claim_preparation, Mapping) and claim_preparation.get("state") == "claim_required")
+    if bind_native_claim:
+        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
+
+        # Only the live guard may mark immutable bytes as eligible for a
+        # capture-bound claim. Reading or presenting old bytes cannot add it.
+        payload = {**payload, NATIVE_CLAIM_REPLAY_BOUND: True}
     try:
         captured_digest = capture_decision(capture_directory, payload)
         if getattr(args, "prepared_capture_reference", False) and payload.get("ok") is True:
@@ -432,7 +443,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):
@@ -451,7 +461,7 @@ def _emit_quota_result(
             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)
+                reader_argv=reader_argv, claim_source_sha256=captured_digest if bind_native_claim else None)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index c812f619a4..8acae4f670 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -630,7 +630,8 @@ 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, capture_id: str, decision_identity: tuple[str, str],
-                                 progressive_enabled: bool, reader_argv: list[str] | None) -> dict[str, Any]:
+                                 progressive_enabled: bool, reader_argv: list[str] | None,
+                                 claim_source_sha256: str | None = 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, prepare_codex_context,
@@ -644,7 +645,7 @@ def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespa
         capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1],
         surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet",
         capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None),
-        progressive_enabled=progressive_enabled)
+        progressive_enabled=progressive_enabled, claim_source_sha256=claim_source_sha256)
     recovery = prepared.delivery.get("recovery")
     view = native_context_capture_view(prepared.delivery, capture_directory, context_directory=directory,
         progressive_enabled=progressive_enabled, initial_read_plan=prepared.initial_read_plan,
@@ -768,6 +769,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             pass  # The complete original reference remains the fallback below.
     try:
         from ..control_plane.turn_driver.codex_context import codex_context_directory, prepare_codex_context
+        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
 
         if not activation["enabled"]:
             raise ValueError("context delivery is off")
@@ -808,6 +810,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             continuity_generation=getattr(args, "_context_delivery_generation", None),
             progressive_enabled=activation["progressive_enabled"], selection_command_args_template=transport,
             captured_decision_path=str(path) if args.initial_read and activation["progressive_enabled"] else None,
+            claim_source_sha256=args.capture_sha256 if payload.get(NATIVE_CLAIM_REPLAY_BOUND) is True else None,
             hook_session=hook_session)
         projected = prepared.delivery
         native_prepared = True
@@ -1279,6 +1282,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,
         )
@@ -1297,11 +1301,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
@@ -1340,24 +1347,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..9fba9a8906 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -620,6 +620,7 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                           continuity_generation: str | None = None,
                           include_initial_read_plan: bool = True,
                           progressive_enabled: bool | None = None,
+                          claim_source_sha256: str | None = None,
                           selection_command_args_template: str | None = None,
                           captured_decision_path: str | None = None,
                           hook_session: str | None = None) -> NativeContextPreparation:
@@ -677,7 +678,9 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                 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)
+                    preparation = compact_execution_preparation_request(
+                        payload, claim_source_sha256=claim_source_sha256,
+                    )
                     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 dc3cdb4840..949914ec96 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py
@@ -7,20 +7,32 @@
 from __future__ import annotations
 
 import json
-import re
 import subprocess
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Any
 
 from ...command_invocation import current_loopx_cli_argv
+from ..content_digest import BARE_SHA256_PATTERN
 
 
 CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
+NATIVE_CLAIM_REPLAY_BOUND = "native_claim_replay_bound"
+
+
+class LegacyNativePreparationCapture(ValueError):
+    """The capture cannot identify any claim made by an older Host."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            "This saved capture predates capture-bound 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."""
 
@@ -63,6 +75,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,
@@ -76,6 +93,7 @@ def compact_execution_preparation_request(
             "runtime_root": str(runtime_root),
         },
         **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}),
+        **claim_binding,
     }
 
 
@@ -174,7 +192,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()
@@ -190,20 +207,19 @@ def execute_codex_execution_preparation(
     ):
         raise ValueError("Codex execution preparation requires a physical workspace")
     invoke = runner or _default_runner
-    claim_args = expected_claim
     claim_operation_id = None
-    if context_directory is not None:
-        if capture_directory is None or not turn_instance_id:
+    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("source_sha256")
-        if not isinstance(source_sha256, str) or re.fullmatch(r"[0-9a-f]{64}", source_sha256) is None:
+        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")
-        # The same capture can reappear after a lost readiness result. Let the
-        # managed claim owner recover the original receipt and current lease.
         claim_operation_id = f"todo-claim:codex-capture:{source_sha256}"
-        claim_args = [*expected_claim, "--claim-operation-id", claim_operation_id]
+        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(
-        claim_args,
+        expected_claim,
         registry_path=registry_path,
         runtime_root=runtime_root,
         cwd=workspace,
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 4920c2eb38..2222576bc7 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import timedelta
+import hashlib
 import json
 import os
 from pathlib import Path
@@ -69,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 = "",
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime = tmp_path / "runtime"
@@ -77,7 +80,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(
@@ -106,7 +109,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,
     )
@@ -114,6 +117,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",
@@ -202,16 +219,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,
@@ -237,7 +253,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":
@@ -304,7 +320,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
@@ -317,13 +332,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
@@ -385,9 +397,9 @@ def hook():
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("released", [False, True], ids=["same-execution", "released-execution"])
+@pytest.mark.parametrize("case", ["same-execution", "released-execution", "legacy-released-execution"])
 def test_native_hook_replays_the_original_claim_after_guard_timeout(
-    tmp_path, monkeypatch, capsys, provider, released,
+    tmp_path, monkeypatch, capsys, provider, case,
 ):
     import io
     from loopx.cli import main
@@ -402,13 +414,17 @@ def test_native_hook_replays_the_original_claim_after_guard_timeout(
     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 == "legacy-released-execution"
+    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)
+        "--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"
-    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=turn)
+    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 = [], []
@@ -441,22 +457,42 @@ def read(*args):
         assert code == 0 and payload["ok"], payload
         return payload
 
-    assert hook() == {}
-    original_id = claims[0]["claim_operation_id"]
+    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"])
 
+    if legacy:
+        legacy_result = hook()
+        guidance = legacy_result["hookSpecificOutput"]["additionalContext"]
+        assert "predates capture-bound Host claims" in guidance and "fresh capture" in guidance
+        assert "host_execution_prepared" not in guidance
+    assert _present_saved_guard(tmp_path, native, capture, "codex_app_heartbeat", turn) == digest
     result = hook()
+    if legacy:
+        assert result == legacy_result
     current = read("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
-    assert len(claims) == 2
-    if released:
+    assert source.read_bytes() == original_bytes
+    if legacy:
+        assert not claims and not guards
+        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
@@ -583,22 +619,22 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
 ):
     from loopx.control_plane.turn_driver.codex_context import (
         codex_context_hook,
-        prepare_codex_context,
     )
 
+    tmp_path = tmp_path.resolve()
+    context_root = tmp_path / "context"
+    capture_root = tmp_path / "capture"
+    context_root.mkdir(mode=0o700)
     before = _run(
         tmp_path,
         monkeypatch,
         provider,
         [] if lease_state == "missing" else [_lease(expired=True)],
         "--runtime-profile", "codex_app_ssh_goal",
-        "--begin-turn",
+        "--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"]
-    context_root = tmp_path / "context"
-    capture_root = tmp_path / "capture"
-    context_root.mkdir(mode=0o700)
-    capture_root.mkdir(mode=0o700)
     foreign_workspace = tmp_path / "project-b"
     foreign_workspace.mkdir()
     monkeypatch.setenv("CODEX_THREAD_ID", "codex-hook-session")
@@ -610,15 +646,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
         "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
+    _present_saved_guard(tmp_path, context_root, capture_root, "codex_app_ssh_goal", turn)
 
     command = [
         sys.executable, "-m", "loopx.entrypoint",

From 97799d322e75ec6b7efd77487482c6606698fbc1 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 20:58:56 +0800
Subject: [PATCH 3/6] Keep native claim capture binding within the quota
 decision budget

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota.py         | 28 +++++++++++++++++-----------
 loopx/cli_commands/quota_capture.py |  5 +++--
 2 files changed, 20 insertions(+), 13 deletions(-)

diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index f4fb6298a3..da421f6707 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -329,6 +329,21 @@ def _emit_bounded_settlement(payload, args, *, runtime_root, reader_argv, print_
     return 0 if payload.get("ok") is True else 1
 
 
+def _bind_native_claim_capture(
+    payload: dict[str, object], args: argparse.Namespace, capture_directory: Path | None, native_context: bool,
+) -> tuple[dict[str, object], bool]:
+    preparation = payload.get("execution_preparation")
+    if (args.quota_command == "should-run" and native_context
+            and capture_directory is not None and payload.get("ok") is True
+            and isinstance(preparation, Mapping) and preparation.get("state") == "claim_required"):
+        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
+
+        # Only the live guard can bind an original claim. Presenting saved
+        # bytes cannot make an older capture eligible.
+        return {**payload, NATIVE_CLAIM_REPLAY_BOUND: True}, True
+    return payload, False
+
+
 def _emit_quota_result(
     payload: dict[str, object],
     args: argparse.Namespace,
@@ -387,16 +402,7 @@ def _emit_quota_result(
             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))
-    claim_preparation = payload.get("execution_preparation")
-    bind_native_claim = (args.quota_command == "should-run" and native_context
-            and capture_directory is not None and payload.get("ok") is True
-            and isinstance(claim_preparation, Mapping) and claim_preparation.get("state") == "claim_required")
-    if bind_native_claim:
-        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
-
-        # Only the live guard may mark immutable bytes as eligible for a
-        # capture-bound claim. Reading or presenting old bytes cannot add it.
-        payload = {**payload, NATIVE_CLAIM_REPLAY_BOUND: True}
+    payload, bind_native_claim = _bind_native_claim_capture(payload, args, capture_directory, native_context)
     try:
         captured_digest = capture_decision(capture_directory, payload)
         if getattr(args, "prepared_capture_reference", False) and payload.get("ok") is True:
@@ -461,7 +467,7 @@ def _emit_quota_result(
             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, claim_source_sha256=captured_digest if bind_native_claim else None)
+                reader_argv=reader_argv, captured_claim_replay=bind_native_claim)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 8acae4f670..e1ebd2679b 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -631,7 +631,7 @@ 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, capture_id: str, decision_identity: tuple[str, str],
                                  progressive_enabled: bool, reader_argv: list[str] | None,
-                                 claim_source_sha256: str | None = None) -> dict[str, Any]:
+                                 captured_claim_replay: bool = False) -> 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, prepare_codex_context,
@@ -645,7 +645,8 @@ def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespa
         capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1],
         surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet",
         capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None),
-        progressive_enabled=progressive_enabled, claim_source_sha256=claim_source_sha256)
+        progressive_enabled=progressive_enabled,
+        claim_source_sha256=capture_id if captured_claim_replay else None)
     recovery = prepared.delivery.get("recovery")
     view = native_context_capture_view(prepared.delivery, capture_directory, context_directory=directory,
         progressive_enabled=progressive_enabled, initial_read_plan=prepared.initial_read_plan,

From cb249c0a32985e0e8590b1e561b469e5feff4a31 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 21:13:53 +0800
Subject: [PATCH 4/6] Verify the live native guard before replaying a captured
 claim

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       | 10 ++--
 loopx/cli_commands/quota.py                   | 20 +-------
 loopx/cli_commands/quota_capture.py           | 19 ++++----
 .../turn_driver/codex_context.py              | 48 +++++++++++++++++--
 .../codex_execution_preparation.py            |  5 +-
 .../test_quota_execution_preparation.py       | 36 +++++++++-----
 6 files changed, 88 insertions(+), 50 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 72afcd7fba..81a78db507 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -381,15 +381,17 @@ 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. If readiness is interrupted, retrying the
 same original decision capture reuses its managed claim operation. The
-authority owner checks the original lease on replay. Presenting the saved
-decision again preserves its claim binding. Older captures without the
-binding stay readable. Preserve any original claim receipt and use
+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。
 
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index da421f6707..847c10bd28 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -329,21 +329,6 @@ def _emit_bounded_settlement(payload, args, *, runtime_root, reader_argv, print_
     return 0 if payload.get("ok") is True else 1
 
 
-def _bind_native_claim_capture(
-    payload: dict[str, object], args: argparse.Namespace, capture_directory: Path | None, native_context: bool,
-) -> tuple[dict[str, object], bool]:
-    preparation = payload.get("execution_preparation")
-    if (args.quota_command == "should-run" and native_context
-            and capture_directory is not None and payload.get("ok") is True
-            and isinstance(preparation, Mapping) and preparation.get("state") == "claim_required"):
-        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
-
-        # Only the live guard can bind an original claim. Presenting saved
-        # bytes cannot make an older capture eligible.
-        return {**payload, NATIVE_CLAIM_REPLAY_BOUND: True}, True
-    return payload, False
-
-
 def _emit_quota_result(
     payload: dict[str, object],
     args: argparse.Namespace,
@@ -402,9 +387,8 @@ def _emit_quota_result(
             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))
-    payload, bind_native_claim = _bind_native_claim_capture(payload, args, capture_directory, native_context)
     try:
-        captured_digest = capture_decision(capture_directory, payload)
+        captured_digest = capture_decision(capture_directory, payload, native_args=args if native_context else None)
         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
@@ -467,7 +451,7 @@ def _emit_quota_result(
             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, captured_claim_replay=bind_native_claim)
+                reader_argv=reader_argv)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index e1ebd2679b..ff9a79b7f8 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -630,8 +630,7 @@ 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, capture_id: str, decision_identity: tuple[str, str],
-                                 progressive_enabled: bool, reader_argv: list[str] | None,
-                                 captured_claim_replay: bool = False) -> dict[str, Any]:
+                                 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, prepare_codex_context,
@@ -645,8 +644,7 @@ def project_native_quota_context(payload: dict[str, Any], args: argparse.Namespa
         capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1],
         surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet",
         capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None),
-        progressive_enabled=progressive_enabled,
-        claim_source_sha256=capture_id if captured_claim_replay else None)
+        progressive_enabled=progressive_enabled)
     recovery = prepared.delivery.get("recovery")
     view = native_context_capture_view(prepared.delivery, capture_directory, context_directory=directory,
         progressive_enabled=progressive_enabled, initial_read_plan=prepared.initial_read_plan,
@@ -770,7 +768,6 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             pass  # The complete original reference remains the fallback below.
     try:
         from ..control_plane.turn_driver.codex_context import codex_context_directory, prepare_codex_context
-        from ..control_plane.turn_driver.codex_execution_preparation import NATIVE_CLAIM_REPLAY_BOUND
 
         if not activation["enabled"]:
             raise ValueError("context delivery is off")
@@ -811,7 +808,6 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
             continuity_generation=getattr(args, "_context_delivery_generation", None),
             progressive_enabled=activation["progressive_enabled"], selection_command_args_template=transport,
             captured_decision_path=str(path) if args.initial_read and activation["progressive_enabled"] else None,
-            claim_source_sha256=args.capture_sha256 if payload.get(NATIVE_CLAIM_REPLAY_BOUND) is True else None,
             hook_session=hook_session)
         projected = prepared.delivery
         native_prepared = True
@@ -957,13 +953,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], *,
+                     native_args: argparse.Namespace | None = None) -> 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_args is not None and native_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=native_args.codex_context_dir,
+                context_root=getattr(native_args, "codex_context_root", None))
+        return digest
     except (OSError, ValueError) as exc:
         raise QuotaDecisionTransportError(
             "Decision capture failed after guard evaluation; the Turn receipt may "
diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index 9fba9a8906..d66f32294a 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -210,6 +210,42 @@ 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:
+        if _state(root).get("session_id") in (None, session):
+            write_decision_capture(root / "claim-source.json",
+                {"session_id": 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 +457,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
 
@@ -620,7 +659,6 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap
                           continuity_generation: str | None = None,
                           include_initial_read_plan: bool = True,
                           progressive_enabled: bool | None = None,
-                          claim_source_sha256: str | None = None,
                           selection_command_args_template: str | None = None,
                           captured_decision_path: str | None = None,
                           hook_session: str | None = None) -> NativeContextPreparation:
@@ -674,13 +712,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, claim_source_sha256=claim_source_sha256,
-                    )
                     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 949914ec96..2c11461295 100644
--- a/loopx/control_plane/turn_driver/codex_execution_preparation.py
+++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py
@@ -17,15 +17,14 @@
 
 
 CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
-NATIVE_CLAIM_REPLAY_BOUND = "native_claim_replay_bound"
 
 
 class LegacyNativePreparationCapture(ValueError):
-    """The capture cannot identify any claim made by an older Host."""
+    """The capture cannot prove a live guard bound its claim."""
 
     def __init__(self) -> None:
         super().__init__(
-            "This saved capture predates capture-bound Host claims. Preserve any existing claim receipt. "
+            "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."
         )
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 2222576bc7..88dadc1fa1 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -397,7 +397,10 @@ def hook():
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("case", ["same-execution", "released-execution", "legacy-released-execution"])
+@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,
 ):
@@ -415,7 +418,7 @@ def test_native_hook_replays_the_original_claim_after_guard_timeout(
              "hook_event_name": "SessionStart", "source": "startup"}
     assert codex_context.codex_context_hook(str(native), event) == {}
     capture = tmp_path / "before-claim"
-    legacy = case == "legacy-released-execution"
+    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",
@@ -472,19 +475,27 @@ def read(*args):
         read("task-lease", "release", "--goal-id", GOAL, "--todo-id", TODO, "--owner", AGENT,
             "--expected-version", str(lease["version"]), "--idempotency-key", lease["idempotency_key"])
 
-    if legacy:
-        legacy_result = hook()
-        guidance = legacy_result["hookSpecificOutput"]["additionalContext"]
-        assert "predates capture-bound Host claims" in guidance and "fresh capture" in guidance
-        assert "host_execution_prepared" not in guidance
-    assert _present_saved_guard(tmp_path, native, capture, "codex_app_heartbeat", turn) == digest
+    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()
-    if legacy:
-        assert result == legacy_result
     current = read("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
     assert source.read_bytes() == original_bytes
-    if legacy:
-        assert not claims and not guards
+    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
@@ -625,6 +636,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
     context_root = tmp_path / "context"
     capture_root = tmp_path / "capture"
     context_root.mkdir(mode=0o700)
+    monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
     before = _run(
         tmp_path,
         monkeypatch,

From 42362791fca3b342b4841fa7bd2058c9bf59b2a7 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 21:20:04 +0800
Subject: [PATCH 5/6] Keep native claim provenance inside the capture writer
 boundary

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/cli_commands/quota.py                            | 2 +-
 loopx/cli_commands/quota_capture.py                    | 8 ++++----
 tests/architecture/test_content_digest_single_owner.py | 2 ++
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 847c10bd28..741e45b729 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -388,7 +388,7 @@ def _emit_quota_result(
         )
     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, native_args=args if native_context else None)
+        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
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index ff9a79b7f8..98a40bf9d2 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -954,18 +954,18 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None:
 
 
 def capture_decision(directory: Path | None, payload: dict[str, object], *,
-                     native_args: argparse.Namespace | None = None) -> str | None:
+                     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:
         digest = write_decision_capture(directory / "decision.json", payload, publish=os.replace)
-        if native_args is not None and native_args.quota_command == "should-run" and payload.get("ok") is True:
+        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=native_args.codex_context_dir,
-                context_root=getattr(native_args, "codex_context_root", None))
+            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(
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",

From b341dd80a3cbba8d9edfed01116dce0ec35dd24c Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 21:23:41 +0800
Subject: [PATCH 6/6] Bind explicit Codex preflight to its existing native
 session

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../turn_driver/codex_context.py              | 10 +++++---
 .../test_quota_execution_preparation.py       | 24 +++++++++++--------
 2 files changed, 21 insertions(+), 13 deletions(-)

diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py
index d66f32294a..72abed39c0 100644
--- a/loopx/control_plane/turn_driver/codex_context.py
+++ b/loopx/control_plane/turn_driver/codex_context.py
@@ -222,9 +222,13 @@ def record_native_claim_source(
     if location is None:
         return
     with _locked(location) as root:
-        if _state(root).get("session_id") in (None, session):
-            write_decision_capture(root / "claim-source.json",
-                {"session_id": session, "sha256": digest}, publish=os.replace)
+        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(
diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py
index 88dadc1fa1..8f93c31cf8 100644
--- a/tests/control_plane/test_quota_execution_preparation.py
+++ b/tests/control_plane/test_quota_execution_preparation.py
@@ -625,8 +625,9 @@ 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,
@@ -637,6 +638,16 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
     capture_root = tmp_path / "capture"
     context_root.mkdir(mode=0o700)
     monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
+    foreign_workspace = tmp_path / "project-b"
+    foreign_workspace.mkdir()
+    event = {
+        "session_id": "codex-hook-session",
+        "cwd": str(foreign_workspace.resolve()),
+        "hook_event_name": "SessionStart",
+        "source": "startup",
+    }
+    if startup == "before-guard":
+        assert codex_context_hook(str(context_root), event) == {}
     before = _run(
         tmp_path,
         monkeypatch,
@@ -647,17 +658,10 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project(
         "--decision-output-dir", str(capture_root), context_delivery="on",
     )
     turn = before["heartbeat_receipt"]["turn_instance_id"]
-    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) == {}
+    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 = [