From 73b7a0509b956a38d2cfa0de2768158a6a433268 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 17:30:09 -0500 Subject: [PATCH 1/2] fix: Advance style coverage with verified incremental appends --- docs/resource-budgets.md | 66 ++ tests/test-l0-style-accumulator-747.py | 14 +- tests/test-l0-style-incremental-747.py | 956 ++++++++++++++++++ ...st-maintenance-foreground-admission-753.py | 10 + truememory/engine.py | 18 +- truememory/maintenance.py | 302 +++++- 6 files changed, 1340 insertions(+), 26 deletions(-) create mode 100644 tests/test-l0-style-incremental-747.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 813c54aa..056c548a 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -1286,3 +1286,69 @@ This checkpoint does not activate engine routing, automatic migration, incremental coverage, style status or coordinator style-only jobs. Those remaining migration steps are required before release. Synthetic transaction proofs do not establish native performance or retrieval quality. + +### Incremental style coverage checkpoint + +Single-memory adds can now advance an explicitly enrolled, current style +checkpoint. They capture complete successful provenance and the attempt +baseline before insertion, inside the existing source writer transaction. +The source epoch, correction count and nonappend revision must stay unchanged; +revision and insert count must each increase by exactly 1. The inserted ID +must exceed the preceding high watermark and equal the new high watermark. +The inserted source row and unchanged style checkpoint are verified before +the existing raw-sum updater runs. Profile reads use entity-key lookups; +source verification uses the inserted primary key. No full source or profile +scan, schema enrollment or fallback rebuild runs on add. + +Hash version 2, accumulator schema, exact owned trigger definitions and the +dependency identity are checked again after source insertion, after the raw +append and after successful provenance publication. An ordinary profile write +must produce the exact invalidated projection of the captured checkpoint; +excluded inputs must leave that checkpoint unchanged. The final successful +row must match the intended source, output count and both baselines exactly. +Changed metadata or provenance rolls back style publication instead of +certifying it. Pending category writes compare every checkpoint field, so +they cannot overwrite a newer failure, owner or attempt count. + +Incremental and tracked style work conservatively defer schemas with any +extra MAIN trigger on `entity_style_vectors`, any MAIN trigger on +`maintenance_layers` or `metadata`, or any TEMP trigger on these tables. Only the three +fixed owned output trigger names with their exact definitions permit append +coverage; a similar name or a TEMP shadow is insufficient. These checks read +schema metadata and never scan source rows or profiles. Unsupported schemas +also skip pending checkpoint writes, since those writes could themselves +invoke an unsupported trigger. Preparation, running diagnostics, tracked +output admission and deferred restoration recheck under writer ownership +before mutating affected rows. The private dependency remains deferred with +`StyleTriggersUnsupported`; future coordinator/status routing must preserve +that status rather than certify or retry unsupported work as successful. +Metadata triggers are included because publishing the hash marker can invoke +profile mutations after a rebuild has written its output. + +The raw sum and both provenance baselines publish in one nested savepoint. +An existing entity keeps the output count; a new entity adds 1. Directive or +empty-sender rows advance source coverage without touching profiles. Short +texts with zero vectors still increment message counts. Identical content in +two different source rows counts twice; replaying an already consumed append +proof cannot count the same row again. The public updater, getter and batch +arithmetic are unchanged. Incremental and batch accumulation orders may have +different floating-point rounding; this checkpoint introduces no tolerance. + +Missing, dirty, legacy, future-format or running coverage skips unsafe profile +updates while retaining the source add. A newly missed append invalidates only +its captured successful/running checkpoint and clears its attempt baseline; +it cannot overwrite a newer invalidation. An existing failed build retains its +retry baseline. Failed precomputation or publication leaves bounded categorical +evidence, and unexpected SQL errors are failures rather than writer deferrals. +If writing that pending evidence also fails, the result reports failure and +does not claim durable invalidation; source freshness still shows the uncovered +append. Normal source commit owns the final publication. Caller transactions +retain their final commit or rollback, and savepoint rollback/release failure +escapes rather than reporting a successful add. + +The synthetic gate starts with one tracked empty bootstrap, performs 75 actual +engine adds and probes tracked eligibility after every add. The required number +of additional full builds is 0; successful and attempted insert baselines both +advance by 75. This checkpoint does not activate startup migration, style-only +coordinator jobs, consolidation/bulk style routing or bounded public style +health. Those routing steps remain required before release. diff --git a/tests/test-l0-style-accumulator-747.py b/tests/test-l0-style-accumulator-747.py index 290b9af5..f343405e 100644 --- a/tests/test-l0-style-accumulator-747.py +++ b/tests/test-l0-style-accumulator-747.py @@ -699,16 +699,22 @@ def test_actual_engine_add_catches_style_deferral_and_keeps_source(self) -> None selected = [node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "rebuild_transaction"] exec(compile(ast.Module(body=selected, type_ignores=[]), "rebuild_source.py", "exec"), rebuild.__dict__) + maintenance = types.ModuleType("synthetic_append_maintenance") + modules = {"truememory.personality_style_vec": STYLE, "truememory.rebuild_source": rebuild, + "truememory.maintenance": maintenance, "truememory.storage": STORAGE, + "truememory._platform": load_stdlib_source("_platform")} + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, fromlist: tuple[str, ...] = (), level: int = 0) -> object: - if name == "truememory.personality_style_vec": - return STYLE - if name == "truememory.rebuild_source": - return rebuild + if name in modules: + return modules[name] if name.split(".", 1)[0] not in sys.stdlib_module_names: raise AssertionError("Unexpected import in engine.add: " + name) return builtins.__import__(name, globals, locals, fromlist, level) + maintenance.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import) + path = ROOT / "truememory/maintenance.py" + exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), maintenance.__dict__) tree = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8")) engine_class = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "TrueMemoryEngine") add = next(node for node in engine_class.body if isinstance(node, ast.FunctionDef) and node.name == "add") diff --git a/tests/test-l0-style-incremental-747.py b/tests/test-l0-style-incremental-747.py new file mode 100644 index 00000000..360f8652 --- /dev/null +++ b/tests/test-l0-style-incremental-747.py @@ -0,0 +1,956 @@ +"""Atomic incremental style coverage with actual source and in-memory SQLite.""" + +import ast +import builtins +import json +import logging +import runpy +import sqlite3 +import sys +import threading +import types +import unittest +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager, nullcontext +from pathlib import Path +from unittest.mock import Mock, patch + + +ROOT = Path(__file__).resolve().parents[1] +BASE = runpy.run_path(str(Path(__file__).with_name("test-l0-style-maintenance-747.py"))) +STORAGE, STYLE, MAINTENANCE = BASE["STORAGE"], BASE["STYLE"], BASE["MAINTENANCE"] +ACCUMULATOR = BASE["ACCUMULATOR"] + + +def load_add() -> types.FunctionType: + rebuild = types.ModuleType("synthetic_incremental_rebuild") + rebuild.__dict__.update(sqlite3=sqlite3, Iterator=Iterator, AbstractContextManager=AbstractContextManager, + contextmanager=contextmanager, nullcontext=nullcontext) + source = ast.parse((ROOT / "truememory/rebuild_source.py").read_text(encoding="utf-8")) + selected = [node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == "rebuild_transaction"] + exec(compile(ast.Module(body=selected, type_ignores=[]), "synthetic-rebuild", "exec"), rebuild.__dict__) + modules = {"truememory.personality_style_vec": STYLE, "truememory.rebuild_source": rebuild, + "truememory.maintenance": MAINTENANCE} + + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple = (), level: int = 0) -> object: + if name in modules: + return modules[name] + if name.split(".", 1)[0] not in sys.stdlib_module_names: + raise AssertionError("Non-stdlib import forbidden: " + name) + return builtins.__import__(name, globals, locals, fromlist, level) + + tree = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8")) + engine = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "TrueMemoryEngine") + add = next(node for node in engine.body if isinstance(node, ast.FunctionDef) and node.name == "add") + namespace = {"__builtins__": dict(vars(builtins), __import__=safe_import), "MAX_CONTENT_LENGTH": 50000, + "insert_message": STORAGE.insert_message, "_update_style_vec": STYLE.update_entity_style_vector_incremental, + "logger": logging.getLogger("synthetic-style-incremental")} + exec(compile(ast.Module(body=[add], type_ignores=[]), "actual-engine-add", "exec"), namespace) + return namespace["add"] + + +ADD = load_add() + + +class BoundaryFailure: + def __init__(self, conn: sqlite3.Connection, boundary: str) -> None: + self.conn, self.boundary = conn, boundary + self.denied = 0 + + def __getattr__(self, name: str) -> object: + return getattr(self.conn, name) + + def commit(self) -> None: + self.execute("COMMIT") + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + match = (sql == "COMMIT" if self.boundary == "commit" else + sql == "RELEASE SAVEPOINT rebuild_source" if self.boundary == "caller_release" else + sql.startswith("RELEASE truememory_style_coverage_")) + if match and not self.denied: + self.denied += 1 + action = ((sqlite3.SQLITE_TRANSACTION, "COMMIT", None) if sql == "COMMIT" else + (sqlite3.SQLITE_SAVEPOINT, "RELEASE", sql.split()[-1])) + fault = ACCUMULATOR["StatementFailure"](self.conn, sql, action) + return fault.execute(sql, *args) + return self.conn.execute(sql, *args) + + +class StyleIncrementalTests(unittest.TestCase): + def connection(self, *, bootstrap: bool = True, cache: int = 100) -> sqlite3.Connection: + conn = sqlite3.connect(":memory:", cached_statements=cache) + self.addCleanup(conn.close) + conn.executescript(STORAGE._SCHEMA_SQL) + STORAGE._initialize_maintenance_tracking(conn) + conn.execute("CREATE TABLE sentinel(value TEXT)") + conn.commit() + if bootstrap: + self.assertEqual(MAINTENANCE.run_style_maintenance(conn).outcome, "success_empty") + return conn + + def engine(self, conn: sqlite3.Connection) -> types.SimpleNamespace: + engine = types.SimpleNamespace(conn=conn, _has_vectors=False, _has_personality=False, _has_style_vec=True, + _write_lock=threading.Lock(), _ensure_connection=lambda: None, + _maybe_auto_consolidate=lambda: None) + engine.add = types.MethodType(ADD, engine) + return engine + + def checkpoint(self, conn: sqlite3.Connection) -> tuple | None: + return MAINTENANCE._read_layer_row(conn, "style_vectors") + + def rows(self, conn: sqlite3.Connection) -> list[tuple]: + return conn.execute("SELECT * FROM entity_style_vectors ORDER BY entity").fetchall() + + def state(self, conn: sqlite3.Connection) -> object: + spec = MAINTENANCE.style_layer_spec(conn) + return MAINTENANCE.read_layer_states(conn, (spec,))[spec.layer] + + def append(self, conn: sqlite3.Connection, previous: object, *, content: str = "synthetic ordinary", + sender: str = "Synthetic", directive: bool = False, message_id: int | None = None, + publish: object = None) -> object: + if message_id is None: + message_id = STORAGE.insert_message(conn, {"content": content, "sender": sender, "directive": directive}) + vector = STYLE.compute_style_vector(content) + return MAINTENANCE._publish_style_append( + conn, previous, message_id, sender=sender, content=content, directive=directive, precomputed=True, + publish=publish or (lambda: STYLE.update_entity_style_vector_incremental( + conn, sender, content, _pre_computed_vec=vector)), + ) + + def test_75_actual_adds_advance_raw_sums_and_attempts_without_any_full_build(self) -> None: + conn = self.connection() + engine = self.engine(conn) + expected = {} + original = STYLE._compute_entity_style_vectors + with patch.object(STYLE, "_compute_entity_style_vectors", wraps=original) as builds: + for index in range(75): + sender = "" if index % 11 == 0 else ("Synthetic" if index % 2 else "SYNTHETIC") + directive = index % 13 == 0 + text = "a" if index % 7 == 0 else "synthetic ordinary " + str(index) + result = engine.add(text, sender=sender, directive=directive) + self.assertEqual(result["id"], index + 1) + if sender and not directive: + total, count = expected.setdefault(sender.lower(), [[0.0] * STYLE.DIM, 0]) + vector = STYLE.compute_style_vector(text) + expected[sender.lower()][0] = [one + two for one, two in zip(total, vector)] + expected[sender.lower()][1] = count + 1 + state = self.state(conn) + self.assertEqual((state.successful_revision, state.attempted_revision, state.attempted_insert_count), + (index + 1, index + 1, index + 1)) + self.assertEqual((state.successful_coverage, state.full_rebuild_required), ("complete", False)) + self.assertEqual(MAINTENANCE.run_style_maintenance(conn, threshold=25).outcome, "current") + self.assertEqual(builds.call_count, 0) + self.assertEqual(self.state(conn).output_count, len(expected)) + for entity, (total, count) in expected.items(): + row = conn.execute("SELECT vector_sum,message_count,accumulator_version,vector FROM entity_style_vectors WHERE entity=?", + (entity,)).fetchone() + self.assertEqual(row[:3], (STYLE._STYLE_SUM.pack(*total), count, 1)) + self.assertEqual(json.loads(row[3]), STYLE._style_profile_from_sum(total, count)) + self.assertFalse(conn.in_transaction) + + def test_new_entities_case_merging_duplicates_and_zero_vectors_count_once(self) -> None: + conn = self.connection() + engine = self.engine(conn) + for sender, text in (("One", "a"), ("ONE", "ab"), ("One", "ab"), ("Two", "synthetic second")): + engine.add(text, sender=sender) + self.assertEqual(conn.execute("SELECT entity,message_count FROM entity_style_vectors ORDER BY entity").fetchall(), + [("one", 3), ("two", 1)]) + self.assertEqual(STYLE.get_entity_style_vector(conn, "one"), [0.0] * STYLE.DIM) + self.assertEqual(self.state(conn).output_count, 2) + + def test_excluded_rows_advance_successful_empty_without_calling_updater(self) -> None: + conn = self.connection() + with patch.dict(ADD.__globals__, _update_style_vec=lambda *args, **kwargs: self.fail("Excluded profile write")): + self.engine(conn).add("synthetic directive", sender="Synthetic", directive=True) + self.engine(conn).add("synthetic anonymous", sender="") + self.assertEqual(self.rows(conn), []) + self.assertEqual((self.state(conn).outcome, self.state(conn).attempted_insert_count), ("success_empty", 2)) + + def test_no_enrollment_and_no_rebuild_when_coverage_is_missing_or_dirty(self) -> None: + for mode in ("missing", "dirty", "legacy", "running", "failed", "future", "hash", "triggers"): + with self.subTest(mode=mode): + conn = self.connection(bootstrap=mode != "missing") + if mode == "dirty": + conn.execute("UPDATE maintenance_layers SET full_rebuild_required=1 WHERE layer='style_vectors'") + elif mode == "legacy": + conn.execute("INSERT INTO entity_style_vectors(entity,vector) VALUES ('synthetic','[]')") + elif mode in {"running", "failed"}: + conn.execute("UPDATE maintenance_layers SET outcome=? WHERE layer='style_vectors'", (mode,)) + elif mode == "future": + conn.execute("ALTER TABLE entity_style_vectors RENAME TO old_style") + conn.execute("CREATE TABLE entity_style_vectors(entity TEXT PRIMARY KEY, vector TEXT, message_count INTEGER, " + "updated_at TEXT, vector_sum BLOB, accumulator_version INTEGER DEFAULT 9)") + for name, definition in STORAGE._STYLE_OUTPUT_TRIGGERS.items(): + conn.execute("DROP TRIGGER " + name) + conn.execute(definition) + self.assertEqual(self.state(conn).successful_coverage, "complete") + elif mode == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif mode == "triggers": + conn.execute("DROP TRIGGER truememory_style_output_insert") + conn.commit() + before, checkpoint = self.rows(conn), self.checkpoint(conn) + updater = Mock(side_effect=AssertionError("Unsafe profile write")) + with patch.dict(ADD.__globals__, _update_style_vec=updater), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError("No fallback rebuild")) as builds: + self.engine(conn).add("synthetic pending", sender="Synthetic") + updater.assert_not_called() + builds.assert_not_called() + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + if mode == "missing": + self.assertIsNone(self.checkpoint(conn)) + self.assertEqual(conn.execute("SELECT count(*) FROM maintenance_layers").fetchone()[0], 8) + if mode == "failed": + self.assertEqual(self.checkpoint(conn), checkpoint) + + def test_incomplete_attempt_or_source_proof_cannot_advance_coverage(self) -> None: + mutations = ( + "UPDATE maintenance_layers SET attempted_insert_count=NULL WHERE layer='style_vectors'", + "UPDATE maintenance_layers SET successful_coverage='unverified' WHERE layer='style_vectors'", + "UPDATE maintenance_layers SET output_count=-1 WHERE layer='style_vectors'", + "UPDATE maintenance_layers SET builder_version=9 WHERE layer='style_vectors'", + "UPDATE maintenance_source_state SET revision=1", + ) + for sql in mutations: + with self.subTest(sql=sql): + conn = self.connection() + conn.execute(sql) + conn.commit() + updater = Mock(side_effect=AssertionError("Incomplete proof")) + with patch.dict(ADD.__globals__, _update_style_vec=updater): + self.engine(conn).add("synthetic pending", sender="Synthetic") + updater.assert_not_called() + self.assertEqual(self.rows(conn), []) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + + def test_precompute_failure_preserves_source_and_invalidates_current_attempt(self) -> None: + conn = self.connection() + with patch.object(STYLE, "compute_style_vector", side_effect=ValueError("synthetic failure")): + self.engine(conn).add("synthetic retained source", sender="Synthetic") + state = self.state(conn) + self.assertEqual((state.outcome, state.error_category, state.attempted_revision), + ("pending", "StylePrecomputeFailed", None)) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertEqual(self.rows(conn), []) + self.assertEqual(MAINTENANCE.run_style_maintenance(conn).outcome, "success") + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 1) + + def test_source_insert_hash_trigger_cannot_certify_an_old_hash(self) -> None: + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + conn.execute("CREATE TRIGGER synthetic_source_hash AFTER INSERT ON messages BEGIN " + "UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'; END") + conn.commit() + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual(conn.execute("SELECT value FROM metadata WHERE key='style_vec_hash_version'").fetchone()[0], "1") + self.assertEqual((self.state(conn).outcome, self.state(conn).error_category), ("pending", "StyleHashPending")) + + def test_extra_main_and_temp_output_triggers_defer_without_deleting_other_profiles(self) -> None: + for temporary, name in ((False, "synthetic_extra"), (True, "synthetic_extra"), + (True, "truememory_style_output_insert"), (False, "truememory_style_output_extra")): + with self.subTest(temporary=temporary, name=name): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic other profile", sender="Other") + before, checkpoint = self.rows(conn), self.checkpoint(conn) + conn.execute("CREATE " + ("TEMP " if temporary else "") + "TRIGGER " + name + + " AFTER INSERT ON main.entity_style_vectors WHEN new.entity='synthetic' BEGIN " + "DELETE FROM entity_style_vectors WHERE entity='other'; END") + conn.commit() + updater = Mock(wraps=STYLE.update_entity_style_vector_incremental) + with patch.dict(ADD.__globals__, _update_style_vec=updater): + engine.add("synthetic retained", sender="Synthetic") + updater.assert_not_called() + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + changes = conn.total_changes + result = MAINTENANCE.run_style_maintenance(conn, force=True) + self.assertEqual((result.outcome, result.error_category, result.attempted), + ("deferred", "StyleTriggersUnsupported", False)) + spec = MAINTENANCE.style_layer_spec(conn) + self.assertTrue(spec.resolve_dependency().deferred) + self.assertEqual(MAINTENANCE.run_layers(conn, (spec,), force=True)[0].error_category, + "StyleTriggersUnsupported") + self.assertEqual(conn.total_changes, changes) + + def test_extra_checkpoint_triggers_cannot_run_while_reporting_pending(self) -> None: + for temporary in (False, True): + for action in ("failure", "delete_profile"): + with self.subTest(temporary=temporary, action=action): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic other profile", sender="Other") + before, checkpoint = self.rows(conn), self.checkpoint(conn) + change = ("UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure'," + "run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors';" + if action == "failure" else "DELETE FROM entity_style_vectors WHERE entity='other';") + conn.execute("CREATE " + ("TEMP " if temporary else "") + + "TRIGGER synthetic_checkpoint AFTER UPDATE OF outcome ON main.maintenance_layers " + "WHEN new.layer='style_vectors' AND new.outcome='pending' BEGIN " + change + " END") + conn.commit() + with patch.object(STYLE, "compute_style_vector", side_effect=ValueError("synthetic precompute")): + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual(self.state(conn).dependency.error_category, "StyleTriggersUnsupported") + + def test_metadata_triggers_defer_before_preparation_or_rebuild_writes(self) -> None: + for temporary in (False, True): + for readiness in ("ready", "unenrolled", "repair"): + for borrowed in (False, True): + with self.subTest(temporary=temporary, readiness=readiness, borrowed=borrowed): + conn = self.connection(bootstrap=False) + STORAGE.insert_message(conn, {"content": "synthetic source", "sender": "Synthetic", "directive": False}) + conn.commit() + if readiness != "unenrolled": + self.assertEqual(MAINTENANCE.run_style_maintenance(conn).outcome, "success") + if readiness == "repair": + conn.execute("DROP TRIGGER truememory_style_output_update") + conn.execute("CREATE " + ("TEMP " if temporary else "") + + "TRIGGER synthetic_metadata AFTER UPDATE OF value ON main.metadata " + "WHEN new.key='style_vec_hash_version' BEGIN DELETE FROM entity_style_vectors; END") + conn.commit() + before, checkpoint = self.rows(conn), self.checkpoint(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('synthetic caller')") + changes = conn.total_changes + sql = [] + conn.set_trace_callback(sql.append) + result = MAINTENANCE.run_style_maintenance(conn, force=True, allow_caller_transaction=borrowed) + conn.set_trace_callback(None) + self.assertEqual((result.outcome, result.error_category, result.attempted), + ("deferred", "StyleTriggersUnsupported", False)) + self.assertFalse(any(statement.lstrip().upper().startswith( + ("INSERT", "UPDATE", "DELETE", "REPLACE", "BEGIN", "SAVEPOINT")) for statement in sql)) + self.assertEqual(conn.total_changes, changes) + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertEqual(conn.in_transaction, borrowed) + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), + [("synthetic caller",)] if borrowed else []) + conn.rollback() + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), []) + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + + def test_metadata_triggers_skip_incremental_publication_and_pending_diagnostics(self) -> None: + for temporary in (False, True): + with self.subTest(temporary=temporary): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before, checkpoint = self.rows(conn), self.checkpoint(conn) + conn.execute("CREATE " + ("TEMP " if temporary else "") + + "TRIGGER synthetic_metadata AFTER UPDATE OF value ON main.metadata " + "WHEN new.key='style_vec_hash_version' BEGIN DELETE FROM entity_style_vectors; END") + conn.commit() + updater = Mock(wraps=STYLE.update_entity_style_vector_incremental) + with patch.dict(ADD.__globals__, _update_style_vec=updater): + engine.add("synthetic retained", sender="Synthetic") + updater.assert_not_called() + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual(self.state(conn).dependency.error_category, "StyleTriggersUnsupported") + + def test_metadata_trigger_installed_during_compute_is_rejected_before_publication(self) -> None: + for temporary in (False, True): + with self.subTest(temporary=temporary): + conn = self.connection() + self.engine(conn).add("synthetic baseline", sender="Synthetic") + before, checkpoint = self.rows(conn), self.checkpoint(conn) + compute = STYLE._compute_entity_style_vectors + + def installed_after_compute(*args: object, **kwargs: object) -> object: + value = compute(*args, **kwargs) + conn.execute("CREATE " + ("TEMP " if temporary else "") + + "TRIGGER synthetic_late_metadata AFTER UPDATE OF value ON main.metadata " + "WHEN new.key='style_vec_hash_version' BEGIN DELETE FROM entity_style_vectors; END") + return value + + with patch.object(STYLE, "_compute_entity_style_vectors", installed_after_compute), \ + patch.object(STYLE, "_publish_style_vectors", wraps=STYLE._publish_style_vectors) as publish: + result = MAINTENANCE.run_style_maintenance(conn, force=True) + publish.assert_not_called() + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + self.assertFalse(MAINTENANCE._style_extra_triggers(conn)) + self.assertFalse(conn.in_transaction) + + def test_source_trigger_newer_failure_is_not_overwritten_by_append_reporting(self) -> None: + conn = self.connection() + conn.executescript("""CREATE TRIGGER synthetic_checkpoint AFTER UPDATE OF outcome ON maintenance_layers + WHEN new.layer='style_vectors' AND new.outcome='pending' BEGIN + UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure', + run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors'; END; + CREATE TRIGGER synthetic_source AFTER INSERT ON messages BEGIN + UPDATE maintenance_layers SET outcome='pending' WHERE layer='style_vectors'; END;""") + self.engine(conn).add("synthetic retained", sender="Synthetic") + state = self.state(conn) + self.assertEqual((state.outcome, state.error_category, state.run_generation, state.attempted_insert_count), + ("failed", "NewerFailure", "newer-owner", 999)) + self.assertEqual(self.rows(conn), []) + + def test_preparation_rejects_unsupported_triggers_before_enrollment_or_repair(self) -> None: + for bootstrap in (False, True): + for temporary in (False, True): + with self.subTest(bootstrap=bootstrap, temporary=temporary): + conn = self.connection(bootstrap=bootstrap) + if bootstrap: + conn.execute("DROP TRIGGER truememory_style_output_update") + conn.execute("CREATE " + ("TEMP " if temporary else "") + + "TRIGGER synthetic_preparation AFTER UPDATE ON main.maintenance_layers BEGIN " + "INSERT INTO sentinel VALUES ('unexpected checkpoint mutation'); END") + conn.commit() + before = self.checkpoint(conn) + changes = conn.total_changes + result = MAINTENANCE.run_style_maintenance(conn) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(conn.total_changes, changes) + self.assertEqual(self.checkpoint(conn), before) + self.assertFalse(conn.in_transaction) + + def test_preparation_rechecks_trigger_admission_inside_its_original_writer_boundary(self) -> None: + for borrowed in (False, True): + conn = self.connection(bootstrap=False) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('synthetic caller')") + installed = [] + + class InstallBeforeAdmission: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql in {"BEGIN IMMEDIATE", "SAVEPOINT truememory_style_tracking"} and not installed: + installed.append(True) + conn.execute("CREATE TEMP TRIGGER synthetic_preparation AFTER UPDATE ON main.maintenance_layers BEGIN " + "INSERT INTO sentinel VALUES ('unexpected checkpoint mutation'); END") + return conn.execute(sql, *args) + + result = MAINTENANCE.run_style_maintenance(InstallBeforeAdmission(), allow_caller_transaction=borrowed) + self.assertEqual(result.error_category, "StyleTriggersUnsupported") + self.assertEqual(installed, [True]) + self.assertEqual(conn.in_transaction, borrowed) + self.assertIsNone(self.checkpoint(conn)) + self.assertEqual(conn.execute("SELECT value FROM sentinel").fetchall(), + [("synthetic caller",)] if borrowed else []) + conn.rollback() + + def test_trigger_installed_after_baseline_cannot_fire_on_running_diagnostic(self) -> None: + conn = self.connection() + before = self.checkpoint(conn) + read = MAINTENANCE.read_layer_states + reads = [] + + def installed_after_read(*args: object, **kwargs: object) -> object: + state = read(*args, **kwargs) + reads.append(True) + if len(reads) == 2: + conn.execute("CREATE TEMP TRIGGER synthetic_late_checkpoint AFTER UPDATE ON main.maintenance_layers " + "BEGIN INSERT INTO sentinel VALUES ('unexpected diagnostic write'); END") + return state + + with patch.object(MAINTENANCE, "read_layer_states", installed_after_read), \ + patch.object(STYLE, "_compute_entity_style_vectors", wraps=STYLE._compute_entity_style_vectors) as builds: + result = MAINTENANCE.run_style_maintenance(conn, force=True) + builds.assert_not_called() + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(self.checkpoint(conn), before) + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), []) + self.assertFalse(conn.in_transaction) + + def test_trigger_installed_after_rollback_cannot_fire_on_diagnostic_restoration(self) -> None: + conn = self.connection() + installed = [] + + class InstallAfterRollback: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def rollback(self) -> None: + conn.rollback() + if not installed: + installed.append(True) + conn.execute("CREATE TEMP TRIGGER synthetic_late_restore AFTER UPDATE ON main.maintenance_layers " + "BEGIN INSERT INTO sentinel VALUES ('unexpected restoration'); END") + + with patch.object(STYLE, "_compute_entity_style_vectors", + side_effect=MAINTENANCE.LayerDeferredError("synthetic busy", category="StyleWriterBusy")): + result = MAINTENANCE.run_style_maintenance(InstallAfterRollback(), force=True) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(self.state(conn).outcome, "running") + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), []) + self.assertFalse(conn.in_transaction) + conn.execute("DROP TRIGGER synthetic_late_restore") + self.assertEqual(MAINTENANCE.run_style_maintenance(conn).outcome, "success_empty") + + def test_tracked_output_admission_checks_triggers_before_any_profile_write(self) -> None: + conn = self.connection() + before = self.checkpoint(conn) + compute = STYLE._compute_entity_style_vectors + + def installed_after_compute(*args: object, **kwargs: object) -> object: + value = compute(*args, **kwargs) + conn.execute("CREATE TEMP TRIGGER synthetic_late_output AFTER DELETE ON main.entity_style_vectors " + "BEGIN INSERT INTO sentinel VALUES ('unexpected output publication'); END") + return value + + with patch.object(STYLE, "_compute_entity_style_vectors", installed_after_compute), \ + patch.object(STYLE, "_publish_style_vectors", wraps=STYLE._publish_style_vectors) as publication: + result = MAINTENANCE.run_style_maintenance(conn, force=True) + publication.assert_not_called() + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(self.checkpoint(conn), before) + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), []) + self.assertFalse(conn.in_transaction) + + def test_unsupported_attempt_diagnostic_reports_deferral_without_writing(self) -> None: + for unavailable in (False, True): + with self.subTest(unavailable=unavailable): + conn = self.connection() + before = self.checkpoint(conn) + installed = [] + + def install() -> None: + if not installed: + installed.append(True) + conn.execute("CREATE TEMP TRIGGER synthetic_late_attempt AFTER UPDATE ON main.maintenance_layers " + "BEGIN INSERT INTO sentinel VALUES ('unexpected attempt write'); END") + + class InstallAfterFailure: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def rollback(self) -> None: + conn.rollback() + install() + + if unavailable: + conn.execute("DROP TRIGGER truememory_style_output_update") + spec = MAINTENANCE.style_layer_spec(conn) + read = MAINTENANCE.read_layer_states + reads = [] + + def installed_after_read(*args: object, **kwargs: object) -> object: + states = read(*args, **kwargs) + reads.append(True) + if len(reads) == 2: + install() + return states + + with patch.object(MAINTENANCE, "read_layer_states", installed_after_read): + result, = MAINTENANCE.run_layers(conn, (spec,), force=True) + self.assertEqual(self.checkpoint(conn), before) + self.assertFalse(result.attempted) + else: + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=ValueError("synthetic failure")): + result = MAINTENANCE.run_style_maintenance(InstallAfterFailure(), force=True) + self.assertEqual(self.state(conn).outcome, "running") + self.assertEqual(installed, [True]) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleTriggersUnsupported")) + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), []) + self.assertFalse(conn.in_transaction) + + def test_updater_metadata_and_checkpoint_mutations_rollback_only_style_work(self) -> None: + for change in ("hash", "ready_trigger", "extra_main", "extra_temp", "checkpoint_trigger", "checkpoint", "format", + "metadata_main", "metadata_temp"): + with self.subTest(change=change): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + publish = STYLE.update_entity_style_vector_incremental + + def changed(*args: object, **kwargs: object) -> None: + publish(*args, **kwargs) + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif change == "ready_trigger": + conn.execute("DROP TRIGGER truememory_style_output_insert") + elif change == "checkpoint": + conn.execute("UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure'," + "run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors'") + elif change == "format": + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN synthetic_format TEXT") + else: + table = ("metadata" if change.startswith("metadata_") else + "maintenance_layers" if change == "checkpoint_trigger" else "entity_style_vectors") + conn.execute("CREATE " + ("TEMP " if change in {"extra_temp", "metadata_temp"} else "") + + "TRIGGER synthetic_during_append AFTER UPDATE ON main." + table + + " BEGIN INSERT INTO sentinel VALUES ('synthetic trigger'); END") + + with patch.dict(ADD.__globals__, _update_style_vec=changed): + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual(self.state(conn).outcome, "pending") + self.assertIsNone(self.state(conn).attempted_revision) + self.assertEqual(conn.execute("SELECT value FROM metadata WHERE key='style_vec_hash_version'").fetchone()[0], "2") + self.assertFalse(MAINTENANCE._style_extra_triggers(conn)) + self.assertTrue(STORAGE._style_output_tracking_ready(conn)) + + def test_post_success_metadata_and_provenance_changes_cannot_certify_publication(self) -> None: + for change in ("hash", "checkpoint", "extra_trigger"): + with self.subTest(change=change): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + record = MAINTENANCE.record_layer_success_in_transaction + + def changed(*args: object, **kwargs: object) -> None: + record(*args, **kwargs) + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif change == "checkpoint": + conn.execute("UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure'," + "run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors'") + else: + conn.execute("CREATE TEMP TRIGGER synthetic_after_success AFTER INSERT ON main.entity_style_vectors " + "BEGIN DELETE FROM entity_style_vectors WHERE entity='synthetic'; END") + + with patch.object(MAINTENANCE, "record_layer_success_in_transaction", changed): + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual(self.state(conn).outcome, "pending") + self.assertIsNone(self.state(conn).attempted_revision) + + def test_pending_category_compare_and_swap_preserves_a_newer_whole_row(self) -> None: + conn = self.connection() + changed = [] + + class ChangeBeforeCategoryWrite: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql.startswith("UPDATE maintenance_layers SET error_category=") and not changed: + changed.append(True) + conn.execute("UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure'," + "run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors'") + return conn.execute(sql, *args) + + with patch.object(STYLE, "compute_style_vector", side_effect=ValueError("synthetic precompute")): + self.engine(ChangeBeforeCategoryWrite()).add("synthetic retained", sender="Synthetic") + self.assertEqual(changed, [True]) + state = self.state(conn) + self.assertEqual((state.outcome, state.error_category, state.run_generation, state.attempted_insert_count), + ("failed", "NewerFailure", "newer-owner", 999)) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + + def test_replayed_proof_never_counts_one_source_row_twice(self) -> None: + conn = self.connection() + conn.execute("BEGIN IMMEDIATE") + previous = MAINTENANCE._capture_style_append(conn) + self.assertEqual(self.append(conn, previous).outcome, "success") + before, checkpoint = self.rows(conn), self.checkpoint(conn) + result = self.append(conn, previous, message_id=1) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleCoverageChanged")) + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn), checkpoint) + conn.commit() + + def test_exact_append_rejects_correction_extra_insert_epoch_and_low_id_reuse(self) -> None: + for mode in ("correction", "extra", "epoch", "reuse", "wrong_id", "wrong_content", "counters", "nonappend"): + with self.subTest(mode=mode): + conn = self.connection() + self.engine(conn).add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + conn.execute("BEGIN IMMEDIATE") + previous = MAINTENANCE._capture_style_append(conn) + new_id = STORAGE.insert_message(conn, {"content": "synthetic ordinary", "sender": "Synthetic"}) + if mode == "correction": + conn.execute("UPDATE messages SET content='synthetic correction' WHERE id=1") + elif mode == "extra": + STORAGE.insert_message(conn, {"content": "synthetic extra", "sender": "Synthetic"}) + elif mode == "epoch": + conn.execute("UPDATE maintenance_source_state SET epoch='synthetic-new-epoch'") + elif mode == "reuse": + conn.execute("DELETE FROM messages WHERE id=1") + conn.execute("INSERT INTO messages(id,content,sender) VALUES (1,'synthetic replacement','Synthetic')") + elif mode == "counters": + conn.execute("UPDATE maintenance_source_state SET insert_count=insert_count+1") + elif mode == "nonappend": + conn.execute("UPDATE maintenance_source_state SET nonappend_revision=revision") + result = self.append(conn, previous, message_id=1 if mode == "wrong_id" else new_id, + content="synthetic different" if mode == "wrong_content" else "synthetic ordinary", + publish=lambda: self.fail("Unproven append reached updater")) + self.assertEqual(result.outcome, "deferred") + self.assertEqual(self.rows(conn), before) + conn.rollback() + + def test_unique_schema_and_insert_below_high_watermark_are_not_exact_appends(self) -> None: + for unique in (False, True): + conn = self.connection() + if unique: + conn.execute("CREATE UNIQUE INDEX synthetic_unique ON messages(content)") + conn.execute("BEGIN IMMEDIATE") + previous = MAINTENANCE._capture_style_append(conn) + conn.execute("INSERT INTO messages(id,content,sender) VALUES (?, 'synthetic ordinary','Synthetic')", + (1 if unique else 0,)) + result = self.append(conn, previous, message_id=1 if unique else 0, + publish=lambda: self.fail("Nonappend reached updater")) + self.assertEqual(result.error_category, "StyleAppendUnproven") + self.assertEqual(self.rows(conn), []) + conn.rollback() + + def test_newer_output_invalidation_survives_old_append_proof(self) -> None: + conn = self.connection() + conn.execute("BEGIN IMMEDIATE") + previous = MAINTENANCE._capture_style_append(conn) + new_id = STORAGE.insert_message(conn, {"content": "synthetic ordinary", "sender": "Synthetic"}) + conn.execute("INSERT INTO entity_style_vectors(entity,vector) VALUES ('other','[]')") + checkpoint = self.checkpoint(conn) + result = self.append(conn, previous, message_id=new_id, publish=lambda: self.fail("Stale proof")) + self.assertEqual(result.error_category, "StyleCoverageChanged") + self.assertEqual(self.checkpoint(conn), checkpoint) + conn.rollback() + + def test_running_worker_cannot_restore_over_newly_missed_foreground_append(self) -> None: + conn = self.connection() + before = self.checkpoint(conn) + with MAINTENANCE.maintenance_owner(None) as owner: + conn.execute("UPDATE maintenance_layers SET outcome='running',run_generation=? WHERE layer='style_vectors'", + (owner.generation,)) + conn.commit() + running = self.checkpoint(conn) + self.engine(conn).add("synthetic foreground append", sender="Synthetic") + invalidated = self.checkpoint(conn) + self.assertEqual(self.state(conn).outcome, "pending") + with self.assertRaises(MAINTENANCE.MaintenanceBusyError): + MAINTENANCE._restore_deferred_attempt(conn, "style_vectors", before, running, owner, + protect_rollback=True) + self.assertEqual(self.checkpoint(conn), invalidated) + self.assertEqual(MAINTENANCE.run_style_maintenance(conn).outcome, "success") + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone()[0], 1) + + def test_publication_error_rolls_back_profile_but_keeps_source_and_pending(self) -> None: + for error in (ValueError("synthetic failure"), sqlite3.OperationalError("synthetic SQL failure"), + sqlite3.OperationalError("database is locked")): + with self.subTest(error=type(error).__name__, locked=str(error) == "database is locked"): + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + publish = STYLE.update_entity_style_vector_incremental + + def failed(*args: object, **kwargs: object) -> None: + publish(*args, **kwargs) + raise error + + with patch.dict(ADD.__globals__, _update_style_vec=failed): + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + state = self.state(conn) + self.assertEqual((state.outcome, state.successful_coverage, state.attempted_revision), + ("pending", "unverified", None)) + self.assertEqual(state.error_category, "StyleWriterBusy" if str(error) == "database is locked" else type(error).__name__) + + def test_source_mutation_inside_updater_is_rolled_back_without_erasing_append(self) -> None: + conn = self.connection() + publish = STYLE.update_entity_style_vector_incremental + + def changed(*args: object, **kwargs: object) -> None: + publish(*args, **kwargs) + conn.execute("UPDATE messages SET content='synthetic unapproved correction' WHERE id=1") + + with patch.dict(ADD.__globals__, _update_style_vec=changed): + self.engine(conn).add("synthetic retained", sender="Synthetic") + self.assertEqual(conn.execute("SELECT content FROM messages").fetchone()[0], "synthetic retained") + self.assertEqual(self.rows(conn), []) + self.assertEqual(self.state(conn).error_category, "StyleSourceChanged") + + def test_provenance_failure_rolls_back_the_successful_raw_sum_append(self) -> None: + conn = self.connection() + engine = self.engine(conn) + engine.add("synthetic baseline", sender="Synthetic") + before = self.rows(conn) + record = MAINTENANCE.record_layer_success_in_transaction + + def failed(*args: object, **kwargs: object) -> None: + record(*args, **kwargs) + raise sqlite3.OperationalError("synthetic provenance failure") + + with patch.object(MAINTENANCE, "record_layer_success_in_transaction", failed): + engine.add("synthetic retained", sender="Synthetic") + self.assertEqual(self.rows(conn), before) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 2) + self.assertEqual((self.state(conn).outcome, self.state(conn).attempted_revision), ("pending", None)) + + def test_failed_pending_write_reports_failure_without_claiming_durable_invalidation(self) -> None: + conn = self.connection() + conn.execute("BEGIN IMMEDIATE") + previous = MAINTENANCE._capture_style_append(conn) + checkpoint = self.checkpoint(conn) + new_id = STORAGE.insert_message(conn, {"content": "synthetic ordinary", "sender": "Synthetic"}) + with patch.object(MAINTENANCE, "_STYLE_OUTPUT_INVALIDATE_SQL", "synthetic invalid SQL"): + result = MAINTENANCE._publish_style_append( + conn, previous, new_id, sender="Synthetic", content="synthetic ordinary", directive=False, + precomputed=False, publish=lambda: self.fail("No precomputed vector"), + ) + self.assertEqual((result.outcome, result.error_category, result.coverage, result.pending_caller_commit), + ("failed", "OperationalError", "unverified", True)) + self.assertEqual(self.checkpoint(conn), checkpoint) + state = self.state(conn) + self.assertEqual(MAINTENANCE.layer_freshness(state.source, state, state.dependency), "append_pending") + conn.commit() + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + + def test_read_capture_failure_preserves_source_and_does_not_claim_success(self) -> None: + unusual_error = type("Synthetic Invalid Category", (sqlite3.OperationalError,), {}) + for error in (sqlite3.OperationalError("database table is locked: maintenance_layers"), + sqlite3.OperationalError("synthetic capture failure"), unusual_error("synthetic detail")): + conn = self.connection() + read = MAINTENANCE.read_layer_states + failures = [] + + def unavailable(*args: object, **kwargs: object) -> object: + if not failures: + failures.append(True) + raise error + return read(*args, **kwargs) + + with patch.object(MAINTENANCE, "read_layer_states", unavailable): + self.engine(conn).add("synthetic retained", sender="Synthetic") + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertEqual(self.rows(conn), []) + self.assertEqual(self.state(conn).successful_coverage, "unverified") + self.assertRegex(self.state(conn).error_category, r"\A[A-Za-z][A-Za-z0-9_]{0,63}\Z") + + def test_borrowed_append_success_and_failure_remain_in_caller_transaction(self) -> None: + for failed in (False, True): + conn = self.connection() + before = self.checkpoint(conn) + conn.execute("INSERT INTO sentinel VALUES ('synthetic pending')") + updater = STYLE.update_entity_style_vector_incremental + with patch.dict(ADD.__globals__, _update_style_vec=(lambda *args, **kwargs: (_ for _ in ()).throw(ValueError("synthetic"))) + if failed else updater): + self.engine(conn).add("synthetic caller source", sender="Synthetic") + self.assertTrue(conn.in_transaction) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + conn.rollback() + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 0) + self.assertEqual(self.rows(conn), []) + self.assertEqual(self.checkpoint(conn), before) + + def test_cached_source_commit_and_caller_release_faults_restore_whole_add(self) -> None: + for cache in (0, 5, 100): + for borrowed in (False, True): + with self.subTest(cache=cache, borrowed=borrowed): + conn = self.connection(cache=cache) + if borrowed: + conn.execute("BEGIN") + self.engine(conn).add("synthetic warm release", sender="Synthetic") + conn.rollback() + before = self.checkpoint(conn) + source = MAINTENANCE.read_source_revision(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('synthetic caller')") + fault = BoundaryFailure(conn, "caller_release" if borrowed else "commit") + with self.assertRaises(sqlite3.DatabaseError): + self.engine(fault).add("synthetic rejected publication", sender="Synthetic") + self.assertEqual(fault.denied, 1) + self.assertEqual(conn.in_transaction, borrowed) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], int(borrowed)) + self.assertEqual(self.rows(conn), []) + self.assertEqual(self.checkpoint(conn), before) + self.assertEqual(MAINTENANCE.read_source_revision(conn), source) + + def test_style_savepoint_release_fault_preserves_add_and_invalidates_coverage(self) -> None: + for cache in (0, 5, 100): + conn = self.connection(cache=cache) + fault = BoundaryFailure(conn, "style_release") + self.engine(fault).add("synthetic retained", sender="Synthetic") + self.assertEqual(fault.denied, 1) + self.assertFalse(conn.in_transaction) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + self.assertEqual(self.rows(conn), []) + self.assertEqual((self.state(conn).outcome, self.state(conn).attempted_revision), ("pending", None)) + + def test_rejected_rollback_is_not_swallowed_as_successful_source_add(self) -> None: + conn = self.connection() + + class RefuseRollback: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql.startswith("ROLLBACK TO truememory_style_coverage_"): + raise sqlite3.OperationalError("synthetic rollback refused") + return conn.execute(sql, *args) + + with patch.dict(ADD.__globals__, _update_style_vec=lambda *args, **kwargs: (_ for _ in ()).throw(ValueError("synthetic"))): + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + self.engine(RefuseRollback()).add("synthetic failed source", sender="Synthetic") + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.rows(conn), []) + + def test_cleanup_release_failure_preserves_borrowed_caller_and_rejects_add(self) -> None: + conn = self.connection() + before = self.checkpoint(conn) + conn.execute("INSERT INTO sentinel VALUES ('synthetic caller')") + + class RefuseRelease: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql.startswith("RELEASE truememory_style_coverage_"): + raise sqlite3.OperationalError("synthetic release refused") + return conn.execute(sql, *args) + + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + self.engine(RefuseRelease()).add("synthetic rejected", sender="Synthetic") + self.assertTrue(conn.in_transaction) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone()[0], 1) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.rows(conn), []) + self.assertEqual(self.checkpoint(conn), before) + conn.rollback() + + def test_ended_transaction_cannot_be_reported_as_a_safe_style_failure(self) -> None: + conn = self.connection() + + def ended(*args: object, **kwargs: object) -> None: + conn.rollback() + raise sqlite3.OperationalError("database is locked") + + with patch.dict(ADD.__globals__, _update_style_vec=ended): + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + self.engine(conn).add("synthetic rejected", sender="Synthetic") + self.assertFalse(conn.in_transaction) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + + def test_add_queries_use_only_indexed_source_and_profile_point_reads(self) -> None: + conn = self.connection() + statements = [] + conn.set_trace_callback(statements.append) + self.engine(conn).add("synthetic bounded", sender="Synthetic") + conn.set_trace_callback(None) + reads = [sql for sql in statements if sql.startswith("SELECT") and + ("FROM messages " in sql or "FROM entity_style_vectors " in sql)] + self.assertEqual(len(reads), 3) + for sql in reads: + self.assertIn(" WHERE ", sql) + plans = conn.execute("EXPLAIN QUERY PLAN " + sql).fetchall() + self.assertTrue(all("SEARCH" in row[3] for row in plans), plans) + self.assertFalse(any("COUNT(" in sql.upper() or "MAX(" in sql.upper() for sql in statements if sql.startswith("SELECT"))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test-maintenance-foreground-admission-753.py b/tests/test-maintenance-foreground-admission-753.py index d295387e..e95b4fd8 100644 --- a/tests/test-maintenance-foreground-admission-753.py +++ b/tests/test-maintenance-foreground-admission-753.py @@ -52,6 +52,16 @@ class TestForegroundAdmission(AdmissionFixture): def test_style_computation_runs_outside_capture_and_publication_locks(self) -> None: engine = self.engine() engine._has_style_vec = True + maintenance = self.modules["maintenance"] + maintenance.prepare_style_maintenance(self.conn) + self.conn.execute("BEGIN IMMEDIATE") + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('style_vec_hash_version','2')") + maintenance.record_layer_success_in_transaction( + self.conn, layer="style_vectors", + dependency=maintenance.style_layer_spec(self.conn).resolve_dependency(), + source=maintenance.read_source_revision(self.conn), output_count=0, + ) + self.conn.commit() events = [] style = [0.25, 0.75] diff --git a/truememory/engine.py b/truememory/engine.py index 6411b587..3b963fe4 100644 --- a/truememory/engine.py +++ b/truememory/engine.py @@ -748,6 +748,9 @@ def add( _foreground_model_fence(pre_identity, blocking=not self.conn.in_transaction) if pre_embedding is not None else nullcontext() ), rebuild_transaction(self.conn, write=True): + if self._has_style_vec: + from truememory.maintenance import _capture_style_append, _publish_style_append + style_before = _capture_style_append(self.conn) msg = { "content": content, "sender": sender, @@ -794,12 +797,15 @@ def add( except Exception: logger.debug("Failed to update entity profile for %s during add()", sender, exc_info=True) - # Store pre-computed style vector (DB write only, computation happened outside lock) - if pre_style_vec is not None and not directive: - try: - _update_style_vec(self.conn, sender, content, _pre_computed_vec=pre_style_vec) - except Exception: - logger.debug("Failed to update style vector for %s during add()", sender, exc_info=True) + if self._has_style_vec: + style_result = _publish_style_append( + self.conn, style_before, new_id, sender=sender, content=content, directive=directive, + precomputed=pre_style_vec is not None, + publish=lambda: _update_style_vec(self.conn, sender, content, _pre_computed_vec=pre_style_vec), + ) + if style_result.error_category is not None: + logger.debug("Style append %s for message %s: %s", + style_result.outcome, new_id, style_result.error_category) self._maybe_auto_consolidate() diff --git a/truememory/maintenance.py b/truememory/maintenance.py index e0717b42..44376d14 100644 --- a/truememory/maintenance.py +++ b/truememory/maintenance.py @@ -24,7 +24,10 @@ from typing import NamedTuple from truememory._platform import try_file_lock -from truememory.storage import _prepare_style_maintenance, _style_output_tracking_ready, create_db +from truememory.storage import ( + _STYLE_OUTPUT_INVALIDATE_SQL, _STYLE_OUTPUT_TRIGGERS, + _prepare_style_maintenance, _style_output_tracking_ready, create_db, +) logger = logging.getLogger(__name__) @@ -46,7 +49,7 @@ class LayerDeferredError(RuntimeError): """A narrowly categorized transient condition establishes no attempt baseline.""" def __init__(self, message: str, *, category: str = "ModelBusy") -> None: - if category not in {"ModelBusy", "StyleCancelled", "StyleSourceChanged", "StyleWriterBusy"}: + if category not in {"ModelBusy", "StyleCancelled", "StyleSourceChanged", "StyleWriterBusy", "StyleTriggersUnsupported"}: raise ValueError("Unsupported layer deferral category") self.category = category super().__init__(message) @@ -842,11 +845,15 @@ def _borrowed_transaction( def _record_attempt( conn: sqlite3.Connection, state: LayerState, outcome: str, generation: str, error_category: str | None, *, borrowed: bool = False, -) -> None: +) -> bool | None: source, dependency = state.source, state.dependency if dependency.deferred: raise ValueError("Transient deferral cannot be recorded as a durable attempt") with (_borrowed_transaction(conn) if borrowed else _owned_transaction(conn)): + if state.layer == "style_vectors": + conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if _style_extra_triggers(conn): + return False conn.execute( "INSERT INTO maintenance_layers(layer,builder_version,outcome,attempted_epoch,attempted_revision," "attempted_dependency,attempted_insert_count,run_generation,error_category) VALUES (?,?,?,?,?,?,?,?,?) " @@ -887,6 +894,8 @@ def _restore_deferred_attempt( raise MaintenanceBusyError("Maintenance ownership changed before diagnostic restoration") with _owned_transaction(conn, on_rollback=(lambda: None) if protect_rollback else None): conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if layer == "style_vectors" and _style_extra_triggers(conn): + raise _StyleAppendChanged("StyleTriggersUnsupported") 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): @@ -903,6 +912,8 @@ def _restore_deferred_attempt( def _check_layer_dependency(current: LayerDependency, expected: LayerDependency) -> None: if current.deferred: + if current.error_category == "StyleTriggersUnsupported": + raise LayerDeferredError("Style triggers prevent publication", category="StyleTriggersUnsupported") raise LayerDeferredError("Embedding runtime is temporarily busy") if current != expected: raise MaintenanceUnavailableError("Layer dependency changed before publication") @@ -925,11 +936,10 @@ def run_layers( raise ValueError("Layer adapters belong to another connection") results = [] with maintenance_owner(connection_database_path(conn)) as owner: - def record_attempt(state: LayerState, outcome: str, category: str | None) -> None: + def record_attempt(state: LayerState, outcome: str, category: str | None) -> bool: if borrowed: - _record_attempt(conn, state, outcome, owner.generation, category, borrowed=True) - else: - _record_attempt(conn, state, outcome, owner.generation, category) + return _record_attempt(conn, state, outcome, owner.generation, category, borrowed=True) is not False + return _record_attempt(conn, state, outcome, owner.generation, category) is not False # Validate uniqueness before any builder is invoked. if _read_layer_run_baseline(conn, specs) is None: @@ -958,9 +968,11 @@ def record_attempt(state: LayerState, outcome: str, category: str | None) -> Non started = time.monotonic() baseline_state = state if not state.dependency.available: - 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, state.successful_coverage, borrowed)) + recorded = record_attempt(state, "unavailable", state.dependency.error_category) + results.append(LayerResult(spec.layer, spec.result_key, "unavailable" if recorded else "deferred", + state.output_count, time.monotonic() - started, + state.dependency.error_category if recorded else "StyleTriggersUnsupported", + recorded, state.successful_coverage, borrowed)) continue # This diagnostic commits before the read/compute snapshot starts. previous_row = running_row = None @@ -975,6 +987,10 @@ def confirm_diagnostic_rollback() -> None: with _owned_transaction(conn, on_rollback=( confirm_diagnostic_rollback if spec.layer == "style_vectors" else None )): + if spec.layer == "style_vectors": + conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if _style_extra_triggers(conn): + raise _StyleAppendChanged("StyleTriggersUnsupported") previous_row = _read_layer_row(conn, spec.layer) conn.execute( "INSERT INTO maintenance_layers(layer,builder_version,outcome,run_generation,error_category) " @@ -984,12 +1000,13 @@ def confirm_diagnostic_rollback() -> None: (spec.layer, state.dependency.builder_version, owner.generation), ) running_row = _read_layer_row(conn, spec.layer) - except sqlite3.OperationalError as error: + except (sqlite3.OperationalError, _StyleAppendChanged) as error: + category = error.category if isinstance(error, _StyleAppendChanged) else _style_sqlite_deferral(error) if (spec.layer != "style_vectors" or not diagnostic_rolled_back - or _style_sqlite_deferral(error) != "StyleWriterBusy"): + or category not in {"StyleWriterBusy", "StyleTriggersUnsupported"}): raise results.append(LayerResult(spec.layer, spec.result_key, "deferred", state.output_count, - time.monotonic() - started, "StyleWriterBusy", False, state.successful_coverage)) + time.monotonic() - started, category, False, state.successful_coverage)) continue count = None coverage = state.successful_coverage @@ -1015,7 +1032,7 @@ def confirm_rollback() -> None: output_started = True state = read_layer_states(conn, (spec,))[spec.layer] if state.dependency.deferred: - raise LayerDeferredError("Embedding runtime is temporarily busy") + _check_layer_dependency(state.dependency, state.dependency) if not state.dependency.available: raise MaintenanceUnavailableError("Layer dependency became unavailable") built = True @@ -1050,6 +1067,11 @@ def confirm_rollback() -> None: protect_rollback=True) else: _restore_deferred_attempt(conn, spec.layer, previous_row, running_row, owner) + except _StyleAppendChanged as restore_error: + if spec.layer != "style_vectors" or conn.in_transaction: + raise + error = LayerDeferredError("Style triggers prevent diagnostic restoration", + category=restore_error.category) except sqlite3.OperationalError as restore_error: if (spec.layer != "style_vectors" or conn.in_transaction or _style_sqlite_deferral(restore_error) != "StyleWriterBusy"): @@ -1068,7 +1090,8 @@ def confirm_rollback() -> None: 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(state, outcome, category) + if not record_attempt(state, outcome, category): + outcome, category = "deferred", "StyleTriggersUnsupported" results.append(LayerResult(spec.layer, spec.result_key, outcome, state.output_count, time.monotonic() - started, category, True, state.successful_coverage, borrowed)) if not isinstance(error, Exception): @@ -1088,11 +1111,250 @@ def prepare_style_maintenance( borrowed = conn.in_transaction try: read_source_revision(conn) + if _style_extra_triggers(conn): + raise _StyleAppendChanged("StyleTriggersUnsupported") except sqlite3.OperationalError as error: if _on_safe_failure is not None and conn.in_transaction == borrowed: _on_safe_failure(error) raise - _prepare_style_maintenance(conn, _on_safe_failure=_on_safe_failure) + class GuardedPreparation: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + cursor = conn.execute(sql, *args) + # Storage upgrades the writer with this zero-row statement before + # enrollment or invalidation. Its existing rollback boundary owns + # failures here, including a trigger installed after the first probe. + if sql == "UPDATE maintenance_layers SET layer=layer WHERE 0" and _style_extra_triggers(conn): + raise _StyleAppendChanged("StyleTriggersUnsupported") + return cursor + + _prepare_style_maintenance(GuardedPreparation(), _on_safe_failure=_on_safe_failure) + + +class _StyleAppendChanged(RuntimeError): + """A fixed metadata or checkpoint proof no longer permits publication.""" + + def __init__(self, category: str) -> None: + self.category = category + super().__init__(category) + + +def _style_extra_triggers(conn: sqlite3.Connection) -> bool: + names = tuple(_STYLE_OUTPUT_TRIGGERS) + main = conn.execute( + "SELECT 1 FROM main.sqlite_master WHERE type='trigger' " + "AND lower(tbl_name) IN ('entity_style_vectors','maintenance_layers','metadata') " + "AND NOT (lower(tbl_name)='entity_style_vectors' AND name IN (?,?,?)) LIMIT 1", names, + ).fetchone() + temporary = conn.execute( + "SELECT 1 FROM temp.sqlite_master WHERE type='trigger' " + "AND lower(tbl_name) IN ('entity_style_vectors','maintenance_layers','metadata') LIMIT 1", + ).fetchone() + return main is not None or temporary is not None + + +def _style_append_metadata(conn: sqlite3.Connection) -> tuple: + dependency = style_layer_spec(conn).resolve_dependency() + if dependency.error_category == "StyleTriggersUnsupported": + raise _StyleAppendChanged("StyleTriggersUnsupported") + if not dependency.available: + raise _StyleAppendChanged("StyleMetadataChanged") + schema = tuple(tuple(row) for row in conn.execute("PRAGMA main.table_info(entity_style_vectors)")) + default = next((row[4] for row in schema if row[1] == "accumulator_version"), "unsupported") + if default is not None and default.strip(" ()'\"") not in {"0", "1", "NULL", "null"}: + raise _StyleAppendChanged("StyleAccumulatorUnsupported") + marker = conn.execute("SELECT value FROM metadata WHERE key='style_vec_hash_version'").fetchone() + if marker is None or marker[0] != "2": + raise _StyleAppendChanged("StyleHashPending") + return dependency, schema, tuple(marker) + + +def _style_checkpoint_projection(row: tuple, **changes: object) -> tuple: + fields = dict(zip(_LAYER_ROW_COLUMNS, row)) + fields.update(changes) + return tuple(fields[name] for name in _LAYER_ROW_COLUMNS) + + +def _style_invalidated_checkpoint(row: tuple) -> tuple: + return _style_checkpoint_projection( + row, outcome="pending", full_rebuild_required=1, successful_coverage="unverified", + attempted_epoch=None, attempted_revision=None, attempted_dependency=None, + attempted_insert_count=None, run_generation=None, error_category=None, + ) + + +class _StyleAppendSnapshot(NamedTuple): + connection: sqlite3.Connection + checkpoint: tuple | None + state: LayerState | None + error_category: str | None + failed: bool = False + metadata: tuple | None = None + + +def _valid_style_source(source: SourceRevision) -> bool: + return (isinstance(source.epoch, str) and bool(source.epoch) + and all(type(value) is int and value >= 0 for value in source[1:]) + and source.revision == source.insert_count + source.correction_count + and source.nonappend_revision <= source.revision) + + +def _style_append_error_category(error: Exception) -> str: + category = type(error).__name__[:64] + return category if re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", category) is not None else "Error" + + +def _capture_style_append(conn: sqlite3.Connection) -> _StyleAppendSnapshot: + """Freeze metadata before one source insert, under its existing writer.""" + if not conn.in_transaction: + raise RuntimeError("Style append capture requires the source transaction") + checkpoint = None + state = None + try: + checkpoint = _read_layer_row(conn, "style_vectors") + spec = style_layer_spec(conn) + state = read_layer_states(conn, (spec,))[spec.layer] + if state.dependency.error_category == "StyleTriggersUnsupported": + return _StyleAppendSnapshot(conn, checkpoint, state, "StyleTriggersUnsupported") + if (checkpoint is None or checkpoint[1] != state.dependency.builder_version + or state.outcome not in {"success", "success_empty"} + or state.successful_coverage != "complete" + or layer_freshness(state.source, state, state.dependency) != "current" + or type(state.output_count) is not int or state.output_count < 0 + or (state.outcome == "success_empty") != (state.output_count == 0) + or not _valid_style_source(state.source) + or (state.attempted_epoch, state.attempted_revision, state.attempted_dependency, + state.attempted_insert_count) != (state.source.epoch, state.source.revision, + state.dependency.key, state.source.insert_count)): + return _StyleAppendSnapshot(conn, checkpoint, state, "StyleCoveragePending") + metadata = _style_append_metadata(conn) + except _StyleAppendChanged as error: + return _StyleAppendSnapshot(conn, checkpoint, state, error.category) + except (sqlite3.Error, MaintenanceUnavailableError, StopIteration) as error: + if not conn.in_transaction: + raise _LayerRollbackFailed("Style capture lost the source transaction") from error + category = _style_sqlite_deferral(error) if isinstance(error, sqlite3.OperationalError) else None + return _StyleAppendSnapshot(conn, checkpoint, state, category or _style_append_error_category(error), category is None) + return _StyleAppendSnapshot(conn, checkpoint, state, None, metadata=metadata) + + +@contextmanager +def _style_append_savepoint(conn: sqlite3.Connection) -> Iterator[None]: + """Keep source writes when style fails, only after complete savepoint cleanup.""" + name = "truememory_style_coverage_" + uuid.uuid4().hex + conn.execute(f"SAVEPOINT {name}") + try: + yield + conn.execute(f"RELEASE {name}") + except BaseException: + if not conn.in_transaction: + raise _LayerRollbackFailed("Style append lost the source transaction") + try: + conn.execute(f"ROLLBACK TO {name}") + conn.execute(f"RELEASE {name}") + except BaseException as error: + raise _LayerRollbackFailed("Style append rollback or release failed") from error + raise + + +def _publish_style_append( + conn: sqlite3.Connection, previous: _StyleAppendSnapshot, message_id: int, *, + sender: str, content: str, directive: bool, precomputed: bool, publish: Callable[[], None], +) -> LayerResult: + """Advance one proven append without committing or rebuilding for its caller.""" + if previous.connection is not conn or not conn.in_transaction: + raise RuntimeError("Style append publication requires its original source transaction") + state = previous.state + + def pending(category: str, *, failed: bool = False) -> LayerResult: + # Keep a prior failed attempt's retry baseline. A newly missed append + # supersedes successful/running provenance, never a newer invalidation. + if previous.checkpoint is not None and previous.checkpoint[2] in {"success", "success_empty", "running"}: + try: + with _style_append_savepoint(conn): + if _style_extra_triggers(conn): + category = "StyleTriggersUnsupported" + elif _read_layer_row(conn, "style_vectors") == previous.checkpoint: + conn.execute(_STYLE_OUTPUT_INVALIDATE_SQL) + expected = _style_invalidated_checkpoint(previous.checkpoint) + # The invalidation itself may invoke SQL triggers. Set + # its category only if every field still belongs to it. + if _read_layer_row(conn, "style_vectors") == expected: + predicate = " AND ".join(name + " IS ?" for name in _LAYER_ROW_COLUMNS) + conn.execute("UPDATE maintenance_layers SET error_category=? WHERE " + predicate, + (category, *expected)) + except _LayerRollbackFailed: + raise + except sqlite3.Error as error: + if not conn.in_transaction: + raise _LayerRollbackFailed("Style pending report lost the source transaction") from error + category, failed = _style_append_error_category(error), True + return LayerResult("style_vectors", "style_vectors", "failed" if failed else "deferred", + state.output_count if state is not None else None, 0.0, category, + False, "unverified", True) + + if previous.error_category is not None: + return pending(previous.error_category, failed=previous.failed) + try: + if _read_layer_row(conn, "style_vectors") != previous.checkpoint: + return pending("StyleCoverageChanged") + if _style_append_metadata(conn) != previous.metadata: + return pending("StyleMetadataChanged") + source = read_source_revision(conn) + before = state.source + if (not _valid_style_source(source) or source.epoch != before.epoch + or source.revision != before.revision + 1 or source.insert_count != before.insert_count + 1 + or source.correction_count != before.correction_count + or source.nonappend_revision != before.nonappend_revision + or type(message_id) is not int or message_id <= before.max_seen_message_id + or source.max_seen_message_id != message_id): + return pending("StyleAppendUnproven") + row = conn.execute("SELECT sender,content,directive FROM messages WHERE id=?", (message_id,)).fetchone() + if (not isinstance(sender, str) or not isinstance(content, str) or not content.strip() + or row is None or tuple(row) != (sender, content, int(bool(directive)))): + return pending("StyleAppendUnproven") + included = bool(sender) and not directive + if included and not precomputed: + return pending("StylePrecomputeFailed", failed=True) + with _style_append_savepoint(conn): + count = state.output_count + if included: + exists = conn.execute("SELECT 1 FROM entity_style_vectors WHERE entity=?", (sender.lower(),)).fetchone() + publish() + count += int(exists is None) + expected = _style_invalidated_checkpoint(previous.checkpoint) if included else previous.checkpoint + if _read_layer_row(conn, "style_vectors") != expected: + raise _StyleAppendChanged("StyleCoverageChanged") + if _style_append_metadata(conn) != previous.metadata: + raise _StyleAppendChanged("StyleMetadataChanged") + if read_source_revision(conn) != source: + raise LayerDeferredError("Style source changed during append", category="StyleSourceChanged") + record_layer_success_in_transaction(conn, layer="style_vectors", dependency=state.dependency, + source=source, output_count=count) + success = _style_checkpoint_projection( + previous.checkpoint, builder_version=state.dependency.builder_version, + outcome="success" if count else "success_empty", successful_epoch=source.epoch, + successful_revision=source.revision, successful_dependency=state.dependency.key, + successful_coverage="complete", attempted_epoch=source.epoch, attempted_revision=source.revision, + attempted_dependency=state.dependency.key, attempted_insert_count=source.insert_count, + full_rebuild_required=0, output_count=count, run_generation=None, error_category=None, + ) + if _read_layer_row(conn, "style_vectors") != success: + raise _StyleAppendChanged("StyleCoverageChanged") + if _style_append_metadata(conn) != previous.metadata or read_source_revision(conn) != source: + raise _StyleAppendChanged("StyleMetadataChanged") + except _LayerRollbackFailed: + raise + except Exception as error: + if not conn.in_transaction: + raise _LayerRollbackFailed("Style append lost the source transaction") from error + category = (error.category if isinstance(error, (LayerDeferredError, _StyleAppendChanged)) else + _style_sqlite_deferral(error) if isinstance(error, sqlite3.OperationalError) else None) + return pending(category or _style_append_error_category(error), failed=category is None) + return LayerResult("style_vectors", "style_vectors", "success" if count else "success_empty", + count, 0.0, None, True, "complete", True) def style_layer_spec( @@ -1107,6 +1369,9 @@ def style_layer_spec( } def resolve_dependency() -> LayerDependency: + if _style_extra_triggers(conn): + return make_layer_dependency(1, parameters, available=False, + error_category="StyleTriggersUnsupported")._replace(deferred=True) ready = _style_output_tracking_ready(conn) return make_layer_dependency(1, parameters, available=ready, error_category=None if ready else "StyleTrackingUnprepared") @@ -1179,6 +1444,8 @@ def build(connection: sqlite3.Connection) -> object: # The runner owns final commit. Upgrade before both format and source # validation; a stale reader cannot replace a newer generation. connection.execute("UPDATE messages SET sender=sender WHERE 0") + if _style_extra_triggers(connection): + raise LayerDeferredError("Style triggers prevent tracked publication", category="StyleTriggersUnsupported") style._publish_style_vectors(connection, schema, query, rows, stored_rows, check_cancel=check_cancel, before_replace=compatible) if read_source_revision(connection) != source: @@ -1216,6 +1483,9 @@ def preparation_failure(error: sqlite3.OperationalError) -> None: try: prepare_style_maintenance(conn, _on_safe_failure=preparation_failure) + except _StyleAppendChanged as error: + return LayerResult("style_vectors", "style_vectors", "deferred", None, + time.monotonic() - started, error.category, False, "unverified", borrowed) except LayerDeferredError as error: return LayerResult("style_vectors", "style_vectors", "deferred", None, time.monotonic() - started, error.category, False, "unverified", borrowed) From 157f45189663df8f6067e45a7b8dd2108238dee0 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 18:53:28 -0500 Subject: [PATCH 2/2] fix: Isolate test-owned workers and configuration reads --- tests/test_issue_502_config.py | 32 +++++++++++++++-- tests/test_issue_589_directives_completion.py | 36 ++++++++++++++++--- tests/test_platform_compat.py | 17 +++++++-- 3 files changed, 75 insertions(+), 10 deletions(-) diff --git a/tests/test_issue_502_config.py b/tests/test_issue_502_config.py index 9d03caf0..5bb1fa53 100644 --- a/tests/test_issue_502_config.py +++ b/tests/test_issue_502_config.py @@ -16,6 +16,7 @@ import json import os +from pathlib import Path import threading import time @@ -86,11 +87,14 @@ def test_load_config_returns_cached_copy(self, server, monkeypatch): # Second call within staleness window, file NOT modified on disk # (mtime unchanged) — should use cache, no disk read read_count = 0 - original_read_text = type(server._CONFIG_PATH).read_text + config_path = server._CONFIG_PATH + invoking_thread = threading.get_ident() + original_read_text = type(config_path).read_text - def counting_read(self_path, *args, **kwargs): + def counting_read(self_path: Path, *args: object, **kwargs: object) -> str: nonlocal read_count - read_count += 1 + if self_path == config_path and threading.get_ident() == invoking_thread: + read_count += 1 return original_read_text(self_path, *args, **kwargs) monkeypatch.setattr(type(server._CONFIG_PATH), "read_text", counting_read) @@ -99,6 +103,28 @@ def counting_read(self_path, *args, **kwargs): assert result1 == result2, ( "_load_config() did not return cached result within staleness window" ) + + # Prove that unrelated reads cannot contaminate this caller's count. + # Run controls after the second call so they cannot consume its TTL. + unrelated_path = config_path.with_name("unrelated.json") + unrelated_path.write_text(json.dumps(cfg), encoding="utf-8") + assert json.loads(unrelated_path.read_text(encoding="utf-8")) == cfg + background_results: list[str] = [] + background_errors: list[Exception] = [] + + def background_read() -> None: + try: + background_results.append(config_path.read_text(encoding="utf-8")) + except Exception as error: + background_errors.append(error) + + reader = threading.Thread(target=background_read) + reader.start() + reader.join(timeout=5) + assert not reader.is_alive() + assert not background_errors + assert len(background_results) == 1 + assert json.loads(background_results[0]) == cfg assert read_count == 0, ( f"_load_config() re-read disk {read_count} times within staleness window" ) diff --git a/tests/test_issue_589_directives_completion.py b/tests/test_issue_589_directives_completion.py index ff370c48..4c998b98 100644 --- a/tests/test_issue_589_directives_completion.py +++ b/tests/test_issue_589_directives_completion.py @@ -25,10 +25,29 @@ import logging import sqlite3 +from contextlib import ExitStack +from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest +if TYPE_CHECKING: + from truememory import Memory + + +def _close_and_drain_memory(memory: Memory) -> None: + coordinator = memory._engine._maintenance_coordinator + try: + try: + if coordinator is not None: + coordinator.cancel() + finally: + memory.close() + finally: + if coordinator is not None: + assert coordinator.wait(10), "Test-owned maintenance did not finish" + + # --------------------------------------------------------------------------- # v0.7.5.0 legacy schema (pre-directive) — extracted from the tagged release # --------------------------------------------------------------------------- @@ -253,13 +272,14 @@ def test_issue_589_directive_visible_regardless_of_user_scope(tmp_path): all_got = _load_directives(m) assert len(all_got) == 3 finally: - m.close() + _close_and_drain_memory(m) def test_issue_589_directive_injection_cap(tmp_path, caplog): """More directives than the cap -> only DIRECTIVE_LIMIT are injected, a warning is logged, and the block carries an overflow note pointing at truememory_directives.""" + from truememory import Memory from truememory.storage import create_db, insert_message db = tmp_path / "cap.db" @@ -273,8 +293,14 @@ def test_issue_589_directive_injection_cap(tmp_path, caplog): assert DIRECTIVE_LIMIT == 50 - with caplog.at_level(logging.WARNING): - ctx = recall_memories({}, db_path=str(db)) + with patch("truememory.Memory") as factory, ExitStack() as cleanup: + def tracked_memory(*args: object, **kwargs: object) -> Memory: + memory = Memory(*args, **kwargs) + cleanup.callback(_close_and_drain_memory, memory) + return memory + factory.side_effect = tracked_memory + with caplog.at_level(logging.WARNING): + ctx = recall_memories({}, db_path=str(db)) assert "" in ctx block = ctx.split("")[1].split("")[0] @@ -312,7 +338,7 @@ def _boom(*a, **kw): f"directive load failure must be logged, got: {warnings}" ) finally: - m.close() + _close_and_drain_memory(m) # --------------------------------------------------------------------------- @@ -493,7 +519,7 @@ def test_issue_589_directive_full_lifecycle(tmp_path): stats = m._engine.get_stats() assert stats.get("directive_count") == 0 finally: - m.close() + _close_and_drain_memory(m) # --------------------------------------------------------------------------- diff --git a/tests/test_platform_compat.py b/tests/test_platform_compat.py index 6cdfcd24..de48b509 100644 --- a/tests/test_platform_compat.py +++ b/tests/test_platform_compat.py @@ -85,6 +85,7 @@ def _tmp_config_dir(tmp_path, monkeypatch, ms): (home / ".truememory").mkdir() monkeypatch.setattr(ms, "_TRUEMEMORY_DIR", home / ".truememory") monkeypatch.setattr(ms, "_CONFIG_PATH", home / ".truememory" / "config.json") + monkeypatch.setattr(ms, "_CONFIG_LOCK_PATH", home / ".truememory" / "config.json.lock") return home @@ -158,24 +159,36 @@ def test_ingest_cli_save_warns_on_windows_when_api_key_present( tmp_path, monkeypatch, capsys ): from truememory.ingest import cli as ic - cfg_path = tmp_path / ".truememory" / "config.json" + import truememory.mcp_server as ms + home = _tmp_config_dir(tmp_path, monkeypatch, ms) + cfg_path = home / ".truememory" / "config.json" monkeypatch.setattr(ic, "_TRUEMEMORY_CONFIG_PATH", cfg_path) monkeypatch.setattr(ic.sys, "platform", "win32") + assert ic._TRUEMEMORY_CONFIG_PATH == ms._CONFIG_PATH == cfg_path + assert ms._TRUEMEMORY_DIR == cfg_path.parent + assert ms._CONFIG_LOCK_PATH == cfg_path.parent / "config.json.lock" ic._save_truememory_config({"tier": "pro", "openrouter_api_key": "sk-or-fake"}) captured = capsys.readouterr() assert "windows" in captured.err.lower() assert "environment variable" in captured.err.lower() + assert json.loads(cfg_path.read_text(encoding="utf-8")) == {"tier": "pro", "openrouter_api_key": "sk-or-fake"} def test_ingest_cli_save_silent_on_posix(tmp_path, monkeypatch, capsys): from truememory.ingest import cli as ic - cfg_path = tmp_path / ".truememory" / "config.json" + import truememory.mcp_server as ms + home = _tmp_config_dir(tmp_path, monkeypatch, ms) + cfg_path = home / ".truememory" / "config.json" monkeypatch.setattr(ic, "_TRUEMEMORY_CONFIG_PATH", cfg_path) monkeypatch.setattr(ic.sys, "platform", "linux") + assert ic._TRUEMEMORY_CONFIG_PATH == ms._CONFIG_PATH == cfg_path + assert ms._TRUEMEMORY_DIR == cfg_path.parent + assert ms._CONFIG_LOCK_PATH == cfg_path.parent / "config.json.lock" ic._save_truememory_config({"tier": "pro", "openrouter_api_key": "sk-or-fake"}) captured = capsys.readouterr() assert captured.err == "" + assert json.loads(cfg_path.read_text(encoding="utf-8")) == {"tier": "pro", "openrouter_api_key": "sk-or-fake"}