fix(storage): preserve complete raw frontier counts

polylogue/storage/archive_readiness.py

@@ -309,17 +309,26 @@ def raw_materialization_readiness_snapshot(active_archive: Path) -> dict[str, ob
                     # terminal state rather than keep an already repaired plan
                     # blocking until some later inspection happens to run.
                     frontier_post_residual = json.loads(str(frontier_row["post_residual_json"] or "{}"))
-                    postflight_state_counts = frontier_post_residual.get("state_counts")
+                    postflight_state_counts = frontier_post_residual.get("frontier_state_counts")
+                    postflight_residual_state_counts = frontier_post_residual.get("state_counts")
                     scope_state_counts = frontier_scope.get("state_counts")
-                    state_counts_source = (
+                    frontier_state_counts_source = (
                         postflight_state_counts if isinstance(postflight_state_counts, Mapping) else scope_state_counts
                     )
                     frontier_state_counts = {
-                        str(key): int(value) for key, value in dict(state_counts_source or {}).items()
+                        str(key): int(value) for key, value in dict(frontier_state_counts_source or {}).items()
+                    }
+                    blocking_state_counts = {
+                        str(key): int(value)
+                        for key, value in dict(
+                            postflight_residual_state_counts
+                            if isinstance(postflight_residual_state_counts, Mapping)
+                            else frontier_state_counts
+                        ).items()
                     }
                     nonblocking_states = {"proven_current", "superseded"}
                     authority_frontier_blocking_count = sum(
-                        count for state, count in frontier_state_counts.items() if state not in nonblocking_states
+                        count for state, count in blocking_state_counts.items() if state not in nonblocking_states
                     )
                     authority_frontier = {
                         "census_id": str(frontier_row["census_id"]),

polylogue/storage/raw_reconciler.py

@@ -930,14 +930,18 @@ def _state_counts(items: tuple[RawAuthorityFrontierItem, ...]) -> JSONDocument:
 
 
 def _residual(state_counts: JSONDocument) -> JSONDocument:
+    residual_state_counts = {
+        state: count for state, count in state_counts.items() if state != RawAuthorityFrontierState.PROVEN_CURRENT.value
+    }
     return json_document(
         {
             "schema": "polylogue.raw-authority-frontier-residual.v1",
-            "state_counts": {
-                state: count
-                for state, count in state_counts.items()
-                if state != RawAuthorityFrontierState.PROVEN_CURRENT.value
-            },
+            # Retain the residual-only map for plan/retry semantics, but bind
+            # the complete postflight frontier too.  A readiness surface must
+            # be able to report healthy proven-current heads rather than
+            # presenting the residual as a full state inventory.
+            "state_counts": residual_state_counts,
+            "frontier_state_counts": state_counts,
         }
     )
 

tests/unit/storage/test_archive_readiness.py

@@ -48,6 +48,29 @@ def test_readiness_uses_frontier_postflight_not_preapply_scope(tmp_path: Path) -
         source_preconditions={},
         index_preconditions={},
     )
+    preview = record_raw_authority_census(
+        tmp_path,
+        (plan,),
+        selected_plan_ids=set(),
+        executable_plan_ids={plan.plan_id},
+        mode="dry_run",
+        quiescent=True,
+        scope={
+            "schema": "polylogue.raw-authority-frontier-scope.v1",
+            "state_counts": {"missing_source_bytes": 1},
+        },
+        residual={
+            "schema": "polylogue.raw-authority-frontier-residual.v1",
+            "state_counts": {"missing_source_bytes": 1},
+            "frontier_state_counts": {"missing_source_bytes": 1, "proven_current": 2},
+        },
+    )
+    dry_run_snapshot = raw_materialization_readiness_snapshot(tmp_path)
+    dry_run_frontier = cast(Mapping[str, object], dry_run_snapshot["raw_authority_frontier"])
+    assert dry_run_frontier["census_id"] == preview.census_id
+    assert dry_run_frontier["state_counts"] == {"missing_source_bytes": 1, "proven_current": 2}
+    assert dry_run_frontier["blocking_count"] == 1
+
     receipt = record_raw_authority_census(
         tmp_path,
         (plan,),
@@ -62,6 +85,7 @@ def test_readiness_uses_frontier_postflight_not_preapply_scope(tmp_path: Path) -
         residual={
             "schema": "polylogue.raw-authority-frontier-residual.v1",
             "state_counts": {"missing_source_bytes": 1},
+            "frontier_state_counts": {"missing_source_bytes": 1, "proven_current": 2},
         },
     )
     record_raw_replay_outcome(
@@ -79,11 +103,17 @@ def test_readiness_uses_frontier_postflight_not_preapply_scope(tmp_path: Path) -
         tmp_path,
         receipt.census_id,
         post_plans=(),
-        post_residual={"schema": "polylogue.raw-authority-frontier-residual.v1", "state_counts": {}},
+        post_residual={
+            "schema": "polylogue.raw-authority-frontier-residual.v1",
+            "state_counts": {},
+            "frontier_state_counts": {"proven_current": 3},
+        },
     )
 
     snapshot = raw_materialization_readiness_snapshot(tmp_path)
+    frontier = cast(Mapping[str, object], snapshot["raw_authority_frontier"])
 
+    assert frontier["state_counts"] == {"proven_current": 3}
     assert snapshot["raw_authority_frontier_blocking_count"] == 0
     assert raw_materialization_ready(snapshot) is True