From e5e87085df413b6bf3057be2a2eaba9967ee0510 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Fri, 9 Oct 2026 09:32:30 -0500 Subject: [PATCH 1/2] fix: Add reentrant serving admission for tier activation --- docs/resource-budgets.md | 55 +++ tests/test-tier-serving-gate-795.py | 572 ++++++++++++++++++++++++++++ truememory/tier_switch/serving.py | 406 ++++++++++++++++++++ 3 files changed, 1033 insertions(+) create mode 100644 tests/test-tier-serving-gate-795.py create mode 100644 truememory/tier_switch/serving.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 4fbb9b97..cf41e35b 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -2007,3 +2007,58 @@ exact pair generation, final writer ownership and activation remain separate. Synthetic tests use in-memory databases and mocked file observations only. This binding neither removes the public manager's retained source list nor proves the original Mac memory incident fixed. + +### Serving operation admission prerequisite + +`tier_switch/serving.py` adds an unused process-wide reader/exclusive-writer +boundary. `operation_lease(selection_key)` pins a caller-supplied identity for a +whole logical operation; `exclusive_activation()` drains readers and reserved +children before admitting a writer. The key must be an exact tuple of 1 through +8 exact strings, each 1 through 512 characters. It is opaque metadata, not a +certified database generation. Concurrent admitted keys must agree. Errors and +token representations do not include key contents. + +Same-thread reads are reentrant. Unrelated readers wait behind queued writers, +and writers enter in FIFO order. A live reader can call `fork_child()` to reserve +one parallel task before submitting it. The reservation immediately pins the +group and can outlive its parent; its single successful `join()` bypasses a +waiting writer so the parent can wait for its children without deadlocking. +Joined descendants retain the pin until their own contexts exit. This fairness +exception does not bound the duration or number of descendants in one operation. + +Unused reservations must be closed, including a task that never starts or whose +join fails before admission. Such a failure leaves the reservation available for +retry or explicit `close()`; it does not report successful admission. Close is +idempotent for an authentic unused or retired reservation and rejects a running +joined child. Retired identities are weakly retained, so completed reservations +do not accumulate a process-lifetime strong-reference history. Stale, forged, +foreign-gate and already consumed tokens cannot grant admission. + +Read-to-write upgrades fail immediately. Writers may reenter and synchronously +acquire reads, but those reads cannot export cross-thread child reservations. +A writer-owned read still pins admission if its outer writer exits first. +Every context must enter and exit on the same thread; transferring an entered +generator context across threads is unsupported and fails closed. Ordinary +same-thread body exceptions, including `BaseException`, release registrations. + +Optional timeout and absolute monotonic deadline bounds apply to admission, +including the bookkeeping mutex. Read reentry, child reservation and child join +inherit the parent's deadline and cancellation events. Each exclusive entry +uses its own supplied controls. Cancellation is checked before admission and +after waits. A supplied `threading.Event` sets a maximum timed-wait/poll interval +of 0.05 seconds, not a wall-clock return bound. `Condition.wait()` internally +reacquires the bookkeeping mutex without an interruptible deadline; mutex +reacquisition, cleanup and thread scheduling can delay return. Cancellation and +deadline are checked after reacquisition, so expired or cancelled work is still +refused before admission. There is no background polling thread. Admission bounds +do not interrupt an admitted body, native inference, SQLite work or cleanup. + +The module gate spans all databases within one process. Explicit gate instances +and inherited tokens reject a changed process ID before acquiring an inherited +lock; the module default is replaced in a forked child. This is not a +cross-process mutex or selected-generation writer check. No engine, vector, +MCP, manager, configuration or activation caller adopts it in this checkpoint. +The silent equal-dimension encode/table-selection race remains until complete +nested and parallel operations use this boundary with certified identities. +Local tests use only stdlib threads, events and controlled clocks. No serving +latency, native-memory reduction or fix for the original Mac incident is claimed. diff --git a/tests/test-tier-serving-gate-795.py b/tests/test-tier-serving-gate-795.py new file mode 100644 index 00000000..70951a9e --- /dev/null +++ b/tests/test-tier-serving-gate-795.py @@ -0,0 +1,572 @@ +"""Serving admission through stdlib events and controlled condition wait points.""" + +from __future__ import annotations + +import importlib.util +import gc +import math +import sys +import threading +import types +import unittest +import weakref +from pathlib import Path +from unittest.mock import patch + +SOURCE = Path(__file__).resolve().parents[1] / "truememory/tier_switch/serving.py" +KEY = ("synthetic-database", "synthetic-generation") + + +def load() -> types.ModuleType: + spec = importlib.util.spec_from_file_location("synthetic_serving_gate", SOURCE) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class Cancelled(BaseException): + pass + + +class TestServingGate(unittest.TestCase): + def setUp(self) -> None: + self.api = load() + self.gate = self.api.ServingGate() + self.errors = [] + self.threads = [] + self.releases = [] + self.waiting = {} + self.wait_arguments = [] + original = self.gate._condition.wait + + def observed_wait(timeout=None): + name = threading.current_thread().name + self.wait_arguments.append((name, timeout)) + if name in self.waiting: + self.waiting[name].set() + return original(timeout) + + self.gate._condition.wait = observed_wait + self.addCleanup(self.cleanup_threads) + + def cleanup_threads(self) -> None: + for event in self.releases: + event.set() + for thread in self.threads: + thread.join(timeout=3) + self.assertFalse(any(thread.is_alive() for thread in self.threads), "Synthetic thread did not finish") + + def event(self) -> threading.Event: + event = threading.Event() + self.releases.append(event) + return event + + def start(self, name, work): + done = threading.Event() + self.waiting[name] = threading.Event() + def run(): + try: + work() + except BaseException as error: + self.errors.append(error) + finally: + done.set() + thread = threading.Thread(target=run, name=name, daemon=True) + self.threads.append(thread) + thread.start() + return done + + def reached(self, event: threading.Event) -> None: + self.assertTrue(event.wait(timeout=3), "Synthetic event was not reached") + + def writer(self, entered, release=None, **kwargs): + def work(): + with self.gate.exclusive_activation(**kwargs): + entered.set() + if release is not None: + self.reached(release) + return work + + def assert_idle(self) -> None: + with self.gate.exclusive_activation(timeout=1): + self.assertFalse(self.gate._readers) + self.assertFalse(self.gate._children) + self.assertFalse(self.gate._waiting_writers) + self.assertFalse(self.gate._writers) + self.assertEqual(self.errors, []) + + def test_concurrent_readers_and_same_thread_nested_calls(self) -> None: + child_entered, child_release = threading.Event(), self.event() + def reader(): + with self.gate.operation_lease(KEY) as lease: + self.assertEqual(lease.selection_key, KEY) + child_entered.set() + self.reached(child_release) + with self.gate.operation_lease(KEY) as outer: + done = self.start("reader", reader) + self.reached(child_entered) + with self.gate.operation_lease(KEY) as inner: + self.assertEqual(outer.selection_key, inner.selection_key) + self.assertIsNot(inner, outer) + child_release.set() + self.reached(done) + self.assert_idle() + + def test_waiting_writer_blocks_unrelated_reader_but_not_nested_reader(self) -> None: + writer_entered, writer_release, reader_entered = threading.Event(), self.event(), threading.Event() + with self.gate.operation_lease(KEY): + writer_done = self.start("writer", self.writer(writer_entered, writer_release)) + self.reached(self.waiting["writer"]) + def reader(): + with self.gate.operation_lease(KEY): + reader_entered.set() + reader_done = self.start("late-reader", reader) + self.reached(self.waiting["late-reader"]) + with self.gate.operation_lease(KEY): + self.assertFalse(writer_entered.is_set()) + self.assertFalse(reader_entered.is_set()) + self.reached(writer_entered) + self.assertFalse(reader_entered.is_set()) + writer_release.set() + self.reached(writer_done) + self.reached(reader_done) + self.assert_idle() + + def test_reserved_child_outlives_parent_and_joins_ahead_of_waiting_writer(self) -> None: + writer_entered, child_entered, child_release = threading.Event(), threading.Event(), self.event() + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + writer_done = self.start("writer", self.writer(writer_entered)) + self.reached(self.waiting["writer"]) + def task(): + with child.join() as joined: + self.assertEqual(joined.selection_key, KEY) + child_entered.set() + self.reached(child_release) + child_done = self.start("child", task) + self.reached(child_entered) + self.assertFalse(writer_entered.is_set()) + child_release.set() + self.reached(child_done) + self.reached(writer_done) + with self.assertRaises(self.api.ServingLeaseError): + with child.join(): + self.fail("Consumed reservation was reused") + child.close() + self.assert_idle() + + def test_parent_can_wait_for_parallel_children_with_a_writer_queued(self) -> None: + writer_entered = threading.Event() + with self.gate.operation_lease(KEY) as lease: + children = [lease.fork_child() for _ in range(3)] + writer_done = self.start("writer", self.writer(writer_entered)) + self.reached(self.waiting["writer"]) + child_events = [] + for index, child in enumerate(children): + def task(reservation=child): + with reservation.join() as joined: + with self.gate.operation_lease(KEY): + self.assertEqual(joined.selection_key, KEY) + child_events.append(self.start(f"child-{index}", task)) + for done in child_events: + self.reached(done) + self.assertFalse(writer_entered.is_set()) + self.reached(writer_done) + self.assert_idle() + + def test_unused_reservation_close_unpins_writer_and_is_idempotent(self) -> None: + entered = threading.Event() + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + done = self.start("writer", self.writer(entered)) + self.reached(self.waiting["writer"]) + child.close() + child.close() + self.reached(done) + self.assertTrue(entered.is_set()) + with self.assertRaises(self.api.ServingLeaseError): + with child.join(): + self.fail("Closed reservation was joined") + self.assert_idle() + + def test_active_join_cannot_be_closed_or_joined_twice(self) -> None: + entered, release = threading.Event(), self.event() + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + def task(): + with child.join(): + entered.set() + self.reached(release) + done = self.start("child", task) + self.reached(entered) + with self.assertRaises(self.api.ServingLeaseError): + child.close() + with self.assertRaises(self.api.ServingLeaseError): + with child.join(): + self.fail("Concurrent duplicate join was admitted") + self.assertEqual(len(self.gate._readers), 1) + release.set() + self.reached(done) + self.assert_idle() + + def test_child_can_reserve_grandchild_without_extending_writer_privilege(self) -> None: + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + with child.join() as joined: + grandchild = joined.fork_child() + with grandchild.join() as final: + self.assertEqual(final.selection_key, KEY) + self.assert_idle() + + def test_read_upgrade_refuses_without_waiting_or_poisoning_queue(self) -> None: + with self.gate.operation_lease(KEY): + with self.assertRaises(self.api.ServingLeaseError): + with self.gate.exclusive_activation(): + self.fail("Read was upgraded") + self.assertFalse(self.gate._waiting_writers) + self.assert_idle() + + def test_writer_reentry_and_sync_reader_are_safe_but_not_exportable(self) -> None: + with self.gate.exclusive_activation(): + with self.gate.exclusive_activation(): + with self.gate.operation_lease(KEY) as lease: + with self.gate.exclusive_activation(): + self.assertEqual(lease.selection_key, KEY) + with self.assertRaises(self.api.ServingLeaseError): + lease.fork_child() + self.assert_idle() + + def test_writer_owned_read_still_pins_after_outer_writer_exits_first(self) -> None: + writer = self.gate.exclusive_activation() + writer.__enter__() + reader = self.gate.operation_lease(KEY) + lease = reader.__enter__() + writer.__exit__(None, None, None) + entered = threading.Event() + done = self.start("next-writer", self.writer(entered)) + self.reached(self.waiting["next-writer"]) + with self.gate.operation_lease(KEY) as nested: + with self.assertRaises(self.api.ServingLeaseError): + nested.fork_child() + self.assertFalse(entered.is_set()) + reader.__exit__(None, None, None) + self.reached(done) + with self.assertRaises(self.api.ServingLeaseError): + lease.fork_child() + self.assert_idle() + + def test_writer_fifo_and_cancelled_writer_wakes_unrelated_readers(self) -> None: + order, first_release = [], self.event() + def first(): + with self.gate.exclusive_activation(): + order.append("first") + self.reached(first_release) + def second(): + with self.gate.exclusive_activation(): + order.append("second") + with self.gate.operation_lease(KEY): + first_done = self.start("first-writer", first) + self.reached(self.waiting["first-writer"]) + second_done = self.start("second-writer", second) + self.reached(self.waiting["second-writer"]) + # The first writer must remain ahead until its release event. + self.reached(self.waiting["second-writer"]) + first_release.set() + self.reached(first_done) + self.reached(second_done) + self.assertEqual(order, ["first", "second"]) + self.assert_idle() + + def test_cancellation_while_blocker_is_held_uses_bounded_condition_wait(self) -> None: + cancel, done, reader_entered = threading.Event(), None, threading.Event() + outcomes = [] + def waiting_writer(): + try: + with self.gate.exclusive_activation(cancelled=cancel): + self.fail("Cancelled writer admitted") + except self.api.ServingLeaseCancelled: + outcomes.append("cancelled") + with self.gate.operation_lease(KEY): + done = self.start("cancel-writer", waiting_writer) + self.reached(self.waiting["cancel-writer"]) + def late_reader(): + with self.gate.operation_lease(KEY): + reader_entered.set() + late_done = self.start("reader-after-cancel", late_reader) + self.reached(self.waiting["reader-after-cancel"]) + cancel.set() + self.reached(done) + self.reached(late_done) + self.assertTrue(reader_entered.is_set()) + self.assertEqual(outcomes, ["cancelled"]) + waits = [timeout for name, timeout in self.wait_arguments if name == "cancel-writer"] + self.assertTrue(waits) + self.assertTrue(all(0 < timeout <= 0.05 for timeout in waits)) + self.assert_idle() + + def test_expired_admission_and_inherited_child_cancellation_release_resources(self) -> None: + cancel = threading.Event() + with self.gate.operation_lease(KEY, cancelled=cancel) as lease: + child = lease.fork_child() + cancel.set() + with self.assertRaises(self.api.ServingLeaseCancelled): + with child.join(): + self.fail("Child escaped inherited cancellation") + self.assertIn(child, self.gate._children) + child.close() + self.assertFalse(self.gate._children) + with patch.object(self.api.time, "monotonic", return_value=100.0): + for context in (self.gate.operation_lease(KEY, deadline=100.0), + self.gate.exclusive_activation(deadline=100.0)): + with self.assertRaises(self.api.ServingLeaseTimeout): + with context: + self.fail("Expired admission ran") + self.assert_idle() + + def test_deadline_consumed_during_wait_refuses_and_removes_writer_ticket(self) -> None: + clock = [100.0] + with self.gate.operation_lease(KEY): + def timed_wait(timeout=None): + clock[0] = 102.0 + # A separate thread is needed to avoid the explicit upgrade guard. + outcomes = [] + def waiter(): + try: + with self.gate.exclusive_activation(deadline=101.0): + self.fail("Expired queued writer ran") + except self.api.ServingLeaseTimeout: + outcomes.append("expired") + with patch.object(self.api.time, "monotonic", side_effect=lambda: clock[0]), patch.object(self.gate._condition, "wait", timed_wait): + done = self.start("deadline-writer", waiter) + self.reached(done) + self.assertEqual(outcomes, ["expired"]) + self.assertFalse(self.gate._waiting_writers) + self.assert_idle() + + def test_initial_bookkeeping_lock_timeout_does_not_wait_again_for_cleanup(self) -> None: + class RefusingCondition: + def __init__(self): + self.calls = [] + def acquire(self, **kwargs): + self.calls.append(kwargs) + clock[0] = 102.0 + return False + def release(self): + raise AssertionError("Unacquired lock released") + clock = [100.0] + condition = RefusingCondition() + for kind in ("reader", "writer"): + gate = self.api.ServingGate() + gate._condition = condition + clock[0] = 100.0 + with patch.object(self.api.time, "monotonic", side_effect=lambda: clock[0]): + context = gate.operation_lease(KEY, deadline=101.0) if kind == "reader" else gate.exclusive_activation(deadline=101.0) + with self.assertRaises(self.api.ServingLeaseTimeout): + with context: + self.fail("Timed-out bookkeeping admission ran") + self.assertFalse(gate._readers) + self.assertFalse(gate._writers) + self.assertEqual(len(condition.calls), 2) + self.assertTrue(all("timeout" in call for call in condition.calls)) + + def test_inherited_controls_apply_before_mutex_for_nested_read_fork_and_join(self) -> None: + for operation in ("nested", "fork", "join"): + for mode in ("deadline", "cancelled"): + with self.subTest(operation=operation, mode=mode): + gate = self.api.ServingGate() + clock, cancelled = [100.0], threading.Event() + options = {"deadline": 101.0} if mode == "deadline" else {"cancelled": cancelled} + with patch.object(self.api.time, "monotonic", side_effect=lambda: clock[0]): + parent = gate.operation_lease(KEY, **options) + lease = parent.__enter__() + child = lease.fork_child() if operation == "join" else None + if child is not None: + parent.__exit__(None, None, None) + real_condition = gate._condition + arguments = [] + class RefusingCondition: + def acquire(self, **kwargs): + arguments.append(kwargs) + clock[0] = 102.0 + cancelled.set() + return False + def release(self): + raise AssertionError("Unacquired inherited lock released") + gate._condition = RefusingCondition() + expected = self.api.ServingLeaseTimeout if mode == "deadline" else self.api.ServingLeaseCancelled + try: + with self.assertRaises(expected): + if operation == "fork": + lease.fork_child() + else: + context = child.join() if child is not None else gate.operation_lease(KEY) + with context: + self.fail("Inherited admission bound was ignored") + finally: + gate._condition = real_condition + self.assertEqual(len(arguments), 1) + self.assertIn("timeout", arguments[0]) + if mode == "cancelled": + self.assertLessEqual(arguments[0]["timeout"], 0.05) + clock[0] = 100.0 + cancelled.clear() + if child is not None: + self.assertIn(child, gate._children) + with child.join() as joined: + self.assertEqual(joined.selection_key, KEY) + else: + if operation == "fork": + retry = lease.fork_child() + retry.close() + else: + with gate.operation_lease(KEY): + pass + parent.__exit__(None, None, None) + with gate.exclusive_activation(): + self.assertFalse(gate._children) + self.assertFalse(gate._readers) + + def test_cancelled_unadmitted_child_can_be_closed_without_joining(self) -> None: + cancel = threading.Event() + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + cancel.set() + with self.assertRaises(self.api.ServingLeaseCancelled): + with child.join(cancelled=cancel): + self.fail("Cancelled child entered") + self.assertIn(child, self.gate._children) + child.close() + self.assert_idle() + + def test_retired_authenticity_registry_does_not_retain_closed_children(self) -> None: + references = [] + with self.gate.operation_lease(KEY) as lease: + for _ in range(32): + child = lease.fork_child() + references.append(weakref.ref(child)) + child.close() + del child + gc.collect() + self.assertTrue(all(reference() is None for reference in references)) + self.assertEqual(len(self.gate._retired_children), 0) + self.assert_idle() + + def test_child_join_failure_and_body_baseexception_release_exactly_once(self) -> None: + sentinel = Cancelled("synthetic interruption") + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + with self.assertRaises(Cancelled) as caught: + with child.join(): + raise sentinel + self.assertIs(caught.exception, sentinel) + self.assertFalse(self.gate._children) + for context in (self.gate.operation_lease(KEY), self.gate.exclusive_activation()): + with self.assertRaises(Cancelled) as caught: + with context: + raise sentinel + self.assertIs(caught.exception, sentinel) + self.assert_idle() + + def test_wait_baseexception_unregisters_writer_and_notifies_readers(self) -> None: + sentinel = Cancelled("synthetic wait interruption") + outcomes = [] + with self.gate.operation_lease(KEY): + with patch.object(self.gate._condition, "wait", side_effect=sentinel): + def waiter(): + try: + with self.gate.exclusive_activation(): + self.fail("Writer should not enter") + except Cancelled as error: + outcomes.append(error) + done = self.start("interrupted-writer", waiter) + self.reached(done) + self.assertEqual(outcomes, [sentinel]) + self.assertFalse(self.gate._waiting_writers) + self.assert_idle() + + def test_foreign_thread_cannot_fork_or_inspect_parent_lease(self) -> None: + outcomes = [] + with self.gate.operation_lease(KEY) as lease: + def wrong_thread(): + for call in (lease.fork_child, lambda: lease.selection_key): + try: + call() + except self.api.ServingLeaseError: + outcomes.append("refused") + done = self.start("foreign-reader", wrong_thread) + self.reached(done) + self.assertEqual(outcomes, ["refused", "refused"]) + self.assert_idle() + + def test_forged_foreign_stale_and_process_inherited_tokens_refuse(self) -> None: + for kind in (self.api.OperationLease, self.api.ChildReservation): + with self.assertRaises(self.api.ServingLeaseError): + kind() + forged = object.__new__(self.api.ChildReservation) + with self.assertRaises(self.api.ServingLeaseError): + forged.close() + object.__setattr__(forged, "_gate", self.gate) + object.__setattr__(forged, "_closed", True) + with self.assertRaises(self.api.ServingLeaseError): + forged.close() + with self.gate.operation_lease(KEY) as lease: + child = lease.fork_child() + self.assertNotIn(KEY[0], repr(child)) + foreign = self.api.ServingGate() + with self.assertRaises(self.api.ServingLeaseError): + foreign._close_child(child) + self.gate._condition.acquire() + try: + with patch.object(self.api.os, "getpid", return_value=self.gate._pid + 1): + with self.assertRaises(self.api.ServingLeaseError): + child.close() + with self.assertRaises(self.api.ServingLeaseError): + with self.gate.operation_lease(KEY): + self.fail("Inherited explicit gate admitted") + finally: + self.gate._condition.release() + child.close() + with self.assertRaises(self.api.ServingLeaseError): + lease.fork_child() + self.assert_idle() + + def test_module_gate_reset_rejects_inherited_tokens_and_allows_fresh_operations(self) -> None: + old = self.api._DEFAULT_GATE + with self.api.operation_lease(KEY) as lease: + child = lease.fork_child() + with patch.object(self.api.os, "getpid", return_value=old._pid + 1): + self.api._after_fork() + self.assertIsNot(self.api._DEFAULT_GATE, old) + with self.api.operation_lease(KEY) as fresh: + self.assertEqual(fresh.selection_key, KEY) + with self.assertRaises(self.api.ServingLeaseError): + with child.join(): + self.fail("Inherited reservation joined") + child.close() + + def test_keys_are_inert_bounded_exact_types_and_never_disclosed(self) -> None: + class Hostile(str): + def __eq__(self, other): + raise AssertionError("User comparator invoked") + def __len__(self): + raise AssertionError("User length invoked") + for key in (["synthetic"], (), ("",), ("x" * 513,), ("x",) * 9, (Hostile("secret-key"),), (True,)): + with self.assertRaises(self.api.ServingLeaseError) as caught: + with self.gate.operation_lease(key): + self.fail("Invalid key admitted") + self.assertNotIn("secret-key", str(caught.exception)) + with self.gate.operation_lease(KEY): + with self.assertRaises(self.api.ServingLeaseError) as caught: + with self.gate.operation_lease(("different-private-key",)): + self.fail("Selection mixed within operation") + self.assertNotIn("private", str(caught.exception)) + for value in (True, math.inf, math.nan, "1", 10**1000, -1): + with self.assertRaises(self.api.ServingLeaseError): + with self.gate.operation_lease(KEY, timeout=value): + self.fail("Invalid timeout admitted") + self.assert_idle() + + +if __name__ == "__main__": + unittest.main() diff --git a/truememory/tier_switch/serving.py b/truememory/tier_switch/serving.py new file mode 100644 index 00000000..769625b1 --- /dev/null +++ b/truememory/tier_switch/serving.py @@ -0,0 +1,406 @@ +"""Process-local serving admission; no model, database or caller integration. + +Readers and reserved children pin one caller-supplied selection key. Exclusive +activation drains those registrations. This gate does not validate a database +generation or apply a selection, and is not a cross-process publication lock. +""" + +from __future__ import annotations + +import math +import os +import threading +import time +import weakref +from collections import deque +from collections.abc import Iterator +from contextlib import AbstractContextManager, contextmanager +from dataclasses import dataclass + +SelectionKey = tuple[str, ...] +_CANCEL_POLL_SECONDS = 0.05 + + +class ServingLeaseError(RuntimeError): + """Serving admission or an opaque token's lifecycle is invalid.""" + + +class ServingLeaseTimeout(TimeoutError): + """The serving admission deadline expired.""" + + +class ServingLeaseCancelled(ServingLeaseError): + """Serving admission was cancelled before work could start.""" + + +@dataclass(frozen=True) +class _Control: + deadline: float | None + events: tuple[threading.Event, ...] + + +def _control(timeout: float | None, deadline: float | None, cancelled: threading.Event | None) -> _Control: + for value in (timeout, deadline): + try: + if value is not None and (type(value) not in (int, float) or not math.isfinite(value)): + raise ServingLeaseError("Serving admission requires a finite time bound") + except OverflowError: + raise ServingLeaseError("Serving admission requires a finite time bound") from None + if timeout is not None and timeout < 0: + raise ServingLeaseError("Serving admission timeout must be nonnegative") + if cancelled is not None and type(cancelled) is not threading.Event: + raise ServingLeaseError("Serving cancellation requires a threading Event") + if timeout is not None: + relative = time.monotonic() + timeout + deadline = relative if deadline is None else min(relative, deadline) + return _Control(deadline, () if cancelled is None else (cancelled,)) + + +def _combine(parent: _Control, child: _Control) -> _Control: + bounds = tuple(value for value in (parent.deadline, child.deadline) if value is not None) + events = parent.events + tuple(event for event in child.events if all(event is not old for old in parent.events)) + return _Control(min(bounds) if bounds else None, events) + + +def _check(control: _Control) -> None: + if any(threading.Event.is_set(event) for event in control.events): + raise ServingLeaseCancelled("Serving admission was cancelled") + if control.deadline is not None and time.monotonic() >= control.deadline: + raise ServingLeaseTimeout("Serving admission deadline expired") + + +def _wait_seconds(control: _Control) -> float | None: + seconds = None if control.deadline is None else max(0.0, control.deadline - time.monotonic()) + if control.events: + seconds = _CANCEL_POLL_SECONDS if seconds is None else min(seconds, _CANCEL_POLL_SECONDS) + return None if seconds is None else min(seconds, threading.TIMEOUT_MAX) + + +def _selection_key(value: SelectionKey) -> SelectionKey: + # Exact built-in types exclude user equality/hash callbacks under the lock. + if (type(value) is not tuple or not 1 <= len(value) <= 8 + or any(type(part) is not str or not 1 <= len(part) <= 512 for part in value)): + raise ServingLeaseError("Serving selection requires a bounded immutable key") + return value + + +class _OpaqueToken: + __slots__ = ("_gate", "_closed", "__weakref__") + + def __new__(cls) -> _OpaqueToken: + raise ServingLeaseError("Serving tokens are issued only by their gate") + + def __setattr__(self, name: str, value: object) -> None: + raise ServingLeaseError("Serving tokens are opaque") + + def __repr__(self) -> str: + return "" + + +class OperationLease(_OpaqueToken): + __slots__ = () + + @property + def selection_key(self) -> SelectionKey: + return _issuer(self)._lease_key(self) + + def fork_child(self) -> ChildReservation: + """Reserve one child now; close it if the task never enters join().""" + return _issuer(self)._fork_child(self) + + +class ChildReservation(_OpaqueToken): + __slots__ = () + + @contextmanager + def join( + self, *, timeout: float | None = None, deadline: float | None = None, + cancelled: threading.Event | None = None, + ) -> Iterator[OperationLease]: + """Consume once on admission; an unadmitted reservation still needs close().""" + gate = _issuer(self) + with gate._join_child(self, _control(timeout, deadline, cancelled)) as lease: + yield lease + + def close(self) -> None: + """Release an unused reservation; never release a running joined child.""" + _issuer(self)._close_child(self) + + +def _issuer(token: _OpaqueToken) -> ServingGate: + gate = getattr(token, "_gate", None) + if type(token) not in (OperationLease, ChildReservation) or type(gate) is not ServingGate: + raise ServingLeaseError("Invalid serving admission token") + gate._check_process() + return gate + + +@dataclass(frozen=True) +class _Reader: + thread: threading.Thread + key: SelectionKey + control: _Control + writer_owned: bool + reservation: ChildReservation | None = None + + +@dataclass +class _Reserved: + key: SelectionKey + control: _Control + joined: OperationLease | None = None + + +class ServingGate: + """Concurrent readers, FIFO exclusive writers, and explicit pinned children. + + An explicit gate is invalid after fork. Module-level functions receive a new + child-process gate through register_at_fork; inherited tokens remain invalid. + """ + + def __init__(self) -> None: + self._pid = os.getpid() + self._condition = threading.Condition(threading.Lock()) + self._local = threading.local() + self._readers: dict[OperationLease, _Reader] = {} + self._children: dict[ChildReservation, _Reserved] = {} + self._retired_children: weakref.WeakSet[ChildReservation] = weakref.WeakSet() + self._writers: dict[object, threading.Thread] = {} + self._waiting_writers: deque[object] = deque() + + def _check_process(self) -> None: + if os.getpid() != self._pid: + raise ServingLeaseError("Serving state cannot be inherited by another process") + + @contextmanager + def _locked(self, control: _Control | None = None) -> Iterator[None]: + self._check_process() + acquired = False + try: + while not acquired: + if control is None: + acquired = self._condition.acquire() + else: + _check(control) + seconds = _wait_seconds(control) + acquired = self._condition.acquire() if seconds is None else self._condition.acquire(timeout=seconds) + self._check_process() + if control is not None: + _check(control) + yield + finally: + if acquired: + self._condition.release() + + def _token(self, kind: type[OperationLease] | type[ChildReservation]) -> OperationLease | ChildReservation: + token = object.__new__(kind) + object.__setattr__(token, "_gate", self) + object.__setattr__(token, "_closed", False) + return token + + def _thread_reader(self, thread: threading.Thread) -> _Reader | None: + if thread is not threading.current_thread(): + raise ServingLeaseError("Serving thread state belongs to another thread") + readers = getattr(self._local, "readers", ()) + return readers[-1] if readers else None + + def _writer_thread(self) -> threading.Thread | None: + return next(iter(self._writers.values()), None) + + def _check_key(self, key: SelectionKey) -> None: + current = next((reader.key for reader in self._readers.values()), None) + if current is None: + current = next((child.key for child in self._children.values()), None) + if current is not None and current != key: + raise ServingLeaseError("Serving selection differs from an admitted operation") + + def _owned_reader(self, lease: OperationLease) -> _Reader: + if type(lease) is not OperationLease or getattr(lease, "_gate", None) is not self: + raise ServingLeaseError("Invalid serving operation lease") + reader = self._readers.get(lease) + if reader is None or reader.thread is not threading.current_thread(): + raise ServingLeaseError("Serving operation lease is closed or belongs to another thread") + return reader + + def _register_reader(self, lease: OperationLease, reader: _Reader) -> None: + readers = (*getattr(self._local, "readers", ()), reader) + self._readers[lease] = reader + self._local.readers = readers + + def _lease_key(self, lease: OperationLease) -> SelectionKey: + with self._locked(): + return self._owned_reader(lease).key + + def _release_reader(self, lease: OperationLease) -> None: + with self._locked(): + reader = self._readers.get(lease) + if reader is not None: + self._owned_reader(lease) + del self._readers[lease] + self._local.readers = tuple(item for item in self._local.readers if item is not reader) + if reader.reservation is not None: + del self._children[reader.reservation] + self._retired_children.add(reader.reservation) + object.__setattr__(reader.reservation, "_closed", True) + object.__setattr__(lease, "_closed", True) + self._condition.notify_all() + + @contextmanager + def operation_lease( + self, selection_key: SelectionKey, *, timeout: float | None = None, + deadline: float | None = None, cancelled: threading.Event | None = None, + ) -> Iterator[OperationLease]: + key, control = _selection_key(selection_key), _control(timeout, deadline, cancelled) + lease = self._token(OperationLease) + thread = threading.current_thread() + self._check_process() + parent = self._thread_reader(thread) + if parent is not None: + control = _combine(parent.control, control) + entered = False + try: + with self._locked(control): + entered = True + while True: + _check(control) + self._check_key(key) + writer = self._writer_thread() + if writer is thread or (writer is None and (parent is not None or not self._waiting_writers)): + self._register_reader(lease, _Reader(thread, key, control, writer is thread or bool(parent and parent.writer_owned))) + break + self._condition.wait(_wait_seconds(control)) + yield lease + finally: + if entered: + self._release_reader(lease) + + def _fork_child(self, lease: OperationLease) -> ChildReservation: + self._check_process() + # The owner's record is immutable and cannot be released by another + # thread. Revalidate membership after acquiring the condition as well. + control = self._owned_reader(lease).control + with self._locked(control): + reader = self._owned_reader(lease) + _check(reader.control) + if reader.writer_owned: + raise ServingLeaseError("Writer-owned reads cannot grant cross-thread admission") + child = self._token(ChildReservation) + self._children[child] = _Reserved(reader.key, reader.control) + return child + + def _close_child(self, child: ChildReservation) -> None: + with self._locked(): + if type(child) is not ChildReservation or getattr(child, "_gate", None) is not self: + raise ServingLeaseError("Invalid serving child reservation") + reserved = self._children.get(child) + if reserved is None: + if child not in self._retired_children: + raise ServingLeaseError("Invalid serving child reservation") + return + if reserved.joined is not None: + raise ServingLeaseError("A joined child must release its own operation lease") + del self._children[child] + self._retired_children.add(child) + object.__setattr__(child, "_closed", True) + self._condition.notify_all() + + @contextmanager + def _join_child(self, child: ChildReservation, control: _Control) -> Iterator[OperationLease]: + self._check_process() + if type(child) is not ChildReservation or getattr(child, "_gate", None) is not self: + raise ServingLeaseError("Invalid serving child reservation") + # Snapshot immutable inherited bounds before mutex admission. The + # reservation is revalidated under the condition before consumption. + snapshot = self._children.get(child) + if snapshot is None: + raise ServingLeaseError("Serving child reservation is closed or already joined") + control = _combine(snapshot.control, control) + thread = threading.current_thread() + parent = self._thread_reader(thread) + if parent is not None: + control = _combine(parent.control, control) + lease = self._token(OperationLease) + entered = False + try: + with self._locked(control): + entered = True + reserved = self._children.get(child) + if reserved is None or reserved.joined is not None: + raise ServingLeaseError("Serving child reservation is closed or already joined") + control = _combine(reserved.control, control) + _check(control) + self._check_key(reserved.key) + if self._writers: + raise ServingLeaseError("A reserved child cannot inherit writer ownership") + self._register_reader(lease, _Reader(thread, reserved.key, control, False, child)) + reserved.joined = lease + yield lease + finally: + if entered: + self._release_reader(lease) + + @contextmanager + def exclusive_activation( + self, *, timeout: float | None = None, deadline: float | None = None, + cancelled: threading.Event | None = None, + ) -> Iterator[None]: + control = _control(timeout, deadline, cancelled) + ticket = object() + thread = threading.current_thread() + entered = False + try: + with self._locked(control): + entered = True + if self._writer_thread() is thread: + self._writers[ticket] = thread + else: + if self._thread_reader(thread) is not None: + raise ServingLeaseError("A serving read lease cannot be upgraded to activation") + self._waiting_writers.append(ticket) + while True: + _check(control) + if (not self._writers and not self._readers and not self._children + and self._waiting_writers[0] is ticket): + self._waiting_writers.popleft() + self._writers[ticket] = thread + break + self._condition.wait(_wait_seconds(control)) + yield + finally: + if entered: + with self._locked(): + owner = self._writers.get(ticket) + if owner is not None and owner is not threading.current_thread(): + raise ServingLeaseError("Activation lease belongs to another thread") + self._writers.pop(ticket, None) + if ticket in self._waiting_writers: + self._waiting_writers.remove(ticket) + self._condition.notify_all() + + +_DEFAULT_GATE = ServingGate() + + +def _after_fork() -> None: + global _DEFAULT_GATE + _DEFAULT_GATE = ServingGate() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_after_fork) + + +def operation_lease( + selection_key: SelectionKey, *, timeout: float | None = None, deadline: float | None = None, + cancelled: threading.Event | None = None, +) -> AbstractContextManager[OperationLease]: + """Acquire the process-wide serving boundary for one logical operation.""" + return _DEFAULT_GATE.operation_lease(selection_key, timeout=timeout, deadline=deadline, cancelled=cancelled) + + +def exclusive_activation( + *, timeout: float | None = None, deadline: float | None = None, + cancelled: threading.Event | None = None, +) -> AbstractContextManager[None]: + """Drain process-wide operations before a caller's activation work.""" + return _DEFAULT_GATE.exclusive_activation(timeout=timeout, deadline=deadline, cancelled=cancelled) From e3e2c9b9d70d2016172451d7141789bf3b0595f1 Mon Sep 17 00:00:00 2001 From: buildingjoshbetter Date: Fri, 9 Oct 2026 09:32:30 -0500 Subject: [PATCH 2/2] fix: Certify durable tier selection before activation --- docs/resource-budgets.md | 68 +++ tests/test-tier-activation-journal-795.py | 645 ++++++++++++++++++++++ truememory/tier_switch/activation.py | 383 +++++++++++++ truememory/tier_switch/cache.py | 22 +- truememory/tier_switch/job.py | 54 ++ truememory/tier_switch/source.py | 29 + 6 files changed, 1197 insertions(+), 4 deletions(-) create mode 100644 tests/test-tier-activation-journal-795.py create mode 100644 truememory/tier_switch/activation.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index cf41e35b..e6dd556d 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -2062,3 +2062,71 @@ The silent equal-dimension encode/table-selection race remains until complete nested and parallel operations use this boundary with certified identities. Local tests use only stdlib threads, events and controlled clocks. No serving latency, native-memory reduction or fix for the original Mac incident is claimed. + +### Database-only certified tier selection (#795) + +`tier_switch.activation` is an unused database publication API. It does not +change config, model defaults, rerankers, serving state, or public tier callers. +The caller must establish coherent old serving state, an inactive target pair, +cooperative maintenance/stable-path ownership, and a drained local serving +boundary before final publication. These preconditions are not inferred or +proved by the journal. There is no legacy bootstrap, active-force/reset, or +alternate table slot. First selection with `expected_generation=None` certifies +only the newly completed target; it does not certify the old legacy vectors. + +`stage_activation_intent(conn, job, expected_generation=..., reranker_id=...)` +commits an immutable intent before source planning. An identical current intent +is idempotent without data writes; a conflicting intent for the same live job +is refused. A new intent is refused while the current selection still awaits +config acknowledgement, preserving its reconciliation result until that CAS +completes. Staged intents must name the current selected generation, or None +only when no selection exists. A replacement job must first satisfy the selected-job admission +protocol. The supplied effective reranker identity is frozen, not loaded or +resolved from environment/config. Embedding tier/model/dimension/group come +from the exact `EmbeddingTarget`; neither identity is substituted during SQL. + +The two versioned metadata records hold bounded descriptors, scalar counters, +hashes, intent/job/selection generations, source identity, and explicit pair +names. Each JSON value is limited to 16,384 UTF-8 bytes. Reads test type and +`length(CAST(value AS BLOB))` before copying the value into Python, including +NUL-tail cases. Parsers reject duplicate/unknown fields and invalid scalar +types. Metadata and registry must already have supported ordinary schemas; +TEMP objects, extra indexes, and side-effect triggers are refused. No schema +creation occurs during staging, certification, acknowledgement, or readback. + +`commit_certified_selection(conn, intent, job, plan)` owns one BEGIN IMMEDIATE. +It checks the exact current selected job and previous selection generation, +intent, frozen target, native vec0 cosine schema/dimension, complete source +manifest, paired receipt, and source revision/current coverage. The final pair +digest audit reads bounded pages and costs O(N) once at final certification, +not after each source page. New writer-side source/job helpers do not mutate, +commit, or roll back. They require caller-owned BEGIN IMMEDIATE; Python sqlite3 +can verify an active transaction but cannot expose whether it is DEFERRED or +IMMEDIATE, so they cannot independently prove the required writer lock kind. + +Successful publication writes exact registry model/dimension/pair/count/cursor, +embedder metadata, selection, and intent result in the same transaction, then +returns a config-pending selected record. Metadata and additive registry +`commit=False` writes use UPSERT without deleting the primary key. The old +registry default retains REPLACE plus commit. Successful job retirement deletes +only its exact selected marker inside that same transaction; the journal keeps +its job identity. Incoming foreign keys to the marker are refused even with +enforcement disabled, using at most 4,097 bounded table-name observations to +enforce a 4,096-table limit and 512-byte name bound. Rollback restores the marker; +successful commit allows normal admission of a new job without a false failed +or cancelled label. + +Commit/rollback/cancellation errors propagate, including exceptions raised after +a successful commit. No handler claims rollback or completion from an ambiguous +outcome or writes a follow-up status row. `read_activation_state` is separate +bounded readback on a known-safe connection, not fresh vector certification or +permission to reuse an uncertain connection. `acknowledge_config(generation=...)` +is an idempotent generation CAS after caller-confirmed config mirroring; it +performs no config write and acknowledges no process's runtime state. + +Safe tests use in-memory SQLite, actual source/marker control flow, synthetic +vectors, stub file observations, and explicit vec0 schema doubles. The opt-in +`TRUEMEMORY_TEST_NATIVE_VEC=1` smoke uses actual in-memory vec0 tables on GPUBox. +Runtime reconciliation, operation leases, base/pro config-only publication, and +manager/MCP/CLI adoption remain separate work. This API has no claimed public +memory reduction and does not prove the original 26.24 GB Mac incident fixed. diff --git a/tests/test-tier-activation-journal-795.py b/tests/test-tier-activation-journal-795.py new file mode 100644 index 00000000..647be2c1 --- /dev/null +++ b/tests/test-tier-activation-journal-795.py @@ -0,0 +1,645 @@ +"""Database journal faults with synthetic in-memory SQL and no model imports.""" + +from __future__ import annotations + +import builtins +import dataclasses +import json +import os +import runpy +import sqlite3 +import struct +import sys +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +BASE = runpy.run_path(str(Path(__file__).with_name("test-tier-job-identity-795.py"))) +ROOT = Path(__file__).resolve().parents[1] +TABLES = ("vec_messages_custom", "vec_messages_sep_custom") + + +class Rows: + def __init__(self, values: list[tuple]) -> None: + self.values = values + + def fetchone(self): + return self.values[0] if self.values else None + + +class JournalConnection(BASE["MockFileConnection"]): + fake_vec_schema = True + vec_declaration: str | None = None + armed = False + fault_contains: str | None = None + fault: BaseException | None = None + commit_mode = "" + rollback_mode = "" + commits = 0 + rollbacks = 0 + record_payload_reads = 0 + + def execute(self, sql: str, parameters: tuple = ()): + if sql == "SELECT value FROM main.metadata WHERE key=?" and parameters[0] in { + "tier_selected_v1", "tier_activation_v1", + }: + self.record_payload_reads += 1 + if self.armed and self.fault_contains is not None and self.fault_contains in sql: + raise self.fault + if (self.fake_vec_schema and sql == "SELECT sql FROM main.sqlite_master WHERE name=? COLLATE NOCASE" + and parameters[0] in TABLES): + ddl = self.vec_declaration or ( + f'CREATE VIRTUAL TABLE "{parameters[0]}" USING vec0(embedding float[2] distance_metric=cosine)' + ) + return Rows([(ddl,)]) + return super().execute(sql, parameters) + + def commit(self) -> None: + self.commits += 1 + if self.armed and self.commit_mode == "before": + raise self.fault + if self.armed and self.commit_mode == "retained": + return + super().commit() + if self.armed and self.commit_mode == "after": + raise self.fault + + def rollback(self) -> None: + self.rollbacks += 1 + if self.armed and self.rollback_mode == "before": + raise self.fault + if self.armed and self.rollback_mode == "retained": + return + super().rollback() + if self.armed and self.rollback_mode == "after": + raise self.fault + + +class TestActivationJournal(unittest.TestCase): + native_pair = False + + def setUp(self) -> None: + self.fixture = BASE["TestTierJobIdentity"]() + self.fixture.setUp() + self.addCleanup(self.fixture.doCleanups) + self.modules = self.fixture.modules + + def safe_import(name, globals=None, locals=None, fromlist=(), level=0): + if name in {"torch", "numpy", "sentence_transformers", "model2vec", "sqlite_vec"}: + raise AssertionError("Journal imports must stay model/native free") + if name == "psutil": + return types.SimpleNamespace() + if name.startswith("truememory."): + short = name.removeprefix("truememory.") + if short in self.modules: + return self.modules[short] + raise AssertionError("Unexpected application import: " + name) + return builtins.__import__(name, globals, locals, fromlist, level) + + for name in ("tier_config", "tier_switch.cache", "tier_switch.source", "tier_switch.activation"): + module = types.ModuleType("synthetic_activation_" + name.replace(".", "_")) + sys.modules[module.__name__] = module + module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import) + path = ROOT / "truememory" / (name.replace(".", "/") + ".py") + exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), module.__dict__) + self.modules[name] = module + self.api = self.modules["tier_switch.activation"] + self.job_api = self.modules["tier_switch.job"] + self.source = self.modules["tier_switch.source"] + self.conn = BASE["CONNECT"](":memory:", factory=JournalConnection) + self.addCleanup(self.conn.close) + self.conn.executescript(""" + CREATE TABLE messages(id INTEGER PRIMARY KEY, content TEXT, sender TEXT, recipient TEXT, timestamp TEXT); + CREATE TABLE metadata(key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT); + CREATE TABLE vector_cache_registry ( + tier_group TEXT PRIMARY KEY, vec_table TEXT NOT NULL, sep_table TEXT NOT NULL, + last_embedded_id INTEGER DEFAULT 0, vector_count INTEGER DEFAULT 0, + model_name TEXT, embedding_dim INTEGER DEFAULT 256, last_updated REAL, created REAL + ); + CREATE TABLE old_serving_pair(embedding BLOB); + INSERT INTO old_serving_pair(rowid,embedding) VALUES(9,x'0102'); + INSERT INTO metadata(key,value) VALUES('embed_model','old-synthetic'),('embed_dim','7'); + """) + if self.native_pair: + import sqlite_vec + self.conn.fake_vec_schema = False + self.conn.enable_load_extension(True) + try: + sqlite_vec.load(self.conn) + finally: + self.conn.enable_load_extension(False) + for name in TABLES: + self.conn.execute(f'CREATE VIRTUAL TABLE "{name}" USING vec0(embedding float[2] distance_metric=cosine)') + else: + for name in TABLES: + self.conn.execute(f'CREATE TABLE "{name}"(embedding BLOB)') + self.modules["rebuild_source"].ensure_rebuild_tracking(self.conn) + self.target = self.modules["embedding_target"].EmbeddingTarget("custom", "synthetic/encoder", 2, "custom") + self.job = self.job_api.select_tier_job(self.conn, self.target) + self.reranker = "synthetic/reranker" + + def stage(self, *, expected=None, reranker=None): + return self.api.stage_activation_intent( + self.conn, self.job, expected_generation=expected, reranker_id=reranker or self.reranker, + ) + + def build(self, ids=(-3, 0, 5), *, intent=None): + if ids: + self.conn.executemany("INSERT INTO messages VALUES(?,?,?,?,?)", [ + (mid, f"synthetic-{mid}", "sender", "recipient", "2026-01-01") for mid in ids + ]) + self.conn.commit() + intent = self.stage() if intent is None else intent + plan = self.source.plan_tier_source( + self.conn, model_id=self.target.model_id, dimension=self.target.dimension, targets=self.target.tables, + ) + plan = self.source.initialize_tier_source(self.conn, plan) + while plan.manifest.consumed < plan.manifest.total: + page = self.source.read_tier_page(self.conn, plan, size=2) + count = len(page.rows) + plan, pending = self.source.publish_tier_prefix( + self.conn, plan, page, completion=[struct.pack("2f", 1, 0)] * count, + separation=[struct.pack("2f", 0, 1)] * count, + ) + self.assertEqual(pending.rows, ()) + completion = self.source.finish_tier_source(self.conn, plan) + self.assertTrue(completion.current_source_complete) + return intent, completion.plan + + def commit(self, intent, plan): + return self.api.commit_certified_selection(self.conn, intent, self.job, plan) + + def marker(self): + return self.conn.execute(f"SELECT * FROM {self.job_api._SELECTION_TABLE}").fetchone() + + def assert_old(self) -> None: + self.assertEqual(self.conn.execute("SELECT value FROM main.metadata WHERE key='embed_model'").fetchone(), ("old-synthetic",)) + self.assertEqual(self.conn.execute("SELECT value FROM main.metadata WHERE key='embed_dim'").fetchone(), ("7",)) + self.assertIsNone(self.conn.execute("SELECT value FROM main.metadata WHERE key='tier_selected_v1'").fetchone()) + self.assertEqual(self.conn.execute("SELECT * FROM old_serving_pair").fetchall(), [(b"\x01\x02",)]) + + def test_owned_certification_exact_records_one_commit_and_job_retirement(self) -> None: + intent, plan = self.build() + before = self.conn.commits + selected = self.commit(intent, plan) + self.assertEqual(self.conn.commits, before + 1) + self.assertFalse(self.conn.in_transaction) + self.assertIsNone(self.marker()) + self.assertEqual(selected.target, self.target) + self.assertEqual(selected.reranker_id, self.reranker) + self.assertEqual(selected.tables, TABLES) + self.assertEqual((selected.vector_count, selected.cursor), (3, 5)) + self.assertFalse(selected.config_acknowledged) + self.assertEqual(self.api.read_activation_state(self.conn).selection, selected) + self.assertEqual(self.api.read_activation_state(self.conn).intent.state, "db_selected") + self.assertEqual(self.conn.execute("SELECT vec_table,sep_table,last_embedded_id,vector_count,model_name,embedding_dim FROM vector_cache_registry").fetchone(), + (*TABLES, 5, 3, "synthetic/encoder", 2)) + self.assertEqual(self.conn.execute("SELECT * FROM old_serving_pair").fetchall(), [(b"\x01\x02",)]) + self.assertEqual(self.conn.execute("SELECT value FROM metadata WHERE key='embed_model'").fetchone(), ("synthetic/encoder",)) + self.assertEqual(self.conn.execute("SELECT value FROM metadata WHERE key='embed_dim'").fetchone(), ("2",)) + new_job = self.job_api.select_tier_job(self.conn, self.target) + self.assertNotEqual(new_job.job_id, self.job.job_id) + self.assertEqual(self.marker()[1], new_job.job_id) + with self.assertRaises(self.job_api.TierJobSelectionError): + self.conn.execute("BEGIN IMMEDIATE") + try: + self.job_api._retire_selected_tier_job_in_writer(self.conn, self.job) + finally: + self.conn.rollback() + self.assertEqual(self.marker()[1], new_job.job_id) + + def test_empty_source_is_certified_without_fabricated_last_id(self) -> None: + intent, plan = self.build(ids=()) + result = self.commit(intent, plan) + self.assertEqual((result.vector_count, result.cursor), (0, None)) + self.assertEqual(self.conn.execute("SELECT last_embedded_id,vector_count FROM vector_cache_registry").fetchone(), (0, 0)) + + def test_stage_is_idempotent_and_conflicting_live_intent_refuses(self) -> None: + intent = self.stage() + before = self.conn.total_changes + self.assertEqual(self.stage(), intent) + self.assertEqual(self.conn.total_changes, before) + with self.assertRaises(self.api.TierActivationError): + self.stage(reranker="synthetic/other") + self.assertEqual(self.api.read_activation_state(self.conn).intent, intent) + self.assert_old() + + def test_replacement_job_may_replace_abandoned_intent_only_after_binding(self) -> None: + first = self.stage() + self.job_api.end_selected_tier_job(self.conn, self.job, outcome="failed") + self.job = self.job_api.select_tier_job(self.conn, self.target, previous_job_id=self.job.job_id) + second = self.stage() + self.assertNotEqual(first.intent_id, second.intent_id) + self.assertEqual(second.job_id, self.job.job_id) + self.assert_old() + + def test_config_acknowledgement_is_generation_cas_and_no_runtime_claim(self) -> None: + intent, plan = self.build() + selected = self.commit(intent, plan) + with self.assertRaises(self.api.TierActivationError): + self.api.acknowledge_config(self.conn, generation="0" * 32) + self.assertFalse(self.api.read_activation_state(self.conn).selection.config_acknowledged) + acked = self.api.acknowledge_config(self.conn, generation=selected.generation) + self.assertTrue(acked.config_acknowledged) + before = self.conn.total_changes + self.assertEqual(self.api.acknowledge_config(self.conn, generation=selected.generation), acked) + self.assertEqual(self.conn.total_changes, before) + self.assertNotIn("runtime", dataclasses.asdict(acked)) + self.job = self.job_api.select_tier_job(self.conn, self.target) + self.stage(expected=selected.generation) + with self.assertRaises(self.api.TierActivationError): + self.api.acknowledge_config(self.conn, generation=selected.generation) + + def test_missing_or_wrong_expected_generation_never_overwrites_selection(self) -> None: + with self.assertRaises(self.api.TierActivationError): + self.stage(expected="0" * 32) + intent, plan = self.build() + selected = self.commit(intent, plan) + self.job = self.job_api.select_tier_job(self.conn, self.target) + for expected in (None, "0" * 32, True): + with self.subTest(expected=expected), self.assertRaises(self.api.TierActivationError): + self.stage(expected=expected) + self.assertEqual(self.api.read_activation_state(self.conn).selection, selected) + + def test_final_writer_rejects_changed_selected_generation(self) -> None: + intent, plan = self.build() + previous = self.commit(intent, plan) + self.api.acknowledge_config(self.conn, generation=previous.generation) + self.job = self.job_api.select_tier_job(self.conn, self.target) + intent = self.stage(expected=previous.generation) + plan = self.source.plan_tier_source(self.conn, model_id=self.target.model_id, + dimension=2, targets=TABLES) + changed = dataclasses.replace(previous, generation="0" * 32) + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_selected_v1'", (self.api._dump(changed),)) + self.conn.commit() + marker = self.marker() + with self.assertRaisesRegex(self.api.TierActivationError, "Selected generation changed"): + self.commit(intent, plan) + self.assertEqual(self.marker(), marker) + raw = self.conn.execute("SELECT value FROM main.metadata WHERE key='tier_selected_v1'").fetchone()[0] + self.assertEqual(self.api._parse(raw, self.api.TierSelection), changed) + with self.assertRaises(self.api.TierActivationError): + self.api.read_activation_state(self.conn) + + def test_staged_intent_must_match_current_selection_or_absence(self) -> None: + intent = self.stage() + malformed = dataclasses.replace(intent, expected_generation="0" * 32) + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_activation_v1'", (self.api._dump(malformed),)) + self.conn.commit() + with self.assertRaisesRegex(self.api.TierActivationError, "Selected generation changed"): + self.api.read_activation_state(self.conn) + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_activation_v1'", (self.api._dump(intent),)) + self.conn.commit() + intent, plan = self.build(intent=intent) + selected = self.commit(intent, plan) + for expected in (None, "0" * 32, selected.generation): + malformed = dataclasses.replace(intent, expected_generation=expected) + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_activation_v1'", (self.api._dump(malformed),)) + self.conn.commit() + with self.subTest(expected=expected), self.assertRaises(self.api.TierActivationError): + self.api.read_activation_state(self.conn) + self.assertNotEqual(selected.generation, "0" * 32) + + def test_new_intent_preserves_pending_config_acknowledgement_then_retries(self) -> None: + intent, plan = self.build() + selected = self.commit(intent, plan) + previous_state = self.api.read_activation_state(self.conn) + self.job = self.job_api.select_tier_job(self.conn, self.target) + before = self.conn.total_changes + with self.assertRaisesRegex(self.api.TierActivationError, "awaits config acknowledgement"): + self.stage(expected=selected.generation) + self.assertEqual(self.conn.total_changes, before) + self.assertEqual(self.api.read_activation_state(self.conn), previous_state) + acked = self.api.acknowledge_config(self.conn, generation=selected.generation) + self.assertTrue(acked.config_acknowledged) + next_intent = self.stage(expected=selected.generation) + self.assertEqual(next_intent.expected_generation, selected.generation) + self.assertNotEqual(next_intent.intent_id, intent.intent_id) + + def test_frozen_custom_identity_never_reresolves_configuration(self) -> None: + intent, plan = self.build() + with patch.object(self.modules["tier_config"], "get_tier_config", side_effect=AssertionError("No config reads")): + result = self.commit(intent, plan) + self.assertEqual(result.target, self.target) + self.assertEqual(result.reranker_id, self.reranker) + + def test_idempotent_intent_does_not_invalidate_existing_source_plan(self) -> None: + intent, plan = self.build() + before = self.conn.total_changes + self.assertEqual(self.stage(), intent) + self.assertEqual(self.conn.total_changes, before) + self.assertEqual(self.commit(intent, plan).manifest_generation, plan.manifest.generation) + + def test_borrowed_transactions_and_read_only_helpers_never_end_caller_work(self) -> None: + intent, plan = self.build() + self.conn.execute("BEGIN IMMEDIATE") + before = self.conn.total_changes + self.job_api.check_selected_tier_job_in_writer(self.conn, self.job) + self.source.require_current_completion_in_writer(self.conn, plan) + self.assertEqual(self.conn.total_changes, before) + self.assertTrue(self.conn.in_transaction) + for call in (lambda: self.stage(), lambda: self.commit(intent, plan), + lambda: self.api.read_activation_state(self.conn), + lambda: self.api.acknowledge_config(self.conn, generation="0" * 32)): + with self.assertRaises(self.api.TierActivationError): + call() + self.assertTrue(self.conn.in_transaction) + self.assertEqual(self.conn.total_changes, before) + self.conn.rollback() + for call in (lambda: self.job_api.check_selected_tier_job_in_writer(self.conn, self.job), + lambda: self.source.require_current_completion_in_writer(self.conn, plan)): + with self.assertRaises((self.job_api.TierJobSelectionError, self.source.TierSourceUntrusted)): + call() + + def test_post_stage_unrelated_status_write_invalidates_plan(self) -> None: + intent, plan = self.build() + self.conn.execute("INSERT INTO metadata VALUES('unrelated','status',NULL)") + self.conn.commit() + with self.assertRaises(self.source.TierSourceUntrusted): + self.commit(intent, plan) + self.assert_old() + self.assertIsNotNone(self.marker()) + + def test_writer_rechecks_source_marker_receipt_and_target(self) -> None: + cases = ( + "INSERT INTO messages VALUES(8,'tail','sender','recipient','time')", + "UPDATE messages SET content='corrected' WHERE id=0", + f"UPDATE {self.job_api._SELECTION_TABLE} SET state='cancelled'", + "UPDATE metadata SET value='{}' WHERE key LIKE 'tier_pair_receipt_v1:%'", + "UPDATE vec_messages_custom SET embedding=x'0000000000000000' WHERE rowid=0", + ) + for mutation in cases: + with self.subTest(mutation=mutation): + fixture = TestActivationJournal() + fixture.setUp() + try: + intent, plan = fixture.build() + fixture.conn.execute(mutation) + fixture.conn.commit() + with self.assertRaises((fixture.api.TierActivationError, fixture.source.TierSourceUntrusted, + fixture.job_api.TierJobSelectionError)): + fixture.commit(intent, plan) + fixture.assert_old() + finally: + fixture.doCleanups() + + def test_revalidation_rejects_same_width_corrupted_pair(self) -> None: + intent, plan = self.build() + self.conn.execute("UPDATE vec_messages_custom SET embedding=? WHERE rowid=0", (struct.pack("2f", 0, 0),)) + self.conn.commit() + # Bypass only the cheap change counter to exercise the independent final + # digest audit; no forged receipt is supplied. + plan = dataclasses.replace(plan, changes=self.conn.total_changes) + with self.assertRaises(self.source.TierSourceUntrusted): + self.commit(intent, plan) + self.assert_old() + + def test_incomplete_fabricated_and_wrong_identity_certificates_refuse(self) -> None: + intent, plan = self.build() + altered = ( + dataclasses.replace(plan, action="resume"), + dataclasses.replace(plan, digest="0" * 64), + dataclasses.replace(plan, manifest=dataclasses.replace(plan.manifest, dimension=3)), + dataclasses.replace(plan, manifest=dataclasses.replace(plan.manifest, model="synthetic/wrong")), + dataclasses.replace(plan, manifest=dataclasses.replace(plan.manifest, complete=False)), + ) + for current in altered: + with self.subTest(plan=current.manifest.dimension), self.assertRaises((self.api.TierActivationError, self.source.TierSourceUntrusted)): + self.commit(intent, current) + self.assert_old() + with self.assertRaises(self.api.TierActivationError): + self.commit(dataclasses.replace(intent, reranker_id="synthetic/changed"), plan) + + def test_schema_dimension_metric_and_ordinary_table_refuse(self) -> None: + intent, plan = self.build() + for ddl in ( + "CREATE TABLE vec_messages_custom(embedding BLOB)", + "CREATE VIRTUAL TABLE vec_messages_custom USING vec0(embedding float[3] distance_metric=cosine)", + "CREATE VIRTUAL TABLE vec_messages_custom USING vec0(embedding float[2] distance_metric=l2)", + ): + self.conn.vec_declaration = ddl + with self.subTest(ddl=ddl), self.assertRaises(self.api.TierActivationError): + self.commit(intent, plan) + self.assert_old() + + def test_trigger_side_effects_and_temp_case_aliases_refuse_before_writes(self) -> None: + for ddl in ( + "CREATE TRIGGER bad_metadata AFTER UPDATE ON metadata BEGIN UPDATE messages SET content='changed'; END", + "CREATE TRIGGER bad_registry AFTER INSERT ON vector_cache_registry BEGIN DELETE FROM messages; END", + "CREATE INDEX bad_index ON metadata(value)", + "CREATE TEMP TABLE METADATA(key TEXT,value TEXT)", + "CREATE TEMP TABLE VECTOR_CACHE_REGISTRY(tier_group TEXT)", + ): + with self.subTest(ddl=ddl): + fixture = TestActivationJournal() + fixture.setUp() + try: + intent, plan = fixture.build() + fixture.conn.execute(ddl) + before = fixture.conn.total_changes + with self.assertRaises(fixture.api.TierActivationError): + fixture.commit(intent, plan) + self.assertEqual(fixture.conn.total_changes, before) + fixture.assert_old() + finally: + fixture.doCleanups() + + def test_marker_incoming_foreign_key_refuses_even_with_enforcement_off(self) -> None: + for enforcement in (False, True): + with self.subTest(enforcement=enforcement): + fixture = TestActivationJournal() + fixture.setUp() + try: + fixture.conn.execute(f"PRAGMA foreign_keys={int(enforcement)}") + fixture.conn.execute(f"CREATE TABLE marker_child(id INTEGER REFERENCES {fixture.job_api._SELECTION_TABLE.upper()}(singleton) ON DELETE CASCADE)") + fixture.conn.execute("INSERT INTO marker_child VALUES(1)") + fixture.conn.commit() + intent, plan = fixture.build() + with self.assertRaises(fixture.job_api.TierJobSelectionError): + fixture.commit(intent, plan) + self.assertEqual(fixture.conn.execute("SELECT * FROM marker_child").fetchall(), [(1,)]) + self.assertIsNotNone(fixture.marker()) + fixture.assert_old() + finally: + fixture.doCleanups() + + def test_registry_and_metadata_upserts_do_not_delete_parent_rows(self) -> None: + self.conn.execute("PRAGMA foreign_keys=ON") + self.conn.executescript(""" + INSERT INTO vector_cache_registry(tier_group,vec_table,sep_table) VALUES('custom','old','old_sep'); + CREATE TABLE metadata_child(parent TEXT REFERENCES metadata(key) ON DELETE CASCADE); + INSERT INTO metadata_child VALUES('embed_model'); + CREATE TABLE registry_child(parent TEXT REFERENCES vector_cache_registry(tier_group) ON DELETE CASCADE); + INSERT INTO registry_child VALUES('custom'); + """) + intent, plan = self.build() + self.commit(intent, plan) + self.assertEqual(self.conn.execute("SELECT * FROM metadata_child").fetchall(), [("embed_model",)]) + self.assertEqual(self.conn.execute("SELECT * FROM registry_child").fetchall(), [("custom",)]) + + def test_marker_retirement_bounds_table_names_before_materialization(self) -> None: + self.conn.execute('CREATE TABLE "' + 'x' * 600 + '"(id INTEGER)') + intent, plan = self.build() + with self.assertRaisesRegex(self.job_api.TierJobSelectionError, "schema exceeds"): + self.commit(intent, plan) + self.assertIsNotNone(self.marker()) + self.assert_old() + + def test_each_publication_write_failure_rolls_back_marker_registry_and_journal(self) -> None: + for fault in ("DELETE FROM main.truememory_tier_selected_job_v1", "INSERT INTO vector_cache_registry", + "INSERT INTO main.metadata"): + with self.subTest(fault=fault): + fixture = TestActivationJournal() + fixture.setUp() + try: + intent, plan = fixture.build() + marker = fixture.marker() + failure = sqlite3.OperationalError("synthetic statement failure") + fixture.conn.armed, fixture.conn.fault_contains, fixture.conn.fault = True, fault, failure + with self.assertRaises(sqlite3.OperationalError) as caught: + fixture.commit(intent, plan) + self.assertIs(caught.exception, failure) + fixture.conn.armed = False + self.assertFalse(fixture.conn.in_transaction) + self.assertEqual(fixture.marker(), marker) + fixture.assert_old() + self.assertEqual(fixture.api.read_activation_state(fixture.conn).intent.state, "staged") + finally: + fixture.doCleanups() + + def test_commit_before_after_and_noop_never_fake_success(self) -> None: + for mode in ("before", "after", "retained"): + with self.subTest(mode=mode): + fixture = TestActivationJournal() + fixture.setUp() + try: + intent, plan = fixture.build() + failure = BASE["Cancelled"]("synthetic commit cancellation") + fixture.conn.armed, fixture.conn.commit_mode, fixture.conn.fault = True, mode, failure + with self.assertRaises((BASE["Cancelled"], fixture.api.TierActivationError)): + fixture.commit(intent, plan) + fixture.conn.armed = False + self.assertFalse(fixture.conn.in_transaction) + state = fixture.api.read_activation_state(fixture.conn) + if mode == "after": + self.assertIsNotNone(state.selection) + self.assertFalse(state.selection.config_acknowledged) + self.assertIsNone(fixture.marker()) + else: + fixture.assert_old() + self.assertIsNotNone(fixture.marker()) + finally: + fixture.doCleanups() + + def test_uncertain_rollback_propagates_without_followup_status_sql(self) -> None: + for mode in ("before", "after", "retained"): + with self.subTest(mode=mode): + fixture = TestActivationJournal() + fixture.setUp() + try: + intent, plan = fixture.build() + fixture.conn.armed = True + fixture.conn.fault_contains = "INSERT INTO main.metadata" + fixture.conn.fault = MemoryError("synthetic rollback failure") + fixture.conn.rollback_mode = mode + before = fixture.conn.commits + with self.assertRaises((MemoryError, fixture.api.TierActivationError)): + fixture.commit(intent, plan) + self.assertEqual(fixture.conn.commits, before) + self.assertEqual(fixture.conn.in_transaction, mode != "after") + if fixture.conn.in_transaction: + with self.assertRaises(fixture.api.TierActivationError): + fixture.api.read_activation_state(fixture.conn) + fixture.conn.armed = False + fixture.conn.rollback() + fixture.assert_old() + self.assertIsNotNone(fixture.marker()) + finally: + fixture.doCleanups() + + def test_read_snapshot_cleanup_never_retries_uncertain_rollback(self) -> None: + self.stage() + for mode in ("before", "after", "retained"): + with self.subTest(mode=mode): + failure = sqlite3.OperationalError("synthetic read rollback failure") + self.conn.armed, self.conn.rollback_mode, self.conn.fault = True, mode, failure + before = self.conn.rollbacks + try: + with self.assertRaises((sqlite3.OperationalError, self.api.TierActivationError)) as caught: + self.api.read_activation_state(self.conn) + self.assertEqual(self.conn.rollbacks, before + 1) + self.assertEqual(self.conn.in_transaction, mode != "after") + if mode != "retained": + self.assertIs(caught.exception, failure) + finally: + self.conn.armed = False + self.conn.rollback() + self.assert_old() + + def test_oversized_and_nul_tail_records_reject_before_payload_read(self) -> None: + self.stage() + for value in ("x" * 20000, "{}\x00" + "x" * 20000, b"{}"): + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_activation_v1'", (value,)) + self.conn.commit() + before = self.conn.record_payload_reads + with self.assertRaises(self.api.TierActivationError): + self.api.read_activation_state(self.conn) + self.assertEqual(self.conn.record_payload_reads, before) + + def test_strict_json_fields_types_and_frozen_identity(self) -> None: + intent = self.stage() + valid = json.loads(self.api._dump(intent)) + malformed = ["[]", "{}", '{"version":1,"version":1}', self.api._dump(intent) + "\x00"] + for key, value in (("version", True), ("unexpected", 1), ("expected_generation", False), ("state", "complete")): + item = dict(valid) + item[key] = value + malformed.append(json.dumps(item)) + for field, value in (("dimension", True), ("model_id", "x" * 513), ("tier", "pro")): + item = dict(valid, target=dict(valid["target"], **{field: value})) + malformed.append(json.dumps(item)) + for raw in malformed: + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_activation_v1'", (raw,)) + self.conn.commit() + with self.subTest(raw=raw[:30]), self.assertRaises(self.api.TierActivationError): + self.api.read_activation_state(self.conn) + with self.assertRaises(dataclasses.FrozenInstanceError): + intent.reranker_id = "changed" + + def test_selected_record_counts_tables_hashes_and_ack_are_strict(self) -> None: + intent, plan = self.build() + selected = self.commit(intent, plan) + valid = json.loads(self.api._dump(selected)) + for field, value in (("vector_count", True), ("vector_count", -1), ("vector_count", 2**63), + ("cursor", None), ("cursor", 2**63), ("config_acknowledged", 1), + ("tables", list(reversed(TABLES))), ("pair_digest", "wrong"), + ("job_id", "0" * 32), ("config_acknowledged", True)): + self.conn.execute("UPDATE metadata SET value=? WHERE key='tier_selected_v1'", (json.dumps(dict(valid, **{field: value})),)) + self.conn.commit() + with self.subTest(field=field, value=value), self.assertRaises(self.api.TierActivationError): + self.api.read_activation_state(self.conn) + + def test_registry_default_commits_and_additive_mode_preserves_borrowed_transaction(self) -> None: + registry = self.modules["tier_switch.cache"].VectorCacheRegistry + before = self.conn.commits + registry.set(self.conn, "custom", model_name="synthetic/encoder", embedding_dim=2) + self.assertEqual(self.conn.commits, before + 1) + self.conn.execute("BEGIN") + registry.set(self.conn, "custom", model_name="synthetic/other", embedding_dim=3, commit=False) + self.assertTrue(self.conn.in_transaction) + self.conn.rollback() + self.assertEqual(self.conn.execute("SELECT model_name,embedding_dim FROM vector_cache_registry").fetchone(), ("synthetic/encoder", 2)) + + +@unittest.skipUnless(os.environ.get("TRUEMEMORY_TEST_NATIVE_VEC") == "1", "Native vec0 gate belongs to GPUBox") +class TestNativeActivationJournal(unittest.TestCase): + def test_actual_vec0_certification_and_job_retirement(self) -> None: + fixture = TestActivationJournal() + fixture.native_pair = True + fixture.setUp() + self.addCleanup(fixture.doCleanups) + fixture.test_owned_certification_exact_records_one_commit_and_job_retirement() + + +if __name__ == "__main__": + unittest.main() diff --git a/truememory/tier_switch/activation.py b/truememory/tier_switch/activation.py new file mode 100644 index 00000000..4b4c431a --- /dev/null +++ b/truememory/tier_switch/activation.py @@ -0,0 +1,383 @@ +"""Database-only certification of an explicitly selected, inactive tier pair. + +Callers own maintenance, coherent old serving state, inactive target admission, +and the drained local serving boundary. None is inferred from a database row. +This module never applies config/runtime state or certifies legacy vectors. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import sqlite3 +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import asdict, dataclass, replace + +from truememory.embedding_target import EmbeddingTarget +from truememory.tier_switch.cache import VectorCacheRegistry +from truememory.tier_switch.job import ( + TierJob, _retire_selected_tier_job_in_writer, check_selected_tier_job_in_writer, +) +from truememory.tier_switch.source import TierSourcePlan, require_current_completion_in_writer + +_SELECTED = "tier_selected_v1" +_INTENT = "tier_activation_v1" +_MAX_RECORD_BYTES = 16384 +_MAX_SCHEMA_BYTES = 65536 +_TRACKERS = {"canonical-v1", "rebuild-bridge-v1", "rebuild-bridge-conservative-v1"} +_METADATA_SQL = "CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT)" +_REGISTRY_SQL = """CREATE TABLE vector_cache_registry ( + tier_group TEXT PRIMARY KEY, vec_table TEXT NOT NULL, sep_table TEXT NOT NULL, + last_embedded_id INTEGER DEFAULT 0, vector_count INTEGER DEFAULT 0, + model_name TEXT, embedding_dim INTEGER DEFAULT 256, last_updated REAL, created REAL +)""" + + +class TierActivationError(RuntimeError): + """Selection cannot be certified; no runtime/config success is implied.""" + + +def _hex(value: object, size: int) -> bool: + return type(value) is str and re.fullmatch(r"[0-9a-f]{" + str(size) + "}", value) is not None + + +def _text(value: object, limit: int) -> bool: + if type(value) is not str or not 1 <= len(value) <= limit or "\x00" in value: + return False + try: + return len(value.encode("utf-8")) <= limit * 4 + except UnicodeError: + return False + + +def _identity(target: EmbeddingTarget, reranker_id: str, job_id: str, + tracker: str, source_epoch: str, source_schema_signature: str) -> None: + if (type(target) is not EmbeddingTarget or not _text(target.model_id, 512) + or any(type(value) is not str for value in (target.tier, target.tier_group)) + or not _text(reranker_id, 512) + or re.fullmatch(r"[\w][\w.\-]*(/[\w][\w.\-]*)?", reranker_id) is None + or not _hex(job_id, 32) or type(tracker) is not str or tracker not in _TRACKERS + or not _text(source_epoch, 128) or not _hex(source_schema_signature, 64)): + raise TierActivationError("Invalid frozen activation identity") + EmbeddingTarget.from_wire(target.to_wire()) + + +@dataclass(frozen=True) +class ActivationIntent: + intent_id: str + expected_generation: str | None + target: EmbeddingTarget + reranker_id: str + job_id: str + tracker: str + source_epoch: str + source_schema_signature: str + state: str = "staged" + + def __post_init__(self) -> None: + _identity(self.target, self.reranker_id, self.job_id, self.tracker, + self.source_epoch, self.source_schema_signature) + if (not _hex(self.intent_id, 32) + or (self.expected_generation is not None and not _hex(self.expected_generation, 32)) + or type(self.state) is not str or self.state not in {"staged", "db_selected", "config_acknowledged"}): + raise TierActivationError("Invalid activation intent") + + +@dataclass(frozen=True) +class TierSelection: + generation: str + intent_id: str + target: EmbeddingTarget + reranker_id: str + tables: tuple[str, str] + job_id: str + tracker: str + source_epoch: str + source_schema_signature: str + manifest_generation: str + manifest_hash: str + receipt_hash: str + pair_digest: str + schema: str + vector_count: int + cursor: int | None + config_acknowledged: bool = False + + def __post_init__(self) -> None: + _identity(self.target, self.reranker_id, self.job_id, self.tracker, + self.source_epoch, self.source_schema_signature) + if (any(not _hex(value, 32) for value in (self.generation, self.intent_id, self.manifest_generation)) + or any(not _hex(value, 64) for value in (self.manifest_hash, self.receipt_hash, self.pair_digest, self.schema)) + or type(self.tables) is not tuple or self.tables != self.target.tables + or type(self.vector_count) is not int or not 0 <= self.vector_count <= 2**63 - 1 + or (self.cursor is not None and (type(self.cursor) is not int or not -(2**63) <= self.cursor < 2**63)) + or ((self.vector_count == 0) != (self.cursor is None)) + or type(self.config_acknowledged) is not bool): + raise TierActivationError("Invalid certified selection") + + +@dataclass(frozen=True) +class ActivationState: + selection: TierSelection | None + intent: ActivationIntent | None + + +def _dump(record: ActivationIntent | TierSelection) -> str: + value = asdict(record) + value["target"] = record.target.to_wire() + raw = json.dumps(dict(value, version=1), sort_keys=True, separators=(",", ":"), allow_nan=False) + if len(raw.encode("utf-8")) > _MAX_RECORD_BYTES: + raise TierActivationError("Activation record exceeds protocol bounds") + return raw + + +def _unique(pairs: list[tuple[str, object]]) -> dict[str, object]: + value = dict(pairs) + if len(value) != len(pairs): + raise TierActivationError("Duplicate activation record field") + return value + + +def _parse(raw: str, kind: type[ActivationIntent] | type[TierSelection]) -> ActivationIntent | TierSelection: + try: + value = json.loads(raw, object_pairs_hook=_unique) + if (type(value) is not dict or set(value) != set(kind.__dataclass_fields__) | {"version"} + or type(value["version"]) is not int or value.pop("version") != 1): + raise TierActivationError("Unsupported activation record schema") + value["target"] = EmbeddingTarget.from_wire(value["target"]) + if kind is TierSelection: + if type(value["tables"]) is not list or len(value["tables"]) != 2: + raise TierActivationError("Invalid selected table pair") + value["tables"] = tuple(value["tables"]) + return kind(**value) + except (ValueError, TypeError, KeyError, OverflowError, RecursionError) as exc: + raise TierActivationError("Malformed activation record") from exc + + +def _normalized(sql: str) -> str: + return re.sub(r"\s*([(),])\s*", r"\1", " ".join(sql.strip().rstrip(";").split())).lower() + + +def _schema_sql(conn: sqlite3.Connection, name: str) -> str: + shape = conn.execute( + "SELECT type, typeof(sql), length(CAST(sql AS BLOB)) FROM main.sqlite_master " + "WHERE name=? COLLATE NOCASE", (name,), + ).fetchone() + if shape is None or shape[:2] != ("table", "text") or not 1 <= shape[2] <= _MAX_SCHEMA_BYTES: + raise TierActivationError("Initialized bounded table schema is required") + return conn.execute("SELECT sql FROM main.sqlite_master WHERE name=? COLLATE NOCASE", (name,)).fetchone()[0] + + +def _guard_storage(conn: sqlite3.Connection) -> None: + if conn.execute("SELECT 1 FROM temp.sqlite_master LIMIT 1").fetchone() is not None: + raise TierActivationError("TEMP objects make activation storage ambiguous") + if conn.execute("SELECT 1 FROM pragma_database_list WHERE name NOT IN ('main','temp') LIMIT 1").fetchone(): + raise TierActivationError("Activation requires a single MAIN database") + for name, expected in (("metadata", _METADATA_SQL), ("vector_cache_registry", _REGISTRY_SQL)): + if _normalized(_schema_sql(conn, name)) != _normalized(expected): + raise TierActivationError("Unsupported activation storage schema") + if conn.execute( + "SELECT 1 FROM main.sqlite_master WHERE tbl_name=? COLLATE NOCASE " + "AND (type='trigger' OR (type='index' AND sql IS NOT NULL)) LIMIT 1", (name,), + ).fetchone(): + raise TierActivationError("Activation storage has unsupported side effects") + + +def _guard_pair(conn: sqlite3.Connection, target: EmbeddingTarget) -> None: + for name in target.tables: + declaration = ( + r"CREATE\s+VIRTUAL\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?" + + r'(?:"' + name + r'"|`' + name + r'`|\[' + name + r'\]|' + name + r')' + + r"\s+USING\s+vec0\s*\(\s*embedding\s+float\[" + str(target.dimension) + + r"\]\s+distance_metric\s*=\s*cosine\s*\)\s*" + ) + if re.fullmatch(declaration, _schema_sql(conn, name), re.IGNORECASE) is None: + raise TierActivationError("Selected pair must have the exact prepared vec0 schema") + + +def _read(conn: sqlite3.Connection, key: str, kind: type[ActivationIntent] | type[TierSelection] + ) -> ActivationIntent | TierSelection | None: + shape = conn.execute( + "SELECT typeof(value), length(CAST(value AS BLOB)) FROM main.metadata WHERE key=?", (key,), + ).fetchone() + if shape is None: + return None + if shape[0] != "text" or not 1 <= shape[1] <= _MAX_RECORD_BYTES: + raise TierActivationError("Activation record exceeds its type or byte bound") + raw = conn.execute("SELECT value FROM main.metadata WHERE key=?", (key,)).fetchone()[0] + return _parse(raw, kind) + + +def _put(conn: sqlite3.Connection, key: str, value: str) -> None: + # Preserve the primary key: REPLACE could cascade an incoming foreign key. + conn.execute( + "INSERT INTO main.metadata(key,value) VALUES (?,?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value), + ) + + +@contextmanager +def _transaction(conn: sqlite3.Connection, *, write: bool) -> Iterator[None]: + if conn.in_transaction: + raise TierActivationError("Activation requires an owned clean connection") + conn.execute("BEGIN IMMEDIATE" if write else "BEGIN") + try: + yield + if write: + conn.commit() + if conn.in_transaction: + raise TierActivationError("Activation transaction did not commit") + except BaseException: + if conn.in_transaction: + conn.rollback() + if conn.in_transaction: + raise TierActivationError("Activation rollback is uncertain") + raise + else: + # Cleanup failure is terminal. Keeping this outside the handler above + # prevents a second rollback after an uncertain first attempt. + if not write: + conn.rollback() + if conn.in_transaction: + raise TierActivationError("Activation snapshot did not close") + + +def _state(conn: sqlite3.Connection) -> ActivationState: + selected, intent = _read(conn, _SELECTED, TierSelection), _read(conn, _INTENT, ActivationIntent) + if selected is not None and intent is None: + raise TierActivationError("Selected record has no activation intent") + if intent is not None and intent.state == "staged": + _expected(selected, intent.expected_generation) + if selected is not None and not selected.config_acknowledged: + raise TierActivationError("Pending config selection cannot have a new staged intent") + if intent is not None and intent.state != "staged": + if (selected is None or selected.intent_id != intent.intent_id + or selected.target != intent.target or selected.reranker_id != intent.reranker_id + or (selected.job_id, selected.tracker, selected.source_epoch, selected.source_schema_signature) + != (intent.job_id, intent.tracker, intent.source_epoch, intent.source_schema_signature) + or selected.config_acknowledged != (intent.state == "config_acknowledged")): + raise TierActivationError("Selected record and activation result disagree") + return ActivationState(selected, intent) + + +def read_activation_state(conn: sqlite3.Connection) -> ActivationState: + """Read committed journal state separately after reopening a safe connection. + + This is not config/runtime reconciliation, fresh vector recertification, or + permission to reuse a connection with uncertain rollback state. + """ + with _transaction(conn, write=False): + _guard_storage(conn) + return _state(conn) + + +def _expected(selection: TierSelection | None, generation: str | None) -> None: + if generation is not None and not _hex(generation, 32): + raise TierActivationError("Invalid expected selection generation") + if (selection.generation if selection else None) != generation: + raise TierActivationError("Selected generation changed") + + +def stage_activation_intent(conn: sqlite3.Connection, job: TierJob, *, + expected_generation: str | None, reranker_id: str) -> ActivationIntent: + """Commit intent before source planning; identical current intent is idempotent. + + None expects absent selection and permits only a later certified target. It + does not bootstrap/certify the legacy serving pair. A different job can + replace an abandoned staged intent only after selected-job admission has + replaced its marker. Conflicting intent for a live job is refused. + """ + intent = ActivationIntent(uuid.uuid4().hex, expected_generation, job.target, reranker_id, + job.job_id, job.tracker, job.source_epoch, job.source_schema_signature) + with _transaction(conn, write=True): + _guard_storage(conn) + check_selected_tier_job_in_writer(conn, job) + state = _state(conn) + _expected(state.selection, expected_generation) + if state.selection is not None and not state.selection.config_acknowledged: + raise TierActivationError("Current selection awaits config acknowledgement") + if state.intent is not None and state.intent.job_id == job.job_id: + if state.intent == replace(intent, intent_id=state.intent.intent_id): + return state.intent + raise TierActivationError("A conflicting intent already owns this live job") + _put(conn, _INTENT, _dump(intent)) + return intent + + +def commit_certified_selection(conn: sqlite3.Connection, intent: ActivationIntent, + job: TierJob, plan: TierSourcePlan) -> TierSelection: + """Certify and select atomically, then retire this exact successful marker. + + Caller must drain serving operations, prove an inactive target, and retain + maintenance/stable-path ownership. No native/config/runtime action occurs. + Commit/rollback exceptions propagate; use separate safe readback to learn + whether durable selection occurred. A returned result is still config-pending. + """ + if type(intent) is not ActivationIntent or intent.state != "staged": + raise TierActivationError("A staged immutable intent is required") + if (intent.target != job.target + or (intent.job_id, intent.tracker, intent.source_epoch, intent.source_schema_signature) + != (job.job_id, job.tracker, job.source_epoch, job.source_schema_signature)): + raise TierActivationError("Activation intent belongs to a different job") + with _transaction(conn, write=True): + _guard_storage(conn) + check_selected_tier_job_in_writer(conn, job) + state = _state(conn) + _expected(state.selection, intent.expected_generation) + if state.intent != intent: + raise TierActivationError("Activation intent changed") + if (plan.manifest.model != intent.target.model_id or plan.manifest.dimension != intent.target.dimension + or plan.manifest.targets != intent.target.tables + or plan.manifest.source is None + or (plan.manifest.source.tracker, plan.manifest.source.revision.epoch, + plan.manifest.source.schema_signature) + != (intent.tracker, intent.source_epoch, intent.source_schema_signature)): + raise TierActivationError("Paired certificate belongs to a different target or source") + _guard_pair(conn, intent.target) + require_current_completion_in_writer(conn, plan) + selection = TierSelection( + uuid.uuid4().hex, intent.intent_id, intent.target, intent.reranker_id, intent.target.tables, + intent.job_id, intent.tracker, intent.source_epoch, intent.source_schema_signature, + plan.manifest.generation, hashlib.sha256(plan.saved_manifest.encode()).hexdigest(), + hashlib.sha256(plan.saved_receipt.encode()).hexdigest(), plan.digest, plan.schema, + plan.prefix_count + plan.manifest.outputs, plan.manifest.cursor, + ) + # The marker helper rechecks the exact row and refuses all incoming FKs. + # Both registry/metadata writes preserve their primary keys and cannot + # invoke triggers under the exact schemas accepted above. + _retire_selected_tier_job_in_writer(conn, job) + VectorCacheRegistry.set( + conn, intent.target.tier_group, vec_table=selection.tables[0], sep_table=selection.tables[1], + last_embedded_id=selection.cursor if selection.cursor is not None else 0, + vector_count=selection.vector_count, model_name=intent.target.model_id, + embedding_dim=intent.target.dimension, commit=False, + ) + _put(conn, "embed_model", intent.target.model_id) + _put(conn, "embed_dim", str(intent.target.dimension)) + _put(conn, _SELECTED, _dump(selection)) + _put(conn, _INTENT, _dump(replace(intent, state="db_selected"))) + return selection + + +def acknowledge_config(conn: sqlite3.Connection, *, generation: str) -> TierSelection: + """Generation-CAS acknowledgement after caller-confirmed config mirroring. + + This performs no config write and acknowledges no process's runtime state. + """ + if not _hex(generation, 32): + raise TierActivationError("A concrete selected generation is required") + with _transaction(conn, write=True): + _guard_storage(conn) + state = _state(conn) + _expected(state.selection, generation) + if state.intent.state == "staged" or state.intent.intent_id != state.selection.intent_id: + raise TierActivationError("A newer activation intent is pending") + if state.selection.config_acknowledged: + return state.selection + selected = replace(state.selection, config_acknowledged=True) + _put(conn, _SELECTED, _dump(selected)) + _put(conn, _INTENT, _dump(replace(state.intent, state="config_acknowledged"))) + return selected diff --git a/truememory/tier_switch/cache.py b/truememory/tier_switch/cache.py index 0b0f6aa9..c3eee48a 100644 --- a/truememory/tier_switch/cache.py +++ b/truememory/tier_switch/cache.py @@ -75,21 +75,35 @@ def set( vector_count: int = 0, model_name: str | None = None, embedding_dim: int = 256, + commit: bool = True, ) -> None: - """Insert or replace a cache entry.""" + """Replace and commit by default; commit=False preserves the primary key. + + The additive mode uses UPSERT inside the caller's transaction so that + metadata publication cannot cause an incoming foreign-key delete cascade. + """ now = time.time() vt = vec_table or f"vec_messages_{group}" st = sep_table or f"vec_messages_sep_{group}" mn = model_name or model_name_for_group(group) + insertion = "INSERT OR REPLACE" if commit else "INSERT" + conflict = "" if commit else ( + " ON CONFLICT(tier_group) DO UPDATE SET " + "vec_table=excluded.vec_table, sep_table=excluded.sep_table, " + "last_embedded_id=excluded.last_embedded_id, vector_count=excluded.vector_count, " + "model_name=excluded.model_name, embedding_dim=excluded.embedding_dim, " + "last_updated=excluded.last_updated, created=excluded.created" + ) conn.execute( - "INSERT OR REPLACE INTO vector_cache_registry " + f"{insertion} INTO vector_cache_registry " "(tier_group, vec_table, sep_table, last_embedded_id, " "vector_count, model_name, embedding_dim, last_updated, created) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)" + conflict, (group, vt, st, last_embedded_id, vector_count, mn, embedding_dim, now, now), ) - conn.commit() + if commit: + conn.commit() @staticmethod def update_progress( diff --git a/truememory/tier_switch/job.py b/truememory/tier_switch/job.py index 5bdc95cd..ead335d5 100644 --- a/truememory/tier_switch/job.py +++ b/truememory/tier_switch/job.py @@ -346,6 +346,60 @@ def check_selected_tier_job(conn: sqlite3.Connection, job: TierJob) -> None: raise TierJobSelectionError("Selected background job could not be validated") from None +def check_selected_tier_job_in_writer(conn: sqlite3.Connection, job: TierJob) -> None: + """Read only inside the caller's owned BEGIN IMMEDIATE transaction. + + Python sqlite3 exposes whether a transaction exists, not its lock kind. + The caller must establish writer ownership; this helper neither acquires it + nor commits/rolls back caller work. Stable-path/maintenance ownership is + still required, just as for the public clean-connection check. + """ + if not conn.in_transaction: + raise TierJobSelectionError("Selected job certification requires caller writer ownership") + values = _selection_values(job) + shape = conn.execute( + "SELECT typeof(sql), length(CAST(sql AS BLOB)) FROM main.sqlite_master " + "WHERE name = ? COLLATE NOCASE", (_SELECTION_TABLE,), + ).fetchone() + if shape is None or shape[0] != "text" or not 1 <= shape[1] <= 8192: + raise TierJobSelectionError("Selected job schema exceeds its protocol bounds") + _selection_source(conn, job) + _selection_schema(conn) + if _selection_row(conn) != (*values, "selected"): + raise TierJobSelectionError("Background job is no longer selected") + _selection_source(conn, job) + + +def _retire_selected_tier_job_in_writer(conn: sqlite3.Connection, job: TierJob) -> None: + """Retire this exact successful marker only within certified selection SQL. + + Reject incoming foreign keys before DELETE, including when enforcement is + disabled on this connection. The bounded schema walk never copies table + definitions or unbounded names into Python. No commit or rollback occurs. + """ + check_selected_tier_job_in_writer(conn, job) + count = 0 + for (name,) in conn.execute( + "SELECT CASE WHEN typeof(name)='text' AND length(CAST(name AS BLOB)) BETWEEN 1 AND 512 " + "THEN name END FROM main.sqlite_master WHERE type='table' LIMIT 4097", + ): + count += 1 + if name is None or count > 4096: + raise TierJobSelectionError("Database schema exceeds safe marker retirement bounds") + if conn.execute( + 'SELECT 1 FROM pragma_foreign_key_list(?, \'main\') WHERE "table" = ? COLLATE NOCASE LIMIT 1', + (name, _SELECTION_TABLE), + ).fetchone() is not None: + raise TierJobSelectionError("Selected job marker has an incoming foreign key") + where = " AND ".join(f"{name} = ?" for name in _SELECTION_FIELDS) + changed = conn.execute( + f"DELETE FROM main.{_SELECTION_TABLE} WHERE singleton=1 AND state='selected' AND {where}", + _selection_values(job), + ).rowcount + if changed != 1: + raise TierJobSelectionError("Selected job changed before retirement") + + @contextmanager def open_selected_tier_job(job: TierJob) -> Iterator[sqlite3.Connection]: """Reopen under worker ownership and validate the committed selection.""" diff --git a/truememory/tier_switch/source.py b/truememory/tier_switch/source.py index 3e2d4c48..7afeefdc 100644 --- a/truememory/tier_switch/source.py +++ b/truememory/tier_switch/source.py @@ -462,3 +462,32 @@ def finish_tier_source(conn: sqlite3.Connection, plan: TierSourcePlan) -> TierSo and current.high_id == plan.manifest.cursor) updated = _persist(conn, replace(plan, action="complete", manifest=replace(plan.manifest, complete=True))) return TierSourceCompletion(updated, True, covers_current) + + +def require_current_completion_in_writer(conn: sqlite3.Connection, plan: TierSourcePlan) -> None: + """Read-only final certification inside an owned BEGIN IMMEDIATE. + + sqlite3 cannot expose transaction lock kind: the caller must hold the writer + and cooperative pair ownership. Unlike finish, this never persists or ends + a transaction. One bounded-memory full pair audit occurs at activation, + not after every page. Strict native table/model validation remains caller + responsibility; the adapter attests stored bytes and source coverage. + """ + if not conn.in_transaction: + raise TierSourceUntrusted("Current completion requires caller writer ownership") + if (plan.action != "complete" or not plan.manifest.complete + or plan.manifest.consumed != plan.manifest.total + or plan.saved_manifest is None or plan.saved_receipt is None): + raise TierSourceUntrusted("A finished paired generation is required") + _check(conn, plan) + parsed = _parse_manifest(plan.saved_manifest, plan.manifest.model, + plan.manifest.dimension, plan.manifest.targets) + if parsed != plan.manifest or _parse_receipt(plan.saved_receipt, parsed, plan.schema) != ( + plan.digest, plan.prefix_count, plan.prefix_cursor, + ): + raise TierSourceUntrusted("Completion does not match its persisted paired certificate") + current = capture_source(conn) + if (current.total != plan.prefix_count + plan.manifest.outputs + or current.high_id != plan.manifest.cursor): + raise TierSourceUntrusted("Captured pair does not cover the current source") + _audit_pair(conn, plan)