diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md
index 3f8b1b01..662152a0 100644
--- a/docs/resource-budgets.md
+++ b/docs/resource-budgets.md
@@ -994,3 +994,114 @@ the current definition checks exposing that interval. Arbitrary direct FTS
 mutations can also defeat synchronization. This checkpoint adds neither a
 schema-cookie fence nor a readiness gate to FTS search, and does not certify
 either kind of external mutation.
+
+### Foreground model admission before writer ownership
+
+Owned foreground add and incremental vector publication now acquire the model
+identity fence before opening the SQLite write transaction. Add first owns its
+engine write lock, so another mutation on that same engine cannot hold the
+engine lock while waiting for a model fence held by the queued add. The model
+fence validates the captured generation and remains held through commit or
+rollback. There is one model acquisition, with no nested acquisition of the
+nonreentrant model lock.
+
+This permits an owned foreground call to wait for a concurrent cluster source
+snapshot without holding SQLite's writer or failing a valid vector publication.
+Model loading and inference finish before owned writer admission. A model
+generation change during loading, encoding or admission still rejects stale
+publication; add and update propagate those typed errors instead of silently
+falling back to a source-only write. Update retains its existing distinction:
+capture failure precedes its source write, while a later vector-publication
+failure can occur after the source update committed.
+
+Caller transactions use nonblocking model capture and publication admission.
+Add and update classify the connection during model capture under their engine
+write lock, so another same-engine mutation's transaction cannot be mistaken
+for borrowed caller work. They release that lock before encoding.
+Capture accepts only a cached model observed together with its identity under
+the model lock. Busy ownership or an unloaded model produces a typed retry
+error before any additional source write; the caller's transaction remains
+intact. Cached-model inference can still run inside a caller-held transaction.
+The publication fence protects the nested savepoint release, not a later outer
+commit controlled by that caller. Streamed rebuilds retain nonblocking model
+fences after writer admission and use the same cached-only borrowed capture.
+
+SQLite and event regressions cover actual clustering snapshot contention,
+same-engine mutation ordering, stale generations, borrowed sentinels, atomic
+add publication, and model release after admission/publication/commit failures.
+This checkpoint changes admission ordering rather than bounding wait duration:
+cluster snapshots may still hold the model lock while scanning the corpus.
+Native integration and latency remain separate acceptance checks.
+
+### Engine maintenance routing
+
+Engine startup, committed mutations and later safe API boundaries now use the
+same durable eight-layer planner. The default append trigger remains
+`insert_count - attempted_insert_count >= 25`; 24 committed appends do not
+qualify and the 25th does, including across engine instances and reopen.
+`TRUEMEMORY_AUTO_CONSOLIDATE_EVERY` retains its positive-integer override.
+Initial historical or empty input attempts once. Corrections, source epochs and
+dependency changes follow their separate eligibility rules. A successful empty
+cluster result no longer starts every layer again on each open. Source edits
+committed through another handle become visible at the next safe API boundary;
+there is no idle polling timer. Uncommitted caller changes do not schedule work.
+
+Foreground scheduling reads eight checkpoint records, source counters and small
+dependency metadata. It does not scan source/vector rows, import native
+frameworks, wait for the model lock or perform consolidation in `add()`. The
+foreground write lock is acquired nonblocking for this probe. Package versions
+are cached in three records and refreshed on connection setup and worker/manual
+setup. One extension-failure record is bound to the probed model/table identity
+and version snapshot. Actual worker extension loading supplies that evidence;
+metadata alone is not proof of availability. Epoch checks prevent a late worker
+from overwriting newer shared capability evidence, while that worker still uses
+its own actual failure for its current attempt. A transient model-busy probe
+cannot erase an existing extension-failure baseline.
+
+Each canonical file has at most one active owner and one pending notification
+generation. A notification stores only a generation and threshold, not an
+engine, caller connection or source content. Multiple foreground notifications
+replace that single pending slot. A same-process owner's release services an
+existing pending notification; it creates no notification of its own. A pending
+coordinator behind a local owner is retained only until that exact owner's
+release or cancellation, so closing the last engine does not lose the wake.
+Cross-process contention reports busy/pending and needs a later API boundary;
+this checkpoint has no automatic cross-process release notification. Deferred
+model ownership and failed attempts do not schedule their own retry loops.
+
+Workers create and close their own file connection. `engine.close()` detaches
+and closes only the foreground connection; it does not cancel or join a shared
+worker. Cancellation clears queued notifications but retains ownership until
+active computation reaches a cooperative boundary and its connection closes.
+Thread-start failure closes an unclaimed owner descriptor; interruption after
+the target started leaves teardown with that exact worker and cannot cancel a
+later successor. Child processes discard inherited owner descriptors, pending
+references and registry state. Daemon threads do not guarantee completion when
+the process exits; durable abandoned-attempt recovery remains necessary.
+
+Explicit `consolidate()` forces one pass with the same eight result keys plus
+the existing nonpersisted preference extraction result. A busy owner produces
+nine explicit `BUSY` entries and does not queue a hidden forced pass. A file
+call without an existing transaction uses a dedicated connection. An existing
+caller transaction uses the original connection's protected savepoints and
+reports `pending caller commit`; its writes and maintenance output remain
+rollbackable together. `:memory:` automatic work reports pending without opening
+an empty replacement database or creating a worker. Explicit manual maintenance
+uses the original in-memory handle. Caller-held writer and later outer-commit
+limitations from the borrowed-runner contract still apply.
+
+Successful manual summary/cluster publication restores the calling engine's
+corresponding capability flags, including successful empty output. Failed or
+unavailable layers never enable those flags or clear a previously usable
+capability. The existing automatic-consolidation opt-out remains until explicit
+manual repair. Cluster retrieval still requires live vector availability.
+`get_stats()` exposes bounded coordinator and per-layer outcome, freshness,
+coverage and categorical error state; it does not expose source text, paths or
+exception messages. Unverified vector-generation and legacy-contact coverage
+remain explicit limitations, rather than complete success claims.
+
+This routing checkpoint preserves builder algorithms and the A17 foreground
+vector-publication guards. Ingest database replacement, profile/style getter
+freshness and custom-schema source tracking remain separate work. Synthetic
+SQLite and event tests establish scheduling and ownership behavior; native
+latency, throughput and memory acceptance still require measurement.
diff --git a/tests/test-maintenance-adapters-753.py b/tests/test-maintenance-adapters-753.py
index a0454f79..ebc0601c 100644
--- a/tests/test-maintenance-adapters-753.py
+++ b/tests/test-maintenance-adapters-753.py
@@ -57,7 +57,11 @@ def setUp(self) -> None:
             ("vec_source_v1:vec_messages_edge", '{"version":1,"generation":"synthetic-a","cursor":0,"complete":true}'),
         ])
         self.conn.commit()
-        self.enterContext(patch.dict(sys.modules, {"truememory.vector_search": self.vector}))
+        # Detached production workers must not discover this fixture's model
+        # or hold its lock through the process-wide module registry.
+        self.enterContext(patch.object(MAINTENANCE, "sys", types.SimpleNamespace(
+            modules={"truememory.vector_search": self.vector}, version_info=sys.version_info,
+        )))
         self.enterContext(patch.object(MAINTENANCE, "importlib", types.SimpleNamespace(
             import_module=lambda name: self.modules[name], metadata=importlib.metadata, util=importlib.util,
         )))
diff --git a/tests/test-maintenance-engine-routing-753.py b/tests/test-maintenance-engine-routing-753.py
new file mode 100644
index 00000000..5c8c02ff
--- /dev/null
+++ b/tests/test-maintenance-engine-routing-753.py
@@ -0,0 +1,783 @@
+"""Engine routing with real SQLite, event-controlled owners and synthetic models."""
+
+import ast
+import builtins
+import gc
+import logging
+import os
+import select
+import signal
+import runpy
+import sqlite3
+import sys
+import threading
+import time
+import types
+import unittest
+import warnings
+import weakref
+from contextlib import contextmanager
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ADAPTERS = runpy.run_path(str(Path(__file__).with_name("test-maintenance-adapters-753.py")))
+STORAGE, MAINTENANCE = ADAPTERS["STORAGE"], ADAPTERS["MAINTENANCE"]
+
+
+def stdlib_module(name: str, modules: dict) -> types.ModuleType:
+    module = types.ModuleType("synthetic_routing_" + name)
+
+    def safe_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name in modules:
+            return modules[name]
+        if name.startswith("truememory"):
+            raise ImportError("Synthetic boundary excludes native modules")
+        return builtins.__import__(name, globals, locals, fromlist, level)
+
+    module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import)
+    path = ROOT / "truememory" / (name + ".py")
+    with patch.dict(sys.modules, {module.__name__: module}):
+        exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), module.__dict__)
+    return module
+
+
+def load_engine(modules: dict) -> types.ModuleType:
+    module = types.ModuleType("synthetic_routing_engine")
+
+    def safe_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name in modules:
+            return modules[name]
+        if name.startswith("truememory"):
+            raise ImportError("Synthetic boundary excludes native modules")
+        return builtins.__import__(name, globals, locals, fromlist, level)
+
+    module.__dict__.update({"__builtins__": dict(vars(builtins), __import__=safe_import),
+        "__file__": str(ROOT / "truememory/engine.py"), "Path": Path, "sqlite3": sqlite3,
+        "threading": threading, "time": time, "os": os, "warnings": warnings,
+        "logger": logging.getLogger("synthetic-routing"), "MAX_CONTENT_LENGTH": 50_000,
+        "_env_int": lambda name, default, **kwargs: default})
+    source = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8"))
+    nodes = [ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0)]
+    for node in source.body:
+        if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
+            nodes.append(node)
+        elif isinstance(node, ast.ImportFrom) and node.module == "truememory.storage":
+            module.__dict__.update({item.asname or item.name: getattr(STORAGE, item.name) for item in node.names})
+        elif isinstance(node, ast.Assign):
+            for target in node.targets:
+                if isinstance(target, ast.Name):
+                    if target.id.startswith("_HAS_"):
+                        module.__dict__[target.id] = False
+                    elif target.id in {"_ALLOWED_TABLES", "_ALLOWED_COLUMNS", "_ALL_VEC_TABLES", "_SQLITE_IN_CHUNK"}:
+                        nodes.append(node)
+    module.__dict__["_HAS_CONSOLIDATION"] = True
+    exec(compile(ast.fix_missing_locations(ast.Module(body=nodes, type_ignores=[])),
+                 str(ROOT / "truememory/engine.py"), "exec"), module.__dict__)
+    return module
+
+
+class RoutingFixture(ADAPTERS["AdapterFixture"]):
+    def setUp(self) -> None:
+        style = stdlib_module("personality_style_vec", {})
+        self.enterContext(patch.dict(sys.modules, {"truememory.personality_style_vec": style}))
+        super().setUp()
+        self.modules.update({"truememory.storage": STORAGE, "truememory.maintenance": MAINTENANCE,
+                             "truememory.vector_search": self.vector,
+                             "sqlite_vec": types.SimpleNamespace(load=lambda conn: None)})
+        self.modules["truememory.rebuild_source"] = stdlib_module("rebuild_source", self.modules)
+        self.engine_module = load_engine(self.modules)
+        create = STORAGE.create_db
+
+        class ExtensionBoundary:
+            def __init__(boundary, conn):
+                boundary.conn = conn
+
+            def __getattr__(boundary, name):
+                return getattr(boundary.conn, name)
+
+            def enable_load_extension(boundary, enabled):
+                # The fixture uses ordinary BLOB tables and a fake sqlite-vec
+                # loader, including hosts compiled without extension support.
+                self.assertIs(type(enabled), bool)
+
+        self.enterContext(patch.object(MAINTENANCE, "create_db", lambda path: ExtensionBoundary(create(path))))
+        self.enterContext(patch.object(self.engine_module, "create_db", lambda path: ExtensionBoundary(create(path))))
+        self.engine = self.new_engine(self.conn)
+        self.coordinator = self.engine._get_maintenance_coordinator()
+        self.addCleanup(self.coordinator.wait, 3)
+        self.addCleanup(self.coordinator.cancel)
+
+    def new_engine(self, conn: sqlite3.Connection | None = None):
+        engine = self.engine_module.TrueMemoryEngine(self.path)
+        engine.conn = conn or self.open_db()
+        engine.ready = True
+        engine._has_consolidation = False
+        self.addCleanup(engine.close)
+        return engine
+
+    def completed(self) -> None:
+        self.assertTrue(self.coordinator.wait(3), "Synthetic maintenance did not terminate")
+        self.assertFalse(self.coordinator.snapshot()["active"])
+
+    def fake_report(self):
+        return MAINTENANCE.MaintenanceReport((), "SKIPPED (synthetic)")
+
+
+class TestCoordinatorHandoff(RoutingFixture):
+    def test_local_owner_retains_one_pending_wake_after_last_engine_close(self) -> None:
+        path = self.path.with_name("synthetic-detached.sqlite")
+        engine = self.engine_module.TrueMemoryEngine(path)
+        engine.conn, engine.ready = STORAGE.create_db(path), True
+        coordinator = engine._get_maintenance_coordinator()
+        reference, engine_reference = weakref.ref(coordinator), weakref.ref(engine)
+        entered, release = threading.Event(), threading.Event()
+        observed = []
+
+        def work(conn, owner, **kwargs):
+            observed.append(kwargs["threshold"])
+            entered.set()
+            self.assertTrue(release.wait(3))
+            return self.fake_report()
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance", work):
+            try:
+                with MAINTENANCE.maintenance_owner(path):
+                    coordinator.request_layers(threshold=31)
+                    coordinator.request_layers(threshold=37)
+                    self.assertEqual(len(MAINTENANCE._owner_waiters), 1)
+                    engine.close()
+                    del engine, coordinator
+                    gc.collect()
+                    self.assertIsNone(engine_reference())
+                    self.assertIsNotNone(reference())
+                self.assertTrue(entered.wait(2))
+                self.assertEqual(MAINTENANCE._owner_waiters, {})
+                coordinator = reference()
+                self.assertIsNotNone(coordinator)
+            finally:
+                release.set()
+                if reference() is not None:
+                    self.assertTrue(reference().wait(3))
+        self.assertEqual(observed, [37])
+
+    def test_cancel_releases_local_waiter_and_admission_cannot_restore_it(self) -> None:
+        original = MAINTENANCE._acquire_owner
+
+        def cancelled_admission(path, on_busy=None):
+            self.coordinator.cancel()
+            return original(path, on_busy=on_busy)
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance") as work:
+            with MAINTENANCE.maintenance_owner(self.path):
+                self.coordinator.request_layers()
+                self.assertEqual(len(MAINTENANCE._owner_waiters), 1)
+                self.coordinator.cancel()
+                self.assertEqual(MAINTENANCE._owner_waiters, {})
+                with patch.object(MAINTENANCE, "_acquire_owner", cancelled_admission):
+                    self.assertFalse(self.coordinator.request_layers())
+                self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+                self.assertEqual(MAINTENANCE._owner_waiters, {})
+            work.assert_not_called()
+        self.assertEqual(self.coordinator.status, ("cancelled", None))
+
+    def test_delayed_start_exception_cannot_cancel_successor_or_its_pending_wake(self) -> None:
+        start = threading.Thread.start
+        entered, release = threading.Event(), threading.Event()
+        attempts, observed, cancellation = [], [], []
+
+        def work(conn, coordinator, **kwargs):
+            observed.append(kwargs["threshold"])
+            if kwargs["threshold"] == 31:
+                entered.set()
+                self.assertTrue(release.wait(3))
+                cancellation.append(kwargs["cancel"].is_set())
+            return self.fake_report()
+
+        def late_failure(worker):
+            attempts.append(worker)
+            start(worker)
+            if len(attempts) == 1:
+                self.completed()
+                self.assertTrue(self.coordinator.request_layers(threshold=31))
+                self.assertTrue(entered.wait(2))
+                self.assertFalse(self.coordinator.request_layers(threshold=37))
+                raise RuntimeError("synthetic delayed start failure")
+
+        with patch.object(MAINTENANCE.threading.Thread, "start", late_failure), \
+             patch.object(MAINTENANCE, "run_engine_maintenance", work):
+            try:
+                with self.assertRaisesRegex(RuntimeError, "synthetic delayed start failure"):
+                    self.coordinator.request_layers()
+                self.assertFalse(self.coordinator._cancel.is_set())
+                self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1)
+            finally:
+                release.set()
+                self.completed()
+        self.assertEqual(observed, [25, 31, 37])
+        self.assertEqual(cancellation, [False])
+
+    @unittest.skipUnless(hasattr(os, "fork"), "POSIX fork required")
+    def test_fork_discards_inherited_pending_worker_and_rejects_stale_coordinator(self) -> None:
+        entered, release = threading.Event(), threading.Event()
+
+        def active(conn, cancel):
+            entered.set()
+            self.assertTrue(release.wait(3))
+
+        self.coordinator.request(active)
+        self.assertTrue(entered.wait(2))
+        self.coordinator.request_layers()
+        waiting_path = self.path.with_name("synthetic-fork-waiter.sqlite")
+        self.enterContext(MAINTENANCE.maintenance_owner(waiting_path))
+        waiting = MAINTENANCE.get_coordinator(waiting_path)
+        self.addCleanup(waiting.cancel)
+        waiting.request_layers()
+        self.assertEqual(len(MAINTENANCE._owner_waiters), 1)
+        read_fd, write_fd = os.pipe()
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            child = os.fork()
+        if child == 0:
+            os.close(read_fd)
+            try:
+                rejected = False
+                try:
+                    self.coordinator.request_layers()
+                except RuntimeError:
+                    rejected = True
+                fresh = MAINTENANCE.get_coordinator(self.path)
+                clean = (not fresh.snapshot()["active"] and fresh.snapshot()["pending_generations"] == 0
+                         and not MAINTENANCE._owner_waiters)
+                os.write(write_fd, b"PASS" if rejected and clean else b"FAIL")
+            finally:
+                os.close(write_fd)
+                os._exit(0)
+        os.close(write_fd)
+        try:
+            self.assertTrue(select.select([read_fd], [], [], 2)[0], "Synthetic child did not answer")
+            self.assertEqual(os.read(read_fd, 4), b"PASS")
+        finally:
+            os.close(read_fd)
+            if os.waitpid(child, os.WNOHANG)[0] == 0:
+                os.kill(child, signal.SIGKILL)
+                os.waitpid(child, 0)
+            self.coordinator.cancel()
+            release.set()
+            self.completed()
+
+    def test_start_failure_services_only_a_real_later_notification(self) -> None:
+        start = threading.Thread.start
+        attempts, observed = [], []
+
+        def fail_once(worker):
+            attempts.append(True)
+            if len(attempts) == 1:
+                self.assertFalse(self.coordinator.request_layers(threshold=31))
+                raise RuntimeError("synthetic start failure")
+            return start(worker)
+
+        with patch.object(MAINTENANCE.threading.Thread, "start", fail_once), \
+             patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(kwargs["threshold"]) or self.fake_report()):
+            with self.assertRaisesRegex(RuntimeError, "synthetic start failure"):
+                self.coordinator.request_layers()
+            self.completed()
+        self.assertEqual(observed, [31])
+        self.assertEqual(len(attempts), 2)
+
+    def test_start_failure_without_notification_has_no_retry(self) -> None:
+        with patch.object(MAINTENANCE.threading.Thread, "start", side_effect=RuntimeError("synthetic")) as start:
+            with self.assertRaises(RuntimeError):
+                self.coordinator.request_layers()
+            self.completed()
+        start.assert_called_once()
+        self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+        self.assertEqual(self.coordinator.status, ("failed", "worker_start"))
+
+    def test_close_detaches_without_cancelling_or_retaining_engine(self) -> None:
+        entered, release = threading.Event(), threading.Event()
+
+        def active(conn, cancel):
+            entered.set()
+            self.assertTrue(release.wait(2))
+            self.assertFalse(cancel.is_set())
+            self.assertEqual(conn.execute("SELECT 1").fetchone()[0], 1)
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: self.fake_report()):
+            self.coordinator.request(active)
+            self.assertTrue(entered.wait(2))
+            engine = self.engine_module.TrueMemoryEngine(self.path)
+            engine.conn, engine.ready = self.open_db(), True
+            reference = weakref.ref(engine)
+            engine._maybe_auto_consolidate()
+            self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1)
+            engine.close()
+            del engine
+            gc.collect()
+            self.assertIsNone(reference())
+            self.assertFalse(self.coordinator.wait(0))
+            release.set()
+            self.completed()
+        self.assertEqual(self.coordinator.status, ("success", None))
+
+    def test_busy_admission_and_concurrent_owner_release_cannot_lose_wake(self) -> None:
+        held, release, leaving, ran = (threading.Event() for _ in range(4))
+        original_acquire, original_release = MAINTENANCE._acquire_owner, MAINTENANCE._release_owner
+
+        def owner():
+            with MAINTENANCE.maintenance_owner(self.path):
+                held.set()
+                self.assertTrue(release.wait(2))
+
+        def release_owner(fd):
+            leaving.set()
+            original_release(fd)
+
+        def acquire(path, on_busy=None):
+            def busy():
+                on_busy()
+                release.set()
+                self.assertTrue(leaving.wait(2))
+            return original_acquire(path, on_busy=busy if on_busy is not None else None)
+
+        owner_thread = threading.Thread(target=owner)
+        with patch.object(MAINTENANCE, "_release_owner", release_owner), \
+             patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: ran.set() or self.fake_report()):
+            owner_thread.start()
+            try:
+                self.assertTrue(held.wait(2))
+                with patch.object(MAINTENANCE, "_acquire_owner", acquire):
+                    self.assertFalse(self.coordinator.request_layers())
+                    self.assertTrue(ran.wait(2))
+                    self.completed()
+            finally:
+                release.set()
+                owner_thread.join(3)
+        self.assertFalse(owner_thread.is_alive())
+
+    def test_many_active_notifications_coalesce_to_one_static_successor(self) -> None:
+        entered, release = threading.Event(), threading.Event()
+        observed = []
+
+        def active(conn, cancel):
+            entered.set()
+            self.assertTrue(release.wait(2))
+
+        def successor(conn, coordinator, **kwargs):
+            observed.append((conn is self.conn, kwargs["threshold"]))
+            return self.fake_report()
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance", successor):
+            self.assertTrue(self.coordinator.request(active))
+            self.assertTrue(entered.wait(2))
+            for threshold in range(1, 101):
+                self.assertFalse(self.coordinator.request_layers(threshold=threshold))
+            self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1)
+            self.assertFalse(self.coordinator.wait(0))
+            release.set()
+            self.completed()
+        self.assertEqual(observed, [(False, 100)])
+        self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+
+    def test_local_synchronous_owner_release_services_pending_wake(self) -> None:
+        observed = []
+        with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()):
+            with MAINTENANCE.maintenance_owner(self.path):
+                self.assertFalse(self.coordinator.request_layers())
+                self.assertEqual(self.coordinator.status, ("busy", None))
+                self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1)
+                self.assertEqual(observed, [])
+            self.completed()
+        self.assertEqual(observed, [True])
+
+    def test_external_busy_requires_new_boundary_and_does_not_spin(self) -> None:
+        observed = []
+        with patch.object(MAINTENANCE, "try_file_lock", return_value=None), \
+             patch.object(MAINTENANCE.threading, "Thread") as thread:
+            self.assertFalse(self.coordinator.request_layers())
+            thread.assert_not_called()
+        self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1)
+        self.assertEqual(MAINTENANCE._owner_waiters, {})
+        with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()):
+            self.assertTrue(self.coordinator.request_layers())
+            self.completed()
+        self.assertEqual(observed, [True])
+
+    def test_wake_between_descriptor_release_and_idle_transition_is_not_lost(self) -> None:
+        released, finish = threading.Event(), threading.Event()
+        original_release = MAINTENANCE._release_owner
+        observed = []
+        first = []
+
+        def pause(fd):
+            original_release(fd)
+            if not first:
+                first.append(True)
+                released.set()
+                self.assertTrue(finish.wait(2))
+
+        with patch.object(MAINTENANCE, "_release_owner", pause), \
+             patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()):
+            self.assertTrue(self.coordinator.request(lambda conn, cancel: None))
+            self.assertTrue(released.wait(2))
+            self.assertFalse(self.coordinator.request_layers())
+            self.assertFalse(self.coordinator.wait(0))
+            finish.set()
+            self.completed()
+        self.assertEqual(observed, [True])
+
+    def test_model_deferral_has_no_self_generated_successor(self) -> None:
+        observed = []
+        report = MAINTENANCE.MaintenanceReport((MAINTENANCE.LayerResult(
+            "clusters", "cluster_messages", "deferred", None, 0., "ModelBusy", False),), "SKIPPED")
+        with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or report):
+            self.assertTrue(self.coordinator.request_layers())
+            self.completed()
+        self.assertEqual(observed, [True])
+        self.assertEqual(self.coordinator.status, ("deferred", "ModelBusy"))
+        self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+
+    def test_cancellation_drops_pending_but_keeps_active_ownership(self) -> None:
+        entered, release = threading.Event(), threading.Event()
+
+        def active(conn, cancel):
+            entered.set()
+            self.assertTrue(release.wait(2))
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance") as successor:
+            self.coordinator.request(active)
+            self.assertTrue(entered.wait(2))
+            self.coordinator.request_layers()
+            self.coordinator.cancel()
+            self.assertFalse(self.coordinator.wait(0))
+            self.assertFalse(self.coordinator.request_layers())
+            self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+            release.set()
+            self.completed()
+            successor.assert_not_called()
+        self.assertEqual(self.coordinator.status, ("cancelled", None))
+
+
+class TestEngineRouting(RoutingFixture):
+    def test_status_reports_categorical_layer_failure_and_caller_transaction(self) -> None:
+        self.add(1)
+        with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic private sentinel")):
+            self.engine.consolidate()
+        state = self.engine.get_stats()["maintenance"]
+        self.assertEqual(len(state["layers"]), 8)
+        self.assertEqual(state["layers"]["clusters"]["outcome"], "failed")
+        self.assertEqual(state["layers"]["clusters"]["error_category"], "RuntimeError")
+        self.assertEqual(state["layers"]["clusters"]["freshness"], "untrusted")
+        self.assertNotIn("synthetic private sentinel", str(state))
+        self.assertNotIn(str(self.path), str(state))
+        self.conn.execute("UPDATE messages SET content='synthetic uncommitted status'")
+        self.assertTrue(self.engine.get_stats()["maintenance"]["pending_caller_commit"])
+        self.assertTrue(self.conn.in_transaction)
+        self.conn.rollback()
+
+    def test_manual_repair_restores_actual_cluster_search_and_automatic_eligibility(self) -> None:
+        self.add(5)
+        self.native.fit = lambda: setattr(self.native, "labels", [0, 0, 1, 1, 2])
+        self.engine._has_clustering = False
+        self.engine._has_vectors = True
+        self.modules["truememory.agentic_search"] = types.SimpleNamespace(normalize_scores=lambda rows: rows)
+        with patch.object(self.coordinator, "request_layers") as request, \
+             patch.object(self.engine, "search", return_value=[{"id": 1, "score": .5}]), \
+             patch.object(self.engine, "_entity_focused_search", return_value=[]), \
+             patch.object(self.engine, "_check_sufficiency", return_value=True), \
+             patch.object(self.engine, "_apply_surprise_boost", side_effect=lambda rows: rows), \
+             patch.object(self.engine, "_clean_results", side_effect=lambda rows, *_args, **_kwargs: rows), \
+             patch.object(self.engine_module, "search_clustered", return_value=[{"id": 2, "score": .5}], create=True) as search:
+            self.engine._maybe_auto_consolidate()
+            request.assert_not_called()
+            self.engine.search_agentic("synthetic query", use_hyde=False, use_reranker=False)
+            search.assert_not_called()
+            result = self.engine.consolidate()
+            self.assertIn("3 clusters", result["cluster_messages"])
+            self.assertTrue(self.engine._has_clustering)
+            self.assertTrue(self.engine._has_consolidation)
+            rows = self.engine.search_agentic("synthetic query", use_hyde=False, use_reranker=False)
+            search.assert_called_once()
+            self.assertEqual([row["id"] for row in rows], [1, 2])
+            request.assert_not_called()
+            self.conn.execute("UPDATE messages SET content='synthetic repaired correction' WHERE id=1")
+            self.conn.commit()
+            self.engine._maybe_auto_consolidate()
+            request.assert_called_once_with(threshold=25)
+
+    def test_failed_manual_repair_does_not_enable_capabilities_or_remove_prior_success(self) -> None:
+        self.engine._has_clustering = False
+        with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic")), \
+             patch.object(self.modules["truememory.consolidation"], "build_summaries", side_effect=RuntimeError("synthetic")):
+            result = self.engine.consolidate()
+            self.assertIn("ERROR", result["cluster_messages"])
+            self.assertIn("ERROR", result["build_summaries"])
+            self.assertFalse(self.engine._has_clustering)
+            self.assertFalse(self.engine._has_consolidation)
+            self.engine._has_clustering = self.engine._has_consolidation = True
+            self.engine.consolidate()
+            self.assertTrue(self.engine._has_clustering)
+            self.assertTrue(self.engine._has_consolidation)
+
+    def test_unavailable_manual_cluster_repair_preserves_only_prior_capability(self) -> None:
+        self.engine._has_clustering = False
+        with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")):
+            self.assertIn("UNAVAILABLE", self.engine.consolidate()["cluster_messages"])
+            self.assertFalse(self.engine._has_clustering)
+            self.engine._has_clustering = True
+            self.assertIn("UNAVAILABLE", self.engine.consolidate()["cluster_messages"])
+            self.assertTrue(self.engine._has_clustering)
+
+    def test_first_open_manual_does_not_start_its_own_competitor(self) -> None:
+        self.add(1)
+        engine = self.engine_module.TrueMemoryEngine(self.path)
+        self.addCleanup(engine.close)
+        with patch.object(self.coordinator, "request_layers") as request:
+            result = engine.consolidate()
+            request.assert_not_called()
+        self.assertTrue(engine.ready)
+        self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS))
+
+    def test_current_empty_startup_and_rolled_back_correction_do_not_schedule(self) -> None:
+        self.engine.consolidate()
+        self.engine._has_consolidation = True
+        with patch.object(self.coordinator, "request_layers") as request:
+            self.engine._maybe_startup_consolidate()
+            self.engine._ensure_connection()
+            request.assert_not_called()
+        self.add(1)
+        self.engine.consolidate()
+        with patch.object(self.coordinator, "request_layers") as request:
+            self.conn.execute("UPDATE messages SET content='synthetic uncommitted'")
+            self.engine._ensure_connection()
+            self.assertEqual(self.engine._maintenance_pending_reason, "pending_caller_commit")
+            self.conn.rollback()
+            self.engine._ensure_connection()
+            request.assert_not_called()
+            self.conn.execute("UPDATE messages SET content='synthetic committed'")
+            self.conn.commit()
+            self.engine._ensure_connection()
+            request.assert_called_once_with(threshold=25)
+
+    def test_add_probe_uses_no_corpus_scan_native_import_or_blocking_model_lock(self) -> None:
+        self.engine.consolidate()
+        self.engine._has_consolidation = True
+        operations = []
+        underlying = self.vector._lock
+
+        class ProbeLock:
+            def acquire(lock, blocking=True):
+                self.assertFalse(blocking)
+                return underlying.acquire(blocking=False)
+
+            def release(lock):
+                underlying.release()
+
+        def authorize(action, table, column, *_args):
+            if action == sqlite3.SQLITE_READ:
+                operations.append((table, column))
+            return sqlite3.SQLITE_OK
+
+        self.conn.set_authorizer(authorize)
+        try:
+            with patch.object(self.vector, "_lock", ProbeLock()), \
+                 patch.object(MAINTENANCE, "_installed_dependency", side_effect=AssertionError("per-add package scan")), \
+                 patch.object(self.coordinator, "request_layers", side_effect=AssertionError("unexpected work")):
+                self.engine._maybe_auto_consolidate()
+                self.assertEqual(self.engine._maintenance_pending_reason, "awaiting_eligibility")
+        finally:
+            self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK)
+        self.assertFalse({table for table, _ in operations}.intersection({"messages", "vec_messages_edge", "summaries", "cluster_centroids"}))
+
+    def test_source_commit_followed_by_a17_vector_error_still_notifies(self) -> None:
+        self.add(1)
+        self.engine.consolidate()
+        self.engine._has_vectors = self.engine._has_consolidation = True
+        changed = type("VectorPublicationChanged", (RuntimeError,), {})
+
+        @contextmanager
+        def publication(*args):
+            raise changed("synthetic model replacement")
+            yield
+
+        def capture(conn: sqlite3.Connection) -> tuple[object, str]:
+            self.assertIs(conn, self.engine.conn)
+            return object(), "synthetic-identity"
+
+        fields = {"_capture_rebuild_model": capture,
+                  "_encode_with_mps_fallback": lambda *args: [[1., 2.]],
+                  "_build_sep_text": lambda *args: "synthetic separated text",
+                  "_foreground_vector_publication": publication,
+                  "serialize_f32": lambda values: b"synthetic-unused-vector",
+                  "_write_embedder_metadata_no_commit": lambda conn: None,
+                  "VectorPublicationChanged": changed}
+        with patch.dict(self.vector.__dict__, fields), patch.object(self.coordinator, "request_layers") as request:
+            with self.assertRaisesRegex(changed, "Source update committed"):
+                self.engine.update(1, content="synthetic corrected source")
+            request.assert_called_once_with(threshold=25)
+        self.assertFalse(self.conn.in_transaction)
+        self.assertFalse(self.engine._write_lock.locked())
+        self.assertEqual(self.conn.execute("SELECT content FROM messages WHERE id=1").fetchone()[0], "synthetic corrected source")
+
+    def test_manual_busy_is_truthful_and_does_not_queue_forced_work(self) -> None:
+        entered, release = threading.Event(), threading.Event()
+
+        def active(conn, cancel):
+            entered.set()
+            self.assertTrue(release.wait(2))
+
+        self.coordinator.request(active)
+        self.assertTrue(entered.wait(2))
+        try:
+            result = self.engine.consolidate()
+            self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS))
+            self.assertTrue(all(value.startswith("BUSY") for value in result.values()))
+            self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0)
+        finally:
+            release.set()
+            self.completed()
+
+    def test_file_manual_uses_dedicated_connection_all_eight_and_preferences(self) -> None:
+        self.add(5)
+        seen = []
+        original = MAINTENANCE.run_engine_maintenance
+
+        def run(conn, coordinator, **kwargs):
+            seen.append(conn)
+            self.assertIsNot(conn, self.conn)
+            self.assertTrue(self.engine._write_lock.acquire(blocking=False))
+            self.engine._write_lock.release()
+            return original(conn, coordinator, **kwargs)
+
+        with patch.object(MAINTENANCE, "run_engine_maintenance", run), \
+             patch.object(self.coordinator, "request_layers") as automatic:
+            result = self.engine.consolidate()
+            automatic.assert_not_called()
+        self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS))
+        self.assertTrue(all("ERROR" not in value and "UNAVAILABLE" not in value for value in result.values()), result)
+        self.assertIn("vector_generation_unverified", result["cluster_messages"])
+        with self.assertRaises(sqlite3.ProgrammingError):
+            seen[0].execute("SELECT 1")
+
+    def test_24_then_25_committed_adds_across_engines_use_durable_counts(self) -> None:
+        self.engine.consolidate()
+        second = self.new_engine()
+        self.engine._has_consolidation = second._has_consolidation = True
+        with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request:
+            for index in range(24):
+                engine = self.engine if index % 2 else second
+                engine.add("synthetic append", sender="synthetic-sender")
+            request.assert_not_called()
+            second.add("synthetic twenty fifth", sender="synthetic-sender")
+            self.completed()
+            self.assertGreaterEqual(request.call_count, 1)
+        states = MAINTENANCE.read_layer_states(self.conn, self.specs())
+        self.assertTrue(all(state.attempted_insert_count == 25 for state in states.values()), states)
+
+    def test_borrowed_manual_preserves_caller_writes_and_rollback(self) -> None:
+        self.add(1, commit=False)
+        with patch.object(self.engine_module, "create_db", side_effect=AssertionError("new handle")):
+            result = self.engine.consolidate()
+        self.assertTrue(all("pending caller commit" in value for value in result.values()), result)
+        self.assertTrue(self.conn.in_transaction)
+        self.conn.rollback()
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0)
+        self.assertEqual(self.output(), ([], []))
+
+    def test_in_memory_auto_stays_pending_and_manual_uses_original_rows(self) -> None:
+        conn = STORAGE.create_db(":memory:")
+        self.addCleanup(conn.close)
+        engine = self.engine_module.TrueMemoryEngine(":memory:")
+        engine.conn, engine.ready = conn, True
+        self.addCleanup(engine.close)
+        conn.execute("INSERT INTO messages(content,timestamp) VALUES ('synthetic in memory','2026-01-01')")
+        conn.commit()
+        with patch.object(MAINTENANCE, "create_db", side_effect=AssertionError("new handle")), \
+             patch.object(MAINTENANCE.threading, "Thread", side_effect=AssertionError("memory worker")):
+            engine._maybe_auto_consolidate()
+            self.assertEqual(engine._maintenance_coordinator.status, ("pending_in_memory", None))
+            result = engine.consolidate()
+        self.assertIn("1 episodes", result["detect_episodes"])
+        self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1)
+
+
+class TestCapabilityEvidence(RoutingFixture):
+    def test_transient_model_busy_cannot_clear_prior_extension_failure_baseline(self) -> None:
+        self.add(5)
+        with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")):
+            self.engine.consolidate()
+            prior = self.coordinator.capability_snapshot()[2]
+            self.add(1)
+            worker = MAINTENANCE.create_db(self.path)
+            try:
+                with self.vector._lock:
+                    MAINTENANCE._prepare_worker_extensions(worker, self.coordinator)
+            finally:
+                worker.close()
+        self.assertEqual(self.coordinator.capability_snapshot()[2], prior)
+        self.assertEqual(MAINTENANCE.plan_layers(self.conn, MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)), ())
+
+    def test_reopen_during_worker_failure_preserves_current_run_local_evidence(self) -> None:
+        self.add(1)
+
+        def fail_and_refresh(conn):
+            self.coordinator.refresh_capabilities()
+            raise RuntimeError("synthetic load failure after reopen")
+
+        with patch.object(self.modules["sqlite_vec"], "load", fail_and_refresh):
+            result = self.engine.consolidate()
+        self.assertIn("VectorExtensionUnavailable", result["cluster_messages"])
+        self.assertIsNone(self.coordinator.capability_snapshot()[2])
+
+    def test_worker_extension_failure_keeps_same_key_and_25_insert_baseline(self) -> None:
+        self.add(5)
+        loader = self.modules["sqlite_vec"]
+        with patch.object(loader, "load", side_effect=RuntimeError("synthetic private sentinel")):
+            result = self.engine.consolidate()
+            self.assertIn("VectorExtensionUnavailable", result["cluster_messages"])
+            self.engine._has_consolidation = True
+            with patch.object(self.coordinator, "request_layers") as request:
+                self.engine.add("synthetic sixth")
+                self.engine._ensure_connection()
+                request.assert_not_called()
+            state = MAINTENANCE.read_layer_states(self.conn, MAINTENANCE.engine_layer_specs(self.conn, self.coordinator))["clusters"]
+            self.assertEqual(state.attempted_insert_count, 5)
+            self.assertEqual(state.dependency.key, state.attempted_dependency)
+        repaired = self.engine.consolidate()
+        self.assertNotIn("UNAVAILABLE", repaired["cluster_messages"])
+        self.assertIsNone(self.coordinator.capability_snapshot()[2])
+
+    def test_reopen_refreshes_static_evidence_and_new_model_invalidates_failure_key(self) -> None:
+        self.add(1)
+        with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")):
+            self.engine.consolidate()
+        epoch, versions, failure = self.coordinator.capability_snapshot()
+        self.assertEqual(len(versions), 3)
+        self.assertIsNotNone(failure)
+        self.vector.EMBEDDING_MODEL = "synthetic-model-c"
+        dependency = MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)[0].resolve_dependency()
+        self.assertTrue(dependency.available)
+        engine = self.new_engine()
+        self.assertIs(engine._get_maintenance_coordinator(), self.coordinator)
+        self.assertGreater(self.coordinator.capability_snapshot()[0], epoch)
+        self.assertIsNone(self.coordinator.capability_snapshot()[2])
+
+    def test_old_worker_cannot_publish_capability_error_after_refresh(self) -> None:
+        epoch, _, _ = self.coordinator.capability_snapshot()
+        failure = MAINTENANCE.make_layer_dependency(1, available=False, error_category="VectorExtensionUnavailable")
+        self.coordinator.refresh_capabilities()
+        self.coordinator._publish_extension_evidence(epoch, failure)
+        self.assertIsNone(self.coordinator.capability_snapshot()[2])
+
+    def test_actual_connection_unavailability_is_not_overridden_by_cached_success(self) -> None:
+        dependency = self.spec().resolve_dependency()
+        unavailable = MAINTENANCE.make_layer_dependency(1, MAINTENANCE._dependency_parameters(dependency),
+                                                       available=False, error_category="VectorExtensionUnavailable")
+        with patch.object(MAINTENANCE, "_cluster_schedule_dependency", return_value=unavailable):
+            observed = MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)[0].resolve_dependency()
+        self.assertEqual(observed, unavailable)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test-maintenance-fixture-isolation-753.py b/tests/test-maintenance-fixture-isolation-753.py
new file mode 100644
index 00000000..f52397c5
--- /dev/null
+++ b/tests/test-maintenance-fixture-isolation-753.py
@@ -0,0 +1,220 @@
+"""Cross-fixture worker isolation using actual fixtures and SQLite in memory."""
+
+import ast
+import builtins
+import sqlite3
+import sys
+import threading
+import types
+import unittest
+import uuid
+from collections.abc import Callable
+from pathlib import Path
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def fixture_loader(process_sys: object) -> Callable[[str], dict]:
+    """Keep fixture imports and module registries inside one synthetic process."""
+    loaded = {}
+
+    def safe_import(name: str, globals: dict | None = None, locals: dict | None = None,
+                    fromlist: tuple = (), level: int = 0) -> object:
+        if name == "sys":
+            return process_sys
+        if name == "runpy":
+            return types.SimpleNamespace(run_path=load)
+        if name == "truememory.personality_style_vec":
+            return process_sys.modules[name]
+        if name.startswith("truememory") or name.split(".")[0] in {"torch", "numpy", "hdbscan", "sqlite_vec"}:
+            raise AssertionError("Unexpected application/native import: " + name)
+        return builtins.__import__(name, globals, locals, fromlist, level)
+
+    def safe_exec(code: object, namespace: dict) -> None:
+        namespace.setdefault("__builtins__", fixture_builtins)
+        exec(code, namespace)
+
+    fixture_builtins = dict(vars(builtins), __import__=safe_import, exec=safe_exec)
+
+    def load(path: str) -> dict:
+        path = Path(path)
+        if path.name in loaded:
+            return loaded[path.name]
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        # Existing fixture loaders also read production source. Pin their
+        # decoding in this harness on Python 3.10/Windows without editing it.
+        for node in ast.walk(tree):
+            if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
+                    and node.func.attr == "read_text" and not node.args
+                    and not any(keyword.arg == "encoding" for keyword in node.keywords)):
+                node.keywords.append(ast.keyword(arg="encoding", value=ast.Constant(value="utf-8")))
+        namespace = {"__file__": str(path), "__name__": "synthetic_fixture_" + path.stem,
+                     "__builtins__": fixture_builtins}
+        exec(compile(ast.fix_missing_locations(tree), str(path), "exec"), namespace)
+        loaded[path.name] = namespace
+        return namespace
+
+    return load
+
+
+class TestFixtureIsolation(unittest.TestCase):
+    def setUp(self) -> None:
+        values = vars(sys).copy()
+        values["modules"] = {}
+        self.process_sys = types.SimpleNamespace(**values)
+        load = fixture_loader(self.process_sys)
+        self.routing = load(str(ROOT / "tests/test-maintenance-engine-routing-753.py"))
+        self.adapters = self.routing["ADAPTERS"]
+        self.storage = self.routing["STORAGE"]
+        self.maintenance = self.routing["MAINTENANCE"]
+        _, self.foreign_vector, _ = self.adapters["CLUSTERS"]["load_clustering"]()
+        self.process_sys.modules["truememory.vector_search"] = self.foreign_vector
+        load_stdlib = self.routing["stdlib_module"]
+        style = load_stdlib("personality_style_vec", {})
+        self.process_sys.modules["truememory.personality_style_vec"] = style
+        self.enter_context(patch.dict(self.routing, {"stdlib_module": lambda name, modules:
+            style if name == "personality_style_vec" else load_stdlib(name, modules)}))
+        self.registry_before = self.process_sys.modules.copy()
+        # Primitive loaders have their own import guard. Bind their sys view
+        # explicitly so both modules start in the same synthetic process.
+        self.enter_context(patch.object(self.maintenance, "sys", self.process_sys))
+        self.foreign = types.ModuleType("synthetic_foreign_maintenance")
+        self.foreign.__dict__.update(self.maintenance.__dict__)
+        # Functions need their own globals to represent an unrelated worker.
+        source = (ROOT / "truememory/maintenance.py").read_text(encoding="utf-8")
+        exec(compile(source, "synthetic-foreign-maintenance", "exec"), self.foreign.__dict__)
+        self.foreign.sys = self.process_sys
+
+        self.uri = "file:synthetic-fixture-" + uuid.uuid4().hex + "?mode=memory&cache=shared"
+        self.connect = sqlite3.connect
+        self.keeper = self.connect(self.uri, uri=True, check_same_thread=False)
+        self.addCleanup(self.keeper.close)
+        self.create = self.storage.create_db
+        self.coordinator = self.maintenance.MaintenanceCoordinator(None)
+        fixture_test = self
+
+        class MemoryDirectory:
+            name = ":memory:"
+
+            def cleanup(self) -> None:
+                pass
+
+        class MemoryFixture(self.routing["RoutingFixture"]):
+            def open_db(self) -> sqlite3.Connection:
+                conn = fixture_test.open_memory()
+                self.addCleanup(conn.close)
+                return conn
+
+        self.enter_context(patch.dict(self.adapters, {"tempfile": types.SimpleNamespace(
+            TemporaryDirectory=lambda **kwargs: MemoryDirectory())}))
+        self.enter_context(patch.object(self.storage, "create_db", self.open_memory))
+        self.enter_context(patch.object(self.maintenance, "get_coordinator", lambda path: self.coordinator))
+        self.fixture = MemoryFixture()
+        self.addCleanup(self.fixture.doCleanups)
+        self.fixture.setUp()
+
+    def enter_context(self, context: object) -> object:
+        result = context.__enter__()
+        self.addCleanup(context.__exit__, None, None, None)
+        return result
+
+    def open_memory(self, path: object = None) -> sqlite3.Connection:
+        sqlite_view = types.SimpleNamespace(**dict(vars(sqlite3), connect=lambda *args, **kwargs:
+            self.connect(self.uri, uri=True, check_same_thread=False)))
+        with patch.object(self.storage, "sqlite3", sqlite_view):
+            return self.create(":memory:")
+
+    def states(self, conn: sqlite3.Connection | None = None) -> dict:
+        conn = conn or self.fixture.engine.conn
+        specs = self.maintenance.engine_layer_specs(conn, self.coordinator)
+        return self.maintenance.read_layer_states(conn, specs)
+
+    def reopen(self) -> object:
+        self.fixture.engine.close()
+        reopened = self.fixture.new_engine()
+        reopened._has_consolidation = True
+        return reopened
+
+    def test_fixture_does_not_publish_its_vector_into_process_registry(self) -> None:
+        self.assertIs(self.process_sys.modules["truememory.vector_search"], self.foreign_vector)
+        self.assertIs(self.maintenance.sys.modules["truememory.vector_search"], self.fixture.vector)
+        self.assertIsNot(self.maintenance.sys.modules, self.process_sys.modules)
+        self.assertEqual(self.process_sys.modules, self.registry_before)
+
+    def test_interloper_cannot_defer_bootstrap_or_reset_reopen_threshold(self) -> None:
+        foreign_conn = self.connect(":memory:", check_same_thread=False)
+        self.fixture.conn.backup(foreign_conn)
+        entered, release = threading.Event(), threading.Event()
+        observed = []
+
+        class BlockedSnapshot:
+            def __getattr__(boundary, name: str) -> object:
+                return getattr(foreign_conn, name)
+
+            def execute(boundary, sql: str, *args: object) -> sqlite3.Cursor:
+                if sql.startswith("SELECT type,name,rootpage,sql"):
+                    entered.set()
+                    if not release.wait(3):
+                        raise AssertionError("Synthetic worker was not released")
+                return foreign_conn.execute(sql, *args)
+
+        worker_coordinator = self.foreign.MaintenanceCoordinator(None)
+
+        def work(conn: sqlite3.Connection, cancel: threading.Event) -> object:
+            observed.append(self.foreign.sys.modules["truememory.vector_search"])
+            self.foreign._cluster_schedule_dependency(conn, versions=dict(self.fixture.versions))
+            return self.foreign.MaintenanceReport((), "SKIPPED (synthetic)")
+
+        with patch.object(self.foreign, "create_db", lambda path: BlockedSnapshot()), \
+             patch.object(self.foreign, "_release_owner", lambda fd: None):
+            worker = threading.Thread(target=worker_coordinator._run, args=(-1, work))
+            worker.start()
+            try:
+                self.assertTrue(entered.wait(3))
+                result = self.fixture.engine.consolidate()
+                states = self.states()
+            finally:
+                release.set()
+                worker.join(3)
+                foreign_conn.close()
+            self.assertFalse(worker.is_alive())
+        self.assertEqual(worker_coordinator.status, ("success", None))
+        self.assertNotIn("DEFERRED", result["cluster_messages"])
+        self.assertEqual(observed, [self.foreign_vector])
+        self.assertEqual(len(states), 8)
+        self.assertTrue(all(state.attempted_insert_count == 0 for state in states.values()), states)
+        self.assertTrue(all(state.outcome == "success_empty" for state in states.values()), states)
+        self.fixture.add(24)
+        before = self.maintenance.read_source_revision(self.fixture.conn)
+        versions = self.coordinator.capability_snapshot()[1]
+        reopened = self.reopen()
+        with patch.object(self.coordinator, "request_layers") as request:
+            reopened._maybe_auto_consolidate()
+            request.assert_not_called()
+            self.assertEqual(self.maintenance.read_source_revision(reopened.conn), before)
+            self.assertEqual(self.coordinator.capability_snapshot()[1], versions)
+            reopened.add("synthetic twenty fifth")
+            request.assert_called_once_with(threshold=25)
+
+    def test_intentionally_deferred_bootstrap_still_recovers_at_twenty_four(self) -> None:
+        with self.fixture.vector._lock:
+            result = self.fixture.engine.consolidate()
+        self.assertEqual(result["cluster_messages"], "DEFERRED (ModelBusy)")
+        self.assertIsNone(self.states()["clusters"].attempted_insert_count)
+        self.fixture.add(24)
+        reopened = self.reopen()
+        with patch.object(self.coordinator, "request_layers") as request:
+            reopened._maybe_auto_consolidate()
+            request.assert_called_once_with(threshold=25)
+        report = self.maintenance.run_engine_maintenance(
+            reopened.conn, self.coordinator, threshold=25, prepare_extensions=False)
+        self.assertEqual([item.layer for item in report.results if item.attempted], ["clusters"])
+        states = self.states(reopened.conn)
+        self.assertEqual(states["clusters"].attempted_insert_count, 24)
+        self.assertTrue(all(state.attempted_insert_count == 0 for name, state in states.items() if name != "clusters"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test-maintenance-foreground-admission-753.py b/tests/test-maintenance-foreground-admission-753.py
new file mode 100644
index 00000000..d295387e
--- /dev/null
+++ b/tests/test-maintenance-foreground-admission-753.py
@@ -0,0 +1,491 @@
+"""Foreground publication admission with real SQLite and synthetic models."""
+
+import runpy
+import sqlite3
+import threading
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+FOREGROUND = runpy.run_path(str(Path(__file__).with_name("test-rebuild-foreground-publication-756.py")))
+CLUSTERS = runpy.run_path(str(Path(__file__).with_name("test-cluster-publication-749.py")))
+
+
+class AdmissionFixture(unittest.TestCase):
+    setUp = FOREGROUND["TestForegroundPublication"].setUp
+    seed = FOREGROUND["TestForegroundPublication"].seed
+    ids = FOREGROUND["TestForegroundPublication"].ids
+    writer = FOREGROUND["TestForegroundPublication"].writer
+    engine = FOREGROUND["TestForegroundPublication"].engine
+
+    def start(self, action, name="synthetic-worker"):
+        results, done = [], threading.Event()
+
+        def run():
+            try:
+                results.append(action())
+            except BaseException as error:
+                results.append(error)
+            finally:
+                done.set()
+
+        thread = threading.Thread(target=run, name=name, daemon=True)
+        thread.start()
+        self.addCleanup(thread.join, 3)
+        return thread, results, done
+
+    def released(self):
+        self.assertTrue(self.vector._lock.acquire(blocking=False))
+        self.vector._lock.release()
+
+    def sentinel(self):
+        self.conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-caller','pending')")
+
+    def preserved(self):
+        self.assertTrue(self.conn.in_transaction)
+        self.assertEqual(self.conn.execute("SELECT value FROM metadata WHERE key='synthetic-caller'").fetchone()[0], "pending")
+
+
+class TestForegroundAdmission(AdmissionFixture):
+    def test_style_computation_runs_outside_capture_and_publication_locks(self) -> None:
+        engine = self.engine()
+        engine._has_style_vec = True
+        events = []
+        style = [0.25, 0.75]
+
+        def compute(content: str) -> list[float]:
+            self.assertFalse(engine._write_lock.locked())
+            self.assertFalse(self.vector._lock.locked())
+            self.assertFalse(self.conn.in_transaction)
+            events.append("compute")
+            return style
+
+        def publish(conn: sqlite3.Connection, sender: str, content: str,
+                    _pre_computed_vec: list[float]) -> None:
+            self.assertTrue(engine._write_lock.locked())
+            self.assertTrue(self.vector._lock.locked())
+            self.assertTrue(conn.in_transaction)
+            self.assertIs(_pre_computed_vec, style)
+            events.append("publish")
+
+        module = types.ModuleType("synthetic_style")
+        module.compute_style_vector = compute
+        with patch.dict(self.modules, personality_style_vec=module), \
+                patch.dict(engine.add.__globals__, _update_style_vec=publish):
+            engine.add("synthetic styled content", sender="synthetic sender")
+        self.assertEqual(events, ["compute", "publish"])
+        self.assertEqual(self.ids(), [1])
+        self.assertEqual(self.ids("vec_messages_sep"), [1])
+
+    def test_actual_cluster_snapshot_delays_add_before_writer_then_add_commits_both_vectors(self):
+        engine = self.engine()
+        cluster, runtime, _native = CLUSTERS["load_clustering"]()
+        encoding, snapshot, admission, release = (threading.Event() for _ in range(4))
+        underlying = self.vector._lock
+        fixture = self
+
+        class ObservedLock:
+            def acquire(self, blocking=True):
+                if threading.current_thread().name == "synthetic-add" and snapshot.is_set():
+                    fixture.assertTrue(blocking)
+                    fixture.assertFalse(engine.conn.in_transaction)
+                    fixture.assertTrue(engine._write_lock.locked())
+                    admission.set()
+                return underlying.acquire(blocking=blocking)
+
+            def release(self):
+                underlying.release()
+
+            def __enter__(self):
+                self.acquire()
+                return self
+
+            def __exit__(self, *_args):
+                self.release()
+
+        self.vector._lock = runtime._lock = ObservedLock()
+        runtime.EMBEDDING_MODEL = self.vector.EMBEDDING_MODEL
+        cluster_conn, independent = self.writer(), self.writer()
+        original = cluster._get_all_embeddings
+
+        def read(conn):
+            snapshot.set()
+            self.assertTrue(release.wait(3))
+            return original(conn)
+
+        def encode(texts):
+            if len(self.calls) == 1:
+                encoding.set()
+                self.assertTrue(snapshot.wait(3))
+            return [[1., 0.] for _ in texts]
+
+        def consolidate():
+            self.assertTrue(encoding.wait(3))
+            return cluster.cluster_messages(cluster_conn)
+
+        self.on_encode = encode
+        with patch.object(cluster, "_get_all_embeddings", read):
+            cluster_thread, cluster_result, cluster_done = self.start(consolidate)
+            add_thread, result, done = self.start(lambda: engine.add("synthetic first add"), "synthetic-add")
+            try:
+                self.assertTrue(admission.wait(2))
+                self.assertFalse(done.is_set())
+                independent.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-independent','committed')")
+                independent.commit()
+                self.assertFalse(self.conn.in_transaction)
+            finally:
+                release.set()
+                add_thread.join(3)
+                cluster_thread.join(3)
+        self.assertTrue(done.is_set())
+        self.assertTrue(cluster_done.is_set())
+        self.assertIsInstance(cluster_result[0], (int, RuntimeError), cluster_result)
+        self.assertIsInstance(result[0], dict, result)
+        self.assertEqual(self.ids(), [1])
+        self.assertEqual(self.ids("vec_messages_sep"), [1])
+        self.assertEqual(self.conn.execute("SELECT content FROM messages").fetchone()[0], "synthetic first add")
+        self.released()
+
+    def test_writer_owner_defers_nonblocking_model_fence_and_foreground_finishes(self):
+        owner = self.writer()
+        owner.execute("BEGIN IMMEDIATE")
+        admission, attempted = threading.Event(), threading.Event()
+        underlying = self.vector._lock
+
+        class AdmittedLock:
+            def acquire(self, blocking=True):
+                acquired = underlying.acquire(blocking=blocking)
+                if acquired and threading.current_thread().name == "synthetic-publish":
+                    admission.set()
+                return acquired
+
+            def release(self):
+                underlying.release()
+
+        self.vector._lock = AdmittedLock()
+        identity = (0, self.vector.EMBEDDING_MODEL, 2)
+
+        def publish():
+            with self.vector._foreground_vector_publication(self.conn, identity, 1):
+                self.conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-published','yes')")
+
+        thread, result, done = self.start(publish, "synthetic-publish")
+        try:
+            self.assertTrue(admission.wait(2))
+            with self.assertRaisesRegex(RuntimeError, "model is busy"):
+                with self.vector._rebuild_model_fence(identity):
+                    attempted.set()
+            self.assertFalse(attempted.is_set())
+        finally:
+            owner.rollback()
+            thread.join(3)
+        self.assertTrue(done.is_set())
+        self.assertEqual(result, [None])
+        self.released()
+
+    def test_add_waiting_for_same_engine_update_does_not_own_model(self):
+        self.seed((1,))
+        engine = self.engine()
+        update_ready, add_waiting = threading.Event(), threading.Event()
+        underlying = engine._write_lock
+
+        class EngineLock:
+            def __enter__(self):
+                if threading.current_thread().name == "synthetic-add":
+                    add_waiting.set()
+                underlying.acquire()
+                return self
+
+            def __exit__(self, *_args):
+                underlying.release()
+
+        engine._write_lock = EngineLock()
+        original = engine.update.__globals__["update_message"]
+
+        def update(conn, memory_id, **fields):
+            result = original(conn, memory_id, **fields)
+            update_ready.set()
+            self.assertTrue(add_waiting.wait(3))
+            self.released()
+            return result
+
+        with patch.dict(engine.update.__globals__, update_message=update):
+            update_thread, updated, update_done = self.start(lambda: engine.update(1, "synthetic updated"), "synthetic-update")
+            self.assertTrue(update_ready.wait(2))
+            add_thread, added, add_done = self.start(lambda: engine.add("synthetic added"), "synthetic-add")
+            update_thread.join(3)
+            add_thread.join(3)
+        self.assertTrue(update_done.is_set())
+        self.assertTrue(add_done.is_set())
+        self.assertIsInstance(updated[0], dict, updated)
+        self.assertIsInstance(added[0], dict, added)
+        self.assertEqual(self.ids(), [1, 2])
+        self.assertEqual(self.ids("vec_messages_sep"), [1, 2])
+
+    def test_same_engine_writer_finishes_before_capture_classifies_transaction(self) -> None:
+        self.seed((1,))
+        for operation in ("add", "update"):
+            with self.subTest(operation=operation):
+                engine = self.engine()
+                writer_ready, waiting, release = (threading.Event() for _ in range(3))
+                underlying = engine._write_lock
+
+                class EngineLock:
+                    def __enter__(self) -> "EngineLock":
+                        if threading.current_thread().name == "synthetic-capture":
+                            waiting.set()
+                        underlying.acquire()
+                        return self
+
+                    def __exit__(self, *_args: object) -> None:
+                        underlying.release()
+
+                engine._write_lock = EngineLock()
+                self.vector._model = None
+                loads = []
+
+                def load() -> object:
+                    self.assertFalse(self.conn.in_transaction)
+                    self.assertTrue(underlying.locked())
+                    loads.append(True)
+                    self.vector._model = self.model
+                    return self.model
+
+                def writer() -> None:
+                    with engine._write_lock:
+                        self.conn.execute("BEGIN IMMEDIATE")
+                        self.conn.execute("INSERT INTO metadata(key,value) VALUES (?, 'pending')",
+                                          ("synthetic-owner-" + operation,))
+                        writer_ready.set()
+                        self.assertTrue(release.wait(3))
+                        self.conn.commit()
+
+                with patch.object(self.vector, "get_model", load):
+                    writer_thread, written, writer_done = self.start(writer)
+                    self.assertTrue(writer_ready.wait(2))
+                    action = (lambda: engine.add("synthetic added")) if operation == "add" else (
+                        lambda: engine.update(1, "synthetic updated")
+                    )
+                    mutation_thread, result, done = self.start(action, "synthetic-capture")
+                    try:
+                        self.assertTrue(waiting.wait(2))
+                        self.assertFalse(done.is_set())
+                        self.assertEqual(loads, [])
+                    finally:
+                        release.set()
+                        mutation_thread.join(3)
+                        writer_thread.join(3)
+                self.assertTrue(writer_done.is_set())
+                self.assertEqual(written, [None])
+                self.assertTrue(done.is_set())
+                self.assertIsInstance(result[0], dict, result)
+                self.assertEqual(loads, [True])
+                self.assertFalse(self.conn.in_transaction)
+                self.released()
+
+    def test_generation_change_during_model_load_rejects_add_and_update_before_source_write(self):
+        self.seed((1,))
+        engine = self.engine()
+
+        def changed():
+            self.vector._model_generation += 1
+            return self.model
+
+        with patch.object(self.vector, "get_model", changed):
+            for action in (lambda: engine.add("synthetic rejected"), lambda: engine.update(1, "synthetic rejected")):
+                with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "changed while loading"):
+                    action()
+        self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")])
+        self.assertEqual(self.ids(), [])
+        self.released()
+
+    def test_generation_change_while_waiting_rejects_before_opening_transaction(self):
+        captured = (0, self.vector.EMBEDDING_MODEL, 2)
+        attempted = threading.Event()
+        underlying = self.vector._lock
+
+        class WaitingLock:
+            def acquire(self, blocking=True):
+                attempted.set()
+                return underlying.acquire(blocking=blocking)
+
+            def release(self):
+                underlying.release()
+
+        self.vector._lock = WaitingLock()
+        underlying.acquire()
+
+        def publish():
+            with self.vector._foreground_vector_publication(self.conn, captured, 1):
+                self.fail("Changed generation entered publication")
+
+        thread, results, done = self.start(publish)
+        try:
+            self.assertTrue(attempted.wait(2))
+            self.assertFalse(done.is_set())
+            self.assertFalse(self.conn.in_transaction)
+            self.vector._model_generation += 1
+        finally:
+            underlying.release()
+            thread.join(3)
+        self.assertIsInstance(results[0], self.vector.VectorPublicationChanged)
+        self.assertIn("model changed", str(results[0]))
+        self.assertFalse(self.conn.in_transaction)
+        self.released()
+
+    def test_add_failure_after_source_insert_rolls_back_source_and_vectors(self):
+        engine = self.engine()
+        with patch.object(self.vector, "_validate_foreground_vector_target", side_effect=self.vector.VectorPublicationChanged("synthetic target changed")):
+            with self.assertRaises(self.vector.VectorPublicationChanged):
+                engine.add("synthetic rollback")
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0)
+        self.assertEqual(self.ids(), [])
+        self.assertEqual(self.ids("vec_messages_sep"), [])
+        self.assertFalse(self.conn.in_transaction)
+        self.released()
+
+    def test_admission_begin_and_commit_failures_release_model(self):
+        real = self.conn
+
+        class FailedConnection:
+            def __init__(self, phase):
+                self.phase = phase
+
+            def __getattr__(self, name):
+                return getattr(real, name)
+
+            def execute(self, sql, *args):
+                if self.phase == "begin" and sql == "BEGIN IMMEDIATE":
+                    raise sqlite3.OperationalError("synthetic writer admission")
+                return real.execute(sql, *args)
+
+            def commit(self):
+                if self.phase == "commit":
+                    raise sqlite3.OperationalError("synthetic commit failure")
+                return real.commit()
+
+        for phase in ("begin", "commit"):
+            with self.subTest(phase=phase):
+                with self.assertRaises(sqlite3.OperationalError):
+                    self.engine(FailedConnection(phase)).add("synthetic rollback")
+                self.assertFalse(real.in_transaction)
+                self.assertEqual(real.execute("SELECT count(*) FROM messages").fetchone()[0], 0)
+                self.assertEqual(self.ids(), [])
+                self.released()
+
+    def test_owned_model_load_and_encoding_finish_before_writer(self):
+        def load():
+            self.assertFalse(self.conn.in_transaction)
+            return self.model
+
+        def encode(texts):
+            self.assertFalse(self.conn.in_transaction)
+            self.released()
+            return [[1., 0.] for _ in texts]
+
+        self.on_encode = encode
+        with patch.object(self.vector, "get_model", load):
+            self.engine().add("synthetic outside writer")
+        self.assertEqual(self.ids(), [1])
+
+
+class TestBorrowedAdmission(AdmissionFixture):
+    def actions(self):
+        engine = self.engine()
+        return (
+            ("add", lambda: engine.add("synthetic rejected")),
+            ("update", lambda: engine.update(1, "synthetic rejected")),
+            ("embed_single", lambda: self.vector.embed_single(self.conn, 1, "synthetic-1")),
+            ("completion", lambda: self.vector.build_vectors(self.conn)),
+            ("separation", lambda: self.vector.build_separation_vectors(self.conn)),
+        )
+
+    def test_busy_capture_defers_every_mutation_without_waiting_or_loading(self):
+        self.seed((1,))
+        self.sentinel()
+        calls = []
+
+        class BusyLock:
+            def acquire(self, blocking=True):
+                calls.append(blocking)
+                if blocking:
+                    raise AssertionError("Borrowed caller attempted a blocking model acquisition")
+                return False
+
+        with patch.object(self.vector, "_lock", BusyLock()), patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")):
+            for name, action in self.actions():
+                with self.subTest(action=name):
+                    with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "model is busy"):
+                        action()
+                    self.preserved()
+        self.assertEqual(calls, [False] * 5)
+        self.assertEqual(self.calls, [])
+        self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")])
+        self.assertEqual(self.ids(), [])
+        self.conn.rollback()
+        self.released()
+
+    def test_unloaded_capture_defers_every_mutation_and_preserves_sentinel(self):
+        self.seed((1,))
+        self.sentinel()
+        self.vector._model = None
+        with patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")) as loader:
+            for name, action in self.actions():
+                with self.subTest(action=name):
+                    with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "not loaded"):
+                        action()
+                    self.preserved()
+                    self.released()
+            loader.assert_not_called()
+        self.assertEqual(self.calls, [])
+        self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")])
+        self.conn.rollback()
+
+    def test_borrowed_loaded_snapshot_skips_loader_and_add_remains_rollbackable(self):
+        self.sentinel()
+        with patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")) as loader:
+            item = self.engine().add("synthetic borrowed")
+            loader.assert_not_called()
+        self.preserved()
+        self.assertEqual(self.ids(), [item["id"]])
+        self.assertEqual(self.ids("vec_messages_sep"), [item["id"]])
+        self.conn.rollback()
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0)
+        self.assertEqual(self.ids(), [])
+        self.released()
+
+    def test_borrowed_publication_contention_preserves_caller_transaction(self):
+        self.sentinel()
+        held, release = threading.Event(), threading.Event()
+
+        def hold():
+            with self.vector._lock:
+                held.set()
+                self.assertTrue(release.wait(3))
+
+        holder = []
+
+        def encode(texts):
+            if len(self.calls) == 2:
+                holder.append(self.start(hold)[0])
+                self.assertTrue(held.wait(2))
+            return [[1., 0.] for _ in texts]
+
+        self.on_encode = encode
+        try:
+            with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "model is busy"):
+                self.engine().add("synthetic rejected")
+            self.preserved()
+            self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0)
+        finally:
+            release.set()
+            for thread in holder:
+                thread.join(3)
+        self.conn.rollback()
+        self.released()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test-rebuild-foreground-publication-756.py b/tests/test-rebuild-foreground-publication-756.py
index 6c71ac5e..5e8b0b7d 100644
--- a/tests/test-rebuild-foreground-publication-756.py
+++ b/tests/test-rebuild-foreground-publication-756.py
@@ -273,7 +273,7 @@ def __getattr__(self, name):
 
             def execute(self, sql, parameters=()):
                 if sql == "BEGIN IMMEDIATE" or sql == "UPDATE messages SET id = id WHERE 0":
-                    case.assertFalse(case.vector._lock.locked())
+                    case.assertTrue(case.vector._lock.locked())
                     boundaries.append("writer")
                 if sql == "RELEASE SAVEPOINT rebuild_source":
                     case.assertTrue(case.vector._lock.locked())
@@ -296,13 +296,13 @@ def commit(self):
     def test_embed_owned_commit_stays_inside_model_fence(self):
         self._assert_foreground_writer_and_terminal_fence("embed", False)
 
-    def test_embed_caller_writer_upgrade_precedes_fence_and_release_stays_inside(self):
+    def test_embed_caller_fence_precedes_writer_upgrade_and_contains_release(self):
         self._assert_foreground_writer_and_terminal_fence("embed", True)
 
     def test_add_owned_commit_stays_inside_model_fence(self):
         self._assert_foreground_writer_and_terminal_fence("add", False)
 
-    def test_add_caller_writer_upgrade_precedes_fence_and_release_stays_inside(self):
+    def test_add_caller_fence_precedes_writer_upgrade_and_contains_release(self):
         self._assert_foreground_writer_and_terminal_fence("add", True)
 
     def test_denied_add_savepoint_release_rolls_back_only_add(self):
diff --git a/tests/test-rebuild-source-streaming-756.py b/tests/test-rebuild-source-streaming-756.py
index ccf10ac2..5ca9d5ff 100644
--- a/tests/test-rebuild-source-streaming-756.py
+++ b/tests/test-rebuild-source-streaming-756.py
@@ -78,6 +78,7 @@ def setUp(self):
         self.addCleanup(self.conn.close)
         self.conn.executescript("CREATE TABLE vec_messages(embedding BLOB); CREATE TABLE vec_messages_sep(embedding BLOB);")
         self.model = object()
+        self.vector._model = self.model
         self.vector.get_model = lambda: self.model
         self.calls = []
         self.on_encode = None
diff --git a/tests/test_consolidation_gaps.py b/tests/test_consolidation_gaps.py
index 3c92e12c..5344444d 100644
--- a/tests/test_consolidation_gaps.py
+++ b/tests/test_consolidation_gaps.py
@@ -1,134 +1,60 @@
-"""Regression tests for consolidation gaps (clustering + preferences).
-
-Verifies that consolidate() calls cluster_messages and extract_preferences —
-features that were previously only available via ingest().
-"""
-from __future__ import annotations
-
-import tempfile
-import os
-from unittest.mock import patch, MagicMock
-
-
-from truememory.engine import TrueMemoryEngine
-from truememory.storage import create_db
-
-
-def _make_engine_with_messages(n=20):
-    """Create an engine with messages in a temp DB (no vectors needed).
-
-    Disables startup consolidation to prevent background threads from
-    accessing SQLite cross-thread (causes segfault on Python 3.14).
-    """
-    td = tempfile.mkdtemp()
-    db = os.path.join(td, "test.db")
-    conn = create_db(db)
-    for i in range(n):
-        conn.execute(
-            "INSERT INTO messages (content, sender, recipient, timestamp, category, modality) "
-            "VALUES (?, ?, '', '', '', '')",
-            (f"Message {i} about topic {i % 5}", "alice" if i % 2 == 0 else "bob"),
-        )
-    conn.commit()
-    conn.close()
-
-    eng = TrueMemoryEngine(db_path=db)
-    eng._has_consolidation = False
-    eng._ensure_connection()
-    return eng, td
-
-
-def test_consolidate_calls_cluster_messages():
-    """consolidate() must call cluster_messages when clustering is available."""
-    eng, td = _make_engine_with_messages()
-    eng._has_vectors = True
-
-    with patch("truememory.engine._HAS_CLUSTERING", True), \
-         patch("truememory.clustering.cluster_messages", return_value=3) as mock_cm:
-        result = eng.consolidate()
-
-    assert "cluster_messages" in result
-    assert "ERROR" not in result["cluster_messages"]
-    assert "3 clusters" in result["cluster_messages"]
-    mock_cm.assert_called_once_with(eng.conn)
-    eng.close()
-
-
-def test_consolidate_calls_extract_preferences():
-    """consolidate() must call extract_preferences when personality is available."""
-    eng, td = _make_engine_with_messages()
-
-    with patch("truememory.engine._HAS_PERSONALITY", True), \
-         patch("truememory.personality.extract_preferences", return_value={}) as mock_ep:
-        result = eng.consolidate()
-
-    assert "extract_preferences" in result
-    assert "ERROR" not in result["extract_preferences"]
-    mock_ep.assert_called_once_with(eng.conn)
-    eng.close()
-
-
-def test_consolidate_skips_clustering_without_vectors():
-    """consolidate() must skip clustering when vectors are unavailable."""
-    eng, td = _make_engine_with_messages()
-    eng._has_vectors = False
-
-    result = eng.consolidate()
-    assert "cluster_messages" not in result
-    eng.close()
-
-
-def test_consolidate_handles_cluster_error_gracefully():
-    """consolidate() must catch and report clustering errors."""
-    eng, td = _make_engine_with_messages()
-    eng._has_vectors = True
-
-    with patch("truememory.engine._HAS_CLUSTERING", True), \
-         patch("truememory.clustering.cluster_messages", side_effect=RuntimeError("boom")):
-        result = eng.consolidate()
-
-    assert "cluster_messages" in result
-    assert "ERROR" in result["cluster_messages"]
-    eng.close()
-
-
-def test_auto_consolidation_threshold_lowered():
-    """Default auto-consolidation threshold is 25 (was 100)."""
-    eng, td = _make_engine_with_messages()
-    assert eng._auto_consolidate_threshold == 25
-    eng.close()
-
-
-def test_startup_consolidation_triggers_when_clusters_empty():
-    """On startup, if messages exist but clusters are empty, consolidate fires."""
-    eng, td = _make_engine_with_messages(n=30)
-    eng._has_consolidation = True
-
-    import truememory.engine as _eng_mod
-    with patch.object(_eng_mod.threading, "Thread") as mock_thread_cls:
-        mock_thread = MagicMock()
-        mock_thread_cls.return_value = mock_thread
-        eng._maybe_startup_consolidate()
-
-    mock_thread_cls.assert_called_once()
-    assert mock_thread_cls.call_args.kwargs["name"] == "startup-consolidate"
-    mock_thread.start.assert_called_once()
-    eng.close()
-
-
-def test_startup_consolidation_skips_when_few_messages():
-    """Don't consolidate on startup if fewer messages than threshold."""
-    eng, td = _make_engine_with_messages(n=5)
-
-    consolidate_called = False
-    orig = eng._bg_consolidate
-    def spy():
-        nonlocal consolidate_called
-        consolidate_called = True
-        return orig()
-
-    eng._bg_consolidate = spy
-    eng._has_consolidation = True
-    eng._maybe_startup_consolidate()
-    assert not consolidate_called
-    eng.close()
+"""Behavioral coverage for all maintenance layers and startup provenance."""
+
+import runpy
+from pathlib import Path
+from unittest.mock import patch
+
+
+_ROUTING = runpy.run_path(str(Path(__file__).with_name("test-maintenance-engine-routing-753.py")))
+
+
+class TestConsolidationGaps(_ROUTING["RoutingFixture"]):
+    def test_clustering_count_matches_real_published_centroids(self):
+        self.add(5)
+        self.native.fit = lambda: setattr(self.native, "labels", [0, 0, 1, 1, 2])
+        result = self.engine.consolidate()
+        self.assertIn("3 clusters", result["cluster_messages"])
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM cluster_centroids").fetchone()[0], 3)
+
+    def test_preferences_runs_on_the_worker_connection(self):
+        function = self.modules["truememory.personality"].extract_preferences
+        with patch.object(self.modules["truememory.personality"], "extract_preferences", wraps=function) as preference:
+            result = self.engine.consolidate()
+        self.assertIn("extract_preferences", result)
+        self.assertNotIn("ERROR", result["extract_preferences"])
+        preference.assert_called_once()
+        self.assertIsNot(preference.call_args.args[0], self.conn)
+
+    def test_missing_vector_index_reports_unavailable_without_skipping_siblings(self):
+        self.conn.execute("DROP TABLE vec_messages_edge")
+        self.conn.commit()
+        result = self.engine.consolidate()
+        self.assertIn("UNAVAILABLE", result["cluster_messages"])
+        self.assertIn("build_summaries", result)
+        self.assertNotIn("ERROR", result["build_summaries"])
+
+    def test_cluster_error_is_categorical_and_siblings_still_run(self):
+        with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic private sentinel")):
+            result = self.engine.consolidate()
+        self.assertEqual(result["cluster_messages"], "ERROR (RuntimeError)")
+        self.assertNotIn("synthetic private sentinel", str(result))
+        self.assertNotIn("ERROR", result["detect_episodes"])
+
+    def test_startup_attempts_initial_small_corpus_once(self):
+        self.add(5)
+        self.engine._has_consolidation = True
+        self.engine._maybe_startup_consolidate()
+        self.completed()
+        self.assertEqual(self.state().outcome, "success_empty")
+        with patch.object(self.coordinator, "request_layers") as request:
+            self.engine._maybe_startup_consolidate()
+            request.assert_not_called()
+
+    def test_successful_empty_clusters_do_not_retrigger_all_layers(self):
+        self.add(30)
+        self.engine.consolidate()
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM cluster_centroids").fetchone()[0], 0)
+        self.engine._has_consolidation = True
+        with patch.object(self.coordinator, "request_layers") as request:
+            self.engine._maybe_startup_consolidate()
+            request.assert_not_called()
diff --git a/tests/test_hardening.py b/tests/test_hardening.py
index 654845ed..a8950826 100644
--- a/tests/test_hardening.py
+++ b/tests/test_hardening.py
@@ -188,15 +188,19 @@ class TestAutoConsolidation:
 
     def test_auto_consolidation_doesnt_fire_below_threshold(self):
         eng, td = _make_engine(n_messages=0)
+        # Establish the initial successful/empty or unavailable attempt. The
+        # threshold applies after that durable baseline, including on reopen.
+        eng.consolidate()
         eng._has_consolidation = True
         eng._auto_consolidate_threshold = 25
+        eng.conn.executemany("INSERT INTO messages(content) VALUES (?)", [("synthetic append",)] * 24)
+        eng.conn.commit()
 
-        import truememory.engine as _eng_mod
-        with patch.object(_eng_mod.threading, "Thread") as mock_thread_cls:
-            for i in range(24):
+        with patch.object(eng._maintenance_coordinator, "request_layers") as request:
+            for _ in range(3):
                 eng._maybe_auto_consolidate()
 
-        mock_thread_cls.assert_not_called()
+        request.assert_not_called()
         eng.close()
 
     def test_startup_consolidation_no_crash_on_closed_conn(self):
diff --git a/tests/test_issue_498_auto_consolidate.py b/tests/test_issue_498_auto_consolidate.py
index 901d7f73..4295082f 100644
--- a/tests/test_issue_498_auto_consolidate.py
+++ b/tests/test_issue_498_auto_consolidate.py
@@ -1,37 +1,49 @@
-"""Regression test for M1 #498: auto-consolidation trigger.
-
-Consolidation should run automatically after a configurable number of
-add() calls, not require manual truememory_consolidate invocation.
-"""
-
-import inspect
-import unittest
-
-
-class TestAutoConsolidation(unittest.TestCase):
-
-    def test_auto_consolidate_threshold_exists(self):
-        from truememory.engine import TrueMemoryEngine
-        source = inspect.getsource(TrueMemoryEngine.__init__)
-        self.assertIn("_auto_consolidate_threshold", source)
-        self.assertIn("TRUEMEMORY_AUTO_CONSOLIDATE_EVERY", source)
-
-    def test_add_calls_maybe_auto_consolidate(self):
-        from truememory.engine import TrueMemoryEngine
-        source = inspect.getsource(TrueMemoryEngine.add)
-        self.assertIn("_maybe_auto_consolidate", source,
-                       "add() should call _maybe_auto_consolidate()")
-
-    def test_maybe_auto_consolidate_checks_threshold(self):
-        from truememory.engine import TrueMemoryEngine
-        source = inspect.getsource(TrueMemoryEngine._maybe_auto_consolidate)
-        self.assertIn("_auto_consolidate_threshold", source)
-        self.assertIn("_has_consolidation", source)
-
-    def test_bg_consolidate_exists(self):
-        from truememory.engine import TrueMemoryEngine
-        self.assertTrue(hasattr(TrueMemoryEngine, "_bg_consolidate"))
-
-
-if __name__ == "__main__":
-    unittest.main()
+"""Durable automatic maintenance thresholds across engine instances."""
+
+import runpy
+from pathlib import Path
+from unittest.mock import patch
+
+
+_ROUTING = runpy.run_path(str(Path(__file__).with_name("test-maintenance-engine-routing-753.py")))
+
+
+class TestAutoConsolidation(_ROUTING["RoutingFixture"]):
+    def assert_empty_bootstrap(self) -> None:
+        maintenance = _ROUTING["MAINTENANCE"]
+        states = maintenance.read_layer_states(
+            self.conn, maintenance.engine_layer_specs(self.conn, self.coordinator))
+        self.assertEqual(len(states), 8)
+        self.assertTrue(all(state.outcome == "success_empty" for state in states.values()), states)
+        self.assertTrue(all(state.attempted_insert_count == 0 for state in states.values()), states)
+
+    def test_default_threshold_is_twenty_five(self):
+        self.assertEqual(self.engine._auto_consolidate_threshold, 25)
+
+    def test_configured_threshold_uses_committed_source_inserts(self):
+        self.engine.consolidate()
+        self.assert_empty_bootstrap()
+        self.engine._auto_consolidate_threshold = 3
+        self.engine._has_consolidation = True
+        with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request:
+            for _ in range(2):
+                self.engine.add("synthetic committed append")
+            request.assert_not_called()
+            self.engine.add("synthetic threshold append")
+            self.completed()
+            self.assertTrue(request.called)
+            self.assertTrue(all(call.kwargs["threshold"] == 3 for call in request.call_args_list))
+
+    def test_close_and_reopen_does_not_reset_threshold_progress(self):
+        self.engine.consolidate()
+        self.assert_empty_bootstrap()
+        self.add(24)
+        self.engine.close()
+        reopened = self.new_engine()
+        reopened._has_consolidation = True
+        with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request:
+            reopened._maybe_auto_consolidate()
+            request.assert_not_called()
+            reopened.add("synthetic twenty fifth")
+            self.completed()
+            self.assertTrue(request.called)
diff --git a/tests/test_issue_511_write_path_opt.py b/tests/test_issue_511_write_path_opt.py
index c2f1cbe4..0d1858a3 100644
--- a/tests/test_issue_511_write_path_opt.py
+++ b/tests/test_issue_511_write_path_opt.py
@@ -5,7 +5,9 @@
 M16: Fallback embed_single inside write lock — removed in favor of skip-and-log
 """
 
+import ast
 import inspect
+import textwrap
 import unittest
 
 
@@ -30,13 +32,17 @@ class TestM15StyleVecOutsideLock(unittest.TestCase):
 
     def test_add_precomputes_style_vec(self):
         from truememory.engine import TrueMemoryEngine
-        source = inspect.getsource(TrueMemoryEngine.add)
-        lock_pos = source.find("self._write_lock")
-        precompute_pos = source.find("compute_style_vector")
-        self.assertGreater(precompute_pos, -1,
-                           "add() should reference compute_style_vector for pre-computation")
-        self.assertLess(precompute_pos, lock_pos,
-                        "compute_style_vector call should appear before _write_lock")
+        tree = ast.parse(textwrap.dedent(inspect.getsource(TrueMemoryEngine.add)))
+        calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call)
+                 and isinstance(node.func, ast.Name) and node.func.id == "compute_style_vector"]
+        self.assertEqual(len(calls), 1)
+        for block in (node for node in ast.walk(tree) if isinstance(node, ast.With)):
+            self.assertNotIn(calls[0], list(ast.walk(block)),
+                             "Style computation must run outside lock and transaction contexts")
+        publications = [node for node in ast.walk(tree) if isinstance(node, ast.Call)
+                        and isinstance(node.func, ast.Name) and node.func.id == "_update_style_vec"]
+        self.assertEqual(len(publications), 1)
+        self.assertLess(calls[0].lineno, publications[0].lineno)
 
     def test_style_vec_accepts_precomputed(self):
         from truememory.personality_style_vec import update_entity_style_vector_incremental
diff --git a/tests/test_issue_590_build_vectors_batch.py b/tests/test_issue_590_build_vectors_batch.py
index a678b1c8..1f11dc2a 100644
--- a/tests/test_issue_590_build_vectors_batch.py
+++ b/tests/test_issue_590_build_vectors_batch.py
@@ -73,6 +73,8 @@ def _make_server(monkeypatch, tmp_path):
 def _seed_messages(ms, n: int) -> None:
     """Add *n* messages to the database via the server module."""
     m = ms._get_memory()
+    # These focused tests own explicit rebuild maintenance.
+    m._engine._has_consolidation = 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 3a168421..a85b8530 100644
--- a/tests/test_issue_647_vector_rebuild.py
+++ b/tests/test_issue_647_vector_rebuild.py
@@ -81,6 +81,8 @@ def _make_server(monkeypatch, tmp_path):
 
 def _seed_messages(ms, n: int) -> None:
     m = ms._get_memory()
+    # These focused tests own explicit rebuild maintenance.
+    m._engine._has_consolidation = 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 0285b066..37e7b4d8 100644
--- a/truememory/engine.py
+++ b/truememory/engine.py
@@ -31,6 +31,10 @@
 import time
 import sqlite3
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from truememory.maintenance import LayerResult, MaintenanceCoordinator
 
 # ───────────────────────────────────────────────────────────────────────────
 # Core modules (always available)
@@ -358,18 +362,18 @@ def __init__(self, db_path: str | Path = None,
         # Coordination flag for background NaN re-embed thread (#485).
         self._nan_migration_in_progress = False
 
-        # Auto-consolidation: run L5 consolidation every N adds (#498)
-        self._adds_since_consolidation = 0
         self._auto_consolidate_threshold = _env_int(
             "TRUEMEMORY_AUTO_CONSOLIDATE_EVERY", 25, lo=1
         )
-        self._consolidation_thread: threading.Thread | None = None
+        self._maintenance_coordinator = None
+        self._maintenance_handle = None
+        self._maintenance_pending_reason = "not_connected"
 
     # ──────────────────────────────────────────────────────────────────────
     # Auto-connect (production API)
     # ──────────────────────────────────────────────────────────────────────
 
-    def _ensure_connection(self) -> None:
+    def _ensure_connection(self, *, _suppress_maintenance: bool = False) -> None:
         """Open (or create) the database and load extensions if needed.
 
         Called automatically by :meth:`add`, :meth:`search`, and other
@@ -385,6 +389,8 @@ def _ensure_connection(self) -> None:
             # through to reconnect so the server self-heals once the FS recovers.
             try:
                 self.conn.execute("PRAGMA schema_version")
+                if not _suppress_maintenance:
+                    self._maybe_auto_consolidate()
                 return
             except sqlite3.Error:
                 logger.warning(
@@ -400,6 +406,8 @@ def _ensure_connection(self) -> None:
 
         with self._init_lock:
             if self.conn is not None:
+                if not _suppress_maintenance:
+                    self._maybe_auto_consolidate()
                 return
 
             # Create parent directory if using a real path
@@ -581,7 +589,8 @@ def _bg_wrapper(db_path):
             self._purge_legacy_entity_profile_summaries()
 
             self.ready = True
-            self._maybe_startup_consolidate()
+            if not _suppress_maintenance:
+                self._maybe_startup_consolidate()
 
     def _purge_legacy_entity_profile_summaries(self) -> None:
         """Delete legacy ``period='entity_profile'`` summary rows once.
@@ -656,25 +665,8 @@ def _purge_legacy_entity_profile_summaries(self) -> None:
             )
 
     def _maybe_startup_consolidate(self) -> None:
-        """Trigger background consolidation on startup if data looks stale."""
-        if not self._has_consolidation or self.conn is None:
-            return
-        try:
-            msg_count = self.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
-            if msg_count < self._auto_consolidate_threshold:
-                return
-            cluster_count = self.conn.execute(
-                "SELECT COUNT(*) FROM message_clusters"
-            ).fetchone()[0]
-            if cluster_count == 0:
-                self._consolidation_thread = threading.Thread(
-                    target=self._bg_consolidate,
-                    daemon=True,
-                    name="startup-consolidate",
-                )
-                self._consolidation_thread.start()
-        except Exception:
-            pass
+        """Use durable per-layer provenance, including successful empty work."""
+        self._maybe_auto_consolidate()
 
     # ──────────────────────────────────────────────────────────────────────
     # Production CRUD API
@@ -724,14 +716,18 @@ def add(
         pre_embedding = None
         pre_sep_embedding = None
         if self._has_vectors:
+            from truememory.vector_search import VectorPublicationChanged
             try:
                 from truememory.vector_search import (
                     _capture_rebuild_model, _encode_with_mps_fallback, _build_sep_text,
                 )
-                model, pre_identity = _capture_rebuild_model()
+                with self._write_lock:
+                    model, pre_identity = _capture_rebuild_model(self.conn)
                 pre_embedding = _encode_with_mps_fallback(model, [content])[0]
                 sep_text = _build_sep_text(sender, recipient, timestamp, content)
                 pre_sep_embedding = _encode_with_mps_fallback(model, [sep_text])[0]
+            except VectorPublicationChanged:
+                raise
             except Exception:
                 logger.debug("Failed to pre-compute embedding during add()", exc_info=True)
 
@@ -743,13 +739,15 @@ def add(
             except Exception:
                 logger.debug("Failed to pre-compute style vector during add()", exc_info=True)
 
+        from contextlib import nullcontext
         from truememory.rebuild_source import rebuild_transaction
-        publication_fence = None
         if pre_embedding is not None:
             from truememory.vector_search import _foreground_model_fence
-            publication_fence = _foreground_model_fence(pre_identity)
 
-        with self._write_lock, rebuild_transaction(self.conn, write=True, publication_fence=publication_fence):
+        with self._write_lock, (
+            _foreground_model_fence(pre_identity, blocking=not self.conn.in_transaction)
+            if pre_embedding is not None else nullcontext()
+        ), rebuild_transaction(self.conn, write=True):
             msg = {
                 "content": content,
                 "sender": sender,
@@ -816,30 +814,54 @@ def add(
             "metadata": metadata or {},
         }
 
+    def _get_maintenance_coordinator(self) -> MaintenanceCoordinator:
+        from truememory.maintenance import get_coordinator
+        coordinator = self._maintenance_coordinator
+        if coordinator is not None:
+            try:
+                coordinator.status
+            except RuntimeError:
+                coordinator = None
+        if coordinator is None or self._maintenance_handle is not self.conn:
+            coordinator = get_coordinator(self.db_path)
+            coordinator.refresh_capabilities()
+            self._maintenance_coordinator = coordinator
+            self._maintenance_handle = self.conn
+        return coordinator
+
     def _maybe_auto_consolidate(self) -> None:
-        """Trigger background consolidation after threshold adds."""
-        self._adds_since_consolidation += 1
-        if self._adds_since_consolidation < self._auto_consolidate_threshold:
-            return
-        if not self._has_consolidation:
+        """Probe fixed metadata without waiting on foreground or model work."""
+        if not self._has_consolidation or self.conn is None:
             return
-        if (self._consolidation_thread is not None
-                and self._consolidation_thread.is_alive()):
+        if not self._write_lock.acquire(blocking=False):
+            self._maintenance_pending_reason = "foreground_busy"
             return
-        self._adds_since_consolidation = 0
-        self._consolidation_thread = threading.Thread(
-            target=self._bg_consolidate,
-            daemon=True,
-            name="auto-consolidate",
-        )
-        self._consolidation_thread.start()
+        coordinator = None
+        eligible = ()
+        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
+            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"
+        except Exception as error:
+            self._maintenance_pending_reason = type(error).__name__[:64]
+        finally:
+            self._write_lock.release()
+        if eligible and coordinator is not None:
+            try:
+                coordinator.request_layers(threshold=self._auto_consolidate_threshold)
+            except Exception as error:
+                self._maintenance_pending_reason = type(error).__name__[:64]
 
     def _bg_consolidate(self) -> None:
-        """Run consolidation in a background thread with its own connection."""
-        try:
-            self.consolidate()
-        except Exception:
-            logger.debug("Auto-consolidation failed", exc_info=True)
+        """Compatibility entry point; the coordinator owns the worker handle."""
+        self._maybe_auto_consolidate()
 
     def delete(self, memory_id: int) -> bool:
         """Delete a memory by ID.
@@ -848,7 +870,10 @@ def delete(self, memory_id: int) -> bool:
         """
         self._ensure_connection()
         with self._write_lock:
-            return delete_message(self.conn, memory_id)
+            deleted = delete_message(self.conn, memory_id)
+        if deleted:
+            self._maybe_auto_consolidate()
+        return deleted
 
     def delete_all(self, user_id: str | None = None) -> bool:
         """Delete all memories, optionally filtered by user.
@@ -1015,147 +1040,51 @@ def delete_all(self, user_id: str | None = None) -> bool:
                 logger.warning("Failed to rebuild FTS index during delete_all", exc_info=True)
 
             self.conn.commit()
-            return deleted
+        if deleted:
+            self._maybe_auto_consolidate()
+        return deleted
+
+    def _apply_manual_maintenance_capabilities(self, results: tuple[LayerResult, ...]) -> None:
+        completed = {result.layer for result in results
+                     if result.outcome in {"success", "success_empty"}}
+        if "summaries" in completed:
+            self._has_consolidation = True
+        if "clusters" in completed:
+            self._has_clustering = True
 
     def consolidate(self) -> dict[str, str]:
-        """Run all consolidation layers (L0-L5) under the write lock.
-
-        Returns timing stats for each step.
-        """
-        import time as _time
-
-        self._ensure_connection()
-        stats: dict[str, str] = {}
-
-        try:
-            from truememory.consolidation import (
-                build_summaries,
-                detect_contradictions,
-                build_structured_facts,
-            )
-        except (ImportError, ModuleNotFoundError):
-            stats["consolidation"] = "SKIPPED (module not available)"
-            return stats
-
-        try:
-            from truememory.predictive import build_surprise_index
-        except (ImportError, ModuleNotFoundError):
-            build_surprise_index = None
-
-        try:
-            from truememory.temporal import detect_episodes, detect_landmark_events
-        except (ImportError, ModuleNotFoundError):
-            detect_episodes = None
-            detect_landmark_events = None
-
-        try:
-            from truememory.personality import build_dunbar_hierarchy
-        except (ImportError, ModuleNotFoundError):
-            build_dunbar_hierarchy = None
-
-        _cluster_messages = None
-        if _HAS_CLUSTERING and self._has_vectors:
-            try:
-                from truememory.clustering import cluster_messages as _cm
-                _cluster_messages = _cm
-            except (ImportError, ModuleNotFoundError):
-                pass
-
-        _extract_preferences = None
-        if _HAS_PERSONALITY:
-            try:
-                from truememory.personality import extract_preferences as _ep
-                _extract_preferences = _ep
-            except (ImportError, ModuleNotFoundError):
-                pass
-
+        """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,
+        )
+        self._ensure_connection(_suppress_maintenance=True)
         with self._write_lock:
-            if _cluster_messages:
-                try:
-                    t0 = _time.time()
-                    n = _cluster_messages(self.conn)
-                    stats["cluster_messages"] = f"{n} clusters in {_time.time() - t0:.3f}s"
-                    self._has_clustering = True
-                except Exception as exc:
-                    stats["cluster_messages"] = f"ERROR: {exc}"
-
-            if _extract_preferences:
-                try:
-                    t0 = _time.time()
-                    _extract_preferences(self.conn)
-                    stats["extract_preferences"] = f"{_time.time() - t0:.3f}s"
-                except Exception as exc:
-                    stats["extract_preferences"] = f"ERROR: {exc}"
-
-            try:
-                t0 = _time.time()
-                build_summaries(self.conn)
-                stats["build_summaries"] = f"{_time.time() - t0:.3f}s"
-            except Exception as exc:
-                stats["build_summaries"] = f"ERROR: {exc}"
-
-            try:
-                t0 = _time.time()
-                detect_contradictions(self.conn)
-                stats["detect_contradictions"] = f"{_time.time() - t0:.3f}s"
-            except Exception as exc:
-                stats["detect_contradictions"] = f"ERROR: {exc}"
-
-            try:
-                t0 = _time.time()
-                n = build_structured_facts(self.conn)
-                stats["structured_facts"] = f"{n} facts in {_time.time() - t0:.3f}s"
-            except Exception as exc:
-                stats["structured_facts"] = f"ERROR: {exc}"
-
-            if build_surprise_index:
-                try:
-                    t0 = _time.time()
-                    build_surprise_index(self.conn)
-                    stats["build_surprise_index"] = f"{_time.time() - t0:.3f}s"
-                except Exception as exc:
-                    stats["build_surprise_index"] = f"ERROR: {exc}"
-
-            if detect_episodes:
-                try:
-                    t0 = _time.time()
-                    ep = detect_episodes(self.conn)
-                    stats["detect_episodes"] = f"{ep} episodes in {_time.time() - t0:.3f}s"
-                except Exception as exc:
-                    stats["detect_episodes"] = f"ERROR: {exc}"
-
-            if detect_landmark_events:
+            coordinator = self._get_maintenance_coordinator()
+            coordinator.refresh_capabilities()
+            borrowed = self.conn.in_transaction
+            if borrowed or coordinator.path is None:
                 try:
-                    t0 = _time.time()
-                    lm = detect_landmark_events(self.conn)
-                    stats["detect_landmarks"] = f"{lm} events in {_time.time() - t0:.3f}s"
-                except Exception as exc:
-                    stats["detect_landmarks"] = f"ERROR: {exc}"
-
-            if build_dunbar_hierarchy:
+                    report = run_engine_maintenance(self.conn, coordinator, force=True,
+                        threshold=self._auto_consolidate_threshold, prepare_extensions=False,
+                        allow_caller_transaction=borrowed)
+                except MaintenanceBusyError:
+                    return maintenance_busy_result()
+                self._apply_manual_maintenance_capabilities(report.results)
+                return format_maintenance_report(report)
+        try:
+            with maintenance_owner(coordinator.path):
+                connection = create_db(coordinator.path)
                 try:
-                    t0 = _time.time()
-                    primary = None
-                    try:
-                        row = self.conn.execute(
-                            "SELECT sender, COUNT(*) as cnt FROM messages "
-                            "WHERE sender != '' AND sender IS NOT NULL "
-                            "GROUP BY sender ORDER BY cnt DESC LIMIT 1"
-                        ).fetchone()
-                        if row and row[0] and row[0].strip():
-                            primary = row[0]
-                    except Exception:
-                        pass
-                    result = build_dunbar_hierarchy(self.conn, primary_entity=primary)
-                    n_rel = len(result) if isinstance(result, dict) else result
-                    stats["dunbar_hierarchy"] = f"{n_rel} relationships in {_time.time() - t0:.3f}s"
-                except Exception as exc:
-                    stats["dunbar_hierarchy"] = f"ERROR: {exc}"
-
-            self.conn.commit()
-
-        self._has_consolidation = True
-        return stats
+                    report = run_engine_maintenance(connection, coordinator, force=True,
+                                                   threshold=self._auto_consolidate_threshold)
+                finally:
+                    connection.close()
+        except MaintenanceBusyError:
+            return maintenance_busy_result()
+        with self._write_lock:
+            self._apply_manual_maintenance_capabilities(report.results)
+        return format_maintenance_report(report)
 
     def update(self, memory_id: int, content: str | None = None, **fields) -> dict | None:
         """Update a memory.
@@ -1176,11 +1105,13 @@ def update(self, memory_id: int, content: str | None = None, **fields) -> dict |
         pre_embedding = None
         pre_sep_embedding = None
         if content is not None and self._has_vectors:
+            from truememory.vector_search import VectorPublicationChanged
             try:
                 from truememory.vector_search import (
                     _capture_rebuild_model, _encode_with_mps_fallback, _build_sep_text,
                 )
-                model, pre_identity = _capture_rebuild_model()
+                with self._write_lock:
+                    model, pre_identity = _capture_rebuild_model(self.conn)
                 pre_embedding = _encode_with_mps_fallback(model, [content])[0]
                 row = self.conn.execute(
                     "SELECT sender, recipient, timestamp FROM messages WHERE id = ?",
@@ -1192,44 +1123,52 @@ def update(self, memory_id: int, content: str | None = None, **fields) -> dict |
                     timestamp_val = fields.get("timestamp", row[2])
                     sep_text = _build_sep_text(sender_val, recipient_val, timestamp_val, content)
                     pre_sep_embedding = _encode_with_mps_fallback(model, [sep_text])[0]
+            except VectorPublicationChanged:
+                raise
             except Exception:
                 logger.debug("Failed to pre-compute embedding during update()", exc_info=True)
 
-        with self._write_lock:
-            if content is not None:
-                fields["content"] = content
+        source_committed = False
+        try:
+            with self._write_lock:
+                if content is not None:
+                    fields["content"] = content
 
-            ok = update_message(self.conn, memory_id, **fields)
-            if not ok:
-                return None
+                ok = update_message(self.conn, memory_id, **fields)
+                source_committed = ok and not self.conn.in_transaction
+                if not ok:
+                    return None
 
-            if pre_embedding is not None:
-                from truememory.vector_search import VectorPublicationChanged, _foreground_vector_publication
-                try:
-                    from truememory.vector_search import serialize_f32, _write_embedder_metadata_no_commit
-                    with _foreground_vector_publication(self.conn, pre_identity, memory_id) as (vec_tbl, sep_tbl):
-                        try:
-                            self.conn.execute(f"DELETE FROM {vec_tbl} WHERE rowid = ?", (memory_id,))
-                        except Exception:
-                            logger.debug("Failed to delete old vector embedding for message %d", memory_id, exc_info=True)
-                        self.conn.execute(
-                            f"INSERT INTO {vec_tbl}(rowid, embedding) VALUES (?, ?)",
-                            (memory_id, serialize_f32(pre_embedding)),
-                        )
-                        if pre_sep_embedding is not None:
+                if pre_embedding is not None:
+                    from truememory.vector_search import VectorPublicationChanged, _foreground_vector_publication
+                    try:
+                        from truememory.vector_search import serialize_f32, _write_embedder_metadata_no_commit
+                        with _foreground_vector_publication(self.conn, pre_identity, memory_id) as (vec_tbl, sep_tbl):
                             try:
-                                self.conn.execute(f"DELETE FROM {sep_tbl} WHERE rowid = ?", (memory_id,))
+                                self.conn.execute(f"DELETE FROM {vec_tbl} WHERE rowid = ?", (memory_id,))
                             except Exception:
-                                logger.debug("Failed to delete old sep vector for message %d", memory_id, exc_info=True)
+                                logger.debug("Failed to delete old vector embedding for message %d", memory_id, exc_info=True)
                             self.conn.execute(
-                                f"INSERT INTO {sep_tbl}(rowid, embedding) VALUES (?, ?)",
-                                (memory_id, serialize_f32(pre_sep_embedding)),
+                                f"INSERT INTO {vec_tbl}(rowid, embedding) VALUES (?, ?)",
+                                (memory_id, serialize_f32(pre_embedding)),
                             )
-                        _write_embedder_metadata_no_commit(self.conn)
-                except VectorPublicationChanged as exc:
-                    raise VectorPublicationChanged("Source update committed; retry update to publish vectors") from exc
-                except Exception:
-                    logger.warning("Vector embedding failed for message %d", memory_id, exc_info=True)
+                            if pre_sep_embedding is not None:
+                                try:
+                                    self.conn.execute(f"DELETE FROM {sep_tbl} WHERE rowid = ?", (memory_id,))
+                                except Exception:
+                                    logger.debug("Failed to delete old sep vector for message %d", memory_id, exc_info=True)
+                                self.conn.execute(
+                                    f"INSERT INTO {sep_tbl}(rowid, embedding) VALUES (?, ?)",
+                                    (memory_id, serialize_f32(pre_sep_embedding)),
+                                )
+                            _write_embedder_metadata_no_commit(self.conn)
+                    except VectorPublicationChanged as exc:
+                        raise VectorPublicationChanged("Source update committed; retry update to publish vectors") from exc
+                    except Exception:
+                        logger.warning("Vector embedding failed for message %d", memory_id, exc_info=True)
+        finally:
+            if source_committed:
+                self._maybe_auto_consolidate()
 
         return self.get(memory_id)
 
@@ -2557,6 +2496,28 @@ def get_stats(self) -> dict:
         """Return ingestion and search statistics."""
         self._ensure_connection()
         stats = dict(self.stats)
+        if self._write_lock.acquire(blocking=False):
+            try:
+                from truememory.maintenance import engine_layer_specs, layer_freshness, read_layer_states
+                coordinator = self._get_maintenance_coordinator()
+                maintenance = coordinator.snapshot()
+                maintenance["last_results"] = [result._asdict() for result in maintenance["last_results"]]
+                maintenance["pending_reason"] = self._maintenance_pending_reason
+                maintenance["pending_caller_commit"] = self.conn.in_transaction
+                states = read_layer_states(self.conn, engine_layer_specs(self.conn, coordinator))
+                maintenance["layers"] = {name: {
+                    "outcome": state.outcome, "freshness": layer_freshness(state.source, state, state.dependency),
+                    "coverage": state.successful_coverage, "error_category": state.error_category,
+                    "dependency_error": state.dependency.error_category, "output_count": state.output_count,
+                    "pending_inserts": state.source.insert_count - (state.attempted_insert_count or 0),
+                } for name, state in states.items()}
+                stats["maintenance"] = maintenance
+            except Exception as error:
+                stats["maintenance"] = {"status": "unavailable", "error_category": type(error).__name__[:64]}
+            finally:
+                self._write_lock.release()
+        else:
+            stats["maintenance"] = {"status": "pending", "pending_reason": "foreground_busy"}
 
         # Add live DB stats if connected.
         if self.conn:
@@ -2579,19 +2540,18 @@ def get_stats(self) -> dict:
         return stats
 
     def close(self):
-        """Close database connection and join background threads."""
-        if self._consolidation_thread is not None and self._consolidation_thread.is_alive():
-            try:
-                self._consolidation_thread.join(timeout=5)
-            except Exception:
-                pass
-        if self.conn:
-            try:
-                self.conn.close()
-            except Exception:
-                logger.debug("Failed to close database connection", exc_info=True)
-            self.conn = None
-            self.ready = False
+        """Detach this engine; a shared worker retains its own connection."""
+        with self._init_lock, self._write_lock:
+            self._maintenance_coordinator = None
+            self._maintenance_handle = None
+            self._maintenance_pending_reason = "not_connected"
+            if self.conn:
+                try:
+                    self.conn.close()
+                except sqlite3.Error:
+                    logger.debug("Failed to close database connection", exc_info=True)
+                self.conn = None
+                self.ready = False
 
     def __enter__(self):
         return self
diff --git a/truememory/maintenance.py b/truememory/maintenance.py
index 325ac2c8..670f0202 100644
--- a/truememory/maintenance.py
+++ b/truememory/maintenance.py
@@ -1,7 +1,6 @@
-"""Database maintenance ownership and committed source revision primitives.
+"""Database maintenance ownership, source revisions and durable layer scheduling.
 
-The engine scheduler is not routed here yet. No models, polling loop or automatic
-builder invocation is introduced by importing this module.
+Importing this module loads no models and starts no worker or polling loop.
 """
 
 import hashlib
@@ -104,27 +103,41 @@ def connection_database_path(conn: sqlite3.Connection) -> Path | None:
 _coordinators: weakref.WeakValueDictionary = weakref.WeakValueDictionary()
 _held_fds: set[int] = set()
 _held_paths: dict[Path, int] = {}
+_owner_waiters: dict[Path, tuple[int, int, "MaintenanceCoordinator"]] = {}
 _thread_owners = threading.local()
 
 
-def _acquire_owner(path: Path) -> int | None:
+def _acquire_owner(path: Path, on_busy: Callable[[], None] | None = None) -> int | None:
     with _registry_lock:
         if path in _held_paths:
+            if on_busy is not None:
+                on_busy()
             return None
         fd = try_file_lock(Path(str(path) + ".maintenance.lock"))
         if fd is not None:
             _held_fds.add(fd)
             _held_paths[path] = fd
+        elif on_busy is not None:
+            on_busy()
         return fd
 
 
 def _release_owner(fd: int) -> None:
+    notify = None
     with _registry_lock:
         os.close(fd)
         _held_fds.discard(fd)
         for path, claimed_fd in tuple(_held_paths.items()):
             if claimed_fd == fd:
                 del _held_paths[path]
+                waiter = _owner_waiters.get(path)
+                if waiter is not None and waiter[0] == fd:
+                    del _owner_waiters[path]
+                    notify = waiter[2]
+                else:
+                    notify = _coordinators.get(path)
+    if notify is not None:
+        notify._owner_released()
 
 
 @contextmanager
@@ -169,13 +182,14 @@ def maintenance_owner(db_path: str | os.PathLike[str] | None) -> Iterator[Mainte
 
 
 def _after_fork() -> None:
-    global _registry_lock, _coordinators, _held_fds, _held_paths, _thread_owners
+    global _registry_lock, _coordinators, _held_fds, _held_paths, _owner_waiters, _thread_owners
     # The child must not prolong its parent's ownership by retaining an
     # inherited descriptor. Inherited coordinator objects reject child use.
     for fd in _held_fds:
         os.close(fd)
     _held_fds = set()
     _held_paths = {}
+    _owner_waiters = {}
     _thread_owners = threading.local()
     _coordinators = weakref.WeakValueDictionary()
     _registry_lock = threading.Lock()
@@ -203,6 +217,12 @@ def __init__(self, path: Path | None) -> None:
         self._done.set()
         self._status = "pending"
         self._error_category: str | None = None
+        self._notification_generation = 0
+        self._pending: tuple[int, int] | None = None
+        self._last_report = None
+        self._capability_epoch = 0
+        self._capability_versions: tuple[tuple[str, str | None], ...] | None = None
+        self._extension_failure: LayerDependency | None = None
 
     def _check_process(self) -> None:
         if self._pid != os.getpid():
@@ -214,6 +234,83 @@ def status(self) -> tuple[str, str | None]:
         with self._mutex:
             return self._status, self._error_category
 
+    def snapshot(self) -> dict:
+        self._check_process()
+        with self._mutex:
+            return {"status": self._status, "error_category": self._error_category,
+                    "active": self._active, "pending_generations": int(self._pending is not None),
+                    "capability_epoch": self._capability_epoch,
+                    "last_results": self._last_report.results if self._last_report is not None else ()}
+
+    def refresh_capabilities(self, *, reset_extension: bool = True) -> None:
+        """Refresh three package records at open/manual setup, never per add."""
+        self._check_process()
+        versions = tuple((module, _installed_dependency(module, distribution)) for module, distribution in (
+            ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"),
+        ))
+        with self._mutex:
+            self._capability_epoch += 1
+            if reset_extension or versions != self._capability_versions:
+                self._extension_failure = None
+            self._capability_versions = versions
+
+    def capability_snapshot(self) -> tuple[int, tuple, "LayerDependency | None"]:
+        self._check_process()
+        with self._mutex:
+            return self._capability_epoch, self._capability_versions or (), self._extension_failure
+
+    def _publish_extension_evidence(self, epoch: int, dependency: "LayerDependency | None") -> None:
+        with self._mutex:
+            if epoch == self._capability_epoch:
+                self._extension_failure = dependency
+
+    def _reserve_locked(self) -> None:
+        self._active = True
+        self._done.clear()
+        self._cancel.clear()
+        self._status = "starting"
+        self._error_category = None
+
+    def request_layers(self, *, threshold: int = 25) -> 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")
+        with self._mutex:
+            if self.path is None:
+                self._status = "pending_in_memory"
+                return False
+            if self._active and self._cancel.is_set():
+                return False
+            self._notification_generation += 1
+            self._pending = (self._notification_generation, threshold)
+            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:
+        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 work(conn: sqlite3.Connection, cancel: threading.Event) -> "MaintenanceReport":
+            return run_engine_maintenance(conn, self, threshold=pending[1], cancel=cancel)
+        return self._launch(work, pending)
+
+    def _owner_released(self) -> None:
+        self._check_process()
+        with self._mutex:
+            pending = self._take_pending_locked()
+        if pending is not None:
+            try:
+                self._launch_layers(pending)
+            except Exception:
+                # The launch path records a bounded failure. Notification must
+                # not replace a synchronous owner's unrelated operation error.
+                pass
+
     def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) -> bool:
         """Start one owned worker, or report busy/in-memory pending without retry."""
         self._check_process()
@@ -223,11 +320,21 @@ def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) -
                 return False
             if self._active:
                 return False
-            self._active = True
-            self._done.clear()
-            self._cancel.clear()
-            self._status = "starting"
-            self._error_category = None
+            self._reserve_locked()
+        return self._launch(work)
+
+    def _launch(self, work: Callable, pending: tuple[int, int] | None = None) -> 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 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()
 
         fd = None
         handoff = None
@@ -235,12 +342,8 @@ def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) -
         try:
             # Serializing descriptor acquisition with fork registration makes
             # every inherited owner descriptor visible to the child cleanup.
-            fd = _acquire_owner(self.path)
+            fd = _acquire_owner(self.path, on_busy=busy)
             if fd is None:
-                with self._mutex:
-                    self._active = False
-                    self._status = "busy"
-                    self._done.set()
                 return False
             handoff = {"fd": fd, "started": False}
 
@@ -266,7 +369,10 @@ def run() -> None:
             if started:
                 # start() can be interrupted after launching the target. That
                 # worker owns cleanup until it really finishes, even on error.
-                self._cancel.set()
+                with self._mutex:
+                    if self._active and self._thread is worker:
+                        self._pending = None
+                        self._cancel.set()
                 raise
             if fd is not None:
                 _release_owner(fd)
@@ -275,7 +381,14 @@ def run() -> None:
                 self._thread = None
                 self._status = "failed"
                 self._error_category = "worker_start"
-                self._done.set()
+                successor = self._take_pending_locked()
+                if successor is None:
+                    self._done.set()
+            if successor is not None:
+                try:
+                    self._launch_layers(successor)
+                except Exception:
+                    pass
             raise
 
     def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], None]) -> None:
@@ -287,11 +400,15 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No
                 try:
                     if not self._cancel.is_set():
                         conn = create_db(self.path)
-                        work(conn, self._cancel)
+                        report = work(conn, self._cancel)
                         if conn.in_transaction:
                             conn.rollback()
                             raise RuntimeError("Maintenance work left an unfinished transaction")
                         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:
@@ -307,14 +424,29 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No
             _release_owner(fd)
             with self._mutex:
                 self._active = False
+                self._thread = None
                 self._status = outcome
                 self._error_category = error_category
-                self._done.set()
+                if self._cancel.is_set():
+                    self._pending = None
+                pending = self._take_pending_locked()
+                if pending is None:
+                    self._done.set()
+            if pending is not None:
+                try:
+                    self._launch_layers(pending)
+                except Exception:
+                    pass  # Launch already recorded the categorical failure.
 
     def cancel(self) -> None:
         """Signal a phase boundary; never release ownership of active work."""
         self._check_process()
-        self._cancel.set()
+        with _registry_lock, self._mutex:
+            self._pending = None
+            self._cancel.set()
+            waiter = _owner_waiters.get(self.path)
+            if waiter is not None and waiter[2] is self:
+                del _owner_waiters[self.path]
 
     def wait(self, timeout: float | None = None) -> bool:
         """Wait for this worker if requested; report whether teardown completed."""
@@ -860,12 +992,15 @@ def _installed_dependency(module: str, distribution: str) -> str | None:
         return None
 
 
-def _cluster_schedule_dependency(conn: sqlite3.Connection) -> LayerDependency:
+def _cluster_schedule_dependency(
+    conn: sqlite3.Connection, *, versions: dict[str, str | None] | None = None,
+) -> LayerDependency:
     parameters = {"min_cluster_size": 10, "min_samples": 5, "metric": "euclidean",
                   "cluster_selection_method": "eom"}
-    versions = {module: _installed_dependency(module, distribution) for module, distribution in (
-        ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"),
-    )}
+    if versions is None:
+        versions = {module: _installed_dependency(module, distribution) for module, distribution in (
+            ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"),
+        )}
     parameters["versions"] = versions
 
     def unavailable(category: str) -> LayerDependency:
@@ -1042,3 +1177,136 @@ def dunbar_dependency() -> LayerDependency:
     return tuple(specs[name] for name in (
         "clusters", "summaries", "contradictions", "structured_facts", "surprise", "episodes", "landmarks", "dunbar",
     ))
+
+
+class MaintenanceReport(NamedTuple):
+    results: tuple[LayerResult, ...]
+    preferences: str
+
+
+def _dependency_parameters(dependency: LayerDependency) -> dict | None:
+    if dependency.deferred:
+        return None
+    return json.loads(dependency.key)["parameters"]
+
+
+def engine_layer_specs(
+    conn: sqlite3.Connection, coordinator: MaintenanceCoordinator, *, evidence: tuple | None = None,
+) -> tuple[LayerSpec, ...]:
+    """Small live probes plus one bounded worker-capability evidence record."""
+    _, versions, failure = evidence if evidence is not None else coordinator.capability_snapshot()
+
+    def resolve() -> LayerDependency:
+        dependency = _cluster_schedule_dependency(conn, versions=dict(versions))
+        if (failure is not None and not dependency.deferred
+                and _dependency_parameters(dependency) == _dependency_parameters(failure)):
+            return failure
+        return dependency
+
+    return tuple(spec._replace(resolve_dependency=resolve) if spec.layer == "clusters" else spec
+                 for spec in all_layer_specs(conn))
+
+
+def _prepare_worker_extensions(conn: sqlite3.Connection, coordinator: MaintenanceCoordinator) -> tuple:
+    coordinator.refresh_capabilities(reset_extension=False)
+    epoch, versions, previous_failure = coordinator.capability_snapshot()
+    failed = False
+    enabled = False
+    try:
+        if dict(versions).get("sqlite_vec") is None:
+            failed = True
+        else:
+            module = importlib.import_module("sqlite_vec")
+            conn.enable_load_extension(True)
+            enabled = True
+            module.load(conn)
+    except Exception:
+        failed = True
+    finally:
+        if enabled:
+            conn.enable_load_extension(False)
+    dependency = _cluster_schedule_dependency(conn, versions=dict(versions))
+    failure = previous_failure if failed else None
+    if failed and not dependency.deferred:
+        parameters = _dependency_parameters(dependency)
+        if parameters is not None and "identity_sha256" in parameters:
+            failure = make_layer_dependency(dependency.builder_version, parameters,
+                                            available=False, error_category="VectorExtensionUnavailable")
+    elif dependency.error_category == "VectorExtensionUnavailable":
+        failure = dependency
+    coordinator._publish_extension_evidence(epoch, failure)
+    return epoch, versions, failure
+
+
+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,
+) -> MaintenanceReport:
+    """Use only the supplied owned worker/original explicitly borrowed handle."""
+    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,
+                         allow_caller_transaction=allow_caller_transaction)
+    preferences = "SKIPPED (no maintenance attempt)"
+    if cancel is not None and cancel.is_set():
+        preferences = "CANCELLED"
+    elif force or any(result.attempted for result in results):
+        started = time.monotonic()
+        try:
+            importlib.import_module("truememory.personality").extract_preferences(conn)
+        except (ImportError, AttributeError):
+            preferences = "UNAVAILABLE (DependencyMissing)"
+        except Exception as error:
+            preferences = "ERROR (" + type(error).__name__[:64] + ")"
+        else:
+            preferences = f"{time.monotonic() - started:.3f}s"
+        if conn.in_transaction and allow_caller_transaction:
+            preferences += " (pending caller commit)"
+    return MaintenanceReport(results, preferences)
+
+
+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"
+    for outcome in ("failed", "unavailable", "deferred", "abandoned"):
+        for result in report.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):
+        return "pending", None
+    if any(result.coverage != "complete" for result in report.results):
+        return "completed_with_limits", None
+    return "success", None
+
+
+_MAINTENANCE_RESULT_KEYS = (
+    "cluster_messages", "build_summaries", "detect_contradictions", "structured_facts",
+    "build_surprise_index", "detect_episodes", "detect_landmarks", "dunbar_hierarchy", "extract_preferences",
+)
+
+
+def maintenance_busy_result() -> dict[str, str]:
+    return dict.fromkeys(_MAINTENANCE_RESULT_KEYS, "BUSY (maintenance owner active)")
+
+
+def format_maintenance_report(report: MaintenanceReport) -> dict[str, str]:
+    counts = {"clusters": "clusters", "structured_facts": "facts", "episodes": "episodes",
+              "landmarks": "events", "dunbar": "relationships"}
+    output = {}
+    for result in report.results:
+        if result.outcome in {"success", "success_empty"}:
+            prefix = f"{result.output_count} {counts[result.layer]} in " if result.layer in counts else ""
+            value = prefix + f"{result.elapsed_seconds:.3f}s"
+            if result.coverage != "complete":
+                value += " (" + result.coverage + ")"
+        else:
+            label = "ERROR" if result.outcome == "failed" else result.outcome.upper()
+            value = label + (" (" + result.error_category + ")" if result.error_category else "")
+        if result.pending_caller_commit:
+            value += " (pending caller commit)"
+        output[result.result_key] = value
+    output["extract_preferences"] = report.preferences
+    return output
diff --git a/truememory/vector_search.py b/truememory/vector_search.py
index 507e4bb9..965ff0ab 100644
--- a/truememory/vector_search.py
+++ b/truememory/vector_search.py
@@ -739,13 +739,23 @@ def _encode_with_mps_fallback(model, texts, **kwargs):
 """
 
 
-def _capture_rebuild_model() -> tuple[object, tuple[int, str, int]]:
+def _capture_rebuild_model(conn: sqlite3.Connection | None = None) -> tuple[object, tuple[int, str, int]]:
+    """Caller transactions can borrow a cached model, but cannot wait or load."""
+    if conn is not None and conn.in_transaction:
+        if not _lock.acquire(blocking=False):
+            raise VectorPublicationChanged("Embedding model is busy; retry outside the caller transaction")
+        try:
+            if _model is None:
+                raise VectorPublicationChanged("Embedding model is not loaded; load it outside the caller transaction")
+            return _model, (_model_generation, EMBEDDING_MODEL, _embedding_dim)
+        finally:
+            _lock.release()
     with _lock:
         before = (_model_generation, EMBEDDING_MODEL)
     model = get_model()
     with _lock:
         if before != (_model_generation, EMBEDDING_MODEL):
-            raise RuntimeError("Embedding model changed while loading rebuild model")
+            raise VectorPublicationChanged("Embedding model changed while loading rebuild model")
         return model, (_model_generation, EMBEDDING_MODEL, _embedding_dim)
 
 
@@ -767,8 +777,9 @@ class VectorPublicationChanged(RuntimeError):
 
 
 @contextmanager
-def _foreground_model_fence(identity: tuple[int, str, int]) -> Iterator[None]:
-    if not _lock.acquire(blocking=False):
+def _foreground_model_fence(identity: tuple[int, str, int], *, blocking: bool = False) -> Iterator[None]:
+    """Blocking admission is permitted only before an owned writer starts."""
+    if not _lock.acquire(blocking=blocking):
         raise VectorPublicationChanged("Embedding model is busy; retry vector publication")
     try:
         if identity != (_model_generation, EMBEDDING_MODEL, _embedding_dim):
@@ -815,7 +826,7 @@ def _foreground_vector_publication(
 ) -> Iterator[tuple[str, str]]:
     from truememory.rebuild_source import rebuild_transaction
 
-    with rebuild_transaction(conn, write=True, publication_fence=_foreground_model_fence(identity)):
+    with _foreground_model_fence(identity, blocking=not conn.in_transaction), rebuild_transaction(conn, write=True):
         yield _validate_foreground_vector_target(conn, identity, message_id)
 
 
@@ -856,7 +867,7 @@ def _build_streamed_vectors(
                     save_manifest(conn, key, empty)
                     _clear_build_in_progress(conn, table)
                     return 0
-        model, identity = _capture_rebuild_model()
+        model, identity = _capture_rebuild_model(conn)
         with rebuild_transaction(conn, write=True, publication_fence=_rebuild_model_fence(identity)):
             manifest = load_manifest(conn, key)
             resumable = (manifest is not None and not manifest.complete
@@ -1191,7 +1202,7 @@ def embed_single(conn: sqlite3.Connection, message_id: int, content: str) -> Non
         message_id: The ``messages.id`` of the row being embedded.
         content:    The text to embed.
     """
-    model, identity = _capture_rebuild_model()
+    model, identity = _capture_rebuild_model(conn)
     embedding = _encode_with_mps_fallback(model, [content])[0]
     normed = _normalize_for_cosine(embedding)
     if normed is None: