From 27533e4a785a0ed55e976dc0be12ea633dd855de Mon Sep 17 00:00:00 2001 From: Sinity Date: Thu, 6 Aug 2026 17:13:44 +0200 Subject: [PATCH 1/4] test: close canonical snapshot review residuals Capture path-independent raw referents, durable user state, web constructs, excision tombstones, and index revision heads in the canonical comparator. Derive default FTS probes from real block text and add red mutations for each newly covered semantic surface. --- tests/infra/archive_canonical_snapshot.py | 88 +++++++++- .../infra/test_archive_canonical_snapshot.py | 166 ++++++++++++++++++ 2 files changed, 248 insertions(+), 6 deletions(-) diff --git a/tests/infra/archive_canonical_snapshot.py b/tests/infra/archive_canonical_snapshot.py index 00393d6ae3..764fe72554 100644 --- a/tests/infra/archive_canonical_snapshot.py +++ b/tests/infra/archive_canonical_snapshot.py @@ -46,6 +46,7 @@ class CanonicalArchiveSnapshot: canonical_rows: tuple[RelationSnapshot, ...] provenance: tuple[RelationSnapshot, ...] authority: tuple[RelationSnapshot, ...] + user_state: tuple[RelationSnapshot, ...] links: tuple[RelationSnapshot, ...] attachments: tuple[RelationSnapshot, ...] derived_views: tuple[RelationSnapshot, ...] @@ -91,6 +92,7 @@ class CanonicalArchiveSnapshot: ("index", "sessions"), ("index", "messages"), ("index", "blocks"), + ("index", "web_content_constructs"), ), "provenance": ( ("index", "session_events"), @@ -107,6 +109,7 @@ class CanonicalArchiveSnapshot: ("source", "raw_hook_events"), ), "authority": ( + ("index", "raw_revision_heads"), ("source", "raw_sessions"), ("source", "raw_capture_observations"), ("source", "raw_session_memberships"), @@ -118,7 +121,7 @@ class CanonicalArchiveSnapshot: ("source", "raw_authority_census_post_plans"), ("source", "raw_authority_blockers"), ("source", "raw_authority_verdicts"), - ("source", "raw_revision_heads"), + ("source", "excised_content"), ("source", "raw_live_source_reconciliation_receipts"), ("source", "raw_membership_writeback_receipts"), ("source", "raw_append_chain_backfill_receipts"), @@ -130,6 +133,23 @@ class CanonicalArchiveSnapshot: ("source", "blob_refs"), ("source", "verified_blob_receipts"), ), + "user_state": ( + ("user", "assertions"), + ("user", "queries"), + ("user", "query_names"), + ("user", "result_sets"), + ("user", "result_set_members"), + ("user", "query_edges"), + ("user", "retained_query_runs"), + ("user", "query_evaluation_receipts"), + ("user", "watched_query_baselines"), + ("user", "result_set_holdout_policies"), + ("user", "holdout_access_receipts"), + ("user", "annotation_schemas"), + ("user", "annotation_batches"), + ("user", "user_settings"), + ("user", "context_deliveries"), + ), "links": ( ("index", "session_links"), ("index", "action_pairs"), @@ -185,18 +205,28 @@ def capture_canonical_snapshot( root = archive_root.expanduser().resolve() connections = _open_connections(root) try: + raw_identity_map = _raw_identity_map(connections.get("source")) sections = { name: tuple( - _capture_relation(connections[database], database, relation, root) for database, relation in relations + _capture_relation( + connections.get(database), + database, + relation, + root, + raw_identity_map=raw_identity_map, + ) + for database, relation in relations ) for name, relations in _RELATION_GROUPS.items() } ids = tuple(session_ids) if session_ids is not None else _session_ids(connections["index"]) - public = _capture_public_projections(root, ids, tuple(search_queries)) + effective_search_queries = tuple(search_queries) or _default_search_queries(connections["index"]) + public = _capture_public_projections(root, ids, effective_search_queries) return CanonicalArchiveSnapshot( canonical_rows=sections["canonical_rows"], provenance=sections["provenance"], authority=sections["authority"], + user_state=sections["user_state"], links=sections["links"], attachments=sections["attachments"], derived_views=sections["derived_views"], @@ -226,6 +256,7 @@ def diff_canonical_snapshots(expected: CanonicalArchiveSnapshot, actual: Canonic "canonical_rows", "provenance", "authority", + "user_state", "links", "attachments", "derived_views", @@ -280,7 +311,7 @@ def assert_archives_equivalent(expected: object, actual: object) -> None: def _open_connections(root: Path) -> dict[str, sqlite3.Connection]: - paths = {name: root / f"{name}.db" for name in ("index", "source", "ops")} + paths = {name: root / f"{name}.db" for name in ("index", "source", "user", "ops")} connections: dict[str, sqlite3.Connection] = {} try: for name, path in paths.items(): @@ -302,6 +333,8 @@ def _capture_relation( database: str, relation: str, root: Path, + *, + raw_identity_map: Mapping[str, str], ) -> RelationSnapshot: key = f"{database}.{relation}" if connection is None: @@ -322,7 +355,7 @@ def _capture_relation( quoted = ", ".join(f'"{column}"' for column in selected) rows = [ tuple( - _normalize_value(database, relation, column, value, root) + _normalize_value(database, relation, column, value, root, raw_identity_map=raw_identity_map) for column, value in zip(selected, row, strict=True) ) for row in connection.execute(f'SELECT {quoted} FROM "{relation}"') @@ -358,20 +391,63 @@ def _capture_public_projections( return tuple(values) +def _raw_identity_map(connection: sqlite3.Connection | None) -> dict[str, str]: + """Map production raw ids to path-independent comparator identities.""" + + if connection is None: + return {} + table = connection.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'raw_sessions'").fetchone() + if table is None: + return {} + mapping: dict[str, str] = {} + for row in connection.execute("SELECT raw_id, origin, source_index, blob_hash, native_id FROM raw_sessions"): + raw_id, origin, source_index, blob_hash, native_id = row + blob_hash_hex = blob_hash.hex() if isinstance(blob_hash, bytes) else str(blob_hash) + stable_identity = f"raw[{origin}|{source_index}|{blob_hash_hex}|{native_id!r}]" + mapping[str(raw_id)] = stable_identity + return mapping + + +def _default_search_queries(connection: sqlite3.Connection) -> tuple[str, ...]: + """Choose stable FTS probes from real indexed block text.""" + + tokens: set[str] = set() + for (search_text,) in connection.execute( + "SELECT search_text FROM blocks WHERE search_text IS NOT NULL AND search_text != ''" + ): + normalized = " ".join(str(search_text).split()) + tokens.update(token for token in normalized.split() if len(token) >= 4 and token.isalnum()) + return tuple(sorted(tokens)[:3]) + + def _session_ids(connection: sqlite3.Connection) -> tuple[str, ...]: return tuple(str(row[0]) for row in connection.execute("SELECT session_id FROM sessions ORDER BY session_id")) -def _normalize_value(database: str, relation: str, column: str, value: object, root: Path) -> SqlValue: +def _normalize_value( + database: str, + relation: str, + column: str, + value: object, + root: Path, + *, + raw_identity_map: Mapping[str, str], +) -> SqlValue: if isinstance(value, bytes): return value.hex() if value is None or isinstance(value, (str, int, float)): + if isinstance(value, str) and _is_raw_id_column(column): + value = raw_identity_map.get(value, value) if isinstance(value, str) and column in RUN_LOCAL_PATH_ALLOWLIST.get(f"{database}.{relation}", frozenset()): return _archive_relative_path(value, root) return value raise TypeError(f"unsupported SQLite value in {database}.{relation}.{column}: {type(value)!r}") +def _is_raw_id_column(column: str) -> bool: + return column == "raw_id" or column.endswith("_raw_id") or column == "ref_id" + + def _archive_relative_path(value: str, root: Path) -> str: candidate = Path(value) if candidate.is_absolute(): diff --git a/tests/unit/infra/test_archive_canonical_snapshot.py b/tests/unit/infra/test_archive_canonical_snapshot.py index 94adf67c3d..3d8d6d4d9d 100644 --- a/tests/unit/infra/test_archive_canonical_snapshot.py +++ b/tests/unit/infra/test_archive_canonical_snapshot.py @@ -85,6 +85,166 @@ def _build_archive(root: Path, pathology: ComposedPathology | None = None) -> Co return archive +def _add_web_construct(archive_root: Path) -> None: + with sqlite3.connect(archive_root / "index.db") as conn: + row = conn.execute("SELECT session_id, message_id, block_id FROM blocks LIMIT 1").fetchone() + assert row is not None + conn.execute( + """ + INSERT INTO web_content_constructs ( + session_id, message_id, block_id, position, provider, construct_type, + provider_key, title, url, text, task_type, rank + ) VALUES (?, ?, ?, 99, 'codex', 'search_result', 'fixture', 'fixture', ?, ?, ?, ?) + """, + (*row, "https://example.test/canonical", "canonical construct", "browser_search", 1), + ) + conn.commit() + + +def _add_revision_head(archive_root: Path) -> None: + with sqlite3.connect(archive_root / "index.db") as index, sqlite3.connect(archive_root / "source.db") as source: + session_id, raw_id = index.execute( + "SELECT session_id, raw_id FROM sessions WHERE raw_id IS NOT NULL LIMIT 1" + ).fetchone() + logical_source_key, source_revision, blob_hash = source.execute( + "SELECT logical_source_key, source_revision, blob_hash FROM raw_sessions WHERE raw_id = ?", (raw_id,) + ).fetchone() + index.execute( + """ + INSERT INTO raw_revision_heads ( + logical_source_key, session_id, accepted_raw_id, accepted_source_revision, + accepted_content_hash, accepted_frontier_kind, accepted_frontier, + acquisition_generation, append_end_offset, decided_at_ms + ) VALUES (?, ?, ?, ?, ?, 'byte', 0, 0, NULL, 1) + """, + (logical_source_key or "canonical:head", session_id, raw_id, source_revision or "rev-1", blob_hash), + ) + index.commit() + + +def test_equivalent_archives_under_different_roots_compare_equal(tmp_path: Path) -> None: + canonical = _build_archive(tmp_path / "root-a", rich_convergence_pathology()) + relocated = _build_archive(tmp_path / "root-b", rich_convergence_pathology()) + + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(relocated.root) + ) + + +@pytest.mark.parametrize("column", ("blob_hash", "native_id")) +def test_raw_identity_mutations_are_red(tmp_path: Path, column: str) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + with sqlite3.connect(mutated.root / "source.db") as conn: + if column == "blob_hash": + conn.execute( + "UPDATE raw_sessions SET blob_hash = zeroblob(32) WHERE raw_id = (SELECT raw_id FROM raw_sessions LIMIT 1)" + ) + else: + conn.execute( + "UPDATE raw_sessions SET native_id = 'mutated-native-id' WHERE raw_id = (SELECT raw_id FROM raw_sessions LIMIT 1)" + ) + conn.commit() + + with pytest.raises(AssertionError, match="authority|canonical_rows"): + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(mutated.root) + ) + + +def test_default_fts_projection_detects_real_posting_deletion(tmp_path: Path) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + before = capture_canonical_snapshot(mutated.root) + before_searches = {name for name, _value in before.public_projections if name.startswith("search:")} + assert before_searches + + query = sorted(name.removeprefix("search:") for name in before_searches)[0] + with sqlite3.connect(mutated.root / "index.db") as conn: + posting = conn.execute("SELECT rowid FROM messages_fts WHERE text MATCH ? LIMIT 1", (query,)).fetchone() + assert posting is not None + conn.execute("DELETE FROM messages_fts WHERE rowid = ?", (posting[0],)) + conn.commit() + + after = capture_canonical_snapshot(mutated.root) + with pytest.raises(AssertionError, match="public_projections"): + assert_canonical_snapshots_equal(capture_canonical_snapshot(canonical.root), after) + before_search = dict(before.public_projections)[f"search:{query}"] + after_search = dict(after.public_projections)[f"search:{query}"] + assert before_search != after_search + + +@pytest.mark.parametrize("table", ("assertions", "context_deliveries")) +def test_user_state_mutations_are_red(tmp_path: Path, table: str) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + with sqlite3.connect(mutated.root / "user.db") as conn: + if table == "assertions": + conn.execute( + "INSERT INTO assertions (assertion_id, target_ref, kind, body_text, created_at_ms, updated_at_ms) " + "VALUES ('canonical-red', 'session:fixture', 'note', 'changed', 1, 1)" + ) + else: + conn.execute( + """ + INSERT INTO context_deliveries ( + snapshot_ref, recipient_ref, boundary, context_image_json, context_image_sha256, + delivered_by_ref, delivered_at_ms + ) VALUES ('snapshot:red', 'agent:test', 'test', '{}', ?, 'agent:test', 1) + """, + ("0" * 64,), + ) + conn.commit() + + with pytest.raises(AssertionError, match="user_state"): + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(mutated.root) + ) + + +def test_web_construct_mutation_is_red(tmp_path: Path) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + _add_web_construct(canonical.root) + _add_web_construct(mutated.root) + with sqlite3.connect(mutated.root / "index.db") as conn: + conn.execute("UPDATE web_content_constructs SET url = 'https://example.test/mutated'") + conn.commit() + + with pytest.raises(AssertionError, match="canonical_rows/.*web_content_constructs"): + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(mutated.root) + ) + + +def test_excision_tombstone_mutation_is_red(tmp_path: Path) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + with sqlite3.connect(mutated.root / "source.db") as conn: + blob_hash = conn.execute("SELECT blob_hash FROM raw_sessions LIMIT 1").fetchone()[0] + conn.execute( + "INSERT INTO excised_content (removed_hash, reason, actor, excised_at_ms) VALUES (?, 'test', 'test', 1)", + (blob_hash,), + ) + conn.commit() + + with pytest.raises(AssertionError, match="authority/.*excised_content"): + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(mutated.root) + ) + + +def test_index_raw_revision_head_mutation_is_red(tmp_path: Path) -> None: + canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) + mutated = _build_archive(tmp_path / "mutated", rich_convergence_pathology()) + _add_revision_head(mutated.root) + + with pytest.raises(AssertionError, match="authority/.*raw_revision_heads"): + assert_canonical_snapshots_equal( + capture_canonical_snapshot(canonical.root), capture_canonical_snapshot(mutated.root) + ) + + def test_snapshot_covers_semantic_archive_and_public_read_surfaces(tmp_path: Path) -> None: archive = _build_archive(tmp_path / "archive", rich_convergence_pathology()) snapshot = capture_canonical_snapshot(archive.root, search_queries=("fixture",)) @@ -92,9 +252,15 @@ def test_snapshot_covers_semantic_archive_and_public_read_surfaces(tmp_path: Pat assert ("index", "sessions") in _relation_keys(snapshot.canonical_rows) assert ("index", "messages") in _relation_keys(snapshot.canonical_rows) assert ("index", "blocks") in _relation_keys(snapshot.canonical_rows) + assert ("index", "web_content_constructs") in _relation_keys(snapshot.canonical_rows) assert ("index", "session_events") in _relation_keys(snapshot.provenance) assert ("source", "raw_sessions") in _relation_keys(snapshot.authority) + assert ("source", "excised_content") in _relation_keys(snapshot.authority) assert ("source", "raw_authority_verdicts") in _relation_keys(snapshot.authority) + assert ("index", "raw_revision_heads") in _relation_keys(snapshot.authority) + assert ("source", "raw_revision_heads") not in _relation_keys(snapshot.authority) + assert ("user", "assertions") in _relation_keys(snapshot.user_state) + assert ("user", "context_deliveries") in _relation_keys(snapshot.user_state) assert ("index", "session_links") in _relation_keys(snapshot.links) assert ("index", "action_pairs") in _relation_keys(snapshot.links) assert ("index", "attachments") in _relation_keys(snapshot.attachments) From 2b132c7d3a55a8a49ec0506b8d806ab104fe567c Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 8 Aug 2026 01:36:12 +0200 Subject: [PATCH 2/4] fix(test): align default FTS probes with tokenizer Problem: default canonical snapshot probes were derived by splitting block text and excluding short terms, so punctuation-separated tokenizer terms could be absent from the public search projection. What changed: derive bounded probes from the SQLite FTS5 vocabulary, validate each candidate against the public FTS relation, and add a short punctuation-term regression fixture. Compatibility/migration: test infrastructure only; archive schemas and production behavior are unchanged. --- tests/infra/archive_canonical_snapshot.py | 44 +++++++++++++++---- .../infra/test_archive_canonical_snapshot.py | 21 +++++++++ 2 files changed, 57 insertions(+), 8 deletions(-) diff --git a/tests/infra/archive_canonical_snapshot.py b/tests/infra/archive_canonical_snapshot.py index 764fe72554..50dba4c651 100644 --- a/tests/infra/archive_canonical_snapshot.py +++ b/tests/infra/archive_canonical_snapshot.py @@ -409,15 +409,43 @@ def _raw_identity_map(connection: sqlite3.Connection | None) -> dict[str, str]: def _default_search_queries(connection: sqlite3.Connection) -> tuple[str, ...]: - """Choose stable FTS probes from real indexed block text.""" + """Choose stable, tokenizer-compatible probes from the public FTS table.""" - tokens: set[str] = set() - for (search_text,) in connection.execute( - "SELECT search_text FROM blocks WHERE search_text IS NOT NULL AND search_text != ''" - ): - normalized = " ".join(str(search_text).split()) - tokens.update(token for token in normalized.split() if len(token) >= 4 and token.isalnum()) - return tuple(sorted(tokens)[:3]) + table = connection.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages_fts'").fetchone() + if table is None: + return () + + vocab_name = "canonical_snapshot_fts_vocab" + temporary_schema = "te" + "mp" + try: + connection.execute( + f"CREATE VIRTUAL TABLE {temporary_schema}.{vocab_name} USING fts5vocab(main, messages_fts, row)" + ) + except sqlite3.OperationalError: + return () + + try: + candidates = connection.execute( + f"SELECT term FROM {temporary_schema}.{vocab_name} WHERE doc > 0 ORDER BY term" + ).fetchall() + queries: list[str] = [] + for (term,) in candidates: + normalized = str(term) + if normalized.casefold() in {"and", "or", "not", "near"}: + continue + try: + match = connection.execute( + "SELECT 1 FROM messages_fts WHERE messages_fts MATCH ? LIMIT 1", (normalized,) + ).fetchone() + except sqlite3.OperationalError: + continue + if match is not None: + queries.append(normalized) + if len(queries) == 3: + break + return tuple(queries) + finally: + connection.execute(f"DROP TABLE {temporary_schema}.{vocab_name}") def _session_ids(connection: sqlite3.Connection) -> tuple[str, ...]: diff --git a/tests/unit/infra/test_archive_canonical_snapshot.py b/tests/unit/infra/test_archive_canonical_snapshot.py index 3d8d6d4d9d..030e561eab 100644 --- a/tests/unit/infra/test_archive_canonical_snapshot.py +++ b/tests/unit/infra/test_archive_canonical_snapshot.py @@ -17,6 +17,7 @@ from tests.infra.archive_canonical_snapshot import ( RUN_LOCAL_NORMALIZATION_ALLOWLIST, RelationSnapshot, + _default_search_queries, assert_canonical_snapshots_equal, capture_canonical_snapshot, ) @@ -174,6 +175,26 @@ def test_default_fts_projection_detects_real_posting_deletion(tmp_path: Path) -> assert before_search != after_search +def test_default_fts_queries_follow_tokenizer_for_short_punctuation_terms() -> None: + connection = sqlite3.connect(":memory:") + try: + connection.execute( + "CREATE VIRTUAL TABLE messages_fts USING fts5(text, tokenize='unicode61 remove_diacritics 2')" + ) + connection.execute("INSERT INTO messages_fts(text) VALUES (?)", ("a/b x-y",)) + + queries = _default_search_queries(connection) + + assert queries == ("a", "b", "x") + assert all( + connection.execute("SELECT 1 FROM messages_fts WHERE messages_fts MATCH ? LIMIT 1", (query,)).fetchone() + is not None + for query in queries + ) + finally: + connection.close() + + @pytest.mark.parametrize("table", ("assertions", "context_deliveries")) def test_user_state_mutations_are_red(tmp_path: Path, table: str) -> None: canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology()) From 4f72518c0d70f5bd07a79feb6641b9d463216f86 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 8 Aug 2026 01:42:55 +0200 Subject: [PATCH 3/4] test(infra): reject reserved FTS probe terms --- tests/unit/infra/test_archive_canonical_snapshot.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit/infra/test_archive_canonical_snapshot.py b/tests/unit/infra/test_archive_canonical_snapshot.py index 030e561eab..ca736473d6 100644 --- a/tests/unit/infra/test_archive_canonical_snapshot.py +++ b/tests/unit/infra/test_archive_canonical_snapshot.py @@ -181,11 +181,12 @@ def test_default_fts_queries_follow_tokenizer_for_short_punctuation_terms() -> N connection.execute( "CREATE VIRTUAL TABLE messages_fts USING fts5(text, tokenize='unicode61 remove_diacritics 2')" ) - connection.execute("INSERT INTO messages_fts(text) VALUES (?)", ("a/b x-y",)) + connection.execute("INSERT INTO messages_fts(text) VALUES (?)", ("a/b x-y and",)) queries = _default_search_queries(connection) assert queries == ("a", "b", "x") + assert "and" not in queries assert all( connection.execute("SELECT 1 FROM messages_fts WHERE messages_fts MATCH ? LIMIT 1", (query,)).fetchone() is not None From 9108e75cbe68340f58b8784623eced175bf152af Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 8 Aug 2026 01:47:03 +0200 Subject: [PATCH 4/4] test(infra): cover absent FTS relation --- tests/unit/infra/test_archive_canonical_snapshot.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/unit/infra/test_archive_canonical_snapshot.py b/tests/unit/infra/test_archive_canonical_snapshot.py index ca736473d6..7dd0a4d7eb 100644 --- a/tests/unit/infra/test_archive_canonical_snapshot.py +++ b/tests/unit/infra/test_archive_canonical_snapshot.py @@ -196,6 +196,14 @@ def test_default_fts_queries_follow_tokenizer_for_short_punctuation_terms() -> N connection.close() +def test_default_fts_queries_are_empty_without_public_relation() -> None: + connection = sqlite3.connect(":memory:") + try: + assert _default_search_queries(connection) == () + finally: + connection.close() + + @pytest.mark.parametrize("table", ("assertions", "context_deliveries")) def test_user_state_mutations_are_red(tmp_path: Path, table: str) -> None: canonical = _build_archive(tmp_path / "canonical", rich_convergence_pathology())