fix(repair): harden raw authority convergence

devtools/command_catalog.py

@@ -606,7 +606,7 @@ def to_dict(self) -> dict[str, object]:
         ),
         examples=(
             "devtools workspace raw-authority-scale-proof --json",
-            "devtools workspace raw-authority-scale-proof --components 10163 --raws 15264 --pass-limit 64 --keep --json",
+            "devtools workspace raw-authority-scale-proof --components 10163 --raws 15264 --expanded-raws 21398 --pass-limit 64 --keep --json",
         ),
     ),
     CommandSpec(

devtools/raw_authority_scale_proof.py

@@ -19,6 +19,7 @@
 import sqlite3
 import tempfile
 import time
+from collections import Counter
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import TextIO, cast
@@ -258,12 +259,16 @@ def _write_payload(path: Path, *, native_id: str, revision: int, target_size: in
             remaining -= amount
 
 
-def _independent_payload(*, native_id: str, target_size: int) -> bytes:
-    """Build one bounded standalone JSONL raw without a disk staging file."""
-    header = f'{{"type":"session_meta","payload":{{"id":"{native_id}","timestamp":"2026-07-15T00:00:00Z"}}}}\n'.encode()
-    if target_size < len(header):
-        raise ValueError("scenario payload allocation cannot preserve valid JSONL evidence")
-    return header + (b" " * (target_size - len(header)))
+def _explicit_component_cohorts(
+    *, components: int, direct_candidates: int, expanded_candidates: int
+) -> tuple[tuple[int, int, int], ...]:
+    """Build an exact aggregate topology without inventing an unexpanded corpus."""
+    direct = _component_counts(direct_candidates, components=components)
+    expanded = _component_counts(expanded_candidates, components=components)
+    return tuple(
+        (raw_count, direct_count, count)
+        for (raw_count, direct_count), count in sorted(Counter(zip(expanded, direct, strict=True)).items())
+    )
 
 
 def _component_counts(total: int, *, components: int) -> list[int]:
@@ -507,6 +512,12 @@ def _record_repair_pass(
     wall_ms = int((time.perf_counter() - started) * 1000)
     after = _process_sample()
     metrics = result.metrics
+    hard_failure_metrics = (
+        "raw_materialization_plan_conservation_error_count",
+        "raw_materialization_unresolved_blocker_count",
+    )
+    if not result.success and any(float(metrics.get(key, 0)) > 0 for key in hard_failure_metrics):
+        raise RuntimeError(f"raw-authority scale proof repair pass failed: {result.detail}")
     candidate_value = metrics.get("raw_materialization_candidate_count")
     executable_candidate_value = metrics.get("raw_materialization_executable_candidate_count", candidate_value)
     if (
@@ -566,6 +577,7 @@ def run_raw_authority_scale_proof(
     *,
     components: int = 16,
     raws: int = 24,
+    expanded_raws: int | None = None,
     scenario: RawAuthorityScaleScenario | None = None,
     pass_limit: int = 4,
     keep: bool = False,
@@ -582,11 +594,17 @@ def run_raw_authority_scale_proof(
     if scenario is None:
         if components < 1 or raws < components:
             raise ValueError("require components >= 1 and raws >= components")
+        expanded = raws if expanded_raws is None else expanded_raws
         scenario = RawAuthorityScaleScenario(
             components=components,
             direct_candidates=raws,
-            expanded_candidates=raws,
-            total_payload_bytes=raws * 1024,
+            expanded_candidates=expanded,
+            total_payload_bytes=expanded * 1024,
+            component_cohorts=(
+                _explicit_component_cohorts(components=components, direct_candidates=raws, expanded_candidates=expanded)
+                if expanded != raws
+                else None
+            ),
         )
     if components != 16 and components != scenario.components:
         raise ValueError("components and scenario.components disagree")
@@ -601,6 +619,7 @@ def run_raw_authority_scale_proof(
         max_memory_full_avg10=max_memory_full_avg10,
     )
     generation_samples = [admission_sample]
+    replay_samples: list[ProcessSample] = []
 
     def check_generation_pressure() -> None:
         sample = _process_sample()
@@ -611,6 +630,15 @@ def check_generation_pressure() -> None:
         )
         generation_samples.append(sample)
 
+    def check_replay_pressure() -> None:
+        sample = _process_sample()
+        _assert_admission(
+            sample,
+            max_io_full_avg10=max_io_full_avg10,
+            max_memory_full_avg10=max_memory_full_avg10,
+        )
+        replay_samples.append(sample)
+
     root = workdir.expanduser().resolve() / "raw-authority-scale-proof"
     if root.exists():
         shutil.rmtree(root)
@@ -647,21 +675,16 @@ def check_generation_pressure() -> None:
                         if not _uses_independent_component_members(scenario)
                         else f"{session_native_id}-member-{member:05d}"
                     )
-                    if _uses_independent_component_members(scenario):
-                        blob_hash, blob_size = publisher.write_from_bytes(
-                            _independent_payload(native_id=row_native_id, target_size=row_size)
-                        )
-                    else:
-                        payload_path = temporary_root / f"{source_label}.jsonl"
-                        _write_payload(
-                            payload_path,
-                            native_id=row_native_id,
-                            revision=member,
-                            target_size=row_size,
-                            previous=previous,
-                        )
-                        blob_hash, blob_size = publisher.write_from_path(payload_path)
-                        payload_path.unlink()
+                    payload_path = temporary_root / f"{source_label}.jsonl"
+                    _write_payload(
+                        payload_path,
+                        native_id=row_native_id,
+                        revision=member,
+                        target_size=row_size,
+                        previous=None if _uses_independent_component_members(scenario) else previous,
+                    )
+                    blob_hash, blob_size = publisher.write_from_path(payload_path)
+                    payload_path.unlink()
                     previous = publisher.blob_path(blob_hash)
                     source_path = component_source_path
                     pending_rows.append((row_native_id, source_path, blob_hash, blob_size, terminalized, component))
@@ -793,6 +816,7 @@ def check_generation_pressure() -> None:
         )
     pass_receipts: list[RawAuthorityScalePass] = []
     for number in range(1, (scenario.components * 3) + 4):
+        check_replay_pressure()
         pass_receipt, _digest = _record_repair_pass(
             number=number,
             mode="apply",
@@ -807,6 +831,7 @@ def check_generation_pressure() -> None:
         raise RuntimeError("raw-authority scale proof did not drain bounded apply passes")
     fixed_point_digests: list[str] = []
     for _ in range(2):
+        check_replay_pressure()
         pass_receipt, digest = _record_repair_pass(
             number=len(pass_receipts) + 1,
             mode="dry_run",
@@ -878,6 +903,7 @@ def check_generation_pressure() -> None:
         "achieved_shape": achieved_shape,
         "admission_sample": asdict(admission_sample),
         "generation_samples": [asdict(sample) for sample in generation_samples],
+        "replay_samples": [asdict(sample) for sample in replay_samples],
         "passes": [asdict(item) for item in pass_receipts],
         "fixed_point_digests": fixed_point_digests,
         "receipt": receipt.to_payload(),
@@ -893,6 +919,12 @@ def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int:
     parser.add_argument("--workdir", type=Path, default=Path(".cache") / "raw-authority-scale-proof")
     parser.add_argument("--components", type=int, default=16)
     parser.add_argument("--raws", type=int, default=24)
+    parser.add_argument(
+        "--expanded-raws",
+        type=int,
+        default=None,
+        help="Exact expanded authority candidates; defaults to --raws when omitted.",
+    )
     parser.add_argument(
         "--scenario-profile",
         type=Path,
@@ -938,6 +970,7 @@ def main(argv: list[str] | None = None, *, stdout: TextIO | None = None) -> int:
         args.workdir,
         components=args.components,
         raws=args.raws,
+        expanded_raws=args.expanded_raws,
         scenario=scenario,
         pass_limit=args.pass_limit,
         keep=args.keep,

polylogue/storage/archive_readiness.py

@@ -107,6 +107,7 @@ def raw_materialization_ready(readiness: Mapping[str, Any] | object | None) -> b
         "unchecked",
         "affected_unchecked",
         "raw_authority_frontier_blocking_count",
+        "raw_authority_blocker_count",
         "raw_authority_pending_census_count",
     )
     return all(_read_int(readiness, key) == 0 for key in blocking_keys)

polylogue/storage/raw_authority.py

@@ -778,6 +778,32 @@ def record_raw_authority_census(
     scope_json = _canonical_json(scope)
     residual_json = _canonical_json(residual)
     with closing(sqlite3.connect(archive_root / "source.db")) as conn, conn:
+        # A frontier preview authorizes an immutable execution exactly once.
+        # This runs in the census INSERT transaction, so two offline callers
+        # cannot both turn one unchanged preview into execution receipts.
+        if mode == "apply" and scope.get("schema") == "polylogue.raw-authority-frontier-scope.v1":
+            preview_census_id = scope.get("preview_census_id")
+            if isinstance(preview_census_id, str) and preview_census_id and selected_plan_ids:
+                placeholders = ",".join("?" for _ in selected_plan_ids)
+                duplicate = conn.execute(
+                    f"""
+                    SELECT existing.census_id
+                    FROM raw_authority_censuses AS existing
+                    JOIN raw_authority_census_plans AS cp ON cp.census_id = existing.census_id
+                    WHERE existing.mode = 'apply'
+                      AND json_extract(existing.scope_json, '$.schema') =
+                          'polylogue.raw-authority-frontier-scope.v1'
+                      AND json_extract(existing.scope_json, '$.preview_census_id') = ?
+                      AND cp.selected = 1
+                      AND cp.plan_id IN ({placeholders})
+                    LIMIT 1
+                    """,
+                    (preview_census_id, *sorted(selected_plan_ids)),
+                ).fetchone()
+                if duplicate is not None:
+                    raise RuntimeError(
+                        f"raw authority frontier preview plan is already claimed by census {duplicate[0]}"
+                    )
         previous = conn.execute(
             """
             SELECT census_id, sequence_no, inventory_digest, residual_digest,
@@ -1367,8 +1393,8 @@ def recover_interrupted_raw_authority_censuses(
             JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id
             WHERE c.lifecycle_status = 'planned'
               AND cp.selected = 1 AND cp.outcome_recorded = 0
-              AND COALESCE(json_extract(p.authority_witness_json, '$.schema'), '') !=
-                  'polylogue.raw-authority-frontier-plan.v1'
+              AND COALESCE(json_extract(c.scope_json, '$.schema'), '') !=
+                  'polylogue.raw-authority-frontier-scope.v1'
             ORDER BY c.sequence_no, cp.ordinal
             """
         ).fetchall()
@@ -1379,15 +1405,8 @@ def recover_interrupted_raw_authority_censuses(
                 SELECT census_id, scope_json
                 FROM raw_authority_censuses
                 WHERE lifecycle_status = 'planned'
-                  AND EXISTS (
-                      SELECT 1
-                      FROM raw_authority_census_plans AS cp
-                      JOIN raw_authority_plans AS p ON p.plan_id = cp.plan_id
-                      WHERE cp.census_id = raw_authority_censuses.census_id
-                        AND cp.selected = 1 AND cp.outcome_recorded = 0
-                        AND COALESCE(json_extract(p.authority_witness_json, '$.schema'), '') !=
-                            'polylogue.raw-authority-frontier-plan.v1'
-                  )
+                  AND COALESCE(json_extract(scope_json, '$.schema'), '') !=
+                      'polylogue.raw-authority-frontier-scope.v1'
                 ORDER BY sequence_no
                 """
             )

polylogue/storage/raw_reconciler.py

@@ -1381,22 +1381,13 @@ def recover_interrupted_raw_authority_frontier(config: Config) -> tuple[str, ...
                 receipt,
             )
             record_raw_replay_outcome(root, census_id, outcome)
-        elif related and all(
-            item.state in {RawAuthorityFrontierState.PROVEN_CURRENT, RawAuthorityFrontierState.SUPERSEDED}
-            for item in related
-        ):
-            outcome = RawReplayPlanOutcome(
-                plan.plan_id,
-                plan.input_raw_ids,
-                RawReplayPlanStatus.EXECUTED,
-                "interrupted application recovered from typed terminal frontier states",
-                "none",
-                receipt,
-            )
-            record_raw_replay_outcome(root, census_id, outcome)
         else:
             from polylogue.storage.raw_authority import reject_stale_raw_replay_plan
 
+            # Terminal-looking frontier items do not prove this exact immutable
+            # strategy ran: ordinary ingest may have terminalized only part of
+            # a multi-input plan.  Without its strategy receipt, never mint an
+            # EXECUTED outcome during crash recovery.
             reject_stale_raw_replay_plan(root, census_id, plan, receipt)
         recovered.append(plan.plan_id)
     post_state_counts = _state_counts(current_items)