Recover native execution preparation without repeating claims

docs/reference/required-work-context.md

@@ -391,6 +391,12 @@ returns its writer digest with the fields needed to verify the same Todo and
 Turn. The hook verifies that digest when it opens the original file. If the
 file changed, the receipt preserves the claim and pins recovery to the original
 bytes before work can continue.
+The private native journal reserves automatic preparation before claiming a
+bound capture. If readiness is interrupted, the hook offers one readback
+for the original claim receipt and current lease. Follow the captured readiness
+check for the same Todo and Turn once the active lease matches. An authorized
+retry of the same capture uses the same managed claim operation. 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

loopx/cli_commands/quota_capture.py

@@ -1284,6 +1284,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
         )
         from ..control_plane.turn_driver.codex_execution_preparation import (
             codex_execution_preparation_context,
+            codex_execution_preparation_readback_context,
             execute_codex_execution_preparation,
         )
 
@@ -1301,11 +1302,16 @@ 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)
+                request = None
+                additional = None
                 try:
-                    request = native_execution_preparation_request(directory, event)
+                    workspace = Path(event.get("cwd") or "")
+                    if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve():
+                        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 request.get("readback_only") is True:
+                            additional = codex_execution_preparation_readback_context(request)
+                        else:
                             capture = (
                                 Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready"
                                 if request.get("turn_instance_id") else None
@@ -1344,24 +1350,22 @@ 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),
-                                ])),
-                            }
+                            additional = codex_execution_preparation_context(receipt)
                 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
+                    # Reconcile only the attempted source. A concurrent guard
+                    # can replace recovery, so it must never be reserved here.
+                    if request is not None and request.get("turn_instance_id"):
+                        readback = native_execution_preparation_request(directory, event,
+                            readback_for=request["source_sha256"])
+                        if readback is not None:
+                            additional = codex_execution_preparation_readback_context(readback)
+                if additional is not None:
+                    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, additional])),
+                    }
             else:
                 payload = {}
         except (OSError, RuntimeError, subprocess.SubprocessError, ValueError, TypeError, RecursionError):

loopx/control_plane/turn_driver/codex_context.py

@@ -27,6 +27,7 @@
     NATIVE_CONTEXT_LARGE_OUTPUT_BYTES, NATIVE_CONTEXT_HISTORY_TOKENS, NATIVE_CONTEXT_OUTPUT_ENV,
     native_context_page_budget,
 )
+from ..content_digest import BARE_SHA256_PATTERN
 from .codex_execution_preparation import compact_execution_preparation_request
 
 # Native adapter activation reuses scheduler-owned host declarations. These
@@ -391,9 +392,9 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str
 
 
 def native_execution_preparation_request(
-    directory: str | None, event: dict[str, Any],
+    directory: str | None, event: dict[str, Any], *, readback_for: str | None = None,
 ) -> dict[str, Any] | None:
-    """Read one compact Host action from this exact native session journal."""
+    """Reserve one automatic try or surface one readback for the same source."""
 
     if (directory is None or not native_context_event_eligible(event)
             or event.get("hook_event_name") not in {"SessionStart", "PostToolUse"}):
@@ -411,17 +412,36 @@ def native_execution_preparation_request(
                 return None
             request = recovery.get("execution_preparation")
             source_sha256 = recovery.get("sha256")
-            if not isinstance(request, dict) or not isinstance(source_sha256, str):
+            if (not isinstance(request, dict) or not isinstance(source_sha256, str)
+                    or BARE_SHA256_PATTERN.fullmatch(source_sha256) is None):
                 return None
+            if readback_for is not None and readback_for != source_sha256:
+                return None
+            action = {**request, "source_sha256": source_sha256}
             previous = state.get("host_execution_preparation")
-            if (
-                event.get("hook_event_name") != "SessionStart"
-                and isinstance(previous, dict)
-                and previous.get("source_sha256") == source_sha256
-                and previous.get("state") == "ready"
-            ):
+            if isinstance(previous, dict) and previous.get("source_sha256") == source_sha256:
+                if previous.get("state") == "attempted":
+                    if previous.get("readback_sent") is True:
+                        return None
+                    previous["readback_sent"] = True
+                    _save(root, state)
+                    return {**action, "readback_only": True}
+                if previous.get("state") == "ready" and event.get("hook_event_name") != "SessionStart":
+                    return None
+            if readback_for is not None:
                 return None
-            return {**request, "source_sha256": source_sha256}
+            if request.get("turn_instance_id"):
+                # Commit the reservation before the ordinary claim can run.
+                # An interrupted Hook may have committed the claim; only the
+                # exact provider receipt and current lease answer that question.
+                state["host_execution_preparation"] = {
+                    "source_sha256": source_sha256,
+                    "state": "attempted",
+                    "todo_id": request.get("todo_id"),
+                    "turn_instance_id": request.get("turn_instance_id"),
+                }
+                _save(root, state)
+            return action
     except (OSError, ValueError, TypeError, RecursionError):
         return None
 

loopx/control_plane/turn_driver/codex_execution_preparation.py

@@ -7,12 +7,14 @@
 from __future__ import annotations
 
 import json
+import shlex
 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]]
@@ -133,6 +135,45 @@ def prepared_capture_reference(payload: Mapping[str, Any], digest: str) -> dict[
     }
 
 
+def _native_claim_operation_id(request: Mapping[str, Any]) -> str:
+    """Reuse the captured managed claim identity on an authorized retry."""
+    digest = request.get("source_sha256")
+    if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None:
+        raise ValueError("Native execution preparation requires its original capture digest")
+    return f"todo-claim:codex-capture:{digest}"
+
+
+def codex_execution_preparation_readback_context(request: Mapping[str, Any]) -> str | None:
+    """Expose only existing receipt and lease readers after an interrupted preparation."""
+    goal, agent, todo, turn = (request.get(key) for key in ("goal_id", "agent_id", "todo_id", "turn_instance_id"))
+    route = request.get("authority_route")
+    if not isinstance(route, Mapping) or not all(
+        isinstance(part, str) and part for part in (goal, agent, todo, turn)
+    ):
+        return None
+    registry_path, runtime_root = route.get("registry_path"), route.get("runtime_root")
+    if not all(isinstance(part, str) and Path(part).is_absolute() for part in (registry_path, runtime_root)):
+        return None
+    operation_id = _native_claim_operation_id(request)
+    base = [*current_loopx_cli_argv(), "--registry", registry_path,
+            "--runtime-root", runtime_root, "--format", "json"]
+    facts = {
+        "event": "host_execution_preparation_readback_required",
+        "goal_id": goal, "agent_id": agent, "todo_id": todo, "turn_instance_id": turn,
+        "claim_operation_id": operation_id,
+        "receipt_command": shlex.join([*base, "todo", "receipt", "--goal-id", goal, "--operation-id", operation_id]),
+        "lease_command": shlex.join([*base, "task-lease", "inspect", "--goal-id", goal, "--todo-id", todo]),
+    }
+    return (
+        "Automatic preparation started for this captured Turn and may still be finishing. "
+        "Read the original claim receipt and current lease before deciding how to recover. "
+        "With the matching active lease, follow the current captured same-Todo and same-Turn readiness check before work. "
+        "For an authorized retry of this claim, pass the original --claim-operation-id. "
+        "A retired lease requires a fresh eligible capture. "
+        "Receipt: " + json.dumps(facts, ensure_ascii=False, separators=(",", ":"))
+    )
+
+
 def execute_codex_execution_preparation(
     request: Mapping[str, Any],
     *,
@@ -189,16 +230,24 @@ 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")
+        claim_operation_id = _native_claim_operation_id(request)
+        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 +308,6 @@ def execute_codex_execution_preparation(
         if not isinstance(digest, str) or len(digest) != 64:
             raise ValueError("The guard did not return its published capture digest")
         receipt["capture_sha256"] = digest
-    claim_reference = claimed.get("claim_operation_id")
     if isinstance(claim_reference, str) and claim_reference:
         receipt["claim_operation_id"] = claim_reference
     # The Host already paid for the captured complete decision. Its caller

tests/control_plane/test_codex_context_lifecycle.py

@@ -78,6 +78,14 @@ def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatc
     }
     child_event = {**event, "hook_event_name": "PostToolUse", "agent_id": "child"}
     assert context.native_execution_preparation_request(str(tmp_path), child_event) is None
+    eligible = {**event, "hook_event_name": "PostToolUse"}
+    assert context.native_execution_preparation_request(
+        str(tmp_path), eligible, readback_for="0" * 64,
+    ) is None
+    assert context.native_execution_preparation_request(str(tmp_path), eligible) == {
+        **request, "readback_only": True,
+    }
+    assert context.native_execution_preparation_request(str(tmp_path), eligible) is None
     receipt = {
         "event": "host_execution_prepared",
         "todo_id": "todo_fixture",
@@ -99,6 +107,12 @@ def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatc
     assert context.native_execution_preparation_request(
         str(tmp_path), {**event, "hook_event_name": "SessionStart"},
     )["todo_id"] == "todo_fixture"
+    payload["heartbeat_receipt"] = {"turn_instance_id": "next-turn"}
+    _, next_wire = project(tmp_path, capture_id="next", payload=payload)
+    assert next_wire["recovery"]["sha256"] != request["source_sha256"]
+    assert context.native_execution_preparation_request(str(tmp_path), eligible) == {
+        **request, "turn_instance_id": "next-turn", "source_sha256": next_wire["recovery"]["sha256"],
+    }
 
 
 def test_shared_root_read_does_not_create_unknown_session_or_follow_alias(tmp_path, monkeypatch):

tests/control_plane/test_quota_execution_preparation.py

@@ -387,6 +387,105 @@ 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_recovers_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(name="PostToolUse"):
+        data = json.dumps({**event, "hook_event_name": name}).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
+
+    context = hook()["hookSpecificOutput"]["additionalContext"]
+    request = json.loads(context.rsplit("Receipt: ", 1)[1])
+    assert request["event"] == "host_execution_preparation_readback_required"
+    assert (request["goal_id"], request["agent_id"], request["todo_id"], request["turn_instance_id"]) == (
+        GOAL, AGENT, TODO, turn,
+    )
+    original_id = claims[0]["claim_operation_id"]
+    assert request["claim_operation_id"] == original_id
+    assert original_id.startswith("todo-claim:codex-capture:")
+
+    def projected_read(name):
+        result = subprocess.run(shlex.split(request[name]), cwd=tmp_path, capture_output=True,
+                                check=True, text=True)
+        return json.loads(result.stdout)
+
+    original = projected_read("receipt_command")
+    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"])
+
+    assert hook() == {}
+    assert hook() == {}
+    resume = hook("SessionStart").get("hookSpecificOutput", {}).get("additionalContext", "")
+    assert "host_execution_preparation_readback_required" not in resume
+    assert len(claims) == 1 and guards == [turn]
+
+    current = projected_read("lease_command")
+    if released:
+        assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"]
+    else:
+        assert current["active"] and current["lease"] == lease
+        ready = read("quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT,
+            "--context-delivery", "off", "--codex-app", "--turn-instance-id", turn, "--todo-id", TODO)
+        assert ready["execution_preparation"]["state"] == "ready"
+        assert ready["heartbeat_receipt"]["turn_instance_id"] == turn
+        assert projected_read("lease_command")["lease"] == lease
+
+    readback = projected_read("receipt_command")
+    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(