"""Real SQLite reader replay with synthetic rows/vectors and a raw SQL writer. Usage: python snapshot-probe.py /path/to/mempalace No model, Chroma service, MCP server, or application ingest is run. """ import hashlib import importlib import os from pathlib import Path import platform import sqlite3 import sys import tempfile import types import numpy as np PIN = "4f7745afbd9d0197c6475b4c7ac846d4932eaeda" root = Path(sys.argv[1]).resolve() hashes = { "mempalace/write_routing.py": "8ab8717f160cd49890034dc8b8879cb2892a555a8e9b4b5e4cfc69ac644db705", "mempalace/_once.py": "bdfdb15206fcfc2fcc142b12823a70c3bcdf7addfe08a5d33f6332021a8be329", "mempalace/_env.py": "6226fa8a0d279c74e7a12dcbc3f1679f1a2234eb7b6aa707890f4df943bfd611", "mempalace/paths.py": "279b13a5de59af1f8cc20657d47e4003d6fb0c62a2bd64d7d7f2fc1bcf375801", "mempalace/fsio.py": "509b5a23278c1c91d9970a611139668d238095559535318b8ddd6a4075e561f8", "mempalace/config.py": "865bb5ff19ce2509cbf9c9d5509518bde70374be246f94b61348bdd705dcea14", "mempalace/timeutil.py": "e799237ee39c0b461655dcabd99a384e20e4a207f24885bfc40dddf55db50b48", "mempalace/lexical.py": "e3ba2cda925fe8167d547d7966ff84d51bf417e2d15390d9abd87287c6b81fb0", "mempalace/metadata_types.py": "7348b76e11574af0e55d3ae32c112898703e2637d9a2e2fdb1ff7a77968b444b", "mempalace/stored_text.py": "24438375609d82f20b2081ce559eae8a4dc2ca1602aa3657379a0532f49b043a", "mempalace/backends/_magic.py": "198264607f7ed6b9f83594c2241b0f81341bd49484b79d3e3bad6547d437b656", "mempalace/backends/_filters.py": "900fa2f77df64afd57698ad95adcf6cdbf7388d21045286ba90a071301af3536", "mempalace/backends/base.py": "f0f491b23eeae9a9a87e93e947b9557d0b6b255e0c3a8550e971e0a5f65a2d0e", "mempalace/backends/sqlite_exact.py": "f40a4ae3cb29235860e4091635c8fb3501c1cad150f24b3c36a6c12a7eb0f6c4", } for name, digest in hashes.items(): assert hashlib.sha256((root / name).read_bytes()).hexdigest() == digest, name # Bypass package initializers which load other storage integrations. All # imported implementation/dependency files are the original pinned files. for name in ["mempalace", "mempalace.backends"]: p = types.ModuleType(name) p.__path__ = [str(root / name.replace(".", "/"))] sys.modules[name] = p m = importlib.import_module("mempalace.backends.sqlite_exact") print(f"Python {platform.python_version()}; SQLite {sqlite3.sqlite_version}; NumPy {np.__version__}; source {PIN}") cases = 0 def passed(text): global cases cases += 1 print(text) def query(col): return col.query(query_embeddings=[[0.0, 1.0]], n_results=1, include=["documents", "distances"]) def insert(conn, doc_id, text, vector): conn.execute( "INSERT INTO documents(collection_id,id,document,metadata_json,embedding,dim,created_at,updated_at) " "VALUES (1,?,?,?, ?,2,?,?)", (doc_id, text, "{}", np.array(vector, dtype=np.float32).tobytes(), "2026-01-01", "2026-01-01"), ) with tempfile.TemporaryDirectory(prefix="mp-reader-replay-") as d: db = str(Path(d) / "sqlite_exact.sqlite3") seed = sqlite3.connect(db) m.SQLiteExactBackend()._init_schema(seed) seed.execute("INSERT INTO collections(id,name,dimension,created_at) VALUES (1,'test',2,'2026-01-01')") insert(seed, "a", "alpha", [1, 0]) m._bump_write_generation(seed) seed.commit() seed.close() assert not Path(db + "-wal").exists() backend = m.SQLiteExactBackend() writer = None try: col = backend.get_collection(d, "test", options={"read_only": True}) original = col._handle assert original.immutable assert query(col).ids == [["a"]] cache = original._vector_cache[1] assert query(col).ids == [["a"]] and original._vector_cache[1] is cache passed("clean_database: immutable=True; top=a; warm matrix reused") # A fabricated empty, incomplete sidecar set; no writer owns it. Path(db + "-wal").touch() try: assert query(col).ids == [["a"]] assert col._handle is original and original._vector_cache[1] is cache cold = m.SQLiteExactBackend() try: try: cold.get_collection(d, "test", options={"read_only": True}) except m.BackendError as exc: assert "incomplete WAL sidecar set" in str(exc) else: raise AssertionError("cold reader accepted lone sidecar") finally: cold.close() finally: Path(db + "-wal").unlink(missing_ok=True) passed("empty_lone_wal: warm snapshot retained; cold reader rejected") writer = sqlite3.connect(db) insert(writer, "b", "bravo", [0, 1]) m._bump_write_generation(writer) writer.commit() assert Path(db + "-wal").stat().st_size > 0 assert Path(db + "-shm").exists() frozen_count = original.conn.execute("SELECT count(*) FROM documents").fetchone()[0] assert frozen_count == 1 passed(f"direct_old_immutable_connection: rows={frozen_count} after peer commit") result = query(col) assert result.ids == [["b"]] and result.distances == [[0.0]] assert original.retired and original.closed and not col._handle.immutable passed("public_query_after_wal_commit: old handle retired; immutable=False; top=b; distance=0.0") warm = col._handle._vector_cache[1] writer.execute("UPDATE documents SET embedding=? WHERE id='b'", (np.array([0, -1], dtype=np.float32).tobytes(),)) m._bump_write_generation(writer) writer.commit() assert query(col).ids == [["a"]] assert col._handle._vector_cache[1] is not warm passed("second_peer_commit: data-version invalidates matrix; top=a") writer.close() writer = None backend.close_palace(d) cleanup = sqlite3.connect(db) cleanup.execute("PRAGMA wal_checkpoint(TRUNCATE)") cleanup.close() col = backend.get_collection(d, "test", options={"read_only": True}) assert not Path(db + "-wal").exists() and col._handle.immutable query(col) passed("after_explicit_checkpoint_and_close: clean database reopens immutable=True") # Real SQL transaction/checkpoint; simulated unchanged stat metadata. before = backend._database_signature(db) real_stat = os.stat frozen_stat = real_stat(db) writer = sqlite3.connect(db) writer.execute("UPDATE documents SET document='omega' WHERE id='a'") m._bump_write_generation(writer) writer.commit() writer.close() writer = None assert not Path(db + "-wal").exists() def coarse_stat(path, *args, **kwargs): return frozen_stat if os.fspath(path) == db else real_stat(path, *args, **kwargs) try: os.stat = coarse_stat after = backend._database_signature(db) assert before[:5] == after[:5] assert before[-1][:-1] == after[-1][:-1] assert before[-1][-1] != after[-1][-1] assert query(col).documents == [["omega"]] finally: os.stat = real_stat passed("simulated_unchanged_stat: five header pragmas unchanged; generation changed; document=omega") try: col._handle.conn.execute("UPDATE documents SET document='forbidden'") except sqlite3.OperationalError: pass else: raise AssertionError("read-only connection allowed update") passed("read_only_connection: SQL update refused") backend.close_palace(d) try: query(col) except m.BackendClosedError: pass else: raise AssertionError("explicitly closed wrapper reconnected") passed("explicit_close: old collection remains closed") finally: if writer is not None: writer.close() backend.close() assert cases == 9 print("PASS: 9 reader scenarios; raw SQL fixture writes; no full ingest, embeddings model, or MCP execution")