diff --git a/CHANGELOG.md b/CHANGELOG.md index 7775417b..6ab19568 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,25 @@ adheres to [Semantic Versioning](https://semver.org/). ### Fixed +- **SQLite requests no longer share transaction ownership across concurrent + MCP workers (HC-CORTEX-002).** The fallback store previously exposed one + process-wide connection with `check_same_thread=False`, so an acknowledged + request could commit unfinished writes from a rejected request. Each + active execution thread now owns the native connection it uses, while the + handler boundary tracks exact handles propagated through + `asyncio.to_thread`, rolls back + unfinished request work, rejects false success, and quarantines a connection + when rollback fails. An unrecoverable rollback failure on the in-memory + lifetime anchor now invalidates the registry instead of silently exposing a + fresh empty database. Non-anchor request handles are released after + finalization, and relative database paths are frozen at construction so + worker opens cannot drift after a process `cwd` change. Nested request scopes + fail closed. Deterministic regressions cover the original supersede/insert + store-path interference, reused workers, nested offload, unnamed handlers, + cleanup failure, close/reopen persistence, FTS, vectors, foreign keys, and + integrity checks. The + fresh-process, cross-backend load ladder is tracked separately and is not + claimed by this correction. - `tests_py/invariants/test_pg_schema_provision_live.py` minted its throwaway low-privilege PostgreSQL role with a literal password. It now uses `secrets.token_urlsafe(16)` per run — the role never outlives the diff --git a/docs/adr/ADR-0055-sqlite-connection-per-execution-thread.md b/docs/adr/ADR-0055-sqlite-connection-per-execution-thread.md new file mode 100644 index 00000000..fd2afd9e --- /dev/null +++ b/docs/adr/ADR-0055-sqlite-connection-per-execution-thread.md @@ -0,0 +1,230 @@ +# ADR-0055: Request-Scoped SQLite Connections With Thread Ownership + +**Status:** Proposed +**Date:** 2026-08-30 +**Decision-makers:** cdeust +**Related:** HC-CORTEX-002; ADR-0045 R6; I10 + +## Context + +`SqliteMemoryStore` owns one `sqlite3.Connection` configured with +`check_same_thread=False`. `safe_handler` executes concurrent MCP calls on +worker threads, so unrelated operations share one implicit transaction. + +The HC-CORTEX-002 baseline at Cortex revision +`8f5ae3b87b6969f3abcb3736859febfdab69304a` injected a failure after a +supersession insert and compare-and-set. A concurrent acknowledged insert +committed the shared connection before the rejected supersession rolled back. +The rejected memory, its FTS row, and its `superseded_by_id` edge remained, +while `PRAGMA integrity_check` still returned `ok`. + +A second RED fixture rejected an `insert_memory` after its first write, then +ran an unrelated acknowledged request on the same reused worker. A connection +per thread alone still let the second request commit the first request's dirty +transaction. Thread ownership is therefore necessary but not sufficient: the +request boundary must finalize unfinished work before a worker is reused. + +This is the documented SQLite boundary, not a damaged database file: + +- Python's `sqlite3` documentation says that disabling `check_same_thread` + permits cross-thread access but requires the caller to serialize writes to + avoid corruption: . +- SQLite provides isolation between separate connections and explicitly no + isolation between operations on the same connection: + . +- Python exposes `Connection.in_transaction` as the read-only view of SQLite's + low-level autocommit state, and documents `rollback()` as reverting a pending + transaction: . +- WAL permits simultaneous readers and a single serialized writer across + separate connections: . + +## Decision + +The SQLite adapter and handler boundary will enforce both ownership levels: + +1. Each execution thread lazily receives one native connection to the same + database file. The on-disk path is resolved once at registry construction, + so a later process `cwd` change cannot redirect worker handles. +2. Every `safe_handler` path is offloaded through `asyncio.to_thread`; an + optional tool name controls admission and named metrics, not execution or + isolation. The request installs a thread-safe `ContextVar` state that + records the exact registry/connection identities touched by the handler + and any nested `to_thread` calls. Python documents that `to_thread` copies + the current context into its worker: + . +3. On an exception, the boundary rolls back every pending connection recorded + for that request. If a handler tries to return success with pending work, + the boundary rolls it back and raises + `UncommittedSqliteTransactionError`; it cannot emit a false + acknowledgement. A rollback failure quarantines and closes that native + connection so later work cannot commit its unknown state. The original + handler exception remains the reported failure. + After finalization, every non-anchor request-opened handle is closed and + removed from the registry. While healthy, one anchor remains until + store shutdown so a named in-memory database cannot disappear between + requests. If rollback itself fails on that last in-memory keeper, the + registry becomes explicitly unusable; it never silently reopens an empty + database as recovered state. +4. Nested handler transaction scopes are rejected before the inner handler + starts. This change does not implement savepoints and therefore does not + pretend that an inner request can independently commit or roll back. +5. The primary on-disk connection requests WAL once and records a warning if + SQLite retains another mode; the benchmark records the observed mode. + Every connection enables foreign keys and loads + `sqlite-vec` when vector support was enabled by the primary connection. + SQLite documents both WAL persistence and the in-memory `MEMORY`/`OFF` + restriction: . +6. The existing `_raw_conn` and psycopg-compatible `_conn` surfaces remain + stable proxies, so the existing store/mixin contract does not gain backend + or thread branching. +7. Store shutdown, after requests are quiescent, closes every connection + registered by the process. `check_same_thread=False` remains intentional + only so that shutdown can close worker-owned handles centrally; normal SQL + operations are routed to their owner by the proxy. +8. Direct `SqliteMemoryStore(":memory:")` fixtures use a uniquely named + in-memory URI so worker connections observe the same ephemeral database. + They use SQLite's `MEMORY` journal mode because WAL is unavailable for an + in-memory database. This is the only shared-cache use. It is test-only; + production/plugin construction supplies the configured on-disk fallback + path. SQLite discourages shared cache for production and recommends WAL: + . + +Transaction ownership follows the connection that performed the operation, +and request finalization prevents a dirty transaction from crossing a worker +reuse boundary. A commit or rollback from another request cannot affect it. + +## Options Considered + +### A. Global handler semaphore + +| Dimension | Assessment | +|---|---| +| Complexity | Low | +| Integrity | Incomplete | +| Scalability | Poor | + +Serializing every SQLite-backed MCP handler would close the demonstrated tool +race, but direct store callers and any internal concurrency would remain +unsafe. It would also serialize read-side ranking work that does not need the +writer lock. Rejected because the invariant belongs at the storage boundary. + +### B. Transaction-aware lock around the shared connection + +| Dimension | Assessment | +|---|---| +| Complexity | Medium | +| Integrity | Sensitive to every error path | +| Scalability | One connection | + +A re-entrant lock could be retained from the first implicit write until +`commit` or `rollback`. The lock would have to infer transaction ownership, +survive cursor and raw-connection paths, and release correctly after every +exception. SQLite would still define same-connection read/write interleaving +as non-isolated. Rejected as a second transaction manager layered over +SQLite's own. + +### C. Request-scoped worker connections plus one lifetime anchor + +| Dimension | Assessment | +|---|---| +| Complexity | Medium | +| Integrity | Native SQLite isolation | +| Scalability | Concurrent readers; one SQLite writer | + +Accepted. Separate connections use SQLite's documented isolation unit, while +the handler scope closes the reused-worker gap demonstrated by the second RED +fixture and releases its non-anchor handles after finalization. The lifetime +anchor preserves the named in-memory fixture. Existing multi-statement methods +keep one stable connection for their complete call. + +### D. Autocommit plus explicit unit-of-work contexts everywhere + +| Dimension | Assessment | +|---|---| +| Complexity | High | +| Integrity | Explicit | +| Migration risk | High | + +This would require auditing and rewriting every multi-statement method across +the store and shared mixins. It may be useful for a later typed unit-of-work +API, but it is not the smallest correction for the proven ownership defect. + +## Consequences + +- Unfinished work on a connection recorded by the request scope cannot be + committed or rolled back by another worker. +- A failed request cannot leave recorded, uncommitted writes for a later + request to commit, and a successful response cannot acknowledge a recorded, + unfinished transaction. +- When the file is actually in WAL mode, WAL can serve readers while SQLite + serializes the sole writer. The requested and observed modes are distinct + benchmark evidence. +- The first SQLite call in a request pays one connection-open cost; later calls + on that request's worker reuse the handle. Non-anchor request handles are + then released, so inactive per-request executors do not accumulate + connections. +- Concurrent writers may queue or return SQLite's documented busy result. + The HC-CORTEX-002 load ladder must publish error/retry and saturation data; + this ADR does not invent a throughput threshold. +- On the current `safe_handler` host path, retained connection count returns to + the construction-time anchor after requests become quiescent; peak handles + follow the active request workers. Direct applications outside a request + scope that create arbitrary ephemeral threads retain those handles until + `SqliteMemoryStore.close()`; reclamation outside the host path is not claimed. +- Shutdown assumes the store is quiescent; closing a handle while a request is + using it remains outside the lifecycle contract. +- A rollback failure on the in-memory anchor is unrecoverable without a durable + backing file. The triggering handler keeps its original error, and every + later registry access fails explicitly until the store is closed. +- The request boundary does not undo writes that a store method has already + committed before a later method in the same handler fails. Handler-wide + atomic units of work require a separate explicit transaction API and are + outside this correction. HC-CORTEX-002 claims only that unfinished work from + a rejected request cannot cross a request/worker boundary or be finalized by + an unrelated request. +- Cancelling the coroutine awaiting `asyncio.to_thread` does not stop the + native worker. A cancelled or transport-lost call is therefore + `indeterminate`, not a rejected operation: the benchmark must wait for + worker quiescence and reconcile its operation identifier against storage + before assigning an outcome. This ADR does not claim cancellable + handler-wide atomicity. +- `ContextVar` propagation is guaranteed for `asyncio.to_thread`, not for an + arbitrary manually-created `threading.Thread`, and a child task that outlives + its handler is outside the quiescent request contract. +- The in-memory compatibility path uses SQLite's documented named-memory + shared-cache mechanism only because separate `:memory:` opens otherwise + create different databases: . + +## Verification + +- Preserve the deterministic rejected-supersede/concurrent-insert ledger as a + regression: rejected row/FTS/edge count is zero; acknowledged insert count is + one. +- Preserve the failed-request/acknowledged-next-request fixture on one reused + worker, plus a fixture proving that apparent success with pending work is + rejected and rolled back. +- Prove nested `to_thread` work is finalized through its exact native handle, + nested handler scopes fail before inner work, and concurrent unnamed + `safe_handler` calls do not share the event-loop thread's connection. +- Force rollback cleanup to fail and prove the connection is quarantined, the + original handler error is preserved, and a later acknowledged request opens + a clean on-disk replacement. Prove the equivalent in-memory anchor failure + invalidates the registry instead of exposing a fresh empty database. +- Verify the same fixture after closing and reopening the database. +- Verify worker handles cannot be reused after registry close, future worker + handles receive `sqlite-vec`, and observed journal modes are `wal` for the + file fixture and `memory` for the in-memory fixture. +- Repeat direct and nested-worker requests and prove registry cardinality + returns to its pre-request value after each one. +- Construct from a relative on-disk path, change process `cwd`, and prove a + later worker still opens the original database and creates no second file. +- Run `PRAGMA integrity_check` and `PRAGMA foreign_key_check`. +- Prove two worker threads receive isolated transaction outcomes on both an + on-disk fixture and the in-memory compatibility path. +- Run the preregistered concurrency ladder twice and publish throughput, + latency percentiles, queueing/busy errors, retries, resources, and recovery. +- Do not count a client cancellation or transport timeout as a rejection. + Record it as `indeterminate`, wait for worker quiescence, and reconcile it + before closing the store or publishing the cell. +- Run the matched PostgreSQL reference cell; PostgreSQL storage code is + unchanged. diff --git a/mcp_server/handlers/request_transaction.py b/mcp_server/handlers/request_transaction.py new file mode 100644 index 00000000..1ac1b8d5 --- /dev/null +++ b/mcp_server/handlers/request_transaction.py @@ -0,0 +1,17 @@ +"""Backend transaction finalization at the MCP handler boundary.""" + +from __future__ import annotations + +from collections.abc import Iterator +from contextlib import contextmanager + +from mcp_server.infrastructure.sqlite_request_scope import ( + sqlite_request_scope, +) + + +@contextmanager +def handler_transaction_scope() -> Iterator[None]: + """Reject or roll back unfinished SQLite work at the handler boundary.""" + with sqlite_request_scope(): + yield diff --git a/mcp_server/infrastructure/sqlite_compat.py b/mcp_server/infrastructure/sqlite_compat.py index 9c7fb935..177a461e 100644 --- a/mcp_server/infrastructure/sqlite_compat.py +++ b/mcp_server/infrastructure/sqlite_compat.py @@ -21,7 +21,7 @@ import sqlite3 from datetime import datetime -from typing import Any +from typing import Any, Protocol from mcp_server.infrastructure.sqlite_sql_translate import ( _returning_was_stripped, @@ -65,6 +65,28 @@ def _adapt_datetime_iso(value: datetime) -> str: sqlite3.register_adapter(datetime, _adapt_datetime_iso) +class SqliteConnectionLike(Protocol): + """Native or thread-local connection surface used by the SQL adapter.""" + + row_factory: Any + + def execute(self, sql: str, parameters: Any = (), /) -> sqlite3.Cursor: ... + + def executemany(self, sql: str, parameters: Any, /) -> sqlite3.Cursor: ... + + def cursor(self, /) -> sqlite3.Cursor: ... + + def executescript(self, sql: str, /) -> Any: ... + + def commit(self) -> None: ... + + def rollback(self) -> None: ... + + def close(self) -> None: ... + + def enable_load_extension(self, enabled: bool) -> None: ... + + class _CompatCursor: """Wraps a sqlite3.Cursor to mimic psycopg result access.""" @@ -169,7 +191,7 @@ class PsycopgCompatConnection: will work transparently with this wrapper. """ - def __init__(self, conn: sqlite3.Connection) -> None: + def __init__(self, conn: SqliteConnectionLike) -> None: self._real = conn def execute( diff --git a/mcp_server/infrastructure/sqlite_connection_registry.py b/mcp_server/infrastructure/sqlite_connection_registry.py new file mode 100644 index 00000000..63a4fb34 --- /dev/null +++ b/mcp_server/infrastructure/sqlite_connection_registry.py @@ -0,0 +1,216 @@ +"""Thread-confined SQLite connections behind one stable store facade. + +SQLite defines transaction isolation at the connection boundary. The MCP +server executes synchronous handlers on worker threads, so each execution +thread must own the connection whose commit or rollback ends its transaction. +The handler scope then rolls back unfinished work before a worker is reused. +""" + +from __future__ import annotations + +import logging +import sqlite3 +import threading +import uuid +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from mcp_server.infrastructure.sqlite_request_scope import ( + register_request_connection, +) + +_VectorLoader = Callable[[sqlite3.Connection], None] +logger = logging.getLogger(__name__) + + +class SqliteConnectionRegistry: + """Own request-scoped thread handles plus one store-lifetime anchor.""" + + def __init__(self, path: str) -> None: + self._target, self._uri = _connection_target(path) + self._local = threading.local() + self._lock = threading.Lock() + self._connections: list[sqlite3.Connection] = [] + self._anchor_connection: sqlite3.Connection | None = None + self._vector_loader: _VectorLoader | None = None + self._invalid_reason: str | None = None + self._closed = False + self._anchor_connection = self._open_locked() + self._local.connection = self._anchor_connection + + def connection(self) -> sqlite3.Connection: + """Return the connection owned by the calling execution thread.""" + with self._lock: + if self._closed: + raise sqlite3.ProgrammingError("SQLite connection registry is closed") + if self._invalid_reason is not None: + raise sqlite3.ProgrammingError(self._invalid_reason) + connection = getattr(self._local, "connection", None) + if connection not in self._connections: + connection = self._open_locked() + if self._anchor_connection is None: + self._anchor_connection = connection + self._local.connection = connection + register_request_connection(self, connection) + return connection + + def enable_vector_extension(self, loader: _VectorLoader) -> None: + """Load an optional extension on present and future connections.""" + with self._lock: + for connection in self._connections: + _load_extension(connection, loader) + self._vector_loader = loader + + def close(self) -> None: + """Close every worker connection after the store becomes quiescent.""" + with self._lock: + if self._closed: + return + self._closed = True + connections = tuple(self._connections) + self._connections.clear() + self._anchor_connection = None + self._local.connection = None + for connection in connections: + connection.close() + + def rollback_request_connection(self, connection: sqlite3.Connection) -> bool: + """Roll back one exact request handle; quarantine it on failure.""" + with self._lock: + if connection not in self._connections: + return False + if not connection.in_transaction: + return False + try: + _rollback_native(connection) + except Exception: + self._discard_connection(connection) + raise + return True + + def release_request_connection(self, connection: sqlite3.Connection) -> bool: + """Close a request-owned handle unless it anchors the store lifetime.""" + return self._discard_connection(connection, preserve_anchor=True) + + def _discard_connection( + self, + connection: sqlite3.Connection, + *, + preserve_anchor: bool = False, + ) -> bool: + with self._lock: + if preserve_anchor and connection is self._anchor_connection: + return False + if connection not in self._connections: + return False + self._connections.remove(connection) + if connection is self._anchor_connection: + self._anchor_connection = None + if self._uri: + self._invalid_reason = ( + "SQLite in-memory registry invalidated because its " + "anchor rollback failed" + ) + if getattr(self._local, "connection", None) is connection: + self._local.connection = None + try: + connection.close() + except sqlite3.Error: + logger.exception("SQLite connection release or quarantine close failed") + return True + + def _open_locked(self) -> sqlite3.Connection: + connection = sqlite3.connect( + self._target, + uri=self._uri, + check_same_thread=False, + detect_types=sqlite3.PARSE_DECLTYPES, + ) + try: + expected_journal_mode = "memory" if self._uri else "wal" + _configure( + connection, + expected_journal_mode=( + expected_journal_mode if not self._connections else None + ), + ) + if self._vector_loader is not None: + _load_extension(connection, self._vector_loader) + except BaseException: + connection.close() + raise + self._connections.append(connection) + return connection + + +class ThreadLocalSqliteConnection: + """Stable connection-shaped proxy resolving the caller's native handle.""" + + def __init__(self, registry: SqliteConnectionRegistry) -> None: + self._registry = registry + + def execute(self, sql: str, parameters: Any = ()) -> sqlite3.Cursor: + return self._registry.connection().execute(sql, parameters) + + def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor: + return self._registry.connection().executemany(sql, parameters) + + def cursor(self) -> sqlite3.Cursor: + return self._registry.connection().cursor() + + def executescript(self, sql: str) -> sqlite3.Cursor: + return self._registry.connection().executescript(sql) + + def commit(self) -> None: + self._registry.connection().commit() + + def rollback(self) -> None: + self._registry.connection().rollback() + + def close(self) -> None: + self._registry.close() + + @property + def row_factory(self) -> Any: + return self._registry.connection().row_factory + + @row_factory.setter + def row_factory(self, value: Any) -> None: + self._registry.connection().row_factory = value + + def enable_load_extension(self, enabled: bool) -> None: + self._registry.connection().enable_load_extension(enabled) + + +def _connection_target(path: str) -> tuple[str, bool]: + if path != ":memory:": + return str(Path(path).resolve()), False + name = f"cortex-{uuid.uuid4().hex}" + return f"file:{name}?mode=memory&cache=shared", True + + +def _configure( + connection: sqlite3.Connection, *, expected_journal_mode: str | None +) -> None: + connection.row_factory = sqlite3.Row + if expected_journal_mode is not None: + row = connection.execute("PRAGMA journal_mode=WAL").fetchone() + actual_mode = str(row[0]).lower() + if actual_mode != expected_journal_mode: + logger.warning( + "SQLite requested WAL but retained journal_mode=%s", actual_mode + ) + connection.execute("PRAGMA foreign_keys=ON") + + +def _load_extension(connection: sqlite3.Connection, loader: _VectorLoader) -> None: + connection.enable_load_extension(True) + try: + loader(connection) + finally: + connection.enable_load_extension(False) + + +def _rollback_native(connection: sqlite3.Connection) -> None: + connection.rollback() diff --git a/mcp_server/infrastructure/sqlite_request_scope.py b/mcp_server/infrastructure/sqlite_request_scope.py new file mode 100644 index 00000000..2bd770db --- /dev/null +++ b/mcp_server/infrastructure/sqlite_request_scope.py @@ -0,0 +1,120 @@ +"""Request-scoped finalization for exact native SQLite handles.""" + +from __future__ import annotations + +import logging +import sqlite3 +import threading +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Protocol + +logger = logging.getLogger(__name__) + + +class UncommittedSqliteTransactionError(RuntimeError): + """A successful request tried to return with uncommitted SQLite work.""" + + +class NestedSqliteRequestScopeError(RuntimeError): + """A handler attempted to start a second request transaction boundary.""" + + +class SqliteRequestRegistry(Protocol): + """Registry operations needed to finalize one request's native handles.""" + + def rollback_request_connection(self, connection: sqlite3.Connection) -> bool: ... + + def release_request_connection(self, connection: sqlite3.Connection) -> bool: ... + + +_RequestEntry = tuple[SqliteRequestRegistry, sqlite3.Connection] + + +class _RequestConnections: + """Thread-safe identity set shared through copied asyncio contexts.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._entries: dict[int, _RequestEntry] = {} + + def add( + self, registry: SqliteRequestRegistry, connection: sqlite3.Connection + ) -> None: + with self._lock: + self._entries.setdefault(id(connection), (registry, connection)) + + def snapshot(self) -> tuple[_RequestEntry, ...]: + with self._lock: + return tuple(self._entries.values()) + + +_request_connections: ContextVar[_RequestConnections | None] = ContextVar( + "sqlite_request_connections", default=None +) + + +@contextmanager +def sqlite_request_scope() -> Iterator[None]: + """Finalize every SQLite transaction touched by one handler request.""" + existing = _request_connections.get() + if existing is not None: + raise NestedSqliteRequestScopeError( + "nested handler transaction scopes are unsupported" + ) + connections = _RequestConnections() + token = _request_connections.set(connections) + try: + yield + except BaseException: + _finalize_request_connections(connections, preserve_original=True) + raise + else: + dirty = _finalize_request_connections(connections, preserve_original=False) + if dirty: + raise UncommittedSqliteTransactionError( + f"request left {dirty} uncommitted SQLite transaction(s)" + ) + finally: + _request_connections.reset(token) + + +def register_request_connection( + registry: SqliteRequestRegistry, connection: sqlite3.Connection +) -> None: + """Record the exact handle used in the propagated request context.""" + connections = _request_connections.get() + if connections is not None: + connections.add(registry, connection) + + +def _rollback_connections( + connections: tuple[_RequestEntry, ...], + *, + preserve_original: bool, +) -> int: + dirty = 0 + first_error: Exception | None = None + for registry, connection in connections: + try: + dirty += int(registry.rollback_request_connection(connection)) + except Exception as exc: + first_error = first_error or exc + logger.exception("SQLite request rollback failed at handler boundary") + if first_error is not None and not preserve_original: + raise first_error + return dirty + + +def _finalize_request_connections( + connections: _RequestConnections, + *, + preserve_original: bool, +) -> int: + snapshot = connections.snapshot() + try: + return _rollback_connections(snapshot, preserve_original=preserve_original) + finally: + for registry, connection in snapshot: + registry.release_request_connection(connection) diff --git a/mcp_server/infrastructure/sqlite_store.py b/mcp_server/infrastructure/sqlite_store.py index 202ce22d..8088bed6 100644 --- a/mcp_server/infrastructure/sqlite_store.py +++ b/mcp_server/infrastructure/sqlite_store.py @@ -25,7 +25,14 @@ import numpy as np from mcp_server.shared.temporal_normalize import normalize_date_to_iso -from mcp_server.infrastructure.sqlite_compat import PsycopgCompatConnection +from mcp_server.infrastructure.sqlite_compat import ( + PsycopgCompatConnection, + SqliteConnectionLike, +) +from mcp_server.infrastructure.sqlite_connection_registry import ( + SqliteConnectionRegistry, + ThreadLocalSqliteConnection, +) from mcp_server.infrastructure.sqlite_schema import ( COLUMN_BACKFILLS, CURRENT_MEMORIES_VIEW_DDL, @@ -119,6 +126,8 @@ class SqliteMemoryStore( ): """SQLite + FTS5 + sqlite-vec storage engine for Cortex memory system.""" + _raw_conn: SqliteConnectionLike + def __init__(self, db_path: str = "", embedding_dim: int = 384) -> None: self._embedding_dim = embedding_dim self._has_vec = False @@ -126,16 +135,9 @@ def __init__(self, db_path: str = "", embedding_dim: int = 384) -> None: if path != ":memory:": Path(path).parent.mkdir(parents=True, exist_ok=True) _register_json_codec() - raw = sqlite3.connect( - path, - check_same_thread=False, - detect_types=sqlite3.PARSE_DECLTYPES, - ) - raw.row_factory = sqlite3.Row - raw.execute("PRAGMA journal_mode=WAL") - raw.execute("PRAGMA foreign_keys=ON") - self._raw_conn = raw - self._conn = PsycopgCompatConnection(raw) + self._connection_registry = SqliteConnectionRegistry(path) + self._raw_conn = ThreadLocalSqliteConnection(self._connection_registry) + self._conn = PsycopgCompatConnection(self._raw_conn) self._init_schema() def _init_schema(self) -> None: @@ -317,9 +319,7 @@ def _try_load_vec(self) -> None: try: import sqlite_vec # noqa: PLC0415, F401 — optional dependency ([sqlite] extra); imported where used so environments without it keep working - self._raw_conn.enable_load_extension(True) - sqlite_vec.load(self._raw_conn) - self._raw_conn.enable_load_extension(False) + self._connection_registry.enable_vector_extension(sqlite_vec.load) self._conn.execute(MEMORIES_VEC_DDL) self._conn.commit() self._has_vec = True @@ -494,12 +494,12 @@ def supersede_atomic( """Insert ``data`` as the supersessor of ``target_id``'s head, atomically. SQLite parity for PgMemoryStore.supersede_atomic. One transaction on the - single connection inserts the new row (supersedes_id = the walked head) - and stamps that head's ``superseded_by_id`` — a compare-and-set that - lands only while the head is still open. On a lost CAS the transaction - rolls back (the insert, its FTS and vec rows all undone — no orphan is - ever committed) and we rebase onto the moved head, bounded by - _SUPERSEDE_REBASE_ATTEMPTS. + current execution thread's connection inserts the new row + (supersedes_id = the walked head) and stamps that head's + ``superseded_by_id`` — a compare-and-set that lands only while the head + is still open. On a lost CAS the transaction rolls back (the insert, + its FTS and vec rows all undone — no orphan is ever committed) and we + rebase onto the moved head, bounded by _SUPERSEDE_REBASE_ATTEMPTS. Returns ``(new_id, head_id)`` on success (``head_id`` == ``target_id`` unless a race rebased us), ``(None, last_head_id)`` when the bounded @@ -720,12 +720,10 @@ def update_memory_extinction( # ── Connection acquisition (PgMemoryStore parity) ───────────────── # # PgMemoryStore splits connections across an interactive pool and a batch - # pool so long-running jobs cannot starve the hot path. SQLite has no - # such split to make: the store owns exactly one WAL-mode connection, and - # a second competing connection is what produces `database is locked` / - # stale-read failures under WAL. Both accessors therefore yield the same - # persistent connection — the identical shape PgMemoryStore itself yields - # when POOL_DISABLED is set (pg_store.py acquire_* kill-switch path). + # pool so long-running jobs cannot starve the hot path. SQLite instead + # binds each transaction to its execution thread's persistent connection; + # both accessors expose the same stable facade, which resolves that native + # connection at every call. # # These exist so handlers stay backend-agnostic: anchor.py, get_rules.py, # the codebase_analyze/backfill/consolidation writers all call diff --git a/mcp_server/infrastructure/sqlite_store_entities.py b/mcp_server/infrastructure/sqlite_store_entities.py index 74882e3d..7f1a67ac 100644 --- a/mcp_server/infrastructure/sqlite_store_entities.py +++ b/mcp_server/infrastructure/sqlite_store_entities.py @@ -2,10 +2,12 @@ from __future__ import annotations -import sqlite3 - from typing import Any -from mcp_server.infrastructure.sqlite_compat import PsycopgCompatConnection + +from mcp_server.infrastructure.sqlite_compat import ( + PsycopgCompatConnection, + SqliteConnectionLike, +) from mcp_server.shared.code_tokenize import expand_fts_query @@ -13,7 +15,7 @@ class SqliteEntityMixin: """Entity persistence operations on SQLite.""" _conn: PsycopgCompatConnection - _raw_conn: sqlite3.Connection + _raw_conn: SqliteConnectionLike def _normalize_memory_row(self, row: dict) -> dict: """Provided by SqliteMemoryStore.""" diff --git a/mcp_server/infrastructure/sqlite_store_receipts.py b/mcp_server/infrastructure/sqlite_store_receipts.py index 19884cc7..678639bf 100644 --- a/mcp_server/infrastructure/sqlite_store_receipts.py +++ b/mcp_server/infrastructure/sqlite_store_receipts.py @@ -12,14 +12,17 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from mcp_server.infrastructure.sqlite_compat import PsycopgCompatConnection + from mcp_server.infrastructure.sqlite_compat import ( + PsycopgCompatConnection, + SqliteConnectionLike, + ) class SqliteReceiptsMixin: """Append-only injection receipts (blame path T1).""" _conn: PsycopgCompatConnection - _raw_conn: sqlite3.Connection + _raw_conn: SqliteConnectionLike def insert_injection_receipt( self, diff --git a/mcp_server/infrastructure/sqlite_store_relationships.py b/mcp_server/infrastructure/sqlite_store_relationships.py index a7019a40..990a8ace 100644 --- a/mcp_server/infrastructure/sqlite_store_relationships.py +++ b/mcp_server/infrastructure/sqlite_store_relationships.py @@ -2,17 +2,19 @@ from __future__ import annotations -import sqlite3 - -from mcp_server.infrastructure.sqlite_compat import PsycopgCompatConnection from typing import Any +from mcp_server.infrastructure.sqlite_compat import ( + PsycopgCompatConnection, + SqliteConnectionLike, +) + class SqliteRelationshipMixin: """Relationship persistence operations on SQLite.""" _conn: PsycopgCompatConnection - _raw_conn: sqlite3.Connection + _raw_conn: SqliteConnectionLike def update_relationships_weight_batch( self, updates: list[tuple[int, float]] diff --git a/mcp_server/infrastructure/sqlite_store_stats.py b/mcp_server/infrastructure/sqlite_store_stats.py index 76a99df0..457cc53d 100644 --- a/mcp_server/infrastructure/sqlite_store_stats.py +++ b/mcp_server/infrastructure/sqlite_store_stats.py @@ -3,16 +3,20 @@ from __future__ import annotations import sqlite3 -from mcp_server.infrastructure.sqlite_compat import PsycopgCompatConnection -from typing import Any from datetime import datetime, timezone +from typing import Any + +from mcp_server.infrastructure.sqlite_compat import ( + PsycopgCompatConnection, + SqliteConnectionLike, +) class SqliteStatsMixin: """Diagnostics, consolidation stages, CLS queries on SQLite.""" _conn: PsycopgCompatConnection - _raw_conn: sqlite3.Connection + _raw_conn: SqliteConnectionLike def _normalize_memory_row(self, row: dict) -> dict: """Provided by SqliteMemoryStore.""" diff --git a/mcp_server/tool_error_handler.py b/mcp_server/tool_error_handler.py index e33c344d..8374f782 100644 --- a/mcp_server/tool_error_handler.py +++ b/mcp_server/tool_error_handler.py @@ -9,6 +9,12 @@ DB methods) run on a worker thread instead of blocking the event loop +HC-CORTEX-002 adds a transaction-finalization boundary around every handler: +unfinished SQLite work is rolled back on failure, while apparent success with +an open transaction is rejected instead of emitting a false acknowledgement. +Registered PostgreSQL MCP tools already used the named offload path and retain +that behavior; unnamed compatibility calls now use the same offload boundary. + Issue #17 (PSGSupport): handlers that declare ``output_schema`` were rejected by FastMCP with ``structured_content must be a dict or None. Got str: '{...}'`` because this wrapper used to ``json.dumps`` the @@ -54,6 +60,7 @@ async def tool_remember(...) -> dict: from mcp_server.shared.json_native import to_json_native from mcp_server.handlers.admission import admit +from mcp_server.handlers.request_transaction import handler_transaction_scope from mcp_server.observability import metrics logger = logging.getLogger(__name__) @@ -156,7 +163,8 @@ def _run_coroutine_on_thread( """ loop = asyncio.new_event_loop() try: - return loop.run_until_complete(handler_fn(args)) + with handler_transaction_scope(): + return loop.run_until_complete(handler_fn(args)) finally: try: loop.close() @@ -173,19 +181,16 @@ async def safe_handler( ) -> dict[str, Any]: """Call a handler and return its dict, catching errors gracefully. - When ``tool_name`` is provided: + Every handler runs on a worker thread via ``asyncio.to_thread``. When + ``tool_name`` is provided: * The call is gated by the per-tool admission semaphore (Phase 5 step 5). Bounds concurrency so one client cannot DoS a tool by hammering it. - * The handler runs on a worker thread via ``asyncio.to_thread`` - (Phase 5 step 4). The handler body — which calls sync DB - methods — no longer blocks the event loop, and two concurrent - tool invocations genuinely run in parallel (the pool gives each - worker its own DB connection). + * Duration and outcome metrics include the tool name. - When ``tool_name`` is omitted the call runs in-line on the caller's - event loop without admission (backward-compat for code paths not - yet migrated). + When ``tool_name`` is omitted, admission and named metrics remain disabled, + but offload and transaction isolation are preserved. This keeps concurrent + compatibility calls from sharing the event-loop thread's SQLite handle. Contract (issue #17 — Liskov enforcement across all MCP handlers): precondition: ``handler_fn`` is an async callable returning a dict. @@ -221,7 +226,7 @@ async def safe_handler( {"tool": tool_name, "status": "ok"}, ) else: - result = await handler_fn(args) + result = await asyncio.to_thread(_run_coroutine_on_thread, handler_fn, args) # Defensive: every handler must already return a dict per its # ``output_schema``. If a handler regresses to None we surface # an empty dict so the MCP SDK's structured-content validator diff --git a/tests_py/infrastructure/test_sqlite_connection_registry.py b/tests_py/infrastructure/test_sqlite_connection_registry.py new file mode 100644 index 00000000..2c02548f --- /dev/null +++ b/tests_py/infrastructure/test_sqlite_connection_registry.py @@ -0,0 +1,137 @@ +"""Lifecycle and optional-extension contract for thread-local SQLite handles.""" + +from __future__ import annotations + +import sqlite3 +import threading +from pathlib import Path + +import numpy as np +import pytest + +from mcp_server.infrastructure.sqlite_connection_registry import ( + SqliteConnectionRegistry, +) +from mcp_server.infrastructure.sqlite_store import SqliteMemoryStore + +# source: repository worker teardown convention in +# tests_py/infrastructure/test_workflow_graph_source_ast.py. +_THREAD_SAFETY_BOUND_SECONDS = 5 + + +def _join(thread: threading.Thread) -> None: + thread.join(timeout=_THREAD_SAFETY_BOUND_SECONDS) + assert not thread.is_alive() + + +def test_registry_owns_and_closes_every_thread_connection(tmp_path: Path) -> None: + registry = SqliteConnectionRegistry(str(tmp_path / "registry.sqlite3")) + main_connection = registry.connection() + worker_connections: list[sqlite3.Connection] = [] + post_close_errors: list[Exception] = [] + worker_ready = threading.Event() + close_finished = threading.Event() + + def worker_lifecycle() -> None: + worker_connections.append(registry.connection()) + worker_ready.set() + assert close_finished.wait(timeout=_THREAD_SAFETY_BOUND_SECONDS) + try: + registry.connection() + except Exception as exc: # pragma: no cover - diagnostic boundary + post_close_errors.append(exc) + + worker = threading.Thread(target=worker_lifecycle) + + worker.start() + assert worker_ready.wait(timeout=_THREAD_SAFETY_BOUND_SECONDS) + assert len(worker_connections) == 1 + assert worker_connections[0] is not main_connection + + registry.close() + close_finished.set() + _join(worker) + for connection in (main_connection, worker_connections[0]): + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + connection.execute("SELECT 1") + assert len(post_close_errors) == 1 + assert isinstance(post_close_errors[0], sqlite3.ProgrammingError) + assert "registry is closed" in str(post_close_errors[0]) + with pytest.raises(sqlite3.ProgrammingError, match="registry is closed"): + registry.connection() + + +def test_future_worker_loads_enabled_vector_extension() -> None: + pytest.importorskip("sqlite_vec") + store = SqliteMemoryStore() + if not store.has_vec: + pytest.skip("sqlite-vec cannot be loaded on this SQLite build") + observed: list[tuple[int, int]] = [] + errors: list[Exception] = [] + + def insert_vector() -> None: + try: + embedding = np.zeros(384, dtype=np.float32).tobytes() + memory_id = store.insert_memory( + {"content": "worker-vector", "embedding": embedding} + ) + count = store._raw_conn.execute( + "SELECT COUNT(*) FROM memories_vec WHERE rowid = ?", (memory_id,) + ).fetchone()[0] + observed.append((memory_id, count)) + except Exception as exc: # pragma: no cover - diagnostic boundary + errors.append(exc) + + worker = threading.Thread(target=insert_vector) + worker.start() + _join(worker) + + assert errors == [] + assert len(observed) == 1 + assert observed[0][1] == 1 + store.close() + + +def test_journal_mode_matches_storage_kind(tmp_path: Path) -> None: + file_registry = SqliteConnectionRegistry(str(tmp_path / "journal.sqlite3")) + memory_registry = SqliteConnectionRegistry(":memory:") + + file_mode = file_registry.connection().execute("PRAGMA journal_mode").fetchone()[0] + memory_mode = ( + memory_registry.connection().execute("PRAGMA journal_mode").fetchone()[0] + ) + assert file_mode == "wal" + assert memory_mode == "memory" + file_registry.close() + memory_registry.close() + + +def test_relative_database_path_is_stable_after_cwd_change( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + relative_path = Path("relative.sqlite3") + store = SqliteMemoryStore(str(relative_path)) + moved_cwd = tmp_path / "moved-cwd" + moved_cwd.mkdir() + monkeypatch.chdir(moved_cwd) + observed_ids: list[int] = [] + errors: list[Exception] = [] + + def insert_from_worker() -> None: + try: + observed_ids.append( + store.insert_entity({"name": "stable-path", "type": "test"}) + ) + except Exception as exc: # pragma: no cover - diagnostic boundary + errors.append(exc) + + worker = threading.Thread(target=insert_from_worker) + worker.start() + _join(worker) + store.close() + + assert errors == [] + assert len(observed_ids) == 1 + assert (tmp_path / relative_path).is_file() + assert not (moved_cwd / relative_path).exists() diff --git a/tests_py/infrastructure/test_sqlite_request_transaction.py b/tests_py/infrastructure/test_sqlite_request_transaction.py new file mode 100644 index 00000000..31d2da09 --- /dev/null +++ b/tests_py/infrastructure/test_sqlite_request_transaction.py @@ -0,0 +1,233 @@ +"""HC-CORTEX-002 request-boundary transaction ownership regressions.""" + +from __future__ import annotations + +import asyncio +import sqlite3 +import threading +from pathlib import Path + +import pytest +from mcp.server.mcpserver.exceptions import ToolError + +from mcp_server.infrastructure import sqlite_connection_registry as registry_module +from mcp_server.infrastructure.sqlite_request_scope import ( + UncommittedSqliteTransactionError, +) +from mcp_server.infrastructure.sqlite_store import SqliteMemoryStore +from mcp_server.tool_error_handler import _run_coroutine_on_thread, safe_handler + +# source: repository worker teardown convention in +# tests_py/infrastructure/test_workflow_graph_source_ast.py. +_THREAD_SAFETY_BOUND_SECONDS = 5 + + +def test_success_cannot_acknowledge_uncommitted_work(tmp_path: Path) -> None: + store = SqliteMemoryStore(str(tmp_path / "unfinished-request.sqlite3")) + + async def unfinished_request(_args: dict) -> dict: + store._conn.execute( + "INSERT INTO entities (name, type) VALUES (?, ?)", + ("uncommitted", "test"), + ) + return {"acknowledged": True} + + with pytest.raises(UncommittedSqliteTransactionError, match="uncommitted"): + _run_coroutine_on_thread(unfinished_request, {}) + count = store._conn.execute( + "SELECT COUNT(*) AS count FROM entities WHERE name = ?", ("uncommitted",) + ).fetchone()["count"] + assert count == 0 + store.close() + + +def test_request_tracks_transaction_opened_by_nested_worker(tmp_path: Path) -> None: + store = SqliteMemoryStore(str(tmp_path / "nested-worker.sqlite3")) + + async def nested_worker_request(_args: dict) -> dict: + await asyncio.to_thread( + store._conn.execute, + "INSERT INTO memories (content) VALUES (?)", + ("nested-worker-uncommitted",), + ) + return {"acknowledged": True} + + with pytest.raises(UncommittedSqliteTransactionError, match="uncommitted"): + _run_coroutine_on_thread(nested_worker_request, {}) + count = store._conn.execute( + "SELECT COUNT(*) AS count FROM memories WHERE content = ?", + ("nested-worker-uncommitted",), + ).fetchone()["count"] + assert count == 0 + store.close() + + +def test_request_releases_connection_owned_by_nested_worker(tmp_path: Path) -> None: + store = SqliteMemoryStore(str(tmp_path / "nested-worker-lifecycle.sqlite3")) + initial_connections = len(store._connection_registry._connections) + + async def nested_worker_request(args: dict) -> dict: + entity_id = await asyncio.to_thread( + store.insert_entity, + {"name": args["name"], "type": "test"}, + ) + return {"entity_id": entity_id} + + try: + for name in ("first-request", "second-request"): + result = _run_coroutine_on_thread(nested_worker_request, {"name": name}) + assert result["entity_id"] > 0 + assert len(store._connection_registry._connections) == initial_connections + finally: + store.close() + + +def test_safe_handler_releases_connection_from_ephemeral_outer_executor( + tmp_path: Path, +) -> None: + store = SqliteMemoryStore(str(tmp_path / "outer-worker-lifecycle.sqlite3")) + initial_connections = len(store._connection_registry._connections) + + async def direct_request(args: dict) -> dict: + entity_id = store.insert_entity({"name": args["name"], "type": "test"}) + return {"entity_id": entity_id} + + try: + for name in ("first-request", "second-request"): + result = asyncio.run(safe_handler(direct_request, {"name": name})) + assert result["entity_id"] > 0 + assert len(store._connection_registry._connections) == initial_connections + finally: + store.close() + + +def test_nested_safe_handler_is_rejected_before_inner_work(tmp_path: Path) -> None: + store = SqliteMemoryStore(str(tmp_path / "nested-handler.sqlite3")) + inner_entered: list[bool] = [] + + async def rejected_inner(_args: dict) -> dict: + inner_entered.append(True) + store._conn.execute( + "INSERT INTO memories (content) VALUES (?)", ("rejected-inner",) + ) + raise RuntimeError("inner rejected") + + async def outer(_args: dict) -> dict: + with pytest.raises(ToolError): + await safe_handler(rejected_inner, {}) + entity_id = store.insert_entity({"name": "ack-outer", "type": "test"}) + return {"entity_id": entity_id} + + result = asyncio.run(safe_handler(outer, {})) + assert result["entity_id"] > 0 + assert inner_entered == [] + count = store._conn.execute( + "SELECT COUNT(*) AS count FROM memories WHERE content = ?", + ("rejected-inner",), + ).fetchone()["count"] + assert count == 0 + store.close() + + +def test_concurrent_unadmitted_handlers_do_not_share_transaction( + tmp_path: Path, +) -> None: + store = SqliteMemoryStore(str(tmp_path / "unadmitted-concurrency.sqlite3")) + rejected_written = threading.Event() + acknowledged_attempting = threading.Event() + + async def rejected(_args: dict) -> dict: + store._conn.execute( + "INSERT INTO memories (content) VALUES (?)", ("rejected-inline",) + ) + rejected_written.set() + assert await asyncio.to_thread( + acknowledged_attempting.wait, _THREAD_SAFETY_BOUND_SECONDS + ) + raise RuntimeError("reject during concurrent request") + + async def acknowledged(_args: dict) -> dict: + assert await asyncio.to_thread( + rejected_written.wait, _THREAD_SAFETY_BOUND_SECONDS + ) + acknowledged_attempting.set() + entity_id = store.insert_entity({"name": "ack-inline", "type": "test"}) + return {"entity_id": entity_id} + + async def run_both() -> list[object]: + return await asyncio.gather( + safe_handler(rejected, {}), + safe_handler(acknowledged, {}), + return_exceptions=True, + ) + + results = asyncio.run(run_both()) + assert isinstance(results[0], ToolError) + assert isinstance(results[1], dict) + count = store._conn.execute( + "SELECT COUNT(*) AS count FROM memories WHERE content = ?", + ("rejected-inline",), + ).fetchone()["count"] + assert count == 0 + store.close() + + +def test_rollback_failure_quarantines_dirty_connection( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + db_path = tmp_path / "rollback-quarantine.sqlite3" + store = SqliteMemoryStore(str(db_path)) + + def fail_rollback(_connection: object) -> None: + raise RuntimeError("injected rollback failure") + + async def rejected(_args: dict) -> dict: + store._conn.execute( + "INSERT INTO memories (content) VALUES (?)", ("rejected-cleanup",) + ) + raise ValueError("original handler failure") + + async def acknowledged(_args: dict) -> dict: + entity_id = store.insert_entity({"name": "ack-next", "type": "test"}) + return {"entity_id": entity_id} + + monkeypatch.setattr(registry_module, "_rollback_native", fail_rollback) + with pytest.raises(ValueError, match="original handler failure"): + _run_coroutine_on_thread(rejected, {}) + result = _run_coroutine_on_thread(acknowledged, {}) + assert result["entity_id"] > 0 + store.close() + + reopened = SqliteMemoryStore(str(db_path)) + rejected_count = reopened._conn.execute( + "SELECT COUNT(*) AS count FROM memories WHERE content = ?", + ("rejected-cleanup",), + ).fetchone()["count"] + acknowledged_count = reopened._conn.execute( + "SELECT COUNT(*) AS count FROM entities WHERE name = ?", ("ack-next",) + ).fetchone()["count"] + assert rejected_count == 0 + assert acknowledged_count == 1 + reopened.close() + + +def test_in_memory_anchor_rollback_failure_invalidates_registry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = SqliteMemoryStore() + + def fail_rollback(_connection: object) -> None: + raise RuntimeError("injected anchor rollback failure") + + async def rejected(_args: dict) -> dict: + store._conn.execute( + "INSERT INTO memories (content) VALUES (?)", ("rejected-anchor",) + ) + raise ValueError("original in-memory handler failure") + + monkeypatch.setattr(registry_module, "_rollback_native", fail_rollback) + with pytest.raises(ValueError, match="original in-memory handler failure"): + _run_coroutine_on_thread(rejected, {}) + with pytest.raises(sqlite3.ProgrammingError, match="anchor rollback failed"): + store._conn.execute("SELECT COUNT(*) FROM memories") + store.close() diff --git a/tests_py/infrastructure/test_sqlite_transaction_isolation.py b/tests_py/infrastructure/test_sqlite_transaction_isolation.py new file mode 100644 index 00000000..57abdb0c --- /dev/null +++ b/tests_py/infrastructure/test_sqlite_transaction_isolation.py @@ -0,0 +1,233 @@ +"""HC-CORTEX-002: a worker owns its complete SQLite transaction. + +The fault fixture is the smallest baseline reproduction from ADR-0055. A +supersede is rejected after its insert and compare-and-set while an unrelated +insert attempts to commit. The external ledger, not SQLite's physical +integrity result, decides which rows are allowed to survive. +""" + +from __future__ import annotations + +import sqlite3 +import threading +from dataclasses import dataclass, field +from pathlib import Path + +import numpy as np +import pytest + +from mcp_server.infrastructure.sqlite_store import SqliteMemoryStore +from mcp_server.tool_error_handler import _run_coroutine_on_thread + +# source: repository worker teardown convention in +# tests_py/infrastructure/test_workflow_graph_source_ast.py. +_THREAD_SAFETY_BOUND_SECONDS = 5 + + +class InjectedRollbackError(RuntimeError): + """Fault after the supersede CAS but before its commit.""" + + +def _memory(content: str) -> dict[str, object]: + embedding = np.zeros(384, dtype=np.float32).tobytes() + return { + "content": content, + "domain": "hc-cortex-002", + "embedding": embedding, + "heat": 0.5, + } + + +@dataclass +class OperationLedger: + acknowledged_ids: list[int] = field(default_factory=list) + rejected_operations: list[str] = field(default_factory=list) + busy_retries: int = 0 + errors: list[str] = field(default_factory=list) + + +class ConcurrentFaultFixture: + def __init__(self, store: SqliteMemoryStore, target_id: int) -> None: + self.store = store + self.target_id = target_id + self.ledger = OperationLedger() + self.writer_ready = threading.Event() + self.fault_window_open = threading.Event() + self.first_write_finished = threading.Event() + self.rollback_finished = threading.Event() + + def run(self) -> OperationLedger: + original_transfer = self.store._transfer_anchor + self.store._transfer_anchor = self._inject_after_cas # type: ignore[method-assign] + writer = threading.Thread(target=self._run_insert) + superseder = threading.Thread(target=self._run_supersede) + try: + writer.start() + self._wait(self.writer_ready, "writer connection was not ready") + superseder.start() + superseder.join(timeout=_THREAD_SAFETY_BOUND_SECONDS) + writer.join(timeout=_THREAD_SAFETY_BOUND_SECONDS) + assert not superseder.is_alive() and not writer.is_alive() + return self.ledger + finally: + self.store._transfer_anchor = original_transfer # type: ignore[method-assign] + + def _inject_after_cas(self, _head_id: int, _new_id: int) -> None: + self.fault_window_open.set() + self._wait(self.first_write_finished, "concurrent write never finished") + raise InjectedRollbackError("fault after supersede compare-and-set") + + def _run_supersede(self) -> None: + try: + self.store.supersede_atomic(_memory("rejected-supersede"), self.target_id) + except InjectedRollbackError: + self.ledger.rejected_operations.append("rejected-supersede") + except Exception as exc: # pragma: no cover - diagnostic boundary + self._record_error("supersede", exc) + finally: + self.rollback_finished.set() + + def _run_insert(self) -> None: + self.store._raw_conn.execute("PRAGMA busy_timeout=0") + self.writer_ready.set() + self._wait(self.fault_window_open, "fault window never opened") + try: + self._insert_acknowledged() + except sqlite3.OperationalError as exc: + if "locked" not in str(exc).lower(): + self._record_error("insert", exc) + else: + self.ledger.busy_retries += 1 + self.first_write_finished.set() + self._wait(self.rollback_finished, "fault rollback never finished") + self._retry_after_rollback() + else: + self.first_write_finished.set() + + def _retry_after_rollback(self) -> None: + try: + self._insert_acknowledged() + except Exception as exc: # pragma: no cover - diagnostic boundary + self._record_error("insert-retry", exc) + + def _insert_acknowledged(self) -> None: + memory_id = self.store.insert_memory(_memory("acknowledged-insert")) + self.ledger.acknowledged_ids.append(memory_id) + + def _record_error(self, operation: str, exc: Exception) -> None: + self.ledger.errors.append(f"{operation}:{type(exc).__name__}:{exc}") + + @staticmethod + def _wait(event: threading.Event, message: str) -> None: + assert event.wait(timeout=_THREAD_SAFETY_BOUND_SECONDS), message + + +def _snapshot( + store: SqliteMemoryStore, +) -> tuple[list[dict], list[dict], list[int] | None]: + rows = store._conn.execute( + "SELECT id, content, superseded_by_id FROM memories ORDER BY id" + ).fetchall() + fts_rows = store._conn.execute( + "SELECT rowid, content FROM memories_fts ORDER BY rowid" + ).fetchall() + vec_ids = None + if store.has_vec: + vec_ids = [ + row["rowid"] + for row in store._conn.execute( + "SELECT rowid FROM memories_vec ORDER BY rowid" + ).fetchall() + ] + return rows, fts_rows, vec_ids + + +def _assert_ledger_matches_store( + ledger: OperationLedger, + target_id: int, + rows: list[dict], + fts_rows: list[dict], + vec_ids: list[int] | None, +) -> None: + contents = [row["content"] for row in rows] + fts_contents = [row["content"] for row in fts_rows] + target = next(row for row in rows if row["id"] == target_id) + assert ledger.rejected_operations == ["rejected-supersede"] + assert len(ledger.acknowledged_ids) == 1 + assert ledger.busy_retries == 1 + assert ledger.errors == [] + assert contents == ["target", "acknowledged-insert"] + assert fts_contents == ["target", "acknowledged-insert"] + assert target["superseded_by_id"] is None + if vec_ids is not None: + assert vec_ids == [target_id, ledger.acknowledged_ids[0]] + + +def test_rejected_transaction_cannot_be_committed_by_another_worker( + tmp_path: Path, +) -> None: + db_path = tmp_path / "transaction-isolation.sqlite3" + store = SqliteMemoryStore(str(db_path)) + had_vec = store.has_vec + target_id = store.insert_memory(_memory("target")) + + ledger = ConcurrentFaultFixture(store, target_id).run() + rows, fts_rows, vec_ids = _snapshot(store) + _assert_ledger_matches_store(ledger, target_id, rows, fts_rows, vec_ids) + assert store._conn.execute("PRAGMA integrity_check").fetchone() == { + "integrity_check": "ok" + } + assert store._conn.execute("PRAGMA foreign_key_check").fetchall() == [] + store.close() + + reopened = SqliteMemoryStore(str(db_path)) + assert reopened.has_vec is had_vec + persisted_rows, persisted_fts, persisted_vec = _snapshot(reopened) + _assert_ledger_matches_store( + ledger, target_id, persisted_rows, persisted_fts, persisted_vec + ) + reopened.close() + + +def test_in_memory_workers_keep_transaction_ownership() -> None: + store = SqliteMemoryStore() + target_id = store.insert_memory(_memory("target")) + + ledger = ConcurrentFaultFixture(store, target_id).run() + rows, fts_rows, vec_ids = _snapshot(store) + _assert_ledger_matches_store(ledger, target_id, rows, fts_rows, vec_ids) + store.close() + + +def test_failed_request_cannot_leak_into_reused_worker(tmp_path: Path) -> None: + db_path = tmp_path / "request-boundary.sqlite3" + store = SqliteMemoryStore(str(db_path)) + store._conn.execute("DROP TABLE memories_fts") + store._conn.commit() + + async def rejected_request(_args: dict) -> dict: + store.insert_memory(_memory("rejected-partial")) + return {"acknowledged": True} + + async def acknowledged_request(_args: dict) -> dict: + entity_id = store.insert_entity({"name": "acknowledged", "type": "test"}) + return {"entity_id": entity_id} + + with pytest.raises(sqlite3.OperationalError, match="memories_fts"): + _run_coroutine_on_thread(rejected_request, {}) + result = _run_coroutine_on_thread(acknowledged_request, {}) + assert result["entity_id"] > 0 + store.close() + + reopened = SqliteMemoryStore(str(db_path)) + rejected_count = reopened._conn.execute( + "SELECT COUNT(*) AS count FROM memories WHERE content = ?", + ("rejected-partial",), + ).fetchone()["count"] + acknowledged_count = reopened._conn.execute( + "SELECT COUNT(*) AS count FROM entities WHERE name = ?", + ("acknowledged",), + ).fetchone()["count"] + assert rejected_count == 0 + assert acknowledged_count == 1 + reopened.close()