diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 8e3d5fc6fe..0bbff0aa14 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -121,6 +121,18 @@ intentionally have no LoopX Core and must remain uncontaminated controls. ### C1 source audit: shared owners and the remaining split +Bound native execution preparation retains the ordinary managed claim reference +and captures its same-Turn ready decision before CLI compaction. The Host hands +that source to the existing saved-capture presentation/reader owner, rather than +discarding it and making the model evaluate another guard to obtain its current +contract. Goal, Agent, Todo and Turn must match before publication. Required +material reads, including an explicit complete Goal read, remain required; a +ready receipt or saved file does not fulfill them. Optional presentation failure +retains the original digest-bound recovery. Explicit off is inert, and an +unbound request keeps its original readiness route without inventing a Turn. +This is a bounded Host handoff repair, not C1 cutover, installed adoption or +qualification of model cost and long-horizon benefit. + The source already shares more than the output names suggest. `_emit_quota_result` captures the complete decision **before** `_project_quota_cli_payload` chooses ordinary CLI compaction or `buildTurnEnvelope`. The latter delegates to the diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index 878188180b..67f4497243 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -96,6 +96,14 @@ ### C1 源码核对：已有共享 owner 与剩余分叉 +绑定 Turn 的原生执行准备保留普通 managed claim 原引用，并在 CLI 压缩前捕获同一 +Turn 的 ready 决策。Host 将该原件交给既有 saved-capture 展示/reader owner，不丢弃 +它再要求模型重算 guard 才能取得当前契约。发布前须核对 Goal、Agent、Todo 和 Turn。 +必需材料读取（包括显式的完整 Goal 读取）仍须执行，ready 回执或已保存文件不能代替 +读取。可选展示失败保留原摘要绑定的恢复引用；显式 off 不产生动作，未绑定请求沿原 +readiness 路径，不补造 Turn。这只是有界的 Host 交接修复，不代表 C1 切换、已安装 +采用，或模型费用与长程收益已取得资格。 + 源码实际共享的部分比输出名称体现的更多。`_emit_quota_result` 在 `_project_quota_cli_payload` 选择普通 CLI 压缩或 `buildTurnEnvelope` **之前**， 已经捕获完整决策。后者委托 TypeScript owner，Python 模块是传输 adapter。 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 6b7a407374..9c598223b0 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -1154,10 +1154,15 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca if request is not None: workspace = Path(event.get("cwd") or "") if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve(): + capture = ( + Path(tempfile.mkdtemp(prefix="prepared-", dir=directory)) / "ready" + if request.get("turn_instance_id") else None + ) receipt = execute_codex_execution_preparation( request, cwd=workspace.resolve(), scenario=activation["scenario"], + capture_directory=capture, ) record_native_execution_preparation( directory, @@ -1165,6 +1170,21 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca source_sha256=request["source_sha256"], receipt=receipt, ) + # 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. + receipt["context"] = project_saved_context_capture(argparse.Namespace(**{ + **vars(args), "quota_command": "context-project", + "capture_file": str(source), + "capture_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "goal_id": request["goal_id"], "agent_id": request["agent_id"], + "turn_instance_id": request["turn_instance_id"], + })) existing = payload.get("hookSpecificOutput") existing_context = ( str(existing.get("additionalContext") or "") diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index a29d19100f..34251eef4a 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -6,6 +6,7 @@ """ from __future__ import annotations +import hashlib import json import subprocess from collections.abc import Callable, Mapping, Sequence @@ -124,6 +125,7 @@ def execute_codex_execution_preparation( *, cwd: Path, scenario: str, + capture_directory: Path | None = None, timeout: float = 8, runner: CommandRunner | None = None, ) -> dict[str, Any]: @@ -173,7 +175,7 @@ def execute_codex_execution_preparation( ): raise ValueError("Codex execution preparation requires a physical workspace") invoke = runner or _default_runner - _run_json( + claimed = _run_json( expected_claim, registry_path=registry_path, runtime_root=runtime_root, @@ -190,6 +192,7 @@ def execute_codex_execution_preparation( "--context-delivery", "off", "--todo-id", todo_id, *(["--turn-instance-id", turn_instance_id] if turn_instance_id else []), + *(["--decision-output-dir", str(capture_directory)] if capture_directory is not None else []), ] after = _run_json( guard, @@ -199,6 +202,13 @@ def execute_codex_execution_preparation( timeout=timeout / 2, runner=invoke, ) + if capture_directory is not None: + from .decision_capture import read_context_capture_file + + source = capture_directory / "decision.json" + _, _, after = read_context_capture_file( + str(source), hashlib.sha256(source.read_bytes()).hexdigest(), private=True, + ) preparation = after.get("execution_preparation") selected = after.get("selected_todo") receipt = after.get("heartbeat_receipt") @@ -207,12 +217,16 @@ def execute_codex_execution_preparation( and receipt.get("turn_instance_id") == turn_instance_id ) ready = ( - after.get("should_run") is True + after.get("ok") is True + and after.get("goal_id") == goal_id + and after.get("should_run") is True and after.get("normal_delivery_allowed") is True and isinstance(selected, Mapping) and selected.get("todo_id") == todo_id and isinstance(preparation, Mapping) and preparation.get("state") == "ready" + and preparation.get("todo_id") == todo_id + and preparation.get("agent_id") == agent_id and preparation.get("delivery_allowed") is True and same_turn ) @@ -220,7 +234,7 @@ def execute_codex_execution_preparation( raise RuntimeError( "LoopX Host claim completed without a same-Todo ready postcondition" ) - return { + receipt = { "event": "host_execution_prepared", "goal_id": goal_id, "agent_id": agent_id, @@ -229,6 +243,12 @@ def execute_codex_execution_preparation( "execution_preparation": dict(preparation), "normal_delivery_allowed": True, } + 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 + # presents that source through the normal reader instead of admitting again. + return receipt def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str: @@ -239,6 +259,11 @@ def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str: "rechecked the same Todo/Turn before this model step. Treat the earlier " "claim_required state as superseded by this ready receipt; do not repeat " "its claim or guard solely for that lease state. Current gates and later " - "settlement checks still apply. Receipt: " + "settlement checks still apply. " + + ("Read the ready context in this receipt for current work and settlement, " + "replacing the earlier pre-claim view. Retain its original claim reference " + "for managed progress or completion. A saved context is not proof of " + "model consumption. " if "context" in receipt else "") + + "Receipt: " + json.dumps(dict(receipt), ensure_ascii=False, separators=(",", ":")) ) diff --git a/tests/control_plane/test_codex_execution_preparation.py b/tests/control_plane/test_codex_execution_preparation.py index dd8cbd0904..153b8459df 100644 --- a/tests/control_plane/test_codex_execution_preparation.py +++ b/tests/control_plane/test_codex_execution_preparation.py @@ -78,15 +78,18 @@ def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None: assert compact_execution_preparation_request(malformed) is None -def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path) -> None: - request = compact_execution_preparation_request(_packet(tmp_path)) +@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)) assert request is not None 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}) + return _completed({"ok": True, "claimed": True, + "claim_operation_id": "todo-claim:goal-example:todo_example:original"}) return _completed({ "ok": True, "goal_id": GOAL, @@ -100,18 +103,19 @@ def runner(command, *, cwd, timeout): "agent_id": AGENT, "delivery_allowed": True, }, - "heartbeat_receipt": {"turn_instance_id": "turn-example"}, + **({"heartbeat_receipt": {"turn_instance_id": turn}} if turn is not None else {}), }) receipt = execute_codex_execution_preparation( request, cwd=tmp_path.resolve(), - scenario="codex_app_ssh_goal", + scenario=scenario, 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 len(calls) == 2 for call in calls: assert call[call.index("--registry") + 1] == ( @@ -119,12 +123,17 @@ 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"] - assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", "turn-example"] + 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 context = codex_execution_preparation_context(receipt) assert "do not repeat" in context and '"state":"ready"' in context -def test_executor_rejects_changed_todo_or_turn(tmp_path: Path) -> None: +@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 @@ -133,15 +142,18 @@ def runner(command, *, cwd, timeout): return _completed({"ok": True}) return _completed({ "ok": True, + "goal_id": "goal-other" if changed == "goal" else GOAL, "should_run": True, "normal_delivery_allowed": True, - "selected_todo": {"todo_id": "todo-other"}, + "selected_todo": {"todo_id": "todo-other" if changed == "todo" else TODO}, "execution_preparation": { "schema_version": "execution_preparation_v0", "state": "ready", + "todo_id": TODO, + "agent_id": "agent-other" if changed == "agent" else AGENT, "delivery_allowed": True, }, - "heartbeat_receipt": {"turn_instance_id": "turn-other"}, + "heartbeat_receipt": {"turn_instance_id": "turn-other" if changed == "turn" else "turn-example"}, }) with pytest.raises(RuntimeError, match="same-Todo ready"): diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index bc5733821a..214ec8c6f8 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -4,6 +4,7 @@ import json import os from pathlib import Path +import shlex import subprocess import sys @@ -174,6 +175,106 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard( assert after["normal_delivery_allowed"] is True +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("scenario", ["codex_app_heartbeat", "codex_app_ssh_goal"]) +def test_native_hook_returns_readable_ready_source_and_original_claim( + tmp_path, monkeypatch, provider, scenario, +): + from loopx.control_plane.turn_driver import codex_context + + tmp_path = tmp_path.resolve() + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("CODEX_THREAD_ID", "preparation-session") + native = tmp_path / "native" + native.mkdir(mode=0o700) + event = {"session_id": "preparation-session", "cwd": str(tmp_path), + "hook_event_name": "SessionStart", "source": "startup"} + 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."} + before = _run(tmp_path, monkeypatch, provider, [], "--runtime-profile", scenario, + "--turn-instance-id", "preparation-turn", "--todo-id", TODO, todos=[original]) + 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.project_codex_context(before, directory=str(native), capture_directory=capture, + goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="before-claim") + 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, "--codex-context-dir", str(native)] + hook_event = {**event, "hook_event_name": "PostToolUse"} + log = tmp_path / "runtime" / "goals" / GOAL / "rollout-event-log.jsonl" + unchanged = log.read_bytes() + disabled = subprocess.run(["off" if item == "on" else item for item in hook], + input=json.dumps(hook_event), cwd=tmp_path, capture_output=True, text=True, timeout=30) + assert disabled.returncode == 0 and json.loads(disabled.stdout) == {} + assert log.read_bytes() == unchanged + completed = subprocess.run(hook, input=json.dumps(hook_event), cwd=tmp_path, + capture_output=True, text=True, timeout=30) + assert completed.returncode == 0, completed.stderr + additional = json.loads(completed.stdout)["hookSpecificOutput"]["additionalContext"] + receipt = json.loads(additional.rsplit("Receipt: ", 1)[1]) + assert receipt["event"] == "host_execution_prepared" + 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:") + ready = receipt["context"] + read = shlex.split(ready["recovery"]["command"]) + pointers = ["/execution_preparation", "/selected_todo", "/heartbeat_receipt", "/interaction_contract/cli_channel", + "/interaction_contract/agent_channel/work_context", "/interaction_contract/agent_channel/required_reads"] + for pointer in pointers: + read.extend(("--select", pointer)) + completed = subprocess.run(read, cwd=tmp_path, capture_output=True, text=True, timeout=30) + assert completed.returncode == 0, completed.stderr + page = json.loads(completed.stdout) + assert page["eof"] is True and page["offset"] == 0 + selected = {item["pointer"]: item["value"] for item in json.loads(page["content"])["selections"]} + assert selected["/execution_preparation"]["state"] == "ready" + assert selected["/heartbeat_receipt"]["turn_instance_id"] == "preparation-turn" + plan = selected["/interaction_contract/cli_channel"]["settlement_plan"] + assert plan["identity"]["todo_id"] == TODO and plan["identity"]["turn_instance_id"] == "preparation-turn" + sources = selected["/interaction_contract/agent_channel/work_context"]["sources"] + task = next(item["content"]["todo"] for item in sources if item["source"] == "selected_todo") + # The v0 owner may deduplicate short text already in selected_todo. + # The complete request must survive either original placement, without + # replacing it with the selected summary when that summary is shorter. + assert task.get("text", selected["/selected_todo"]["text"]) == original["text"] + assert task["note"] == original["note"] + goal_read = next(item for item in selected["/interaction_contract/agent_channel/required_reads"] + if item["source"] == "goal_state") + goal = subprocess.run(shlex.split(goal_read["command"]), capture_output=True, text=True, timeout=30) + assert goal.returncode == 0 and goal.stdout == goal_text + code, current = run_json_cli_result("todo", "list", "--goal-id", GOAL, "--todo-id", TODO, + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", cwd=tmp_path) + assert code == 0 + code, progress = run_json_cli_result("todo", "update", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claim-operation-id", claim_reference, + "--update-operation-id", "prepared-progress", + "--update-expected-provider-revision", current["authority_read"]["provider_revision"], + "--note", original["note"] + " Checkpoint verified.", + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", cwd=tmp_path) + assert code == 0 and progress["ok"], progress + events = log.read_bytes() + repeat = subprocess.run(hook, input=json.dumps(hook_event), cwd=tmp_path, + capture_output=True, text=True, timeout=30) + assert repeat.returncode == 0 and "host_execution_prepared" not in repeat.stdout + assert log.read_bytes() == events + guards = [json.loads(line) for line in events.splitlines() + if json.loads(line)["event_kind"] == "quota_should_run"] + assert len(guards) == 2 + assert [item["details"]["delivery_allowed"] for item in guards] == [False, True] + source, _ = codex_context.read_codex_context_reference(str(native), goal_id=GOAL, agent_id=AGENT) + source = Path(source) + source.write_bytes(source.read_bytes() + b" ") + corrupted = subprocess.run(read, cwd=tmp_path, capture_output=True, text=True, timeout=30) + assert corrupted.returncode != 0 and json.loads(corrupted.stdout)["ok"] is False + assert log.read_bytes() == events + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("multiple_candidates", [False, True]) def test_projected_claim_prepares_the_same_receipt_bound_turn(