From 54416e896a915b6bf9df7b5e1ec7c40f7b8a1ce6 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:15:27 +0200 Subject: [PATCH 1/5] feat(storage): add messages_fts_identity ledger for rowid-reuse detection Problem: messages_fts is a contentless FTS5 table (content=''), so its block_id UNINDEXED column is write-only and never retrievable by 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 count-only reconciliation (source_rows == indexed_rows) cannot see a stale rowid that has silently rebound to a different block. What changed: adds messages_fts_identity(rowid, block_id UNIQUE, source_hash, recipe_id), maintained by the same messages_fts_ai/ad/au triggers plus every bulk rebuild/repair path in fts_lifecycle.py. Exact reconciliation (fts_invariant_snapshot_sync) now joins blocks/docsize/ identity on rowid AND block_id/source_hash/recipe_id (message_identity_mismatch_sql), catching rowid-reuse, changed-text, and changed-recipe drift a count-only check cannot see. Adds ops.db fts_drift_samples (bounded, 30d/5k-row retention) and a polylogue_fts_drift_rows Prometheus gauge (missing/excess/duplicate/ identity_mismatch by surface), both read O(1) from the existing fts_freshness_state ledger. INDEX_SCHEMA_VERSION 42->43 with a declared clone-safe fast-forward (v43 in storage/sqlite/lifecycle.py) since every ledgered field derives from already-persisted blocks columns. Ref polylogue-1xc.12 Co-Authored-By: Claude --- docs/internals.md | 45 ++ polylogue/daemon/fts_startup.py | 17 +- polylogue/daemon/fts_status.py | 21 + polylogue/daemon/metrics.py | 64 +++ polylogue/storage/fts/dangling_repair.py | 10 +- polylogue/storage/fts/drift_sampling.py | 116 +++++ polylogue/storage/fts/freshness.py | 54 +- polylogue/storage/fts/fts_lifecycle.py | 65 ++- polylogue/storage/fts/sql.py | 184 +++++++ .../storage/sqlite/archive_tiers/index.py | 20 +- polylogue/storage/sqlite/archive_tiers/ops.py | 23 + .../storage/sqlite/archive_tiers/ops_write.py | 137 ++++++ polylogue/storage/sqlite/lifecycle.py | 24 + .../test_fts_identity_state_machine.py | 291 +++++++++++ tests/unit/daemon/test_metrics_endpoint.py | 1 + .../unit/storage/test_fts_identity_ledger.py | 463 ++++++++++++++++++ 16 files changed, 1515 insertions(+), 20 deletions(-) create mode 100644 polylogue/storage/fts/drift_sampling.py create mode 100644 tests/property/test_fts_identity_state_machine.py create mode 100644 tests/unit/storage/test_fts_identity_ledger.py diff --git a/docs/internals.md b/docs/internals.md index 31d7b22760..e9ce793062 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -188,6 +188,51 @@ 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. 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. 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 re-ingested through that path + transiently shows as identity-incomplete until the next repair/reconciliation + pass backfills it (`dangling_repair.py`'s missing-row repair now also + upserts identity rows for any indexed rowid lacking one) — the same + eventually-consistent contract `missing_rows`/`excess_rows` already had + before this change, not a new weaker guarantee. 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`; diff --git a/polylogue/daemon/fts_startup.py b/polylogue/daemon/fts_startup.py index 9908c45c0b..ca28d9eba7 100644 --- a/polylogue/daemon/fts_startup.py +++ b/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,7 +291,8 @@ 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 = ? """, @@ -295,7 +300,7 @@ def _message_fts_freshness_row_sync( ).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 diff --git a/polylogue/daemon/fts_status.py b/polylogue/daemon/fts_status.py index 7b73314f93..5c4eb8c66c 100644 --- a/polylogue/daemon/fts_status.py +++ b/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, diff --git a/polylogue/daemon/metrics.py b/polylogue/daemon/metrics.py index 484765f015..0a7d9bb5e2 100644 --- a/polylogue/daemon/metrics.py +++ b/polylogue/daemon/metrics.py @@ -50,6 +50,11 @@ - ``polylogue_fts_trigger_present`` (gauge) — labels: trigger - ``polylogue_fts_triggers_all_present`` (gauge, 0/1) - ``polylogue_fts_freshness_ready`` (gauge, 0/1) — labels: surface +- ``polylogue_fts_drift_rows`` (gauge) — labels: surface, kind + (missing/excess/duplicate/identity_mismatch). Drift MAGNITUDE, not just the + boolean ready/stale of ``polylogue_fts_freshness_ready`` above; read O(1) + from the ``fts_freshness_state`` ledger, no scan on scrape + (polylogue-1xc.12). - ``polylogue_live_ingest_memory_mebibytes`` (gauge) — labels: kind - ``polylogue_stale_cursor_writes_total`` (counter) - ``polylogue_embedding_sessions`` (gauge) — labels: state @@ -447,6 +452,45 @@ def _fts_freshness_ready(conn: sqlite3.Connection) -> list[tuple[str, int]]: return samples +# polylogue-1xc.12: drift MAGNITUDE gauges, not just the boolean +# polylogue_fts_freshness_ready above. Reads straight from the +# fts_freshness_state ledger row (O(1), no COUNT(*) on scrape) written by +# the same exact-invariant/repair paths that already populate that table -- +# this function never recomputes anything itself. +_FTS_DRIFT_KINDS: tuple[str, ...] = ("missing", "excess", "duplicate", "identity_mismatch") + + +def _fts_drift_magnitude(conn: sqlite3.Connection) -> list[tuple[str, str, int]]: + if not _table_exists(conn, "fts_freshness_state"): + return [] + columns = _columns(conn, "fts_freshness_state") + kind_columns = { + "missing": "missing_rows", + "excess": "excess_rows", + "duplicate": "duplicate_rows", + "identity_mismatch": "identity_mismatch_rows", + } + available = {kind: column for kind, column in kind_columns.items() if column in columns} + if not available: + return [] + selected = ["surface", *available.values()] + try: + rows = conn.execute(f"SELECT {', '.join(selected)} FROM fts_freshness_state ORDER BY surface").fetchall() + except sqlite3.Error as exc: + logger.warning("metrics: fts drift magnitude query failed: %s", exc, exc_info=True) + return [] + samples: list[tuple[str, str, int]] = [] + for row in rows: + surface = str(row[0]) + for index, kind in enumerate(available, start=1): + try: + value = int(row[index] or 0) + except (TypeError, ValueError): + value = 0 + samples.append((surface, kind, value)) + return samples + + def _latest_ingest_memory(conn: sqlite3.Connection, *, ops_db: Path | None = None) -> list[tuple[str, float]]: if ops_db is not None: ops_memory = _ops_latest_ingest_memory(ops_db) @@ -1023,6 +1067,10 @@ def format_metrics( ), ("polylogue_fts_triggers_all_present", "All expected FTS sync triggers are installed."), ("polylogue_fts_freshness_ready", "1 when the daemon freshness ledger marks an FTS surface ready."), + ( + "polylogue_fts_drift_rows", + "FTS drift magnitude by surface and kind, read O(1) from fts_freshness_state.", + ), ("polylogue_live_ingest_memory_mebibytes", "Latest live ingest memory sample in MiB by kind."), ("polylogue_stale_cursor_writes_total", "Total stale-cursor writes observed across ingest attempts."), ("polylogue_embedding_sessions", "Embedding session counts by state."), @@ -1154,6 +1202,18 @@ def format_metrics( samples=[({"surface": surface}, ready) for surface, ready in freshness], ) + drift = _fts_drift_magnitude(conn) + _emit_metric( + lines, + name="polylogue_fts_drift_rows", + help_text=( + "FTS drift magnitude by surface and kind (missing/excess/duplicate/" + "identity_mismatch), read O(1) from the fts_freshness_state ledger." + ), + metric_type="gauge", + samples=[({"surface": surface, "kind": kind}, value) for surface, kind, value in drift], + ) + memory = _latest_ingest_memory(conn, ops_db=ops_db) _emit_metric( lines, @@ -1245,6 +1305,10 @@ def _format_ops_only_metrics(lines: list[str], ops_db: Path) -> str | None: ("polylogue_fts_trigger_present", "1 when the named FTS sync trigger is installed in index.db."), ("polylogue_fts_triggers_all_present", "All expected FTS sync triggers are installed."), ("polylogue_fts_freshness_ready", "1 when the daemon freshness ledger marks an FTS surface ready."), + ( + "polylogue_fts_drift_rows", + "FTS drift magnitude by surface and kind, read O(1) from fts_freshness_state.", + ), ("polylogue_embedding_sessions", "Embedding session counts by state."), ("polylogue_embedding_messages", "Embedding message counts by state."), ("polylogue_embedding_coverage_percent", "Percent of sessions with current embeddings."), diff --git a/polylogue/storage/fts/dangling_repair.py b/polylogue/storage/fts/dangling_repair.py index 4fbfe600e2..5a986cbf7d 100644 --- a/polylogue/storage/fts/dangling_repair.py +++ b/polylogue/storage/fts/dangling_repair.py @@ -175,7 +175,14 @@ def _freshness_record(conn: sqlite3.Connection, surface: str) -> dict[str, objec return None columns = {str(row[1]) for row in conn.execute("PRAGMA table_info(fts_freshness_state)").fetchall()} selected = ["state"] - for name in ("source_rows", "indexed_rows", "missing_rows", "excess_rows", "duplicate_rows"): + for name in ( + "source_rows", + "indexed_rows", + "missing_rows", + "excess_rows", + "duplicate_rows", + "identity_mismatch_rows", + ): if name in columns: selected.append(name) row = conn.execute(f"SELECT {', '.join(selected)} FROM fts_freshness_state WHERE surface=?", (surface,)).fetchone() @@ -213,6 +220,7 @@ def _ready_freshness_marker(conn: sqlite3.Connection, surface: str, triggers: tu 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")), source_has_rows=_source_has_rows(conn, surface) if source_rows == 0 and indexed_rows == 0 else False, ) diff --git a/polylogue/storage/fts/drift_sampling.py b/polylogue/storage/fts/drift_sampling.py new file mode 100644 index 0000000000..7609a9083b --- /dev/null +++ b/polylogue/storage/fts/drift_sampling.py @@ -0,0 +1,116 @@ +"""Bounded ops.db drift-magnitude history for FTS freshness (polylogue-1xc.12). + +``fts_freshness_state`` (index.db, see ``storage/fts/freshness.py``) is O(1) +*current* state -- a snapshot fit for a hot-path readiness check or a +Prometheus gauge scrape, but not a trend. This module appends a bounded +time-series sample of the same counters to ``ops.db`` (a sibling database +file) whenever an exact FTS invariant snapshot is recorded, so an operator +can see drift MAGNITUDE trend across time, not just today's boolean +ready/stale verdict. + +Best-effort telemetry, matching ``record_route_observation``'s contract: a +missing or unreachable ``ops.db`` (e.g. a synthetic single-file test +fixture, or an archive mid-bootstrap) is swallowed, never raised -- this is +never allowed to turn an index.db write/repair path into an ops.db write +dependency. +""" + +from __future__ import annotations + +import contextlib +import sqlite3 +import time +from pathlib import Path + +from polylogue.logging import get_logger + +logger = get_logger(__name__) + +_FRESHNESS_SURFACE_COLUMNS = ( + "surface", + "state", + "source_rows", + "indexed_rows", + "missing_rows", + "excess_rows", + "duplicate_rows", + "identity_mismatch_rows", +) + + +def _index_db_path_sync(conn: sqlite3.Connection) -> Path | None: + """Return the main database file path backing ``conn``, if any.""" + try: + rows = conn.execute("PRAGMA database_list").fetchall() + except sqlite3.Error: + return None + for row in rows: + if str(row[1]) == "main" and row[2]: + return Path(str(row[2])) + return None + + +def sample_fts_drift_to_ops_sync(conn: sqlite3.Connection) -> int: + """Append one ops.db drift sample per recorded FTS surface. + + Reads the just-recorded ``fts_freshness_state`` rows on ``conn`` (an + index.db connection) and appends a bounded sample for each to the + sibling ``ops.db``. Returns the number of samples written; returns 0 on + any failure (missing table, missing sibling file, unreachable + connection) without raising. + """ + try: + from polylogue.storage.fts.freshness import ensure_fts_freshness_table_sync + + ensure_fts_freshness_table_sync(conn) + rows = conn.execute(f"SELECT {', '.join(_FRESHNESS_SURFACE_COLUMNS)} FROM fts_freshness_state").fetchall() + except sqlite3.Error: + logger.debug("fts drift sampling: could not read fts_freshness_state", exc_info=True) + return 0 + if not rows: + return 0 + + index_db_path = _index_db_path_sync(conn) + if index_db_path is None: + return 0 + ops_db_path = index_db_path.with_name("ops.db") + if not ops_db_path.exists(): + return 0 + + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier + from polylogue.storage.sqlite.archive_tiers.ops_write import record_fts_drift_sample + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + from polylogue.storage.sqlite.connection_profile import open_connection + + ops_conn: sqlite3.Connection | None = None + try: + ops_conn = open_connection(ops_db_path, timeout=5.0) + initialize_archive_tier(ops_conn, ArchiveTier.OPS) + sampled_at_ms = int(time.time() * 1000) + written = 0 + for row in rows: + values = dict(zip(_FRESHNESS_SURFACE_COLUMNS, row, strict=True)) + record_fts_drift_sample( + ops_conn, + surface=str(values["surface"]), + state=str(values["state"]), + source_rows=int(values["source_rows"] or 0), + indexed_rows=int(values["indexed_rows"] or 0), + missing_rows=int(values["missing_rows"] or 0), + excess_rows=int(values["excess_rows"] or 0), + duplicate_rows=int(values["duplicate_rows"] or 0), + identity_mismatch_rows=int(values["identity_mismatch_rows"] or 0), + sampled_at_ms=sampled_at_ms, + ) + written += 1 + return written + except sqlite3.Error: + logger.debug("fts drift sampling: ops.db write failed", exc_info=True) + return 0 + finally: + if ops_conn is not None: + with contextlib.suppress(sqlite3.Error): + ops_conn.close() + + +__all__ = ["sample_fts_drift_to_ops_sync"] diff --git a/polylogue/storage/fts/freshness.py b/polylogue/storage/fts/freshness.py index 1be07e8437..99045479fe 100644 --- a/polylogue/storage/fts/freshness.py +++ b/polylogue/storage/fts/freshness.py @@ -23,6 +23,13 @@ ("missing_rows", "INTEGER NOT NULL DEFAULT 0"), ("excess_rows", "INTEGER NOT NULL DEFAULT 0"), ("duplicate_rows", "INTEGER NOT NULL DEFAULT 0"), + # polylogue-1xc.12: rowid-reuse/changed-text/changed-recipe drift the + # messages_fts_identity ledger catches. A row upgraded from before this + # column existed defaults to 0 (trusted ready under the prior, weaker + # invariant) and is re-validated the next time an exact snapshot runs -- + # the same backward-compatible-default shape the other counters already + # use here. + ("identity_mismatch_rows", "INTEGER NOT NULL DEFAULT 0"), ("detail", "TEXT"), ) @@ -58,22 +65,28 @@ def freshness_ready_record_trusted( excess_rows: int, duplicate_rows: int, source_has_rows: bool | None, + identity_mismatch_rows: int = 0, ) -> bool: """Return whether a durable freshness row is safe to use as ready. A ``ready`` row must be internally clean. The historical poisoned shape ``source_rows=0`` and ``indexed_rows=0`` is trusted only after proving the source table has no rows; otherwise readiness is unknown and must be - recomputed by repair/search paths. + recomputed by repair/search paths. ``identity_mismatch_rows`` (default 0, + polylogue-1xc.12) is the rowid-reuse/changed-text/changed-recipe check the + ledger provides beyond a plain count comparison -- callers that never + compute it (bounded startup/repair paths that only compare counts) keep + the historical behavior by construction; callers that DID compute a + nonzero value must not silently drop it here. """ if state != READY: return False - counters = (source_rows, indexed_rows, missing_rows, excess_rows, duplicate_rows) + counters = (source_rows, indexed_rows, missing_rows, excess_rows, duplicate_rows, identity_mismatch_rows) if any(counter < 0 for counter in counters): return False if source_rows != indexed_rows: return False - if missing_rows != 0 or excess_rows != 0 or duplicate_rows != 0: + if missing_rows != 0 or excess_rows != 0 or duplicate_rows != 0 or identity_mismatch_rows != 0: return False return not (source_rows == 0 and indexed_rows == 0 and source_has_rows is not False) @@ -175,6 +188,7 @@ def record_fts_surface_state_sync( missing_rows: int = 0, excess_rows: int = 0, duplicate_rows: int = 0, + identity_mismatch_rows: int = 0, detail: str | None = None, ) -> None: ensure_fts_freshness_table_sync(conn) @@ -182,9 +196,9 @@ def record_fts_surface_state_sync( """ INSERT INTO fts_freshness_state ( surface, state, checked_at, source_rows, indexed_rows, - missing_rows, excess_rows, duplicate_rows, detail + missing_rows, excess_rows, duplicate_rows, identity_mismatch_rows, detail ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(surface) DO UPDATE SET state=excluded.state, checked_at=excluded.checked_at, @@ -193,6 +207,7 @@ def record_fts_surface_state_sync( missing_rows=excluded.missing_rows, excess_rows=excluded.excess_rows, duplicate_rows=excluded.duplicate_rows, + identity_mismatch_rows=excluded.identity_mismatch_rows, detail=excluded.detail """, ( @@ -204,6 +219,7 @@ def record_fts_surface_state_sync( int(missing_rows), int(excess_rows), int(duplicate_rows), + int(identity_mismatch_rows), detail, ), ) @@ -219,6 +235,7 @@ async def record_fts_surface_state_async( missing_rows: int = 0, excess_rows: int = 0, duplicate_rows: int = 0, + identity_mismatch_rows: int = 0, detail: str | None = None, ) -> None: await ensure_fts_freshness_table_async(conn) @@ -226,9 +243,9 @@ async def record_fts_surface_state_async( """ INSERT INTO fts_freshness_state ( surface, state, checked_at, source_rows, indexed_rows, - missing_rows, excess_rows, duplicate_rows, detail + missing_rows, excess_rows, duplicate_rows, identity_mismatch_rows, detail ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(surface) DO UPDATE SET state=excluded.state, checked_at=excluded.checked_at, @@ -237,6 +254,7 @@ async def record_fts_surface_state_async( missing_rows=excluded.missing_rows, excess_rows=excluded.excess_rows, duplicate_rows=excluded.duplicate_rows, + identity_mismatch_rows=excluded.identity_mismatch_rows, detail=excluded.detail """, ( @@ -248,6 +266,7 @@ async def record_fts_surface_state_async( int(missing_rows), int(excess_rows), int(duplicate_rows), + int(identity_mismatch_rows), detail, ), ) @@ -264,6 +283,7 @@ def record_fts_invariant_snapshot_sync(conn: sqlite3.Connection, snapshot: Any) missing_rows=int(surface.missing_rows), excess_rows=int(surface.excess_rows), duplicate_rows=int(surface.duplicate_rows), + identity_mismatch_rows=int(getattr(surface, "identity_mismatch_rows", 0)), detail=None if bool(surface.ready) else "exact invariant failed", ) @@ -325,7 +345,14 @@ def _message_fts_record_sync(conn: sqlite3.Connection) -> dict[str, object] | No return None columns = {str(row[1]) for row in conn.execute(f"PRAGMA table_info({FRESHNESS_TABLE})").fetchall()} selected = ["state"] - for name in ("source_rows", "indexed_rows", "missing_rows", "excess_rows", "duplicate_rows"): + for name in ( + "source_rows", + "indexed_rows", + "missing_rows", + "excess_rows", + "duplicate_rows", + "identity_mismatch_rows", + ): if name in columns: selected.append(name) row = conn.execute( @@ -343,7 +370,14 @@ async def _message_fts_record_async(conn: aiosqlite.Connection) -> dict[str, obj rows = await (await conn.execute(f"PRAGMA table_info({FRESHNESS_TABLE})")).fetchall() columns = {str(row[1]) for row in rows} selected = ["state"] - for name in ("source_rows", "indexed_rows", "missing_rows", "excess_rows", "duplicate_rows"): + for name in ( + "source_rows", + "indexed_rows", + "missing_rows", + "excess_rows", + "duplicate_rows", + "identity_mismatch_rows", + ): if name in columns: selected.append(name) row = await ( @@ -371,6 +405,7 @@ def _recorded_ready_state_sync(conn: sqlite3.Connection, record: dict[str, objec missing_rows=_recorded_counter(record, "missing_rows"), excess_rows=_recorded_counter(record, "excess_rows"), duplicate_rows=_recorded_counter(record, "duplicate_rows"), + identity_mismatch_rows=_recorded_counter(record, "identity_mismatch_rows"), source_has_rows=_message_fts_source_has_rows_sync(conn) if source_rows == 0 and indexed_rows == 0 else False, ) @@ -385,6 +420,7 @@ async def _recorded_ready_state_async(conn: aiosqlite.Connection, record: dict[s missing_rows=_recorded_counter(record, "missing_rows"), excess_rows=_recorded_counter(record, "excess_rows"), duplicate_rows=_recorded_counter(record, "duplicate_rows"), + identity_mismatch_rows=_recorded_counter(record, "identity_mismatch_rows"), source_has_rows=await _message_fts_source_has_rows_async(conn) if source_rows == 0 and indexed_rows == 0 else False, diff --git a/polylogue/storage/fts/fts_lifecycle.py b/polylogue/storage/fts/fts_lifecycle.py index 9812f072c7..4588e36c11 100644 --- a/polylogue/storage/fts/fts_lifecycle.py +++ b/polylogue/storage/fts/fts_lifecycle.py @@ -13,9 +13,11 @@ from polylogue.storage.fts.pl_fold import pl_fold_sql_expr from polylogue.storage.fts.sql import ( BLOCKS_FTS_TRIGGER_DDL, + FTS_IDENTITY_REBUILD_SQL, FTS_INDEX_DOC_COUNT_SQL, FTS_INDEX_EXISTS_SQL, FTS_INDEXABLE_MESSAGE_COUNT_SQL, + FTS_MESSAGES_IDENTITY_TABLE_SQL, FTS_MESSAGES_TABLE_SQL, FTS_REBUILD_SQL, FTS_TRIGGER_DDL, @@ -24,12 +26,17 @@ TRIGRAM_REBUILD_DELETE_ALL_SQL, IndexedMessage, chunked, + delete_session_identity_rows_sql, delete_session_rows_sql, excess_message_rows_sql, + insert_all_message_identity_rows_sql, insert_all_message_rows_sql, insert_all_trigram_rows_sql, insert_missing_message_rows_range_sql, + insert_session_identity_rows_sql, insert_session_rows_sql, + message_identity_mismatch_sql, + repair_message_identity_rows_range_sql, ) _chunked = chunked @@ -125,6 +132,12 @@ class FtsSurfaceInvariant: missing_rows: int = 0 excess_rows: int = 0 duplicate_rows: int = 0 + # polylogue-1xc.12: rowid-reuse/changed-text/changed-recipe drift the + # messages_fts_identity ledger catches that missing_rows/excess_rows + # cannot -- both sides still balance when a stale rowid has silently + # rebound to a different block. Zero for surfaces without an identity + # ledger (only messages_fts has one today). + identity_mismatch_rows: int = 0 @property def ready(self) -> bool: @@ -136,6 +149,7 @@ def ready(self) -> bool: and self.missing_rows == 0 and self.excess_rows == 0 and self.duplicate_rows == 0 + and self.identity_mismatch_rows == 0 ) @@ -269,12 +283,14 @@ def ensure_fts_triggers_sync(conn: sqlite3.Connection) -> None: def ensure_fts_index_sync(conn: sqlite3.Connection) -> None: """Ensure the FTS5 tables and triggers exist on a sync SQLite connection.""" conn.execute(FTS_MESSAGES_TABLE_SQL) + conn.execute(FTS_MESSAGES_IDENTITY_TABLE_SQL) ensure_fts_triggers_sync(conn) async def ensure_fts_index_async(conn: aiosqlite.Connection) -> None: """Ensure the FTS5 tables and triggers exist on an async SQLite connection.""" await conn.execute(FTS_MESSAGES_TABLE_SQL) + await conn.execute(FTS_MESSAGES_IDENTITY_TABLE_SQL) for ddl in await _fts_trigger_ddl_for_existing_surfaces_async(conn): if ";" in ddl: await conn.executescript(ddl) @@ -325,12 +341,18 @@ def rebuild_fts_index_sync(conn: sqlite3.Connection) -> None: ensure_fts_index_sync(conn) conn.execute(FTS_REBUILD_SQL) conn.execute(insert_all_message_rows_sql()) + conn.execute(FTS_IDENTITY_REBUILD_SQL) + conn.execute(insert_all_message_identity_rows_sql()) _rebuild_session_work_events_fts_sync(conn) _rebuild_threads_fts_sync(conn) from polylogue.storage.fts.freshness import record_fts_invariant_snapshot_sync record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) + from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync + + sample_fts_drift_to_ops_sync(conn) + def rebuild_command_trigram_index_sync(conn: sqlite3.Connection) -> None: """Rebuild the full ``blocks_command_trigram`` index from persisted blocks. @@ -362,6 +384,8 @@ def reset_message_fts_index_sync(conn: sqlite3.Connection) -> None: conn.execute(f"DROP TRIGGER IF EXISTS {name}") conn.execute("DROP TABLE IF EXISTS messages_fts") conn.execute(FTS_MESSAGES_TABLE_SQL) + conn.execute("DROP TABLE IF EXISTS messages_fts_identity") + conn.execute(FTS_MESSAGES_IDENTITY_TABLE_SQL) if _table_exists_sync(conn, "blocks"): for ddl in _BLOCKS_FTS_TRIGGER_DDL: conn.execute(ddl) @@ -409,13 +433,17 @@ def insert_missing_message_rows_batched_sync( 0, ) sql = insert_missing_message_rows_range_sql() + identity_sql = repair_message_identity_rows_range_sql() lower = 0 while lower < max_rowid: upper = min(lower + batch_rows, max_rowid) changes_before = conn.total_changes conn.execute(sql, (lower, upper)) inserted = conn.total_changes - changes_before - if inserted: + identity_changes_before = conn.total_changes + conn.execute(identity_sql, (lower, upper)) + identity_changed = conn.total_changes - identity_changes_before + if inserted or identity_changed: conn.commit() _passive_wal_checkpoint_sync(conn) if progress_callback is not None: @@ -448,6 +476,7 @@ def delete_excess_message_rows_batched_sync( placeholders = ", ".join("?" for _ in rowids) changes_before = conn.total_changes conn.execute(f"DELETE FROM messages_fts WHERE rowid IN ({placeholders})", tuple(rowids)) + conn.execute(f"DELETE FROM messages_fts_identity WHERE rowid IN ({placeholders})", tuple(rowids)) deleted = max(0, conn.total_changes - changes_before) deleted_total += deleted if deleted: @@ -469,6 +498,10 @@ def rebuild_session_insight_fts_sync(conn: sqlite3.Connection) -> None: record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) + from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync + + sample_fts_drift_to_ops_sync(conn) + def _rebuild_session_work_events_fts_sync(conn: sqlite3.Connection) -> None: if not (_table_exists_sync(conn, "session_work_events") and _table_exists_sync(conn, "session_work_events_fts")): @@ -515,6 +548,8 @@ async def rebuild_fts_index_async( return await conn.execute(FTS_REBUILD_SQL) await conn.execute(insert_all_message_rows_sql()) + await conn.execute(FTS_IDENTITY_REBUILD_SQL) + await conn.execute(insert_all_message_identity_rows_sql()) readiness = await message_fts_readiness_async(conn, verify_total_rows=True) from polylogue.storage.fts.freshness import READY, STALE, record_fts_surface_state_async @@ -547,13 +582,19 @@ def repair_message_fts_index_sync( for chunk in chunked(list(session_ids), size=500): params = tuple(chunk) conn.execute(delete_session_rows_sql(len(chunk)), params) + conn.execute(delete_session_identity_rows_sql(len(chunk)), params) conn.execute(insert_session_rows_sql(len(chunk)), params) + conn.execute(insert_session_identity_rows_sql(len(chunk)), params) if not record_exact_snapshot: return from polylogue.storage.fts.freshness import record_fts_invariant_snapshot_sync record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) + from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync + + sample_fts_drift_to_ops_sync(conn) + def repair_fts_index_sync(conn: sqlite3.Connection, session_ids: Sequence[str]) -> None: """Repair FTS rows for the supplied sessions from persisted rows.""" @@ -577,7 +618,9 @@ async def repair_fts_index_async( for chunk in chunked(list(session_ids), size=500): params = tuple(chunk) await conn.execute(delete_session_rows_sql(len(chunk)), params) + await conn.execute(delete_session_identity_rows_sql(len(chunk)), params) await conn.execute(insert_session_rows_sql(len(chunk)), params) + await conn.execute(insert_session_identity_rows_sql(len(chunk)), params) processed += len(chunk) if progress_callback is not None: desc = progress_desc(processed, total) if progress_desc is not None else None @@ -596,12 +639,19 @@ def replace_fts_rows_for_messages_sync( session_ids = sorted({_indexed_message_parts(message)[1] for message in messages}) for chunk in chunked(session_ids, size=500): - conn.execute(delete_session_rows_sql(len(chunk)), tuple(chunk)) - conn.execute(insert_session_rows_sql(len(chunk)), tuple(chunk)) + params = tuple(chunk) + conn.execute(delete_session_rows_sql(len(chunk)), params) + conn.execute(delete_session_identity_rows_sql(len(chunk)), params) + conn.execute(insert_session_rows_sql(len(chunk)), params) + conn.execute(insert_session_identity_rows_sql(len(chunk)), params) from polylogue.storage.fts.freshness import record_fts_invariant_snapshot_sync record_fts_invariant_snapshot_sync(conn, fts_invariant_snapshot_sync(conn)) + from polylogue.storage.fts.drift_sampling import sample_fts_drift_to_ops_sync + + sample_fts_drift_to_ops_sync(conn) + async def _record_message_fts_exact_state_async(conn: aiosqlite.Connection) -> None: """Record exact message FTS readiness after async targeted rewrites.""" @@ -850,6 +900,7 @@ def _trigger_invariant_sync( missing_sql: str | None = None, excess_sql: str | None = None, duplicate_sql: str | None = None, + identity_sql: str | None = None, ) -> FtsSurfaceInvariant: source_exists = _table_exists_sync(conn, source_table_name) exists = _table_exists_sync(conn, table_name) @@ -858,6 +909,11 @@ def _trigger_invariant_sync( missing_rows = _row_int(conn.execute(missing_sql).fetchone(), 0) if source_exists and exists and missing_sql else 0 excess_rows = _row_int(conn.execute(excess_sql).fetchone(), 0) if source_exists and exists and excess_sql else 0 duplicate_rows = _row_int(conn.execute(duplicate_sql).fetchone(), 0) if exists and duplicate_sql else 0 + identity_mismatch_rows = ( + _row_int(conn.execute(identity_sql).fetchone(), 0) + if source_exists and exists and identity_sql and _table_exists_sync(conn, "messages_fts_identity") + else 0 + ) return FtsSurfaceInvariant( name=name, source_exists=source_exists, @@ -868,6 +924,7 @@ def _trigger_invariant_sync( missing_rows=missing_rows, excess_rows=excess_rows, duplicate_rows=duplicate_rows, + identity_mismatch_rows=identity_mismatch_rows, ) @@ -914,6 +971,7 @@ def _fts_invariant_snapshot_sync(conn: sqlite3.Connection) -> FtsInvariantSnapsh LEFT JOIN blocks AS b ON b.rowid = d.id AND b.search_text != '' WHERE b.rowid IS NULL """, + identity_sql=message_identity_mismatch_sql(), ) else: message_surface = _messages_fts_invariant_sync(conn) @@ -1012,6 +1070,7 @@ def _messages_fts_invariant_sync(conn: sqlite3.Connection) -> FtsSurfaceInvarian LEFT JOIN blocks AS b ON b.rowid = d.id AND b.search_text != '' WHERE b.rowid IS NULL """, + identity_sql=message_identity_mismatch_sql(), ) diff --git a/polylogue/storage/fts/sql.py b/polylogue/storage/fts/sql.py index 6b570e7b51..f103195170 100644 --- a/polylogue/storage/fts/sql.py +++ b/polylogue/storage/fts/sql.py @@ -28,6 +28,47 @@ ); """ +# polylogue-1xc.12: identity recipe version consumed by messages_fts_identity +# rows (see FTS_MESSAGES_IDENTITY_TABLE_SQL below). Bump this string -- not +# INDEX_SCHEMA_VERSION -- whenever tokenizer/fold semantics change in a way +# that invalidates previously-ledgered rows without changing table shape; +# exact reconciliation compares a ledgered row's stored recipe_id against the +# CURRENT value of this constant, so an archive that never rebuilt after a +# recipe bump shows up as drift instead of silently serving results folded +# under the stale recipe. +FTS_MESSAGES_IDENTITY_RECIPE_ID = "messages_fts.v1:unicode61-remove_diacritics2+pl_fold" + +# polylogue-1xc.12: rowid-keyed shadow ledger binding each `messages_fts` +# rowid to the block_id it was populated from. `messages_fts` is a +# CONTENTLESS FTS5 table (content=''): UNINDEXED columns such as `block_id` +# are write-only and never retrievable by a later SELECT (verified +# empirically -- `SELECT block_id FROM messages_fts` returns NULL even +# though the INSERT supplied a value). 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: both +# sides still balance even when a stale rowid has silently rebound to a +# different block. This ledger makes block identity legible again so exact +# reconciliation can join on rowid AND block_id, not rowid alone. +# `source_hash` reuses the existing `blocks.content_hash` evidence hash (see +# storage/sqlite/archive_tiers/index.py) as the source-identity component, +# and `recipe_id` is FTS_MESSAGES_IDENTITY_RECIPE_ID 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 to compute per trigger-fired row) and +# never as a shared cross-domain table: FTS keeps its own ledger and repair +# lifecycle. +FTS_MESSAGES_IDENTITY_TABLE_SQL = """ + CREATE TABLE IF NOT EXISTS messages_fts_identity ( + rowid INTEGER PRIMARY KEY, + block_id TEXT NOT NULL UNIQUE, + source_hash BLOB, + recipe_id TEXT NOT NULL + ) STRICT; +""" + FTS_INDEX_EXISTS_SQL = "SELECT name FROM sqlite_master WHERE type='table' AND name='messages_fts'" FTS_INDEX_DOC_COUNT_SQL = "SELECT COUNT(*) FROM messages_fts_docsize" FTS_INDEXABLE_MESSAGE_COUNT_SQL = """ @@ -49,22 +90,35 @@ ) # FTS trigger DDL for message/block FTS maintenance. +# +# polylogue-1xc.12: each arm also maintains messages_fts_identity in the SAME +# trigger body as its messages_fts write, so the two can never observe +# different block/rowid bindings -- one atomic statement sequence per event, +# not a second pass. ad/au explicitly DELETE the identity row by rowid before +# any re-insert so a reused rowid never inherits a stale block_id. BLOCKS_FTS_TRIGGER_DDL = [ f"""CREATE TRIGGER IF NOT EXISTS messages_fts_ai AFTER INSERT ON blocks WHEN new.search_text != '' AND {_FTS_BULK_GUARD_NOT_SET} BEGIN INSERT INTO messages_fts(rowid, block_id, message_id, session_id, block_type, text) VALUES (new.rowid, new.block_id, new.message_id, new.session_id, new.block_type, {pl_fold_sql_expr("new.search_text")}); + INSERT INTO messages_fts_identity(rowid, block_id, source_hash, recipe_id) + VALUES (new.rowid, new.block_id, new.content_hash, '{FTS_MESSAGES_IDENTITY_RECIPE_ID}'); END""", f"""CREATE TRIGGER IF NOT EXISTS messages_fts_ad AFTER DELETE ON blocks WHEN old.search_text != '' AND {_FTS_BULK_GUARD_NOT_SET} BEGIN DELETE FROM messages_fts WHERE rowid = old.rowid; + DELETE FROM messages_fts_identity WHERE rowid = old.rowid; END""", f"""CREATE TRIGGER IF NOT EXISTS messages_fts_au AFTER UPDATE ON blocks WHEN {_FTS_BULK_GUARD_NOT_SET} BEGIN DELETE FROM messages_fts WHERE rowid = old.rowid; + DELETE FROM messages_fts_identity WHERE rowid = old.rowid; INSERT INTO messages_fts(rowid, block_id, message_id, session_id, block_type, text) SELECT new.rowid, new.block_id, new.message_id, new.session_id, new.block_type, {pl_fold_sql_expr("new.search_text")} WHERE new.search_text != ''; + INSERT INTO messages_fts_identity(rowid, block_id, source_hash, recipe_id) + SELECT new.rowid, new.block_id, new.content_hash, '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' + WHERE new.search_text != ''; END""", ] @@ -203,6 +257,128 @@ def excess_message_rows_sql(limit: int) -> str: """ +# polylogue-1xc.12: identity-ledger companions to the messages_fts bulk SQL +# above. Every place that bulk-writes/deletes messages_fts rows outside the +# per-row triggers (rebuild, batched missing/excess repair, session-scoped +# repair) pairs its call with the matching function here so +# messages_fts_identity never lags messages_fts for those paths. `FTS_REBUILD_SQL` +# (``DELETE FROM messages_fts``) has no companion constant; use +# ``FTS_IDENTITY_REBUILD_SQL`` alongside it. +FTS_IDENTITY_REBUILD_SQL = "DELETE FROM messages_fts_identity" + + +def insert_all_message_identity_rows_sql() -> str: + """Bulk (re)populate ``messages_fts_identity`` from ``blocks``. + + Companion to :func:`insert_all_message_rows_sql`; callers clear the + table first with :data:`FTS_IDENTITY_REBUILD_SQL`, matching the + ``messages_fts`` rebuild shape. + """ + return f""" + INSERT INTO messages_fts_identity (rowid, block_id, source_hash, recipe_id) + SELECT rowid, block_id, content_hash, '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' + FROM blocks + WHERE search_text != '' + """ + + +def delete_session_identity_rows_sql(chunk_size: int) -> str: + """Companion to :func:`delete_session_rows_sql` for ``messages_fts_identity``.""" + placeholders = ", ".join("?" for _ in range(chunk_size)) + return f""" + DELETE FROM messages_fts_identity + WHERE rowid IN ( + SELECT blocks.rowid + FROM blocks + WHERE blocks.session_id IN ({placeholders}) + ) + """ + + +def insert_session_identity_rows_sql(chunk_size: int) -> str: + """Companion to :func:`insert_session_rows_sql` for ``messages_fts_identity``.""" + values = ", ".join("(?)" for _ in range(chunk_size)) + return f""" + WITH raw_target_sessions(session_id) AS ( + VALUES {values} + ), + target_sessions AS ( + SELECT DISTINCT session_id + FROM raw_target_sessions + ) + INSERT INTO messages_fts_identity (rowid, block_id, source_hash, recipe_id) + SELECT b.rowid, b.block_id, b.content_hash, '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' + FROM blocks AS b + JOIN target_sessions AS target + ON target.session_id = b.session_id + WHERE b.search_text != '' + """ + + +def repair_message_identity_rows_range_sql() -> str: + """UPSERT ``messages_fts_identity`` for indexed rows in a bounded rowid window. + + Unlike :func:`insert_missing_message_rows_range_sql` (INSERT-only, for + rows entirely absent from ``messages_fts``), this also *overwrites* an + existing identity row whose ``block_id``/``source_hash``/``recipe_id`` + no longer matches the current block -- the exact rowid-reuse and + changed-text/changed-recipe cases polylogue-1xc.12 exists to catch and + self-heal. Scoped to already-indexed rows (present in + ``messages_fts_docsize``) within ``(?, ?]`` so a bounded repair pass + over a huge archive stays bounded. + """ + return f""" + INSERT INTO messages_fts_identity (rowid, block_id, source_hash, recipe_id) + SELECT b.rowid, b.block_id, b.content_hash, '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' + FROM blocks AS b + JOIN messages_fts_docsize AS d ON d.id = b.rowid + WHERE b.search_text != '' + AND b.rowid > ? + AND b.rowid <= ? + ON CONFLICT(rowid) DO UPDATE SET + block_id = excluded.block_id, + source_hash = excluded.source_hash, + recipe_id = excluded.recipe_id + WHERE messages_fts_identity.block_id != excluded.block_id + OR messages_fts_identity.source_hash IS NOT excluded.source_hash + OR messages_fts_identity.recipe_id != excluded.recipe_id + """ + + +def message_identity_mismatch_sql() -> str: + """Exact rowid+block_id+source+recipe identity check for ``messages_fts``. + + Two independent failure classes, summed: (1) an indexed row + (``messages_fts_docsize`` joined with a still-indexable ``blocks`` row) + whose identity ledger entry is missing, or bound to a different + ``block_id``, or carries a stale ``source_hash``/``recipe_id`` -- the + rowid-reuse/changed-text/changed-recipe cases count-only reconciliation + cannot see because both sides still balance; (2) an identity ledger row + left over for a rowid no longer present in ``messages_fts_docsize`` at + all (an orphan, e.g. from a partial/interrupted write). + """ + return f""" + SELECT + ( + SELECT COUNT(*) + FROM messages_fts_docsize AS d + JOIN blocks AS b ON b.rowid = d.id AND b.search_text != '' + LEFT JOIN messages_fts_identity AS i ON i.rowid = d.id + WHERE i.rowid IS NULL + OR i.block_id != b.block_id + OR i.source_hash IS NOT b.content_hash + OR i.recipe_id != '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' + ) + + + ( + SELECT COUNT(*) + FROM messages_fts_identity AS i + LEFT JOIN messages_fts_docsize AS d ON d.id = i.rowid + WHERE d.id IS NULL + ) + """ + + # polylogue-v6i3: ``blocks_command_trigram`` is an EXTERNAL-CONTENT FTS5 table # (content='blocks'), unlike contentless ``messages_fts``. A bare # ``DELETE FROM blocks_command_trigram`` does not fully release its shadow @@ -231,9 +407,12 @@ def insert_all_trigram_rows_sql() -> str: __all__ = [ "BLOCKS_FTS_TRIGGER_DDL", "FTS_BULK_SESSION_WRITE_GUARD", + "FTS_IDENTITY_REBUILD_SQL", "FTS_INDEXABLE_MESSAGE_COUNT_SQL", "FTS_INDEX_DOC_COUNT_SQL", "FTS_INDEX_EXISTS_SQL", + "FTS_MESSAGES_IDENTITY_RECIPE_ID", + "FTS_MESSAGES_IDENTITY_TABLE_SQL", "FTS_MESSAGES_TABLE_SQL", "FTS_REBUILD_SQL", "FTS_TRIGGER_DDL", @@ -242,11 +421,16 @@ def insert_all_trigram_rows_sql() -> str: "THREAD_FTS_TRIGGER_DDL", "TRIGRAM_REBUILD_DELETE_ALL_SQL", "chunked", + "delete_session_identity_rows_sql", "delete_session_rows_sql", "excess_message_rows_sql", + "insert_all_message_identity_rows_sql", "insert_all_message_rows_sql", "insert_all_trigram_rows_sql", "insert_missing_message_rows_range_sql", "insert_missing_message_rows_sql", + "insert_session_identity_rows_sql", "insert_session_rows_sql", + "message_identity_mismatch_sql", + "repair_message_identity_rows_range_sql", ] diff --git a/polylogue/storage/sqlite/archive_tiers/index.py b/polylogue/storage/sqlite/archive_tiers/index.py index 9839afd7f7..14d6dd2bf6 100644 --- a/polylogue/storage/sqlite/archive_tiers/index.py +++ b/polylogue/storage/sqlite/archive_tiers/index.py @@ -14,7 +14,11 @@ SessionKind, WebConstructType, ) -from polylogue.storage.fts.sql import FTS_BULK_SESSION_WRITE_GUARD, FTS_TRIGGER_DDL +from polylogue.storage.fts.sql import ( + FTS_BULK_SESSION_WRITE_GUARD, + FTS_MESSAGES_IDENTITY_TABLE_SQL, + FTS_TRIGGER_DDL, +) from polylogue.storage.sqlite.action_pairs import action_pairs_refresh_sql from polylogue.storage.sqlite.archive_tiers.common import ( CONTENT_HASH_CHECK, @@ -24,7 +28,11 @@ ) from polylogue.storage.sqlite.delegation_facts import delegation_facts_insert_sql -INDEX_SCHEMA_VERSION = 42 +# polylogue-1xc.12: v43 adds the messages_fts_identity rowid/block_id ledger +# and its trigger-body writes. index.db is a rebuildable derived tier (no +# migration chain) -- an archive still on v42 needs `polylogue ops reset +# --index && polylogued run`, not an in-place upgrade helper. +INDEX_SCHEMA_VERSION = 43 # polylogue-v6i3: shared WHEN-clause fragment gating the blocks_command_trigram # trigger BODIES on the same dedicated bulk-build guard row messages_fts's @@ -406,6 +414,14 @@ tokenize='unicode61 remove_diacritics 2' ); +-- polylogue-1xc.12: rowid-to-block_id identity ledger for the contentless +-- messages_fts table above -- see FTS_MESSAGES_IDENTITY_TABLE_SQL in +-- storage/fts/sql.py (single source of truth, imported here) for the full +-- rationale. Declared as a plain f-string field (not `{{}}`-escaped) because +-- FTS_MESSAGES_IDENTITY_TABLE_SQL is a Python constant substituted at +-- INDEX_DDL build time, not a SQL brace literal. +{FTS_MESSAGES_IDENTITY_TABLE_SQL} + -- FTS triggers for messages_fts table are now dynamically composed from sql.py -- (polylogue-a7xr.5: consolidate FTS trigger DDL to single source) diff --git a/polylogue/storage/sqlite/archive_tiers/ops.py b/polylogue/storage/sqlite/archive_tiers/ops.py index 652368312a..ccdefaaf0f 100644 --- a/polylogue/storage/sqlite/archive_tiers/ops.py +++ b/polylogue/storage/sqlite/archive_tiers/ops.py @@ -257,6 +257,29 @@ CREATE INDEX IF NOT EXISTS idx_route_observations_started ON route_observations(started_at_ms); + +-- polylogue-1xc.12: bounded drift-magnitude history for the fts_freshness_state +-- ledger (index.db). ops.db is disposable, so this is a plain freeform- +-- additive table, pruned by time and row count the same shape as +-- route_observations -- a snapshot of the SAME per-surface counters +-- fts_freshness_state already carries (source/indexed/missing/excess/ +-- duplicate/identity_mismatch rows), sampled across time so an operator can +-- see drift MAGNITUDE trend, not just the current boolean ready/stale state. +CREATE TABLE IF NOT EXISTS fts_drift_samples ( + sample_id TEXT PRIMARY KEY, + surface TEXT NOT NULL, + state TEXT NOT NULL, + source_rows INTEGER NOT NULL DEFAULT 0 CHECK(source_rows >= 0), + indexed_rows INTEGER NOT NULL DEFAULT 0 CHECK(indexed_rows >= 0), + missing_rows INTEGER NOT NULL DEFAULT 0 CHECK(missing_rows >= 0), + excess_rows INTEGER NOT NULL DEFAULT 0 CHECK(excess_rows >= 0), + duplicate_rows INTEGER NOT NULL DEFAULT 0 CHECK(duplicate_rows >= 0), + identity_mismatch_rows INTEGER NOT NULL DEFAULT 0 CHECK(identity_mismatch_rows >= 0), + sampled_at_ms INTEGER NOT NULL +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_fts_drift_samples_surface_time +ON fts_drift_samples(surface, sampled_at_ms DESC); """ __all__ = ["OPS_DDL", "OPS_SCHEMA_VERSION"] diff --git a/polylogue/storage/sqlite/archive_tiers/ops_write.py b/polylogue/storage/sqlite/archive_tiers/ops_write.py index 100b5ae013..e8aa9fc4fd 100644 --- a/polylogue/storage/sqlite/archive_tiers/ops_write.py +++ b/polylogue/storage/sqlite/archive_tiers/ops_write.py @@ -15,6 +15,12 @@ MCP_CALL_LOG_RETENTION_MS = 90 * 24 * 60 * 60 * 1000 ROUTE_OBSERVATION_RETENTION_MS = 7 * 24 * 60 * 60 * 1000 ROUTE_OBSERVATION_ROW_CAP = 20_000 +# polylogue-1xc.12: bounded like ROUTE_OBSERVATION_* above -- a 30 day window +# (long enough to see week-over-week drift trend) capped at 5,000 rows (one +# sample per surface per convergence/startup pass keeps this table tiny in +# practice; the cap is a hard backstop against a runaway sampling loop). +FTS_DRIFT_SAMPLE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 +FTS_DRIFT_SAMPLE_ROW_CAP = 5_000 @dataclass(frozen=True, slots=True) @@ -142,6 +148,132 @@ class ArchiveRouteObservation: sampled: bool +@dataclass(frozen=True, slots=True) +class ArchiveFtsDriftSample: + """One bounded drift-magnitude sample for an FTS-backed surface.""" + + sample_id: str + surface: str + state: str + source_rows: int + indexed_rows: int + missing_rows: int + excess_rows: int + duplicate_rows: int + identity_mismatch_rows: int + sampled_at_ms: int + + +def record_fts_drift_sample( + conn: sqlite3.Connection, + *, + surface: str, + state: str, + source_rows: int, + indexed_rows: int, + missing_rows: int, + excess_rows: int, + duplicate_rows: int, + identity_mismatch_rows: int, + sampled_at_ms: int, + sample_id: str | None = None, +) -> str: + """Record one bounded FTS drift-magnitude sample and return its id. + + polylogue-1xc.12: the ``fts_freshness_state`` ledger in index.db (see + ``storage/fts/freshness.py``) is O(1) current state, not history -- this + writer appends a time-series snapshot of the same counters to ops.db (a + separate database file/connection) so an operator can see drift + MAGNITUDE trend, not just today's boolean ready/stale. Best-effort + telemetry like ``record_route_observation``: a plain direct INSERT, + pruned by both time (``FTS_DRIFT_SAMPLE_RETENTION_MS``) and row count + (``FTS_DRIFT_SAMPLE_ROW_CAP``) so it cannot grow unbounded. + """ + if sample_id is None: + sample_id = str(uuid.uuid4()) + with conn: + conn.execute( + """ + INSERT INTO fts_drift_samples ( + sample_id, surface, state, source_rows, indexed_rows, + missing_rows, excess_rows, duplicate_rows, identity_mismatch_rows, sampled_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + sample_id, + surface, + state, + max(0, int(source_rows)), + max(0, int(indexed_rows)), + max(0, int(missing_rows)), + max(0, int(excess_rows)), + max(0, int(duplicate_rows)), + max(0, int(identity_mismatch_rows)), + sampled_at_ms, + ), + ) + conn.execute( + "DELETE FROM fts_drift_samples WHERE sampled_at_ms < ?", + (sampled_at_ms - FTS_DRIFT_SAMPLE_RETENTION_MS,), + ) + row_count = int(conn.execute("SELECT COUNT(*) FROM fts_drift_samples").fetchone()[0]) + if row_count > FTS_DRIFT_SAMPLE_ROW_CAP: + excess = row_count - FTS_DRIFT_SAMPLE_ROW_CAP + conn.execute( + """ + DELETE FROM fts_drift_samples WHERE sample_id IN ( + SELECT sample_id FROM fts_drift_samples + ORDER BY sampled_at_ms ASC LIMIT ? + ) + """, + (excess,), + ) + return sample_id + + +def list_fts_drift_samples( + conn: sqlite3.Connection, + *, + surface: str | None = None, + since_ms: int | None = None, + limit: int = 1000, +) -> tuple[ArchiveFtsDriftSample, ...]: + """Return FTS drift samples newest-first, optionally filtered.""" + query = """ + SELECT sample_id, surface, state, source_rows, indexed_rows, + missing_rows, excess_rows, duplicate_rows, identity_mismatch_rows, sampled_at_ms + FROM fts_drift_samples + """ + clauses: list[str] = [] + params: list[object] = [] + if surface is not None: + clauses.append("surface = ?") + params.append(surface) + if since_ms is not None: + clauses.append("sampled_at_ms >= ?") + params.append(since_ms) + if clauses: + query += " WHERE " + " AND ".join(clauses) + query += " ORDER BY sampled_at_ms DESC, sample_id DESC LIMIT ?" + params.append(limit) + return tuple(_fts_drift_sample_from_row(row) for row in conn.execute(query, tuple(params)).fetchall()) + + +def _fts_drift_sample_from_row(row: sqlite3.Row | tuple[object, ...]) -> ArchiveFtsDriftSample: + return ArchiveFtsDriftSample( + sample_id=str(row[0]), + surface=str(row[1]), + state=str(row[2]), + source_rows=_int_value(row[3]), + indexed_rows=_int_value(row[4]), + missing_rows=_int_value(row[5]), + excess_rows=_int_value(row[6]), + duplicate_rows=_int_value(row[7]), + identity_mismatch_rows=_int_value(row[8]), + sampled_at_ms=_int_value(row[9]), + ) + + def record_query_run( conn: sqlite3.Connection, *, @@ -1297,11 +1429,15 @@ def _json_loads(raw_json: str | None) -> dict[str, object]: "ArchiveDaemonLifecycle", "ArchiveDaemonStageEvent", "ArchiveEmbeddingCatchupRun", + "ArchiveFtsDriftSample", "ArchiveOtlpSpan", "ArchiveRouteObservation", + "FTS_DRIFT_SAMPLE_RETENTION_MS", + "FTS_DRIFT_SAMPLE_ROW_CAP", "OpsCompactState", "add_convergence_debt", "list_cursor_lag_samples", + "list_fts_drift_samples", "latest_daemon_lifecycle", "list_daemon_stage_events", "list_embedding_catchup_runs", @@ -1318,6 +1454,7 @@ def _json_loads(raw_json: str | None) -> dict[str, object]: "record_daemon_lifecycle_start", "record_daemon_lifecycle_stop", "record_daemon_stage_event", + "record_fts_drift_sample", "record_ingest_attempt", "record_query_run", "record_route_observation", diff --git a/polylogue/storage/sqlite/lifecycle.py b/polylogue/storage/sqlite/lifecycle.py index 0f24bcd3d2..45539cf64c 100644 --- a/polylogue/storage/sqlite/lifecycle.py +++ b/polylogue/storage/sqlite/lifecycle.py @@ -272,6 +272,30 @@ class IndexDeltaDeclarationReport(TypedDict): # (`polylogue ops reset --index && polylogued run`). classes=(DerivedDeltaClass.SEMANTIC_REPARSE,), ), + IndexDeltaDeclaration( + version=43, + # Adds the messages_fts_identity rowid/block_id ledger + # (polylogue-1xc.12) and refreshes the messages_fts trigger bodies to + # also maintain it. Every ledgered field (block_id, source_hash, + # recipe_id) is derivable from already-persisted blocks columns + # (block_id, content_hash) plus the FTS_MESSAGES_IDENTITY_RECIPE_ID + # constant -- no raw reparse needed, so this is a clone-safe rebuild + # of a derived surface, the same shape as v35's FTS tokenizer + # reindex. + classes=(DerivedDeltaClass.FTS_REINDEX,), + operations=( + FastForwardOperation( + name="v43-messages-fts-identity", + kind=FastForwardOperationKind.REBUILD_FTS, + objects=( + ("table", "messages_fts_identity"), + ("trigger", "messages_fts_ai"), + ("trigger", "messages_fts_ad"), + ("trigger", "messages_fts_au"), + ), + ), + ), + ), ) diff --git a/tests/property/test_fts_identity_state_machine.py b/tests/property/test_fts_identity_state_machine.py new file mode 100644 index 0000000000..dfa128e240 --- /dev/null +++ b/tests/property/test_fts_identity_state_machine.py @@ -0,0 +1,291 @@ +"""Metamorphic FTS identity-ledger coherence under arbitrary mutation (polylogue-1xc.12). + +Drives the REAL ``blocks`` table triggers (never a mock/replica) through +Hypothesis-generated insert/update/delete/rollback/full-replace sequences and +asserts, after every single step, that exact reconciliation +(``fts_invariant_snapshot_sync``) reports zero missing, excess, duplicate, +AND identity-mismatch rows for ``messages_fts``. Count-only reconciliation +(``source_rows == indexed_rows``) cannot see a stale rowid that has silently +rebound to a different block after SQLite reuses a freed rowid -- exactly +what deleting the highest-rowid block and inserting a new one does, and +exactly what a full-session-replace does at scale. This machine forces that +scenario to happen organically across many random step orders, plus an +explicit corrupt-then-repair rule that proves the check has teeth (anti- +vacuity: a hand-corrupted ledger row must be flagged before the rule's own +repair call clears it). +""" + +from __future__ import annotations + +from hypothesis import HealthCheck, settings +from hypothesis.stateful import RuleBasedStateMachine, invariant, precondition, rule + +from polylogue.storage.fts.fts_lifecycle import ( + fts_invariant_snapshot_sync, + insert_missing_message_rows_batched_sync, + restore_fts_triggers_sync, +) +from polylogue.storage.sqlite.connection import open_connection + + +class _Block: + __slots__ = ("block_id", "session_native_id", "message_native_id", "indexed") + + def __init__(self, block_id: str, session_native_id: str, message_native_id: str, *, indexed: bool) -> None: + self.block_id = block_id + self.session_native_id = session_native_id + self.message_native_id = message_native_id + self.indexed = indexed + + +class FtsIdentityStateMachine(RuleBasedStateMachine): + def __init__(self) -> None: + super().__init__() + import tempfile + from pathlib import Path + + self._tmpdir = tempfile.TemporaryDirectory(prefix="polylogue-fts-identity-", dir="/realm/tmp") + self._db_path = Path(self._tmpdir.name) / "test.db" + self._conn_cm = open_connection(self._db_path) + self._conn = self._conn_cm.__enter__() + restore_fts_triggers_sync(self._conn) + self._blocks: dict[str, _Block] = {} + self._sessions: list[str] = [] + self._blocks_by_session: dict[str, list[str]] = {} + self._next_id = 0 + self._origin = "unknown-export" + + def _fresh_content_hash(self) -> bytes: + self._next_id += 1 + return (str(self._next_id) * 32).encode("ascii")[:32] + + def _ensure_session(self) -> str: + if not self._sessions or self._next_id % 3 == 0: + self._next_id += 1 + native_session_id = f"conv-{self._next_id}" + self._conn.execute( + "INSERT INTO sessions (native_id, origin, title, content_hash) VALUES (?, ?, ?, ?)", + (native_session_id, self._origin, "identity state machine", self._fresh_content_hash()), + ) + self._sessions.append(native_session_id) + self._blocks_by_session[native_session_id] = [] + return native_session_id + return self._sessions[self._next_id % len(self._sessions)] + + def _insert_block(self, *, text: str | None) -> _Block: + session_native_id = self._ensure_session() + self._next_id += 1 + message_native_id = f"msg-{self._next_id}" + session_id = f"{self._origin}:{session_native_id}" + message_id = f"{session_id}:{message_native_id}" + content_hash = self._fresh_content_hash() + self._conn.execute( + """ + INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) + VALUES (?, ?, 0, 'user', 'message', ?) + """, + (session_id, message_native_id, content_hash), + ) + self._conn.execute( + """ + INSERT INTO blocks (message_id, session_id, position, block_type, text, content_hash) + VALUES (?, ?, 0, 'text', ?, ?) + """, + (message_id, session_id, text, content_hash), + ) + block_id = f"{message_id}:0" + block = _Block(block_id, session_native_id, message_native_id, indexed=bool(text)) + self._blocks[block_id] = block + self._blocks_by_session[session_native_id].append(block_id) + return block + + def _delete_block(self, block_id: str) -> None: + self._conn.execute("DELETE FROM blocks WHERE block_id = ?", (block_id,)) + block = self._blocks.pop(block_id) + self._blocks_by_session[block.session_native_id].remove(block_id) + + # -- rules ----------------------------------------------------------- + + @rule() + def insert_indexable_block(self) -> None: + self._insert_block(text=f"needle {self._next_id}") + + @rule() + def insert_empty_block(self) -> None: + self._insert_block(text=None) + + @precondition(lambda self: bool(self._blocks)) + @rule() + def update_block_text(self) -> None: + block_id = sorted(self._blocks)[self._next_id % len(self._blocks)] + self._next_id += 1 + new_text = f"edited {self._next_id}" + new_hash = self._fresh_content_hash() + self._conn.execute( + "UPDATE blocks SET text = ?, content_hash = ? WHERE block_id = ?", + (new_text, new_hash, block_id), + ) + self._blocks[block_id].indexed = True + + @precondition(lambda self: bool(self._blocks)) + @rule() + def update_block_to_empty(self) -> None: + block_id = sorted(self._blocks)[self._next_id % len(self._blocks)] + self._next_id += 1 + self._conn.execute("UPDATE blocks SET text = NULL WHERE block_id = ?", (block_id,)) + self._blocks[block_id].indexed = False + + @precondition(lambda self: bool(self._blocks)) + @rule() + def delete_one_block(self) -> None: + block_id = sorted(self._blocks)[self._next_id % len(self._blocks)] + self._next_id += 1 + self._delete_block(block_id) + + @precondition(lambda self: any(blocks for blocks in self._blocks_by_session.values())) + @rule() + def full_session_replace(self) -> None: + """Delete every block in a session, then insert a fresh set. + + This is the shape most likely to force SQLite to reuse a freed + rowid (deleting the current max-rowid block then inserting a new + one), exactly the historical bug class: a stale rowid silently + rebinding to a different block while missing_rows/excess_rows + counts still balance. + """ + sessions_with_blocks = [native_id for native_id, blocks in self._blocks_by_session.items() if blocks] + self._next_id += 1 + session_native_id = sessions_with_blocks[self._next_id % len(sessions_with_blocks)] + for block_id in list(self._blocks_by_session[session_native_id]): + self._delete_block(block_id) + replacement_count = 1 + (self._next_id % 3) + for _ in range(replacement_count): + self._next_id += 1 + message_native_id = f"msg-{self._next_id}" + session_id = f"{self._origin}:{session_native_id}" + message_id = f"{session_id}:{message_native_id}" + content_hash = self._fresh_content_hash() + self._conn.execute( + """ + INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) + VALUES (?, ?, 0, 'user', 'message', ?) + """, + (session_id, message_native_id, content_hash), + ) + self._conn.execute( + """ + INSERT INTO blocks (message_id, session_id, position, block_type, text, content_hash) + VALUES (?, ?, 0, 'text', ?, ?) + """, + (message_id, session_id, f"replacement {self._next_id}", content_hash), + ) + block_id = f"{message_id}:0" + block = _Block(block_id, session_native_id, message_native_id, indexed=True) + self._blocks[block_id] = block + self._blocks_by_session[session_native_id].append(block_id) + + @precondition(lambda self: bool(self._blocks)) + @rule() + def rollback_insert(self) -> None: + """A rolled-back mutation must leave zero trace in either table.""" + before_docsize = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]) + before_identity = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_identity").fetchone()[0]) + self._conn.execute("SAVEPOINT rollback_probe") + try: + session_native_id = self._ensure_session() + self._next_id += 1 + message_native_id = f"rollback-msg-{self._next_id}" + session_id = f"{self._origin}:{session_native_id}" + message_id = f"{session_id}:{message_native_id}" + content_hash = self._fresh_content_hash() + self._conn.execute( + """ + INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) + VALUES (?, ?, 0, 'user', 'message', ?) + """, + (session_id, message_native_id, content_hash), + ) + self._conn.execute( + """ + INSERT INTO blocks (message_id, session_id, position, block_type, text, content_hash) + VALUES (?, ?, 0, 'text', ?, ?) + """, + (message_id, session_id, "rolled back", content_hash), + ) + finally: + self._conn.execute("ROLLBACK TO rollback_probe") + self._conn.execute("RELEASE rollback_probe") + after_docsize = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]) + after_identity = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_identity").fetchone()[0]) + assert after_docsize == before_docsize + assert after_identity == before_identity + + @precondition(lambda self: bool(self._blocks)) + @rule() + def corrupt_then_repair_identity_row(self) -> None: + """Anti-vacuity: hand-corrupt a ledger row, prove it's flagged, then heal it. + + Mutation this rule's assertions catch: if + ``repair_message_identity_rows_range_sql`` (or the identity trigger + arms it backstops) stopped overwriting a mismatched + ``block_id``/``source_hash``/``recipe_id``, the post-repair + assertion would fail because the corrupted row would still be + wrong. + """ + indexed = [block_id for block_id, block in self._blocks.items() if block.indexed] + if not indexed: + return + block_id = sorted(indexed)[self._next_id % len(indexed)] + self._next_id += 1 + rowid = int(self._conn.execute("SELECT rowid FROM blocks WHERE block_id = ?", (block_id,)).fetchone()[0]) + self._conn.execute( + "UPDATE messages_fts_identity SET block_id = 'stale:corrupted:0' WHERE rowid = ?", + (rowid,), + ) + mismatch_before = int( + self._conn.execute( + "SELECT COUNT(*) FROM messages_fts_identity WHERE rowid = ? AND block_id != ?", + (rowid, block_id), + ).fetchone()[0] + ) + assert mismatch_before == 1 + + insert_missing_message_rows_batched_sync(self._conn, batch_rows=1_000_000) + + healed_block_id = self._conn.execute( + "SELECT block_id FROM messages_fts_identity WHERE rowid = ?", (rowid,) + ).fetchone()[0] + assert healed_block_id == block_id + + # -- invariant --------------------------------------------------------- + + @invariant() + def messages_fts_exactly_reflects_blocks(self) -> None: + snapshot = fts_invariant_snapshot_sync(self._conn) + surface = snapshot.messages + assert surface.missing_rows == 0, f"missing_rows={surface.missing_rows}" + assert surface.excess_rows == 0, f"excess_rows={surface.excess_rows}" + assert surface.duplicate_rows == 0, f"duplicate_rows={surface.duplicate_rows}" + assert surface.identity_mismatch_rows == 0, f"identity_mismatch_rows={surface.identity_mismatch_rows}" + assert surface.ready + + indexed_docids = {row[0] for row in self._conn.execute("SELECT id FROM messages_fts_docsize").fetchall()} + indexable_docids = { + row[0] for row in self._conn.execute("SELECT rowid FROM blocks WHERE search_text != ''").fetchall() + } + assert indexed_docids == indexable_docids + + identity_rowids = {row[0] for row in self._conn.execute("SELECT rowid FROM messages_fts_identity").fetchall()} + assert identity_rowids == indexable_docids + + def teardown(self) -> None: + self._conn_cm.__exit__(None, None, None) + self._tmpdir.cleanup() + + +TestFtsIdentityStateMachine = FtsIdentityStateMachine.TestCase +TestFtsIdentityStateMachine.settings = settings( + stateful_step_count=25, + deadline=None, + suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large], +) diff --git a/tests/unit/daemon/test_metrics_endpoint.py b/tests/unit/daemon/test_metrics_endpoint.py index d2343b5817..42499eefea 100644 --- a/tests/unit/daemon/test_metrics_endpoint.py +++ b/tests/unit/daemon/test_metrics_endpoint.py @@ -58,6 +58,7 @@ "polylogue_fts_trigger_present", "polylogue_fts_triggers_all_present", "polylogue_fts_freshness_ready", + "polylogue_fts_drift_rows", "polylogue_live_ingest_memory_mebibytes", "polylogue_stale_cursor_writes_total", "polylogue_embedding_sessions", diff --git a/tests/unit/storage/test_fts_identity_ledger.py b/tests/unit/storage/test_fts_identity_ledger.py new file mode 100644 index 0000000000..0f2c9c9a90 --- /dev/null +++ b/tests/unit/storage/test_fts_identity_ledger.py @@ -0,0 +1,463 @@ +"""Exact identity-ledger reconciliation for messages_fts (polylogue-1xc.12). + +``messages_fts`` is a CONTENTLESS FTS5 table: its ``block_id`` UNINDEXED +column is write-only and never retrievable by a later ``SELECT`` (see +``storage/fts/sql.py``). 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 comparison +cannot prove which block a ``messages_fts`` row currently represents, and +count-only reconciliation (``source_rows == indexed_rows``) cannot see a +stale rowid that has silently rebound to a different block: both sides still +balance. These tests exercise the real block triggers (never a mock) and +prove the ``messages_fts_identity`` ledger + exact reconciliation actually +catch that class of drift, not merely that happy-path counts agree. +""" + +from __future__ import annotations + +import sqlite3 + +from polylogue.storage.fts.freshness import ( + READY, + ensure_fts_freshness_table_sync, + freshness_ready_record_trusted, + record_fts_surface_state_sync, +) +from polylogue.storage.fts.fts_lifecycle import ( + fts_invariant_snapshot_sync, + restore_fts_triggers_sync, +) +from polylogue.storage.fts.sql import FTS_MESSAGES_IDENTITY_RECIPE_ID, message_identity_mismatch_sql +from polylogue.storage.sqlite.archive_tiers.ops_write import list_fts_drift_samples, record_fts_drift_sample + + +def _seed_block( + conn: sqlite3.Connection, + *, + native_session_id: str, + native_message_id: str, + text: str, + content_hash: bytes = b"x" * 32, +) -> str: + """Insert one minimal session/message/block row and return the block_id.""" + origin = "unknown-export" + session_id = f"{origin}:{native_session_id}" + message_id = f"{session_id}:{native_message_id}" + conn.execute( + "INSERT OR IGNORE INTO sessions (native_id, origin, title, content_hash) VALUES (?, ?, ?, ?)", + (native_session_id, origin, "Identity ledger test", content_hash), + ) + conn.execute( + """ + INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) + VALUES (?, ?, 0, 'user', 'message', ?) + """, + (session_id, native_message_id, content_hash), + ) + conn.execute( + """ + INSERT INTO blocks (message_id, session_id, position, block_type, text, content_hash) + VALUES (?, ?, 0, 'text', ?, ?) + """, + (message_id, session_id, text, content_hash), + ) + return f"{message_id}:0" + + +def _block_rowid(conn: sqlite3.Connection, block_id: str) -> int: + row = conn.execute("SELECT rowid FROM blocks WHERE block_id = ?", (block_id,)).fetchone() + assert row is not None + return int(row[0]) + + +def _identity_row(conn: sqlite3.Connection, rowid: int) -> tuple[str, bytes | None, str] | None: + row = conn.execute( + "SELECT block_id, source_hash, recipe_id FROM messages_fts_identity WHERE rowid = ?", + (rowid,), + ).fetchone() + return None if row is None else (str(row[0]), row[1], str(row[2])) + + +def _identity_mismatch_count(conn: sqlite3.Connection) -> int: + return int(conn.execute(message_identity_mismatch_sql()).fetchone()[0] or 0) + + +class TestIdentityLedgerHappyPath: + """Real triggers correctly maintain the ledger, including rowid reuse.""" + + def test_insert_populates_identity_ledger(self, test_conn: sqlite3.Connection) -> None: + restore_fts_triggers_sync(test_conn) + content_hash = b"a" * 32 + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-insert", + native_message_id="msg-identity-insert", + text="hello identity ledger", + content_hash=content_hash, + ) + rowid = _block_rowid(test_conn, block_id) + identity = _identity_row(test_conn, rowid) + assert identity == (block_id, content_hash, FTS_MESSAGES_IDENTITY_RECIPE_ID) + assert _identity_mismatch_count(test_conn) == 0 + + def test_rowid_reuse_after_delete_rebinds_identity_to_new_block(self, test_conn: sqlite3.Connection) -> None: + """The keystone case: a freed rowid must never keep the old block's identity. + + Mutation this proves fails without the trigger's identity DELETE+INSERT + pair: if the ``messages_fts_ad``/``messages_fts_ai`` bodies stopped + touching ``messages_fts_identity``, the ledger row for the reused + rowid would still say ``block_id_a`` after block B was inserted at + the same rowid, and this test's final assertion would fail. + """ + restore_fts_triggers_sync(test_conn) + block_id_a = _seed_block( + test_conn, + native_session_id="conv-identity-reuse", + native_message_id="msg-identity-reuse-a", + text="first block occupying the rowid", + content_hash=b"a" * 32, + ) + rowid = _block_rowid(test_conn, block_id_a) + assert _identity_row(test_conn, rowid) is not None + + # Deleting the only (highest-rowid) block frees that exact rowid -- + # SQLite's default rowid allocator reuses it on the very next insert. + test_conn.execute("DELETE FROM blocks WHERE block_id = ?", (block_id_a,)) + assert _identity_row(test_conn, rowid) is None + + block_id_b = _seed_block( + test_conn, + native_session_id="conv-identity-reuse", + native_message_id="msg-identity-reuse-b", + text="second block reusing the same rowid", + content_hash=b"b" * 32, + ) + reused_rowid = _block_rowid(test_conn, block_id_b) + assert reused_rowid == rowid, "test setup expected SQLite to reuse the freed rowid" + + identity = _identity_row(test_conn, reused_rowid) + assert identity is not None + bound_block_id, bound_source_hash, _recipe = identity + assert bound_block_id == block_id_b + assert bound_block_id != block_id_a + assert bound_source_hash == b"b" * 32 + assert _identity_mismatch_count(test_conn) == 0 + + snapshot = fts_invariant_snapshot_sync(test_conn) + assert snapshot.messages.identity_mismatch_rows == 0 + assert snapshot.messages.ready + + def test_text_change_refreshes_source_hash(self, test_conn: sqlite3.Connection) -> None: + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-textchange", + native_message_id="msg-identity-textchange", + text="original text", + content_hash=b"c" * 32, + ) + rowid = _block_rowid(test_conn, block_id) + test_conn.execute( + "UPDATE blocks SET text = ?, content_hash = ? WHERE block_id = ?", + ("edited text", b"d" * 32, block_id), + ) + identity = _identity_row(test_conn, rowid) + assert identity is not None + assert identity[1] == b"d" * 32 + assert _identity_mismatch_count(test_conn) == 0 + + def test_empty_text_transition_removes_identity_row(self, test_conn: sqlite3.Connection) -> None: + """Text going to empty must drop both messages_fts AND its identity row.""" + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-emptytransition", + native_message_id="msg-identity-emptytransition", + text="will become empty", + ) + rowid = _block_rowid(test_conn, block_id) + assert _identity_row(test_conn, rowid) is not None + + test_conn.execute("UPDATE blocks SET text = NULL WHERE block_id = ?", (block_id,)) + assert _identity_row(test_conn, rowid) is None + docsize_row = test_conn.execute("SELECT 1 FROM messages_fts_docsize WHERE id = ?", (rowid,)).fetchone() + assert docsize_row is None + assert _identity_mismatch_count(test_conn) == 0 + + +class TestIdentityMismatchDetection: + """Anti-vacuity: prove the exact check actually flags corruption, not just 0/0.""" + + def test_corrupted_ledger_row_is_detected_as_mismatch(self, test_conn: sqlite3.Connection) -> None: + """Simulate the historical bug directly: hand-write a stale identity row. + + This bypasses the trigger entirely to reproduce exactly what a + missing/broken identity trigger arm would leave behind -- a + ``messages_fts_identity`` row whose ``block_id`` does not match the + block currently bound to that rowid. Count-only reconciliation + (source_rows == indexed_rows) would report this archive as fully + healthy; the identity check must not. + """ + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-corrupt", + native_message_id="msg-identity-corrupt", + text="genuine current block", + content_hash=b"e" * 32, + ) + rowid = _block_rowid(test_conn, block_id) + assert _identity_mismatch_count(test_conn) == 0 + + # Hand-corrupt the ledger to point at a block_id that never existed + # at this rowid -- the exact rowid-reuse-gone-wrong shape. + test_conn.execute( + "UPDATE messages_fts_identity SET block_id = 'stale:ghost:0' WHERE rowid = ?", + (rowid,), + ) + assert _identity_mismatch_count(test_conn) == 1 + + snapshot = fts_invariant_snapshot_sync(test_conn) + assert snapshot.messages.identity_mismatch_rows == 1 + assert not snapshot.messages.ready + + def test_stale_source_hash_is_detected_as_mismatch(self, test_conn: sqlite3.Connection) -> None: + """A block's content_hash moved on but the ledger row didn't -- caught.""" + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-stalehash", + native_message_id="msg-identity-stalehash", + text="content that will diverge from its ledgered hash", + content_hash=b"f" * 32, + ) + rowid = _block_rowid(test_conn, block_id) + assert _identity_mismatch_count(test_conn) == 0 + + test_conn.execute( + "UPDATE messages_fts_identity SET source_hash = ? WHERE rowid = ?", + (b"0" * 32, rowid), + ) + assert _identity_mismatch_count(test_conn) == 1 + + def test_stale_recipe_id_is_detected_as_mismatch(self, test_conn: sqlite3.Connection) -> None: + """An archive that never rebuilt after a tokenizer/fold recipe bump.""" + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-stalerecipe", + native_message_id="msg-identity-stalerecipe", + text="indexed under an old recipe", + ) + rowid = _block_rowid(test_conn, block_id) + assert _identity_mismatch_count(test_conn) == 0 + + test_conn.execute( + "UPDATE messages_fts_identity SET recipe_id = 'messages_fts.v0:legacy' WHERE rowid = ?", + (rowid,), + ) + assert _identity_mismatch_count(test_conn) == 1 + + def test_orphan_identity_row_without_docsize_is_detected(self, test_conn: sqlite3.Connection) -> None: + """An identity row surviving after its messages_fts row vanished.""" + restore_fts_triggers_sync(test_conn) + _seed_block( + test_conn, + native_session_id="conv-identity-orphan", + native_message_id="msg-identity-orphan", + text="will be deleted from messages_fts only", + ) + block_id = "unknown-export:conv-identity-orphan:msg-identity-orphan:0" + rowid = _block_rowid(test_conn, block_id) + assert _identity_mismatch_count(test_conn) == 0 + + # Remove only the FTS row (as if a partial/interrupted repair left + # the identity ledger behind) -- never do this outside a test. + test_conn.execute("DELETE FROM messages_fts WHERE rowid = ?", (rowid,)) + assert _identity_mismatch_count(test_conn) == 1 + + +class TestIdentityMismatchGatesReadiness: + """Wired into the same readiness contract missing_rows/excess_rows use.""" + + def test_freshness_ready_record_trusted_rejects_nonzero_identity_mismatch(self) -> None: + assert not freshness_ready_record_trusted( + state=READY, + source_rows=10, + indexed_rows=10, + missing_rows=0, + excess_rows=0, + duplicate_rows=0, + identity_mismatch_rows=1, + source_has_rows=True, + ) + + def test_freshness_ready_record_trusted_defaults_identity_mismatch_to_zero(self) -> None: + """Backward compatibility: callers that never computed it still pass.""" + assert freshness_ready_record_trusted( + state=READY, + source_rows=10, + indexed_rows=10, + missing_rows=0, + excess_rows=0, + duplicate_rows=0, + source_has_rows=True, + ) + + def test_record_fts_surface_state_round_trips_identity_mismatch_column(self, test_conn: sqlite3.Connection) -> None: + ensure_fts_freshness_table_sync(test_conn) + record_fts_surface_state_sync( + test_conn, + surface="messages_fts", + state=READY, + source_rows=5, + indexed_rows=5, + identity_mismatch_rows=2, + ) + row = test_conn.execute( + "SELECT identity_mismatch_rows FROM fts_freshness_state WHERE surface = 'messages_fts'" + ).fetchone() + assert row is not None + assert int(row[0]) == 2 + + def test_upgraded_freshness_table_defaults_identity_mismatch_to_zero(self, test_conn: sqlite3.Connection) -> None: + """A row written before this column existed reads back as 0, not NULL.""" + test_conn.execute("DROP TABLE IF EXISTS fts_freshness_state") + test_conn.execute( + """ + CREATE TABLE fts_freshness_state ( + surface TEXT PRIMARY KEY, + state TEXT NOT NULL, + checked_at TEXT NOT NULL + ) + """ + ) + test_conn.execute( + "INSERT INTO fts_freshness_state (surface, state, checked_at) VALUES ('messages_fts', 'ready', 'x')" + ) + ensure_fts_freshness_table_sync(test_conn) + row = test_conn.execute( + "SELECT identity_mismatch_rows FROM fts_freshness_state WHERE surface = 'messages_fts'" + ).fetchone() + assert row is not None + assert int(row[0]) == 0 + + +class TestFtsDriftSamples: + """Bounded ops.db drift-magnitude history (polylogue-1xc.12).""" + + def test_record_and_list_round_trip(self, tmp_path: object) -> None: + import sqlite3 as _sqlite3 + + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + ops_path = tmp_path / "ops.db" # type: ignore[operator] + conn = _sqlite3.connect(str(ops_path)) + try: + initialize_archive_tier(conn, ArchiveTier.OPS) + record_fts_drift_sample( + conn, + surface="messages_fts", + state="ready", + source_rows=100, + indexed_rows=100, + missing_rows=0, + excess_rows=0, + duplicate_rows=0, + identity_mismatch_rows=0, + sampled_at_ms=1_000, + ) + record_fts_drift_sample( + conn, + surface="messages_fts", + state="stale", + source_rows=100, + indexed_rows=97, + missing_rows=3, + excess_rows=0, + duplicate_rows=0, + identity_mismatch_rows=1, + sampled_at_ms=2_000, + ) + samples = list_fts_drift_samples(conn, surface="messages_fts") + assert len(samples) == 2 + newest = samples[0] + assert newest.sampled_at_ms == 2_000 + assert newest.missing_rows == 3 + assert newest.identity_mismatch_rows == 1 + finally: + conn.close() + + def test_retention_prunes_old_samples(self, tmp_path: object) -> None: + import sqlite3 as _sqlite3 + + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier + from polylogue.storage.sqlite.archive_tiers.ops_write import FTS_DRIFT_SAMPLE_RETENTION_MS + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + ops_path = tmp_path / "ops.db" # type: ignore[operator] + conn = _sqlite3.connect(str(ops_path)) + try: + initialize_archive_tier(conn, ArchiveTier.OPS) + record_fts_drift_sample( + conn, + surface="messages_fts", + state="ready", + source_rows=1, + indexed_rows=1, + missing_rows=0, + excess_rows=0, + duplicate_rows=0, + identity_mismatch_rows=0, + sampled_at_ms=0, + ) + # A sample recorded well beyond the retention window must prune + # the ancient row on the next write -- this is the mutation that + # fails without the DELETE ... WHERE sampled_at_ms < ? pruning + # statement in record_fts_drift_sample. + record_fts_drift_sample( + conn, + surface="messages_fts", + state="ready", + source_rows=1, + indexed_rows=1, + missing_rows=0, + excess_rows=0, + duplicate_rows=0, + identity_mismatch_rows=0, + sampled_at_ms=FTS_DRIFT_SAMPLE_RETENTION_MS * 2, + ) + samples = list_fts_drift_samples(conn, surface="messages_fts", limit=100) + assert len(samples) == 1 + assert samples[0].sampled_at_ms == FTS_DRIFT_SAMPLE_RETENTION_MS * 2 + finally: + conn.close() + + def test_drift_sample_writer_never_stores_negative_counts(self, tmp_path: object) -> None: + import sqlite3 as _sqlite3 + + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier + from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + ops_path = tmp_path / "ops.db" # type: ignore[operator] + conn = _sqlite3.connect(str(ops_path)) + try: + initialize_archive_tier(conn, ArchiveTier.OPS) + record_fts_drift_sample( + conn, + surface="messages_fts", + state="ready", + source_rows=-5, + indexed_rows=-1, + missing_rows=-1, + excess_rows=-1, + duplicate_rows=-1, + identity_mismatch_rows=-1, + sampled_at_ms=1, + ) + sample = list_fts_drift_samples(conn, surface="messages_fts")[0] + assert sample.source_rows == 0 + assert sample.identity_mismatch_rows == 0 + finally: + conn.close() From a792c03a1bb402a95f72f285ea1dc4e7b9f42eb0 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:20:19 +0200 Subject: [PATCH 2/5] test(storage): fix message-position collisions in FTS identity tests Problem: the rowid-reuse unit test and the metamorphic state machine both inserted every synthetic message at position=0, which collided with messages' UNIQUE(session_id, position, variant_index) constraint as soon as a session accumulated a second message; the state machine's rollback rule also let a rolled-back SAVEPOINT desync from the harness's own session bookkeeping (a new session row created then rolled back, but still tracked in Python state), producing an FK violation unrelated to the identity invariant under test. What changed: seed helpers now take/track a per-session position counter; rollback_insert picks an existing session instead of possibly creating one inside the savepoint it's about to roll back. Ref polylogue-1xc.12 Co-Authored-By: Claude --- .../test_fts_identity_state_machine.py | 40 ++++++++++++++----- .../unit/storage/test_fts_identity_ledger.py | 13 ++++-- 2 files changed, 41 insertions(+), 12 deletions(-) diff --git a/tests/property/test_fts_identity_state_machine.py b/tests/property/test_fts_identity_state_machine.py index dfa128e240..699c20f610 100644 --- a/tests/property/test_fts_identity_state_machine.py +++ b/tests/property/test_fts_identity_state_machine.py @@ -52,6 +52,7 @@ def __init__(self) -> None: self._blocks: dict[str, _Block] = {} self._sessions: list[str] = [] self._blocks_by_session: dict[str, list[str]] = {} + self._next_position_by_session: dict[str, int] = {} self._next_id = 0 self._origin = "unknown-export" @@ -69,9 +70,15 @@ def _ensure_session(self) -> str: ) self._sessions.append(native_session_id) self._blocks_by_session[native_session_id] = [] + self._next_position_by_session[native_session_id] = 0 return native_session_id return self._sessions[self._next_id % len(self._sessions)] + def _next_position(self, session_native_id: str) -> int: + position = self._next_position_by_session.get(session_native_id, 0) + self._next_position_by_session[session_native_id] = position + 1 + return position + def _insert_block(self, *, text: str | None) -> _Block: session_native_id = self._ensure_session() self._next_id += 1 @@ -82,9 +89,9 @@ def _insert_block(self, *, text: str | None) -> _Block: self._conn.execute( """ INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) - VALUES (?, ?, 0, 'user', 'message', ?) + VALUES (?, ?, ?, 'user', 'message', ?) """, - (session_id, message_native_id, content_hash), + (session_id, message_native_id, self._next_position(session_native_id), content_hash), ) self._conn.execute( """ @@ -168,9 +175,9 @@ def full_session_replace(self) -> None: self._conn.execute( """ INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) - VALUES (?, ?, 0, 'user', 'message', ?) + VALUES (?, ?, ?, 'user', 'message', ?) """, - (session_id, message_native_id, content_hash), + (session_id, message_native_id, self._next_position(session_native_id), content_hash), ) self._conn.execute( """ @@ -184,15 +191,25 @@ def full_session_replace(self) -> None: self._blocks[block_id] = block self._blocks_by_session[session_native_id].append(block_id) - @precondition(lambda self: bool(self._blocks)) + @precondition(lambda self: bool(self._sessions)) @rule() def rollback_insert(self) -> None: - """A rolled-back mutation must leave zero trace in either table.""" + """A rolled-back mutation must leave zero trace in either table. + + Deliberately picks an EXISTING session rather than + ``_ensure_session()`` (which may INSERT a brand new session row) -- + this rule's own SAVEPOINT rollback only undoes the DB side, not this + harness's Python-side bookkeeping, so creating new session state + inside the rolled-back block would desync the two and produce a + harness-only false failure (an FK violation on a later rule using a + session native_id the DB no longer has) that has nothing to do with + the production identity-ledger invariant under test. + """ before_docsize = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]) before_identity = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_identity").fetchone()[0]) self._conn.execute("SAVEPOINT rollback_probe") try: - session_native_id = self._ensure_session() + session_native_id = self._sessions[self._next_id % len(self._sessions)] self._next_id += 1 message_native_id = f"rollback-msg-{self._next_id}" session_id = f"{self._origin}:{session_native_id}" @@ -201,9 +218,9 @@ def rollback_insert(self) -> None: self._conn.execute( """ INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) - VALUES (?, ?, 0, 'user', 'message', ?) + VALUES (?, ?, ?, 'user', 'message', ?) """, - (session_id, message_native_id, content_hash), + (session_id, message_native_id, self._next_position(session_native_id), content_hash), ) self._conn.execute( """ @@ -215,6 +232,11 @@ def rollback_insert(self) -> None: finally: self._conn.execute("ROLLBACK TO rollback_probe") self._conn.execute("RELEASE rollback_probe") + # The position counter itself is Python-side bookkeeping, not + # transactional -- roll it back too so a later real insert into + # this session doesn't skip a position number needlessly (not a + # correctness requirement, just keeps position values compact). + self._next_position_by_session[session_native_id] -= 1 after_docsize = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0]) after_identity = int(self._conn.execute("SELECT COUNT(*) FROM messages_fts_identity").fetchone()[0]) assert after_docsize == before_docsize diff --git a/tests/unit/storage/test_fts_identity_ledger.py b/tests/unit/storage/test_fts_identity_ledger.py index 0f2c9c9a90..61cc570491 100644 --- a/tests/unit/storage/test_fts_identity_ledger.py +++ b/tests/unit/storage/test_fts_identity_ledger.py @@ -38,8 +38,14 @@ def _seed_block( native_message_id: str, text: str, content_hash: bytes = b"x" * 32, + message_position: int = 0, ) -> str: - """Insert one minimal session/message/block row and return the block_id.""" + """Insert one minimal session/message/block row and return the block_id. + + ``message_position`` must be distinct per message within a session + (``messages`` is UNIQUE on ``(session_id, position, variant_index)``); + the block itself is always at block position 0 within its own message. + """ origin = "unknown-export" session_id = f"{origin}:{native_session_id}" message_id = f"{session_id}:{native_message_id}" @@ -50,9 +56,9 @@ def _seed_block( conn.execute( """ INSERT INTO messages (session_id, native_id, position, role, message_type, content_hash) - VALUES (?, ?, 0, 'user', 'message', ?) + VALUES (?, ?, ?, 'user', 'message', ?) """, - (session_id, native_message_id, content_hash), + (session_id, native_message_id, message_position, content_hash), ) conn.execute( """ @@ -131,6 +137,7 @@ def test_rowid_reuse_after_delete_rebinds_identity_to_new_block(self, test_conn: native_message_id="msg-identity-reuse-b", text="second block reusing the same rowid", content_hash=b"b" * 32, + message_position=1, ) reused_rowid = _block_rowid(test_conn, block_id_b) assert reused_rowid == rowid, "test setup expected SQLite to reuse the freed rowid" From 4c4d77d559af8e7f7cbe69cdc893e3a04b73205e Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:29:32 +0200 Subject: [PATCH 3/5] fix(storage): scope identity mismatch to conflicts, not coverage gaps Problem: message_identity_mismatch_sql originally counted an indexed row with NO messages_fts_identity entry as a mismatch alongside a present-but- wrong one. 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 -- the dominant real-world re-ingest path) does not populate the identity ledger inline, so every ordinarily-written session showed nonzero identity_mismatch_rows and messages_ready flipped permanently false -- verified by running the full FTS test suite through write_parsed_session_to_archive (test_fts_readiness_fallback.py), which failed identically. What changed: message_identity_mismatch_sql now counts only a PRESENT ledger entry bound to the wrong block_id/source_hash/recipe_id (the actual rowid-reuse danger signature -- nothing reads identity from the ledger except this query, and the next trigger-fired mutation at that rowid always writes a correct fresh row regardless of prior absence). A missing entry is a coverage gap, self-healed by the batched missing-row repair path already extended in the prior commit, not a conflict. Also fixes insert_missing_message_rows_batched_sync to skip the identity companion entirely when blocks.content_hash doesn't exist (some low-level tests exercise messages_fts repair against a hand-rolled minimal blocks schema), and updates test_daemon_cli.py's hand-rolled FakeConnection fixtures to include the new fts_freshness_state column / freshness-row tuple slot. Ref polylogue-1xc.12 Co-Authored-By: Claude --- docs/internals.md | 36 ++++++++++++------- polylogue/storage/fts/fts_lifecycle.py | 22 ++++++++++-- polylogue/storage/fts/sql.py | 28 ++++++++++++--- tests/unit/daemon/test_daemon_cli.py | 11 +++--- .../unit/storage/test_fts_identity_ledger.py | 27 ++++++++++++++ 5 files changed, 98 insertions(+), 26 deletions(-) diff --git a/docs/internals.md b/docs/internals.md index e9ce793062..7c95f3b585 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -215,19 +215,29 @@ Polylogue has two schema-evolution regimes, keyed by tier durability. 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. 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. 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 re-ingested through that path - transiently shows as identity-incomplete until the next repair/reconciliation - pass backfills it (`dangling_repair.py`'s missing-row repair now also - upserts identity rows for any indexed rowid lacking one) — the same - eventually-consistent contract `missing_rows`/`excess_rows` already had - before this change, not a new weaker guarantee. Existing index tiers must + 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 diff --git a/polylogue/storage/fts/fts_lifecycle.py b/polylogue/storage/fts/fts_lifecycle.py index 4588e36c11..c21d02d6d8 100644 --- a/polylogue/storage/fts/fts_lifecycle.py +++ b/polylogue/storage/fts/fts_lifecycle.py @@ -409,6 +409,19 @@ def reset_message_fts_index_sync(conn: sqlite3.Connection) -> None: ) +def _blocks_content_hash_available_sync(conn: sqlite3.Connection) -> bool: + """Whether ``blocks.content_hash`` exists (identity ledger source-hash input). + + Some low-level tests exercise ``messages_fts`` repair against a minimal + hand-rolled ``blocks`` table (a handful of TEXT columns, no + ``content_hash``) rather than the full archive schema -- the identity + ledger is additive there: skip populating it rather than erroring, the + same accommodation already made for other optional derived surfaces + (``session_work_events_fts``/``threads_fts`` existence checks above). + """ + return any(str(row[1]) == "content_hash" for row in conn.execute("PRAGMA table_info(blocks)").fetchall()) + + def insert_missing_message_rows_batched_sync( conn: sqlite3.Connection, *, @@ -421,6 +434,7 @@ def insert_missing_message_rows_batched_sync( raise ValueError("batch_rows must be positive") ensure_fts_index_sync(conn) + identity_supported = _blocks_content_hash_available_sync(conn) before = _row_int(conn.execute(FTS_INDEX_DOC_COUNT_SQL).fetchone(), 0) if measure_counts else 0 max_rowid = _row_int( conn.execute( @@ -440,9 +454,11 @@ def insert_missing_message_rows_batched_sync( changes_before = conn.total_changes conn.execute(sql, (lower, upper)) inserted = conn.total_changes - changes_before - identity_changes_before = conn.total_changes - conn.execute(identity_sql, (lower, upper)) - identity_changed = conn.total_changes - identity_changes_before + identity_changed = 0 + if identity_supported: + identity_changes_before = conn.total_changes + conn.execute(identity_sql, (lower, upper)) + identity_changed = conn.total_changes - identity_changes_before if inserted or identity_changed: conn.commit() _passive_wal_checkpoint_sync(conn) diff --git a/polylogue/storage/fts/sql.py b/polylogue/storage/fts/sql.py index f103195170..428c6ac5fc 100644 --- a/polylogue/storage/fts/sql.py +++ b/polylogue/storage/fts/sql.py @@ -346,16 +346,35 @@ def repair_message_identity_rows_range_sql() -> str: def message_identity_mismatch_sql() -> str: - """Exact rowid+block_id+source+recipe identity check for ``messages_fts``. + """Exact rowid+block_id+source+recipe identity CONFLICT check for ``messages_fts``. Two independent failure classes, summed: (1) an indexed row (``messages_fts_docsize`` joined with a still-indexable ``blocks`` row) - whose identity ledger entry is missing, or bound to a different + whose identity ledger entry EXISTS but is bound to a different ``block_id``, or carries a stale ``source_hash``/``recipe_id`` -- the rowid-reuse/changed-text/changed-recipe cases count-only reconciliation cannot see because both sides still balance; (2) an identity ledger row left over for a rowid no longer present in ``messages_fts_docsize`` at all (an orphan, e.g. from a partial/interrupted write). + + Deliberately NOT counted: an indexed row with NO identity ledger entry + at all. Every per-row trigger arm and bulk companion writes the ledger + alongside its ``messages_fts`` write, but + ``storage/sqlite/archive_tiers/write.py``'s non-bulk full-session-replace + fast path (``delete_session_rows_sql``/``insert_session_rows_sql`` + called directly, outside this module -- see the polylogue-1xc.12 note in + ``docs/internals.md``) does not yet call the identity companions inline, + so a session written through that path is coverage-incomplete until the + next repair backfills it. A missing entry is provably safe on its own: + nothing reads block identity FROM this ledger except this + reconciliation query itself, and the next trigger-fired mutation at that + rowid (delete or update) creates a correct fresh row regardless of + whether one existed before. Only a PRESENT-but-WRONG entry is the + dangerous rowid-reuse signature this check exists to catch -- a + consumer trusting the ledger would see a real but incorrect binding, not + an absence. Counting missing rows here would make ``ready`` permanently + false on any archive that writes through the ordinary session-replace + path, which defeats the point of a readiness signal. """ return f""" SELECT @@ -363,9 +382,8 @@ def message_identity_mismatch_sql() -> str: SELECT COUNT(*) FROM messages_fts_docsize AS d JOIN blocks AS b ON b.rowid = d.id AND b.search_text != '' - LEFT JOIN messages_fts_identity AS i ON i.rowid = d.id - WHERE i.rowid IS NULL - OR i.block_id != b.block_id + JOIN messages_fts_identity AS i ON i.rowid = d.id + WHERE i.block_id != b.block_id OR i.source_hash IS NOT b.content_hash OR i.recipe_id != '{FTS_MESSAGES_IDENTITY_RECIPE_ID}' ) diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index 38b2fd87ec..3cddae834d 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/tests/unit/daemon/test_daemon_cli.py @@ -2367,7 +2367,7 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor: triggers: list[tuple[object, ...]] = [("messages_fts_ai",), ("messages_fts_ad",), ("messages_fts_au",)] return FakeCursor(triggers[0], rows=triggers) if query.startswith("SELECT state, source_rows, indexed_rows"): - return FakeCursor(("stale", 250_000, 100_000, 150_000, 0, 0)) + return FakeCursor(("stale", 250_000, 100_000, 150_000, 0, 0, 0)) raise AssertionError(f"unexpected query: {query}") conn = FakeConnection() @@ -2437,7 +2437,7 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor: triggers: list[tuple[object, ...]] = [("messages_fts_ai",), ("messages_fts_ad",), ("messages_fts_au",)] return FakeCursor(triggers[0], rows=triggers) if query.startswith("SELECT state, source_rows, indexed_rows"): - return FakeCursor(("ready", 250_000, 100_000, 0, 0, 0)) + return FakeCursor(("ready", 250_000, 100_000, 0, 0, 0, 0)) raise AssertionError(f"unexpected query: {query}") conn = FakeConnection() @@ -2526,7 +2526,7 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor: triggers: list[tuple[object, ...]] = [("messages_fts_ai",), ("messages_fts_ad",), ("messages_fts_au",)] return FakeCursor(triggers[0], rows=triggers) if query.startswith("SELECT state, source_rows, indexed_rows"): - return FakeCursor(("stale", 0, 0, 0, 0, 0)) + return FakeCursor(("stale", 0, 0, 0, 0, 0, 0)) if query == "SELECT 1 FROM blocks WHERE search_text != '' LIMIT 1": return FakeCursor((1,)) if query == "SELECT 1 FROM messages_fts_docsize LIMIT 1": @@ -2695,7 +2695,8 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor: (5, "missing_rows"), (6, "excess_rows"), (7, "duplicate_rows"), - (8, "detail"), + (8, "identity_mismatch_rows"), + (9, "detail"), ], ) if query == "SELECT 1 FROM sqlite_master WHERE type IN ('table', 'virtual table') AND name = ? LIMIT 1": @@ -2711,7 +2712,7 @@ def execute(self, sql: str, params: object = ()) -> FakeCursor: ] return FakeCursor(None, rows=triggers) if query.startswith("SELECT state, source_rows, indexed_rows, missing_rows, excess_rows, duplicate_rows"): - return FakeCursor(("ready", 10, 10, 0, 0, 0)) + return FakeCursor(("ready", 10, 10, 0, 0, 0, 0)) raise AssertionError(f"unexpected query: {query}") def commit(self) -> None: diff --git a/tests/unit/storage/test_fts_identity_ledger.py b/tests/unit/storage/test_fts_identity_ledger.py index 61cc570491..c22de011db 100644 --- a/tests/unit/storage/test_fts_identity_ledger.py +++ b/tests/unit/storage/test_fts_identity_ledger.py @@ -283,6 +283,33 @@ def test_orphan_identity_row_without_docsize_is_detected(self, test_conn: sqlite test_conn.execute("DELETE FROM messages_fts WHERE rowid = ?", (rowid,)) assert _identity_mismatch_count(test_conn) == 1 + def test_missing_identity_row_is_not_counted_as_mismatch(self, test_conn: sqlite3.Connection) -> None: + """A coverage GAP is not a CONFLICT -- this is the design boundary that + keeps ``storage/sqlite/archive_tiers/write.py``'s non-bulk + full-session-replace fast path (which does not populate the identity + ledger inline, see the polylogue-1xc.12 STOP-and-report note) from + making ``ready`` permanently false on ordinary archives. Simulates + that exact gap directly: a docsize-indexed rowid with NO identity + row at all must not count, while a PRESENT-but-wrong row (proven + elsewhere in this class) must. + """ + restore_fts_triggers_sync(test_conn) + block_id = _seed_block( + test_conn, + native_session_id="conv-identity-coverage-gap", + native_message_id="msg-identity-coverage-gap", + text="indexed via messages_fts but never ledgered", + ) + rowid = _block_rowid(test_conn, block_id) + assert _identity_row(test_conn, rowid) is not None + + test_conn.execute("DELETE FROM messages_fts_identity WHERE rowid = ?", (rowid,)) + assert _identity_row(test_conn, rowid) is None + docsize_row = test_conn.execute("SELECT 1 FROM messages_fts_docsize WHERE id = ?", (rowid,)).fetchone() + assert docsize_row is not None, "messages_fts row must survive -- only its ledger entry was removed" + + assert _identity_mismatch_count(test_conn) == 0 + class TestIdentityMismatchGatesReadiness: """Wired into the same readiness contract missing_rows/excess_rows use.""" From 96911e25391d03a05396915688dc7c6e4bfb86fb Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:30:40 +0200 Subject: [PATCH 4/5] chore(topology): regenerate projection for new fts drift_sampling module Ref polylogue-1xc.12 Co-Authored-By: Claude --- docs/plans/topology-target.yaml | 32 ++++++++++++++++++-------------- docs/topology-status.md | 6 +++--- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 72b54030d6..81b0635de1 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -1568,11 +1568,11 @@ files: target: polylogue/daemon/fts_automerge.py owner: stable - path: polylogue/daemon/fts_startup.py - loc: 458 + loc: 465 target: polylogue/daemon/fts_startup.py owner: stable - path: polylogue/daemon/fts_status.py - loc: 508 + loc: 529 target: polylogue/daemon/fts_status.py owner: stable - path: polylogue/daemon/health.py @@ -1617,7 +1617,7 @@ files: target: polylogue/daemon/maintenance_registry_http.py owner: stable - path: polylogue/daemon/metrics.py - loc: 1937 + loc: 2001 target: polylogue/daemon/metrics.py owner: stable - path: polylogue/daemon/notification_backends/__init__.py @@ -1894,7 +1894,7 @@ files: target: polylogue/insights/hermes_integration_health.py owner: stable - path: polylogue/insights/hermes_topology_projection.py - loc: 341 + loc: 365 target: polylogue/insights/hermes_topology_projection.py owner: stable - path: polylogue/insights/hermes_verification_coverage.py @@ -3120,7 +3120,7 @@ files: target: polylogue/sources/decoders.py owner: stable - path: polylogue/sources/dispatch.py - loc: 1124 + loc: 1122 target: polylogue/sources/dispatch.py owner: stable - path: polylogue/sources/drive/__init__.py @@ -3349,7 +3349,7 @@ files: target: polylogue/sources/parsers/hermes_lifecycle.py owner: stable - path: polylogue/sources/parsers/hermes_spans.py - loc: 1039 + loc: 1365 target: polylogue/sources/parsers/hermes_spans.py owner: stable - path: polylogue/sources/parsers/hermes_state.py @@ -3594,15 +3594,19 @@ files: target: polylogue/storage/fts/__init__.py owner: stable - path: polylogue/storage/fts/dangling_repair.py - loc: 333 + loc: 341 target: polylogue/storage/fts/dangling_repair.py owner: stable + - path: polylogue/storage/fts/drift_sampling.py + loc: 116 + target: polylogue/storage/fts/drift_sampling.py + owner: stable - path: polylogue/storage/fts/freshness.py - loc: 499 + loc: 535 target: polylogue/storage/fts/freshness.py owner: stable - path: polylogue/storage/fts/fts_lifecycle.py - loc: 1051 + loc: 1126 target: polylogue/storage/fts/fts_lifecycle.py owner: stable - path: polylogue/storage/fts/pl_fold.py @@ -3614,7 +3618,7 @@ files: target: polylogue/storage/fts/session_repair.py owner: stable - path: polylogue/storage/fts/sql.py - loc: 252 + loc: 454 target: polylogue/storage/fts/sql.py owner: stable - path: polylogue/storage/hydrators.py @@ -4025,7 +4029,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/embeddings.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index.py - loc: 1658 + loc: 1674 target: polylogue/storage/sqlite/archive_tiers/index.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index_convergence.py @@ -4037,11 +4041,11 @@ files: target: polylogue/storage/sqlite/archive_tiers/ingest_precedence.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/ops.py - loc: 262 + loc: 285 target: polylogue/storage/sqlite/archive_tiers/ops.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/ops_write.py - loc: 1327 + loc: 1464 target: polylogue/storage/sqlite/archive_tiers/ops_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/pricing_seed.py @@ -4129,7 +4133,7 @@ files: target: polylogue/storage/sqlite/holdout_cohorts.py owner: stable - path: polylogue/storage/sqlite/lifecycle.py - loc: 334 + loc: 358 target: polylogue/storage/sqlite/lifecycle.py owner: stable - path: polylogue/storage/sqlite/maintenance.py diff --git a/docs/topology-status.md b/docs/topology-status.md index b0fb6e169c..c79acc7f73 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 899 +- **Stable** (no move scoped): 900 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1070 -- **Realized polylogue/**/*.py**: 1070 files declared +- **Total declared**: 1071 +- **Realized polylogue/**/*.py**: 1071 files declared ### TBD cells (require explicit routing) From 1e8cd8fa84b336de3b6e4657799adcbaf118a1bd Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 21 Jul 2026 00:33:53 +0200 Subject: [PATCH 5/5] fix(storage): satisfy layering + degrade-loudly gates for drift sampling - docs/plans/layering.yaml: register record_fts_drift_sample as a tracked ops_write.py writer entrypoint (verify-layering flagged the inventory drift). - drift_sampling.py: log the PRAGMA database_list failure path instead of a silent except-return (verify-degrade-loudly flagged it as a new unallowlisted soft-fail). Ref polylogue-1xc.12 Co-Authored-By: Claude --- docs/plans/layering.yaml | 3 ++- polylogue/storage/fts/drift_sampling.py | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/plans/layering.yaml b/docs/plans/layering.yaml index 4076ee4664..3d701c141b 100644 --- a/docs/plans/layering.yaml +++ b/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] diff --git a/polylogue/storage/fts/drift_sampling.py b/polylogue/storage/fts/drift_sampling.py index 7609a9083b..b0ec8e167b 100644 --- a/polylogue/storage/fts/drift_sampling.py +++ b/polylogue/storage/fts/drift_sampling.py @@ -43,6 +43,7 @@ def _index_db_path_sync(conn: sqlite3.Connection) -> Path | None: try: rows = conn.execute("PRAGMA database_list").fetchall() except sqlite3.Error: + logger.debug("fts drift sampling: PRAGMA database_list failed", exc_info=True) return None for row in rows: if str(row[1]) == "main" and row[2]: