test(control-plane): preserve checkpoint fences and frontier recovery

tests/control_plane/test_checkpoint_provider_fence.py

@@ -15,7 +15,9 @@
 from loopx.control_plane.effect_runtime import EffectRuntimeConflict
 from loopx.control_plane.goals import checkpoint_context_io as context_io
 from loopx.control_plane.projects.registry_codec import (
+    ProjectRegistryProtocolError,
     load_project_registry,
+    load_registry as load_runtime_registry,
     source_session_registry_transaction,
 )
 from tests.control_plane.test_checkpoint_read_context import _missing
@@ -162,7 +164,7 @@ def test_exact_source_checkpoint_commits_only_for_the_current_goal_instance(
     tmp_path,
     monkeypatch,
 ):
-    from loopx import state_refresh
+    from loopx import history, state_refresh
 
     current = _missing(tmp_path / "current")
     stale = _missing(tmp_path / "stale")
@@ -173,8 +175,14 @@ def test_exact_source_checkpoint_commits_only_for_the_current_goal_instance(
 
     monkeypatch.setattr(context_io, "load_registry", load_project_registry)
     monkeypatch.setattr(state_refresh, "load_registry", load_project_registry)
+    # This fixture exercises the exact-source commit fence below runtime
+    # admission. Include late-bound Goal-facts reads in that fixture seam;
+    # the real generic reader must still refuse this lifecycle-only profile.
+    monkeypatch.setattr(history, "load_registry", load_project_registry)
 
     _, current_runtime, current_registry, _, _, current_original = current
+    with pytest.raises(ProjectRegistryProtocolError, match="lifecycle-only profile"):
+        load_runtime_registry(current_registry)
     current_context = context_io.read_checkpoint_context(
         registry_path=current_registry,
         runtime_root_override=str(current_runtime),
@@ -199,6 +207,17 @@ def test_exact_source_checkpoint_commits_only_for_the_current_goal_instance(
         .splitlines()[-1]
     )
     assert committed_row["goal_ref"] == goal_ref
+    current_index = current_runtime / "goals" / GOAL_ID / "runs" / "index.jsonl"
+    committed_bytes = current_index.read_bytes()
+    replay = refresh(
+        current_registry,
+        current_runtime,
+        current_context["read_context_id"],
+        goal_ref=goal_ref,
+    )
+    assert replay["idempotent_replay"] is True
+    assert current_index.read_bytes() == committed_bytes
+    assert _spend_run_count(current_runtime) == 0
 
     _, stale_runtime, stale_registry, _, _, _ = stale
     stale_context = context_io.read_checkpoint_context(
@@ -225,6 +244,7 @@ def test_exact_source_checkpoint_commits_only_for_the_current_goal_instance(
         )
     assert rejected.value.diagnostic_code == "stale_goal_instance"
     assert stale_index.read_bytes() == before
+    assert _spend_run_count(stale_runtime) == 0
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])

tests/control_plane/test_goal_frontier_full_source_counts.py

@@ -167,28 +167,54 @@ def test_real_cli_frontier_preserves_canonical_counts_after_display_loss(tmp_pat
     initialize_canonical_authority(runtime, goal["id"], build_todo_runtime_shadow_projection(
         goal_id=goal["id"], todos=fields["agent_todos"]["items"], handoff_mode="soft_claim"),
         state_path=state, provider=provider)
-    state.unlink()
+    original_state = state.read_bytes()
+    quota_args = ("quota", "should-run", "--goal-id", goal["id"],
+        "--agent-id", "agent-a", "--scan-path", str(tmp_path), "--include-detail", "agent-todos")
     try:
+        code, intact = run_json_cli_result(*quota_args, registry_path=registry, runtime_root=runtime)
+        assert code == 0, intact
+        state.unlink()
         code, listed = run_json_cli_result("todo", "list", "--goal-id", goal["id"], "--role", "agent",
             "--limit", "1", "--thin", registry_path=registry, runtime_root=runtime)
         assert code == 0, listed
         assert len(listed["todos"]) == 1
         assert listed["agent_todos"]["deferred_work_counts"]["ready"] == 12
-        code, packet = run_json_cli_result("quota", "should-run", "--goal-id", goal["id"],
-            "--agent-id", "agent-a", "--scan-path", str(tmp_path), "--include-detail", "agent-todos",
-            registry_path=registry, runtime_root=runtime)
-        assert code == 0, packet
+        code, packet = run_json_cli_result(*quota_args, registry_path=registry, runtime_root=runtime)
+        # Canonical work survives display loss, but counts cannot supply the
+        # missing Goal intent or authorize execution.
+        assert code == 1, packet
+        assert packet["status_health_ok"] is False
+        assert packet["should_run"] is False
+        assert packet["normal_delivery_allowed"] is False
         frontier = packet["goal_frontier_projection"]
         assert frontier["normalized_progress"]["agent_advancement_open_count"] == 12
         assert frontier["deferred_successors"]["ready_count"] == 12
         assert frontier["deferred_successors"]["blocked_count"] == 0
         assert frontier["deferred_successors"]["counts_complete"] is True
+        for field in ("normalized_progress", "deferred_successors"):
+            assert frontier[field] == intact["goal_frontier_projection"][field]
         assert not state.exists()
-        (runtime / "authority" / f"{provider}-v0").rename(runtime / "unavailable-provider")
+        state.write_bytes(original_state)
+        code, recovered = run_json_cli_result(*quota_args, registry_path=registry, runtime_root=runtime)
+        assert code == 0, recovered
+        assert state.read_bytes() == original_state
+        for field in ("normalized_progress", "deferred_successors"):
+            assert recovered["goal_frontier_projection"][field] == frontier[field]
+        state.unlink()
+        authority = runtime / "authority" / f"{provider}-v0"
+        unavailable = runtime / "unavailable-provider"
+        authority.rename(unavailable)
         code, failed = run_json_cli_result("todo", "list", "--goal-id", goal["id"],
             registry_path=registry, runtime_root=runtime)
         assert code != 0
         assert failed["legacy_fallback_used"] is False
         assert not state.exists()
+        unavailable.rename(authority)
+        code, restored = run_json_cli_result("todo", "list", "--goal-id", goal["id"], "--role", "agent",
+            "--limit", "1", "--thin", registry_path=registry, runtime_root=runtime)
+        assert code == 0, restored
+        assert len(restored["todos"]) == 1
+        assert restored["agent_todos"]["deferred_work_counts"]["ready"] == 12
+        assert not state.exists()
     finally:
         restart_effect_runtime()