diff --git a/tests/infra/archive_canonical_snapshot.py b/tests/infra/archive_canonical_snapshot.py index 00393d6ae3..50dba4c651 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,91 @@ 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, tokenizer-compatible probes from the public FTS table.""" + + 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, ...]: 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..7dd0a4d7eb 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, ) @@ -85,6 +86,195 @@ 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 + + +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 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 + for query in queries + ) + finally: + 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()) + 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 +282,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)