test(control-plane): separate canonical readback from execution admission

docs/status-data-contract.md

@@ -688,6 +688,20 @@ the live quota-backed queue item for the same `goal_id` when one exists, under
 `global_registry_shadow_findings`, so stale source shadows do not become a second
 quota health blocker while the fact remains visible in `global_registry.findings`.
 
+A Goal can therefore have both a health finding and a live read-model item in
+`attention_queue.items`. Match the required projection as well as `goal_id`:
+the first row for a Goal is not necessarily its Todo or Goal-channel view.
+When the declared Goal document is missing, available canonical Todo, lease and
+acceptance observations remain readable; the `state_file_missing` finding still
+blocks automatic delivery. Readback never reconstructs the document or replaces
+its intent and stop conditions. Restoring the document permits a fresh admission
+check; provider failure still refuses stale display fallback.
+
+同一个 Goal 可以同时有健康告警和权威状态的只读投影。读取 Todo 或 Goal channel
+时须同时匹配 `goal_id` 和所需投影字段，不能假定第一行就是该视图。Goal 文档缺失时，
+可用的 canonical Todo、lease 和验收仍可回读，但不授予自动执行，也不重建文档；恢复
+文档后重新准入，provider 不可用时继续拒绝旧显示数据回退。
+
 ## Contract Health
 
 `contract.ok=false` means the UI should show a blocking health state before

tests/control_plane/test_auxiliary_scoped_gate.py

@@ -98,7 +98,14 @@ def emitted():
                 blocks_agent=AGENT_ID, agent_id=AGENT_ID)
         rc, current = _run_cli(registry, runtime, *guard_args)
         if gate_change == "primary_blocked":
-            assert rc == 1 and current["error_code"] == "heartbeat_receipt_identity_conflict", current
+            # A blocked primary still owes its original Turn a no-spend
+            # closeout. Recovery is not renewed delivery or monitor admission.
+            assert rc == 0, current
+            assert current["effective_action"] == "unsettled_host_turn_recovery"
+            assert current["heartbeat_receipt"]["settlement_identity"] == guard["heartbeat_receipt"]["settlement_identity"]
+            assert current["execution_obligation"]["delivery_allowed"] is False
+            assert current["execution_obligation"]["contract_obligation"] == "close_bound_wait_without_spend"
+            assert "auxiliary_monitor_poll" not in current["interaction_contract"]["cli_channel"]
         else:
             assert rc == 0, current
             offered = current["interaction_contract"]["cli_channel"].get("auxiliary_monitor_poll", {})
@@ -111,6 +118,24 @@ def emitted():
         assert _spend_run_count(runtime) == 0
         row = list_goal_todos(registry_path=source, goal_id=GOAL_ID, todo_id=DUE_MONITOR_TODO_ID)["todo"]
         assert row.get("result_hash") is None
+        if gate_change == "primary_blocked":
+            refresh_args = ("refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+                "--todo-id", TODO_ID, "--turn-instance-id", turn,
+                "--classification", "blocked_primary_observation_denied",
+                "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_gap",
+                "--progress-result-class", "blocked", "--progress-blocker-id", "blocker:primary-unavailable",
+                "--progress-evidence-id", "evidence:auxiliary-observation-denied",
+                "--no-global-sync", "--suppress-external-sinks")
+            for replay in (False, True):
+                rc, closed = _run_cli(registry, runtime, *refresh_args, cwd=project)
+                assert rc == 0, closed
+                assert closed["settlement_progress"]["state"] == "settled"
+                assert closed["settlement_progress"]["closeout_kind"] == "typed_blocked_writeback_no_spend"
+                if replay:
+                    assert closed["idempotent_replay"] is True
+            assert _spend_run_count(runtime) == 0
+            primary = list_goal_todos(registry_path=source, goal_id=GOAL_ID, todo_id=TODO_ID)["todo"]
+            assert primary["status"] == "blocked"
         return
     rc, observed = emitted()
     assert rc == 0, observed

tests/control_plane/test_canonical_goal_channel_coordination.py

@@ -130,8 +130,15 @@ def cli(*args):
     def channel():
         status = cli('status', '--goal-id', GOAL_ID)
         items = status['attention_queue']['items']
-        item = next(row for row in items if row['goal_id'] == GOAL_ID)
-        return item['goal_channel_projection']
+        channels = [row['goal_channel_projection'] for row in items
+                    if row['goal_id'] == GOAL_ID and 'goal_channel_projection' in row]
+        assert len(channels) == 1, items
+        if not state.exists():
+            # The missing Goal document blocks execution independently of the
+            # canonical ownership read model, which remains inspectable.
+            assert any(row['goal_id'] == GOAL_ID and row['status'] == 'state_file_missing'
+                       for row in items)
+        return channels[0]
 
     def assert_current(view):
         assert view['coordination_observation']['source_authority'] == provider + '_v0'

tests/control_plane/test_capability_gate_cli.py

@@ -32,8 +32,7 @@ def test_real_quota_capability_gate_does_not_enable_repair_outputs(tmp_path, mon
         projection, _ = build_runtime_shadow_source_snapshot(goal=goal, runtime_root=runtime,
                                                              state_path=state, registry_path=registry)
         initialize_canonical_authority(runtime, "goal-a", projection, state_path=state, provider=provider)
-        state.unlink()
-    before = state.read_bytes() if state.exists() else None
+    before = state.read_bytes()
     authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a")
     args = ["quota", "should-run", "--goal-id", "goal-a", "--agent-id", "agent-a",
             "--include-detail", "agent-todos", "--scan-path", str(tmp_path)]
@@ -48,5 +47,26 @@ def test_real_quota_capability_gate_does_not_enable_repair_outputs(tmp_path, mon
     ids = [row["todo_id"] for row in gate["runnable_candidates"]]
     assert ids == (["todo_external", "todo_repair"] if network_available else ["todo_repair"])
     assert gate["repair_candidate_count"] == (0 if network_available else 1)
-    assert (state.read_bytes() if state.exists() else None) == before
+    assert state.read_bytes() == before
+
+    # Capability availability cannot replace the Goal's missing intent and
+    # stop conditions. Canonical Todos remain readable without granting work.
+    state.unlink()
+    code, missing = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime)
+    assert code == 1, missing
+    assert missing["status_health_ok"] is False
+    assert missing["should_run"] is False
+    assert missing["normal_delivery_allowed"] is False
+    assert missing["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    assert not state.exists(), "observation must not reconstruct the Goal document"
+    if provider != "markdown":
+        code, listed = run_json_cli_result("todo", "list", "--goal-id", "goal-a",
+                                          registry_path=registry, runtime_root=runtime)
+        assert code == 0, listed
+        assert {row["todo_id"] for row in listed["todos"]} == {"todo_external", "todo_repair", "todo_owner"}
+    state.write_bytes(before)
+    code, recovered = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime)
+    assert code == 0, recovered
+    assert [row["todo_id"] for row in recovered["capability_gate"]["runnable_candidates"]] == ids
+    assert state.read_bytes() == before
     assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == authority

tests/control_plane/test_goal_acceptance_runtime.py

@@ -81,13 +81,21 @@ def test_public_status_cli_retains_safe_contract_sidecar_without_extra_source(tm
     result = subprocess.run([sys.executable, "-m", "loopx.cli", "--registry", str(registry),
         "--format", "json", "status", "--goal-id", goal["id"]], capture_output=True, text=True, timeout=60)
     assert result.returncode == 0, result.stdout + result.stderr
-    items = json.loads(result.stdout)["attention_queue"]["items"]
-    item = next(item for item in items if item["goal_id"] == goal["id"])
+    status = json.loads(result.stdout)
+    items = status["attention_queue"]["items"]
+    summaries = [item for item in items if item["goal_id"] == goal["id"] and "agent_todos" in item]
+    assert len(summaries) == 1, items
+    item = summaries[0]
     contract = item["agent_todos"]["goal_acceptance_contract"]
     assert contract["status"] == "held"
     assert contract["criteria"] == [{"id": "criterion-a", "description": "The check passes"}]
     assert item["agent_todos"]["first_executable_items"] == []
     assert "validation_argv" not in json.dumps(contract)
+    assert any(row["goal_id"] == goal["id"] and row["status"] == "state_file_missing" for row in items)
+    decision = build_quota_should_run(status, goal_id=goal["id"])
+    assert decision["status_health_ok"] is False
+    assert decision["should_run"] is False
+    assert decision["normal_delivery_allowed"] is False
     assert not state.exists()