fix(quota): preserve host poll observations in decision captures

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.

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

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

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;

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

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,19 +49,142 @@ 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)
     assert payload["ok"] is True, payload
     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)