feat(storage): add messages_fts_identity ledger for rowid-reuse detection

docs/internals.md

@@ -188,6 +188,61 @@ Polylogue has two schema-evolution regimes, keyed by tier durability.
   hash, so concurrent publishers of identical bytes remain independent.
   Existing v3 tiers migrate additively through
   `004_blob_publication_reservations.sql` after a verified backup manifest.
+- Index schema version 43 adds `messages_fts_identity`, a rowid-keyed shadow
+  ledger binding each `messages_fts` rowid to the `block_id` it was populated
+  from (polylogue-1xc.12). `messages_fts` is a contentless FTS5 table
+  (`content=''`): its `UNINDEXED` columns (including `block_id`) are
+  write-only and never retrievable by a later `SELECT`. SQLite reuses freed
+  rowids (deleting the highest-rowid block then inserting a new one commonly
+  gets the same rowid back — exactly what a full-session-replace does), so a
+  bare rowid cannot prove which block a `messages_fts` row currently
+  represents; count-only reconciliation (`source_rows == indexed_rows`) is
+  blind to this because both sides still balance even when a stale rowid has
+  silently rebound to a different block. The ledger's `source_hash` column
+  reuses the existing `blocks.content_hash` evidence hash as the
+  source-identity component and `recipe_id` is the versioned
+  `FTS_MESSAGES_IDENTITY_RECIPE_ID` constant as the recipe-identity
+  component — the same subject/source/recipe separation
+  `storage/derivation_identity.py` formalizes for polylogue-wmsc's
+  `DerivationKey`, applied here as a lightweight per-row ledger (not a full
+  `DerivationKey` digest — too expensive per trigger-fired row) and never as
+  a shared cross-domain table: FTS keeps its own ledger and repair lifecycle.
+  The `messages_fts_ai`/`ad`/`au` trigger bodies gain a paired
+  insert/delete/upsert against `messages_fts_identity` in the SAME trigger
+  body as their `messages_fts` write, so the two tables can never observe a
+  different rowid/block_id binding for the same event. Exact reconciliation
+  (`fts_invariant_snapshot_sync`, `storage/fts/sql.py:message_identity_mismatch_sql`)
+  now also joins `blocks`/`messages_fts_docsize`/`messages_fts_identity` on
+  rowid AND `block_id`/`source_hash`/`recipe_id`, catching rowid-reuse,
+  changed-text, and changed-recipe drift that a count-only or rowid-only
+  check cannot see. `message_identity_mismatch_sql` deliberately counts only
+  a PRESENT-but-WRONG ledger entry, never a missing one: nothing reads block
+  identity from the ledger except this reconciliation query itself, and the
+  next trigger-fired mutation at that rowid creates a correct fresh row
+  regardless of whether one existed before, so an absent entry is a coverage
+  gap, not a conflict — counting it would make `ready` permanently false on
+  any archive that writes through the ordinary session-replace path (see
+  below), defeating the point of a readiness signal. Bulk paths outside the
+  per-row triggers (full rebuild, batched missing/excess repair,
+  session-scoped repair in `storage/fts/fts_lifecycle.py`) pair their
+  `messages_fts` writes with the matching identity companion SQL, and the
+  batched missing-row repair also opportunistically UPSERTs a correct entry
+  for any indexed rowid whose ledger entry is missing or wrong. The one
+  exception is `storage/sqlite/archive_tiers/write.py`'s non-bulk
+  full-session-replace fast path (`delete_session_rows_sql`/
+  `insert_session_rows_sql` called directly, bypassing the suspended block
+  triggers): it does not yet call the identity companions inline, so a
+  session written through that path is coverage-incomplete (not
+  conflicting) until the next repair backfills it — a STOP-and-report gap
+  (polylogue-1xc.12): closing it fully needs one additional paired
+  `delete_session_identity_rows_sql`/`insert_session_identity_rows_sql`
+  call at each of write.py's four `delete_session_rows_sql`/
+  `insert_session_rows_sql` call sites. Existing index tiers must
+  be rebuilt from source evidence (`polylogue ops reset --index && polylogued
+  run`) to populate the new ledger for already-indexed rows; a declared
+  clone-safe fast-forward exists (`IndexDeltaDeclaration` v43 in
+  `polylogue/storage/sqlite/lifecycle.py`) since every ledgered field is
+  re-derivable from already-persisted `blocks` columns with no raw reparse.
 - Index schema version 42 stops materializing `session_events` rows for four
   event types that are fully redundant with a sibling typed table
   (`token_count`, `message_usage`, `agent_policy`, `agent_message`;

docs/plans/layering.yaml

@@ -107,7 +107,8 @@ writer_modules:
           durability: disposable
           interruption: restartable
       entrypoints:
-        [add_convergence_debt, record_cursor_lag_sample, record_daemon_stage_event, record_ingest_attempt, record_query_run,
+        [add_convergence_debt, record_cursor_lag_sample, record_daemon_stage_event, record_fts_drift_sample,
+        record_ingest_attempt, record_query_run,
         record_daemon_lifecycle_heartbeat, record_daemon_lifecycle_signal, record_daemon_lifecycle_start,
         record_daemon_lifecycle_stop, record_mcp_call, record_route_observation, upsert_embedding_catchup_run,
         upsert_ingest_cursor, upsert_otlp_span]

polylogue/daemon/fts_startup.py

@@ -79,6 +79,10 @@ def record_fts_freshness_snapshot_sync(conn: sqlite3.Connection) -> None:
         return
     record_fts_invariant_snapshot_sync(conn, snapshot)
 
+    from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync
+
+    sample_fts_drift_to_ops_sync(conn)
+
 
 def active_fts_triggers_sync(conn: sqlite3.Connection) -> tuple[str, ...]:
     """Return the FTS triggers that should exist given the schema present."""
@@ -277,7 +281,7 @@ def _record_optional_fts_surface_debt(db_path: Path | None, error: str) -> None:
 
 def _message_fts_freshness_row_sync(
     conn: sqlite3.Connection,
-) -> tuple[object, object, object, object, object, object] | None:
+) -> tuple[object, object, object, object, object, object, object] | None:
     from polylogue.storage.fts.freshness import (
         MESSAGE_SURFACE,
         ensure_fts_freshness_table_sync,
@@ -287,15 +291,16 @@ def _message_fts_freshness_row_sync(
         ensure_fts_freshness_table_sync(conn)
         row = conn.execute(
             """
-            SELECT state, source_rows, indexed_rows, missing_rows, excess_rows, duplicate_rows
+            SELECT state, source_rows, indexed_rows, missing_rows, excess_rows, duplicate_rows,
+                   identity_mismatch_rows
             FROM fts_freshness_state
             WHERE surface = ?
             """,
             (MESSAGE_SURFACE,),
         ).fetchone()
         if row is None:
             return None
-        return (row[0], row[1], row[2], row[3], row[4], row[5])
+        return (row[0], row[1], row[2], row[3], row[4], row[5], row[6])
     except sqlite3.Error:
         return None
 
@@ -312,7 +317,7 @@ def _message_fts_docsize_has_rows_sync(conn: sqlite3.Connection) -> bool:
 
 
 def _message_fts_freshness_row_ready_sync(
-    conn: sqlite3.Connection, row: tuple[object, object, object, object, object, object] | None
+    conn: sqlite3.Connection, row: tuple[object, object, object, object, object, object, object] | None
 ) -> bool:
     if row is None:
         return False
@@ -324,6 +329,7 @@ def _message_fts_freshness_row_ready_sync(
     missing_rows = _int_or_zero(row[3])
     excess_rows = _int_or_zero(row[4])
     duplicate_rows = _int_or_zero(row[5])
+    identity_mismatch_rows = _int_or_zero(row[6])
     source_has_rows: bool | None = None
     if source_rows == 0 and indexed_rows == 0:
         source_has_rows = _blocks_search_text_has_rows_sync(conn)
@@ -334,12 +340,13 @@ def _message_fts_freshness_row_ready_sync(
         missing_rows=missing_rows,
         excess_rows=excess_rows,
         duplicate_rows=duplicate_rows,
+        identity_mismatch_rows=identity_mismatch_rows,
         source_has_rows=source_has_rows,
     )
 
 
 def _message_fts_freshness_row_stale_sync(
-    row: tuple[object, object, object, object, object, object] | None,
+    row: tuple[object, object, object, object, object, object, object] | None,
 ) -> bool:
     if row is None:
         return False

polylogue/daemon/fts_status.py

@@ -10,6 +10,7 @@
 from polylogue.logging import get_logger
 from polylogue.storage.fts.freshness import STALE, UNKNOWN, freshness_ready_record_trusted
 from polylogue.storage.fts.fts_lifecycle import FtsInvariantSnapshot, FtsSurfaceInvariant, fts_invariant_snapshot_sync
+from polylogue.storage.fts.sql import message_identity_mismatch_sql
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
 
 logger = get_logger(__name__)
@@ -89,6 +90,7 @@ def _freshness_rows(conn: sqlite3.Connection) -> dict[str, dict[str, int | str |
         "missing_rows",
         "excess_rows",
         "duplicate_rows",
+        "identity_mismatch_rows",
     )
     selected = ["surface", "state"]
     selected.extend(name for name in numeric_columns if name in columns)
@@ -105,6 +107,7 @@ def _freshness_rows(conn: sqlite3.Connection) -> dict[str, dict[str, int | str |
             "missing_rows": _int_or_zero(record.get("missing_rows")),
             "excess_rows": _int_or_zero(record.get("excess_rows")),
             "duplicate_rows": _int_or_zero(record.get("duplicate_rows")),
+            "identity_mismatch_rows": _int_or_zero(record.get("identity_mismatch_rows")),
             "detail": None if "detail" not in record or record["detail"] is None else str(record["detail"]),
         }
     return records
@@ -129,6 +132,7 @@ def _surface_payload(surface: FtsSurfaceInvariant) -> dict[str, int | bool | str
         "missing_rows": surface.missing_rows,
         "excess_rows": surface.excess_rows,
         "duplicate_rows": surface.duplicate_rows,
+        "identity_mismatch_rows": surface.identity_mismatch_rows,
         "ready": surface.ready,
         "exact": True,
     }
@@ -178,6 +182,7 @@ def _archive_exact_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | b
             "missing_rows": 0,
             "excess_rows": 0,
             "duplicate_rows": 0,
+            "identity_mismatch_rows": 0,
             "ready": ready,
             "exact": True,
         }
@@ -198,6 +203,7 @@ def _archive_exact_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | b
             "missing_rows": source_rows,
             "excess_rows": 0,
             "duplicate_rows": 0,
+            "identity_mismatch_rows": 0,
             "ready": False,
             "exact": True,
         }
@@ -227,11 +233,17 @@ def _archive_exact_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | b
         or 0
     )
     duplicate_rows = 0
+    identity_mismatch_rows = (
+        int(conn.execute(message_identity_mismatch_sql()).fetchone()[0] or 0)
+        if _table_exists(conn, "messages_fts_identity")
+        else 0
+    )
     ready = (
         triggers_present
         and missing_rows == 0
         and excess_rows == 0
         and duplicate_rows == 0
+        and identity_mismatch_rows == 0
         and source_rows == indexed_rows
     )
     return {
@@ -243,6 +255,7 @@ def _archive_exact_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | b
         "missing_rows": missing_rows,
         "excess_rows": excess_rows,
         "duplicate_rows": duplicate_rows,
+        "identity_mismatch_rows": identity_mismatch_rows,
         "ready": ready,
         "exact": True,
     }
@@ -259,6 +272,7 @@ def _archive_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | bool |
     missing_rows = 0 if freshness is None else _int_or_zero(freshness.get("missing_rows"))
     excess_rows = 0 if freshness is None else _int_or_zero(freshness.get("excess_rows"))
     duplicate_rows = 0 if freshness is None else _int_or_zero(freshness.get("duplicate_rows"))
+    identity_mismatch_rows = 0 if freshness is None else _int_or_zero(freshness.get("identity_mismatch_rows"))
     recorded_state = None if freshness is None else str(freshness.get("state"))
     source_has_rows = (
         _source_has_rows(conn, "blocks")
@@ -275,6 +289,7 @@ def _archive_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | bool |
             missing_rows=missing_rows,
             excess_rows=excess_rows,
             duplicate_rows=duplicate_rows,
+            identity_mismatch_rows=identity_mismatch_rows,
             source_has_rows=source_has_rows,
         )
     )
@@ -292,6 +307,7 @@ def _archive_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | bool |
         "missing_rows": missing_rows,
         "excess_rows": excess_rows,
         "duplicate_rows": duplicate_rows,
+        "identity_mismatch_rows": identity_mismatch_rows,
         "ready": ready,
         "exact": False,
         "freshness_known": freshness_records is not None,
@@ -428,6 +444,9 @@ def fts_readiness_info(dbf: Path, *, exact: bool = False) -> dict[str, object]:
                 missing_rows = 0 if freshness is None else _int_or_zero(freshness.get("missing_rows"))
                 excess_rows = 0 if freshness is None else _int_or_zero(freshness.get("excess_rows"))
                 duplicate_rows = 0 if freshness is None else _int_or_zero(freshness.get("duplicate_rows"))
+                identity_mismatch_rows = (
+                    0 if freshness is None else _int_or_zero(freshness.get("identity_mismatch_rows"))
+                )
                 recorded_state = None if freshness is None else str(freshness.get("state"))
                 source_has_rows = (
                     _source_has_rows(conn, source_table)
@@ -444,6 +463,7 @@ def fts_readiness_info(dbf: Path, *, exact: bool = False) -> dict[str, object]:
                         missing_rows=missing_rows,
                         excess_rows=excess_rows,
                         duplicate_rows=duplicate_rows,
+                        identity_mismatch_rows=identity_mismatch_rows,
                         source_has_rows=source_has_rows,
                     )
                 )
@@ -462,6 +482,7 @@ def fts_readiness_info(dbf: Path, *, exact: bool = False) -> dict[str, object]:
                     "missing_rows": missing_rows,
                     "excess_rows": excess_rows,
                     "duplicate_rows": duplicate_rows,
+                    "identity_mismatch_rows": identity_mismatch_rows,
                     "ready": ready,
                     "exact": False,
                     "freshness_known": freshness_records is not None,