feat(maintenance): raw-authority census reset + orphaned index-seed prune

polylogue/maintenance/raw_authority_reset.py

@@ -0,0 +1,109 @@
+"""Reset the raw-authority census planning ledger so the daemon rebuilds it.
+
+The raw-authority census tables (``raw_authority_censuses`` and its
+``plans``/``blockers``/``census_plans``/``census_post_plans`` children) are
+DERIVED convergence bookkeeping: each census chains to its predecessor
+(``sequence_no + 1``) and carries unresolved plans forward. The ACCEPTED
+materialization state — ``raw_sessions.revision_authority`` and the index's
+``raw_revision_heads`` / ``raw_revision_applications`` — lives OUTSIDE these
+tables and is untouched here.
+
+When the ledger accumulates inconsistent carried-forward state (e.g. a stale-plan
+blocker marks sibling plans ``CARRIED_FORWARD``, and later raw deletions drop
+them out of the recomputed frontier so the finalize postflight
+``persistent ⊄ post_ids`` throws — live incident 2026-07-22 after hook
+de-inflation), no new census can finalize to become a clean baseline, and the
+daemon defers every pass. Emptying the ledger removes the poisoned predecessor:
+the next daemon pass builds census #1 fresh over the current raw set
+(``predecessor = None``), with no carried-forward history and no stale blocker.
+
+``raw_authority_parser_census`` is intentionally KEPT — it holds resource-blocked
+parser fingerprints keyed to raws (FK-cascaded from ``raw_sessions``), not
+census-cycle bookkeeping, and the whale pass consumes it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from polylogue.config import Config
+from polylogue.maintenance.offline_guard import offline_maintenance_block_reason
+from polylogue.paths import render_root
+from polylogue.storage.raw_authority import prune_orphaned_index_revision_seeds as _prune_orphaned_index_revision_seeds
+from polylogue.storage.raw_authority import reset_raw_authority_census_ledger
+
+
+@dataclass(frozen=True, slots=True)
+class RawAuthorityResetReport:
+    """Row counts removed from the census ledger (dry-run or applied)."""
+
+    censuses: int
+    plans: int
+    blockers: int
+    census_plans: int
+    census_post_plans: int
+    applied: bool
+
+
+def _offline_config(archive_root: Path) -> Config:
+    return Config(archive_root=archive_root, render_root=render_root(), sources=[])
+
+
+def reset_raw_authority_census(
+    archive_root: Path,
+    *,
+    backup_manifest: Path | None = None,
+    dry_run: bool = True,
+) -> RawAuthorityResetReport:
+    """Empty the census planning ledger. ``dry_run`` reports counts only."""
+    if not dry_run and (
+        reason := offline_maintenance_block_reason(_offline_config(archive_root), active=True, dry_run=False)
+    ):
+        raise RuntimeError(reason)
+    before = reset_raw_authority_census_ledger(
+        archive_root,
+        backup_manifest=backup_manifest,
+        dry_run=dry_run,
+    )
+
+    return RawAuthorityResetReport(
+        censuses=before.censuses,
+        plans=before.plans,
+        blockers=before.blockers,
+        census_plans=before.census_plans,
+        census_post_plans=before.census_post_plans,
+        applied=not dry_run,
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class IndexSeedPruneReport:
+    """Index revision-authority read-model rows removed (dry-run or applied)."""
+
+    revision_heads: int
+    revision_applications: int
+    applied: bool
+
+
+def prune_orphaned_index_revision_seeds(archive_root: Path, *, dry_run: bool = True) -> IndexSeedPruneReport:
+    """Delete index raw-frontier seeds whose raw is gone from the source tier.
+
+    ``raw_revision_heads`` / ``raw_revision_applications`` are the index's
+    (rebuildable) revision-authority read model. After a source raw is deleted
+    (hook de-inflation), the seeds referencing it become broken predecessor
+    chains — the daemon's raw-frontier integrity check reports them as violated
+    and cannot converge past them. Deleting the seeds whose ``accepted_raw_id`` /
+    ``raw_id`` no longer exists in ``source.raw_sessions`` restores a clean
+    frontier; seeds for present raws are untouched.
+    """
+    if not dry_run and (
+        reason := offline_maintenance_block_reason(_offline_config(archive_root), active=True, dry_run=False)
+    ):
+        raise RuntimeError(reason)
+    counts = _prune_orphaned_index_revision_seeds(archive_root, dry_run=dry_run)
+    return IndexSeedPruneReport(
+        revision_heads=counts.revision_heads,
+        revision_applications=counts.revision_applications,
+        applied=not dry_run,
+    )

polylogue/storage/raw_authority.py

@@ -20,13 +20,43 @@
 
 from polylogue.core.json import JSONDocument, json_document
 from polylogue.logging import get_logger
+from polylogue.storage.archive_identity import ArchiveLocation
+from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from polylogue.storage.sqlite.migration_runner import validate_migration_backup_manifest
 
 RAW_AUTHORITY_PARSER_FINGERPRINT = "revision-membership-v1"
 RAW_AUTHORITY_CENSUS_QUERY_PREFIX = "polylogue://raw-authority-census/"
 RAW_AUTHORITY_DETAIL_QUERY_PREFIX = "polylogue://raw-authority-detail/"
 RAW_AUTHORITY_DETAIL_CHUNK_CHARS = 16_384
 logger = get_logger(__name__)
 
+_RESET_LEDGER_TABLES_CHILD_FIRST = (
+    "raw_authority_blockers",
+    "raw_authority_census_plans",
+    "raw_authority_census_post_plans",
+    "raw_authority_plans",
+    "raw_authority_censuses",
+)
+
+
+@dataclass(frozen=True, slots=True)
+class RawAuthorityCensusResetCounts:
+    """Counts for a reset of derived raw-authority census bookkeeping."""
+
+    censuses: int
+    plans: int
+    blockers: int
+    census_plans: int
+    census_post_plans: int
+
+
+@dataclass(frozen=True, slots=True)
+class OrphanedIndexRevisionSeedCounts:
+    """Counts for rebuildable revision-seed rows absent from source authority."""
+
+    revision_heads: int
+    revision_applications: int
+
 
 class RawReplayPlanStatus(StrEnum):
     EXECUTED = "executed"
@@ -1847,12 +1877,89 @@ def reject_invalid_raw_replay_application(
     return outcome
 
 
+def reset_raw_authority_census_ledger(
+    archive_root: Path,
+    *,
+    backup_manifest: Path | None,
+    dry_run: bool,
+) -> RawAuthorityCensusResetCounts:
+    """Reset derived census bookkeeping after authenticating a source backup."""
+    source_db = archive_root / "source.db"
+    if not source_db.is_file():
+        raise FileNotFoundError(source_db)
+    with closing(sqlite3.connect(source_db)) as conn:
+        conn.execute("PRAGMA foreign_keys = ON")
+        counts = {
+            "raw_authority_censuses": int(conn.execute("SELECT COUNT(*) FROM raw_authority_censuses").fetchone()[0]),
+            "raw_authority_plans": int(conn.execute("SELECT COUNT(*) FROM raw_authority_plans").fetchone()[0]),
+            "raw_authority_blockers": int(conn.execute("SELECT COUNT(*) FROM raw_authority_blockers").fetchone()[0]),
+            "raw_authority_census_plans": int(
+                conn.execute("SELECT COUNT(*) FROM raw_authority_census_plans").fetchone()[0]
+            ),
+            "raw_authority_census_post_plans": int(
+                conn.execute("SELECT COUNT(*) FROM raw_authority_census_post_plans").fetchone()[0]
+            ),
+        }
+        if not dry_run:
+            if backup_manifest is None:
+                raise ValueError("raw-authority census reset requires a verified source backup manifest")
+            validate_migration_backup_manifest(backup_manifest, ArchiveTier.SOURCE, connection=conn)
+            for table in _RESET_LEDGER_TABLES_CHILD_FIRST:
+                conn.execute(f"DELETE FROM {table}")
+            conn.commit()
+    return RawAuthorityCensusResetCounts(
+        censuses=counts["raw_authority_censuses"],
+        plans=counts["raw_authority_plans"],
+        blockers=counts["raw_authority_blockers"],
+        census_plans=counts["raw_authority_census_plans"],
+        census_post_plans=counts["raw_authority_census_post_plans"],
+    )
+
+
+def prune_orphaned_index_revision_seeds(
+    archive_root: Path,
+    *,
+    dry_run: bool,
+) -> OrphanedIndexRevisionSeedCounts:
+    """Prune rebuildable revision seeds that no longer have source authority."""
+    source_db = archive_root / "source.db"
+    index_db = ArchiveLocation.resolve(archive_root).active_index_path
+    if not source_db.is_file() or not index_db.is_file():
+        raise FileNotFoundError(source_db if not source_db.is_file() else index_db)
+    with closing(sqlite3.connect(index_db)) as conn:
+        conn.execute("ATTACH DATABASE ? AS src", (str(source_db),))
+        heads = int(
+            conn.execute(
+                "SELECT COUNT(*) FROM raw_revision_heads WHERE accepted_raw_id NOT IN (SELECT raw_id FROM src.raw_sessions)"
+            ).fetchone()[0]
+        )
+        applications = int(
+            conn.execute(
+                "SELECT COUNT(*) FROM raw_revision_applications WHERE raw_id NOT IN (SELECT raw_id FROM src.raw_sessions)"
+            ).fetchone()[0]
+        )
+        if not dry_run:
+            conn.execute(
+                "DELETE FROM raw_revision_heads WHERE accepted_raw_id NOT IN (SELECT raw_id FROM src.raw_sessions)"
+            )
+            conn.execute(
+                "DELETE FROM raw_revision_applications WHERE raw_id NOT IN (SELECT raw_id FROM src.raw_sessions)"
+            )
+            conn.commit()
+    return OrphanedIndexRevisionSeedCounts(
+        revision_heads=heads,
+        revision_applications=applications,
+    )
+
+
 __all__ = [
     "RAW_AUTHORITY_CENSUS_QUERY_PREFIX",
     "RAW_AUTHORITY_DETAIL_CHUNK_CHARS",
     "RAW_AUTHORITY_DETAIL_QUERY_PREFIX",
     "RAW_AUTHORITY_PARSER_FINGERPRINT",
     "RawAuthorityCensusReceipt",
+    "RawAuthorityCensusResetCounts",
+    "OrphanedIndexRevisionSeedCounts",
     "RawReplayPlan",
     "RawReplayPlanOutcome",
     "RawReplayPlanStatus",
@@ -1872,6 +1979,8 @@ def reject_invalid_raw_replay_application(
     "read_raw_authority_detail",
     "record_raw_authority_census",
     "record_raw_replay_outcome",
+    "reset_raw_authority_census_ledger",
+    "prune_orphaned_index_revision_seeds",
     "reject_invalid_raw_replay_application",
     "reject_stale_raw_replay_plan",
     "resolve_raw_authority_blocker",