diff --git a/docs/internals.md b/docs/internals.md index 31d7b22760..7c95f3b585 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -188,6 +188,61 @@ Polylogue has two schema-evolution regimes, keyed by tier durability. hash, so concurrent publishers of identical bytes remain independent. Existing v3 tiers migrate additively through `004_blob_publication_reservations.sql` after a verified backup manifest. +- Index schema version 43 adds `messages_fts_identity`, a rowid-keyed shadow + ledger binding each `messages_fts` rowid to the `block_id` it was populated + from (polylogue-1xc.12). `messages_fts` is a contentless FTS5 table + (`content=''`): its `UNINDEXED` columns (including `block_id`) are + write-only and never retrievable by a later `SELECT`. SQLite reuses freed + rowids (deleting the highest-rowid block then inserting a new one commonly + gets the same rowid back — exactly what a full-session-replace does), so a + bare rowid cannot prove which block a `messages_fts` row currently + represents; count-only reconciliation (`source_rows == indexed_rows`) is + blind to this because both sides still balance even when a stale rowid has + silently rebound to a different block. The ledger's `source_hash` column + reuses the existing `blocks.content_hash` evidence hash as the + source-identity component and `recipe_id` is the versioned + `FTS_MESSAGES_IDENTITY_RECIPE_ID` constant as the recipe-identity + component — the same subject/source/recipe separation + `storage/derivation_identity.py` formalizes for polylogue-wmsc's + `DerivationKey`, applied here as a lightweight per-row ledger (not a full + `DerivationKey` digest — too expensive per trigger-fired row) and never as + a shared cross-domain table: FTS keeps its own ledger and repair lifecycle. + The `messages_fts_ai`/`ad`/`au` trigger bodies gain a paired + insert/delete/upsert against `messages_fts_identity` in the SAME trigger + body as their `messages_fts` write, so the two tables can never observe a + different rowid/block_id binding for the same event. Exact reconciliation + (`fts_invariant_snapshot_sync`, `storage/fts/sql.py:message_identity_mismatch_sql`) + now also joins `blocks`/`messages_fts_docsize`/`messages_fts_identity` on + rowid AND `block_id`/`source_hash`/`recipe_id`, catching rowid-reuse, + changed-text, and changed-recipe drift that a count-only or rowid-only + check cannot see. `message_identity_mismatch_sql` deliberately counts only + a PRESENT-but-WRONG ledger entry, never a missing one: nothing reads block + identity from the ledger except this reconciliation query itself, and the + next trigger-fired mutation at that rowid creates a correct fresh row + regardless of whether one existed before, so an absent entry is a coverage + gap, not a conflict — counting it would make `ready` permanently false on + any archive that writes through the ordinary session-replace path (see + below), defeating the point of a readiness signal. Bulk paths outside the + per-row triggers (full rebuild, batched missing/excess repair, + session-scoped repair in `storage/fts/fts_lifecycle.py`) pair their + `messages_fts` writes with the matching identity companion SQL, and the + batched missing-row repair also opportunistically UPSERTs a correct entry + for any indexed rowid whose ledger entry is missing or wrong. The one + exception is `storage/sqlite/archive_tiers/write.py`'s non-bulk + full-session-replace fast path (`delete_session_rows_sql`/ + `insert_session_rows_sql` called directly, bypassing the suspended block + triggers): it does not yet call the identity companions inline, so a + session written through that path is coverage-incomplete (not + conflicting) until the next repair backfills it — a STOP-and-report gap + (polylogue-1xc.12): closing it fully needs one additional paired + `delete_session_identity_rows_sql`/`insert_session_identity_rows_sql` + call at each of write.py's four `delete_session_rows_sql`/ + `insert_session_rows_sql` call sites. Existing index tiers must + be rebuilt from source evidence (`polylogue ops reset --index && polylogued + run`) to populate the new ledger for already-indexed rows; a declared + clone-safe fast-forward exists (`IndexDeltaDeclaration` v43 in + `polylogue/storage/sqlite/lifecycle.py`) since every ledgered field is + re-derivable from already-persisted `blocks` columns with no raw reparse. - Index schema version 42 stops materializing `session_events` rows for four event types that are fully redundant with a sibling typed table (`token_count`, `message_usage`, `agent_policy`, `agent_message`; 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/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) 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..b0ec8e167b --- /dev/null +++ b/polylogue/storage/fts/drift_sampling.py @@ -0,0 +1,117 @@ +"""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: + 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]: + 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..c21d02d6d8 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) @@ -385,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, *, @@ -397,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( @@ -409,13 +447,19 @@ 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_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) if progress_callback is not None: @@ -448,6 +492,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 +514,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 +564,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 +598,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 +634,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 +655,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 +916,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 +925,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 +940,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 +987,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 +1086,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..428c6ac5fc 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,146 @@ 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 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 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 + ( + SELECT COUNT(*) + FROM messages_fts_docsize AS d + JOIN blocks AS b ON b.rowid = d.id AND b.search_text != '' + 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}' + ) + + + ( + 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 +425,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 +439,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..699c20f610 --- /dev/null +++ b/tests/property/test_fts_identity_state_machine.py @@ -0,0 +1,313 @@ +"""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_position_by_session: dict[str, int] = {} + 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] = [] + 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 + 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 (?, ?, ?, 'user', 'message', ?) + """, + (session_id, message_native_id, self._next_position(session_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 (?, ?, ?, 'user', 'message', ?) + """, + (session_id, message_native_id, self._next_position(session_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._sessions)) + @rule() + def rollback_insert(self) -> None: + """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._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}" + 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 (?, ?, ?, 'user', 'message', ?) + """, + (session_id, message_native_id, self._next_position(session_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") + # 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 + 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_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/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..c22de011db --- /dev/null +++ b/tests/unit/storage/test_fts_identity_ledger.py @@ -0,0 +1,497 @@ +"""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, + message_position: int = 0, +) -> str: + """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}" + 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 (?, ?, ?, 'user', 'message', ?) + """, + (session_id, native_message_id, message_position, 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, + 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" + + 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 + + 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.""" + + 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()