fix(maintenance): prune orphaned raw-authority plans in hook-deinflation repair

polylogue/maintenance/hook_deinflation.py

@@ -43,6 +43,27 @@
     "source_path LIKE '%/hooks/pending/%' AND source_path IN (SELECT source_path FROM raw_hook_events)"
 )
 
+# Raw-authority tables reference raws by JSON string (input_raw_ids_json), not by
+# FK, so deleting a raw does NOT cascade-clean its frontier plans/blockers/census
+# rows. A plan whose EVERY input raw is gone is unprocessable and must be pruned,
+# or the daemon reconciler throws ("duplicate strategy did not reach its typed
+# terminal postcondition") on the dangling plan (live incident 2026-07-22, first
+# convergence pass after the hook de-inflation). These are children of
+# raw_authority_plans with NO ACTION (RESTRICT) FKs, so delete children first.
+# Set-based (single pass, LEFT JOIN on the raw_id PK): expand each plan's input
+# raws, GROUP BY plan, keep plans whose inputs are ALL missing. The equivalent
+# correlated-subquery form ran ~26 billion ops (>1h) on the live archive; this
+# form completes in ~0.3s. A plan with empty inputs produces no json_each rows
+# and is correctly excluded (it is not orphaned-by-missing-raw).
+_PURELY_ORPHANED_PLANS_SQL = """
+    SELECT p.plan_id
+    FROM raw_authority_plans p, json_each(p.input_raw_ids_json) j
+    LEFT JOIN raw_sessions r ON r.raw_id = j.value
+    GROUP BY p.plan_id
+    HAVING SUM(CASE WHEN r.raw_id IS NULL THEN 1 ELSE 0 END) > 0
+       AND SUM(CASE WHEN r.raw_id IS NOT NULL THEN 1 ELSE 0 END) = 0
+"""
+
 
 @dataclass(frozen=True, slots=True)
 class HookDeinflationReport:
@@ -53,6 +74,7 @@ class HookDeinflationReport:
     raw_hook_events_before: int
     raw_hook_events_after: int
     hook_blob_refs_retained: int
+    orphaned_authority_plans: int
     applied: bool
 
 
@@ -65,6 +87,46 @@ def _load_ids(conn: sqlite3.Connection, table: str, ids: list[str]) -> None:
     conn.executemany(f"INSERT OR IGNORE INTO {table}(id) VALUES (?)", ((rid,) for rid in ids))
 
 
+def _count_orphaned_authority_plans(conn: sqlite3.Connection) -> int:
+    return int(conn.execute(f"SELECT COUNT(*) FROM ({_PURELY_ORPHANED_PLANS_SQL})").fetchone()[0])
+
+
+def _delete_orphaned_authority(conn: sqlite3.Connection) -> int:
+    """Prune raw-authority plans whose every input raw is gone, plus their
+    blocker/census children. Children first (NO ACTION FKs to plans).
+
+    ``_orphan_plans`` is created WITH a primary-key index on ``plan_id``: the
+    child tables (census_plans ~405k, census_post_plans ~324k rows) don't index
+    ``plan_id`` as a leading column, so each ``DELETE ... WHERE plan_id IN
+    (SELECT plan_id FROM _orphan_plans)`` full-scans the child once and needs an
+    O(log) membership probe per row. Without the PK the probe degrades to a full
+    scan of the 64k-row orphan set — ~26 billion ops, observed as a >1h hang on
+    the live archive."""
+    conn.execute("CREATE TEMP TABLE _orphan_plans (plan_id TEXT PRIMARY KEY)")
+    conn.execute(f"INSERT INTO _orphan_plans (plan_id) {_PURELY_ORPHANED_PLANS_SQL}")
+    count = int(conn.execute("SELECT COUNT(*) FROM _orphan_plans").fetchone()[0])
+    # Existing child indexes on plan_id are partial (WHERE resolved_at_ms IS NULL
+    # / WHERE selected = 1) or non-leading (PK is (census_id, plan_id)), so the FK
+    # RESTRICT check on the parent plan delete would full-scan each child per plan
+    # (~51 billion ops, the observed >1h hang). Full temporary plan_id indexes make
+    # both the IN-delete and the RESTRICT check index-driven; dropped before commit
+    # so the durable schema is unchanged.
+    child_indexes = {
+        "_tmp_hookdeflate_blk_plan": "raw_authority_blockers",
+        "_tmp_hookdeflate_cp_plan": "raw_authority_census_plans",
+        "_tmp_hookdeflate_cpp_plan": "raw_authority_census_post_plans",
+    }
+    for index_name, table in child_indexes.items():
+        conn.execute(f"CREATE INDEX {index_name} ON {table}(plan_id)")
+    for table in child_indexes.values():
+        conn.execute(f"DELETE FROM {table} WHERE plan_id IN (SELECT plan_id FROM _orphan_plans)")
+    conn.execute("DELETE FROM raw_authority_plans WHERE plan_id IN (SELECT plan_id FROM _orphan_plans)")
+    for index_name in child_indexes:
+        conn.execute(f"DROP INDEX {index_name}")
+    conn.execute("DROP TABLE _orphan_plans")
+    return count
+
+
 def repair_hook_session_inflation(archive_root: Path, *, dry_run: bool = True) -> HookDeinflationReport:
     """Delete hook-derived session rows from both tiers, keeping hook evidence.
 
@@ -109,6 +171,9 @@ def repair_hook_session_inflation(archive_root: Path, *, dry_run: bool = True) -
                 raw_hook_events_before=raw_hook_events_before,
                 raw_hook_events_after=raw_hook_events_before,
                 hook_blob_refs_retained=hook_blob_refs,
+                # Currently-orphaned plans (e.g. a prior incomplete repair left
+                # some); apply also prunes any this run newly orphans.
+                orphaned_authority_plans=_count_orphaned_authority_plans(source_conn),
                 applied=False,
             )
 
@@ -140,6 +205,11 @@ def repair_hook_session_inflation(archive_root: Path, *, dry_run: bool = True) -
         # write_source_hook_event path) and raw_hook_events. Delete only the
         # spurious session rows; FK cascade removes their parser-census rows.
         source_conn.execute(f"DELETE FROM raw_sessions WHERE {_HOOK_RAW_PREDICATE}")
+        # Prune raw-authority plans/blockers/census now dangling on the deleted
+        # raws (no FK to raw_sessions, so no cascade did this). Runs in the same
+        # transaction, after the raw delete, so "orphaned" is computed against
+        # the post-delete raw set.
+        orphaned_authority = _delete_orphaned_authority(source_conn)
         raw_hook_events_after = int(source_conn.execute("SELECT COUNT(*) FROM raw_hook_events").fetchone()[0])
         hook_blob_refs = int(
             source_conn.execute(
@@ -153,5 +223,6 @@ def repair_hook_session_inflation(archive_root: Path, *, dry_run: bool = True) -
         raw_hook_events_before=raw_hook_events_before,
         raw_hook_events_after=raw_hook_events_after,
         hook_blob_refs_retained=hook_blob_refs,
+        orphaned_authority_plans=orphaned_authority,
         applied=True,
     )

tests/unit/maintenance/test_hook_deinflation.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import sqlite3
 from pathlib import Path
 
@@ -105,6 +107,54 @@ def test_repair_removes_hook_sessions_keeps_evidence_and_real_sessions(tmp_path:
     assert rows == {"conv-real", "empty-real"}  # hook shells gone, real + empty-real kept
 
 
+def _seed_authority_plan(source_db: Path, *, plan_id: str, input_raw_id: str, with_blocker: bool) -> None:
+    """Seed a raw-authority frontier plan (and optional blocker) for a raw."""
+    with sqlite3.connect(source_db) as conn:
+        conn.execute("PRAGMA foreign_keys = OFF")  # seeding only; repair runs with FK on
+        conn.execute(
+            "INSERT OR IGNORE INTO raw_authority_censuses (census_id, sequence_no, scope_json, residual_json, "
+            "parser_fingerprint, mode, lifecycle_status, quiescent, inventory_digest, residual_digest, "
+            "plan_count, executable_plan_count, residual_plan_count, created_at_ms) "
+            "VALUES ('c1',1,'{}','{}','fp','census','finalized',1,'d','d',1,1,0,1)"
+        )
+        digest = hashlib.sha256(plan_id.encode()).hexdigest()  # 64 hex chars (CHECK)
+        conn.execute(
+            "INSERT INTO raw_authority_plans (plan_id, input_digest, input_raw_ids_json, logical_keys_json, "
+            "authority_witness_json, source_preconditions_json, index_preconditions_json, created_at_ms) "
+            "VALUES (?,?,?,?,?,?,?,1)",
+            (plan_id, digest, json.dumps([input_raw_id]), "[]", "{}", "{}", "{}"),
+        )
+        if with_blocker:
+            conn.execute(
+                "INSERT INTO raw_authority_blockers (blocker_id, plan_id, census_id, reason, expected_json, "
+                "observed_json, created_at_ms) VALUES (?,?, 'c1', 'r', '{}', '{}', 1)",
+                (f"blk:{plan_id}", plan_id),
+            )
+
+
+def test_repair_prunes_orphaned_raw_authority_plans(tmp_path: Path) -> None:
+    """Deleting hook raws leaves their frontier plans dangling (no FK); the repair
+    must prune them or the daemon reconciler throws on the dangling plan."""
+    initialize_active_archive_root(tmp_path)
+    with ArchiveStore.open_existing(tmp_path, read_only=False) as archive:
+        hook_raw = _seed_hook_raw(archive, event_id="evt-a", session_id="sess-1")
+        real_raw = _seed_real_raw(archive, native="conv-real")
+        archive.commit()
+
+    source_db = tmp_path / "source.db"
+    _seed_authority_plan(source_db, plan_id="plan-hook", input_raw_id=hook_raw, with_blocker=True)
+    _seed_authority_plan(source_db, plan_id="plan-real", input_raw_id=real_raw, with_blocker=False)
+
+    report = repair_hook_session_inflation(tmp_path, dry_run=False)
+    assert report.orphaned_authority_plans == 1  # only the hook plan is orphaned
+
+    with sqlite3.connect(source_db) as conn:
+        plans = {r[0] for r in conn.execute("SELECT plan_id FROM raw_authority_plans")}
+        blockers = conn.execute("SELECT COUNT(*) FROM raw_authority_blockers").fetchone()[0]
+    assert plans == {"plan-real"}  # hook plan pruned, real plan kept
+    assert blockers == 0  # the hook plan's blocker pruned with it
+
+
 def test_repair_refuses_to_race_a_running_daemon(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     """apply must refuse when the sole-writer daemon is live (Codex review)."""
     initialize_active_archive_root(tmp_path)