fix(maintenance): type unowned attachments and share the ANALYZE set

docs/maintenance.md

@@ -419,7 +419,7 @@ extensible registry):
 | `tier-schema` | Every tier file (source/index/embeddings/user/ops) exists at its current `PRAGMA user_version`. |
 | `pointer-coherence` | The conventional `index.db` path and the active `.index-active-pointer` generation agree (an interrupted blue-green promotion leaves these diverged — polylogue-k8kj class). |
 | `source-index-coverage` | Every raw logical head is materialized, has an explicit terminal disposition, or is quarantined, and every index session's `raw_id` still resolves to a real raw row (orphans). The raw source population, not the derived census ledger, defines the coverage universe. |
-| `source-conservation` | Every acquired source item (each `raw_sessions` row, hook event, history sidecar) is materialized or carries a typed exclusion citing its rule (revision superseded, byte-duplicate receipt, parse failure, validation rejection, declared non-session artifact kind, decode failure, census verdict, pending); a raw row whose source file no longer exists on disk is `source_missing` when its raw payload bytes are still retained and `source_lost` when they are not. Reverse: every session traces to a raw row that is not a declared non-session artifact (phantom sessions, polylogue-b508, are reported and never deleted), and every message, block, and attachment ref traces to its owner. Unexplained, unclassified, lost-source, orphan, and phantom terms block; pending is a warning. The acceptance instrument for a rebuilt archive: zero blocking terms. |
+| `source-conservation` | Every acquired source item (each `raw_sessions` row, hook event, history sidecar) is materialized or carries a typed exclusion citing its rule (revision superseded, byte-duplicate receipt, parse failure, validation rejection, declared non-session artifact kind, decode failure, census verdict, pending); a raw row whose source file no longer exists on disk is `source_missing` when its raw payload bytes are still retained and `source_lost` when they are not. Reverse: every session traces to a raw row that is not a declared non-session artifact (phantom sessions, polylogue-b508, are reported and never deleted), and every message, block, and attachment ref traces to its owner. An attachment with no ref splits on `ref_count`: `attachment_unowned` (ref_count 0) is the writer's typed retention of an owner-ambiguous attachment and is explained, while `attachment_unreferenced` (non-zero ref_count) lost its refs without the ref-count sweep and blocks. Unexplained, unclassified, lost-source, orphan, and phantom terms block; pending is a warning. The acceptance instrument for a rebuilt archive: zero blocking terms. |
 | `fts-parity` | `messages_fts`/`blocks_command_trigram` exactly cover their source `blocks` rows, archive-wide, with the worst-offending sessions surfaced by name. |
 | `lineage-sanity` | `session_links.resolved_dst_session_id` and `branch_point_message_id` resolve to real sessions/messages (the latter is deliberately not a foreign key — see the data-model docs). |
 | `planner-stats` | `sqlite_stat1` covers `blocks`/`messages`/`session_links`/`action_pairs` (warn-level: a fresh generation without `ANALYZE` picks pathological query plans, polylogue-l3tk class). |

polylogue/maintenance/archive_verification.py

@@ -71,16 +71,16 @@
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
+from polylogue.storage.sqlite.maintenance import PLANNER_STATS_COVERED_TABLES
 
 logger = get_logger(__name__)
 
 #: Default cap on per-check sample evidence (worst sessions, offending ids, ...).
 DEFAULT_SAMPLE_LIMIT = 10
 
 
-#: index-tier tables the planner-stats check expects ``ANALYZE`` coverage for
-#: (polylogue-l3tk: fresh generations without stats pick pathological plans).
-_PLANNER_STATS_COVERED_TABLES: tuple[str, ...] = ("blocks", "messages", "session_links", "action_pairs")
+#: index-tier tables the planner-stats check expects ``ANALYZE`` coverage for.
+_PLANNER_STATS_COVERED_TABLES = PLANNER_STATS_COVERED_TABLES
 
 
 @dataclass

polylogue/maintenance/rebuild_index.py

@@ -35,6 +35,7 @@
 from polylogue.storage.sqlite.action_pairs import rebuild_all_action_pairs_sync
 from polylogue.storage.sqlite.connection_profile import BULK_BUILD_WRITE_CONNECTION_PRAGMA_STATEMENTS
 from polylogue.storage.sqlite.delegation_facts import rebuild_all_delegation_facts_sync
+from polylogue.storage.sqlite.maintenance import PLANNER_STATS_COVERED_TABLES
 
 if TYPE_CHECKING:
     from polylogue.operations.candidate_build import CandidateBuildRequest
@@ -54,13 +55,10 @@
 # so analyzing their virtual-table backing stores does not improve any replay
 # plan and can dominate a large archive's checkpoint.  These row stores are
 # the tables used by the writer-hot replacement/link/action-pair queries.
-_PLANNER_STATS_ANALYZE_STATEMENTS = (
-    "ANALYZE sessions",
-    "ANALYZE messages",
-    "ANALYZE blocks",
-    "ANALYZE session_links",
-    "ANALYZE action_pairs",
-)
+# ``sessions`` is analyzed in addition to the verified set because replay's
+# session-scoped lookups depend on it; the verified set itself is shared so the
+# replay route cannot drift from what planner-stats requires.
+_PLANNER_STATS_ANALYZE_STATEMENTS = tuple(f"ANALYZE {table}" for table in ("sessions", *PLANNER_STATS_COVERED_TABLES))
 
 logger = get_logger(__name__)
 

polylogue/maintenance/source_conservation.py

@@ -69,6 +69,7 @@
 _TERM_BLOCK_ORPHAN = "block_orphan"
 _TERM_ATTACHMENT_REF_ORPHAN = "attachment_ref_orphan"
 _TERM_ATTACHMENT_UNREFERENCED = "attachment_unreferenced"
+_TERM_ATTACHMENT_UNOWNED = "attachment_unowned"
 
 _RULES: dict[str, str] = {
     _TERM_SOURCE_MISSING: ("acquired source file no longer exists on disk; the archive retains its raw payload bytes"),
@@ -98,7 +99,14 @@
     _TERM_MESSAGE_ORPHAN: "message names no session",
     _TERM_BLOCK_ORPHAN: "block names no message",
     _TERM_ATTACHMENT_REF_ORPHAN: "attachment ref names no message",
-    _TERM_ATTACHMENT_UNREFERENCED: "attachment has no ref and therefore no source lineage",
+    _TERM_ATTACHMENT_UNREFERENCED: (
+        "attachment has no ref yet still carries a non-zero ref_count; its refs went away "
+        "without the ref-count sweep, so the row is unreachable from every read path"
+    ),
+    _TERM_ATTACHMENT_UNOWNED: (
+        "attachment was written unreferenced because its owning message is ambiguous "
+        "(ref_count 0, never swept); identity and bytes are retained as evidence"
+    ),
 }
 
 _BLOCKING: frozenset[str] = frozenset(
@@ -510,6 +518,8 @@ def audit_source_conservation(
     attachment_ref_orphans: list[tuple[Any, ...]] = []
     attachment_unreferenced_count = 0
     attachment_unreferenced: list[tuple[Any, ...]] = []
+    attachment_unowned_count = 0
+    attachment_unowned: list[tuple[Any, ...]] = []
     if table_exists(conn, "attachment_refs", schema="idx_tier"):
         attachment_ref_orphan_count = int(
             conn.execute(
@@ -527,18 +537,43 @@ def audit_source_conservation(
             """,
             (sample_limit,),
         ).fetchall()
+        # A ref-less attachment splits on ``ref_count``. The writer inserts an
+        # owner-ambiguous row with ref_count 0 and keeps it out of the sweep,
+        # so ref_count 0 means "never had a ref" -- explained, non-blocking.
+        # Any ref-less row whose ref_count is non-zero was refreshed while refs
+        # existed and then lost them without the sweep running: it is
+        # unreachable from every read path and blocks.
+        unreferenced_predicate = """
+            NOT EXISTS (SELECT 1 FROM idx_tier.attachment_refs ar WHERE ar.attachment_id = a.attachment_id)
+        """
         attachment_unreferenced_count = int(
             conn.execute(
-                """
+                f"""
                 SELECT COUNT(*) FROM idx_tier.attachments a
-                WHERE NOT EXISTS (SELECT 1 FROM idx_tier.attachment_refs ar WHERE ar.attachment_id = a.attachment_id)
+                WHERE {unreferenced_predicate} AND a.ref_count != 0
                 """
             ).fetchone()[0]
         )
         attachment_unreferenced = conn.execute(
-            """
+            f"""
             SELECT a.attachment_id FROM idx_tier.attachments a
-            WHERE NOT EXISTS (SELECT 1 FROM idx_tier.attachment_refs ar WHERE ar.attachment_id = a.attachment_id)
+            WHERE {unreferenced_predicate} AND a.ref_count != 0
+            LIMIT ?
+            """,
+            (sample_limit,),
+        ).fetchall()
+        attachment_unowned_count = int(
+            conn.execute(
+                f"""
+                SELECT COUNT(*) FROM idx_tier.attachments a
+                WHERE {unreferenced_predicate} AND a.ref_count = 0
+                """
+            ).fetchone()[0]
+        )
+        attachment_unowned = conn.execute(
+            f"""
+            SELECT a.attachment_id FROM idx_tier.attachments a
+            WHERE {unreferenced_predicate} AND a.ref_count = 0
             LIMIT ?
             """,
             (sample_limit,),
@@ -603,6 +638,11 @@ def _term(
                 attachment_unreferenced_count,
                 _sample(attachment_unreferenced, sample_limit),
             ),
+            _term(
+                _TERM_ATTACHMENT_UNOWNED,
+                attachment_unowned_count,
+                _sample(attachment_unowned, sample_limit),
+            ),
         )
     )
     return SourceConservationReport(

polylogue/storage/sqlite/maintenance.py

@@ -10,6 +10,20 @@
 DEFAULT_OPTIMIZE_ANALYSIS_LIMIT = 1_000
 ARCHIVE_TIER_OPTIMIZE_FILES = ("source.db", "index.db", "embeddings.db", "user.db", "ops.db")
 
+#: Index-tier tables that must carry measured ``ANALYZE`` statistics
+#: (polylogue-l3tk: a generation without stats picks pathological plans for
+#: writer-hot session-scoped queries). Every route that produces a queryable
+#: index tier -- the rebuild replay, and the test archive builders -- ANALYZEs
+#: this set, and the ``planner-stats`` verification check reads the same
+#: constant, so a table added here cannot go uncovered on only one route.
+PLANNER_STATS_COVERED_TABLES: tuple[str, ...] = ("blocks", "messages", "session_links", "action_pairs")
+
+
+def analyze_planner_stats_tables(conn: sqlite3.Connection, *, tables: tuple[str, ...] | None = None) -> None:
+    """Run ``ANALYZE`` over each covered table on an open write connection."""
+    for table in tables if tables is not None else PLANNER_STATS_COVERED_TABLES:
+        conn.execute(f"ANALYZE {table}")
+
 
 @dataclass(frozen=True, slots=True)
 class SqliteOptimizeObservation:

tests/infra/convergence_harness.py

@@ -49,6 +49,7 @@
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.archive_tiers.write import write_parsed_session_to_archive
 from polylogue.storage.sqlite.connection_profile import open_connection
+from polylogue.storage.sqlite.maintenance import analyze_planner_stats_tables
 from tests.infra.pathology_composer import (
     ComposedPathology,
     compose_append_revision_chain,
@@ -642,8 +643,7 @@ def _acquired_at_ms(index: int) -> int:
 
 def _analyze_registry_tables(index_db: Path) -> None:
     with closing(open_connection(index_db)) as conn:
-        for table in ("blocks", "messages", "action_pairs"):
-            conn.execute(f"ANALYZE {table}")
+        analyze_planner_stats_tables(conn)
         conn.commit()
 
 

tests/infra/schema_inference.py

@@ -7,6 +7,7 @@
 
 from polylogue.storage.blob_store import BlobStore
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
+from polylogue.storage.sqlite.maintenance import analyze_planner_stats_tables
 
 
 def seed_schema_inference_archive(root: Path) -> Path:
@@ -60,7 +61,5 @@ def seed_schema_inference_archive(root: Path) -> Path:
             VALUES ('codex-session:session:0.0', 'codex-session:session', 0, 'text', 'hello')
             """
         )
-        conn.execute("ANALYZE blocks")
-        conn.execute("ANALYZE messages")
-        conn.execute("ANALYZE action_pairs")
+        analyze_planner_stats_tables(conn)
     return ground_truth

tests/unit/maintenance/test_archive_verification.py

@@ -35,6 +35,7 @@
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS, initialize_active_archive_root
 from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceArtifact, upsert_raw_artifact
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
+from polylogue.storage.sqlite.maintenance import analyze_planner_stats_tables
 from tests.infra.pathology_zoo import (
     CLAUDE_VINTAGE_LIVE_PROOF_LOGICAL_SOURCE_KEY,
     CLAUDE_VINTAGE_LIVE_PROOF_ORIGIN,
@@ -1238,6 +1239,71 @@ def test_empty_covered_table_without_stats_is_not_missing_coverage(tmp_path: Pat
     assert check.evidence["missing_tables"] == []
 
 
+def _populate_session_links(index_db: Path) -> None:
+    conn = _connect(index_db)
+    try:
+        session_id = str(conn.execute("SELECT session_id FROM sessions LIMIT 1").fetchone()[0])
+        conn.execute(
+            """
+            INSERT INTO session_links(
+                src_session_id, dst_origin, dst_native_id, link_type,
+                confidence, evidence_json, observed_at_ms
+            ) VALUES (?, 'codex-session', 'parent', 'resume', 1.0, '{}', 100)
+            """,
+            (session_id,),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def test_convergence_builder_analyze_covers_every_verified_table(tmp_path: Path) -> None:
+    """The archive builders ANALYZE the set planner-stats verifies.
+
+    Drives the builders' own ANALYZE seam rather than a restatement of it, so
+    a builder that reverts to a hand-copied table list fails here.
+
+    Anti-vacuity: the control below ANALYZEs the drifted set and still warns,
+    so this passing case cannot be green because the check stopped noticing an
+    uncovered populated table.
+    """
+    from tests.infra.convergence_harness import _analyze_registry_tables
+
+    _seed_coherent_archive(tmp_path)
+    _populate_session_links(tmp_path / "index.db")
+    _analyze_registry_tables(tmp_path / "index.db")
+
+    check = _check(verify_archive(tmp_path, checks=("planner-stats",)), "planner-stats")
+
+    assert check.status is OutcomeStatus.OK, check.summary
+    assert check.evidence["missing_tables"] == []
+
+
+def test_populated_session_links_left_unanalyzed_still_warns(tmp_path: Path) -> None:
+    """Control for the test above: a populated covered table with no stats warns.
+
+    This is the exact shape the archive builders produced while they ANALYZEd
+    a hand-copied list that had drifted from the verified set.
+
+    Anti-vacuity: vacuous if planner-stats stopped covering session_links, in
+    which case no ANALYZE set could make it warn.
+    """
+    _seed_coherent_archive(tmp_path)
+    _populate_session_links(tmp_path / "index.db")
+    conn = _connect(tmp_path / "index.db")
+    try:
+        conn.execute("DELETE FROM sqlite_stat1 WHERE tbl = 'session_links'")
+        analyze_planner_stats_tables(conn, tables=("blocks", "messages", "action_pairs"))
+        conn.commit()
+    finally:
+        conn.close()
+
+    check = _check(verify_archive(tmp_path, checks=("planner-stats",)), "planner-stats")
+
+    assert check.status is OutcomeStatus.WARNING
+    assert check.evidence["missing_tables"] == ["session_links"]
+
+
 def test_missing_archive_root_reports_skips_not_crashes(tmp_path: Path) -> None:
     empty_root = tmp_path / "does-not-exist"
 

tests/unit/maintenance/test_source_conservation.py

@@ -447,3 +447,67 @@ def test_check_json_carries_every_term_with_its_rule(tmp_path: Path) -> None:
         assert isinstance(term, dict)
         assert isinstance(term["rule"], str) and term["rule"]
         assert isinstance(term["blocking"], bool)
+
+
+def _insert_attachment(conn: sqlite3.Connection, *, attachment_id: str, ref_count: int) -> None:
+    conn.execute(
+        """
+        INSERT INTO attachments(attachment_id, display_name, media_type, byte_count, acquisition_status, ref_count)
+        VALUES (?, 'a.png', 'image/png', 0, 'unfetched', ?)
+        """,
+        (attachment_id, ref_count),
+    )
+
+
+def test_owner_ambiguous_attachment_types_as_unowned_and_does_not_block(tmp_path: Path) -> None:
+    """An attachment written unreferenced because its owner is ambiguous is explained.
+
+    The writer inserts such a row with ``ref_count`` 0 and deliberately keeps
+    it out of the ref-count sweep, so it never had a ref to lose.
+
+    Anti-vacuity: without the ``ref_count`` split every ref-less attachment
+    types as ``attachment_unreferenced``, which blocks, and the check goes red.
+    """
+    _seed(tmp_path)
+    index_conn = sqlite3.connect(tmp_path / "index.db")
+    try:
+        _insert_attachment(index_conn, attachment_id="unowned-1", ref_count=0)
+        index_conn.commit()
+    finally:
+        index_conn.close()
+
+    check = _run(tmp_path)
+    assert check.status is OutcomeStatus.OK, check.summary
+    assert _count(check, "attachment_unowned") == 1
+    assert _terms(check)["attachment_unowned"]["blocking"] is False
+    assert _terms(check)["attachment_unowned"]["sample"] == ["unowned-1"]
+    assert _count(check, "attachment_unreferenced") == 0
+    assert check.evidence["blocking_count"] == 0
+    assert "attachment_unreferenced" not in check.summary
+
+
+def test_refless_attachment_with_stale_ref_count_still_blocks(tmp_path: Path) -> None:
+    """A row whose refs went away without the sweep is unreachable and blocks.
+
+    Its non-zero ``ref_count`` is the evidence that refs once existed: the
+    sweep would have recomputed it to 0 and deleted the row.
+
+    Anti-vacuity: if the split classified every ref-less attachment as the
+    explained ``attachment_unowned`` term, this archive would verify green and
+    the genuine orphan would go unreported.
+    """
+    _seed(tmp_path)
+    index_conn = sqlite3.connect(tmp_path / "index.db")
+    try:
+        _insert_attachment(index_conn, attachment_id="orphan-1", ref_count=2)
+        index_conn.commit()
+    finally:
+        index_conn.close()
+
+    check = _run(tmp_path)
+    assert check.status is OutcomeStatus.ERROR, check.summary
+    assert _count(check, "attachment_unreferenced") == 1
+    assert _terms(check)["attachment_unreferenced"]["blocking"] is True
+    assert _terms(check)["attachment_unreferenced"]["sample"] == ["orphan-1"]
+    assert _count(check, "attachment_unowned") == 0
+    assert check.evidence["blocking_count"] == 1