diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 4a991d51e0..b0084a8d6c 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -418,21 +418,30 @@ 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. If readiness is interrupted, retrying the -same original decision capture reuses its managed claim operation. The -authority owner checks the original lease on replay. The native session -keeps the latest eligible live guard digest separately from the capture. -Presenting the exact saved decision preserves that binding. Older captures -and copies with different bytes remain readable. Preserve any original claim receipt and use -`quota should-run` with the same Goal, Agent, Todo and Turn and a fresh -capture if preparation is still required. A retired lease requires a fresh -eligible preparation capture. - -中文：Host 重试绑定原始决策 capture 的领取操作，由权威 owner 验证原租约。 -原生会话独立保存最新合格准入的摘要。重新展示逐字节一致的快照会保留绑定。 -旧快照和字节不同的副本仍可读，保留已有领取回执。 -仍需准备时，用原 Goal、Agent、Todo 和 Turn 运行 `quota should-run`， -生成新的 capture，再按当前准入结果准备。已退役的租约需要新的合格 capture。 +bytes before work can continue. +The private native journal reserves automatic preparation before claiming a +bound capture and records the verified ready capture before projecting it. If +readiness or hook delivery is interrupted, the hook repeatably surfaces +readback for the original claim receipt, current lease and any verified ready +capture until delivery is confirmed, without repeating the claim or guard. +Follow the captured readiness check for the same Todo and Turn once the active +lease matches. An authorized retry of the same bound capture reuses its +managed claim operation, and the authority owner checks the original lease on +replay. The native session keeps the latest eligible live guard digest +separately from the capture. Presenting the exact saved decision preserves +that binding. Older captures and copies with different bytes remain readable. +Preserve any original claim receipt and use `quota should-run` with the same +Goal, Agent, Todo and Turn and a fresh capture if preparation is still +required. A retired lease requires a fresh eligible preparation capture. + +中文：私有原生日志在对绑定 capture 发起领取前先登记自动准备，并在投影前记录已校验的 +ready capture。若就绪检查或 Hook 送达中断，Hook 在确认送达前可重复提供原领取回执、 +当前租约及已校验 ready capture 的回读入口，不再重复自动领取或准入。活动租约匹配后， +沿同一 Todo 与 Turn 完成就绪检查。对同一绑定 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_capture.py b/loopx/cli_commands/quota_capture.py index 94d407a79d..e6cf3fb6bd 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -460,7 +460,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, - hook_session=hook_session) + hook_session=hook_session, saved_capture=True) projected = prepared.delivery initial_read_plan = prepared.initial_read_plan native_prepared = True @@ -939,11 +939,12 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca from ..control_plane.turn_driver.codex_context import ( CONTEXT_LIFECYCLE_HOOKS, codex_context_directory, codex_context_hook, native_context_event_eligible, native_execution_preparation_request, - record_native_execution_preparation, + record_native_execution_preparation, record_native_execution_preparation_delivered, ) from ..control_plane.turn_driver.codex_execution_preparation import ( LegacyNativePreparationCapture, codex_execution_preparation_context, + codex_execution_preparation_readback_context, execute_codex_execution_preparation, ) @@ -961,14 +962,18 @@ 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 preparation_context = "" 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 "claim_source_sha256" not in request: - raise LegacyNativePreparationCapture() + if "claim_source_sha256" not in request: + raise LegacyNativePreparationCapture() + if request.get("readback_only") is True: + preparation_context = codex_execution_preparation_readback_context(request) or "" + else: capture = ( Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready" if request.get("turn_instance_id") else None @@ -995,6 +1000,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca # pinned full recovery even if native projection # fails. It never rereads authority or admits work. digest = receipt.pop("capture_sha256") + receipt.pop("capture_path", None) receipt["context"] = project_saved_context_capture(argparse.Namespace(**{ **vars(args), "quota_command": "context-project", "capture_file": str(source), @@ -1008,13 +1014,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()) preparation_context = codex_execution_preparation_context(receipt) + record_native_execution_preparation_delivered( + directory, + event, + source_sha256=request["source_sha256"], + ) 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 + # 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: + preparation_context = codex_execution_preparation_readback_context(readback) or "" if preparation_context: existing = payload.get("hookSpecificOutput") existing_context = str(existing.get("additionalContext") or "") if isinstance(existing, dict) else "" diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index c949912597..84ea0a2125 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/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 @@ -402,6 +403,17 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str if name == "PostToolUse": checkpoint_native_read(read_state, updated, event, [page for page in pages if "read_sections" not in page]) + if same and any(page.get("ok") is True for page in pages): + previous = updated.get("host_execution_preparation") + if ( + isinstance(previous, dict) + and previous.get("source_sha256") == (recovery or {}).get("sha256") + and previous.get("delivery_confirmed") is not True + ): + updated["host_execution_preparation"] = { + **previous, + "delivery_confirmed": True, + } if updated != state: _save(root, updated) saved = updated.get("recovery") @@ -431,9 +443,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 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"}): @@ -451,19 +463,56 @@ 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 - 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 readback_for is not None and readback_for != source_sha256: return None action = {**request, "source_sha256": source_sha256} - if not _original_native_claim_source(root, session, action.get("claim_source_sha256")): + claim_source_sha256 = action.get("claim_source_sha256") + if not _original_native_claim_source(root, session, claim_source_sha256): action.pop("claim_source_sha256", None) + if readback_for is not None: + return None + return action + previous = state.get("host_execution_preparation") + if isinstance(previous, dict) and ( + previous.get("source_sha256") == source_sha256 + or ( + isinstance(claim_source_sha256, str) + and previous.get("claim_source_sha256") == claim_source_sha256 + ) + ): + if previous.get("state") in {"attempted", "ready"}: + if ( + event.get("hook_event_name") != "SessionStart" + and previous.get("delivery_confirmed") is True + ): + return None + readback = {**action, "readback_only": True} + if isinstance(previous.get("claim_operation_id"), str) and previous["claim_operation_id"]: + readback["claim_operation_id"] = previous["claim_operation_id"] + if isinstance(previous.get("ready_capture"), dict): + readback["ready_capture"] = dict(previous["ready_capture"]) + return readback + if readback_for is not None: + return None + 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, + **( + {"claim_source_sha256": claim_source_sha256} + if isinstance(claim_source_sha256, str) + else {} + ), + "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 @@ -491,8 +540,25 @@ def record_native_execution_preparation( or recovery.get("sha256") != source_sha256 ): raise ValueError("execution-preparation source moved before receipt publication") + request = recovery.get("execution_preparation") + claim_source_sha256 = request.get("claim_source_sha256") if isinstance(request, dict) else None + ready_capture = None + capture_path = receipt.get("capture_path") + capture_sha256 = receipt.get("capture_sha256") + if ( + isinstance(capture_path, str) + and capture_path + and isinstance(capture_sha256, str) + and BARE_SHA256_PATTERN.fullmatch(capture_sha256) is not None + ): + ready_capture = {"path": capture_path, "sha256": capture_sha256} state["host_execution_preparation"] = { "source_sha256": source_sha256, + **( + {"claim_source_sha256": claim_source_sha256} + if isinstance(claim_source_sha256, str) + else {} + ), "state": "ready", "todo_id": receipt.get("todo_id"), **( @@ -500,10 +566,42 @@ def record_native_execution_preparation( if receipt.get("turn_instance_id") else {} ), + **( + {"claim_operation_id": receipt["claim_operation_id"]} + if isinstance(receipt.get("claim_operation_id"), str) and receipt["claim_operation_id"] + else {} + ), + **({"ready_capture": ready_capture} if ready_capture is not None else {}), } _save(root, state) +def record_native_execution_preparation_delivered( + directory: str | None, + event: dict[str, Any], + *, + source_sha256: str, +) -> None: + """Mark a ready or terminal preparation receipt delivered for this source.""" + + if directory is None or not native_context_event_eligible(event): + return + with _locked(directory) as root: + state = _state(root) + previous = state.get("host_execution_preparation") + if ( + state.get("session_id") == event.get("session_id") + and state.get("cwd") == event.get("cwd") + and isinstance(previous, dict) + and previous.get("source_sha256") == source_sha256 + ): + state["host_execution_preparation"] = { + **previous, + "delivery_confirmed": True, + } + _save(root, state) + + def _context_pages(value: object) -> list[dict[str, Any]]: """Decode native shell/MCP JSON records; never classify arbitrary prose. @@ -555,6 +653,9 @@ def _verified_context_pages(state: dict[str, Any], pages: list[dict[str, Any]]) """ references = [value for key in ("recovery", "pending") if isinstance(value := state.get(key), dict)] + prepared = state.get("host_execution_preparation") + if isinstance(prepared, dict) and isinstance(ready_ref := prepared.get("ready_capture"), dict): + references.append(ready_ref) files: dict[str, bytes] = {} result = [] for page in pages: @@ -665,7 +766,8 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap progressive_enabled: bool | None = None, selection_command_args_template: str | None = None, captured_decision_path: str | None = None, - hook_session: str | None = None) -> NativeContextPreparation: + hook_session: str | None = None, + saved_capture: bool = False) -> NativeContextPreparation: """Commit current recovery before optional projection or candidate updates. Serialization prevents invalidation from moving the epoch during preparation. @@ -718,16 +820,41 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap 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) + bound_claim_source = ( + capture_id + if preparation and _original_native_claim_source(root, session, capture_id, bind_unattached=True) + else None + ) def publish_recovery(reference: dict[str, Any]) -> None: + claim_source_sha256 = bound_claim_source + if ( + preparation + and claim_source_sha256 is None + and not saved_capture + and preparation.get("turn_instance_id") + ): + claim_source_sha256 = ( + capture_id + if BARE_SHA256_PATTERN.fullmatch(capture_id) is not None + else reference["sha256"] + ) + write_decision_capture( + root / "claim-source.json", + {"session_id": session, "sha256": claim_source_sha256}, + publish=os.replace, + ) + published_preparation = ( + compact_execution_preparation_request(payload, claim_source_sha256=claim_source_sha256) + if preparation and claim_source_sha256 is not None + else preparation + ) reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"], page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=current_loopx_cli_argv()) 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, - **({"execution_preparation": preparation} if preparation else {})} + **({"execution_preparation": published_preparation} if published_preparation else {})} state["pending"] = {"path": reference["path"], "sha256": reference["sha256"], "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} _save(root, state) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 2c11461295..9185c6fd27 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 shlex import subprocess from collections.abc import Callable, Mapping, Sequence from pathlib import Path @@ -151,6 +152,72 @@ 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("claim_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, lease and ready-capture 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 = request.get("claim_operation_id") + if not isinstance(operation_id, str) or not operation_id: + operation_id = _native_claim_operation_id(request) + base = [*current_loopx_cli_argv(), "--registry", registry_path, + "--runtime-root", runtime_root, "--format", "json"] + facts: dict[str, Any] = { + "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]), + } + ready_capture = request.get("ready_capture") + if isinstance(ready_capture, Mapping): + ready_path = ready_capture.get("path") + ready_sha256 = ready_capture.get("sha256") + if ( + isinstance(ready_path, str) + and ready_path + and isinstance(ready_sha256, str) + and BARE_SHA256_PATTERN.fullmatch(ready_sha256) is not None + ): + from .decision_capture import context_capture_read_command, read_context_capture_file + + try: + read_context_capture_file(ready_path, ready_sha256, private=True) + except (OSError, ValueError, KeyError, TypeError, RecursionError): + pass + else: + recovery_command = context_capture_read_command( + Path(ready_path), ready_sha256, reader_argv=current_loopx_cli_argv(), + ) + facts["capture_sha256"] = ready_sha256 + facts["context"] = { + "capture_sha256": ready_sha256, + "recovery_command": recovery_command, + } + 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], *, @@ -288,6 +355,7 @@ 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 + receipt["capture_path"] = str(capture_directory / "decision.json") 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_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py index edcaf7be86..e6cfcfb145 100644 --- a/tests/control_plane/test_codex_context_lifecycle.py +++ b/tests/control_plane/test_codex_context_lifecycle.py @@ -61,27 +61,44 @@ def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatc "heartbeat_receipt": {"turn_instance_id": "turn-fixture"}, } _, wire = project(tmp_path, payload=payload) - request = context.native_execution_preparation_request( - str(tmp_path), {**event, "hook_event_name": "PostToolUse"}, - ) + source_sha256 = wire["recovery"]["sha256"] + eligible = {**event, "hook_event_name": "PostToolUse"} + request = context.native_execution_preparation_request(str(tmp_path), eligible) assert request == { "goal_id": "fixture", "agent_id": "agent", "todo_id": "todo_fixture", - "claim_args": payload["execution_preparation"]["claim"]["cli_args"], + "claim_args": [ + *payload["execution_preparation"]["claim"]["cli_args"], + "--claim-operation-id", + f"todo-claim:codex-capture:{source_sha256}", + ], "authority_route": { "registry_path": str(tmp_path / "registry.json"), "runtime_root": str(tmp_path / "runtime"), }, "turn_instance_id": "turn-fixture", - "source_sha256": wire["recovery"]["sha256"], + "claim_source_sha256": source_sha256, + "source_sha256": source_sha256, } child_event = {**event, "hook_event_name": "PostToolUse", "agent_id": "child"} assert context.native_execution_preparation_request(str(tmp_path), child_event) is None + 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), {**event, "hook_event_name": "SessionStart"}, + ) == { + **request, "readback_only": True, + } receipt = { "event": "host_execution_prepared", "todo_id": "todo_fixture", "turn_instance_id": "turn-fixture", + "claim_operation_id": f"todo-claim:codex-capture:{source_sha256}", } state_before_child = (tmp_path / "session.json").read_bytes() context.record_native_execution_preparation( @@ -93,12 +110,38 @@ def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatc str(tmp_path), event, source_sha256=request["source_sha256"], receipt=receipt, ) + assert context.native_execution_preparation_request(str(tmp_path), eligible) == { + **request, + "readback_only": True, + "claim_operation_id": f"todo-claim:codex-capture:{source_sha256}", + } assert context.native_execution_preparation_request( - str(tmp_path), {**event, "hook_event_name": "PostToolUse"}, - ) is None + str(tmp_path), {**event, "hook_event_name": "SessionStart"}, + ) == { + **request, + "readback_only": True, + "claim_operation_id": f"todo-claim:codex-capture:{source_sha256}", + } + deliver(tmp_path, event, read_pages(tmp_path, wire)) + assert context.native_execution_preparation_request(str(tmp_path), eligible) is None assert context.native_execution_preparation_request( str(tmp_path), {**event, "hook_event_name": "SessionStart"}, - )["todo_id"] == "todo_fixture" + )["readback_only"] is True + payload["heartbeat_receipt"] = {"turn_instance_id": "next-turn"} + _, next_wire = project(tmp_path, capture_id="next", payload=payload) + next_sha256 = next_wire["recovery"]["sha256"] + assert next_sha256 != source_sha256 + assert context.native_execution_preparation_request(str(tmp_path), eligible) == { + **request, + "claim_args": [ + *payload["execution_preparation"]["claim"]["cli_args"], + "--claim-operation-id", + f"todo-claim:codex-capture:{next_sha256}", + ], + "turn_instance_id": "next-turn", + "claim_source_sha256": next_sha256, + "source_sha256": next_sha256, + } def test_shared_root_read_does_not_create_unknown_session_or_follow_alias(tmp_path, monkeypatch): diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index f3b86cd897..391d00c8e5 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -417,7 +417,8 @@ def hook(): @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("case", [ - "same-execution", "released-execution", "legacy-released-execution", + "same-execution", "released-execution", "after-ready-same-execution", + "after-ready-released-execution", "legacy-released-execution", "reencoded-released-execution", "legacy-marked-released-execution", ]) def test_native_hook_replays_the_original_claim_after_guard_timeout( @@ -438,7 +439,8 @@ def test_native_hook_replays_the_original_claim_after_guard_timeout( assert codex_context.codex_context_hook(str(native), event) == {} capture = tmp_path / "before-claim" legacy = case.startswith("legacy-") - released = case != "same-execution" + after_ready = case.startswith("after-ready-") + released = not case.endswith("same-execution") native_args = () if legacy else ("--defer-context-presentation", "--codex-context-dir", str(native)) before = _run(tmp_path, monkeypatch, provider, [], "--codex-app", "--turn-instance-id", turn, "--todo-id", TODO, "--decision-output-dir", str(capture), @@ -449,26 +451,33 @@ def test_native_hook_replays_the_original_claim_after_guard_timeout( digest = _present_saved_guard(tmp_path, native, capture, "codex_app_heartbeat", turn) run = codex_execution_preparation._default_runner + record_ready = codex_context.record_native_execution_preparation 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: + if not after_ready and 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 + def interrupt_after_ready(*args, **kwargs): + record_ready(*args, **kwargs) + if after_ready: + raise RuntimeError("interrupted after ready receipt") + monkeypatch.setattr(codex_execution_preparation, "_default_runner", interrupt_first_guard) + monkeypatch.setattr(codex_context, "record_native_execution_preparation", interrupt_after_ready) 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() + 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 @@ -485,29 +494,34 @@ def read(*args): legacy_claim = read(*before["execution_preparation"]["claim"]["cli_args"]) original_id = legacy_claim["claim_operation_id"] else: - assert hook() == {} + first_context = hook()["hookSpecificOutput"]["additionalContext"] + first_request = json.loads(first_context.rsplit("Receipt: ", 1)[1]) + assert first_request["event"] == "host_execution_preparation_readback_required" original_id = claims[0]["claim_operation_id"] assert original_id == f"todo-claim:codex-capture:{digest}" + assert first_request["claim_operation_id"] == original_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"]) - 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 + if legacy or unverified_copy: + presented_capture = capture + 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() + resume = hook("SessionStart") current = read("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO) assert source.read_bytes() == original_bytes if legacy or unverified_copy: @@ -516,19 +530,31 @@ def read(*args): assert "host_execution_prepared" not in guidance assert len(claims) == len(guards) == (0 if legacy else 1) assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"] - elif released: - assert len(claims) == 2 - assert result == {} and not claims[1]["ok"] - assert not current["active"] and current["lease"]["lease_epoch"] == lease["lease_epoch"] - assert guards == [turn] else: - assert len(claims) == 2 - context = result["hookSpecificOutput"]["additionalContext"] - receipt = json.loads(context.rsplit("Receipt: ", 1)[1]) - assert receipt["claim_operation_id"] == claims[1]["claim_operation_id"] == original_id - assert receipt["turn_instance_id"] == turn and receipt["execution_preparation"]["state"] == "ready" - assert current["active"] and current["lease"] == lease - assert guards == [turn, turn] + assert len(claims) == 1 and guards == [turn] + for observed in (result, resume): + context = observed["hookSpecificOutput"]["additionalContext"] + receipt = json.loads(context.rsplit("Receipt: ", 1)[1]) + assert receipt["event"] == "host_execution_preparation_readback_required" + assert receipt["claim_operation_id"] == original_id + assert (receipt["goal_id"], receipt["agent_id"], receipt["todo_id"], receipt["turn_instance_id"]) == ( + GOAL, AGENT, TODO, turn, + ) + if after_ready: + assert receipt["context"]["recovery_command"] + for command_key in ("receipt_command", "lease_command"): + proc = subprocess.run( + shlex.split(receipt[command_key]), cwd=tmp_path, capture_output=True, text=True, check=True, + ) + assert json.loads(proc.stdout)["ok"] + 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 readback = read("todo", "receipt", "--goal-id", GOAL, "--operation-id", original_id) assert readback["cursor"] == original["cursor"] and readback["receipts"] == original["receipts"]