Resolve compact plan history before considering a new review

Local checkpoint, NOT_REVIEWED. Reuse the exact wave source before deciding whether a recorded subject is absent. Compact history without its source remains unavailable, never a new paid cycle or synthetic verdict. The real nine-wave regression preserves 27 physical calls and nine paid cycles. Focused 218 tests and Ruff passed; root owns full phase verification and review. Not published.

ouroboros/tools/plan_review.py

@@ -525,11 +525,13 @@ async def _run_plan_review_async(ctx: ToolContext, request: _PlanRequest, *, col
     # The declaration wraps the builder ONLY when non-empty: zero-arg stubs of the builder stay valid.
     slots_fn = (lambda: _plan_review_slots(default_effort=request.reviewer_effort)) if request.reviewer_effort else _plan_review_slots
     existing = plan_review_wave(state, fingerprint)
-    if existing is not None and not isinstance(existing.get("spec"), dict):
-        existing = None  # C-09: a COMPACTED row (no frozen spec) is never authority
     if existing is not None:
         try:
+            # A compact index is not authority; resolve its retained exact source
+            # before deciding that this subject needs another paid review.
             existing = _authority_wave(state_root, task_id, existing)
+            if not isinstance(existing.get("spec"), dict):
+                raise PlanReviewSourceUnavailable("Recorded plan has no complete spec source")
             historical = _completed_historical_feedback(ctx, existing)
         except (OSError, ValueError, json.JSONDecodeError):
             return _plan_unavailable(
@@ -995,5 +997,3 @@ def _apply_disposition(ctx: ToolContext, disposition: dict) -> str:
     )
     return _publish_rendered_wave(ctx, stored, cap=cap, cycles_paid=cycles_paid,
                                   enforcement=enforcement, notes=list(closure["notes"]))
-
-# ------------------------------------------------------------------------ rendering

ouroboros/tools/plan_review_artifacts.py

@@ -105,7 +105,7 @@ def authority_wave(drive_root: Any, task_id: str, hot_wave: Optional[dict]) -> O
         return None
     ref = hot_wave.get("wave_artifact") if isinstance(hot_wave.get("wave_artifact"), dict) else {}
     if not ref:
-        if hot_wave.get("spec_in_artifact") or hot_wave.get("spec_body_truncated"):
+        if hot_wave.get("compact") or hot_wave.get("spec_in_artifact") or hot_wave.get("spec_body_truncated"):
             raise PlanReviewSourceUnavailable("PLAN_REVIEW_SOURCE_UNAVAILABLE: full spec has no artifact reference")
         return hot_wave
     if not drive_root or not task_id:

tests/test_plan_review_historical_replay.py

@@ -158,3 +158,57 @@ def test_historical_text_cannot_create_another_host_control_line(tmp_path):
                                   cached=True, historical_feedback=feedback)
     assert '> PLAN_REVIEW_CONTROL_JSON:' in shown
     assert _parse_plan_review_control(shown) == (old["aggregate"], old["closed"])
+
+
+@pytest.mark.parametrize("missing", [None, "wave", "spec", "legacy_no_sources"])
+def test_compacted_historical_wave_resolves_sources_before_any_new_send(harness, monkeypatch, missing):  # noqa: F811
+    from ouroboros import review_custody
+
+    monkeypatch.setenv("OUROBOROS_REVIEW_MAX_CYCLES", "unlimited")
+    executor = _install_real_substrate(monkeypatch)
+    ctx = harness.make_ctx()
+    try:
+        _call(ctx)
+        fp = _state(harness)["waves"][-1]["request_fingerprint"]
+        _call(ctx, spec={**DECK_SPEC, "in_scope": ["six slides"]})
+        executor.release.set()
+        assert _wait_until(lambda: len(tr.plan_review_wave(_state(harness), fp).get("historical_supplements") or []) == 3)
+        for index in range(7):
+            spec = {**DECK_SPEC, "in_scope": [f"another plan {index}"]}
+            _call(ctx, spec=spec)
+            assert _wait_until(lambda: not review_custody._ACTIVE)
+            _call(ctx, spec=spec)
+        old = copy.deepcopy(tr.plan_review_wave(_state(harness), fp))
+        last_selected_fp = _state(harness)["current_attempt"]["fingerprint"]
+        assert old["compact"] and "spec" not in old
+        assert executor.execute_calls == 27 and _state(harness)["cycles_paid"] == 9
+        artifact_root = task_artifact_dir_path(harness.drive, ctx.task_id, create=False)
+        assert (artifact_root / old["wave_artifact"]["path"]).is_file()
+        assert (artifact_root / old["spec_source_ref"]["path"]).is_file()
+        if missing in {"wave", "spec"}:
+            ref = old["wave_artifact" if missing == "wave" else "spec_source_ref"]
+            (artifact_root / ref["path"]).unlink()
+        elif missing == "legacy_no_sources":
+            def remove_sources(value):
+                target = next(row for row in value["waves"] if row["request_fingerprint"] == fp)
+                target.pop("wave_artifact", None)
+                target.pop("spec_source_ref", None)
+                return value
+            tr._update_plan_review_state(harness.drive, ctx.task_id, remove_sources)
+            old = copy.deepcopy(tr.plan_review_wave(_state(harness), fp))
+        for _ in range(2):
+            result = _call(ctx)
+            assert executor.execute_calls == 27
+            actual = _state(harness)
+            assert actual["cycles_paid"] == 9
+            assert tr.plan_review_wave(actual, fp) == old
+            assert actual["current_attempt"]["fingerprint"] == (last_selected_fp if missing else fp)
+            if missing:
+                assert "PLAN_REVIEW_SOURCE_UNAVAILABLE" in result and "completed historical responses" not in result
+            else:
+                assert "free read of the completed historical responses" in result
+                assert _control(result) == {"outcome": old["aggregate"], "closed": old["closed"]}
+                for row in old["historical_supplements"]:
+                    assert row["source_ref"]["path"] in result
+    finally:
+        executor.release.set()