diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 7ac6fe66e1..22eae0aac9 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -442,6 +442,9 @@ 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. +The native hook executes only a preparation bound to a captured Turn. An +ordinary claim recommendation remains available to its caller; reading one +or reading an original receipt does not trigger an automatic claim. 中文：私有原生日志在对绑定 capture 发起领取前先登记自动准备，并在投影前记录已校验的 ready capture。若就绪检查或 Hook 送达中断，Hook 可重复提供原领取回执、当前租约 @@ -452,6 +455,8 @@ ready capture。若就绪检查或 Hook 送达中断，Hook 可重复提供原 快照会保留绑定。旧快照和字节不同的副本仍可读，保留已有领取回执。仍需准备时，用原 Goal、Agent、Todo 和 Turn 运行 `quota should-run`，生成新的 capture，再按当前准入 结果准备。已退役的租约需要新的合格 capture。 +原生 Hook 只执行与已捕获 Turn 绑定的准备。调用方仍可读取普通领取建议；读取建议或原 +领取回执都不会触发自动领取。 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 512335a3e5..9d1db26ee5 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -844,21 +844,18 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve(): request = native_execution_preparation_request(directory, event) if request is not None: - if "claim_source_sha256" not in request: + if "claim_source_sha256" not in request or not request.get("turn_instance_id"): 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 - ) + capture = Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready" receipt = execute_codex_execution_preparation( request, cwd=workspace.resolve(), scenario=activation["scenario"], capture_directory=capture, - context_directory=directory if capture is not None else None, + context_directory=Path(directory), ) record_native_execution_preparation( directory, @@ -869,25 +866,22 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca # Publish the decision that the Host just verified; # a model must not need a second guard to obtain its # Goal, task, requirements and settlement contract. - if capture is not None: - source = capture / "decision.json" - # The existing saved-capture owner preserves a - # pinned full recovery even if native projection - # fails. It never rereads authority or admits work. - digest = receipt.pop("capture_sha256") - receipt.pop("capture_path", None) - receipt["context"] = project_saved_context_capture(argparse.Namespace(**{ - **vars(args), "quota_command": "context-project", - "capture_file": str(source), - "capture_sha256": digest, - "goal_id": request["goal_id"], "agent_id": request["agent_id"], - "turn_instance_id": request["turn_instance_id"], - }), hook_session=event["session_id"]) - if receipt["context"].get("ok") is False: - from ..command_invocation import current_loopx_cli_argv - - receipt["context"]["recovery_command"] = context_capture_read_command( - source, digest, reader_argv=current_loopx_cli_argv()) + source = capture / "decision.json" + # The saved-capture reader never rereads authority or admits work. + digest = receipt.pop("capture_sha256") + receipt.pop("capture_path") + receipt["context"] = project_saved_context_capture(argparse.Namespace(**{ + **vars(args), "quota_command": "context-project", + "capture_file": str(source), + "capture_sha256": digest, + "goal_id": request["goal_id"], "agent_id": request["agent_id"], + "turn_instance_id": request["turn_instance_id"], + }), hook_session=event["session_id"]) + if receipt["context"].get("ok") is False: + from ..command_invocation import current_loopx_cli_argv + + receipt["context"]["recovery_command"] = context_capture_read_command( + source, digest, reader_argv=current_loopx_cli_argv()) preparation_context = codex_execution_preparation_context(receipt) except LegacyNativePreparationCapture as error: preparation_context = str(error) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index 9185c6fd27..c57735fe99 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -2,7 +2,7 @@ The TypeScript control plane remains the decision owner. This adapter accepts only its exact ordinary claim, executes it through the public CLI, and proves -the same Todo (and Turn when present) is ready before returning a Host receipt. +the captured Todo and Turn are ready before returning a Host receipt. """ from __future__ import annotations @@ -223,12 +223,12 @@ def execute_codex_execution_preparation( *, cwd: Path, scenario: str, - capture_directory: Path | None = None, - context_directory: Path | None = None, + capture_directory: Path, + context_directory: Path, timeout: float = 8, runner: CommandRunner | None = None, ) -> dict[str, Any]: - """Execute one projected claim and prove same-identity readiness. + """Execute one journal-bound claim and publish its ready capture. The hook never selects work. Unknown/malformed requests fail before a subprocess. The ordinary claim remains the authority and rechecks all @@ -272,18 +272,18 @@ def execute_codex_execution_preparation( or workspace.resolve() != workspace ): raise ValueError("Codex execution preparation requires a physical workspace") - invoke = runner or _default_runner - claim_operation_id = None - if context_directory is not None or "source_sha256" in request: - if context_directory is None or capture_directory is None or not turn_instance_id: - raise ValueError("Native execution preparation requires a bound captured Turn") - source_sha256 = request.get("claim_source_sha256") - if not isinstance(source_sha256, str) or BARE_SHA256_PATTERN.fullmatch(source_sha256) is None: - raise ValueError("Native execution preparation requires its original capture digest") - claim_operation_id = f"todo-claim:codex-capture:{source_sha256}" - expected_claim += ["--claim-operation-id", claim_operation_id] + source_sha256 = request.get("source_sha256") + if ( + context_directory is None or capture_directory is None or not turn_instance_id + or request.get("readback_only") is True + or not isinstance(source_sha256, str) or BARE_SHA256_PATTERN.fullmatch(source_sha256) is None + ): + raise ValueError("Native execution preparation requires a bound captured Turn") + claim_operation_id = _native_claim_operation_id(request) + expected_claim += ["--claim-operation-id", claim_operation_id] if request.get("claim_args") != expected_claim: raise ValueError("invalid Codex execution-preparation claim arguments") + invoke = runner or _default_runner claimed = _run_json( expected_claim, registry_path=registry_path, @@ -293,7 +293,7 @@ def execute_codex_execution_preparation( runner=invoke, ) claim_reference = claimed.get("claim_operation_id") - if claim_operation_id is not None and claim_reference != claim_operation_id: + if claim_reference != claim_operation_id: raise RuntimeError("Native execution preparation did not recover its original managed claim") guard = [ "quota", "should-run", @@ -301,12 +301,9 @@ def execute_codex_execution_preparation( "--agent-id", agent_id, *( ["--codex-app"] if scenario == "codex_app_heartbeat" else ["--runtime-profile", "codex_app_ssh_goal"] ), - *(["--context-delivery", "on", "--codex-context-dir", str(context_directory), - "--defer-context-presentation"] if context_directory is not None else - ["--context-delivery", "off"]), - "--todo-id", todo_id, - *(["--turn-instance-id", turn_instance_id] if turn_instance_id else []), - *(["--decision-output-dir", str(capture_directory), "--prepared-capture-reference"] if capture_directory is not None else []), + "--context-delivery", "on", "--codex-context-dir", str(context_directory), "--defer-context-presentation", + "--todo-id", todo_id, "--turn-instance-id", turn_instance_id, + "--decision-output-dir", str(capture_directory), "--prepared-capture-reference", ] after = _run_json( guard, @@ -319,10 +316,6 @@ def execute_codex_execution_preparation( preparation = after.get("execution_preparation") selected = after.get("selected_todo") receipt = after.get("heartbeat_receipt") - same_turn = not turn_instance_id or ( - isinstance(receipt, Mapping) - and receipt.get("turn_instance_id") == turn_instance_id - ) ready = ( after.get("ok") is True and after.get("goal_id") == goal_id @@ -335,32 +328,29 @@ def execute_codex_execution_preparation( and preparation.get("todo_id") == todo_id and preparation.get("agent_id") == agent_id and preparation.get("delivery_allowed") is True - and same_turn + and isinstance(receipt, Mapping) + and receipt.get("turn_instance_id") == turn_instance_id ) if not ready: raise RuntimeError( "LoopX Host claim completed without a same-Todo ready postcondition" ) - receipt = { + digest = after.get("capture_sha256") + if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None: + raise ValueError("The guard did not return its published capture digest") + # The caller presents this same complete decision without admitting again. + return { "event": "host_execution_prepared", "goal_id": goal_id, "agent_id": agent_id, "todo_id": todo_id, - **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}), + "turn_instance_id": turn_instance_id, "execution_preparation": dict(preparation), "normal_delivery_allowed": True, + "capture_sha256": digest, + "capture_path": str(capture_directory / "decision.json"), + "claim_operation_id": claim_operation_id, } - if capture_directory is not None: - digest = after.get("capture_sha256") - if not isinstance(digest, str) or len(digest) != 64: - raise ValueError("The guard did not return its published capture digest") - receipt["capture_sha256"] = digest - 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 - # presents that source through the normal reader instead of admitting again. - return receipt def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str: diff --git a/tests/control_plane/test_codex_execution_preparation.py b/tests/control_plane/test_codex_execution_preparation.py index 153b8459df..9eac0acc58 100644 --- a/tests/control_plane/test_codex_execution_preparation.py +++ b/tests/control_plane/test_codex_execution_preparation.py @@ -16,6 +16,10 @@ GOAL = "goal-example" TODO = "todo_example" AGENT = "agent-a" +ORIGINAL_DIGEST = "a" * 64 +OBSERVED_DIGEST = "b" * 64 +READY_DIGEST = "c" * 64 +CLAIM = f"todo-claim:codex-capture:{ORIGINAL_DIGEST}" def _packet(tmp_path: Path, *, turn: str | None = "turn-example") -> dict[str, object]: @@ -55,14 +59,15 @@ def _completed(payload: dict[str, object], *, code: int = 0) -> subprocess.Compl ) -def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None: - packet = _packet(tmp_path) +@pytest.mark.parametrize("turn", ["turn-example", None]) +def test_compact_request_accepts_only_exact_host_action(tmp_path: Path, turn: str | None) -> None: + packet = _packet(tmp_path, turn=turn) request = compact_execution_preparation_request(packet) assert request == { "goal_id": GOAL, "agent_id": AGENT, "todo_id": TODO, - "turn_instance_id": "turn-example", + **({"turn_instance_id": turn} if turn else {}), "claim_args": [ "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, "--agent-id", AGENT, "--claimed-by", AGENT, @@ -78,18 +83,23 @@ def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None: assert compact_execution_preparation_request(malformed) is None -@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"]) -@pytest.mark.parametrize("turn", ["turn-example", None]) -def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path, scenario: str, turn: str | None) -> None: - request = compact_execution_preparation_request(_packet(tmp_path, turn=turn)) +def _bound_request(tmp_path: Path) -> dict: + request = compact_execution_preparation_request(_packet(tmp_path), claim_source_sha256=ORIGINAL_DIGEST) assert request is not None + return {**request, "source_sha256": OBSERVED_DIGEST} + + +@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"]) +def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path, scenario: str) -> None: + request = _bound_request(tmp_path) + capture, native = tmp_path / "ready", tmp_path / "native" calls: list[list[str]] = [] def runner(command, *, cwd, timeout): calls.append(list(command)) if "should-run" not in command: return _completed({"ok": True, "claimed": True, - "claim_operation_id": "todo-claim:goal-example:todo_example:original"}) + "claim_operation_id": CLAIM}) return _completed({ "ok": True, "goal_id": GOAL, @@ -103,19 +113,25 @@ def runner(command, *, cwd, timeout): "agent_id": AGENT, "delivery_allowed": True, }, - **({"heartbeat_receipt": {"turn_instance_id": turn}} if turn is not None else {}), + "heartbeat_receipt": {"turn_instance_id": "turn-example"}, + "capture_sha256": READY_DIGEST, }) receipt = execute_codex_execution_preparation( request, cwd=tmp_path.resolve(), scenario=scenario, + capture_directory=capture, + context_directory=native, runner=runner, ) assert receipt["event"] == "host_execution_prepared" assert receipt["normal_delivery_allowed"] is True - assert receipt["claim_operation_id"] == "todo-claim:goal-example:todo_example:original" + assert receipt["claim_operation_id"] == CLAIM + assert receipt["turn_instance_id"] == "turn-example" + assert receipt["capture_sha256"] == READY_DIGEST + assert receipt["capture_path"] == str(capture / "decision.json") assert len(calls) == 2 for call in calls: assert call[call.index("--registry") + 1] == ( @@ -123,23 +139,23 @@ def runner(command, *, cwd, timeout): ) assert call[call.index("--runtime-root") + 1] == str((tmp_path / "loopx-runtime-a").resolve()) assert calls[0][-len(request["claim_args"]):] == request["claim_args"] - if turn is not None: - assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", turn] - else: - assert calls[1][-2:] == ["--todo-id", TODO] - assert "--turn-instance-id" not in calls[1] and "turn_instance_id" not in receipt + assert calls[0][-2:] == ["--claim-operation-id", CLAIM] + assert calls[1][-12:] == [ + "--context-delivery", "on", "--codex-context-dir", str(native), "--defer-context-presentation", + "--todo-id", TODO, "--turn-instance-id", "turn-example", + "--decision-output-dir", str(capture), "--prepared-capture-reference", + ] context = codex_execution_preparation_context(receipt) assert "do not repeat" in context and '"state":"ready"' in context @pytest.mark.parametrize("changed", ["goal", "agent", "todo", "turn"]) def test_executor_rejects_changed_todo_or_turn(tmp_path: Path, changed: str) -> None: - request = compact_execution_preparation_request(_packet(tmp_path)) - assert request is not None + request = _bound_request(tmp_path) def runner(command, *, cwd, timeout): if "should-run" not in command: - return _completed({"ok": True}) + return _completed({"ok": True, "claim_operation_id": CLAIM}) return _completed({ "ok": True, "goal_id": "goal-other" if changed == "goal" else GOAL, @@ -161,5 +177,30 @@ def runner(command, *, cwd, timeout): request, cwd=tmp_path.resolve(), scenario="codex_app_ssh_goal", + capture_directory=tmp_path / "ready", + context_directory=tmp_path / "native", runner=runner, ) + + +@pytest.mark.parametrize("fault", ["unbound", "missing-source", "missing-turn", "readback-only"]) +def test_executor_does_not_run_a_claim_without_live_binding(tmp_path: Path, fault: str) -> None: + request = _bound_request(tmp_path) + capture, native = tmp_path / "ready", tmp_path / "native" + if fault == "unbound": + request = compact_execution_preparation_request(_packet(tmp_path, turn=None)) + assert request is not None + capture = native = None + elif fault == "missing-source": + request.pop("source_sha256") + elif fault == "missing-turn": + request.pop("turn_instance_id") + else: + request["readback_only"] = True + + with pytest.raises(ValueError, match="bound captured Turn"): + execute_codex_execution_preparation( + request, cwd=tmp_path.resolve(), scenario="codex_app_heartbeat", + capture_directory=capture, context_directory=native, + runner=lambda *_args, **_kwargs: pytest.fail("unbound preparation must not run a subprocess"), + ) diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 514f5fcb18..afc19f9c13 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -452,7 +452,7 @@ def hook(tool_response=None): @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("case", [ "same-execution", "released-execution", "after-ready-same-execution", - "after-ready-released-execution", "legacy-released-execution", + "after-ready-released-execution", "legacy-same-execution", "legacy-released-execution", "reencoded-released-execution", "legacy-marked-released-execution", ]) def test_native_hook_replays_the_original_claim_after_guard_timeout( @@ -563,7 +563,10 @@ def read(*args): 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"] + assert current["active"] is (not released) + assert current["lease"]["lease_epoch"] == lease["lease_epoch"] + if not released: + assert current["lease"] == lease else: assert len(claims) == 1 and guards == [turn] for observed in (result, resume):