diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b3b8e2ee..762659bc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -125,10 +125,22 @@ jobs: os: [ubuntu-latest, macos-latest, windows-latest] python-version: ["3.12"] mcp-version: [minimum, latest] + pydantic-version: [latest] include: - os: ubuntu-latest python-version: "3.10" mcp-version: minimum + pydantic-version: latest + # Exercise the MCP/Pydantic floor pair and the release that removed + # eval_type_backport. Other dependencies still use normal resolution. + - os: ubuntu-latest + python-version: "3.10" + mcp-version: minimum + pydantic-version: "2.11.0" + - os: ubuntu-latest + python-version: "3.12" + mcp-version: minimum + pydantic-version: "2.14.0" steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - name: Set up Python ${{ matrix.python-version }} @@ -149,6 +161,7 @@ jobs: shell: python env: MCP_VERSION: ${{ matrix.mcp-version }} + PYDANTIC_VERSION: ${{ matrix.pydantic-version }} run: | import os from pathlib import Path @@ -161,7 +174,9 @@ jobs: python = env_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python") requirements = [str(wheel.resolve())] if os.environ["MCP_VERSION"] == "minimum": - requirements.append("mcp[cli]==1.19.0") + requirements.append("mcp[cli]==1.21.1") + if os.environ["PYDANTIC_VERSION"] != "latest": + requirements.append("pydantic==" + os.environ["PYDANTIC_VERSION"]) subprocess.run([str(python), "-m", "pip", "install", *requirements], check=True) subprocess.run([str(python), "-m", "pip", "check"], check=True) subprocess.run([str(python), "-I", "tests/mcp-wheel-smoke.py"], check=True) diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 9fda4e0e..90bdc63c 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -37,6 +37,12 @@ jobs: fail-fast: false matrix: mcp-version: [minimum, latest] + pydantic-version: [latest] + include: + - mcp-version: minimum + pydantic-version: "2.11.0" + - mcp-version: minimum + pydantic-version: "2.14.0" steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 @@ -54,6 +60,7 @@ jobs: shell: python env: MCP_VERSION: ${{ matrix.mcp-version }} + PYDANTIC_VERSION: ${{ matrix.pydantic-version }} run: | import os from pathlib import Path @@ -66,7 +73,9 @@ jobs: python = env_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python") requirements = [str(wheel.resolve())] if os.environ["MCP_VERSION"] == "minimum": - requirements.append("mcp[cli]==1.19.0") + requirements.append("mcp[cli]==1.21.1") + if os.environ["PYDANTIC_VERSION"] != "latest": + requirements.append("pydantic==" + os.environ["PYDANTIC_VERSION"]) subprocess.run([str(python), "-m", "pip", "install", *requirements], check=True) subprocess.run([str(python), "-m", "pip", "check"], check=True) subprocess.run([str(python), "-I", "tests/mcp-wheel-smoke.py"], check=True) diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index e1eb661a..fc730218 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -677,3 +677,113 @@ The outer guard therefore adds no third raw-vector/source scan. It does not replace the builder's raw-input fence, certify arbitrary external vector writes, or make a borrowed caller's later COMMIT safe. This checkpoint adds no cluster adapter, engine routing or coverage migration. + +## Eight-layer adapters and borrowed publication + +`all_layer_specs(conn)` assembles clusters, monthly/entity-monthly summaries, +contradictions, structured facts, surprise, episodes, landmarks and Dunbar in +the existing consolidation order. Specs belong to that connection; using them +on another connection is rejected before any builder runs. The original +`nonvector_layer_specs()` API retains its six adapters and defaults. Preferences +remain a separate nonpersisted call. Profiles, styles, entity sheets and engine +scheduling are not enrolled by this checkpoint. + +Successful output and its checkpoint now include `successful_coverage` in the +same transaction. Allowed values are `complete`, `legacy_contacts_unowned`, +`vector_generation_unverified` and `unverified`. Historical rows migrate to +`unverified` without changing source epochs, counters, tracking triggers or +successful provenance. Failed/unavailable attempts retain the previous successful +coverage, just as they retain its output. Coverage describes that publication, +not a later arbitrary external edit of a derived table. Source freshness and +execution outcome remain separate; partial coverage does not disable retrieval +or cause an automatic repeat of otherwise unchanged work. + +Dunbar's primary selection uses the existing sender count/tie query inside the +runner's source snapshot. Counts describe verified generated relationships; +unowned contacts are preserved and reported as `legacy_contacts_unowned`. +`read_dunbar_coverage` remains the coherent live ownership inspection API. +Cluster counts exclude noise, so successful all-noise clustering has zero +centroids and `success_empty` while retaining its noise assignments. Missing +dependencies, missing vector tables and in-progress rebuilds are unavailable, +never successful empty output. A present rebuild descriptor must decode as an +object with format `version` exactly integer `1` and `complete` exactly boolean +`true`; malformed, unsupported or incomplete descriptors are unavailable. +Its SQLite `schema_version` is not the format version. A missing descriptor +retains compatibility with `vector_generation_unverified` coverage. This narrow +envelope check is not a full rebuild/source certificate. Runtime dependency import +failure also records an unavailable attempt against the captured source/dependency, +without a retry loop. Explicit force can retry after repair. + +Cluster scheduling uses a different identity from its publication fence. Its +stable key binds runtime model/dimension/tier, active table/schema, relevant +registry mappings and model fields, embed metadata values, build state and the +whole rebuild descriptor, plus dependency versions and unchanged HDBSCAN +parameters. Ordinary metadata timestamps and registry progress/count fields are +excluded: `insert_count - attempted_insert_count >= 25` must still mean 25 +committed appends, rather than one metadata rewrite. Full identity, including +those excluded values, is still checked by the publication guard. Initial empty +or historical input attempts once; corrections, dependency changes and explicit +force remain independent reasons to run. Failed initial attempts also retain +their durable 25-insert baseline. + +Temporary model-lock contention is a live `deferred` result with `ModelBusy`, +not a dependency-version change or a persisted unavailable attempt. Even forced +work cannot bypass that ownership boundary. An initial busy probe performs no +diagnostic write; planning omits it until another safe boundary can observe the +runtime. `layer_freshness` reports `dependency_deferred` during that interval. +If contention occurs after computation starts, the result has `attempted=True` +to report actual work, but the durable attempt counters and previous successful +coverage remain unchanged. No timer or automatic retry is added. + +An owned runner restores its entire prior checkpoint row, or its prior absence, +only after confirming output rollback. Restoration requires the same process, +ownership token, run generation and complete unchanged `running` diagnostic; +it cannot overwrite a replacement worker's state. Failure to roll back, loss of +ownership, or failed restoration raises instead of reporting clean deferral. +A cancellation racing with a busy result restores that baseline and stops the +remaining layers. Borrowed runs have no precompute diagnostic to restore and +roll back only their layer savepoint. Missing packages, invalid descriptors and +other declared dependency loss still produce durable unavailable attempts. +Only the explicit `ClusterModelBusyError` from the builder/publication boundary +is translated to deferral; it remains a `RuntimeError` subclass for existing +direct callers. + +The cheap cluster probe reads package metadata, existing runtime state and small +SQLite metadata records. It does not import native packages, load models, scan +source/vector rows or wait for the model lock. Execution requires the caller's +connection to have its vector extension loaded and the existing vector runtime +initialized. Actual clustering and guard imports occur only when execution is +attempted. Initial builder acquisition is nonblocking whenever the connection +already has a transaction, including the interval after a successful probe, so +a caller retaining SQLite's writer never waits for another model owner there. +Standalone builder and owned-guard initial capture can still wait for model +ownership before opening their SQLite transaction; the nonblocking probe does +not promise that all execution is wait-free. Tuple and `sqlite3.Row` factories produce the +same scheduling identity and rebuild-state checks. Whole rebuild descriptors +still do not certify every foreground +vector writer or arbitrary vector SQL; published cluster coverage therefore +remains `vector_generation_unverified` even when source freshness is current. + +`run_layers` continues to require a clean transaction by default. The explicit +`allow_caller_transaction=True` option uses protected per-layer savepoints when +the caller already has a transaction. It never commits unrelated caller writes, +including success/failure metadata. No `running` diagnostic is written before +borrowed computation. A caller that already holds a writer, or retains earlier +layer output, retains that writer until its own terminal transaction. The runner +does not promise short writer ownership for this caller-controlled mode. + +Borrowed results carry `pending_caller_commit=True`; other connections cannot see +their output/checkpoint until the caller commits, and caller rollback removes +both. Failed RELEASE rolls back the layer before recording its failed attempt. +If rollback itself is rejected, execution raises immediately without further +diagnostic writes or cleanup release; the caller must roll back that transaction. +Cancellation remains cooperative and preserves caller ownership. + +Cluster borrowing has an explicit separate guard: initial model acquisition is +nonblocking, full identity is rechecked under writer/model ownership, and model +ownership lasts through the runner's savepoint RELEASE or rollback. The lock is +released before returning. This does not certify the runtime model at the +caller's later outer COMMIT. Other adapters with an owned publication guard must +supply an explicit borrowed guard or report unavailable. This checkpoint does +not route engines or ingest, change algorithms or establish a measured latency, +memory or throughput improvement. diff --git a/pyproject.toml b/pyproject.toml index 5014d6f1..f9be3037 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,8 +29,9 @@ dependencies = [ "model2vec>=0.6.0,<1.0", "sqlite-vec>=0.1.0,<1.0", "numpy>=1.24.0,<3.0", - # Tool metadata requires 1.19; SDK 2 removes the FastMCP import path. - "mcp[cli]>=1.19.0,<2.0", + # 1.21.1 removes a private Pydantic import broken by Pydantic 2.14. + # SDK 2 removes the FastMCP import path. + "mcp[cli]>=1.21.1,<2.0", "httpx>=0.27.0,<1.0", "sentence-transformers>=3.0.0,<7.0", "torch>=2.0.0,<3.0", diff --git a/tests/mcp-wheel-smoke.py b/tests/mcp-wheel-smoke.py index 06388fe9..6a0f2522 100644 --- a/tests/mcp-wheel-smoke.py +++ b/tests/mcp-wheel-smoke.py @@ -182,6 +182,16 @@ async def _check_protocol(root: Path, env: dict[str, str], origin: Path) -> None def main() -> None: _require(bool(sys.flags.isolated), "Run this probe with python -I") distribution = importlib.metadata.distribution("truememory") + # Report public package versions before subprocess checks, whose output is + # intentionally redacted. Dependency import failures then remain traceable. + print(json.dumps({ + "stage": "dependency_versions", + "python_version": ".".join(str(part) for part in sys.version_info[:3]), + "truememory_version": distribution.version, + "mcp_version": importlib.metadata.version("mcp"), + "pydantic_version": importlib.metadata.version("pydantic"), + "pydantic_core_version": importlib.metadata.version("pydantic-core"), + }), flush=True) origin = Path(distribution.locate_file("truememory/__init__.py")).resolve() spec = importlib.util.find_spec("truememory") _require(origin.is_file() and spec is not None and spec.origin is not None, diff --git a/tests/test-maintenance-adapters-753.py b/tests/test-maintenance-adapters-753.py new file mode 100644 index 00000000..a0454f79 --- /dev/null +++ b/tests/test-maintenance-adapters-753.py @@ -0,0 +1,953 @@ +"""Eight real maintenance adapters with SQLite and synthetic native boundaries.""" + +import ast +import importlib +import runpy +import sqlite3 +import struct +import sys +import tempfile +import threading +import types +import unittest +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +TESTS = Path(__file__).resolve().parent +PRIMITIVES = runpy.run_path(str(TESTS / "test-maintenance-source-revision-753.py")) +STORAGE, MAINTENANCE = PRIMITIVES["STORAGE"], PRIMITIVES["MAINTENANCE"] +CLUSTERS = runpy.run_path(str(TESTS / "test-cluster-publication-749.py")) + + +def source_modules() -> dict[str, types.ModuleType]: + modules = {} + for name in ("consolidation", "predictive", "temporal", "personality"): + module = types.ModuleType("synthetic_adapter_" + name) + path = ROOT / "truememory" / (name + ".py") + tree = ast.parse(path.read_text()) + tree.body = [node for node in tree.body if not ( + isinstance(node, ast.ImportFrom) and node.module in {"truememory.storage", "truememory.fts_search"} + )] + module._initialize_dunbar_ownership = STORAGE._initialize_dunbar_ownership + module._dunbar_ownership_ready = STORAGE._dunbar_ownership_ready + exec(compile(tree, str(path), "exec"), module.__dict__) + modules["truememory." + name] = module + return modules + + +class AdapterFixture(unittest.TestCase): + def setUp(self) -> None: + directory = tempfile.TemporaryDirectory(prefix="synthetic-adapters-") + self.addCleanup(directory.cleanup) + self.path = Path(directory.name) / "synthetic.sqlite" + self.conn = self.open_db() + self.modules = source_modules() + self.cluster, self.vector, self.native = CLUSTERS["load_clustering"]() + self.modules["truememory.clustering"] = self.cluster + self.native.fit = lambda: setattr(self.native, "labels", [-1] * len(self.native.inputs[-1])) + self.conn.execute("CREATE TABLE vec_messages_edge(embedding BLOB)") + self.conn.execute( + "INSERT INTO vector_cache_registry VALUES " + "('edge','vec_messages_edge','vec_messages_sep_edge',0,0,'synthetic-model-a',2,0,0)" + ) + self.conn.executemany("INSERT INTO metadata(key,value,updated_at) VALUES (?,?,'synthetic-0')", [ + ("embed_model", "synthetic-model-a"), ("embed_dim", "2"), + ("vec_source_v1:vec_messages_edge", '{"version":1,"generation":"synthetic-a","cursor":0,"complete":true}'), + ]) + self.conn.commit() + self.enterContext(patch.dict(sys.modules, {"truememory.vector_search": self.vector})) + self.enterContext(patch.object(MAINTENANCE, "importlib", types.SimpleNamespace( + import_module=lambda name: self.modules[name], metadata=importlib.metadata, util=importlib.util, + ))) + self.versions = {"numpy": "synthetic-1", "hdbscan": "synthetic-1", "sqlite_vec": "synthetic-1"} + self.enterContext(patch.object(MAINTENANCE, "_installed_dependency", lambda name, dist: self.versions[name])) + + # TestCase.enterContext was added after the oldest supported Python. + def enterContext(self, context): + result = context.__enter__() + self.addCleanup(context.__exit__, None, None, None) + return result + + def open_db(self) -> sqlite3.Connection: + conn = STORAGE.create_db(self.path) + conn.execute("PRAGMA busy_timeout=100") + self.addCleanup(conn.close) + return conn + + def add(self, count: int = 1, *, commit: bool = True) -> None: + for _ in range(count): + cursor = self.conn.execute( + "INSERT INTO messages(content,sender,recipient,timestamp,category) VALUES " + "('synthetic team launched Cedar','synthetic-primary','synthetic-contact','2026-01-01','synthetic-session')" + ) + self.conn.execute("INSERT INTO vec_messages_edge(rowid,embedding) VALUES (?,?)", (cursor.lastrowid, struct.pack("2f", 1, 2))) + self.conn.execute("UPDATE metadata SET updated_at=? WHERE key IN ('embed_model','embed_dim')", (str(cursor.lastrowid),)) + self.conn.execute("UPDATE vector_cache_registry SET last_embedded_id=?,vector_count=?,last_updated=?", + (cursor.lastrowid, cursor.lastrowid, cursor.lastrowid)) + if commit: + self.conn.commit() + + def specs(self, conn: sqlite3.Connection | None = None): + return MAINTENANCE.all_layer_specs(conn or self.conn) + + def spec(self, layer: str = "clusters"): + return next(spec for spec in self.specs() if spec.layer == layer) + + def run_layer(self, layer: str = "clusters", **kwargs): + return MAINTENANCE.run_layers(self.conn, (self.spec(layer),), **kwargs)[0] + + def state(self, layer: str = "clusters", conn: sqlite3.Connection | None = None): + conn = conn or self.conn + spec = next(spec for spec in self.specs(conn) if spec.layer == layer) + return MAINTENANCE.read_layer_states(conn, (spec,))[layer] + + def output(self, conn: sqlite3.Connection | None = None) -> tuple: + conn = conn or self.conn + return (conn.execute("SELECT * FROM message_clusters ORDER BY message_id").fetchall(), + conn.execute("SELECT * FROM cluster_centroids ORDER BY cluster_id").fetchall()) + + +class TestEightAdapters(AdapterFixture): + def test_all_eight_actual_builders_and_original_six_contract(self) -> None: + self.add(5) + self.native.fit = lambda: setattr(self.native, "labels", [0, 0, -1, 1, 1]) + with patch.object(self.modules["truememory.personality"], "build_entity_profiles", side_effect=AssertionError("not enrolled")), \ + patch.object(self.modules["truememory.personality"], "extract_preferences", side_effect=AssertionError("separate nonpersisted call")): + results = MAINTENANCE.run_layers(self.conn, self.specs()) + self.assertEqual([item.result_key for item in results], [ + "cluster_messages", "build_summaries", "detect_contradictions", "structured_facts", + "build_surprise_index", "detect_episodes", "detect_landmarks", "dunbar_hierarchy", + ]) + self.assertTrue(all(item.outcome in {"success", "success_empty"} for item in results), results) + self.assertEqual(results[0].output_count, 2) + self.assertEqual(results[-1].output_count, 1) + self.assertEqual(results[0].coverage, "vector_generation_unverified") + self.assertTrue(all(item.coverage == "complete" for item in results[1:])) + self.assertEqual(len(MAINTENANCE.nonvector_layer_specs()), 6) + self.assertTrue(all(spec.publication_guard is None for spec in MAINTENANCE.nonvector_layer_specs())) + self.assertFalse(self.conn.in_transaction) + + def test_empty_all_layers_checkpoint_once_across_reopen(self) -> None: + results = MAINTENANCE.run_layers(self.conn, self.specs()) + self.assertTrue(all(item.outcome == "success_empty" for item in results), results) + reopened = self.open_db() + self.assertEqual(MAINTENANCE.plan_layers(reopened, self.specs(reopened)), ()) + self.assertTrue(all(not item.attempted for item in MAINTENANCE.run_layers(reopened, self.specs(reopened)))) + + def test_all_noise_retains_assignments_as_success_empty(self) -> None: + self.add(5) + result = self.run_layer() + self.assertEqual((result.outcome, result.output_count, result.coverage), + ("success_empty", 0, "vector_generation_unverified")) + self.assertEqual(len(self.output()[0]), 5) + self.assertTrue(all(row[1:] == (-1, 1) for row in self.output()[0])) + + def test_missing_hdbscan_and_runtime_import_failure_are_unavailable_with_siblings(self) -> None: + self.add(1) + self.versions["hdbscan"] = None + results = MAINTENANCE.run_layers(self.conn, self.specs()) + self.assertEqual(results[0].outcome, "unavailable") + self.assertTrue(all(item.outcome in {"success", "success_empty"} for item in results[1:])) + self.versions["hdbscan"] = "synthetic-1" + self.native.missing = True + result = self.run_layer() + self.assertEqual((result.outcome, result.error_category), ("unavailable", "LayerUnavailableError")) + self.assertIsNone(self.state().successful_epoch) + self.assertFalse(self.run_layer().attempted) + + def test_missing_vector_table_and_active_rebuild_do_not_certify_empty(self) -> None: + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('vec_build_state:vec_messages_edge','in_progress')") + self.conn.commit() + self.assertEqual(self.run_layer().outcome, "unavailable") + self.assertIsNone(self.state().successful_epoch) + self.conn.execute("UPDATE metadata SET value='complete' WHERE key='vec_build_state:vec_messages_edge'") + self.conn.execute("ALTER TABLE vec_messages_edge RENAME TO synthetic_old_vectors") + self.conn.commit() + result = self.run_layer() + self.assertEqual((result.outcome, result.error_category), ("unavailable", "VectorIndexMissing")) + + def test_manifest_absence_is_unverified_and_invalid_or_incomplete_is_unavailable(self) -> None: + self.conn.execute("DELETE FROM metadata WHERE key='vec_source_v1:vec_messages_edge'") + self.conn.commit() + self.assertEqual(self.run_layer().coverage, "vector_generation_unverified") + for descriptor in (b'synthetic bytes', 'synthetic invalid JSON', 'null', '[]', '{}', '{"version":2,"complete":true}', + '{"version":true,"complete":true}', '{"version":1,"complete":1}', + '{"version":1,"complete":"true"}', '{"version":1,"complete":false}'): + with self.subTest(descriptor=descriptor): + self.conn.execute("INSERT OR REPLACE INTO metadata(key,value) VALUES ('vec_source_v1:vec_messages_edge',?)", + (descriptor,)) + self.conn.commit() + result = self.run_layer() + self.assertEqual(result.outcome, "unavailable") + self.assertEqual(result.coverage, "vector_generation_unverified") + + def test_dunbar_primary_ties_legacy_contacts_and_empty_retirement(self) -> None: + self.add(3) + self.conn.execute( + "INSERT INTO entity_relationships(entity_a,entity_b,relationship_type) " + "VALUES ('synthetic-primary','synthetic-contact','contact')" + ) + self.conn.commit() + result = self.run_layer("dunbar") + self.assertEqual((result.output_count, result.coverage), (1, "legacy_contacts_unowned")) + self.assertEqual(self.conn.execute("SELECT count(*) FROM entity_relationships").fetchone()[0], 2) + self.conn.execute("DELETE FROM messages") + self.conn.commit() + result = self.run_layer("dunbar") + self.assertEqual((result.outcome, result.output_count, result.coverage), ("success_empty", 0, "legacy_contacts_unowned")) + self.assertEqual(self.conn.execute("SELECT count(*) FROM entity_relationships").fetchone()[0], 1) + + def test_actual_primary_selection_matches_existing_tied_sender_query(self) -> None: + self.conn.executemany("INSERT INTO messages(content,sender,recipient) VALUES ('synthetic',?,?)", [ + ("synthetic-beta", "synthetic-contact"), ("synthetic-alpha", "synthetic-other"), + ]) + self.conn.commit() + expected = self.conn.execute( + "SELECT sender,COUNT(*) AS cnt FROM messages WHERE sender != '' AND sender IS NOT NULL " + "GROUP BY sender ORDER BY cnt DESC LIMIT 1" + ).fetchone()[0].lower() + self.run_layer("dunbar") + self.assertEqual(self.conn.execute("SELECT DISTINCT entity_a FROM entity_relationships").fetchall(), [(expected,)]) + + +class TestAdapterScheduling(AdapterFixture): + def test_24_actual_appends_with_metadata_progress_rewrites_wait_for_25(self) -> None: + self.run_layer() + original_key = self.spec().resolve_dependency().key + self.add(24) + self.assertEqual(self.spec().resolve_dependency().key, original_key) + self.assertFalse(self.run_layer().attempted) + self.add(1) + self.assertTrue(self.run_layer().attempted) + self.assertEqual(self.state().attempted_insert_count, 25) + + def test_failed_and_unavailable_initial_cluster_attempts_use_25_insert_baseline(self) -> None: + for unavailable in (False, True): + with self.subTest(unavailable=unavailable): + self.conn.execute("DELETE FROM maintenance_layers WHERE layer='clusters'") + self.conn.commit() + self.versions["hdbscan"] = None if unavailable else "synthetic-1" + with patch.object(self.cluster, "cluster_messages", side_effect=ValueError("synthetic failure")): + result = self.run_layer() + self.assertEqual(result.outcome, "unavailable" if unavailable else "failed") + self.add(1) + self.assertFalse(self.run_layer().attempted) + self.add(23) + self.assertFalse(self.run_layer().attempted) + self.add(1) + self.assertTrue(self.run_layer().attempted) + + def test_correction_replace_dependency_and_force_wake_below_25(self) -> None: + self.add(1) + self.run_layer() + for sql in ("UPDATE messages SET content='synthetic changed' WHERE id=1", + "INSERT OR REPLACE INTO messages(id,content) VALUES (1,'synthetic replacement')"): + self.conn.execute(sql) + self.conn.commit() + self.assertTrue(self.run_layer().attempted) + self.versions["hdbscan"] = "synthetic-2" + self.assertTrue(self.run_layer().attempted) + self.assertTrue(self.run_layer(force=True).attempted) + self.assertFalse(self.run_layer().attempted) + + def test_manifest_progress_and_same_dimensional_model_identity_change_key(self) -> None: + previous = self.spec().resolve_dependency().key + self.conn.execute("UPDATE metadata SET value=? WHERE key='vec_source_v1:vec_messages_edge'", + ('{"version":1,"generation":"synthetic-a","cursor":1,"complete":true}',)) + self.conn.commit() + current = self.spec().resolve_dependency().key + self.assertNotEqual(current, previous) + self.vector.EMBEDDING_MODEL = "synthetic-model-c" + self.assertNotEqual(self.spec().resolve_dependency().key, current) + + def test_foreground_eligibility_never_imports_native_modules_or_waits_for_model(self) -> None: + original = MAINTENANCE.importlib.import_module + imported = [] + + def safe_import(name): + imported.append(name) + self.assertNotIn(name, {"numpy", "hdbscan", "sqlite_vec", "truememory.clustering", "truememory.vector_search"}) + return original(name) + + with patch.object(MAINTENANCE.importlib, "import_module", safe_import): + self.vector._lock.acquire() + try: + self.assertEqual(self.spec().resolve_dependency().error_category, "ModelBusy") + planned = MAINTENANCE.plan_layers(self.conn, self.specs()) + self.assertEqual(len(planned), 7) + self.assertNotIn("clusters", {spec.layer for spec in planned}) + finally: + self.vector._lock.release() + self.assertNotIn("truememory.clustering", imported) + self.assertFalse(self.conn.in_transaction) + + def test_adapters_reject_wrong_connection_before_running_or_recording(self) -> None: + other = self.open_db() + with self.assertRaisesRegex(ValueError, "another connection"): + MAINTENANCE.run_layers(other, self.specs()) + self.assertEqual(other.execute("SELECT DISTINCT outcome FROM maintenance_layers").fetchall(), [("pending",)]) + + +class TestTransientModelContention(AdapterFixture): + def row(self) -> tuple | None: + row = self.conn.execute("SELECT * FROM maintenance_layers WHERE layer='clusters'").fetchone() + return tuple(row) if row is not None else None + + def release_model(self) -> None: + if self.vector._lock.locked(): + self.vector._lock.release() + + def block_during_compute(self, cancel: threading.Event | None = None) -> None: + def fit(): + self.native.labels = [-1] * len(self.native.inputs[-1]) + self.vector._lock.acquire() + if cancel is not None: + cancel.set() + self.native.fit = fit + self.addCleanup(self.release_model) + + def test_initial_busy_and_force_never_write_or_fabricate_attempt(self) -> None: + self.add(5) + for absent in (False, True): + if absent: + self.conn.execute("DELETE FROM maintenance_layers WHERE layer='clusters'") + self.conn.commit() + old, changes = self.row(), self.conn.total_changes + self.vector._lock.acquire() + try: + for force in (False, True): + with self.subTest(absent=absent, force=force): + result = self.run_layer(force=force) + self.assertEqual((result.outcome, result.error_category, result.attempted), + ("deferred", "ModelBusy", False)) + self.assertEqual(self.row(), old) + self.assertEqual(self.conn.total_changes, changes) + self.assertEqual(MAINTENANCE.plan_layers(self.conn, (self.spec(),), force=force), ()) + self.assertEqual(MAINTENANCE.layer_freshness( + self.state().source, self.state(), self.spec().resolve_dependency()), "dependency_deferred") + finally: + self.vector._lock.release() + self.assertEqual(self.run_layer().outcome, "success_empty") + self.assertFalse(self.run_layer().attempted) + + def test_busy_after_one_append_preserves_twenty_five_append_baseline(self) -> None: + self.add(5) + self.run_layer() + previous = self.row() + self.add(1) + self.assertFalse(self.run_layer().attempted) + self.vector._lock.acquire() + try: + self.assertEqual(self.run_layer(force=True).outcome, "deferred") + finally: + self.vector._lock.release() + self.assertEqual(self.row(), previous) + self.assertFalse(self.run_layer().attempted) + self.add(23) + self.assertFalse(self.run_layer().attempted) + self.add(1) + self.assertTrue(self.run_layer().attempted) + + def test_mid_compute_busy_restores_every_prior_field_and_original_output(self) -> None: + self.add(5) + self.run_layer() + self.conn.execute("UPDATE maintenance_layers SET builder_version=7,outcome='failed'," + "error_category='SyntheticFailure',run_generation='synthetic-old' WHERE layer='clusters'") + self.conn.commit() + previous, output = self.row(), self.output() + self.add(1) + self.block_during_compute() + result = self.run_layer(force=True) + self.release_model() + self.assertEqual((result.outcome, result.attempted), ("deferred", True)) + self.assertEqual(self.row(), previous) + self.assertEqual(self.output(), output) + self.assertFalse(self.conn.in_transaction) + self.assertFalse(self.run_layer().attempted) + + def test_mid_compute_busy_restores_absent_row_only_after_output_rollback(self) -> None: + self.add(1) + self.conn.execute("DELETE FROM maintenance_layers WHERE layer='clusters'") + self.conn.commit() + self.block_during_compute() + original, observations = MAINTENANCE._restore_deferred_attempt, [] + + def restore(*args): + observations.append((self.conn.in_transaction, self.output())) + return original(*args) + + with patch.object(MAINTENANCE, "_restore_deferred_attempt", restore): + result = self.run_layer() + self.release_model() + self.assertEqual((result.outcome, result.attempted), ("deferred", True)) + self.assertEqual(observations, [(False, ([], []))]) + self.assertIsNone(self.row()) + self.native.fit = lambda: setattr(self.native, "labels", [-1]) + self.assertTrue(self.run_layer().attempted) + + def test_busy_at_postbuild_dependency_and_outer_guard_rolls_back_checkpoint(self) -> None: + self.add(1) + self.run_layer() + self.add(1) + old_row, old_output = self.row(), self.output() + for boundary in ("dependency", "guard"): + with self.subTest(boundary=boundary): + spec, after_build = self.spec(), [] + + def count(conn): + after_build.append(0) + return spec.count_output(conn) + + def resolve(): + if after_build: + after_build[0] += 1 + if boundary == "dependency" and after_build[0] == 1: + self.vector._lock.acquire() + dependency = spec.resolve_dependency() + if boundary == "guard" and after_build and after_build[0] == 2: + self.vector._lock.acquire() + return dependency + + try: + result = MAINTENANCE.run_layers(self.conn, (spec._replace( + count_output=count, resolve_dependency=resolve),), force=True)[0] + finally: + self.release_model() + self.assertEqual((result.outcome, result.attempted), ("deferred", True)) + self.assertEqual(self.row(), old_row) + self.assertEqual(self.output(), old_output) + self.assertFalse(self.run_layer().attempted) + + def test_plain_runtime_error_is_a_failed_attempt_not_a_busy_deferral(self) -> None: + self.add(1) + with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("Embedding model is busy")): + result = self.run_layer() + self.assertEqual((result.outcome, result.error_category, result.attempted), ("failed", "RuntimeError", True)) + self.assertEqual(self.row()[2], "failed") + + def test_cancellation_racing_busy_restores_baseline_and_stops_siblings(self) -> None: + self.add(1) + previous = self.row() + cancel = threading.Event() + self.block_during_compute(cancel) + results = MAINTENANCE.run_layers(self.conn, self.specs(), cancel=cancel) + self.release_model() + self.assertEqual(len(results), 1) + self.assertEqual((results[0].outcome, results[0].attempted), ("deferred", True)) + self.assertEqual(self.row(), previous) + self.assertEqual(self.output(), ([], [])) + + def test_changed_diagnostic_is_not_overwritten_even_with_same_run_generation(self) -> None: + self.add(1) + self.block_during_compute() + restore = MAINTENANCE._restore_deferred_attempt + writer = self.open_db() + + def replace_diagnostic(*args): + writer.execute("UPDATE maintenance_layers SET error_category='ExternalChange' WHERE layer='clusters'") + writer.commit() + return restore(*args) + + with patch.object(MAINTENANCE, "_restore_deferred_attempt", replace_diagnostic): + with self.assertRaisesRegex(MAINTENANCE.MaintenanceBusyError, "diagnostic changed"): + self.run_layer() + self.assertEqual(self.conn.execute("SELECT error_category FROM maintenance_layers WHERE layer='clusters'").fetchone()[0], + "ExternalChange") + self.assertEqual(self.output(), ([], [])) + self.assertFalse(self.conn.in_transaction) + + def test_changed_owner_or_generation_is_not_overwritten(self) -> None: + self.add(1) + self.block_during_compute() + restore = MAINTENANCE._restore_deferred_attempt + for kind in ("owner", "generation"): + with self.subTest(kind=kind): + def change_ownership(conn, layer, previous, running, owner): + if kind == "generation": + conn.execute("UPDATE maintenance_layers SET run_generation='synthetic-other' WHERE layer=?", (layer,)) + conn.commit() + return restore(conn, layer, previous, running, owner) + owners = MAINTENANCE._thread_owners.owners + owners[owner.path] = owner._replace(generation="synthetic-other") + try: + return restore(conn, layer, previous, running, owner) + finally: + owners[owner.path] = owner + + with patch.object(MAINTENANCE, "_restore_deferred_attempt", change_ownership): + with self.assertRaises(MAINTENANCE.MaintenanceBusyError): + self.run_layer(force=True) + self.release_model() + self.assertEqual(self.row()[2], "running") + if kind == "generation": + self.assertEqual(self.conn.execute("SELECT run_generation FROM maintenance_layers WHERE layer='clusters'").fetchone()[0], + "synthetic-other") + self.assertEqual(self.output(), ([], [])) + + def test_failed_restoration_commit_propagates_and_leaves_running_diagnostic(self) -> None: + self.add(1) + self.block_during_compute() + restore, rejected = MAINTENANCE._restore_deferred_attempt, [] + + def fail_commit(*args): + def authorize(action, first, *_args): + if action == sqlite3.SQLITE_TRANSACTION and first == "COMMIT" and not rejected: + rejected.append(True) + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + self.conn.set_authorizer(authorize) + return restore(*args) + + try: + with patch.object(MAINTENANCE, "_restore_deferred_attempt", fail_commit): + with self.assertRaises(sqlite3.DatabaseError): + self.run_layer() + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual(rejected, [True]) + self.assertEqual(self.row()[2], "running") + self.assertFalse(self.conn.in_transaction) + self.assertEqual(self.output(), ([], [])) + + def test_output_rollback_failure_never_restores_diagnostic(self) -> None: + self.add(1) + restore = MAINTENANCE._restore_deferred_attempt + + def fail(conn): + conn.execute("INSERT INTO message_clusters(message_id,cluster_id,noise) VALUES (1,-1,1)") + conn.set_authorizer(lambda action, first, *_args: sqlite3.SQLITE_DENY + if action == sqlite3.SQLITE_TRANSACTION and first == "ROLLBACK" else sqlite3.SQLITE_OK) + raise self.cluster.ClusterModelBusyError("synthetic contention") + + try: + with patch.object(self.cluster, "cluster_messages", fail), \ + patch.object(MAINTENANCE, "_restore_deferred_attempt", wraps=restore) as restoration: + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + self.run_layer() + restoration.assert_not_called() + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.conn.rollback() + self.assertEqual(self.row()[2], "running") + self.assertEqual(self.output(), ([], [])) + + def test_borrowed_busy_rolls_back_only_layer_without_attempt_metadata(self) -> None: + self.add(1, commit=False) + previous = self.row() + self.block_during_compute() + result = self.run_layer(allow_caller_transaction=True) + self.release_model() + self.assertEqual((result.outcome, result.attempted, result.pending_caller_commit), ("deferred", True, True)) + self.assertEqual(self.row(), previous) + self.assertEqual(self.output(), ([], [])) + self.assertTrue(self.conn.in_transaction) + self.conn.commit() + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertEqual(self.output(), ([], [])) + + def test_model_owner_arriving_after_probe_never_blocks_inside_transaction(self) -> None: + self.add(1) + self.run_layer() + self.add(1) + previous, output = self.row(), self.output() + for borrowed in (False, True): + with self.subTest(borrowed=borrowed): + if borrowed: + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-caller','retained')") + spec, attempts = self.spec(), [] + held = threading.Lock() + + class BoundaryLock: + armed = False + + def acquire(lock, blocking=True): + if lock.armed: + attempts.append(blocking) + if blocking: + raise AssertionError("Blocking model acquisition while SQLite transaction is retained") + return held.acquire(blocking=blocking) + + def release(lock): + held.release() + + def __enter__(lock): + lock.acquire() + return lock + + def __exit__(lock, *_args): + lock.release() + + boundary = BoundaryLock() + + def build(conn): + self.assertTrue(conn.in_transaction) + held.acquire() + boundary.armed = True + return spec.build(conn) + + with patch.object(self.vector, "_lock", boundary): + try: + result = MAINTENANCE.run_layers(self.conn, (spec._replace(build=build),), + force=True, allow_caller_transaction=borrowed)[0] + self.assertTrue(held.locked(), "The other model owner's lock must not be released") + finally: + held.release() + self.assertEqual((result.outcome, result.attempted), ("deferred", True)) + self.assertEqual(attempts, [False]) + self.assertEqual(self.row(), previous) + self.assertEqual(self.output(), output) + self.assertEqual(self.conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(self.conn.execute("SELECT value FROM metadata WHERE key='synthetic-caller'").fetchone()[0], "retained") + self.conn.rollback() + self.assertIsNone(self.conn.execute("SELECT value FROM metadata WHERE key='synthetic-caller'").fetchone()) + + def test_standalone_initial_capture_wait_policy_stays_outside_sqlite_transaction(self) -> None: + self.add(1) + attempts = [] + held = threading.Lock() + conn = self.conn + + class RecordingLock: + def acquire(lock, blocking=True): + attempts.append((blocking, conn.in_transaction)) + return held.acquire(blocking=blocking) + + def release(lock): + held.release() + + def __enter__(lock): + lock.acquire() + return lock + + def __exit__(lock, *_args): + lock.release() + + with patch.object(self.vector, "_lock", RecordingLock()): + self.assertEqual(self.cluster.cluster_messages(self.conn), 0) + self.assertEqual(attempts, [(True, False), (False, True)]) + + +class TestRowFactoryCompatibility(AdapterFixture): + def test_row_factory_supports_real_cluster_builder_and_all_adapters(self) -> None: + self.add(5) + original_key = self.spec().resolve_dependency().key + self.conn.row_factory = sqlite3.Row + self.assertEqual(self.cluster.cluster_messages(self.conn), 0) + dependency = self.spec().resolve_dependency() + self.assertTrue(dependency.available, dependency) + self.assertEqual(dependency.key, original_key) + results = MAINTENANCE.run_layers(self.conn, self.specs()) + self.assertTrue(all(item.outcome in {"success", "success_empty"} for item in results), results) + self.assertFalse(self.run_layer().attempted) + self.assertTrue(all(not state.dependency.deferred for state in MAINTENANCE.read_layer_states(self.conn, self.specs()).values())) + + def test_row_factory_detects_explicit_rebuild_in_progress(self) -> None: + self.conn.row_factory = sqlite3.Row + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('vec_build_state:vec_messages_edge','in_progress')") + self.conn.commit() + dependency = self.spec().resolve_dependency() + self.assertEqual((dependency.available, dependency.error_category, dependency.deferred), + (False, "VectorRebuildInProgress", False)) + self.assertEqual(self.run_layer().outcome, "unavailable") + + +class TestCoveragePublication(AdapterFixture): + def test_failure_preserves_success_coverage_and_previous_output(self) -> None: + self.add(1) + self.run_layer() + old_output, old_state = self.output(), self.state() + with patch.object(self.cluster, "cluster_messages", side_effect=ValueError("synthetic failure")): + result = self.run_layer(force=True) + self.assertEqual(result.coverage, "vector_generation_unverified") + self.assertEqual(self.output(), old_output) + self.assertEqual(self.state().successful_coverage, old_state.successful_coverage) + self.assertEqual(self.state().successful_dependency, old_state.successful_dependency) + + def test_output_coverage_checkpoint_visible_together_at_real_commit(self) -> None: + self.add(1) + reader = self.open_db() + observations = [] + + def trace(sql): + if sql == "COMMIT" and self.vector._lock.locked(): + observations.append((self.output(reader), self.state(conn=reader).successful_coverage)) + + spec = self.spec() + original = spec.count_output + + def count(conn): + conn.set_trace_callback(trace) + return original(conn) + + result = MAINTENANCE.run_layers(self.conn, (spec._replace(count_output=count),))[0] + self.conn.set_trace_callback(None) + self.assertEqual(result.outcome, "success_empty") + self.assertEqual(observations, [(([], []), "unverified")]) + self.assertEqual(self.state(conn=reader).successful_coverage, "vector_generation_unverified") + self.assertEqual(len(self.output(reader)[0]), 1) + + def test_coverage_failure_rolls_back_generated_output_and_old_coverage(self) -> None: + self.add(1) + self.run_layer() + previous = self.output(), self.state().successful_coverage + spec = self.spec()._replace(read_coverage=lambda conn: "invalid") + result = MAINTENANCE.run_layers(self.conn, (spec,), force=True)[0] + self.assertEqual(result.outcome, "failed") + self.assertEqual((self.output(), self.state().successful_coverage), previous) + + def test_metadata_timestamp_change_does_not_schedule_but_rejects_inflight_publication(self) -> None: + self.add(1) + spec = self.spec() + original = spec.count_output + + def changed(conn): + result = original(conn) + conn.execute("UPDATE metadata SET updated_at='synthetic-new' WHERE key='embed_model'") + return result + + result = MAINTENANCE.run_layers(self.conn, (spec._replace(count_output=changed),))[0] + self.assertEqual(result.outcome, "failed") + self.assertEqual(self.output(), ([], [])) + self.assertEqual(self.state().successful_coverage, "unverified") + + def test_coverage_migration_preserves_epoch_and_provenance_and_is_idempotent(self) -> None: + self.add(1) + self.run_layer() + token, old = MAINTENANCE.read_source_revision(self.conn), self.state() + self.conn.execute("ALTER TABLE maintenance_layers RENAME TO synthetic_old_layers") + fields = [row[1] for row in self.conn.execute("PRAGMA table_info(synthetic_old_layers)") if row[1] != "successful_coverage"] + self.conn.execute("CREATE TABLE maintenance_layers AS SELECT " + ",".join(fields) + " FROM synthetic_old_layers") + self.conn.execute("CREATE UNIQUE INDEX synthetic_layer_name ON maintenance_layers(layer)") + self.conn.commit() + STORAGE._initialize_maintenance_tracking(self.conn) + self.assertEqual(MAINTENANCE.read_source_revision(self.conn), token) + self.assertEqual(self.state().successful_dependency, old.successful_dependency) + self.assertEqual(self.state().successful_coverage, "unverified") + version, changes = self.conn.execute("PRAGMA schema_version").fetchone()[0], self.conn.total_changes + STORAGE._initialize_maintenance_tracking(self.conn) + self.assertEqual((self.conn.execute("PRAGMA schema_version").fetchone()[0], self.conn.total_changes), (version, changes)) + + def test_owned_commit_failure_preserves_old_output_and_coverage(self) -> None: + self.add(1) + self.run_layer() + previous = self.output(), self.state().successful_coverage + self.native.fit = lambda: setattr(self.native, "labels", [0]) + denied = [] + + def authorize(action, first, *_args): + if action == sqlite3.SQLITE_TRANSACTION and first == "COMMIT" and not denied: + denied.append(True) + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + spec = self.spec() + original = spec.count_output + + def count(conn): + conn.set_authorizer(authorize) + return original(conn) + + try: + result = MAINTENANCE.run_layers(self.conn, (spec._replace(count_output=count),), force=True)[0] + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual((result.outcome, denied), ("failed", [True])) + self.assertEqual((self.output(), self.state().successful_coverage), previous) + self.assertFalse(self.vector._lock.locked()) + + def test_coverage_read_snapshot_does_not_mix_concurrent_publications(self) -> None: + self.add(1) + self.run_layer("dunbar") + writer = self.open_db() + spec = self.spec("dunbar") + with MAINTENANCE.layer_read_snapshot(self.conn, "dunbar", spec.resolve_dependency()) as state: + writer.execute("INSERT INTO entity_relationships(entity_a,entity_b,relationship_type) " + "VALUES ('synthetic-other','synthetic-contact','contact')") + writer.commit() + writer_spec = next(item for item in self.specs(writer) if item.layer == "dunbar") + result = MAINTENANCE.run_layers(writer, (writer_spec,), force=True)[0] + self.assertEqual(result.coverage, "legacy_contacts_unowned") + self.assertEqual(state.successful_coverage, "complete") + self.assertEqual(self.modules["truememory.personality"].read_dunbar_coverage(self.conn)["status"], "managed_only") + self.assertEqual(self.state("dunbar").successful_coverage, "legacy_contacts_unowned") + + +class TestBorrowedPublication(AdapterFixture): + def test_coverage_migration_and_output_remain_inside_caller_transaction(self) -> None: + self.conn.execute("ALTER TABLE maintenance_layers RENAME TO synthetic_old_layers") + fields = [row[1] for row in self.conn.execute("PRAGMA table_info(synthetic_old_layers)") if row[1] != "successful_coverage"] + self.conn.execute("CREATE TABLE maintenance_layers AS SELECT " + ",".join(fields) + " FROM synthetic_old_layers") + self.conn.execute("CREATE UNIQUE INDEX synthetic_layer_name ON maintenance_layers(layer)") + self.conn.commit() + token = MAINTENANCE.read_source_revision(self.conn) + self.add(1, commit=False) + STORAGE._initialize_maintenance_tracking(self.conn) + result = self.run_layer(allow_caller_transaction=True) + self.assertTrue(result.pending_caller_commit) + self.conn.rollback() + self.assertNotIn("successful_coverage", {row[1] for row in self.conn.execute("PRAGMA table_info(maintenance_layers)")}) + self.assertEqual(MAINTENANCE.read_source_revision(self.conn), token) + self.assertEqual(self.output(), ([], [])) + + def test_default_rejects_caller_transaction_and_opt_in_keeps_all_work_rollbackable(self) -> None: + self.add(2, commit=False) + with self.assertRaisesRegex(RuntimeError, "clean transaction"): + MAINTENANCE.run_layers(self.conn, self.specs()) + reader = self.open_db() + results = MAINTENANCE.run_layers(self.conn, self.specs(), allow_caller_transaction=True) + self.assertTrue(all(item.pending_caller_commit for item in results)) + self.assertTrue(all(item.outcome in {"success", "success_empty"} for item in results), results) + self.assertTrue(self.conn.in_transaction) + self.assertEqual(reader.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.state(conn=reader).successful_coverage, "unverified") + self.conn.rollback() + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.output(), ([], [])) + self.assertEqual(self.state().successful_coverage, "unverified") + + def test_read_only_caller_does_not_take_writer_for_running_diagnostic(self) -> None: + self.add(1) + writer = self.open_db() + observed = [] + + def compute(): + self.native.labels = [-1] + writer.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-writer','yes')") + writer.commit() + observed.append(True) + + self.native.fit = compute + self.conn.execute("BEGIN") + # The writer deliberately makes the read snapshot stale. Publication + # and its failure diagnostic cannot upgrade that snapshot either. + with self.assertRaises(sqlite3.OperationalError): + self.run_layer(allow_caller_transaction=True) + self.assertEqual(observed, [True]) + self.assertEqual(self.output(), ([], [])) + self.conn.rollback() + + def test_failed_borrowed_layer_and_successful_sibling_preserve_unrelated_writes(self) -> None: + self.add(1, commit=False) + with patch.object(self.cluster, "cluster_messages", side_effect=ValueError("synthetic failure")): + results = MAINTENANCE.run_layers(self.conn, self.specs(), allow_caller_transaction=True) + self.assertEqual(results[0].outcome, "failed") + self.assertTrue(all(item.outcome in {"success", "success_empty"} for item in results[1:])) + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertTrue(self.conn.in_transaction) + self.conn.rollback() + self.assertEqual(self.state().outcome, "pending") + + def test_cluster_savepoint_release_holds_model_lock_but_return_releases_it(self) -> None: + self.add(1, commit=False) + releases = [] + self.conn.set_trace_callback(lambda sql: releases.append(self.vector._lock.locked()) + if sql.startswith("RELEASE truememory_layer_") else None) + result = self.run_layer(allow_caller_transaction=True) + self.conn.set_trace_callback(None) + self.assertEqual(releases, [True]) + self.assertTrue(result.pending_caller_commit) + self.assertEqual(result.coverage, "vector_generation_unverified") + self.assertFalse(self.vector._lock.locked()) + self.vector.EMBEDDING_MODEL = "synthetic-model-c" + self.conn.commit() + self.assertEqual(self.state().successful_coverage, "vector_generation_unverified") + + def test_successful_release_failure_rolls_back_output_before_caller_commit(self) -> None: + self.add(1, commit=False) + rejected = [] + + def authorize(action, first, second, *_args): + if action == sqlite3.SQLITE_SAVEPOINT and first == "RELEASE" and second.startswith("truememory_layer_") and not rejected: + rejected.append(True) + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + self.conn.set_authorizer(authorize) + try: + result = self.run_layer(allow_caller_transaction=True) + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual(result.outcome, "failed") + self.assertEqual(rejected, [True]) + self.conn.commit() + self.assertEqual(self.output(), ([], [])) + self.assertEqual(self.state().successful_coverage, "unverified") + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + + def test_rejected_rollback_never_continues_with_attempt_write_or_release(self) -> None: + self.add(1, commit=False) + seen = [] + + def authorize(action, first, second, *_args): + if action == sqlite3.SQLITE_SAVEPOINT: + seen.append(first) + if first == "ROLLBACK" and second.startswith("truememory_layer_"): + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + self.conn.set_authorizer(authorize) + try: + with patch.object(self.cluster, "cluster_messages", side_effect=ValueError("synthetic failure")): + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + self.run_layer(allow_caller_transaction=True) + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual(seen, ["BEGIN", "ROLLBACK"]) + self.assertFalse(self.vector._lock.locked()) + self.assertEqual(self.state().outcome, "pending") + self.conn.rollback() + + def test_cancellation_and_keyboard_interrupt_keep_caller_transaction(self) -> None: + self.add(1, commit=False) + cancel = threading.Event() + spec = self.spec() + original = spec.count_output + + def count(conn): + cancel.set() + return original(conn) + + result = MAINTENANCE.run_layers(self.conn, (spec._replace(count_output=count),), + allow_caller_transaction=True, cancel=cancel)[0] + self.assertEqual(result.outcome, "abandoned") + self.assertTrue(result.pending_caller_commit) + self.assertEqual(self.output(), ([], [])) + with patch.object(self.cluster, "cluster_messages", side_effect=KeyboardInterrupt()): + with self.assertRaises(KeyboardInterrupt): + self.run_layer(allow_caller_transaction=True, force=True) + self.assertTrue(self.conn.in_transaction) + self.assertFalse(self.vector._lock.locked()) + self.conn.rollback() + + def test_generic_guard_requires_explicit_borrowed_contract(self) -> None: + self.add(1, commit=False) + spec = self.spec()._replace(borrowed_publication_guard=None) + result = MAINTENANCE.run_layers(self.conn, (spec,), allow_caller_transaction=True)[0] + self.assertEqual(result.outcome, "unavailable") + self.assertEqual(self.output(), ([], [])) + self.conn.rollback() + + def test_borrowed_source_change_after_compute_rolls_back_layer_only(self) -> None: + self.add(1, commit=False) + spec = self.spec() + original = spec.count_output + + def count(conn): + result = original(conn) + conn.execute("UPDATE messages SET content='synthetic stale result' WHERE id=1") + return result + + result = MAINTENANCE.run_layers(self.conn, (spec._replace(count_output=count),), + allow_caller_transaction=True)[0] + self.assertEqual(result.outcome, "failed") + self.assertEqual(self.output(), ([], [])) + self.assertEqual(self.conn.execute("SELECT content FROM messages").fetchone()[0], "synthetic team launched Cedar") + self.assertEqual(self.state().successful_coverage, "unverified") + self.conn.rollback() + + +if __name__ == "__main__": + unittest.main() diff --git a/truememory/clustering.py b/truememory/clustering.py index d97a6d50..b998ce72 100644 --- a/truememory/clustering.py +++ b/truememory/clustering.py @@ -38,6 +38,10 @@ logger = logging.getLogger(__name__) +class ClusterModelBusyError(RuntimeError): + """Model ownership is temporarily unavailable; no input version changed.""" + + # --------------------------------------------------------------------------- # Schema # --------------------------------------------------------------------------- @@ -151,7 +155,7 @@ def hold() -> Iterator[Callable[[], None]]: conn.execute("UPDATE messages SET id=id WHERE 0") acquired = vector_search._lock.acquire(blocking=False) if not acquired: - raise RuntimeError("Embedding model is busy before cluster commit") + raise ClusterModelBusyError("Embedding model is busy before cluster commit") try: yield validate finally: @@ -267,12 +271,16 @@ def cluster_messages( import hdbscan from truememory import vector_search - # Model loading may also hold this lock. Wait before opening our read - # transaction, and release both before any clustering computation. - with vector_search._lock: + # Standalone callers can wait before opening a read transaction. A caller + # transaction may already own SQLite's writer, so never wait inside it. + if not vector_search._lock.acquire(blocking=not conn.in_transaction): + raise ClusterModelBusyError("Embedding model is busy before clustering snapshot") + try: with _cluster_transaction(conn): source_state, categories = _cluster_source_state(conn, vector_search, collect_categories=True) msg_ids, embeddings = _get_all_embeddings(conn) + finally: + vector_search._lock.release() if any(mid not in categories for mid in msg_ids): raise RuntimeError("Clustering found vectors without source messages; rebuild the vector index") @@ -321,7 +329,7 @@ def cluster_messages( # load or a tier change. Keep the model stable through commit. model_lock_acquired = vector_search._lock.acquire(blocking=False) if not model_lock_acquired: - raise RuntimeError("Embedding model is busy; retry clustering after model loading or tier switching") + raise ClusterModelBusyError("Embedding model is busy; retry clustering after model loading or tier switching") current_state, _ = _cluster_source_state(conn, vector_search) if current_state != source_state: raise RuntimeError("Clustering source changed during computation; retry consolidation") diff --git a/truememory/maintenance.py b/truememory/maintenance.py index 51272a21..325ac2c8 100644 --- a/truememory/maintenance.py +++ b/truememory/maintenance.py @@ -4,7 +4,10 @@ builder invocation is introduced by importing this module. """ +import hashlib import importlib +import importlib.metadata +import importlib.util import json import os import re @@ -17,6 +20,7 @@ from collections.abc import Callable, Iterator from contextlib import AbstractContextManager, ExitStack, contextmanager from pathlib import Path +from types import ModuleType from typing import NamedTuple from truememory._platform import try_file_lock @@ -31,6 +35,14 @@ class MaintenanceBusyError(RuntimeError): """Another operation owns maintenance for this database.""" +class LayerUnavailableError(MaintenanceUnavailableError): + """A builder could not use a dependency advertised by its cheap probe.""" + + +class LayerDeferredError(RuntimeError): + """Transient model ownership contention does not establish an attempt baseline.""" + + class MaintenanceOwnership(NamedTuple): path: Path | None process_id: int @@ -327,6 +339,7 @@ class LayerDependency(NamedTuple): builder_version: int available: bool error_category: str | None = None + deferred: bool = False def make_layer_dependency( @@ -353,6 +366,9 @@ class LayerSpec(NamedTuple): build: Callable[[sqlite3.Connection], object] count_output: Callable[[sqlite3.Connection], int] publication_guard: Callable[[sqlite3.Connection], AbstractContextManager[Callable[[], None]]] | None = None + read_coverage: Callable[[sqlite3.Connection], str] | None = None + borrowed_publication_guard: Callable[[sqlite3.Connection], AbstractContextManager[Callable[[], None]]] | None = None + connection: sqlite3.Connection | None = None class LayerState(NamedTuple): @@ -371,6 +387,7 @@ class LayerState(NamedTuple): output_count: int | None run_generation: str | None error_category: str | None + successful_coverage: str = "unverified" class LayerResult(NamedTuple): @@ -381,6 +398,15 @@ class LayerResult(NamedTuple): elapsed_seconds: float error_category: str | None attempted: bool + coverage: str = "unverified" + pending_caller_commit: bool = False + + +_LAYER_COVERAGE = frozenset({"unverified", "complete", "legacy_contacts_unowned", "vector_generation_unverified"}) + + +class _LayerRollbackFailed(RuntimeError): + """Do not continue publication or diagnostics after a rejected rollback.""" def _validate_layer(layer: str) -> None: @@ -390,6 +416,8 @@ def _validate_layer(layer: str) -> None: def layer_freshness(source: SourceRevision, state: LayerState, dependency: LayerDependency) -> str: """Separate a prior valid success from the latest attempt's outcome.""" + if dependency.deferred: + return "dependency_deferred" if (state.successful_epoch != source.epoch or state.successful_revision is None or state.full_rebuild_required or state.output_count is None): return "untrusted" @@ -409,6 +437,8 @@ def read_layer_states(conn: sqlite3.Connection, specs: tuple[LayerSpec, ...]) -> dependencies = {} for spec in specs: _validate_layer(spec.layer) + if spec.connection is not None and spec.connection is not conn: + raise ValueError("Layer adapters belong to another connection") if spec.layer in dependencies: raise ValueError("Duplicate maintenance layer") dependencies[spec.layer] = spec.resolve_dependency() @@ -419,7 +449,7 @@ def read_layer_states(conn: sqlite3.Connection, specs: tuple[LayerSpec, ...]) -> "s.nonappend_revision,s.tracking_ready,l.layer,l.outcome,l.successful_epoch," "l.successful_revision,l.successful_dependency,l.attempted_epoch,l.attempted_revision," "l.attempted_dependency,l.attempted_insert_count,l.full_rebuild_required,l.output_count," - "l.run_generation,l.error_category FROM maintenance_source_state s LEFT JOIN maintenance_layers l " + "l.run_generation,l.error_category,l.successful_coverage FROM maintenance_source_state s LEFT JOIN maintenance_layers l " f"ON l.layer IN ({placeholders}) WHERE s.singleton=1", tuple(dependencies), ).fetchall() except sqlite3.OperationalError as error: @@ -433,7 +463,7 @@ def read_layer_states(conn: sqlite3.Connection, specs: tuple[LayerSpec, ...]) -> states = {} for layer, dependency in dependencies.items(): row = stored.get(layer) - values = row[8:] if row is not None else ("pending", None, None, None, None, None, None, None, 1, None, None, None) + values = row[8:] if row is not None else ("pending", None, None, None, None, None, None, None, 1, None, None, None, "unverified") states[layer] = LayerState(layer, source, dependency, *values) return states @@ -455,6 +485,8 @@ def layer_read_snapshot( def _eligible(state: LayerState, threshold: int, force: bool) -> bool: + if state.dependency.deferred: + return False if state.outcome == "running": return False if force or state.outcome == "abandoned": @@ -486,13 +518,14 @@ def plan_layers( states = read_layer_states(conn, specs) # A previous process may have died while recording "running". Let the # real ownership claim distinguish active work from an abandoned record. - return tuple(spec for spec in specs if states[spec.layer].outcome == "running" - or _eligible(states[spec.layer], threshold, force)) + return tuple(spec for spec in specs if not states[spec.layer].dependency.deferred + and (states[spec.layer].outcome == "running" or _eligible(states[spec.layer], threshold, force))) def record_layer_success_in_transaction( conn: sqlite3.Connection, *, layer: str, dependency: LayerDependency, source: SourceRevision, output_count: int, run_generation: str | None = None, + coverage: str = "complete", ) -> None: """Publish provenance beside completed output, without owning caller commit.""" _validate_layer(layer) @@ -500,26 +533,31 @@ def record_layer_success_in_transaction( raise RuntimeError("Layer publication requires an existing transaction") if not dependency.available or type(output_count) is not int or output_count < 0: raise ValueError("Successful layer publication requires available output") + if coverage not in _LAYER_COVERAGE: + raise ValueError("Invalid layer coverage") if read_source_revision(conn) != source: raise sqlite3.OperationalError("Maintenance source changed before publication") conn.execute( "INSERT INTO maintenance_layers(layer,builder_version,outcome,successful_epoch,successful_revision," "successful_dependency,attempted_epoch,attempted_revision,attempted_dependency,attempted_insert_count," - "full_rebuild_required,output_count,run_generation,error_category) VALUES (?,?,?,?,?,?,?,?,?,?,0,?,?,NULL) " + "full_rebuild_required,output_count,run_generation,error_category,successful_coverage) VALUES (?,?,?,?,?,?,?,?,?,?,0,?,?,NULL,?) " "ON CONFLICT(layer) DO UPDATE SET builder_version=excluded.builder_version,outcome=excluded.outcome," "successful_epoch=excluded.successful_epoch,successful_revision=excluded.successful_revision," "successful_dependency=excluded.successful_dependency,attempted_epoch=excluded.attempted_epoch," "attempted_revision=excluded.attempted_revision,attempted_dependency=excluded.attempted_dependency," "attempted_insert_count=excluded.attempted_insert_count,full_rebuild_required=0," - "output_count=excluded.output_count,run_generation=excluded.run_generation,error_category=NULL", + "output_count=excluded.output_count,run_generation=excluded.run_generation,error_category=NULL," + "successful_coverage=excluded.successful_coverage", (layer, dependency.builder_version, "success" if output_count else "success_empty", source.epoch, source.revision, dependency.key, source.epoch, source.revision, dependency.key, - source.insert_count, output_count, run_generation), + source.insert_count, output_count, run_generation, coverage), ) @contextmanager -def _owned_transaction(conn: sqlite3.Connection) -> Iterator[None]: +def _owned_transaction( + conn: sqlite3.Connection, *, on_rollback: Callable[[], None] | None = None, +) -> Iterator[None]: if conn.in_transaction: raise RuntimeError("Maintenance runner requires a clean transaction boundary") conn.execute("BEGIN") @@ -528,16 +566,55 @@ def _owned_transaction(conn: sqlite3.Connection) -> Iterator[None]: yield conn.execute("COMMIT") completed = True + finally: + if not completed: + if conn.in_transaction: + try: + conn.rollback() + except BaseException as error: + if on_rollback is not None: + raise _LayerRollbackFailed("Output rollback failed") from error + raise + if on_rollback is not None: + on_rollback() + elif on_rollback is not None: + raise _LayerRollbackFailed("Output transaction ended without a confirmed rollback") + + +@contextmanager +def _borrowed_transaction( + conn: sqlite3.Connection, *, on_rollback: Callable[[], None] | None = None, +) -> Iterator[None]: + if not conn.in_transaction: + raise RuntimeError("Borrowed maintenance requires the caller's transaction") + name = "truememory_layer_" + uuid.uuid4().hex + conn.execute(f"SAVEPOINT {name}") + completed = False + try: + yield + conn.execute(f"RELEASE {name}") + completed = True finally: if not completed and conn.in_transaction: - conn.rollback() + try: + conn.execute(f"ROLLBACK TO {name}") + except BaseException as error: + raise _LayerRollbackFailed("Caller must roll back the failed maintenance transaction") from error + if on_rollback is not None: + on_rollback() + conn.execute(f"RELEASE {name}") + elif not completed and on_rollback is not None: + raise _LayerRollbackFailed("Caller transaction ended without a confirmed rollback") def _record_attempt( conn: sqlite3.Connection, state: LayerState, outcome: str, generation: str, error_category: str | None, + *, borrowed: bool = False, ) -> None: source, dependency = state.source, state.dependency - with _owned_transaction(conn): + if dependency.deferred: + raise ValueError("Transient deferral cannot be recorded as a durable attempt") + with (_borrowed_transaction(conn) if borrowed else _owned_transaction(conn)): conn.execute( "INSERT INTO maintenance_layers(layer,builder_version,outcome,attempted_epoch,attempted_revision," "attempted_dependency,attempted_insert_count,run_generation,error_category) VALUES (?,?,?,?,?,?,?,?,?) " @@ -554,75 +631,150 @@ class _LayerCancelled(Exception): """Internal cooperative cancellation before publication.""" +_LAYER_ROW_COLUMNS = ( + "layer", "builder_version", "outcome", "successful_epoch", "successful_revision", "successful_dependency", + "successful_coverage", "attempted_epoch", "attempted_revision", "attempted_dependency", "attempted_insert_count", + "full_rebuild_required", "output_count", "run_generation", "error_category", +) + + +def _read_layer_row(conn: sqlite3.Connection, layer: str) -> tuple | None: + row = conn.execute("SELECT " + ",".join(_LAYER_ROW_COLUMNS) + " FROM maintenance_layers WHERE layer=?", (layer,)).fetchone() + return tuple(row) if row is not None else None + + +def _restore_deferred_attempt( + conn: sqlite3.Connection, layer: str, previous: tuple | None, running: tuple, + owner: MaintenanceOwnership, +) -> None: + """Restore only this owner's unchanged diagnostic, after output rollback.""" + if conn.in_transaction: + raise _LayerRollbackFailed("Output rollback is not complete") + current_owner = getattr(_thread_owners, "owners", {}).get(owner.path) + if current_owner != owner or owner.process_id != os.getpid(): + raise MaintenanceBusyError("Maintenance ownership changed before diagnostic restoration") + with _owned_transaction(conn): + conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + current = _read_layer_row(conn, layer) + if (current != running or current is None or current[2] != "running" + or current[_LAYER_ROW_COLUMNS.index("run_generation")] != owner.generation): + raise MaintenanceBusyError("Maintenance diagnostic changed before restoration") + if previous is None: + conn.execute("DELETE FROM maintenance_layers WHERE layer=? AND outcome='running' AND run_generation=?", + (layer, owner.generation)) + else: + assignments = ",".join(column + "=?" for column in _LAYER_ROW_COLUMNS[1:]) + conn.execute("UPDATE maintenance_layers SET " + assignments + + " WHERE layer=? AND outcome='running' AND run_generation=?", + (*previous[1:], layer, owner.generation)) + + +def _check_layer_dependency(current: LayerDependency, expected: LayerDependency) -> None: + if current.deferred: + raise LayerDeferredError("Embedding runtime is temporarily busy") + if current != expected: + raise MaintenanceUnavailableError("Layer dependency changed before publication") + + def run_layers( conn: sqlite3.Connection, specs: tuple[LayerSpec, ...], *, force: bool = False, threshold: int = 25, cancel: threading.Event | None = None, + allow_caller_transaction: bool = False, ) -> tuple[LayerResult, ...]: - """Run each eligible trusted adapter once on a dedicated connection.""" - if conn.in_transaction: + """Run eligible adapters once; borrowing requires explicit caller opt-in.""" + if type(allow_caller_transaction) is not bool: + raise ValueError("Caller transaction opt-in must be a boolean") + borrowed = conn.in_transaction and allow_caller_transaction + if conn.in_transaction and not borrowed: raise RuntimeError("Maintenance runner requires a clean transaction boundary") if type(threshold) is not int or threshold < 1: raise ValueError("Maintenance threshold must be a positive integer") + if any(spec.connection is not None and spec.connection is not conn for spec in specs): + raise ValueError("Layer adapters belong to another connection") results = [] with maintenance_owner(connection_database_path(conn)) as owner: - if conn.execute( - "SELECT 1 FROM maintenance_layers WHERE outcome='running' AND run_generation IS NOT ? LIMIT 1", - (owner.generation,), - ).fetchone(): - with _owned_transaction(conn): - conn.execute( - "UPDATE maintenance_layers SET outcome='abandoned',error_category='Interrupted' " - "WHERE outcome='running' AND run_generation IS NOT ?", (owner.generation,), - ) + def record_attempt(state: LayerState, outcome: str, category: str | None) -> None: + if borrowed: + _record_attempt(conn, state, outcome, owner.generation, category, borrowed=True) + else: + _record_attempt(conn, state, outcome, owner.generation, category) + # Validate uniqueness before any builder is invoked. read_layer_states(conn, specs) for spec in specs: if cancel is not None and cancel.is_set(): break state = read_layer_states(conn, (spec,))[spec.layer] + if state.dependency.deferred: + results.append(LayerResult(spec.layer, spec.result_key, "deferred", state.output_count, 0.0, + state.dependency.error_category, False, state.successful_coverage, borrowed)) + continue + if state.outcome == "running" and state.run_generation != owner.generation: + # Ownership proves abandonment. Do not acquire a caller's + # writer merely to persist a diagnostic before computation. + state = state._replace(outcome="abandoned") if not _eligible(state, threshold, force): results.append(LayerResult(spec.layer, spec.result_key, layer_freshness(state.source, state, state.dependency), state.output_count, 0.0, - state.error_category, False)) + state.error_category, False, state.successful_coverage, borrowed)) continue started = time.monotonic() + baseline_state = state if not state.dependency.available: - _record_attempt(conn, state, "unavailable", owner.generation, state.dependency.error_category) + record_attempt(state, "unavailable", state.dependency.error_category) results.append(LayerResult(spec.layer, spec.result_key, "unavailable", state.output_count, - time.monotonic() - started, state.dependency.error_category, True)) + time.monotonic() - started, state.dependency.error_category, True, state.successful_coverage, borrowed)) continue # This diagnostic commits before the read/compute snapshot starts. - with _owned_transaction(conn): - conn.execute( - "INSERT INTO maintenance_layers(layer,builder_version,outcome,run_generation,error_category) " - "VALUES (?,?,'running',?,NULL) ON CONFLICT(layer) DO UPDATE SET " - "builder_version=excluded.builder_version,outcome='running'," - "run_generation=excluded.run_generation,error_category=NULL", - (spec.layer, state.dependency.builder_version, owner.generation), - ) + previous_row = running_row = None + if not borrowed: + with _owned_transaction(conn): + previous_row = _read_layer_row(conn, spec.layer) + conn.execute( + "INSERT INTO maintenance_layers(layer,builder_version,outcome,run_generation,error_category) " + "VALUES (?,?,'running',?,NULL) ON CONFLICT(layer) DO UPDATE SET " + "builder_version=excluded.builder_version,outcome='running'," + "run_generation=excluded.run_generation,error_category=NULL", + (spec.layer, state.dependency.builder_version, owner.generation), + ) + running_row = _read_layer_row(conn, spec.layer) count = None + coverage = state.successful_coverage + built = output_started = rolled_back = False + + def confirm_rollback() -> None: + nonlocal rolled_back + rolled_back = True + try: # Factories capture identity before the attempt's read snapshot. # The deferred guard enters only at publication, then outlives # the transaction so both COMMIT and rollback retain ownership. - guard = spec.publication_guard(conn) if spec.publication_guard is not None else None + factory = spec.borrowed_publication_guard if borrowed else spec.publication_guard + if borrowed and spec.publication_guard is not None and factory is None: + raise LayerUnavailableError("The adapter has no borrowed publication contract") + guard = factory(conn) if factory is not None else None with ExitStack() as publication: - with _owned_transaction(conn): + transaction = _borrowed_transaction if borrowed else _owned_transaction + with transaction(conn, on_rollback=confirm_rollback): + output_started = True state = read_layer_states(conn, (spec,))[spec.layer] + if state.dependency.deferred: + raise LayerDeferredError("Embedding runtime is temporarily busy") if not state.dependency.available: raise MaintenanceUnavailableError("Layer dependency became unavailable") + built = True spec.build(conn) if not conn.in_transaction: raise RuntimeError("Layer builder committed the runner transaction") count = spec.count_output(conn) + coverage = spec.read_coverage(conn) if spec.read_coverage is not None else "complete" if cancel is not None and cancel.is_set(): raise _LayerCancelled() - if spec.resolve_dependency() != state.dependency: - raise MaintenanceUnavailableError("Layer dependency changed before publication") + _check_layer_dependency(spec.resolve_dependency(), state.dependency) record_layer_success_in_transaction(conn, layer=spec.layer, dependency=state.dependency, - source=state.source, output_count=count, run_generation=owner.generation) - if spec.resolve_dependency() != state.dependency: - raise MaintenanceUnavailableError("Layer dependency changed before commit") + source=state.source, output_count=count, run_generation=owner.generation, coverage=coverage) + _check_layer_dependency(spec.resolve_dependency(), state.dependency) if guard is not None: # Register release before validation can fail, so # validation errors also roll back under ownership. @@ -631,26 +783,39 @@ def run_layers( if cancel is not None and cancel.is_set(): raise _LayerCancelled() except BaseException as error: + if isinstance(error, _LayerRollbackFailed): + raise + if isinstance(error, LayerDeferredError): + if output_started and not rolled_back: + raise _LayerRollbackFailed("Deferred output has no confirmed rollback") from error + if not borrowed: + _restore_deferred_attempt(conn, spec.layer, previous_row, running_row, owner) + results.append(LayerResult(spec.layer, spec.result_key, "deferred", baseline_state.output_count, + time.monotonic() - started, "ModelBusy", built, baseline_state.successful_coverage, borrowed)) + if cancel is not None and cancel.is_set(): + break + continue interrupted = not isinstance(error, Exception) or isinstance(error, _LayerCancelled) - outcome = "abandoned" if interrupted else ("failed" if state.dependency.available else "unavailable") + unavailable = not state.dependency.available or isinstance(error, LayerUnavailableError) + outcome = "abandoned" if interrupted else ("unavailable" if unavailable else "failed") category = "Cancelled" if interrupted else type(error).__name__[:64] if re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", category) is None: category = "Error" - _record_attempt(conn, state, outcome, owner.generation, category) + record_attempt(state, outcome, category) results.append(LayerResult(spec.layer, spec.result_key, outcome, state.output_count, - time.monotonic() - started, category, True)) + time.monotonic() - started, category, True, state.successful_coverage, borrowed)) if not isinstance(error, Exception): raise if interrupted: break else: results.append(LayerResult(spec.layer, spec.result_key, "success" if count else "success_empty", - count, time.monotonic() - started, None, True)) + count, time.monotonic() - started, None, True, coverage, borrowed)) return tuple(results) def nonvector_layer_specs() -> tuple[LayerSpec, ...]: - """Lazy standard-library adapters; no vector or Dunbar enrollment yet.""" + """The original six standard-library adapters, with unchanged defaults.""" definitions = ( ("summaries", "build_summaries", "consolidation", "build_summaries", "SELECT count(*) FROM summaries WHERE period IN ('monthly','entity_monthly')", {}), @@ -683,3 +848,197 @@ def count(conn: sqlite3.Connection, sql: str = count_sql) -> int: specs.append(LayerSpec(layer, result_key, resolve, build, count)) return tuple(specs) + + +def _installed_dependency(module: str, distribution: str) -> str | None: + """Inspect package metadata without importing a native dependency.""" + try: + if importlib.util.find_spec(module) is None: + return None + return importlib.metadata.version(distribution) + except (ImportError, ValueError, importlib.metadata.PackageNotFoundError): + return None + + +def _cluster_schedule_dependency(conn: sqlite3.Connection) -> LayerDependency: + parameters = {"min_cluster_size": 10, "min_samples": 5, "metric": "euclidean", + "cluster_selection_method": "eom"} + versions = {module: _installed_dependency(module, distribution) for module, distribution in ( + ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"), + )} + parameters["versions"] = versions + + def unavailable(category: str) -> LayerDependency: + return make_layer_dependency(1, parameters, available=False, error_category=category) + + if any(version is None for version in versions.values()): + return unavailable("DependencyMissing") + vector = sys.modules.get("truememory.vector_search") + if vector is None: + return unavailable("VectorRuntimeUnavailable") + if not vector._lock.acquire(blocking=False): + # There is no safe stable-key observation while the model is owned. + # This ephemeral value must never become a durable version or attempt. + return LayerDependency("", 1, False, "ModelBusy", deferred=True) + owned = not conn.in_transaction + try: + if owned: + conn.execute("BEGIN") + group = vector._active_tier_group() + table = vector._active_vec_table(conn) + schema = [tuple(row) for row in conn.execute( + "SELECT type,name,rootpage,sql FROM sqlite_master WHERE name=?", (table,), + )] + registry = [tuple(row) for row in conn.execute( + "SELECT tier_group,vec_table,sep_table,model_name,embedding_dim " + "FROM vector_cache_registry WHERE tier_group=?", (group,), + )] + metadata = [tuple(row) for row in conn.execute( + "SELECT key,value FROM metadata WHERE key IN ('embed_model','embed_dim',?,?) ORDER BY key", + (f"vec_build_state:{table}", f"vec_source_v1:{table}"), + )] + # Timestamp/count rewrites do not mean a new embedding space. Keep + # them in the publication fence, not in the durable scheduling key. + identity = [vector.EMBEDDING_MODEL, vector._embedding_dim, group, table, schema, registry, metadata] + try: + encoded = json.dumps(identity, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + except (TypeError, ValueError): + return unavailable("VectorMetadataInvalid") + parameters["identity_sha256"] = hashlib.sha256(encoded).hexdigest() + if not schema: + return unavailable("VectorIndexMissing") + if any("USING VEC0" in (row[3] or "").upper() for row in schema): + if "vec0" not in {row[0] for row in conn.execute("PRAGMA module_list")}: + return unavailable("VectorExtensionUnavailable") + if (f"vec_build_state:{table}", "in_progress") in metadata: + return unavailable("VectorRebuildInProgress") + descriptor = dict(metadata).get(f"vec_source_v1:{table}") + if descriptor is not None: + try: + manifest = json.loads(descriptor) + except (TypeError, ValueError): + return unavailable("VectorManifestInvalid") + if (not isinstance(manifest, dict) or type(manifest.get("version")) is not int + or manifest["version"] != 1 or type(manifest.get("complete")) is not bool): + return unavailable("VectorManifestInvalid") + if not manifest["complete"]: + return unavailable("VectorRebuildIncomplete") + return make_layer_dependency(1, parameters) + except sqlite3.OperationalError: + return unavailable("VectorSchemaUnavailable") + finally: + try: + if owned and conn.in_transaction: + conn.rollback() + finally: + vector._lock.release() + + +def _cluster_module() -> ModuleType: + try: + return importlib.import_module("truememory.clustering") + except ImportError as error: + raise LayerUnavailableError("Clustering dependency import failed") from error + + +def _cluster_guard(conn: sqlite3.Connection) -> AbstractContextManager[Callable[[], None]]: + module = _cluster_module() + guard = module.cluster_publication_guard(conn) + + @contextmanager + def hold() -> Iterator[Callable[[], None]]: + try: + with guard as validate: + yield validate + except module.ClusterModelBusyError as error: + raise LayerDeferredError("Embedding runtime is temporarily busy") from error + + return hold() + + +def _borrowed_cluster_guard(conn: sqlite3.Connection) -> AbstractContextManager[Callable[[], None]]: + """Fence a borrowed RELEASE, never the caller's later outer COMMIT.""" + module = _cluster_module() + vector = sys.modules.get("truememory.vector_search") + if vector is None: + raise LayerUnavailableError("Embedding runtime is unavailable") + if not vector._lock.acquire(blocking=False): + raise LayerDeferredError("Embedding runtime is temporarily busy") + try: + expected = module._cluster_dependency_identity(conn, vector) + finally: + vector._lock.release() + + @contextmanager + def hold() -> Iterator[Callable[[], None]]: + conn.execute("UPDATE messages SET id=id WHERE 0") + if not vector._lock.acquire(blocking=False): + raise LayerDeferredError("Embedding runtime is temporarily busy") + try: + def validate() -> None: + if module._cluster_dependency_identity(conn, vector) != expected: + raise MaintenanceUnavailableError("Clustering dependency changed before caller savepoint release") + yield validate + finally: + vector._lock.release() + + return hold() + + +def _build_clusters(conn: sqlite3.Connection) -> int: + module = _cluster_module() + try: + return module.cluster_messages(conn) + except module.ClusterModelBusyError as error: + raise LayerDeferredError("Embedding runtime is temporarily busy") from error + except ImportError as error: + raise LayerUnavailableError("Clustering dependency import failed") from error + + +def _build_dunbar(conn: sqlite3.Connection) -> object: + # run_layers supplies the read snapshot, including primary selection. + row = conn.execute( + "SELECT sender,COUNT(*) AS cnt FROM messages WHERE sender != '' AND sender IS NOT NULL " + "GROUP BY sender ORDER BY cnt DESC LIMIT 1" + ).fetchone() + primary = row[0] if row and row[0] and row[0].strip() else None + return importlib.import_module("truememory.personality").build_dunbar_hierarchy(conn, primary_entity=primary) + + +def _dunbar_coverage(conn: sqlite3.Connection) -> str: + coverage = importlib.import_module("truememory.personality").read_dunbar_coverage(conn) + return {"managed_only": "complete", "partial": "legacy_contacts_unowned", "untracked": "unverified"}[coverage["status"]] + + +def all_layer_specs(conn: sqlite3.Connection) -> tuple[LayerSpec, ...]: + """Eight persisted adapters; no imports of native packages or model loads. + + These adapters are bound to this connection. Vector users supply sqlite-vec + and the initialized vector_search runtime. The cheap dependency resolver + reports unavailable otherwise. Preferences remain a separate nonpersisted call. + """ + specs = {spec.layer: spec for spec in nonvector_layer_specs()} + + def dunbar_dependency() -> LayerDependency: + try: + module = importlib.import_module("truememory.personality") + getattr(module, "build_dunbar_hierarchy") + getattr(module, "read_dunbar_coverage") + except (ImportError, AttributeError) as error: + return make_layer_dependency(1, available=False, error_category=type(error).__name__) + return make_layer_dependency(1) + + specs["clusters"] = LayerSpec( + "clusters", "cluster_messages", lambda: _cluster_schedule_dependency(conn), _build_clusters, + lambda connection: connection.execute("SELECT count(*) FROM cluster_centroids").fetchone()[0], + _cluster_guard, lambda _conn: "vector_generation_unverified", _borrowed_cluster_guard, + conn, + ) + specs["dunbar"] = LayerSpec( + "dunbar", "dunbar_hierarchy", dunbar_dependency, _build_dunbar, + lambda connection: importlib.import_module("truememory.personality").read_dunbar_coverage(connection)["managed_rows"], + read_coverage=_dunbar_coverage, + ) + return tuple(specs[name] for name in ( + "clusters", "summaries", "contradictions", "structured_facts", "surprise", "episodes", "landmarks", "dunbar", + )) diff --git a/truememory/storage.py b/truememory/storage.py index 0bd22053..d0a98ff6 100644 --- a/truememory/storage.py +++ b/truememory/storage.py @@ -295,6 +295,8 @@ successful_epoch TEXT, successful_revision INTEGER, successful_dependency TEXT, + successful_coverage TEXT NOT NULL DEFAULT 'unverified' CHECK (successful_coverage IN + ('unverified', 'complete', 'legacy_contacts_unowned', 'vector_generation_unverified')), attempted_epoch TEXT, attempted_revision INTEGER, attempted_dependency TEXT, @@ -403,7 +405,7 @@ def installed() -> dict[str, str]: layers = {row[0] for row in conn.execute("SELECT layer FROM maintenance_layers")} layer_columns = {row[1] for row in conn.execute("PRAGMA table_info(maintenance_layers)")} if (ready and ready[0] and installed() == expected and set(_MAINTENANCE_LAYERS).issubset(layers) - and "attempted_insert_count" in layer_columns): + and {"attempted_insert_count", "successful_coverage"}.issubset(layer_columns)): return owned = not conn.in_transaction @@ -417,6 +419,12 @@ def installed() -> dict[str, str]: "CHECK (attempted_insert_count IS NULL OR " "(typeof(attempted_insert_count) = 'integer' AND attempted_insert_count >= 0))" ) + if "successful_coverage" not in {row[1] for row in conn.execute("PRAGMA table_info(maintenance_layers)")}: + conn.execute( + "ALTER TABLE maintenance_layers ADD COLUMN successful_coverage TEXT NOT NULL DEFAULT 'unverified' " + "CHECK (successful_coverage IN " + "('unverified','complete','legacy_contacts_unowned','vector_generation_unverified'))" + ) columns = {row[1] for row in conn.execute("PRAGMA table_info(messages)")} available = set(_MAINTENANCE_SOURCE_FIELDS).issubset(columns) conn.execute(