diff --git a/assertion_store.py b/assertion_store.py index 6b12b8e3..b4efea05 100644 --- a/assertion_store.py +++ b/assertion_store.py @@ -215,10 +215,16 @@ def _bounded_probability(value: float | None, field: str) -> float | None: class AssertionStore: """SQLite assertion store bound to the same physical DB as ``MessageStore``.""" - def __init__(self, db_path: str | Path, *, read_only: bool = False): + def __init__( + self, + db_path: str | Path, + *, + read_only: bool = False, + db_lock: Any | None = None, + ): self.db_path = Path(db_path) self.read_only = bool(read_only) - self._write_lock = threading.RLock() + self._write_lock = db_lock or threading.RLock() self._conn = self._open_connection() try: self._init_db() @@ -1096,3 +1102,22 @@ def remove_version(self, extraction_version: str) -> dict[str, int]: except Exception: self._conn.execute("ROLLBACK") raise + + +def _synchronized(method): + """Serialize operations that share one SQLite connection across clones.""" + def locked(self, *args, **kwargs): + with self._write_lock: + return method(self, *args, **kwargs) + + return locked + + +for _method_name in ( + "snapshot_source", + "has_current_receipt", + "plan_rebuild", + "query_assertions", + "query_relations", +): + setattr(AssertionStore, _method_name, _synchronized(getattr(AssertionStore, _method_name))) diff --git a/dag.py b/dag.py index 700c0a89..faf9f534 100644 --- a/dag.py +++ b/dag.py @@ -161,10 +161,10 @@ class SummaryDAG: DELETE_SESSION_SCOPE_TABLE = _DELETE_SESSION_SCOPE_TABLE - def __init__(self, db_path: str | Path): + def __init__(self, db_path: str | Path, *, db_lock: object | None = None): self.db_path = Path(db_path) self._conn: Optional[sqlite3.Connection] = None - self._db_lock = threading.RLock() + self._db_lock = db_lock or threading.RLock() self._init_db() @property @@ -889,3 +889,33 @@ def __del__(self) -> None: # pragma: no cover - defensive resource cleanup self.close() except Exception: pass + + +def _synchronized(method): + """Serialize every operation on the shared SQLite connection.""" + def locked(self, *args, **kwargs): + with self._db_lock: + return method(self, *args, **kwargs) + + return locked + + +for _method_name in ( + "reassign_session_nodes", + "get_node", + "get_session_nodes", + "get_session_node_ids_below_depth", + "count_at_depth", + "get_session_node_count", + "get_session_depth_stats", + "get_session_depth_samples", + "get_uncondensed_at_depth", + "search", + "_search_like", + "get_source_nodes", + "source_message_ids", + "_node_matches_source", + "get_source_time_window", + "close", +): + setattr(SummaryDAG, _method_name, _synchronized(getattr(SummaryDAG, _method_name))) diff --git a/engine.py b/engine.py index 97a69ab4..70cf1207 100644 --- a/engine.py +++ b/engine.py @@ -14,8 +14,9 @@ import sqlite3 import threading import time -from collections import deque import uuid +import weakref +from collections import deque from pathlib import Path from typing import Any, Callable, Dict, List, Optional @@ -362,6 +363,128 @@ def _run(self) -> None: _LCM_MESSAGE_PREFIX_FINGERPRINT_LIMIT = 8 +_STORAGE_REGISTRY_LOCK = threading.RLock() +_STORAGE_LOCK_BY_DB_PATH: weakref.WeakValueDictionary[Path, threading.RLock] = ( + weakref.WeakValueDictionary() +) + + +def _storage_lock_for(db_path: str | Path) -> threading.RLock: + """Return the process-wide operation lock for one canonical database.""" + canonical_path = Path(db_path).expanduser().resolve() + with _STORAGE_REGISTRY_LOCK: + return _STORAGE_LOCK_BY_DB_PATH.setdefault(canonical_path, threading.RLock()) + + +class _SharedStorage: + """Reference-counted SQLite helpers shared by engines using one database.""" + + def __init__(self, db_path: str | Path, config: LCMConfig, hermes_home: str): + self.db_path = Path(db_path).expanduser().resolve() + self._lock = threading.RLock() + self._operation_lock = _storage_lock_for(self.db_path) + self._owners = 0 + self._closing = False + self._closed = False + self.store = self.dag = self.lifecycle = self.assertions = self.query_views = None + helpers = [] + try: + # Constructors perform schema and WAL setup. Independent engines for + # one database must not race before their operation locks take over. + with self._operation_lock: + self.store = MessageStore( + self.db_path, + ingest_protection_config=config, + hermes_home=hermes_home, + db_lock=self._operation_lock, + ) + helpers.append(self.store) + self.dag = SummaryDAG(self.db_path, db_lock=self._operation_lock) + helpers.append(self.dag) + if config.temporal_rollups_enabled: + initialize_rollup_invalidation_outbox(self.dag) + self.lifecycle = LifecycleStateStore( + self.db_path, db_lock=self._operation_lock + ) + helpers.append(self.lifecycle) + if bool(getattr(config, "assertions_enabled", False)): + self.assertions = AssertionStore( + self.db_path, db_lock=self._operation_lock + ) + helpers.append(self.assertions) + if bool(getattr(config, "query_views_enabled", False)) or bool( + getattr(config, "adaptive_retrieval_enabled", False) + ): + self.query_views = QueryViewStore( + self.db_path, db_lock=self._operation_lock + ) + helpers.append(self.query_views) + except BaseException: + for helper in reversed(helpers): + try: + helper.close() + except BaseException: + pass + raise + + def acquire(self) -> "_SharedStorage": + # Increment the owner count. A closed or closing bundle must never be + # re-acquired: its SQLite helpers are (or are about to be) closed. + with self._lock: + if self._closed or self._closing: + raise RuntimeError("cannot acquire closed LCM storage") + self._owners += 1 + return self + + def release(self) -> None: + # Decrement the owner count; only the last release triggers teardown. + with self._lock: + if self._owners <= 0: + return + self._owners -= 1 + if self._owners: + return # Other engines still share this bundle; keep it open. + self._closing = True # Last owner: block new acquires, then close. + + failures: list[BaseException] = [] + try: + # Teardown is one path-scoped operation: no independent bundle may + # initialize or use a same-database helper between these closes. + with self._operation_lock: + for helper in ( + self.store, + self.dag, + self.lifecycle, + self.assertions, + self.query_views, + ): + close = getattr(helper, "close", None) + if callable(close): + try: + close() + except BaseException as exc: + failures.append(exc) + finally: + with self._lock: + self._closing = False + self._closed = True + if failures: + raise BaseExceptionGroup("LCM storage close failed", failures) + + +class _StorageOwnership: + """One idempotent storage lease for explicit and finalizer cleanup.""" + + def __init__(self, storage: _SharedStorage): + self._storage: _SharedStorage | None = storage + self._lock = threading.Lock() + + def release(self) -> None: + with self._lock: + storage, self._storage = self._storage, None + if storage is not None: + storage.release() + def _normalize_total_compactions(value: Any) -> int: """Return a persisted compaction total only when it is a valid counter.""" if isinstance(value, bool) or not isinstance(value, int) or value < 0: @@ -387,10 +510,19 @@ class LCMEngine(CompactionMixin, ResetStateMixin, ReconcileMixin, AuxiliarySessi 5. Active context = system prompt + DAG summaries + fresh tail """ - def __init__(self, config: LCMConfig | None = None, - hermes_home: str = ""): + def __init__( + self, + config: LCMConfig | None = None, + hermes_home: str = "", + *, + _storage: _SharedStorage | None = None, + ): self._config = config or LCMConfig.from_env() self._hermes_home = hermes_home + self._storage: _SharedStorage | None = None + self._ownership: _StorageOwnership | None = None + self._ownership_lock = threading.Lock() + self._ownership_finalizer: weakref.finalize | None = None self._assertion_extraction_metrics_lock = threading.RLock() self._assertion_extraction_idle = threading.Event() self._assertion_extraction_idle.set() @@ -406,8 +538,10 @@ def __init__(self, config: LCMConfig | None = None, self._assertion_extraction_last_error = "" self._assertion_extraction_last_model = "" - db_path = self._resolve_db_path(hermes_home) - self._bind_storage(db_path, hermes_home) + if _storage is None: + self._bind_storage(self._resolve_db_path(hermes_home), hermes_home) + else: + self._adopt_storage(_storage.acquire()) self._session_id: str = "" self._session_platform: str = "" @@ -632,41 +766,49 @@ def clone_for_agent(self) -> "LCMEngine": sharing one registered instance across agents can let one conversation rebind another conversation's raw-message ingest and lifecycle state. - The clone shares the same durable SQLite database path/configuration, + The clone shares the process-owned SQLite storage helpers, but gets independent session/cursor/lifecycle runtime state. Runtime model and context-window metadata is copied so the clone is immediately budget-aware even before a compatible Hermes host calls update_model(). """ + storage = self._storage + if storage is None: + raise RuntimeError("cannot clone an LCM engine after it has shut down") clone = type(self)( config=copy.deepcopy(self._config), hermes_home=self._hermes_home, + _storage=storage, ) - clone.model = self.model - clone.base_url = self.base_url - clone.api_key = self.api_key - clone.provider = self.provider - clone.api_mode = self.api_mode - if self._context_length_source: - clone._set_context_length( - self.raw_context_length, - source=self._context_length_source, - model=self.model, - provider=self.provider, - ) - elif self.raw_context_length or self.context_length: - clone._set_context_length( - self.raw_context_length or self.context_length, - source="clone_for_agent", - model=self.model, - provider=self.provider, - ) - # ``update_model()`` authority is a per-runtime lifecycle edge, not - # durable metadata. Compatible hosts call update_model() on the clone - # before binding it; hosts that bind only through on_session_start() - # must still be able to replace the copied prototype route. - clone._update_model_pending_session_start = False - clone._lcm_current_start_allows_bypass_lineage = False - return clone + try: + clone.model = self.model + clone.base_url = self.base_url + clone.api_key = self.api_key + clone.provider = self.provider + clone.api_mode = self.api_mode + if self._context_length_source: + clone._set_context_length( + self.raw_context_length, + source=self._context_length_source, + model=self.model, + provider=self.provider, + ) + elif self.raw_context_length or self.context_length: + clone._set_context_length( + self.raw_context_length or self.context_length, + source="clone_for_agent", + model=self.model, + provider=self.provider, + ) + # ``update_model()`` authority is a per-runtime lifecycle edge, not + # durable metadata. Compatible hosts call update_model() on the clone + # before binding it; hosts that bind only through on_session_start() + # must still be able to replace the copied prototype route. + clone._update_model_pending_session_start = False + clone._lcm_current_start_allows_bypass_lineage = False + return clone + except BaseException: + clone._close_storage() + raise def __deepcopy__(self, memo: dict[int, object]) -> "LCMEngine": """Copy the plugin runtime without pickling SQLite-backed helpers. @@ -675,8 +817,8 @@ def __deepcopy__(self, memo: dict[int, object]) -> "LCMEngine": AIAgent instances. A default object deepcopy walks into MessageStore, SummaryDAG, and LifecycleStateStore sqlite3.Connection handles, which cannot be pickled. LCM already exposes clone_for_agent() as the safe - boundary: share durable configuration/database path, but allocate fresh - per-agent runtime/storage helper objects. + boundary: share durable storage, but allocate fresh per-agent runtime + state. """ clone = self.clone_for_agent() memo[id(self)] = clone @@ -690,72 +832,54 @@ def _resolve_db_path(self, hermes_home: str = "") -> Path: return Path(hermes_home) / "lcm.db" return Path.home() / ".hermes" / "lcm.db" - def _bind_storage(self, db_path: str | Path, hermes_home: str = "") -> None: - """Bind store/DAG/lifecycle helpers to one SQLite database.""" - self._assertions = None - self._query_views = None - self._adaptive_retrieval = None - self._assertion_extractor = None + def _adopt_storage(self, storage: _SharedStorage) -> None: + ownership = _StorageOwnership(storage) + self._storage = storage + self._ownership = ownership try: - self._store = MessageStore( - db_path, - ingest_protection_config=self._config, - hermes_home=hermes_home, - ) - self._dag = SummaryDAG(db_path) - if self._config.temporal_rollups_enabled: - # Install the transaction-coupled summary mutation triggers before - # this engine can publish or delete a DAG node. - initialize_rollup_invalidation_outbox(self._dag) - self._lifecycle = LifecycleStateStore(db_path) - self._assertions = ( - AssertionStore(db_path) - if bool(getattr(self._config, "assertions_enabled", False)) - else None - ) - self._query_views = ( - QueryViewStore(db_path) - if bool(getattr(self._config, "query_views_enabled", False)) - or bool(getattr(self._config, "adaptive_retrieval_enabled", False)) - else None - ) + self._store = storage.store + self._dag = storage.dag + self._lifecycle = storage.lifecycle + self._assertions = storage.assertions + self._query_views = storage.query_views self._adaptive_retrieval = ( AdaptiveRetrievalRegistry(self._query_views) - if bool( - getattr(self._config, "adaptive_retrieval_enabled", False) - ) + if bool(getattr(self._config, "adaptive_retrieval_enabled", False)) else None ) - if ( - self._assertions is not None - and bool(getattr(self._config, "assertion_extraction_enabled", False)) + self._assertion_extractor = None + if self._assertions is not None and bool( + getattr(self._config, "assertion_extraction_enabled", False) ): self._assertion_extractor = ModelAssertionExtractor( self._assertions, model=self._assertion_extraction_model(), timeout_seconds=self._assertion_extraction_timeout(), ) - except Exception: - self._close_storage() + except BaseException: + self._storage = None + self._ownership = None + ownership.release() raise + self._ownership_finalizer = weakref.finalize(self, ownership.release) + + def _bind_storage(self, db_path: str | Path, hermes_home: str = "") -> None: + """Bind a new clone-family bundle using the lock for ``db_path``.""" + self._adopt_storage(_SharedStorage(db_path, self._config, hermes_home).acquire()) def _close_storage(self) -> None: - """Best-effort close of currently bound SQLite helpers.""" - for attr in ( - "_adaptive_retrieval", - "_store", - "_dag", - "_lifecycle", - "_assertions", - "_query_views", - ): - helper = getattr(self, attr, None) - close = getattr(helper, "close", None) - if callable(close): - try: - close() - except Exception: - logger.debug("LCM failed closing %s during profile rebind", attr, exc_info=True) + """Atomically detach and release this engine's storage lease.""" + with self._ownership_lock: + ownership = self._ownership + if ownership is None: + return + self._ownership = None + self._storage = None + finalizer = self._ownership_finalizer + self._ownership_finalizer = None + if finalizer is not None: + finalizer.detach() + ownership.release() def _assertion_extraction_model(self) -> str: return str( @@ -834,10 +958,10 @@ def _rebind_storage_for_home(self, hermes_home: str = "") -> bool: current_store_home = str(getattr(getattr(self, "_store", None), "_hermes_home", "") or "") if current_home == str(hermes_home) and current_store_home == str(hermes_home): return False + db_path = self._resolve_db_path(hermes_home) + self._close_storage() self._hermes_home = hermes_home - store = getattr(self, "_store", None) - if store is not None: - store._hermes_home = hermes_home + self._bind_storage(db_path, hermes_home) self._reset_profile_runtime_state() logger.info("LCM rebound Hermes home for configured database path %s", hermes_home) return True @@ -6657,10 +6781,4 @@ def shutdown(self): self._unregister_active_engine_binding() if self._adaptive_retrieval is not None: self._adaptive_retrieval.close() - self._store.close() - self._dag.close() - self._lifecycle.close() - if self._assertions is not None: - self._assertions.close() - if self._query_views is not None: - self._query_views.close() + self._close_storage() diff --git a/lifecycle_state.py b/lifecycle_state.py index 282ed482..69415624 100644 --- a/lifecycle_state.py +++ b/lifecycle_state.py @@ -54,7 +54,7 @@ class LifecycleState: class LifecycleStateStore: - def __init__(self, db_path: str | Path): + def __init__(self, db_path: str | Path, *, db_lock: Any | None = None): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._conn: Optional[sqlite3.Connection] = None @@ -62,7 +62,7 @@ def __init__(self, db_path: str | Path): # and is shared across the gateway thread, dispatcher, and sub-agents. # Serialize read-modify-write flows so concurrent binds/frontier # advances cannot interleave and regress the checkpoint. - self._lock = threading.RLock() + self._lock = db_lock or threading.RLock() self._init_db() def _init_db(self) -> None: @@ -79,14 +79,15 @@ def _init_db(self) -> None: self._conn.commit() def close(self) -> None: - conn = getattr(self, "_conn", None) - if conn is not None: - try: - conn.execute("PRAGMA wal_checkpoint(PASSIVE)") - except sqlite3.Error: - pass - conn.close() - self._conn = None + with self._lock: + conn = getattr(self, "_conn", None) + if conn is not None: + try: + conn.execute("PRAGMA wal_checkpoint(PASSIVE)") + except sqlite3.Error: + pass + conn.close() + self._conn = None def __del__(self) -> None: # pragma: no cover - defensive resource cleanup try: @@ -105,6 +106,7 @@ def connection(self) -> sqlite3.Connection | None: """ return getattr(self, "_conn", None) + @_synchronized def row_count(self) -> int: row = self._conn.execute("SELECT COUNT(*) AS count FROM lcm_lifecycle_state").fetchone() return int(row["count"] if row else 0) @@ -129,6 +131,7 @@ def _row_to_state(self, row: sqlite3.Row | None) -> LifecycleState | None: updated_at=float(row["updated_at"] or 0.0), ) + @_synchronized def get_by_conversation(self, conversation_id: str | None) -> LifecycleState | None: if not conversation_id: return None @@ -138,6 +141,7 @@ def get_by_conversation(self, conversation_id: str | None) -> LifecycleState | N ).fetchone() return self._row_to_state(row) + @_synchronized def get_by_session(self, session_id: str | None) -> LifecycleState | None: if not session_id: return None @@ -370,6 +374,7 @@ def record_rollover( assert updated is not None return updated + @_synchronized def get_fragmentation_stats(self, state_db_path: str | Path | None = None) -> dict[str, Any]: """Return read-only lifecycle/session fragmentation diagnostics. @@ -627,6 +632,7 @@ def record_debt( self._conn.commit() return self.get_by_conversation(conversation_id) + @_synchronized def clear_debt(self, conversation_id: str | None) -> LifecycleState | None: if not conversation_id: return None @@ -809,6 +815,7 @@ def _session_has_data(session_id: str) -> bool: conn.rollback() raise + @_synchronized def delete_safe_rows_for_sessions( self, session_ids: set[str] | list[str] | tuple[str, ...], diff --git a/query_view_store.py b/query_view_store.py index 7c7385c6..5c0b9e05 100644 --- a/query_view_store.py +++ b/query_view_store.py @@ -532,13 +532,13 @@ def _verify_query_view_schema(conn: sqlite3.Connection) -> list[str]: class QueryViewStore: """Versioned, exact-provenance materialized evidence views.""" - def __init__(self, db_path: str | Path): + def __init__(self, db_path: str | Path, *, db_lock: Any | None = None): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._conn = sqlite3.connect( str(self.db_path), timeout=5.0, check_same_thread=False ) - self._write_lock = threading.RLock() + self._write_lock = db_lock or threading.RLock() try: refuse_schema_version_too_new(self._conn) configure_connection(self._conn) @@ -1272,3 +1272,28 @@ def __del__(self) -> None: # pragma: no cover - defensive cleanup self.close() except Exception: pass + + +def _synchronized(method): + """Serialize operations that share one SQLite connection across clones.""" + def locked(self, *args, **kwargs): + with self._write_lock: + return method(self, *args, **kwargs) + + return locked + + +for _method_name in ( + "corpus_snapshot", + "snapshot_dependency", + "claim_build", + "publish_ready", + "mark_failed", + "reclaim_expired_builds", + "delta_events", + "lookup", + "expire_views", + "purge_expired", + "prune_corpus_events", +): + setattr(QueryViewStore, _method_name, _synchronized(getattr(QueryViewStore, _method_name))) diff --git a/scripts/measure_clone_storage.py b/scripts/measure_clone_storage.py new file mode 100644 index 00000000..2433c8d3 --- /dev/null +++ b/scripts/measure_clone_storage.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Measure LCM SQLite storage sharing across retained engine clones. + +Example: + python scripts/measure_clone_storage.py --samples 25 --clones 10 --json +""" + +from __future__ import annotations + +import argparse +import json +import os +import statistics +import sys +import tempfile +import time +from pathlib import Path +from types import ModuleType + +REPO_ROOT = Path(os.environ.get("LCM_BENCH_REPO_ROOT", Path(__file__).resolve().parents[1])).resolve() +PACKAGE_NAME = "hermes_lcm" +if PACKAGE_NAME not in sys.modules: + package = ModuleType(PACKAGE_NAME) + package.__path__ = [str(REPO_ROOT)] + package.__package__ = PACKAGE_NAME + sys.modules[PACKAGE_NAME] = package + +from hermes_lcm.config import LCMConfig +from hermes_lcm.engine import LCMEngine + + +def _open_fd_count() -> int: + """Return currently open descriptors on platforms exposing /dev/fd.""" + try: + return len(os.listdir("/dev/fd")) + except OSError: + return -1 + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--samples", type=int, default=25, help="Timed samples per measurement.") + parser.add_argument("--clones", type=int, default=10, help="Retained clones for FD and batch measurements.") + parser.add_argument("--database", help="SQLite database path (default: temporary database).") + parser.add_argument("--json", action="store_true", help="Print JSON instead of a readable report.") + return parser.parse_args(argv) + + +def _shutdown_all(engines: list[LCMEngine]) -> None: + for engine in reversed(engines): + engine.shutdown() + + +def run(samples: int, clones: int, database: Path) -> dict[str, float | int]: + if samples < 1 or clones < 1: + raise ValueError("--samples and --clones must both be positive") + + startup_ms: list[float] = [] + for _ in range(samples): + started = time.perf_counter_ns() + engine = LCMEngine(config=LCMConfig(database_path=str(database))) + startup_ms.append((time.perf_counter_ns() - started) / 1_000_000) + engine.shutdown() + + prototype = LCMEngine(config=LCMConfig(database_path=str(database))) + retained: list[LCMEngine] = [] + try: + fd_before = _open_fd_count() + clone_ms: list[float] = [] + for _ in range(samples): + started = time.perf_counter_ns() + clone = prototype.clone_for_agent() + clone_ms.append((time.perf_counter_ns() - started) / 1_000_000) + clone.shutdown() + + started = time.perf_counter_ns() + retained = [prototype.clone_for_agent() for _ in range(clones)] + ten_clone_ms = (time.perf_counter_ns() - started) / 1_000_000 + fd_after = _open_fd_count() + finally: + _shutdown_all(retained) + prototype.shutdown() + + return { + "fd_before_retained_clones": fd_before, + "fd_after_retained_clones": fd_after, + "retained_clone_fd_delta": fd_after - fd_before if fd_before >= 0 and fd_after >= 0 else -1, + "median_clone_setup_ms": statistics.median(clone_ms), + "ten_clone_setup_ms": ten_clone_ms, + "median_initial_startup_ms": statistics.median(startup_ms), + "samples": samples, + "retained_clones": clones, + } + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(argv if argv is not None else sys.argv[1:]) + if args.database: + report = run(args.samples, args.clones, Path(args.database)) + else: + with tempfile.TemporaryDirectory(prefix="lcm-clone-storage-") as directory: + report = run(args.samples, args.clones, Path(directory) / "lcm.db") + + if args.json: + print(json.dumps(report, indent=2, sort_keys=True)) + else: + for name, value in report.items(): + print(f"{name}: {value}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/store.py b/store.py index 7ebfc414..95f2dbe0 100644 --- a/store.py +++ b/store.py @@ -327,7 +327,14 @@ def build_message_fts_spec() -> ExternalContentFtsSpec: class MessageStore: """SQLite-backed immutable message store.""" - def __init__(self, db_path: str | Path, *, ingest_protection_config=None, hermes_home: str = ""): + def __init__( + self, + db_path: str | Path, + *, + ingest_protection_config=None, + hermes_home: str = "", + db_lock: object | None = None, + ): self.db_path = Path(db_path) self._is_memory_database = str(self.db_path) == ":memory:" if not self._is_memory_database: @@ -335,23 +342,8 @@ def __init__(self, db_path: str | Path, *, ingest_protection_config=None, hermes self._ingest_protection_config = ingest_protection_config or LCMConfig(database_path=str(self.db_path)) self._hermes_home = hermes_home or str(self.db_path.parent) self._conn: Optional[sqlite3.Connection] = None - # ``self._conn`` is shared across threads (the connection is opened with - # ``check_same_thread=False``). SQLite's own C-level mutex serializes - # statements at the engine layer, but the Python ``sqlite3`` module - # releases the GIL while the C call runs. Under heavy thread contention - # with concurrent HTTPS clients in the same process, downstream - # operators have observed on-disk corruption that is consistent with - # external bytes landing inside SQLite's write path (e.g. the first - # 28 bytes of the database file replaced with a TLS record header + - # ciphertext while the "SQLit" magic remains intact). - # - # This re-entrant lock is defense-in-depth: it forces all write call - # sites that use ``self._conn`` to be serialized at the Python layer, - # eliminating any window where Python-side buffer reuse or memory - # aliasing could intersect SQLite's flush of a write. It does not - # change semantics for single-threaded callers and adds only a single - # uncontended ``RLock.acquire``/``release`` pair per operation. - self._write_lock = threading.RLock() + # One canonical database path uses one lock across all helper bundles. + self._write_lock = db_lock or threading.RLock() self._init_db() def _init_db(self): @@ -1757,3 +1749,50 @@ def __del__(self) -> None: # pragma: no cover - defensive resource cleanup self.close() except Exception: pass + + +def _synchronized(method): + """Serialize every operation on the shared SQLite connection.""" + def locked(self, *args, **kwargs): + with self._write_lock: + return method(self, *args, **kwargs) + + return locked + + +for _method_name in ( + "append", + "append_batch", + "_append_protected_batch", + "reassign_session_messages", + "delete_session_messages", + "gc_externalized_tool_result", + "pin", + "unpin", + "get", + "get_batch", + "scan_evidence_rows", + "get_range", + "count_session_load_messages", + "load_session_page", + "load_session_window", + "get_session_messages", + "get_session_messages_after", + "get_session_tail", + "get_session_count", + "get_session_token_total", + "get_source_stats", + "scan_session_cleanup_stats", + "scan_session_retention_stats", + "get_source_normalization_plan", + "normalize_legacy_blank_sources", + "get_time_bounds", + "read_metadata_json", + "write_metadata_json", + "commit", + "backup", + "search", + "_search_like", + "close", +): + setattr(MessageStore, _method_name, _synchronized(getattr(MessageStore, _method_name))) diff --git a/tests/test_lcm_core.py b/tests/test_lcm_core.py index ff6725ec..3bc521f6 100644 --- a/tests/test_lcm_core.py +++ b/tests/test_lcm_core.py @@ -7425,9 +7425,9 @@ def test_deepcopy_uses_clone_for_agent_without_copying_sqlite_handles(self, tmp_ assert clone is not prototype assert isinstance(clone, LCMEngine) - assert clone._store is not prototype._store - assert clone._dag is not prototype._dag - assert clone._lifecycle is not prototype._lifecycle + assert clone._store is prototype._store + assert clone._dag is prototype._dag + assert clone._lifecycle is prototype._lifecycle assert clone._config.database_path == prototype._config.database_path assert clone._hermes_home == prototype._hermes_home finally: @@ -7468,9 +7468,9 @@ def host_selects_context_engine(candidate): assert isinstance(clone, LCMEngine) assert clone.name == "lcm" assert clone is not prototype - assert clone._store is not prototype._store - assert clone._dag is not prototype._dag - assert clone._lifecycle is not prototype._lifecycle + assert clone._store is prototype._store + assert clone._dag is prototype._dag + assert clone._lifecycle is prototype._lifecycle assert clone._session_id == "" assert clone._conversation_id == "" assert clone.model == prototype.model @@ -7708,6 +7708,378 @@ def test_deepcopy_preserves_zero_context_metadata_without_pending_authority(self shutdown() +class TestLCMEngineSharedStorage: + def _engine(self, tmp_path): + from hermes_lcm.engine import LCMEngine + + return LCMEngine( + config=LCMConfig(database_path=str(tmp_path / "shared-storage.db")), + hermes_home=str(tmp_path / "hermes"), + ) + + def test_clone_shares_bundle_but_keeps_runtime_and_model_metadata_local(self, tmp_path): + prototype = self._engine(tmp_path) + clone = None + try: + prototype.update_model( + model="prototype-model", + provider="prototype-provider", + base_url="https://prototype.invalid/v1", + api_key="prototype-key", + api_mode="chat", + context_length=128_000, + ) + prototype.on_session_start("prototype", platform="alpha", conversation_id="alpha:1") + clone = prototype.clone_for_agent() + + assert clone._storage is prototype._storage + assert clone._store is prototype._store + assert clone._dag is prototype._dag + assert clone._lifecycle is prototype._lifecycle + assert clone._session_id == "" + assert clone._conversation_id == "" + assert clone.model == "prototype-model" + assert clone.provider == "prototype-provider" + + clone.on_session_start("clone", platform="beta", conversation_id="beta:1") + clone.update_model(model="clone-model", provider="clone-provider", context_length=16_000) + assert prototype._session_id == "prototype" + assert prototype._conversation_id == "alpha:1" + assert prototype.model == "prototype-model" + assert prototype.provider == "prototype-provider" + finally: + prototype.shutdown() + if clone is not None: + clone.shutdown() + + def test_owner_first_shutdown_leaves_clone_operational_and_final_close_is_once(self, tmp_path, monkeypatch): + prototype = self._engine(tmp_path) + clone = prototype.clone_for_agent() + bundle = prototype._storage + close_calls = {"store": 0, "dag": 0, "lifecycle": 0} + try: + for name in close_calls: + helper = getattr(bundle, name) + original = helper.close + + def close(original=original, name=name): + close_calls[name] += 1 + return original() + + monkeypatch.setattr(helper, "close", close) + + prototype.shutdown() + clone._store.append("clone-session", {"role": "user", "content": "still open"}) + assert close_calls == {"store": 0, "dag": 0, "lifecycle": 0} + + clone.shutdown() + clone.shutdown() + assert close_calls == {"store": 1, "dag": 1, "lifecycle": 1} + finally: + prototype.shutdown() + clone.shutdown() + + def test_independent_engines_with_equivalent_paths_keep_bundles_separate_but_share_lock(self, tmp_path): + db_path = tmp_path / "independent.db" + barrier = threading.Barrier(2) + engines = [] + failures = [] + + def construct(path): + try: + barrier.wait() + from hermes_lcm.engine import LCMEngine + + engines.append(LCMEngine(config=LCMConfig(database_path=str(path)))) + except BaseException as exc: + failures.append(exc) + + alternate_path = db_path.parent / "." / db_path.name + threads = [threading.Thread(target=construct, args=(path,)) for path in (db_path, alternate_path)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + try: + assert failures == [] + assert len(engines) == 2 + assert engines[0]._storage is not engines[1]._storage + assert engines[0]._store._write_lock is engines[1]._store._write_lock + assert engines[0]._dag._db_lock is engines[1]._dag._db_lock + assert engines[0]._lifecycle._lock is engines[1]._lifecycle._lock + finally: + for engine in engines: + engine.shutdown() + + def test_configured_path_profile_rebind_detaches_one_clone(self, tmp_path): + db_path = tmp_path / "configured.db" + prototype = self._engine(tmp_path) + prototype._config.database_path = str(db_path) + prototype._rebind_storage_for_home(str(tmp_path / "profile-a")) + clone = prototype.clone_for_agent() + original_storage = prototype._storage + try: + assert clone._storage is original_storage + assert clone._rebind_storage_for_home(str(tmp_path / "profile-b")) is True + assert clone._storage is not original_storage + assert clone._store._write_lock is prototype._store._write_lock + assert prototype._hermes_home == str(tmp_path / "profile-a") + assert clone._hermes_home == str(tmp_path / "profile-b") + prototype._store.append("prototype", {"role": "user", "content": "still live"}) + finally: + prototype.shutdown() + clone.shutdown() + + def test_final_close_attempts_every_helper_and_raises_aggregate_failure(self, tmp_path, monkeypatch): + prototype = self._engine(tmp_path) + bundle = prototype._storage + attempted = [] + try: + for name in ("store", "dag", "lifecycle"): + helper = getattr(bundle, name) + + def close(name=name): + attempted.append(name) + raise RuntimeError(name) + + monkeypatch.setattr(helper, "close", close) + + with pytest.raises(BaseExceptionGroup, match="LCM storage close failed") as exc_info: + prototype.shutdown() + assert attempted == ["store", "dag", "lifecycle"] + assert {str(error) for error in exc_info.value.exceptions} == {"store", "dag", "lifecycle"} + finally: + prototype.shutdown() + + def test_concurrent_clone_writes_and_dag_operations_are_serialized(self, tmp_path): + prototype = self._engine(tmp_path) + clones = [prototype.clone_for_agent() for _ in range(6)] + barrier = threading.Barrier(len(clones)) + failures = [] + + def write(index, clone): + try: + barrier.wait() + session_id = f"session-{index}" + for ordinal in range(20): + clone._store.append( + session_id, + {"role": "user", "content": f"message-{index}-{ordinal}"}, + ) + clone._dag.add_node( + SummaryNode(session_id=session_id, summary=f"node-{index}-{ordinal}") + ) + clone._dag.get_session_depth_stats(session_id) + except BaseException as exc: + failures.append(exc) + + threads = [ + threading.Thread(target=write, args=(index, clone)) + for index, clone in enumerate(clones) + ] + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert failures == [] + for index in range(len(clones)): + session_id = f"session-{index}" + assert prototype._store.get_session_count(session_id) == 20 + assert prototype._dag.get_session_node_count(session_id) == 20 + finally: + for engine in (*clones, prototype): + engine.shutdown() + + def test_shared_lifecycle_reads_wait_for_path_lock(self, tmp_path): + prototype = self._engine(tmp_path) + clone = prototype.clone_for_agent() + started = threading.Event() + finished = threading.Event() + failures = [] + + def read_state(): + try: + started.set() + clone._lifecycle.row_count() + clone._lifecycle.get_by_conversation("missing") + clone._lifecycle.get_by_session("missing") + clone._lifecycle.get_fragmentation_stats() + except BaseException as exc: + failures.append(exc) + finally: + finished.set() + + try: + with prototype._storage._operation_lock: + thread = threading.Thread(target=read_state) + thread.start() + assert started.wait(timeout=1) + assert not finished.wait(timeout=0.05) + thread.join(timeout=2) + assert finished.is_set() + assert failures == [] + finally: + prototype.shutdown() + clone.shutdown() + + def test_dag_delete_callback_runs_outside_path_lock(self, tmp_path): + prototype = self._engine(tmp_path) + prototype._dag.add_node(SummaryNode(session_id="delete", summary="node")) + callback_lock_state = [] + + def on_deleted_batch(_node_ids): + acquired = [] + + def acquire_lock(): + with prototype._storage._operation_lock: + acquired.append(True) + + thread = threading.Thread(target=acquire_lock) + thread.start() + thread.join(timeout=1) + callback_lock_state.append(bool(acquired)) + + try: + assert prototype._dag.delete_session_nodes( + "delete", on_deleted_batch=on_deleted_batch + ) == 1 + assert callback_lock_state == [True] + finally: + prototype.shutdown() + + def test_concurrent_shutdown_releases_one_lease(self, tmp_path): + prototype = self._engine(tmp_path) + clone = prototype.clone_for_agent() + storage = prototype._storage + barrier = threading.Barrier(12) + threads = [ + threading.Thread(target=lambda: (barrier.wait(), prototype.shutdown())) + for _ in range(12) + ] + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert storage._owners == 1 + clone._store.append("live", {"role": "user", "content": "still open"}) + clone.shutdown() + assert storage._closed is True + finally: + prototype.shutdown() + clone.shutdown() + + def test_clone_failure_rolls_back_acquired_lease(self, tmp_path, monkeypatch): + from hermes_lcm.engine import LCMEngine + + prototype = self._engine(tmp_path) + storage = prototype._storage + prototype.raw_context_length = 1_000 + prototype._context_length_source = "test" + monkeypatch.setattr( + LCMEngine, + "_set_context_length", + lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("copy failed")), + ) + try: + with pytest.raises(RuntimeError, match="copy failed"): + prototype.clone_for_agent() + assert storage._owners == 1 + finally: + prototype.shutdown() + assert storage._closed is True + + def test_abandoned_clone_finalizer_releases_lease(self, tmp_path): + import gc + import weakref + + prototype = self._engine(tmp_path) + storage = prototype._storage + clone = prototype.clone_for_agent() + clone_ref = weakref.ref(clone) + del clone + gc.collect() + try: + assert clone_ref() is None + assert storage._owners == 1 + finally: + prototype.shutdown() + assert storage._closed is True + + @staticmethod + def _open_fd_count() -> int: + import os + + for fd_dir in ("/proc/self/fd", "/dev/fd"): + if os.path.isdir(fd_dir): + return len(os.listdir(fd_dir)) + pytest.skip("no file-descriptor listing available on this platform") + + def test_clone_churn_does_not_leak_fds(self, tmp_path): + # Warm up once so lazy imports/caches do not skew the baseline. + warmup = self._engine(tmp_path) + warmup.shutdown() + + baseline = self._open_fd_count() + prototype = self._engine(tmp_path) + clones = [] + try: + with_prototype = self._open_fd_count() + + # Churn: repeatedly create and shut down clones. + for _ in range(10): + churn_clone = prototype.clone_for_agent() + churn_clone.shutdown() + + # Hold live clones concurrently. + clones = [prototype.clone_for_agent() for _ in range(10)] + after_clones = self._open_fd_count() + + # Clones share the prototype's storage bundle: zero new fds. + assert after_clones == with_prototype + finally: + for clone in clones: + clone.shutdown() + prototype.shutdown() + + # Every descriptor opened by the engine family is returned. + assert self._open_fd_count() == baseline + + def test_concurrent_clone_churn_is_thread_safe(self, tmp_path): + prototype = self._engine(tmp_path) + storage = prototype._storage + thread_count = 10 + barrier = threading.Barrier(thread_count) + failures = [] + + def worker(): + try: + barrier.wait() + for _ in range(10): + clone = prototype.clone_for_agent() + clone.shutdown() + except BaseException as exc: + failures.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(thread_count)] + try: + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert failures == [] + # The prototype's lease survived the churn and storage still works. + assert storage._owners == 1 + prototype._store.append("churn", {"role": "user", "content": "still open"}) + finally: + prototype.shutdown() + assert storage._closed is True + + def test_like_fallback_relevance_prefers_multi_term_score_over_single_exact(tmp_path): import hermes_lcm.store as store_module