From dbdb436be33a50f0e99412e4f660525e42cea701 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 20:15:54 -0500 Subject: [PATCH 1/2] fix: Route style maintenance through coordinated ownership --- docs/resource-budgets.md | 115 ++ tests/test-l0-style-routing-747.py | 1415 +++++++++++++++++++ tests/test-rebuild-source-streaming-756.py | 5 +- tests/test_hardening.py | 2 +- tests/test_issue_590_build_vectors_batch.py | 1 + tests/test_issue_647_vector_rebuild.py | 1 + truememory/engine.py | 115 +- truememory/maintenance.py | 550 ++++++- truememory/storage.py | 3 +- 9 files changed, 2120 insertions(+), 87 deletions(-) create mode 100644 tests/test-l0-style-routing-747.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 056c548a..4e21bb4d 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -1352,3 +1352,118 @@ 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. + +### Style routing checkpoint + +Startup, cached-connection and committed-add notifications now plan private +style work independently of optional clustering. Style readiness reads only +fixed source/checkpoint metadata, schema declarations and trigger definitions. +It never reads source contents or profile values, loads a model, enrolls schema +or waits for the foreground engine lock. A compatible complete/current hash2 +checkpoint needs no work; an uncovered append uses the existing threshold of +25 committed inserts. Thus 24 uncovered inserts do not schedule a rebuild and +25 do. Corrections, missing provenance, wrong hash markers and supported owned +trigger repair are immediately eligible. A failed attempt at the same source +and dependency retains its retry baseline. The synthetic 75-add gate starts +from a tracked empty bootstrap and requires 0 additional full builds and +0 style worker requests. + +Failed and unavailable attempts retain that same 25-insert retry budget even +when hash2 or complete coverage is missing. At 0, 1 and 24 subsequent appends, +repeated notifications produce 0 worker opens and 0 integrity scans; at 25, +one retry advances the attempt baseline. Source corrections and changed +dependencies remain immediately eligible. This also bounds retries for an +unsupported future accumulator format. + +The coordinator retains one active worker and one immutable pending request. +Coalescing unions requested work kinds and retains the latest threshold; it +does not turn a busy or failed attempt into an automatic retry loop. Style-only +jobs open an existing canonical database using an escaped `mode=rw` URI. They +retain tuple rows, foreign keys, the 10000 ms busy timeout and existing local +cache/synchronization settings. They do not create a database, set its journal +mode, initialize general schema, load extensions or repair canonical source +tracking. Unsupported MAIN/TEMP triggers on style profiles, checkpoints or +metadata return `deferred` with `StyleTriggersUnsupported`, without enrollment, +diagnostic writes or repeated scheduling. Missing or incompatible canonical +source readiness is `unavailable`; normal schema initialization remains the +repair route. The existing public-layer and combined opener still uses its +general initialization contract. + +After safe metadata preflight, the style-only worker runs `quick_check(1)` +before style writes. This is a background database scan, not a bounded +foreground probe. Corrupt, failed or cancelled checks do not claim a successful +open. Only an exclusively owned connection receives its cancellation handler; +the handler is cleared before cleanup or style execution. Existing SQLite busy +waits can delay cancellation. The worker closes its handle before releasing +canonical ownership, including on setup and execution failure. + +The deprecated `open()` route now schedules guarded work instead of rebuilding +style synchronously or independently claiming hash2. Explicit consolidation +and bulk ingestion use the same private tracked route; bulk publishes style +once and preserves its existing result key. Successful manual and worker +observations are ordered within the process and published after dedicated +connection cleanup. Borrowed work remains pending caller commit, leaves its +original handle open and never clears earlier completed failure evidence. +Rollback restores prior output and provenance. Conditional hash/format +invalidation compares the entire observed checkpoint under writer ownership; +a newer row survives, and lock deferral requires confirmed rollback. + +Owned manual execution carries an unready foreground style observation into +its private report instead of bypassing TEMP-trigger restrictions on a new +connection. Public-layer work retains its existing initialization behavior. +Canonical style preparation validates source readiness under `BEGIN IMMEDIATE` +or an upgraded caller savepoint before enrollment. Diagnostic, restoration +and output paths revalidate at their explicit writer admission points, so a +currently missing or changed source trigger cannot publish certified output. +These checks preserve protected rollback and the standalone low-level contract. +An unstarted worker records its failure before releasing ownership; later +lifecycle cleanup cannot overwrite a newer manual observation. + +Public contracts remain eight maintenance layers and nine consolidation +result keys. Private style results affect aggregate status and appear separately +in `maintenance.style` health. This bounded observation does not connect or +schedule work. It distinguishes missing/unverified provenance, current complete +coverage, successful empty output, failures, unsupported readiness and pending +caller commits. Current database invalidation overrides historical success; +a genuinely committed current checkpoint can supersede historical failure. +Future declared accumulator formats remain unverified until guarded inspection; +health does not scan rows to infer per-row format compatibility. Mixed-case +message deletion normalizes only the style aggregate key to lowercase. + +Exact current source-trigger definitions establish structural readiness only. +They cannot detect a prior drop/change/recreate interval in which changes went +untracked. Matching definitions cannot promote missing or unverified provenance +or clear a known invalidation. These routes retain the existing uninterrupted +canonical-tracking assumption; they add no interval-attestation mechanism or +native performance/retrieval-quality claim. + +Canonical style no-op decisions validate source readiness in the same read +snapshot as their checkpoint and revision. Owned probes roll back only their +own read transaction; borrowed probes retain the caller's transaction. They +perform no source/profile scan, writer admission, rebuild or enrollment. A +trigger lost after the initial scheduling observation returns unverified +`StyleSourceChanged`, including at the per-layer and cancelled-result reads. +Read rollback failure remains an error. Worker observation ordering begins +before ownership binding, so a binding failure reports its category while +preserving a newer manual observation. + +Fixtures that exclusively test public consolidation or explicit vector rebuilds +disable their unrelated style capability before opening or adding. Production +style startup remains independent of the public consolidation flag. The public +threshold still requires an established durable attempt: 24 later inserts +produce no request, and the 25th makes the layer eligible. + +Canonical complete-coverage reads also validate hash2, the declared accumulator +format, exact owned output triggers and enrollment, and unsupported MAIN/TEMP +triggers in the checkpoint's read snapshot. TEMP tables or views that shadow +the five canonical source, checkpoint, metadata and output table names are +unsupported. These checks use schema and keyed metadata reads; they scan no +source or profile rows and acquire no writer. A readiness change after +preparation returns unverified coverage without a durable attempted baseline, +so the next wake can repair supported enrollment or perform hash migration. +Cancelled result reads use the same proof. Initial unverified migrations still +run normally. A real failed builder keeps its error and insert-count baseline; +if old complete coverage loses its metadata proof, only the returned coverage +is downgraded. Its 25-insert retry and explicit force behavior are preserved. +That downgrade survives the output-state reload if a retry fails. Successful +rebuilds publish their newly verified coverage. diff --git a/tests/test-l0-style-routing-747.py b/tests/test-l0-style-routing-747.py new file mode 100644 index 00000000..d20f1ec7 --- /dev/null +++ b/tests/test-l0-style-routing-747.py @@ -0,0 +1,1415 @@ +"""Actual private style routing with synthetic SQLite and stdlib-only loaders. + +Run the InMemoryRouting and CoordinatorRouting classes for local checks. +FileOpenerRouting requires the isolated full-suite host and synthetic files. +""" + +import os +import runpy +import sqlite3 +import tempfile +import threading +import types +import unittest +import uuid +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +BASE = runpy.run_path(str(Path(__file__).with_name("test-l0-style-incremental-747.py"))) +ROUTING = runpy.run_path(str(Path(__file__).with_name("test-maintenance-engine-routing-753.py"))) +STORAGE, STYLE, MAINTENANCE = BASE["STORAGE"], BASE["STYLE"], BASE["MAINTENANCE"] + + +def public_specs(conn: sqlite3.Connection, *_args: object, **_kwargs: object) -> tuple: + layers = ("clusters", "summaries", "contradictions", "structured_facts", "surprise", "episodes", "landmarks", "dunbar") + return tuple(MAINTENANCE.LayerSpec(layer, key, lambda: MAINTENANCE.make_layer_dependency(1), + lambda _conn: None, lambda _conn: 0, connection=conn) + for layer, key in zip(layers, MAINTENANCE._MAINTENANCE_RESULT_KEYS)) + + +class InMemoryRouting(unittest.TestCase): + def connection(self, *, bootstrap: bool = False) -> sqlite3.Connection: + conn = sqlite3.connect(":memory:") + 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) -> object: + modules = {"truememory.storage": STORAGE, "truememory.maintenance": MAINTENANCE, + "truememory.personality_style_vec": STYLE} + modules["truememory.rebuild_source"] = ROUTING["stdlib_module"]("rebuild_source", modules) + module = ROUTING["load_engine"](modules) + module._HAS_STYLE_VEC = True + module._update_style_vec = STYLE.update_entity_style_vector_incremental + engine = module.TrueMemoryEngine(":memory:") + engine.conn, engine.ready = conn, True + engine._has_consolidation = False + engine._has_style_vec = True + engine._maintenance_coordinator = MAINTENANCE.MaintenanceCoordinator(None) + engine._maintenance_handle = conn + engine._engine_test_module = module + return engine + + def add(self, conn: sqlite3.Connection, *, sender: str = "Synthetic") -> None: + STORAGE.insert_message(conn, {"content": "synthetic source", "sender": sender, "directive": False}) + conn.commit() + + def checkpoint(self, conn: sqlite3.Connection) -> tuple | None: + return MAINTENANCE._read_layer_row(conn, "style_vectors") + + def rows(self, conn: sqlite3.Connection) -> list: + return conn.execute("SELECT * FROM entity_style_vectors ORDER BY entity").fetchall() + + def no_native(self) -> object: + original = MAINTENANCE.importlib + return patch.object(MAINTENANCE, "importlib", types.SimpleNamespace(import_module=lambda name: + original.import_module(name) if name == "truememory.personality_style_vec" else + self.fail("Unexpected dependency preparation"))) + + def test_style_only_dispatch_precedes_and_excludes_public_preparation(self) -> None: + conn = self.connection() + self.add(conn) + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + with self.no_native(), patch.object(MAINTENANCE, "_prepare_worker_extensions", side_effect=AssertionError), \ + patch.object(MAINTENANCE, "engine_layer_specs", side_effect=AssertionError): + report = MAINTENANCE.run_engine_maintenance(conn, coordinator, include_layers=False, include_style=True) + self.assertEqual(report.results, ()) + self.assertTrue(report.preferences.startswith("SKIPPED")) + self.assertEqual(report.style_result.outcome, "success") + self.assertEqual(MAINTENANCE.maintenance_report_status(report), ("success", None)) + self.assertEqual(MAINTENANCE.observe_style(conn).health["freshness"], "current") + + def test_combined_style_publishes_before_unrelated_native_failure(self) -> None: + conn = self.connection() + self.add(conn) + with patch.object(MAINTENANCE, "_prepare_worker_extensions", side_effect=RuntimeError("synthetic extension")): + with self.assertRaises(RuntimeError): + MAINTENANCE.run_engine_maintenance(conn, MAINTENANCE.MaintenanceCoordinator(None), include_style=True) + self.assertEqual(MAINTENANCE.observe_style(conn).health["freshness"], "current") + self.assertEqual(len(self.rows(conn)), 1) + + def test_bounded_readiness_and_health_do_not_read_source_or_profiles(self) -> None: + conn = self.connection() + def authorize(action: int, table: str | None, *_args: object) -> int: + return sqlite3.SQLITE_DENY if action == sqlite3.SQLITE_READ and table in {"messages", "entity_style_vectors"} else sqlite3.SQLITE_OK + conn.set_authorizer(authorize) + before = conn.total_changes + with self.no_native(): + observation = MAINTENANCE.observe_style(conn) + health = self.engine(conn).get_style_health() + self.assertTrue(observation.eligible) + self.assertEqual(health["coverage"], "unverified") + self.assertEqual(conn.total_changes, before) + self.assertFalse(conn.in_transaction) + conn.set_authorizer(BASE["ACCUMULATOR"]["allow_authorized_action"]) + + def test_style_plans_with_consolidation_disabled_and_submits_outside_lock(self) -> None: + engine = self.engine(self.connection()) + calls = [] + def requested(**kwargs: object) -> bool: + self.assertFalse(engine._write_lock.locked()) + self.assertFalse(engine.conn.in_transaction) + calls.append(kwargs) + return False + with patch.object(engine._maintenance_coordinator, "request_layers", requested), \ + patch.object(MAINTENANCE, "engine_layer_specs", side_effect=AssertionError): + engine._maybe_auto_consolidate() + self.assertEqual(calls, [{"threshold": 25, "include_layers": False, "include_style": True}]) + + def test_75_actual_adds_make_no_style_request_or_full_rebuild(self) -> None: + conn = self.connection(bootstrap=True) + engine = self.engine(conn) + with patch.object(engine._maintenance_coordinator, "request_layers") as requested, \ + patch.object(STYLE, "_compute_entity_style_vectors", wraps=STYLE._compute_entity_style_vectors) as builds: + for index in range(75): + engine.add("synthetic append " + str(index), sender="Synthetic") + requested.assert_not_called() + builds.assert_not_called() + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone(), (75,)) + self.assertEqual(MAINTENANCE.observe_style(conn).state.attempted_insert_count, 75) + + def test_uncovered_append_threshold_is_24_then_25_and_corrections_are_immediate(self) -> None: + conn = self.connection(bootstrap=True) + for _ in range(24): + self.add(conn) + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.observe_style(conn).health["pending_reason"], "awaiting_threshold") + self.add(conn) + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "success") + conn.execute("UPDATE messages SET content='synthetic correction' WHERE id=1") + conn.commit() + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + + def test_hash_migration_and_owned_trigger_repair_make_current_rows_eligible(self) -> None: + for change in ("hash", "trigger", "coverage"): + with self.subTest(change=change): + conn = self.connection(bootstrap=True) + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif change == "trigger": + conn.execute("DROP TRIGGER truememory_style_output_update") + else: + conn.execute("UPDATE maintenance_layers SET successful_coverage='unverified' WHERE layer='style_vectors'") + conn.commit() + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "success_empty") + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.observe_style(conn).health["coverage"], "complete") + + def test_failed_hash_migration_keeps_attempt_baseline_and_suppresses_reopen_retry(self) -> None: + conn = self.connection(bootstrap=True) + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + conn.commit() + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=ValueError("synthetic failure")): + result = MAINTENANCE.run_routed_style(conn) + self.assertEqual(result.outcome, "failed") + before = self.checkpoint(conn) + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "untrusted") + self.assertEqual(self.checkpoint(conn), before) + + def test_failed_attempts_retry_at_25_appends_without_repeated_worker_opens(self) -> None: + for mode in ("initial", "current", "future"): + conn = self.connection(bootstrap=mode == "current") + self.add(conn) + if mode == "current": + MAINTENANCE.run_routed_style(conn, force=True) + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif mode == "future": + conn.execute("DROP TABLE entity_style_vectors") + 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 NOT NULL DEFAULT 9)") + conn.commit() + if mode == "future": + failed = MAINTENANCE.run_routed_style(conn) + else: + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=ValueError("synthetic build failure")): + failed = MAINTENANCE.run_routed_style(conn) + self.assertEqual(failed.outcome, "unavailable" if mode == "future" else "failed") + engine = self.engine(conn) + coordinator = MAINTENANCE.MaintenanceCoordinator(ROOT / "synthetic-worker-not-created.sqlite") + engine._maintenance_coordinator = coordinator + events = [] + class Opened: + row_factory = None + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql == "PRAGMA quick_check(1)": + events.append("check") + return conn.execute(sql, *args) + def close(self) -> None: + events.append("close") + class ImmediateThread: + def __init__(self, *, target: object, **_kwargs: object) -> None: + self.target = target + def start(self) -> None: + self.target() + def connect(*_args: object, **_kwargs: object) -> object: + events.append("open") + return Opened() + previous = 0 + with patch.object(MAINTENANCE.sqlite3, "connect", connect), \ + patch.object(MAINTENANCE.threading, "Thread", ImmediateThread), \ + patch.object(MAINTENANCE, "_acquire_owner", return_value=123), \ + patch.object(MAINTENANCE, "_release_owner"): + for count in (0, 1, 24, 25): + for _ in range(count - previous): + self.add(conn) + previous = count + with self.subTest(mode=mode, appends=count): + observed = MAINTENANCE.observe_style(conn) + if count < 25: + baseline = self.checkpoint(conn) + self.assertFalse(MAINTENANCE.run_routed_style(conn).attempted) + self.assertEqual(self.checkpoint(conn), baseline) + before_wakes = len(events) + for _ in range(3): + engine._maybe_auto_consolidate() + self.assertEqual(events[before_wakes:], ["open", "check", "close"] if count == 25 else []) + self.assertEqual(observed.eligible, count == 25) + self.assertEqual(observed.eligible, MAINTENANCE._eligible(observed.state, 25, False)) + if count == 25: + result = coordinator._last_report.style_result + self.assertTrue(result.attempted) + self.assertEqual(result.outcome, "unavailable" if mode == "future" else "success") + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + + def test_failed_attempt_correction_or_dependency_repair_retries_before_25(self) -> None: + for change in ("correction", "dependency"): + with self.subTest(change=change): + conn = self.connection() + self.add(conn) + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=ValueError("synthetic failure")): + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "failed") + if change == "correction": + conn.execute("UPDATE messages SET content='synthetic correction' WHERE id=1") + else: + conn.execute("DROP TRIGGER truememory_style_output_update") + conn.commit() + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "success") + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + + def test_future_default_remains_unverified_and_guarded_without_reopen_loop(self) -> None: + conn = self.connection(bootstrap=True) + conn.execute("DROP TABLE entity_style_vectors") + 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 NOT NULL DEFAULT 9)") + conn.commit() + MAINTENANCE.prepare_style_maintenance(conn) + observation = MAINTENANCE.observe_style(conn) + self.assertTrue(observation.eligible) + self.assertFalse(observation.health["format_ready"]) + self.assertEqual(observation.health["coverage"], "unverified") + schema = conn.execute("PRAGMA table_info(entity_style_vectors)").fetchall() + result = MAINTENANCE.run_routed_style(conn) + self.assertEqual((result.outcome, result.error_category), ("unavailable", "StyleAccumulatorUnsupported")) + self.assertEqual(conn.execute("PRAGMA table_info(entity_style_vectors)").fetchall(), schema) + self.assertEqual(self.rows(conn), []) + self.assertFalse(MAINTENANCE.observe_style(conn).eligible) + + def test_hash_invalidation_writer_contention_defers_after_confirmed_rollback(self) -> None: + for borrowed in (False, True): + with self.subTest(borrowed=borrowed): + original = self.connection(bootstrap=True) + original.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + original.commit() + uri = "file:synthetic-style-routing-" + uuid.uuid4().hex + "?mode=memory&cache=shared" + conn = sqlite3.connect(uri, uri=True, timeout=0) + contender = sqlite3.connect(uri, uri=True, timeout=0) + self.addCleanup(conn.close) + self.addCleanup(contender.close) + original.backup(conn) + before = self.checkpoint(conn) + if borrowed: + conn.execute("BEGIN") + observe = MAINTENANCE.observe_style + def contend(*args: object, **kwargs: object) -> object: + result = observe(*args, **kwargs) + contender.execute("BEGIN IMMEDIATE") + contender.execute("INSERT INTO sentinel VALUES ('synthetic writer')") + return result + with patch.object(MAINTENANCE, "observe_style", contend): + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleWriterBusy")) + self.assertEqual(conn.in_transaction, borrowed) + self.assertEqual(self.checkpoint(conn), before) + contender.rollback() + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual(result.outcome, "success_empty") + if borrowed: + conn.rollback() + self.assertEqual(self.checkpoint(conn), before) + + def test_hash_invalidation_preserves_newer_failure_instead_of_forced_overwrite(self) -> None: + for borrowed in (False, True): + with self.subTest(borrowed=borrowed): + conn = self.connection(bootstrap=True) + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + conn.commit() + if borrowed: + conn.execute("BEGIN") + expected = [] + observe = MAINTENANCE.observe_style + def replace(*args: object, **kwargs: object) -> object: + result = observe(*args, **kwargs) + conn.execute("UPDATE maintenance_layers SET outcome='failed',error_category='NewerFailure'," + "run_generation='newer-owner',attempted_insert_count=999 WHERE layer='style_vectors'") + if not borrowed: + conn.commit() + expected.append(self.checkpoint(conn)) + return result + with patch.object(MAINTENANCE, "observe_style", replace), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + result = MAINTENANCE.run_routed_style(conn, force=True, allow_caller_transaction=borrowed) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleSourceChanged")) + self.assertEqual(self.checkpoint(conn), expected[0]) + self.assertEqual(conn.in_transaction, borrowed) + + def test_hash_invalidation_does_not_hide_unknown_errors_or_lost_transactions(self) -> None: + for boundary in ("unknown", "ended", "rollback"): + with self.subTest(boundary=boundary): + conn = self.connection(bootstrap=True) + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + conn.commit() + before = self.checkpoint(conn) + class Fault: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + + def execute(self, sql: str, *args: object) -> sqlite3.Cursor: + if sql == "UPDATE maintenance_layers SET layer=layer WHERE 0": + if boundary == "ended": + conn.rollback() + raise sqlite3.OperationalError("synthetic unknown" if boundary == "unknown" else "database is locked") + return conn.execute(sql, *args) + + def rollback(self) -> None: + if boundary == "rollback": + raise sqlite3.OperationalError("synthetic rollback failure") + conn.rollback() + + expected = sqlite3.OperationalError if boundary == "unknown" else MAINTENANCE._LayerRollbackFailed + # The observation's read-only rollback must remain ordinary; + # inject only at the subsequent writer boundary. + with patch.object(MAINTENANCE, "observe_style", return_value=MAINTENANCE.observe_style(conn)): + with self.assertRaises(expected): + MAINTENANCE.run_routed_style(Fault()) + conn.rollback() + self.assertEqual(self.checkpoint(conn), before) + + def test_unsupported_triggers_do_not_schedule_or_mutate_on_any_guarded_table(self) -> None: + for table in ("entity_style_vectors", "maintenance_layers", "metadata"): + for temporary in (False, True): + with self.subTest(table=table, temporary=temporary): + conn = self.connection() + conn.execute("CREATE " + ("TEMP " if temporary else "") + "TRIGGER synthetic_unsupported AFTER INSERT ON main." + + table + " BEGIN INSERT INTO sentinel VALUES ('unexpected mutation'); END") + conn.commit() + engine = self.engine(conn) + before = conn.total_changes + with patch.object(engine._maintenance_coordinator, "request_layers") as requested: + engine._maybe_auto_consolidate() + requested.assert_not_called() + report = MAINTENANCE.run_engine_maintenance(conn, engine._maintenance_coordinator, + include_layers=False, include_style=True) + self.assertEqual(report.style_result.error_category, "StyleTriggersUnsupported") + self.assertEqual(report.style_result.outcome, "deferred") + self.assertEqual(conn.total_changes, before) + self.assertIsNone(self.checkpoint(conn)) + + def test_source_readiness_rejects_missing_changed_shadowed_and_unready_tracking(self) -> None: + for change in ("missing", "changed", "shadow", "unready", "token", "schema"): + with self.subTest(change=change): + conn = self.connection() + if change == "missing": + conn.execute("DROP TRIGGER messages_maintenance_ai") + elif change == "changed": + conn.execute("DROP TRIGGER messages_maintenance_ai") + conn.execute("CREATE TRIGGER messages_maintenance_ai AFTER INSERT ON messages BEGIN SELECT 1; END") + elif change == "shadow": + conn.execute("CREATE TEMP TRIGGER messages_maintenance_ai AFTER INSERT ON main.messages BEGIN SELECT 1; END") + elif change == "unready": + conn.execute("UPDATE maintenance_source_state SET tracking_ready=0") + elif change == "token": + conn.execute("DELETE FROM maintenance_source_state") + else: + conn.execute("ALTER TABLE messages RENAME COLUMN id TO synthetic_id") + conn.commit() + before = conn.total_changes + observation = MAINTENANCE.observe_style(conn) + self.assertFalse(observation.eligible) + self.assertEqual(observation.health["error_category"], "StyleSourceUnavailable") + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "unavailable") + self.assertEqual(conn.total_changes, before) + + def test_canonical_source_rechecked_under_preparation_and_diagnostic_writer(self) -> None: + for enrolled in (False, True): + for borrowed in (False, True): + with self.subTest(enrolled=enrolled, borrowed=borrowed): + conn = self.connection(bootstrap=enrolled) + self.add(conn) + if enrolled: + MAINTENANCE.run_routed_style(conn, force=True) + baseline, output = self.checkpoint(conn), self.rows(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + observe = MAINTENANCE.observe_style + def invalidate(*args: object, **kwargs: object) -> object: + result = observe(*args, **kwargs) + conn.execute("DROP TRIGGER messages_maintenance_ai") + if not borrowed: + conn.commit() + return result + before = conn.total_changes + with patch.object(MAINTENANCE, "observe_style", invalidate): + result = MAINTENANCE.run_routed_style(conn, force=True, allow_caller_transaction=borrowed) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleSourceChanged")) + self.assertEqual(conn.total_changes, before) + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(self.rows(conn), output) + self.assertEqual(conn.in_transaction, borrowed) + self.assertEqual(MAINTENANCE.observe_style(conn).health["error_category"], "StyleSourceUnavailable") + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + self.assertTrue(MAINTENANCE.observe_style(conn).health["available"] if enrolled else + MAINTENANCE.observe_style(conn).eligible) + + def test_canonical_source_rechecked_before_output_and_rollback_remains_required(self) -> None: + for boundary in ("publication", "rollback_failure"): + with self.subTest(boundary=boundary): + conn = self.connection(bootstrap=True) + self.add(conn) + MAINTENANCE.run_routed_style(conn, force=True) + baseline, output = self.checkpoint(conn), self.rows(conn) + compute = STYLE._compute_entity_style_vectors + def invalidate(*args: object, **kwargs: object) -> object: + result = compute(*args, **kwargs) + conn.execute("DROP TRIGGER messages_maintenance_ai") + return result + class Connection: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def rollback(self) -> None: + missing = conn.execute("SELECT 1 FROM sqlite_master WHERE name='messages_maintenance_ai'").fetchone() is None + if boundary == "rollback_failure" and missing: + raise sqlite3.OperationalError("synthetic rollback failure") + conn.rollback() + with patch.object(STYLE, "_compute_entity_style_vectors", invalidate): + if boundary == "rollback_failure": + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + MAINTENANCE.run_routed_style(Connection(), force=True) + conn.rollback() + else: + result = MAINTENANCE.run_routed_style(Connection(), force=True) + self.assertEqual((result.outcome, result.error_category), ("deferred", "StyleSourceChanged")) + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(self.rows(conn), output) + self.assertFalse(conn.in_transaction) + + def test_current_style_rechecks_source_in_every_baseline_snapshot(self) -> None: + for boundary in ("observation", "first_baseline", "cancelled_baseline"): + for borrowed in (False, True): + with self.subTest(boundary=boundary, borrowed=borrowed): + conn = self.connection(bootstrap=True) + self.add(conn) + MAINTENANCE.run_routed_style(conn, force=True) + baseline, output = self.checkpoint(conn), self.rows(conn) + self.assertEqual(MAINTENANCE.observe_style(conn).health["freshness"], "current") + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + cancel = threading.Event() + original = (MAINTENANCE.observe_style if boundary == "observation" else + MAINTENANCE._read_layer_run_baseline) + changed = False + def invalidate(*args: object, **kwargs: object) -> object: + nonlocal changed + result = original(*args, **kwargs) + if not changed: + changed = True + conn.execute("DROP TRIGGER messages_maintenance_ai") + if not borrowed: + conn.commit() + if boundary == "cancelled_baseline": + cancel.set() + return result + def read_only(action: int, table: str | None, *_args: object) -> int: + if action in {sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE} and table in { + "messages", "entity_style_vectors", "maintenance_layers", "metadata", + }: + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + attribute = "observe_style" if boundary == "observation" else "_read_layer_run_baseline" + before = conn.total_changes + conn.set_authorizer(read_only) + try: + with patch.object(MAINTENANCE, attribute, invalidate), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + result = MAINTENANCE.run_routed_style(conn, force=False, cancel=cancel, + allow_caller_transaction=borrowed) + finally: + conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertTrue(changed) + self.assertEqual((result.outcome, result.error_category, result.coverage, result.attempted), + ("deferred", "StyleSourceChanged", "unverified", False)) + self.assertEqual(conn.total_changes, before) + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(self.rows(conn), output) + self.assertEqual(conn.in_transaction, borrowed) + self.assertEqual(MAINTENANCE.observe_style(conn).health["error_category"], "StyleSourceUnavailable") + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + self.assertEqual(MAINTENANCE.observe_style(conn).health["freshness"], "current") + + def test_current_style_baseline_is_read_only_and_preserves_caller_snapshot(self) -> None: + for borrowed in (False, True): + with self.subTest(borrowed=borrowed): + conn = self.connection(bootstrap=True) + baseline = self.checkpoint(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + observed = [] + validate = MAINTENANCE._validate_routed_style_source + def in_snapshot(connection: sqlite3.Connection) -> None: + observed.append(connection.in_transaction) + validate(connection) + def read_only(action: int, table: str | None, *_args: object) -> int: + if action in {sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE}: + return sqlite3.SQLITE_DENY + if action == sqlite3.SQLITE_READ and table in {"messages", "entity_style_vectors"}: + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + before = conn.total_changes + conn.set_authorizer(read_only) + try: + with patch.object(MAINTENANCE, "_validate_routed_style_source", in_snapshot), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + finally: + conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual((result.outcome, result.coverage, result.attempted), ("current", "complete", False)) + self.assertTrue(observed) + self.assertTrue(all(observed)) + self.assertEqual(conn.total_changes, before) + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + + def test_late_complete_readiness_loss_defers_without_an_attempt_and_can_recover(self) -> None: + changes = {"hash": "StyleHashPending", "format": "StyleAccumulatorUnsupported", + "output_trigger": "StyleTrackingUnprepared", "checkpoint": "StyleTrackingUnprepared"} + changes.update({f"{schema}_{table}": "StyleTriggersUnsupported" for schema in ("main", "temp") + for table in ("entity_style_vectors", "maintenance_layers", "metadata")}) + for change, category in changes.items(): + boundaries = ("observation", "preparation", "first_baseline", "cancelled_baseline") if change in { + "hash", "format", + } else ("preparation", "first_baseline", "cancelled_baseline") + for boundary in boundaries: + for borrowed in (False, True): + with self.subTest(change=change, boundary=boundary, borrowed=borrowed): + conn = self.connection(bootstrap=True) + self.add(conn) + self.assertEqual(MAINTENANCE.run_routed_style(conn, force=True).outcome, "success") + baseline, output = self.checkpoint(conn), self.rows(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + cancel = threading.Event() + attribute = {"observation": "observe_style", "preparation": "_prepare_routed_style_maintenance"}.get( + boundary, "_read_layer_run_baseline") + original = getattr(MAINTENANCE, attribute) + changed = [] + def read_only(action: int, table: str | None, *_args: object) -> int: + if action in {sqlite3.SQLITE_INSERT, sqlite3.SQLITE_UPDATE, sqlite3.SQLITE_DELETE} and table in { + "messages", "entity_style_vectors", "maintenance_layers", "metadata", + }: + return sqlite3.SQLITE_DENY + if action == sqlite3.SQLITE_READ and table in {"messages", "entity_style_vectors"}: + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + def invalidate(*args: object, **kwargs: object) -> object: + result = original(*args, **kwargs) + if not changed: + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif change == "format": + conn.execute("ALTER TABLE entity_style_vectors DROP COLUMN accumulator_version") + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER DEFAULT 9") + elif change == "output_trigger": + conn.execute("DROP TRIGGER truememory_style_output_update") + elif change == "checkpoint": + conn.execute("DELETE FROM maintenance_layers WHERE layer='style_vectors'") + else: + schema, table = change.split("_", 1) + conn.execute(f"CREATE {'TEMP ' if schema == 'temp' else ''}TRIGGER extra_proof " + f"AFTER UPDATE ON main.{table} BEGIN SELECT 1; END") + if not borrowed: + conn.commit() + changed.append((self.checkpoint(conn), self.rows(conn), conn.total_changes)) + if boundary == "cancelled_baseline": + cancel.set() + conn.set_authorizer(read_only) + return result + try: + with patch.object(MAINTENANCE, attribute, invalidate), self.no_native(), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + result = MAINTENANCE.run_routed_style(conn, cancel=cancel, + allow_caller_transaction=borrowed) + finally: + conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertEqual(len(changed), 1) + self.assertEqual((result.outcome, result.error_category, result.coverage, result.attempted), + ("deferred", category, "unverified", False)) + self.assertEqual((self.checkpoint(conn), self.rows(conn), conn.total_changes), changed[0]) + self.assertEqual(conn.in_transaction, borrowed) + if change in {"hash", "output_trigger", "checkpoint"}: + if not borrowed: + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + repaired = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual((repaired.outcome, repaired.coverage), ("success", "complete")) + self.assertTrue(repaired.attempted) + elif change == "format": + failed = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual((failed.outcome, failed.error_category, failed.attempted), + ("unavailable", "StyleAccumulatorUnsupported", True)) + attempt = self.checkpoint(conn) + self.assertFalse(MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed).attempted) + self.assertEqual(self.checkpoint(conn), attempt) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + self.assertEqual((self.checkpoint(conn), self.rows(conn)), (baseline, output)) + + def test_failed_builder_with_old_complete_coverage_keeps_its_retry_budget(self) -> None: + for change, failure in ((change, failure) for change in ("hash", "format") + for failure in (ValueError, MAINTENANCE.LayerUnavailableError)): + for borrowed in (False, True): + with self.subTest(change=change, failure=failure.__name__, borrowed=borrowed): + conn = self.connection(bootstrap=True) + self.add(conn) + self.assertEqual(MAINTENANCE.run_routed_style(conn, force=True).outcome, "success") + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=failure("synthetic failure")): + self.assertEqual(MAINTENANCE.run_routed_style(conn, force=True).outcome, + "failed" if failure is ValueError else "unavailable") + baseline = self.checkpoint(conn) + self.assertEqual(baseline[MAINTENANCE._LAYER_ROW_COLUMNS.index("successful_coverage")], "complete") + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + else: + conn.execute("ALTER TABLE entity_style_vectors DROP COLUMN accumulator_version") + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER DEFAULT 9") + conn.commit() + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + previous = 0 + for count in (0, 1, 24, 25): + for _ in range(count - previous): + STORAGE.insert_message(conn, {"content": "synthetic append", "sender": "Synthetic"}) + if not borrowed: + conn.commit() + previous = count + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual(result.attempted, count == 25) + if count < 25: + self.assertEqual(result.coverage, "unverified") + self.assertEqual(result.error_category, failure.__name__) + self.assertEqual(self.checkpoint(conn), baseline) + else: + self.assertEqual(result.outcome, "success" if change == "hash" else "unavailable") + self.assertEqual(conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + self.assertEqual(self.checkpoint(conn), baseline) + + def test_retry_state_reload_preserves_known_unverified_failure_coverage(self) -> None: + coverage_index = MAINTENANCE._LAYER_ROW_COLUMNS.index("successful_coverage") + for change in ("intact", "hash", "format"): + for failure in (ValueError, MAINTENANCE.LayerUnavailableError): + for borrowed in (False, True): + for force in (False, True): + with self.subTest(change=change, failure=failure.__name__, borrowed=borrowed, force=force): + conn = self.connection(bootstrap=True) + self.add(conn) + self.assertEqual(MAINTENANCE.run_routed_style(conn, force=True).outcome, "success") + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=failure("synthetic failure")): + prior = MAINTENANCE.run_routed_style(conn, force=True) + self.assertEqual(prior.coverage, "complete") + baseline = self.checkpoint(conn) + if change == "hash": + conn.execute("UPDATE metadata SET value='1' WHERE key='style_vec_hash_version'") + elif change == "format": + conn.execute("ALTER TABLE entity_style_vectors DROP COLUMN accumulator_version") + conn.execute("ALTER TABLE entity_style_vectors ADD COLUMN accumulator_version INTEGER DEFAULT 9") + conn.commit() + output = self.rows(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + if not force: + for _ in range(25): + STORAGE.insert_message(conn, {"content": "synthetic append", "sender": "Synthetic"}) + if not borrowed: + conn.commit() + compute = STYLE._compute_entity_style_vectors + with patch.object(STYLE, "_compute_entity_style_vectors", wraps=compute, + side_effect=failure("synthetic failure") if change == "intact" else None): + result = MAINTENANCE.run_routed_style(conn, force=force, allow_caller_transaction=borrowed) + expected = ("success" if change == "hash" else "unavailable" if change == "format" + or failure is MAINTENANCE.LayerUnavailableError else "failed") + self.assertEqual((result.outcome, result.coverage, result.attempted), + (expected, "unverified" if change == "format" else "complete", True)) + self.assertEqual(self.checkpoint(conn)[coverage_index], "complete") + self.assertEqual(conn.in_transaction, borrowed) + if change != "hash": + self.assertEqual(self.rows(conn), output) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(self.rows(conn), output) + + def test_canonical_style_rejects_preexisting_and_late_temp_table_shadows(self) -> None: + for table in ("messages", "maintenance_source_state", "maintenance_layers", "metadata", "entity_style_vectors"): + for kind in ("TABLE", "VIEW"): + for late in (False, True): + for borrowed in (False, True): + with self.subTest(table=table, kind=kind, late=late, borrowed=borrowed): + conn = self.connection(bootstrap=True) + baseline = self.checkpoint(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + def shadow() -> None: + conn.execute(f"CREATE TEMP {kind} {table.upper()} AS SELECT * FROM main.{table}") + if not borrowed: + conn.commit() + prepare = MAINTENANCE._prepare_routed_style_maintenance + def after_prepare(*args: object, **kwargs: object) -> None: + prepare(*args, **kwargs) + shadow() + if not late: + shadow() + before = conn.total_changes + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError), self.no_native(): + if late: + with patch.object(MAINTENANCE, "_prepare_routed_style_maintenance", after_prepare): + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + else: + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual((result.error_category, result.coverage, result.attempted), + ("StyleSourceChanged" if late else "StyleSourceUnavailable", "unverified", False)) + self.assertEqual(conn.total_changes, before) + self.assertEqual(conn.execute("SELECT " + ",".join(MAINTENANCE._LAYER_ROW_COLUMNS) + + " FROM main.maintenance_layers WHERE layer='style_vectors'").fetchone(), baseline) + self.assertEqual(conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + + def test_canonical_style_rejects_case_variant_source_trigger_shadows(self) -> None: + for late in (False, True): + for borrowed in (False, True): + with self.subTest(late=late, borrowed=borrowed): + conn = self.connection(bootstrap=True) + baseline = self.checkpoint(conn) + if borrowed: + conn.execute("INSERT INTO sentinel VALUES ('caller pending')") + def shadow() -> None: + conn.execute("CREATE TEMP TRIGGER MESSAGES_MAINTENANCE_AI AFTER INSERT ON main.messages BEGIN SELECT 1; END") + if not borrowed: + conn.commit() + prepare = MAINTENANCE._prepare_routed_style_maintenance + def after_prepare(*args: object, **kwargs: object) -> None: + prepare(*args, **kwargs) + shadow() + if not late: + shadow() + before = conn.total_changes + with patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + if late: + with patch.object(MAINTENANCE, "_prepare_routed_style_maintenance", after_prepare): + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + else: + result = MAINTENANCE.run_routed_style(conn, allow_caller_transaction=borrowed) + self.assertEqual((result.error_category, result.coverage, result.attempted), + ("StyleSourceChanged" if late else "StyleSourceUnavailable", "unverified", False)) + self.assertEqual(conn.total_changes, before) + self.assertEqual(self.checkpoint(conn), baseline) + self.assertEqual(conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(conn.execute("SELECT * FROM sentinel").fetchall(), [("caller pending",)]) + conn.rollback() + + def test_current_style_read_rollback_failure_is_not_a_safe_deferral(self) -> None: + for failure in ("raises", "no_op"): + with self.subTest(failure=failure): + conn = self.connection(bootstrap=True) + baseline = self.checkpoint(conn) + armed = False + observe = MAINTENANCE.observe_style + def observed(*args: object, **kwargs: object) -> object: + nonlocal armed + result = observe(*args, **kwargs) + armed = True + return result + class Connection: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def rollback(self) -> None: + if armed: + if failure == "raises": + raise sqlite3.OperationalError("synthetic read rollback failure") + return + conn.rollback() + with patch.object(MAINTENANCE, "observe_style", observed): + with self.assertRaises(MAINTENANCE._LayerRollbackFailed): + MAINTENANCE.run_routed_style(Connection()) + self.assertTrue(conn.in_transaction) + conn.rollback() + self.assertEqual(self.checkpoint(conn), baseline) + + def test_in_memory_pending_and_caller_transaction_do_not_start_workers(self) -> None: + engine = self.engine(self.connection()) + with patch.object(MAINTENANCE.threading, "Thread", side_effect=AssertionError): + engine._maybe_auto_consolidate() + self.assertEqual(engine._maintenance_coordinator.status[0], "pending_in_memory") + engine.conn.execute("INSERT INTO sentinel VALUES ('caller')") + with patch.object(engine._maintenance_coordinator, "request_layers") as requested: + engine._maybe_auto_consolidate() + requested.assert_not_called() + self.assertEqual(engine.get_style_health()["pending_reason"], "pending_caller_commit") + engine.conn.rollback() + + def test_borrowed_success_rollback_preserves_failure_and_original_handle(self) -> None: + conn = self.connection(bootstrap=True) + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + failure = MAINTENANCE.LayerResult("style_vectors", "style_vectors", "failed", 0, 0.0, "SyntheticFailure", True, "unverified") + token = coordinator._begin_observation() + old_report = MAINTENANCE.MaintenanceReport((), "SKIPPED", failure) + coordinator._finish_observation(token, old_report, "failed", "SyntheticFailure") + before = self.checkpoint(conn) + conn.execute("INSERT INTO sentinel VALUES ('caller')") + with MAINTENANCE.maintenance_observation(coordinator, borrowed=True) as reports: + report = MAINTENANCE.run_engine_maintenance(conn, coordinator, force=True, include_layers=False, + include_style=True, allow_caller_transaction=True) + reports.append(report) + self.assertEqual(MAINTENANCE.maintenance_report_status(report)[0], "pending") + self.assertIs(coordinator._last_report, old_report) + self.assertEqual(MAINTENANCE.style_health(conn, coordinator)["last_error"], "SyntheticFailure") + self.assertTrue(conn.in_transaction) + conn.rollback() + self.assertEqual(self.checkpoint(conn), before) + self.assertEqual(coordinator.status, ("failed", "SyntheticFailure")) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone(), (0,)) + + def test_historical_success_cannot_hide_new_invalidation_and_committed_success_clears_health(self) -> None: + conn = self.connection(bootstrap=True) + self.add(conn) + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + with MAINTENANCE.maintenance_observation(coordinator) as reports: + reports.append(MAINTENANCE.run_engine_maintenance(conn, coordinator, force=True, include_layers=False, include_style=True)) + conn.execute("DELETE FROM entity_style_vectors") + conn.commit() + self.assertNotEqual(MAINTENANCE.style_health(conn, coordinator)["freshness"], "current") + token = coordinator._begin_observation() + coordinator._finish_observation(token, None, "failed", "SyntheticFailure") + self.assertEqual(MAINTENANCE.run_routed_style(conn).outcome, "success") + self.assertNotIn("last_error", MAINTENANCE.style_health(conn, coordinator)) + + def test_explicit_consolidate_preserves_nine_keys_and_original_connection(self) -> None: + conn = self.connection() + engine = self.engine(conn) + importer = types.SimpleNamespace(import_module=lambda name: STYLE if name == "truememory.personality_style_vec" + else types.SimpleNamespace(extract_preferences=lambda _conn: None)) + with patch.object(MAINTENANCE, "engine_layer_specs", public_specs), patch.object(MAINTENANCE, "importlib", importer), \ + patch.object(MAINTENANCE, "_installed_dependency", return_value=None): + result = engine.consolidate() + with patch.object(engine, "get_clustering_health", return_value={"status": "synthetic"}): + stats = engine.get_stats() + self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS)) + self.assertEqual(len(result), 9) + self.assertEqual(len(STORAGE._MAINTENANCE_LAYERS), 8) + self.assertEqual(len(stats["maintenance"]["layers"]), 8) + self.assertNotIn("style_vectors", stats["maintenance"]["layers"]) + self.assertEqual(stats["maintenance"]["style"]["outcome"], "success_empty") + self.assertIs(engine.conn, conn) + self.assertEqual(engine._maintenance_coordinator._last_report.style_result.outcome, "success_empty") + self.assertFalse(conn.in_transaction) + + def test_explicit_style_failure_changes_aggregate_and_private_health(self) -> None: + conn = self.connection() + engine = self.engine(conn) + importer = types.SimpleNamespace(import_module=lambda name: STYLE if name == "truememory.personality_style_vec" + else types.SimpleNamespace(extract_preferences=lambda _conn: None)) + with patch.object(MAINTENANCE, "engine_layer_specs", public_specs), patch.object(MAINTENANCE, "importlib", importer), \ + patch.object(MAINTENANCE, "_installed_dependency", return_value=None), \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=ValueError("synthetic failure")): + result = engine.consolidate() + self.assertEqual(len(result), 9) + self.assertEqual(engine._maintenance_coordinator.status, ("failed", "ValueError")) + self.assertEqual(engine.get_style_health()["outcome"], "failed") + + def test_owned_manual_preserves_foreground_temp_rejection_on_dedicated_handle(self) -> None: + conn = self.connection() + self.add(conn) + conn.execute("CREATE TEMP TRIGGER synthetic_foreground AFTER INSERT ON main.metadata BEGIN SELECT 1; END") + conn.commit() + engine = self.engine(conn) + coordinator = MAINTENANCE.MaintenanceCoordinator(ROOT / "synthetic-manual-not-created.sqlite") + engine._maintenance_coordinator = coordinator + dedicated = self.connection() + conn.backup(dedicated) + self.assertFalse(MAINTENANCE._style_extra_triggers(dedicated)) + events = [] + class Owned: + def __getattr__(self, name: str) -> object: + return getattr(dedicated, name) + def close(self) -> None: + events.append("close") + importer = types.SimpleNamespace(import_module=lambda name: STYLE if name == "truememory.personality_style_vec" + else types.SimpleNamespace(extract_preferences=lambda _conn: None)) + with patch.object(engine._engine_test_module, "create_db", return_value=Owned()), \ + patch.object(MAINTENANCE, "engine_layer_specs", public_specs), patch.object(MAINTENANCE, "importlib", importer), \ + patch.object(MAINTENANCE, "_prepare_worker_extensions", return_value=None), \ + patch.object(MAINTENANCE, "_installed_dependency", return_value=None), \ + patch.object(MAINTENANCE, "_acquire_owner", return_value=123), \ + patch.object(MAINTENANCE, "_release_owner", side_effect=lambda _fd: events.append("release")): + result = engine.consolidate() + self.assertEqual(len(result), 9) + self.assertEqual(len(coordinator._last_report.results), 8) + self.assertEqual(coordinator._last_report.style_result.error_category, "StyleTriggersUnsupported") + self.assertEqual(coordinator._last_report.style_result.outcome, "deferred") + self.assertEqual(events, ["close", "release"]) + self.assertEqual(self.rows(dedicated), []) + self.assertIsNone(self.checkpoint(dedicated)) + + def test_delete_normalizes_only_the_style_entity_key(self) -> None: + conn = self.connection(bootstrap=True) + engine = self.engine(conn) + engine.add("synthetic source", sender="Synthetic") + self.assertEqual(conn.execute("SELECT entity FROM entity_style_vectors").fetchone(), ("synthetic",)) + conn.executemany("INSERT INTO entity_profiles(entity) VALUES (?)", [("Synthetic",), ("synthetic",)]) + conn.executemany("INSERT INTO summaries(entity,summary) VALUES (?,'synthetic summary')", [("Synthetic",), ("synthetic",)]) + conn.commit() + namespace = STORAGE.__dict__ + original = namespace["__builtins__"]["__import__"] + def safe_import(name: str, *args: object, **kwargs: object) -> object: + if name == "truememory.tier_config": + return types.SimpleNamespace(VALID_TIER_GROUPS=()) + return original(name, *args, **kwargs) + with patch.dict(namespace["__builtins__"], __import__=safe_import): + self.assertTrue(STORAGE.delete_message(conn, 1)) + self.assertEqual(self.rows(conn), []) + self.assertEqual(conn.execute("SELECT entity FROM entity_profiles").fetchall(), [("synthetic",)]) + self.assertEqual(conn.execute("SELECT entity FROM summaries").fetchall(), [("synthetic",)]) + self.assertTrue(MAINTENANCE.observe_style(conn).eligible) + + def test_actual_bulk_style_stage_runs_once_and_reports_borrowed_success(self) -> None: + for borrowed in (False, True): + with self.subTest(borrowed=borrowed): + conn = self.connection() + engine = self.engine(conn) + module = engine._engine_test_module + module._HAS_CONSOLIDATION = False + def load(connection: sqlite3.Connection, _path: Path) -> int: + STORAGE.insert_message(connection, {"content": "synthetic bulk", "sender": "Synthetic", "directive": False}) + if not borrowed: + connection.commit() + return 1 + with patch.object(module, "create_db", return_value=conn), \ + patch.object(module, "load_messages_from_file", load), \ + patch.object(engine._maintenance_coordinator, "observe_clustering_outcome"), \ + patch.object(STYLE, "_compute_entity_style_vectors", wraps=STYLE._compute_entity_style_vectors) as builds: + stats = engine.ingest("synthetic-inline-boundary") + self.assertEqual(builds.call_count, 1) + self.assertTrue(stats["build_style_vectors"].startswith("1 vectors")) + self.assertEqual("pending caller commit" in stats["build_style_vectors"], borrowed) + if not borrowed: + engine.add("synthetic after bulk", sender="Synthetic") + self.assertEqual(builds.call_count, 1) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone(), (2,)) + if borrowed: + self.assertIsNone(engine._maintenance_coordinator._last_report) + conn.rollback() + self.assertIsNone(self.checkpoint(conn)) + self.assertEqual(self.rows(conn), []) + + def test_deprecated_open_schedules_without_synchronous_style_build_or_hash_claim(self) -> None: + conn = self.connection() + engine = self.engine(conn) + module = engine._engine_test_module + engine.db_path = (ROOT / "synthetic-existing-not-created.sqlite") + with patch.object(Path, "exists", return_value=True), \ + patch.object(module.sqlite3, "connect", return_value=conn), \ + patch.object(engine, "_purge_legacy_entity_profile_summaries"), \ + patch.object(engine, "_maybe_auto_consolidate") as planned, \ + patch.object(STYLE, "_compute_entity_style_vectors", side_effect=AssertionError): + with self.assertWarns(DeprecationWarning): + self.assertIs(engine.open(rebuild_vectors=False), engine) + planned.assert_called_once() + self.assertIsNone(conn.execute("SELECT value FROM metadata WHERE key='style_vec_hash_version'").fetchone()) + self.assertIsNone(self.checkpoint(conn)) + + def test_manual_pending_result_and_health_do_not_clear_prior_completed_failure(self) -> None: + conn = self.connection(bootstrap=True) + engine = self.engine(conn) + coordinator = engine._maintenance_coordinator + token = coordinator._begin_observation() + coordinator._finish_observation(token, MAINTENANCE.MaintenanceReport((), "SKIPPED"), "failed", "PriorFailure") + conn.execute("INSERT INTO sentinel VALUES ('caller')") + importer = types.SimpleNamespace(import_module=lambda name: STYLE if name == "truememory.personality_style_vec" + else types.SimpleNamespace(extract_preferences=lambda _conn: None)) + with patch.object(MAINTENANCE, "engine_layer_specs", public_specs), patch.object(MAINTENANCE, "importlib", importer), \ + patch.object(MAINTENANCE, "_installed_dependency", return_value=None): + result = engine.consolidate() + self.assertEqual(len(result), 9) + self.assertTrue(all("pending caller commit" in value for value in result.values())) + self.assertEqual(coordinator.status, ("failed", "PriorFailure")) + self.assertEqual(engine.get_style_health()["last_error"], "PriorFailure") + conn.rollback() + + def test_routed_owned_commit_and_borrowed_release_failures_keep_previous_output(self) -> None: + for borrowed in (False, True): + for cache in (0, 5, 100): + with self.subTest(borrowed=borrowed, cache=cache): + conn = sqlite3.connect(":memory:", cached_statements=cache) + self.addCleanup(conn.close) + conn.executescript(STORAGE._SCHEMA_SQL) + STORAGE._initialize_maintenance_tracking(conn) + conn.commit() + self.add(conn) + MAINTENANCE.run_routed_style(conn) + before, checkpoint = self.rows(conn), self.checkpoint(conn) + if borrowed: + conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-caller','pending')") + wrapper = BASE["BASE"]["PublicationFailure"](conn, borrowed=borrowed) + result = MAINTENANCE.run_routed_style(wrapper, force=True, allow_caller_transaction=borrowed) + self.assertEqual(result.outcome, "failed") + self.assertEqual(wrapper.denied, 1) + self.assertEqual(self.rows(conn), before) + self.assertEqual(self.checkpoint(conn)[3:7], checkpoint[3:7]) + self.assertEqual(conn.in_transaction, borrowed) + if borrowed: + self.assertEqual(conn.execute("SELECT value FROM metadata WHERE key='synthetic-caller'").fetchone(), ("pending",)) + conn.rollback() + self.assertEqual(self.checkpoint(conn), checkpoint) + + def test_cancellation_after_style_compute_prevents_public_stages(self) -> None: + conn = self.connection(bootstrap=True) + before = self.checkpoint(conn) + cancel = threading.Event() + compute = STYLE._compute_entity_style_vectors + def stop_after_compute(*args: object, **kwargs: object) -> object: + result = compute(*args, **kwargs) + cancel.set() + return result + with patch.object(STYLE, "_compute_entity_style_vectors", stop_after_compute), \ + patch.object(MAINTENANCE, "_prepare_worker_extensions", side_effect=AssertionError): + report = MAINTENANCE.run_engine_maintenance(conn, MAINTENANCE.MaintenanceCoordinator(None), + include_style=True, force=True, cancel=cancel) + self.assertEqual(report.results, ()) + self.assertEqual(report.style_result.error_category, "StyleCancelled") + self.assertEqual(self.checkpoint(conn), before) + self.assertFalse(conn.in_transaction) + + def test_health_does_not_connect_when_disconnected_or_foreground_busy(self) -> None: + engine = self.engine(self.connection()) + with patch.object(engine, "_ensure_connection", side_effect=AssertionError), \ + patch.object(engine, "_maybe_auto_consolidate", side_effect=AssertionError): + with engine._write_lock: + self.assertEqual(engine.get_style_health()["pending_reason"], "foreground_busy") + engine.conn = None + self.assertEqual(engine.get_style_health()["pending_reason"], "not_connected") + + def test_integrity_opener_faults_clear_handler_and_close_before_returning(self) -> None: + for failure in ("busy", "cancel", "io", "unknown", "corrupt", "empty", "config"): + with self.subTest(failure=failure): + conn = self.connection() + cancel = threading.Event() + events = [] + handlers = [] + class Connection: + row_factory = None + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def close(self) -> None: + events.append("close") + def set_progress_handler(self, callback: object, interval: int) -> None: + handlers.append(callback) + events.append("clear" if callback is None else "install") + def execute(self, sql: str, *args: object) -> object: + if failure == "config" and sql == "PRAGMA foreign_keys=ON": + raise sqlite3.OperationalError("synthetic configuration") + if sql == "PRAGMA quick_check(1)": + if failure == "cancel": + cancel.set() + self_check = handlers[-1]() + if self_check != 1: + raise AssertionError("Owned cancellation handler did not stop") + raise sqlite3.OperationalError("interrupted") + if failure in {"corrupt", "empty"}: + return types.SimpleNamespace(fetchone=lambda: None if failure == "empty" else ("synthetic corruption",)) + raise sqlite3.OperationalError({"busy": "database is locked", "io": "disk I/O error", + "unknown": "synthetic unexpected"}[failure]) + return conn.execute(sql, *args) + with patch.object(MAINTENANCE.sqlite3, "connect", return_value=Connection()): + expected = MAINTENANCE._StyleOpenRejected if failure in {"busy", "cancel"} else STORAGE.DatabaseOpenError + with self.assertRaises(expected) as raised: + MAINTENANCE._open_style_database((ROOT / "synthetic-not-created.sqlite"), cancel) + self.assertEqual(events[-1], "close") + if failure != "config": + self.assertEqual(events, ["install", "clear", "close"]) + if failure in {"busy", "cancel"}: + self.assertEqual(raised.exception.result.error_category, "StyleWriterBusy" if failure == "busy" else "StyleCancelled") + + def test_integrity_opener_preflight_forbids_general_setup_and_profile_mutation(self) -> None: + conn = self.connection(bootstrap=True) + self.add(conn) + MAINTENANCE.run_routed_style(conn, force=True) + conn.execute("DELETE FROM maintenance_layers WHERE layer='landmarks'") + conn.execute("CREATE TRIGGER synthetic_initializer BEFORE INSERT ON maintenance_layers BEGIN " + "UPDATE entity_style_vectors SET message_count=message_count+1; END") + conn.commit() + class Connection: + row_factory = None + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def close(self) -> None: + pass + def execute(self, sql: str, *args: object) -> object: + if sql == "PRAGMA quick_check(1)": + raise AssertionError("Unsupported opener cannot claim integrity checking") + return conn.execute(sql, *args) + before = conn.total_changes + with patch.object(MAINTENANCE.sqlite3, "connect", return_value=Connection()), \ + patch.object(MAINTENANCE, "create_db", side_effect=AssertionError): + with self.assertRaises(MAINTENANCE._StyleOpenRejected): + MAINTENANCE._open_style_database((ROOT / "synthetic-not-created.sqlite")) + self.assertEqual(conn.total_changes, before) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone(), (1,)) + # Actual general initialization is the negative control: all eight + # BEFORE INSERT triggers run, even for rows ignored by their conflict. + STORAGE._initialize_maintenance_tracking(conn) + self.assertEqual(conn.execute("SELECT message_count FROM entity_style_vectors").fetchone(), (9,)) + + +class CoordinatorRouting(unittest.TestCase): + connection = InMemoryRouting.connection + no_native = InMemoryRouting.no_native + + def test_started_worker_binding_failure_reports_error_and_preserves_newer_evidence(self) -> None: + for newer in (False, True): + with self.subTest(newer=newer): + coordinator = MAINTENANCE.MaintenanceCoordinator(ROOT / "synthetic-binding-not-created.sqlite") + old_report = MAINTENANCE.MaintenanceReport((), "SKIPPED previous") + newer_report = MAINTENANCE.MaintenanceReport((), "SKIPPED newer") + coordinator._last_report = old_report + def fail_identity() -> None: + if newer: + token = coordinator._begin_observation() + coordinator._finish_observation(token, newer_report, "failed", "NewerManual") + raise OSError("synthetic identity failure") + with patch.object(MAINTENANCE, "_acquire_owner", return_value=123), \ + patch.object(MAINTENANCE, "_release_owner") as released, \ + patch.object(MAINTENANCE.uuid, "uuid4", fail_identity), \ + patch.object(MAINTENANCE, "create_db", side_effect=AssertionError) as opened: + self.assertTrue(coordinator.request(lambda *_args: self.fail("Binding failure ran work"))) + self.assertTrue(coordinator.wait(2)) + opened.assert_not_called() + released.assert_called_once_with(123) + self.assertEqual(coordinator.status, ("failed", "NewerManual" if newer else "OSError")) + self.assertIs(coordinator._last_report, newer_report if newer else old_report) + self.assertFalse(coordinator._active) + self.assertIsNone(coordinator._thread) + + def test_mixed_notifications_union_flags_and_keep_latest_threshold(self) -> None: + for reverse in (False, True): + with self.subTest(reverse=reverse): + coordinator = MAINTENANCE.MaintenanceCoordinator((ROOT / "synthetic-not-created.sqlite")) + coordinator._active = True + requests = [(True, False), (False, True)] + if reverse: + requests.reverse() + for index, (layers, style) in enumerate(requests): + self.assertFalse(coordinator.request_layers(threshold=24 + index, include_layers=layers, include_style=style)) + self.assertEqual(coordinator._pending, MAINTENANCE.MaintenanceRequest(2, 25, True, True)) + with self.assertRaises(ValueError): + coordinator.request_layers(include_layers=False, include_style=False) + + def test_busy_admission_merges_consumed_kind_with_newer_pending_kind(self) -> None: + coordinator = MAINTENANCE.MaintenanceCoordinator((ROOT / "synthetic-not-created.sqlite")) + def busy(_path: Path, on_busy: object) -> None: + coordinator.request_layers(threshold=31, include_layers=False, include_style=True) + on_busy() + with patch.object(MAINTENANCE, "_acquire_owner", busy): + self.assertFalse(coordinator.request_layers(threshold=25)) + self.assertEqual(coordinator._pending, MAINTENANCE.MaintenanceRequest(2, 31, True, True)) + + def test_worker_start_failure_unions_real_new_wake_without_self_retry(self) -> None: + for newer in (False, True): + with self.subTest(newer=newer): + coordinator = MAINTENANCE.MaintenanceCoordinator((ROOT / "synthetic-not-created.sqlite")) + coordinator._active = True + def start() -> None: + if newer: + coordinator.request_layers(threshold=30, include_layers=False, include_style=True) + raise RuntimeError("synthetic start") + with patch.object(MAINTENANCE, "_acquire_owner", return_value=123), \ + patch.object(MAINTENANCE, "_release_owner") as released, \ + patch.object(MAINTENANCE.threading, "Thread", return_value=types.SimpleNamespace(start=start)), \ + patch.object(coordinator, "_launch_layers") as successor: + with self.assertRaises(RuntimeError): + coordinator._launch(lambda *_args: None, MAINTENANCE.MaintenanceRequest(0, 25)) + released.assert_called_once_with(123) + self.assertEqual(successor.call_count, int(newer)) + if newer: + self.assertEqual(successor.call_args.args[0], MAINTENANCE.MaintenanceRequest(1, 30, True, True)) + + def test_observation_tokens_reject_late_completion_and_borrowed_results(self) -> None: + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + one, two = coordinator._begin_observation(), coordinator._begin_observation() + newer = MAINTENANCE.MaintenanceReport((), "SKIPPED") + coordinator._finish_observation(two, newer, "failed", "NewerFailure") + coordinator._finish_observation(one, None, "success", None) + self.assertEqual(coordinator.status, ("failed", "NewerFailure")) + self.assertIs(coordinator._last_report, newer) + three = coordinator._begin_observation() + coordinator._finish_observation(three, MAINTENANCE.MaintenanceReport((), "SKIPPED"), "success", None, borrowed=True) + self.assertIs(coordinator._last_report, newer) + + def test_launched_opener_is_immutable_and_style_worker_never_initializes_general_schema(self) -> None: + coordinator = MAINTENANCE.MaintenanceCoordinator((ROOT / "synthetic-not-created.sqlite")) + with patch.object(coordinator, "_launch", return_value=True) as launch: + coordinator._launch_layers(MAINTENANCE.MaintenanceRequest(1, 25, False, True)) + work = launch.call_args.args[0] + self.assertEqual(launch.call_args.kwargs, {"style_only": True}) + coordinator._pending = MAINTENANCE.MaintenanceRequest(2, 30, True, False) + conn = self.connection() + events = [] + class Connection: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def close(self) -> None: + events.append("close") + with patch.object(MAINTENANCE, "_open_style_database", return_value=Connection()) as opener, \ + patch.object(MAINTENANCE, "create_db", side_effect=AssertionError), \ + patch.object(MAINTENANCE, "_release_owner", side_effect=lambda _fd: events.append("release")), \ + patch.object(coordinator, "_launch_layers") as successor, self.no_native(): + coordinator._run(123, work, style_only=True) + opener.assert_called_once() + self.assertEqual(events, ["close", "release"]) + self.assertEqual(coordinator._last_report.style_result.outcome, "success_empty") + successor.assert_called_once_with(MAINTENANCE.MaintenanceRequest(2, 30, True, False)) + + def test_late_worker_cleanup_cannot_replace_newer_manual_failure(self) -> None: + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + conn = self.connection() + events = [] + class Connection: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def close(self) -> None: + events.append("close") + token = coordinator._begin_observation() + coordinator._finish_observation(token, None, "failed", "NewerFailure") + with patch.object(MAINTENANCE, "create_db", return_value=Connection()), \ + patch.object(MAINTENANCE, "_release_owner", side_effect=lambda _fd: events.append("release")): + coordinator._run(123, lambda *_args: MAINTENANCE.MaintenanceReport((), "SKIPPED")) + self.assertEqual(events, ["close", "release"]) + self.assertEqual(coordinator.status, ("failed", "NewerFailure")) + + def test_setup_and_close_failures_retain_report_and_release_after_cleanup(self) -> None: + for boundary in ("setup", "close"): + with self.subTest(boundary=boundary): + coordinator = MAINTENANCE.MaintenanceCoordinator(None) + old = MAINTENANCE.MaintenanceReport((), "SKIPPED") + coordinator._last_report = old + conn = self.connection() + events = [] + class Connection: + def __getattr__(self, name: str) -> object: + return getattr(conn, name) + def close(self) -> None: + events.append("close") + raise sqlite3.OperationalError("synthetic close") + with patch.object(MAINTENANCE, "create_db", side_effect=RuntimeError("synthetic setup") if boundary == "setup" else None, + return_value=Connection()), \ + patch.object(MAINTENANCE, "_release_owner", side_effect=lambda _fd: events.append("release")): + coordinator._run(123, lambda *_args: MAINTENANCE.MaintenanceReport((), "SKIPPED new")) + self.assertIs(coordinator._last_report, old) + self.assertEqual(coordinator.status, ("failed", "RuntimeError" if boundary == "setup" else "OperationalError")) + self.assertEqual(events, ["release"] if boundary == "setup" else ["close", "release"]) + + def test_unstarted_failure_cannot_overwrite_manual_observation_after_owner_release(self) -> None: + coordinator = MAINTENANCE.MaintenanceCoordinator(ROOT / "synthetic-start-not-created.sqlite") + newer = MAINTENANCE.MaintenanceReport((), "SKIPPED newer") + class Unstarted: + def __init__(self, **_kwargs: object) -> None: + pass + def start(self) -> None: + raise RuntimeError("synthetic start failure") + def released(_fd: int) -> None: + token = coordinator._begin_observation() + coordinator._finish_observation(token, newer, "failed", "NewerManual") + with patch.object(MAINTENANCE, "_acquire_owner", return_value=123), \ + patch.object(MAINTENANCE, "_release_owner", released), \ + patch.object(MAINTENANCE.threading, "Thread", Unstarted): + with self.assertRaisesRegex(RuntimeError, "synthetic start failure"): + coordinator.request_layers(include_layers=False, include_style=True) + self.assertEqual(coordinator.status, ("failed", "NewerManual")) + self.assertIs(coordinator._last_report, newer) + self.assertFalse(coordinator._active) + self.assertTrue(coordinator.wait(0)) + + +class FileOpenerRouting(unittest.TestCase): + def setUp(self) -> None: + self.directory = tempfile.TemporaryDirectory(prefix="synthetic-style-routing-") + self.addCleanup(self.directory.cleanup) + self.path = Path(self.directory.name) / "synthetic.sqlite" + conn = STORAGE.create_db(self.path) + conn.close() + + def test_uri_filenames_settings_and_missing_file(self) -> None: + for suffix in ("hash#mark", "question?mark", "percent%mark", "space mark", "unicode-é"): + with self.subTest(suffix=suffix): + if os.name == "nt" and "?" in suffix: + self.skipTest("Question marks are not valid Windows filenames") + path = self.path.with_name(suffix + ".sqlite") + original = STORAGE.create_db(path) + journal = original.execute("PRAGMA journal_mode").fetchone() + original.close() + conn = MAINTENANCE._open_style_database(path) + try: + self.assertIsNone(conn.row_factory) + self.assertEqual(conn.execute("PRAGMA busy_timeout").fetchone(), (10000,)) + self.assertEqual(conn.execute("PRAGMA foreign_keys").fetchone(), (1,)) + self.assertEqual(conn.execute("PRAGMA journal_mode").fetchone(), journal) + self.assertEqual(MAINTENANCE.run_routed_style(conn, connection_owned=True).outcome, "success_empty") + finally: + conn.close() + missing = self.path.with_name("missing.sqlite") + with self.assertRaises(STORAGE.DatabaseOpenError): + MAINTENANCE._open_style_database(missing) + self.assertFalse(missing.exists()) + + def test_unsupported_guard_precedes_general_initializer_side_effects(self) -> None: + conn = sqlite3.connect(self.path) + conn.execute("CREATE TABLE sentinel(value TEXT)") + conn.execute("DELETE FROM maintenance_layers WHERE layer='landmarks'") + conn.execute("CREATE TRIGGER synthetic_extra BEFORE INSERT ON maintenance_layers BEGIN INSERT INTO sentinel VALUES ('unexpected'); END") + conn.commit() + before = conn.total_changes + with self.assertRaises(MAINTENANCE._StyleOpenRejected) as raised: + MAINTENANCE._open_style_database(self.path) + self.assertEqual(raised.exception.result.error_category, "StyleTriggersUnsupported") + self.assertEqual(conn.total_changes, before) + self.assertEqual(conn.execute("SELECT count(*) FROM sentinel").fetchone(), (0,)) + self.assertIsNone(conn.execute("SELECT 1 FROM maintenance_layers WHERE layer='landmarks'").fetchone()) + conn.close() + + def test_missing_source_readiness_and_corruption_never_enroll_style(self) -> None: + for change in ("missing", "changed", "unready", "token", "schema", "metadata"): + with self.subTest(change=change): + path = self.path.with_name("synthetic-" + change + ".sqlite") + conn = STORAGE.create_db(path) + if change in {"missing", "changed"}: + conn.execute("DROP TRIGGER messages_maintenance_ai") + if change == "changed": + conn.execute("CREATE TRIGGER messages_maintenance_ai AFTER INSERT ON messages BEGIN SELECT 1; END") + elif change == "unready": + conn.execute("UPDATE maintenance_source_state SET tracking_ready=0") + elif change == "token": + conn.execute("DELETE FROM maintenance_source_state") + elif change == "schema": + conn.execute("ALTER TABLE messages RENAME COLUMN id TO synthetic_id") + else: + conn.execute("DROP TABLE metadata") + conn.commit() + conn.close() + with self.assertRaises(MAINTENANCE._StyleOpenRejected) as raised: + MAINTENANCE._open_style_database(path) + self.assertEqual(raised.exception.result.error_category, "StyleSourceUnavailable") + conn = sqlite3.connect(path) + self.assertIsNone(conn.execute("SELECT 1 FROM maintenance_layers WHERE layer='style_vectors'").fetchone()) + self.assertEqual(conn.execute("SELECT count(*) FROM entity_style_vectors").fetchone(), (0,)) + conn.close() + damaged = self.path.with_name("synthetic-corrupt.sqlite") + damaged.write_bytes(b"synthetic invalid database") + with self.assertRaises(STORAGE.DatabaseOpenError): + MAINTENANCE._open_style_database(damaged) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test-rebuild-source-streaming-756.py b/tests/test-rebuild-source-streaming-756.py index 5ca9d5ff..bc39586a 100644 --- a/tests/test-rebuild-source-streaming-756.py +++ b/tests/test-rebuild-source-streaming-756.py @@ -928,7 +928,9 @@ def test_actual_deprecated_open_rebuilds_existing_legacy_connection(self): tree = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8")) original = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "TrueMemoryEngine") method = next(node for node in original.body if isinstance(node, ast.FunctionDef) and node.name == "open") - cls = ast.ClassDef(name="LegacyEngine", bases=[], keywords=[], body=[method], decorator_list=[]) + schedule = next(node for node in original.body + if isinstance(node, ast.FunctionDef) and node.name == "_maybe_auto_consolidate") + cls = ast.ClassDef(name="LegacyEngine", bases=[], keywords=[], body=[method, schedule], decorator_list=[]) self.vector._check_embedder_compatibility = lambda _conn: None self.vector.vectors_are_built = lambda _conn, _table: False namespace = dict(__builtins__=self.vector.__dict__["__builtins__"], @@ -941,6 +943,7 @@ def test_actual_deprecated_open_rebuilds_existing_legacy_connection(self): exec(compile(ast.fix_missing_locations(ast.Module(body=[cls], type_ignores=[])), "actual-legacy-open", "exec"), namespace) engine = namespace["LegacyEngine"]() engine.db_path, engine.stats = path, {} + engine._has_consolidation = engine._has_style_vec = False engine._purge_legacy_entity_profile_summaries = lambda: None fake_vec = types.ModuleType("sqlite_vec") fake_vec.load = lambda _conn: None diff --git a/tests/test_hardening.py b/tests/test_hardening.py index a8950826..c0fe657b 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -35,11 +35,11 @@ def _make_engine(n_messages=0, consolidation=False, vectors=False): eng = TrueMemoryEngine(db_path=db) eng._has_consolidation = False + eng._has_style_vec = False eng._ensure_connection() eng._has_consolidation = consolidation eng._has_vectors = vectors eng._has_hybrid = vectors - eng._has_style_vec = False return eng, td diff --git a/tests/test_issue_590_build_vectors_batch.py b/tests/test_issue_590_build_vectors_batch.py index 1f11dc2a..d3821993 100644 --- a/tests/test_issue_590_build_vectors_batch.py +++ b/tests/test_issue_590_build_vectors_batch.py @@ -75,6 +75,7 @@ def _seed_messages(ms, n: int) -> None: m = ms._get_memory() # These focused tests own explicit rebuild maintenance. m._engine._has_consolidation = False + m._engine._has_style_vec = False for i in range(n): m.add(f"memory number {i}", user_id="test") diff --git a/tests/test_issue_647_vector_rebuild.py b/tests/test_issue_647_vector_rebuild.py index a85b8530..bdc1db80 100644 --- a/tests/test_issue_647_vector_rebuild.py +++ b/tests/test_issue_647_vector_rebuild.py @@ -83,6 +83,7 @@ def _seed_messages(ms, n: int) -> None: m = ms._get_memory() # These focused tests own explicit rebuild maintenance. m._engine._has_consolidation = False + m._engine._has_style_vec = False for i in range(n): m.add(f"memory number {i}", user_id="test") diff --git a/truememory/engine.py b/truememory/engine.py index 3b963fe4..d2dd2907 100644 --- a/truememory/engine.py +++ b/truememory/engine.py @@ -246,7 +246,6 @@ def get_vectors_load_error() -> str | None: _HAS_STYLE_VEC = False try: from truememory.personality_style_vec import ( - build_entity_style_vectors, update_entity_style_vector_incremental as _update_style_vec, ) _HAS_STYLE_VEC = True @@ -837,31 +836,44 @@ def _get_maintenance_coordinator(self) -> MaintenanceCoordinator: def _maybe_auto_consolidate(self) -> None: """Probe fixed metadata without waiting on foreground or model work.""" - if not self._has_consolidation or self.conn is None: + if (not self._has_consolidation and not getattr(self, "_has_style_vec", False)) or self.conn is None: return if not self._write_lock.acquire(blocking=False): self._maintenance_pending_reason = "foreground_busy" return coordinator = None eligible = () + style_eligible = False try: if self.conn is None: return if self.conn.in_transaction: self._maintenance_pending_reason = "pending_caller_commit" return - from truememory.maintenance import engine_layer_specs, plan_layers + from truememory.maintenance import engine_layer_specs, observe_style, plan_layers coordinator = self._get_maintenance_coordinator() - eligible = plan_layers(self.conn, engine_layer_specs(self.conn, coordinator), - threshold=self._auto_consolidate_threshold) - self._maintenance_pending_reason = "eligible" if eligible else "awaiting_eligibility" + if getattr(self, "_has_style_vec", False): + observation = observe_style(self.conn, threshold=self._auto_consolidate_threshold) + style_eligible = observation.eligible + self._maintenance_pending_reason = observation.health["pending_reason"] + if self._has_consolidation: + eligible = plan_layers(self.conn, engine_layer_specs(self.conn, coordinator), + threshold=self._auto_consolidate_threshold) + if not getattr(self, "_has_style_vec", False): + self._maintenance_pending_reason = "awaiting_eligibility" + if eligible or style_eligible: + self._maintenance_pending_reason = "eligible" except Exception as error: self._maintenance_pending_reason = type(error).__name__[:64] finally: self._write_lock.release() - if eligible and coordinator is not None: + if (eligible or style_eligible) and coordinator is not None: try: - coordinator.request_layers(threshold=self._auto_consolidate_threshold) + if style_eligible: + coordinator.request_layers(threshold=self._auto_consolidate_threshold, + include_layers=bool(eligible), include_style=True) + else: + coordinator.request_layers(threshold=self._auto_consolidate_threshold) except Exception as error: self._maintenance_pending_reason = type(error).__name__[:64] @@ -1062,7 +1074,7 @@ def consolidate(self) -> dict[str, str]: """Force one pass, preserving caller transactions and shared ownership.""" from truememory.maintenance import ( MaintenanceBusyError, format_maintenance_report, maintenance_busy_result, - maintenance_owner, run_engine_maintenance, + maintenance_observation, observe_style, run_engine_maintenance, _style_unready_result, ) self._ensure_connection(_suppress_maintenance=True) with self._write_lock: @@ -1071,19 +1083,37 @@ def consolidate(self) -> dict[str, str]: borrowed = self.conn.in_transaction if borrowed or coordinator.path is None: try: - report = run_engine_maintenance(self.conn, coordinator, force=True, - threshold=self._auto_consolidate_threshold, prepare_extensions=False, - allow_caller_transaction=borrowed) + with maintenance_observation(coordinator, borrowed=borrowed) as observed: + report = run_engine_maintenance(self.conn, coordinator, force=True, + threshold=self._auto_consolidate_threshold, prepare_extensions=False, + allow_caller_transaction=borrowed, include_style=getattr(self, "_has_style_vec", False)) + if not borrowed and self.conn.in_transaction: + self.conn.rollback() + raise RuntimeError("Maintenance work left an unfinished transaction") + observed.append(report) except MaintenanceBusyError: return maintenance_busy_result() self._apply_manual_maintenance_capabilities(report.results) return format_maintenance_report(report) + foreground_style = None + if getattr(self, "_has_style_vec", False): + observation = observe_style(self.conn, threshold=self._auto_consolidate_threshold) + if observation.state is None: + foreground_style = _style_unready_result(observation.health) try: - with maintenance_owner(coordinator.path): + with maintenance_observation(coordinator) as observed: connection = create_db(coordinator.path) try: report = run_engine_maintenance(connection, coordinator, force=True, - threshold=self._auto_consolidate_threshold) + threshold=self._auto_consolidate_threshold, + include_style=getattr(self, "_has_style_vec", False) and foreground_style is None, + connection_owned=True) + if foreground_style is not None: + report = report._replace(style_result=foreground_style) + if connection.in_transaction: + connection.rollback() + raise RuntimeError("Maintenance work left an unfinished transaction") + observed.append(report) finally: connection.close() except MaintenanceBusyError: @@ -1259,27 +1289,6 @@ def open(self, rebuild_vectors: bool = True) -> "TrueMemoryEngine": # open() path behave identically. self._purge_legacy_entity_profile_summaries() - # Style vector hash migration: Python's hash() was replaced with a - # stable hashlib-based hash in v0.6.3. Existing style vectors were - # computed with the old non-deterministic hash and must be rebuilt. - if "entity_style_vectors" in tables and "metadata" in tables: - try: - row = self.conn.execute( - "SELECT value FROM metadata WHERE key = 'style_vec_hash_version'" - ).fetchone() - if row is None or row[0] != "2": - if _HAS_STYLE_VEC: - from truememory.personality_style_vec import build_entity_style_vectors - build_entity_style_vectors(self.conn) - logger.info("Style vectors rebuilt with stable hash (one-time migration)") - self.conn.execute( - "INSERT OR REPLACE INTO metadata (key, value) VALUES (?, ?)", - ("style_vec_hash_version", "2"), - ) - self.conn.commit() - except Exception: - logger.debug("Style vector migration failed", exc_info=True) - # Load sqlite-vec extension if available. # upgrade DEBUG → WARNING and track failure in a # module-level state so ``truememory_stats.health`` can report @@ -1366,6 +1375,7 @@ def open(self, rebuild_vectors: bool = True) -> "TrueMemoryEngine": self.stats["message_count"] = 0 self.ready = True + self._maybe_auto_consolidate() return self # ────────────────────────────────────────────────────────────────────── @@ -1478,14 +1488,27 @@ def ingest(self, data_path: str | Path) -> dict: # ── 4.5. Build entity style vectors (L0 char-n-gram) ───────────── if _HAS_STYLE_VEC: try: - t0 = time.time() - style_vecs = build_entity_style_vectors(self.conn) - stats["build_style_vectors"] = f"{len(style_vecs)} vectors in {time.time() - t0:.3f}s" + from truememory.maintenance import maintenance_observation, run_engine_maintenance + coordinator = self._get_maintenance_coordinator() + borrowed = self.conn.in_transaction + with maintenance_observation(coordinator, borrowed=borrowed) as observed: + report = run_engine_maintenance(self.conn, coordinator, force=True, prepare_extensions=False, + include_layers=False, include_style=True, allow_caller_transaction=borrowed) + if not borrowed and self.conn.in_transaction: + self.conn.rollback() + raise RuntimeError("Style work left an unfinished transaction") + observed.append(report) + result = report.style_result + if result.outcome in {"success", "success_empty"}: + stats["build_style_vectors"] = f"{result.output_count} vectors in {result.elapsed_seconds:.3f}s" + else: + stats["build_style_vectors"] = result.outcome.upper() + " (" + (result.error_category or "Pending") + ")" + if result.pending_caller_commit: + stats["build_style_vectors"] += " (pending caller commit)" self._has_style_vec = True except Exception as exc: - stats["build_style_vectors"] = f"ERROR: {exc}" + stats["build_style_vectors"] = "ERROR (" + type(exc).__name__[:64] + ")" logger.debug("build_style_vectors failed", exc_info=True) - self._has_style_vec = False else: stats["build_style_vectors"] = "SKIPPED (personality_style_vec module not available)" @@ -2515,6 +2538,17 @@ def get_clustering_health(self) -> dict: finally: self._write_lock.release() + def get_style_health(self) -> dict: + """Observe existing style coverage without connecting or scheduling.""" + from truememory.maintenance import style_health + if not self._write_lock.acquire(blocking=False): + return style_health(pending_reason="foreground_busy") + try: + coordinator = self._maintenance_coordinator if self._maintenance_handle is self.conn else None + return style_health(self.conn, coordinator) + finally: + self._write_lock.release() + def get_stats(self) -> dict: """Return ingestion and search statistics.""" self._ensure_connection() @@ -2543,6 +2577,7 @@ def get_stats(self) -> dict: stats["maintenance"] = {"status": "pending", "pending_reason": "foreground_busy"} stats["clustering"] = self.get_clustering_health() + stats["maintenance"]["style"] = self.get_style_health() # Add live DB stats if connected. if self.conn: diff --git a/truememory/maintenance.py b/truememory/maintenance.py index 44376d14..5b4f0987 100644 --- a/truememory/maintenance.py +++ b/truememory/maintenance.py @@ -25,7 +25,9 @@ from truememory._platform import try_file_lock from truememory.storage import ( + DEFAULT_BUSY_TIMEOUT_MS, DatabaseOpenError, _MAINTENANCE_SOURCE_FIELDS, _STYLE_OUTPUT_INVALIDATE_SQL, _STYLE_OUTPUT_TRIGGERS, + _maintenance_trigger_definitions, _messages_have_rowid_id, _prepare_style_maintenance, _style_output_tracking_ready, create_db, ) @@ -248,6 +250,20 @@ def _after_fork() -> None: ) +class MaintenanceRequest(NamedTuple): + generation: int + threshold: int + include_layers: bool = True + include_style: bool = False + + def merge(self, older: "MaintenanceRequest | None") -> "MaintenanceRequest": + if older is None: + return self + latest = self if self.generation >= older.generation else older + return latest._replace(include_layers=self.include_layers or older.include_layers, + include_style=self.include_style or older.include_style) + + class MaintenanceCoordinator: """One worker per canonical file, with ownership lasting through teardown.""" @@ -263,8 +279,9 @@ def __init__(self, path: Path | None) -> None: self._status = "pending" self._error_category: str | None = None self._notification_generation = 0 - self._pending: tuple[int, int] | None = None + self._pending: MaintenanceRequest | None = None self._last_report = None + self._observation_generation = 0 self._capability_epoch = 0 self._capability_versions: tuple[tuple[str, str | None], ...] | None = None self._extension_failure: LayerDependency | None = None @@ -288,6 +305,29 @@ def snapshot(self) -> dict: "capability_epoch": self._capability_epoch, "last_results": self._last_report.results if self._last_report is not None else ()} + def _begin_observation(self) -> int: + self._check_process() + with self._mutex: + self._observation_generation += 1 + return self._observation_generation + + def _finish_observation(self, token: int, report: "MaintenanceReport | None", outcome: str, + error_category: str | None, *, borrowed: bool = False) -> None: + self._check_process() + with self._mutex: + if token != self._observation_generation or borrowed: + return + self._status = outcome + self._error_category = _style_error_category(error_category) + if report is not None: + self._last_report = report + + def style_observation(self) -> tuple["LayerResult | None", str | None]: + self._check_process() + with self._mutex: + return (self._last_report.style_result if self._last_report is not None else None, + self._error_category) + def refresh_capabilities(self, *, reset_extension: bool = True) -> None: """Refresh three package records at open/manual setup, never per add.""" self._check_process() @@ -343,11 +383,14 @@ def _reserve_locked(self) -> None: self._status = "starting" self._error_category = None - def request_layers(self, *, threshold: int = 25) -> bool: + def request_layers(self, *, threshold: int = 25, include_layers: bool = True, + include_style: bool = False) -> bool: """Coalesce a real foreground wake; no callback captures its engine.""" self._check_process() if type(threshold) is not int or threshold < 1: raise ValueError("Maintenance threshold must be a positive integer") + if type(include_layers) is not bool or type(include_style) is not bool or not (include_layers or include_style): + raise ValueError("Maintenance request requires explicit work kinds") with self._mutex: if self.path is None: self._status = "pending_in_memory" @@ -355,20 +398,24 @@ def request_layers(self, *, threshold: int = 25) -> bool: if self._active and self._cancel.is_set(): return False self._notification_generation += 1 - self._pending = (self._notification_generation, threshold) + self._pending = MaintenanceRequest(self._notification_generation, threshold, + include_layers, include_style).merge(self._pending) pending = self._take_pending_locked() return self._launch_layers(pending) if pending is not None else False - def _take_pending_locked(self) -> tuple[int, int] | None: + def _take_pending_locked(self) -> MaintenanceRequest | None: if self._active or self._pending is None: return None pending, self._pending = self._pending, None self._reserve_locked() return pending - def _launch_layers(self, pending: tuple[int, int]) -> bool: + def _launch_layers(self, pending: MaintenanceRequest) -> bool: def work(conn: sqlite3.Connection, cancel: threading.Event) -> "MaintenanceReport": - return run_engine_maintenance(conn, self, threshold=pending[1], cancel=cancel) + return run_engine_maintenance(conn, self, threshold=pending.threshold, cancel=cancel, + include_layers=pending.include_layers, include_style=pending.include_style, connection_owned=True) + if not pending.include_layers: + return self._launch(work, pending, style_only=True) return self._launch(work, pending) def _owner_released(self) -> None: @@ -395,15 +442,15 @@ def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) - self._reserve_locked() return self._launch(work) - def _launch(self, work: Callable, pending: tuple[int, int] | None = None) -> bool: + def _launch(self, work: Callable, pending: MaintenanceRequest | None = None, *, style_only: bool = False) -> bool: def busy() -> None: # Registry -> coordinator ordering closes the release/admission # race. No coordinator-mutex holder acquires the registry lock. with self._mutex: self._active = False self._status = "cancelled" if self._cancel.is_set() else "busy" - if pending is not None and self._pending is None and not self._cancel.is_set(): - self._pending = pending + if pending is not None and not self._cancel.is_set(): + self._pending = pending.merge(self._pending) if self._pending is not None and self.path in _held_paths: _owner_waiters[self.path] = (_held_paths[self.path], self._pending[0], self) self._done.set() @@ -411,12 +458,15 @@ def busy() -> None: fd = None handoff = None handoff_lock = threading.Lock() + with self._mutex: + launch_token = self._observation_generation try: # Serializing descriptor acquisition with fork registration makes # every inherited owner descriptor visible to the child cleanup. fd = _acquire_owner(self.path, on_busy=busy) if fd is None: return False + launch_token = self._begin_observation() handoff = {"fd": fd, "started": False} def run() -> None: @@ -424,7 +474,10 @@ def run() -> None: handoff["started"] = True worker_fd = handoff.pop("fd", None) if worker_fd is not None: - self._run(worker_fd, work) + if style_only: + self._run(worker_fd, work, style_only=True) + else: + self._run(worker_fd, work) worker = threading.Thread(target=run, daemon=True, name="truememory-maintenance") with self._mutex: @@ -446,13 +499,14 @@ def run() -> None: self._pending = None self._cancel.set() raise + self._finish_observation(launch_token, None, "failed", "worker_start") if fd is not None: _release_owner(fd) with self._mutex: self._active = False self._thread = None - self._status = "failed" - self._error_category = "worker_start" + if self._pending is not None and pending is not None: + self._pending = self._pending.merge(pending) successor = self._take_pending_locked() if successor is None: self._done.set() @@ -463,15 +517,18 @@ def run() -> None: pass raise - def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], None]) -> None: + def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], None], *, style_only: bool = False) -> None: conn = None outcome = "cancelled" error_category = None + report = None + token = None try: + token = self._begin_observation() with _bind_owner(self.path): try: if not self._cancel.is_set(): - conn = create_db(self.path) + conn = _open_style_database(self.path, self._cancel) if style_only else create_db(self.path) report = work(conn, self._cancel) if conn.in_transaction: conn.rollback() @@ -479,8 +536,6 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No outcome = "cancelled" if self._cancel.is_set() else "success" if isinstance(report, MaintenanceReport): outcome, error_category = maintenance_report_status(report, self._cancel) - with self._mutex: - self._last_report = report finally: try: if conn is not None: @@ -490,15 +545,21 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No except BaseException as error: # Thread boundary: expose only a category, never source-bearing # exception text or a traceback through threading.excepthook. - outcome = "cancelled" if isinstance(error, (KeyboardInterrupt, SystemExit)) else "failed" - error_category = type(error).__name__[:64] + if isinstance(error, _StyleOpenRejected): + report = MaintenanceReport((), "SKIPPED (style-only)", error.result) + outcome, error_category = maintenance_report_status(report, self._cancel) + else: + outcome = "cancelled" if isinstance(error, (KeyboardInterrupt, SystemExit)) else "failed" + error_category = type(error).__name__[:64] + report = None finally: + if token is not None: + self._finish_observation(token, report if isinstance(report, MaintenanceReport) else None, + outcome, error_category) _release_owner(fd) with self._mutex: self._active = False self._thread = None - self._status = outcome - self._error_category = error_category if self._cancel.is_set(): self._pending = None pending = self._take_pending_locked() @@ -538,6 +599,23 @@ def get_coordinator(db_path: str | os.PathLike[str]) -> MaintenanceCoordinator: return coordinator +@contextmanager +def maintenance_observation(coordinator: MaintenanceCoordinator, *, borrowed: bool = False) -> Iterator[list]: + """Publish explicit work after caller cleanup and before releasing ownership.""" + with maintenance_owner(coordinator.path): + token = coordinator._begin_observation() + reports = [] + try: + yield reports + except BaseException as error: + coordinator._finish_observation(token, None, "failed", type(error).__name__, borrowed=borrowed) + raise + else: + report = reports[-1] if reports else None + outcome, category = maintenance_report_status(report) if report is not None else ("failed", "MissingReport") + coordinator._finish_observation(token, report, outcome, category, borrowed=borrowed) + + class LayerDependency(NamedTuple): key: str builder_version: int @@ -674,16 +752,53 @@ def read_layer_states(conn: sqlite3.Connection, specs: tuple[LayerSpec, ...]) -> def _read_layer_run_baseline( - conn: sqlite3.Connection, specs: tuple[LayerSpec, ...], + conn: sqlite3.Connection, specs: tuple[LayerSpec, ...], *, _canonical_source: bool = False, ) -> dict[str, LayerState] | None: """Return no style observation only for a read lock with unchanged ownership. This boundary is used before running diagnostics and for cancelled reads, never around builders or publication transactions. + Canonical style readiness and checkpoint values share one read snapshot. """ borrowed = conn.in_transaction try: - return read_layer_states(conn, specs) + if not _canonical_source: + return read_layer_states(conn, specs) + if not borrowed: + conn.execute("BEGIN") + try: + _validate_routed_style_source(conn) + states = read_layer_states(conn, specs) + state = states.get("style_vectors") + if state is not None: + # Canonical preparation already established enrollment. A lost + # proof here is repairable, not a completed unavailable attempt. + if state.dependency.deferred or not state.dependency.available: + raise _StyleAppendChanged(state.dependency.error_category or "StyleMetadataChanged") + if state.successful_coverage == "complete": + try: + dependency, _schema, _marker = _style_append_metadata(conn) + if dependency != state.dependency: + raise _StyleAppendChanged("StyleMetadataChanged") + except _StyleAppendChanged: + if state.outcome not in {"failed", "unavailable"}: + raise + # Keep a real builder's durable retry baseline. Its old + # coverage no longer certifies this read's metadata. + states[state.layer] = state._replace(successful_coverage="unverified") + if not conn.in_transaction: + raise _LayerRollbackFailed("Canonical style read lost its transaction") + return states + finally: + if not borrowed: + if not conn.in_transaction: + raise _LayerRollbackFailed("Canonical style read ended without rollback") + try: + conn.rollback() + except sqlite3.Error as error: + raise _LayerRollbackFailed("Canonical style read rollback failed") from error + if conn.in_transaction: + raise _LayerRollbackFailed("Canonical style read rollback left a transaction") except sqlite3.OperationalError as error: if (len(specs) == 1 and specs[0].layer == "style_vectors" and conn.in_transaction == borrowed and _style_sqlite_deferral(error) == "StyleWriterBusy"): @@ -784,11 +899,11 @@ def record_layer_success_in_transaction( @contextmanager def _owned_transaction( - conn: sqlite3.Connection, *, on_rollback: Callable[[], None] | None = None, + conn: sqlite3.Connection, *, on_rollback: Callable[[], None] | None = None, immediate: bool = False, ) -> Iterator[None]: if conn.in_transaction: raise RuntimeError("Maintenance runner requires a clean transaction boundary") - conn.execute("BEGIN") + conn.execute("BEGIN IMMEDIATE" if immediate else "BEGIN") completed = False automatic_rollback = False try: @@ -844,14 +959,17 @@ def _borrowed_transaction( def _record_attempt( conn: sqlite3.Connection, state: LayerState, outcome: str, generation: str, error_category: str | None, - *, borrowed: bool = False, + *, borrowed: bool = False, canonical_source: bool = False, ) -> 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)): + transaction = _borrowed_transaction if borrowed else _owned_transaction + with (transaction(conn, on_rollback=lambda: None) if canonical_source else transaction(conn)): if state.layer == "style_vectors": conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if canonical_source: + _validate_routed_style_source(conn) if _style_extra_triggers(conn): return False conn.execute( @@ -884,7 +1002,7 @@ def _read_layer_row(conn: sqlite3.Connection, layer: str) -> tuple | None: def _restore_deferred_attempt( conn: sqlite3.Connection, layer: str, previous: tuple | None, running: tuple, - owner: MaintenanceOwnership, *, protect_rollback: bool = False, + owner: MaintenanceOwnership, *, protect_rollback: bool = False, canonical_source: bool = False, ) -> None: """Restore only this owner's unchanged diagnostic, after output rollback.""" if conn.in_transaction: @@ -894,6 +1012,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 canonical_source: + _validate_routed_style_source(conn) if layer == "style_vectors" and _style_extra_triggers(conn): raise _StyleAppendChanged("StyleTriggersUnsupported") current = _read_layer_row(conn, layer) @@ -922,7 +1042,7 @@ def _check_layer_dependency(current: LayerDependency, expected: LayerDependency) 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, + allow_caller_transaction: bool = False, _canonical_source: bool = False, ) -> tuple[LayerResult, ...]: """Run eligible adapters once; borrowing requires explicit caller opt-in.""" if type(allow_caller_transaction) is not bool: @@ -937,17 +1057,21 @@ def run_layers( results = [] with maintenance_owner(connection_database_path(conn)) as owner: def record_attempt(state: LayerState, outcome: str, category: str | None) -> bool: + if _canonical_source: + return _record_attempt(conn, state, outcome, owner.generation, category, + borrowed=borrowed, canonical_source=True) is not False if borrowed: 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: + baseline_options = {"_canonical_source": True} if _canonical_source else {} + if _read_layer_run_baseline(conn, specs, **baseline_options) is None: return (_style_baseline_busy_result(specs[0], borrowed),) for spec in specs: if cancel is not None and cancel.is_set(): break - observed = _read_layer_run_baseline(conn, (spec,)) + observed = _read_layer_run_baseline(conn, (spec,), **baseline_options) if observed is None: results.append(_style_baseline_busy_result(spec, borrowed)) continue @@ -989,6 +1113,8 @@ def confirm_diagnostic_rollback() -> None: )): if spec.layer == "style_vectors": conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if _canonical_source: + _validate_routed_style_source(conn) if _style_extra_triggers(conn): raise _StyleAppendChanged("StyleTriggersUnsupported") previous_row = _read_layer_row(conn, spec.layer) @@ -1003,7 +1129,7 @@ def confirm_diagnostic_rollback() -> None: 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 category not in {"StyleWriterBusy", "StyleTriggersUnsupported"}): + or category not in {"StyleWriterBusy", "StyleTriggersUnsupported", "StyleSourceChanged"}): raise results.append(LayerResult(spec.layer, spec.result_key, "deferred", state.output_count, time.monotonic() - started, category, False, state.successful_coverage)) @@ -1031,6 +1157,9 @@ def confirm_rollback() -> None: ): output_started = True state = read_layer_states(conn, (spec,))[spec.layer] + if (_canonical_source and spec.layer == "style_vectors" + and baseline_state.successful_coverage == "unverified"): + state = state._replace(successful_coverage="unverified") if state.dependency.deferred: _check_layer_dependency(state.dependency, state.dependency) if not state.dependency.available: @@ -1062,7 +1191,10 @@ def confirm_rollback() -> None: raise _LayerRollbackFailed("Deferred output has no confirmed rollback") from error if not borrowed: try: - if spec.layer == "style_vectors": + if _canonical_source: + _restore_deferred_attempt(conn, spec.layer, previous_row, running_row, owner, + protect_rollback=True, canonical_source=True) + elif spec.layer == "style_vectors": _restore_deferred_attempt(conn, spec.layer, previous_row, running_row, owner, protect_rollback=True) else: @@ -1155,6 +1287,305 @@ def _style_extra_triggers(conn: sqlite3.Connection) -> bool: return main is not None or temporary is not None +def _style_error_category(category: str | None) -> str | None: + if category is None: + return None + return category if re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", category) else "Error" + + +def _style_source_ready(conn: sqlite3.Connection) -> None: + """Observe current canonical tracking, never repair or infer its history.""" + required = { + "messages": set(_MAINTENANCE_SOURCE_FIELDS), + "maintenance_source_state": {"singleton", "epoch", "revision", "insert_count", "correction_count", + "max_seen_message_id", "nonappend_revision", "tracking_ready"}, + "maintenance_layers": set(_LAYER_ROW_COLUMNS), + "metadata": {"key", "value"}, + } + shadows = (*required, "entity_style_vectors") + placeholders = ",".join("?" for _ in shadows) + if conn.execute("SELECT 1 FROM temp.sqlite_master WHERE type IN ('table','view') " + f"AND lower(name) IN ({placeholders}) LIMIT 1", shadows).fetchone() is not None: + raise MaintenanceUnavailableError("Canonical style tables are shadowed") + for table, columns in required.items(): + actual = {row[1].lower() for row in conn.execute(f"PRAGMA main.table_xinfo({table})")} + if not columns.issubset(actual): + raise MaintenanceUnavailableError("Canonical source schema is unavailable") + if not _messages_have_rowid_id(conn) or not _valid_style_source(read_source_revision(conn)): + raise MaintenanceUnavailableError("Canonical source identity is unavailable") + definitions = _maintenance_trigger_definitions(conn) + placeholders = ",".join("?" for _ in definitions) + current = {row[0]: " ".join(row[1].strip().rstrip(";").split()) for row in conn.execute( + "SELECT name,sql FROM main.sqlite_master WHERE type='trigger' " + f"AND name IN ({placeholders})", tuple(definitions), + )} + expected = {name: " ".join(sql.split()) for name, sql in definitions.items()} + shadow = conn.execute("SELECT 1 FROM temp.sqlite_master WHERE type='trigger' " + f"AND lower(name) IN ({placeholders}) LIMIT 1", tuple(definitions)).fetchone() + if current != expected or shadow is not None: + raise MaintenanceUnavailableError("Canonical source triggers are unavailable") + + +class StyleObservation(NamedTuple): + state: LayerState | None + checkpoint: tuple | None + eligible: bool + health: dict + + +def observe_style(conn: sqlite3.Connection | None = None, *, threshold: int = 25, + pending_reason: str | None = None) -> StyleObservation: + """Bounded native-free readiness; no profile/source scans or schema writes.""" + if type(threshold) is not int or threshold < 1: + raise ValueError("Maintenance threshold must be a positive integer") + health = {"available": False, "outcome": "pending", "freshness": "unknown", "coverage": "unverified", + "error_category": None, "output_count": None, "pending_reason": pending_reason or "not_connected", + "pending_caller_commit": False, "hash_ready": False, "format_ready": False} + if conn is None: + return StyleObservation(None, None, False, health) + borrowed = conn.in_transaction + health["pending_caller_commit"] = borrowed + owned = False + state = checkpoint = None + eligible = False + try: + if not borrowed: + conn.execute("BEGIN") + owned = True + if _style_extra_triggers(conn): + health.update(outcome="deferred", error_category="StyleTriggersUnsupported", pending_reason="unsupported_triggers") + return StyleObservation(None, None, False, health) + _style_source_ready(conn) + spec = style_layer_spec(conn) + state = read_layer_states(conn, (spec,))[spec.layer] + checkpoint = _read_layer_row(conn, spec.layer) + marker = conn.execute("SELECT value FROM metadata WHERE key='style_vec_hash_version'").fetchone() + hash_ready = marker == ("2",) + health["hash_ready"] = hash_ready + default = next((row[4] for row in conn.execute("PRAGMA main.table_info(entity_style_vectors)") + if row[1] == "accumulator_version"), "unsupported") + format_ready = default is None or default.strip(" ()'\"") in {"0", "1", "NULL", "null"} + health["format_ready"] = format_ready + ready = state.dependency.available + freshness = layer_freshness(state.source, state, state.dependency) + same_failure = (state.outcome in {"failed", "unavailable"} + and state.attempted_epoch == state.source.epoch and state.attempted_revision == state.source.revision + and state.attempted_dependency == state.dependency.key) + scheduled = _eligible(state, threshold, False) + # A terminal attempt owns its retry budget even without successful + # coverage or hash2. Readiness shortcuts cannot bypass that baseline. + eligible = scheduled if state.outcome in {"failed", "unavailable"} else ( + not ready or not hash_ready or not format_ready or state.successful_coverage != "complete" + or state.outcome == "running" or scheduled) + reason = (("tracking_repair" if not ready else "hash_migration" if not hash_ready else + "accumulator_pending" if not format_ready else "eligible") if eligible else + "attempt_unchanged" if same_failure else "current" if freshness == "current" else "awaiting_threshold") + health.update(available=ready and format_ready, outcome=state.outcome, + freshness=freshness if hash_ready and format_ready else "dependency_pending", + coverage=state.successful_coverage if ready and hash_ready and format_ready else "unverified", + error_category=_style_error_category(state.error_category or state.dependency.error_category + or (None if format_ready else "StyleAccumulatorUnsupported")), + output_count=state.output_count, pending_reason=reason) + if borrowed: + eligible = False + health.update(outcome="pending", pending_reason="pending_caller_commit") + except MaintenanceUnavailableError: + state = checkpoint = None + eligible = False + health.update(outcome="unavailable", error_category="StyleSourceUnavailable", pending_reason="source_unavailable") + except sqlite3.Error as error: + state = checkpoint = None + eligible = False + category = _style_sqlite_deferral(error) if isinstance(error, sqlite3.OperationalError) else None + health.update(outcome="deferred" if category else "failed", error_category=category or type(error).__name__, + pending_reason="inspection_failed") + finally: + if owned: + try: + conn.rollback() + except sqlite3.Error as error: + state = checkpoint = None + eligible = False + health.update(available=False, outcome="failed", freshness="unknown", coverage="unverified", + error_category=type(error).__name__, pending_reason="inspection_rollback_failed") + return StyleObservation(state, checkpoint, eligible, health) + + +def style_health(conn: sqlite3.Connection | None = None, coordinator: MaintenanceCoordinator | None = None, + *, pending_reason: str | None = None) -> dict: + health = observe_style(conn, pending_reason=pending_reason).health + if coordinator is not None: + previous, category = coordinator.style_observation() + current = (not health["pending_caller_commit"] and health["freshness"] == "current" + and health["coverage"] == "complete" and health["outcome"] in {"success", "success_empty"}) + if not current: + health["last_error"] = category or (previous.error_category if previous is not None else None) + return health + + +class _StyleOpenRejected(RuntimeError): + def __init__(self, result: LayerResult) -> None: + self.result = result + super().__init__(result.error_category) + + +def _style_unready_result(health: dict) -> LayerResult: + return LayerResult("style_vectors", "style_vectors", health["outcome"], health["output_count"], 0.0, + health["error_category"], False, health["coverage"], health["pending_caller_commit"]) + + +def _open_style_database(path: Path, cancel: threading.Event | None = None) -> sqlite3.Connection: + """Open an existing canonical file without general initialization or migrations.""" + def cancelled() -> bool: + return cancel is not None and cancel.is_set() + + def deferred(category: str) -> _StyleOpenRejected: + return _StyleOpenRejected(LayerResult("style_vectors", "style_vectors", "deferred", None, 0.0, + category, False, "unverified")) + + if cancelled(): + raise deferred("StyleCancelled") + if not isinstance(path, Path) or not path.is_absolute(): + raise ValueError("Style worker requires a canonical file path") + conn = None + progress = False + try: + conn = sqlite3.connect(path.as_uri() + "?mode=rw", uri=True) + conn.row_factory = None + for sql in (f"PRAGMA busy_timeout={DEFAULT_BUSY_TIMEOUT_MS}", "PRAGMA foreign_keys=ON", + "PRAGMA synchronous=NORMAL", "PRAGMA cache_size=-64000", "PRAGMA mmap_size=268435456"): + conn.execute(sql) + if _style_extra_triggers(conn): + raise deferred("StyleTriggersUnsupported") + try: + _style_source_ready(conn) + except MaintenanceUnavailableError as error: + raise _StyleOpenRejected(LayerResult("style_vectors", "style_vectors", "unavailable", None, 0.0, + "StyleSourceUnavailable", False, "unverified")) from error + observation = observe_style(conn) + if observation.state is None: + raise _StyleOpenRejected(_style_unready_result(observation.health)) + if cancelled(): + raise deferred("StyleCancelled") + if cancel is not None: + conn.set_progress_handler(lambda: int(cancelled()), 1000) + progress = True + try: + checked = conn.execute("PRAGMA quick_check(1)").fetchone() + finally: + if progress: + conn.set_progress_handler(None, 0) + progress = False + if cancelled(): + raise deferred("StyleCancelled") + if checked != ("ok",): + raise DatabaseOpenError("Database integrity check failed; restore a known-good backup before retrying") + return conn + except BaseException as error: + if conn is not None: + try: + if progress: + conn.set_progress_handler(None, 0) + if conn.in_transaction: + conn.rollback() + finally: + conn.close() + if isinstance(error, sqlite3.OperationalError): + category = _style_sqlite_deferral(error, cancelled=cancelled()) + if category: + raise deferred(category) from error + if isinstance(error, sqlite3.DatabaseError) and not isinstance(error, DatabaseOpenError): + message = str(error).lower() + if "i/o error" in message: + raise DatabaseOpenError("Database I/O failed; close all TrueMemory processes and retry") from error + raise DatabaseOpenError("Database could not be validated; check access or restore a known-good backup") from error + raise + + +def _validate_routed_style_source(conn: sqlite3.Connection) -> None: + try: + _style_source_ready(conn) + except MaintenanceUnavailableError as error: + raise _StyleAppendChanged("StyleSourceChanged") from error + + +def _prepare_routed_style_maintenance(conn: sqlite3.Connection) -> None: + """Validate canonical tracking under writer ownership before enrollment.""" + borrowed = conn.in_transaction + entered = rolled_back = False + + def confirmed_rollback() -> None: + nonlocal rolled_back + rolled_back = True + + transaction = (_borrowed_transaction(conn, on_rollback=confirmed_rollback) if borrowed else + _owned_transaction(conn, on_rollback=confirmed_rollback, immediate=True)) + try: + if _style_output_tracking_ready(conn): + return + with transaction: + entered = True + if borrowed: + conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + _validate_routed_style_source(conn) + prepare_style_maintenance(conn) + except sqlite3.OperationalError as error: + if (conn.in_transaction == borrowed and (not entered or rolled_back) + and _style_sqlite_deferral(error) == "StyleWriterBusy"): + raise LayerDeferredError("Style preparation writer busy", category="StyleWriterBusy") from error + raise + + +def run_routed_style(conn: sqlite3.Connection, *, force: bool = False, threshold: int = 25, + cancel: threading.Event | None = None, allow_caller_transaction: bool = False, + connection_owned: bool = False) -> LayerResult: + """Canonical engine route; explicit low-level style APIs retain their contract.""" + if cancel is not None and cancel.is_set(): + return LayerResult("style_vectors", "style_vectors", "deferred", None, 0.0, "StyleCancelled", False, + "unverified", conn.in_transaction) + with maintenance_owner(connection_database_path(conn)): + observation = observe_style(conn, threshold=threshold) + if observation.state is None: + return _style_unready_result(observation.health) + if conn.in_transaction and not allow_caller_transaction: + return _style_unready_result(observation.health) + # A wrong marker cannot leave a previously certified checkpoint current. + # Compare the complete old row under writer ownership; preserve failures. + old = observation.checkpoint + if (old is not None and old[2] in {"success", "success_empty"} + and (not observation.health["hash_ready"] or not observation.health["format_ready"] + or observation.state.successful_coverage != "complete")): + borrowed = conn.in_transaction + transaction = _borrowed_transaction if borrowed else _owned_transaction + rolled_back = False + + def confirmed_rollback() -> None: + nonlocal rolled_back + rolled_back = True + + try: + with transaction(conn, on_rollback=confirmed_rollback): + conn.execute("UPDATE maintenance_layers SET layer=layer WHERE 0") + if _style_extra_triggers(conn): + raise LayerDeferredError("Style triggers changed", category="StyleTriggersUnsupported") + try: + _style_source_ready(conn) + except MaintenanceUnavailableError as error: + raise LayerDeferredError("Canonical source readiness changed", category="StyleSourceChanged") from error + if _read_layer_row(conn, "style_vectors") != old: + raise LayerDeferredError("Style checkpoint changed", category="StyleSourceChanged") + conn.execute(_STYLE_OUTPUT_INVALIDATE_SQL) + except (sqlite3.OperationalError, LayerDeferredError) as error: + category = error.category if isinstance(error, LayerDeferredError) else _style_sqlite_deferral(error) + if not rolled_back or conn.in_transaction != borrowed or category is None: + raise + return LayerResult("style_vectors", "style_vectors", "deferred", None, 0.0, + category, False, "unverified", borrowed) + return run_style_maintenance(conn, force=force, threshold=threshold, cancel=cancel, + allow_caller_transaction=allow_caller_transaction, connection_owned=connection_owned, + _canonical_source=True) + + def _style_append_metadata(conn: sqlite3.Connection) -> tuple: dependency = style_layer_spec(conn).resolve_dependency() if dependency.error_category == "StyleTriggersUnsupported": @@ -1359,6 +1790,7 @@ def pending(category: str, *, failed: bool = False) -> LayerResult: def style_layer_spec( conn: sqlite3.Connection, *, cancel: threading.Event | None = None, connection_owned: bool = False, + _canonical_source: bool = False, ) -> LayerSpec: """Describe private style work without installing schema or reading profiles.""" if type(connection_owned) is not bool: @@ -1444,6 +1876,11 @@ 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 _canonical_source: + try: + _validate_routed_style_source(connection) + except _StyleAppendChanged as error: + raise LayerDeferredError("Canonical source readiness changed", category=error.category) from error if _style_extra_triggers(connection): raise LayerDeferredError("Style triggers prevent tracked publication", category="StyleTriggersUnsupported") style._publish_style_vectors(connection, schema, query, rows, stored_rows, @@ -1462,7 +1899,7 @@ def build(connection: sqlite3.Connection) -> object: def run_style_maintenance( conn: sqlite3.Connection, *, force: bool = False, threshold: int = 25, cancel: threading.Event | None = None, allow_caller_transaction: bool = False, - connection_owned: bool = False, + connection_owned: bool = False, _canonical_source: bool = False, ) -> LayerResult: """Explicit tracked style rebuild, separate from all eight public adapters.""" if type(allow_caller_transaction) is not bool: @@ -1482,19 +1919,34 @@ def preparation_failure(error: sqlite3.OperationalError) -> None: raise LayerDeferredError("Style preparation writer busy", category="StyleWriterBusy") from error try: - prepare_style_maintenance(conn, _on_safe_failure=preparation_failure) + if _canonical_source: + _prepare_routed_style_maintenance(conn) + else: + 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) - spec = style_layer_spec(conn, cancel=cancel, connection_owned=connection_owned) - results = run_layers(conn, (spec,), force=force, threshold=threshold, cancel=cancel, - allow_caller_transaction=allow_caller_transaction) - if results: - return results[0] - observed = _read_layer_run_baseline(conn, (spec,)) + routed = {"_canonical_source": True} if _canonical_source else {} + spec = style_layer_spec(conn, cancel=cancel, connection_owned=connection_owned, **routed) + try: + results = run_layers(conn, (spec,), force=force, threshold=threshold, cancel=cancel, + allow_caller_transaction=allow_caller_transaction, **routed) + if results: + return results[0] + observed = _read_layer_run_baseline(conn, (spec,), **routed) + except _StyleAppendChanged as error: + if not _canonical_source or error.category not in { + "StyleSourceChanged", "StyleTrackingUnprepared", "StyleTriggersUnsupported", + "StyleMetadataChanged", "StyleAccumulatorUnsupported", "StyleHashPending", + }: + raise + if conn.in_transaction != borrowed: + raise _LayerRollbackFailed("Canonical readiness rejection changed transaction ownership") from error + return LayerResult("style_vectors", "style_vectors", "deferred", None, + time.monotonic() - started, error.category, False, "unverified", borrowed) if observed is None: return _style_baseline_busy_result(spec, borrowed) state = observed[spec.layer] @@ -1738,6 +2190,7 @@ def dunbar_dependency() -> LayerDependency: class MaintenanceReport(NamedTuple): results: tuple[LayerResult, ...] preferences: str + style_result: LayerResult | None = None def _dependency_parameters(dependency: LayerDependency) -> dict | None: @@ -1908,8 +2361,16 @@ def run_engine_maintenance( conn: sqlite3.Connection, coordinator: MaintenanceCoordinator, *, threshold: int = 25, force: bool = False, cancel: threading.Event | None = None, prepare_extensions: bool = True, allow_caller_transaction: bool = False, + include_layers: bool = True, include_style: bool = False, connection_owned: bool = False, ) -> MaintenanceReport: """Use only the supplied owned worker/original explicitly borrowed handle.""" + style_result = None + if include_style: + style_result = run_routed_style(conn, force=force, threshold=threshold, cancel=cancel, + allow_caller_transaction=allow_caller_transaction, connection_owned=connection_owned) + if not include_layers or (cancel is not None and cancel.is_set()): + return MaintenanceReport((), "CANCELLED" if cancel is not None and cancel.is_set() else + "SKIPPED (style-only)", style_result) evidence = _prepare_worker_extensions(conn, coordinator) if prepare_extensions else None specs = engine_layer_specs(conn, coordinator, evidence=evidence) results = run_layers(conn, specs, threshold=threshold, force=force, cancel=cancel, @@ -1935,21 +2396,22 @@ def run_engine_maintenance( preferences = f"{time.monotonic() - started:.3f}s" if conn.in_transaction and allow_caller_transaction: preferences += " (pending caller commit)" - return MaintenanceReport(results, preferences) + return MaintenanceReport(results, preferences, style_result) def maintenance_report_status(report: MaintenanceReport, cancel: threading.Event | None = None) -> tuple[str, str | None]: if cancel is not None and cancel.is_set(): return "cancelled", "Cancelled" + results = report.results + ((report.style_result,) if report.style_result is not None else ()) for outcome in ("failed", "unavailable", "deferred", "abandoned"): - for result in report.results: + for result in results: if result.outcome == outcome: return outcome, result.error_category if report.preferences.startswith(("ERROR", "UNAVAILABLE")): return "failed", "PreferencesUnavailable" - if any(result.outcome not in {"success", "success_empty", "current"} for result in report.results): + if any(result.pending_caller_commit or result.outcome not in {"success", "success_empty", "current"} for result in results): return "pending", None - if any(result.coverage != "complete" for result in report.results): + if any(result.coverage != "complete" for result in results): return "completed_with_limits", None return "success", None diff --git a/truememory/storage.py b/truememory/storage.py index dc5b6d9e..0ca5f64e 100644 --- a/truememory/storage.py +++ b/truememory/storage.py @@ -1597,7 +1597,8 @@ def delete_message(conn: sqlite3.Connection, msg_id: int) -> bool: for _ent in _entities: for _tbl in ("summaries", "entity_profiles", "entity_style_vectors"): try: - conn.execute(f"DELETE FROM {_tbl} WHERE entity = ?", (_ent,)) + conn.execute(f"DELETE FROM {_tbl} WHERE entity = ?", + (_ent.lower() if _tbl == "entity_style_vectors" else _ent,)) except sqlite3.OperationalError: pass try: From 8cd28ac366f45e2642eabd8c26f3b8ac2ebc4785 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 20:39:15 -0500 Subject: [PATCH 2/2] test: Isolate surprise fixtures from background maintenance --- tests/test_l5_surprise_rerank.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/test_l5_surprise_rerank.py b/tests/test_l5_surprise_rerank.py index 10f72552..65fb28e6 100644 --- a/tests/test_l5_surprise_rerank.py +++ b/tests/test_l5_surprise_rerank.py @@ -60,8 +60,20 @@ def engine_with_surprise(tmp_path, monkeypatch): conn.close() eng = TrueMemoryEngine(db_path) - eng.open(rebuild_vectors=False) - return eng + # These tests own their surprise rows and do not exercise maintenance. + eng._has_consolidation = False + eng._has_style_vec = False + try: + eng.open(rebuild_vectors=False) + yield eng + finally: + coordinator = eng._maintenance_coordinator + try: + if coordinator is not None: + coordinator.cancel() + assert coordinator.wait(10), "Fixture maintenance worker did not stop" + finally: + eng.close() def _fake_results(ids_and_scores, source=None):