diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index a0aa4ec56e..6bb4688325 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -130,6 +130,15 @@ the existing context-delivery owner. C1 must converge the remaining composition and consumer assumptions, rather than introduce another capture service or rewrite already shared delta/native-resume logic. +Optional host polling is part of that original observation. The source-fidelity +slice records `--record-host-poll` once before capture and reuses its receipt in +ordinary CLI, envelope and native-context delivery. Recovery reads do not poll +again. A later capture/handoff failure retains the already-recorded observation +in its error response; it does not imply missing admission or authorize replay. +Default-off behavior and unreadable receipt diagnostics remain unchanged. This +repairs capture completeness, not the unified writer/consumer cutover or a +long-horizon cost-benefit result. + A separate native-presentation slice reuses the committed full recovery as its reader target, avoiding an unused second wire publication. Delta and unwrapped inputs retain their own immutable files and the same reader/lifecycle owners. 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 2f3fbb1b95..41ac2b190c 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 @@ -104,6 +104,12 @@ Quota capture 与 Turn journal capture 都调用 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设，不另建 capture service， 也不重写已经共享的差分与 native-resume 逻辑。 +可选 host poll 也是该次原始观测的一部分。来源保真切片在 capture 前仅记录一次 +`--record-host-poll`，普通 CLI、envelope 与原生上下文交付复用同一回执；恢复读取不再 +轮询。后续 capture/handoff 失败的错误响应仍携带已记录观测，不表示缺少准入，也不 +授权重放。默认关闭行为和不可读回执诊断不变。这只修复 capture 完整性，不代表统一 +writer/consumer 切换或长程成本收益已取得资格。 + 独立的原生展示切片复用已提交的完整恢复原件作为 reader 目标，避免再次发布未使用的 wire 副本。Delta 与未包装输入仍保留各自的不可变文件，reader/lifecycle owner 不变。 同一 107,289 字节合成来源的 capture 写入从 216,754 降为 107,289 字节；完整 reader diff --git a/docs/heartbeat-automation-prompt.md b/docs/heartbeat-automation-prompt.md index 3095bbeafc..b9f8f8c325 100644 --- a/docs/heartbeat-automation-prompt.md +++ b/docs/heartbeat-automation-prompt.md @@ -466,6 +466,16 @@ use the explicit original capture reference. If preparation fails before that can be established, or invalidation also fails, the CLI reports `quota_decision_transport_failed` with the saved decision and receipt. Do not rerun admission to recover a presentation failure. + +When the caller also selects `--record-host-poll`, the existing liveness owner +records that observation once before the original capture. The saved decision, +ordinary/envelope output and native recovery view retain the same +`host_poll_receipt`, including an unreadable-receipt diagnostic. A capture or +handoff failure still reports an already-recorded poll; its count can advance +even when presentation fails. Reading the original evidence never records +another poll, launches work or spends quota. Without this flag, capture and +context delivery do not create a liveness receipt. + `off` creates no native journal, hook configuration or input observation; directories alone never turn it on. This declaration check is not host authentication, execution authority or proof of retained input. Native diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index f2731a0ff0..2e3d752737 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -311,6 +311,12 @@ def _emit_quota_result( payload = {**payload, "replan_read_path": str(capture_directory / "decision.json"), "replan_read_argv": current_loopx_cli_argv()} + if args.quota_command == "should-run" and context is not None: + # Capture the one liveness observation with the decision that produced + # it. Presentation and evidence recovery must not record it again. + attach_host_poll_receipt( + context.status_payload, args, payload, registry_path=registry_path, + ) try: capture_decision(capture_directory, payload) if (payload.get("ok") is False and context_delivery_enabled and capture_directory @@ -343,13 +349,10 @@ def _emit_quota_result( str(capture_directory / "decision.json") if capture_directory is not None else None ), ) - if args.quota_command == "should-run" and context is not None: - attach_host_poll_receipt( - context.status_payload, - args, - payload, - registry_path=registry_path, - ) + if args.quota_command == "should-run" and "host_poll_receipt" in captured_payload: + # Legacy envelope presentation omits this source field. Preserve its + # existing visible receipt by copying the observation, not polling. + payload["host_poll_receipt"] = captured_payload["host_poll_receipt"] succeeded = bool(payload.get("ok")) # Handoff releases the immutable original before any presentation. Native # actual-input observation owns its projection when explicitly selected; diff --git a/loopx/cli_commands/quota_failure_report.py b/loopx/cli_commands/quota_failure_report.py index 839f4def49..ccc32779c5 100644 --- a/loopx/cli_commands/quota_failure_report.py +++ b/loopx/cli_commands/quota_failure_report.py @@ -355,7 +355,7 @@ def quota_decision_transport_failure_payload( } if saved is not None and saved.is_file(): payload["captured_decision"] = {"path": str(saved)} - for key in ("heartbeat_receipt", "rollout_event"): + for key in ("heartbeat_receipt", "rollout_event", "host_poll_receipt"): if key in committed: payload[key] = committed[key] return payload diff --git a/tests/control_plane/test_worker_observation_recovery_cli.py b/tests/control_plane/test_worker_observation_recovery_cli.py index 2d90d85deb..12f9acb740 100644 --- a/tests/control_plane/test_worker_observation_recovery_cli.py +++ b/tests/control_plane/test_worker_observation_recovery_cli.py @@ -9,6 +9,7 @@ import pytest +import loopx from canonical_authority_fixture import isolate_sqlite_runtime from research_worker_readback_fixture import AGENT, GOAL, _write_authority from loopx.control_plane.quota.host_poll_receipts import ( @@ -22,7 +23,8 @@ def _worker_fixture(tmp_path, monkeypatch, provider): isolate_sqlite_runtime(tmp_path, monkeypatch) - monkeypatch.setenv("PYTHONPATH", str(REPO)) + # Subprocesses exercise this owner's installation, including wheel tests. + monkeypatch.setenv("PYTHONPATH", str(Path(loopx.__file__).resolve().parent.parent)) monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "worker-state")) runtime, _ = _write_authority(tmp_path, provider=provider) registry = tmp_path / "registry.json" @@ -47,12 +49,12 @@ def _run_worker(runtime, registry, count, host, agent): ], cwd=host.parent / "project", capture_output=True, text=True, timeout=120) -def _poll(runtime, registry, agent, *, record=False): +def _poll(runtime, registry, agent, *, record=False, extra=(), profile="generic_cli"): result = subprocess.run([ sys.executable, "-m", "loopx.cli", "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "quota", "should-run", "--goal-id", GOAL, "--agent-id", agent, - "--runtime-profile", "generic_cli", *(["--record-host-poll"] if record else []), + "--runtime-profile", profile, *(["--record-host-poll"] if record else []), *extra, ], cwd=registry.parent / "project", capture_output=True, text=True, timeout=90) assert result.returncode == 0, result.stdout + result.stderr payload = json.loads(result.stdout) @@ -60,6 +62,129 @@ def _poll(runtime, registry, agent, *, record=False): return payload +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("wire", ["quota", "envelope", "native"]) +@pytest.mark.parametrize("observation", ["recorded", "unreadable", "off"]) +def test_original_capture_retains_the_emitted_poll_observation( + tmp_path, monkeypatch, provider, wire, observation, +): + """One liveness observation must survive original-evidence recovery.""" + import hashlib + + runtime, registry, count, _ = _worker_fixture(tmp_path, monkeypatch, provider) + receipt = receipt_path_for_goal({"id": GOAL}, tmp_path / "project" / "ACTIVE_GOAL_STATE.md") + damaged = b'{"schema_version":"unknown","worker_observations":{"unresolved":true}}' + if observation == "unreadable": + receipt.parent.mkdir(parents=True, exist_ok=True) + receipt.write_bytes(damaged) + capture = tmp_path / "original-capture" + options = ["--turn-instance-id", "capture-poll", "--decision-output-dir", str(capture), + "--context-delivery", "on" if wire == "native" else "off"] + if wire == "envelope": + options.append("--turn-envelope") + if wire == "native": + native = tmp_path / "native-context" + native.mkdir(mode=0o700) + options.extend(["--codex-context-dir", str(native)]) + emitted = _poll(runtime, registry, AGENT, record=observation != "off", extra=options, + profile="codex_app_heartbeat" if wire == "native" else "generic_cli") + original = (capture / "decision.json").read_bytes() + source = json.loads(original) + assert source["quota"]["spent_slots"] == 0 + assert not count.exists(), "A poll or evidence read must not dispatch the Host" + if observation == "off": + assert "host_poll_receipt" not in source + assert "host_poll_receipt" not in emitted + assert not receipt.exists() + return + if observation == "recorded": + stored = json.loads(receipt.read_bytes()) + expected = {"recorded": True, "poll_count": 1, + "last_poll_at": stored["last_poll_at"], + "expected_continuation": stored["expected_continuation"]} + assert stored["poll_count"] == 1 + else: + expected = {"recorded": False, + "error": "host poll receipt is unreadable; repair its source before updating observations"} + assert receipt.read_bytes() == damaged + assert source["host_poll_receipt"] == expected + before = receipt.read_bytes() + if wire == "native": + # Read precisely the immutable recovery view published to this caller. + reader = shlex.split(emitted["recovery"]["command"]) + elif wire == "envelope": + from loopx.control_plane.quota.turn_envelope import build_turn_envelope + + # Replan readers bind the full observation, including this receipt. + # Their hash changes without changing the selected work or debit. + without_poll = {key: value for key, value in source.items() if key != "host_poll_receipt"} + prior = build_turn_envelope(without_poll) + actual = emitted["action_signature"] + for key in ("decision", "should_run", "effective_action"): + assert emitted[key] == prior[key] + assert emitted["writeback"]["settlement_identity"] == prior["writeback"]["settlement_identity"] + assert actual["source_decision_hash"] != prior["action_signature"]["source_decision_hash"] + assert actual == build_turn_envelope(source)["action_signature"] + review = subprocess.run(shlex.split(emitted["replan_action_packet"]["review"]), + check=True, capture_output=True, text=True, timeout=90) + assert json.loads(review.stdout)["source_hash"] == actual["source_decision_hash"] + reader = shlex.split(emitted["detail_ref"]["full_decision"]) + else: + reader = [sys.executable, "-m", "loopx.cli", "--format", "json", "quota", "context-read", + "--capture-file", str(capture / "decision.json"), + "--capture-sha256", hashlib.sha256(original).hexdigest()] + if wire == "envelope": + recovered = json.loads(subprocess.run(reader, check=True, capture_output=True).stdout) + assert recovered["host_poll_receipt"] == expected + else: + page = subprocess.run([*reader, "--select", "/host_poll_receipt"], + check=True, capture_output=True, text=True, timeout=90) + recovered = json.loads(page.stdout) + assert recovered["ok"] is True and recovered["eof"] is True + assert json.loads(recovered["content"])["selections"][0]["value"] == expected + if wire != "native": + assert emitted["host_poll_receipt"] == expected + assert receipt.read_bytes() == before, "Recovery must not record a second poll" + assert (capture / "decision.json").read_bytes() == original + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_failed_capture_discloses_the_already_recorded_poll( + tmp_path, monkeypatch, capsys, provider, +): + from unittest.mock import patch + + from loopx.cli import main + from loopx.cli_commands import quota, quota_capture + + runtime, registry, count, _ = _worker_fixture(tmp_path, monkeypatch, provider) + original_capture = quota_capture.capture_decision + + def failed_publication(directory, payload): + # Only the capture publisher fails; admission and receipt IO are real. + with patch.object(quota_capture.os, "replace", side_effect=OSError("synthetic capture outage")): + original_capture(directory, payload) + + monkeypatch.setattr(quota, "capture_decision", failed_publication) + monkeypatch.chdir(tmp_path / "project") + capture = tmp_path / "failed-capture" + code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + "--runtime-profile", "generic_cli", "--turn-instance-id", "capture-failure", + "--record-host-poll", "--decision-output-dir", str(capture), + "--context-delivery", "off"]) + failed = json.loads(capsys.readouterr().out) + assert code == 1 and failed["status"] == "decision_transport_failed" + assert failed["should_run"] is False and "captured_decision" not in failed + assert failed["heartbeat_receipt"]["turn_instance_id"] == "capture-failure" + receipt = receipt_path_for_goal({"id": GOAL}, tmp_path / "project" / "ACTIVE_GOAL_STATE.md") + stored = json.loads(receipt.read_bytes()) + assert stored["poll_count"] == 1 + assert failed["host_poll_receipt"] == {"recorded": True, "poll_count": 1, + "last_poll_at": stored["last_poll_at"], "expected_continuation": stored["expected_continuation"]} + assert not list(capture.iterdir()) and not count.exists() + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_canonical_quota_identity_survives_worker_main_and_alias_restart(tmp_path, monkeypatch, provider): runtime, registry, count, host = _worker_fixture(tmp_path, monkeypatch, provider)