From 30372603fbc0d3aa63e3b6972b5e062b30931499 Mon Sep 17 00:00:00 2001 From: Denis H Date: Wed, 15 Jul 2026 23:11:33 +0200 Subject: [PATCH 1/2] fix(memory): restore beam initialization lifecycle --- mnemosyne/core/memory.py | 26 +++++++++++++------------- tests/test_beam.py | 13 +++++++++++++ 2 files changed, 26 insertions(+), 13 deletions(-) diff --git a/mnemosyne/core/memory.py b/mnemosyne/core/memory.py index 75e7dfd6..0d4f9f31 100644 --- a/mnemosyne/core/memory.py +++ b/mnemosyne/core/memory.py @@ -199,6 +199,19 @@ def __init__(self, session_id: str = "default", db_path: Path = None, bank: str self.conn = _get_connection(self.db_path) init_db(self.db_path) + # Phase 8: Streaming + Patterns + Plugins (lazy init) + self._stream = None + self._compressor = None + self._pattern_detector = None + self._delta_sync = None + self._plugin_manager = None + + # Create beam with streaming emitter wired + self.beam = BeamMemory(session_id=session_id, db_path=self.db_path, + author_id=author_id, author_type=author_type, + channel_id=channel_id, + event_emitter=self._stream_emit) + self._closed = False def close(self) -> None: @@ -221,19 +234,6 @@ def __del__(self): except Exception: pass - # Phase 8: Streaming + Patterns + Plugins (lazy init) - self._stream = None - self._compressor = None - self._pattern_detector = None - self._delta_sync = None - self._plugin_manager = None - - # Create beam with streaming emitter wired - self.beam = BeamMemory(session_id=session_id, db_path=self.db_path, - author_id=author_id, author_type=author_type, - channel_id=channel_id, - event_emitter=self._stream_emit) - # ─── Phase 8: Streaming ───────────────────────────────────────── @property diff --git a/tests/test_beam.py b/tests/test_beam.py index c5a091d2..22856dfc 100644 --- a/tests/test_beam.py +++ b/tests/test_beam.py @@ -952,6 +952,19 @@ def test_sleep_consolidated_content_is_recallable(self, temp_db, monkeypatch): class TestMnemosyneIntegration: + def test_constructor_wires_beam_before_close(self, temp_db): + """A close lifecycle must not move BEAM initialization into __del__.""" + mem = Mnemosyne(session_id="lifecycle", db_path=temp_db) + assert isinstance(mem.beam, BeamMemory) + assert mem.beam._event_emitter == mem._stream_emit + + mem.close() + mem.close() # Idempotent cleanup is safe for callers and __del__. + + successor = Mnemosyne(session_id="successor", db_path=temp_db) + assert isinstance(successor.beam, BeamMemory) + successor.close() + def test_legacy_and_beam_dual_write(self, temp_db): mem = Mnemosyne(session_id="s2", db_path=temp_db) mid = mem.remember("Likes pizza", source="preference", importance=0.8) From 5d6da2c03269086679cb4ca6b383cad9e2bea8ad Mon Sep 17 00:00:00 2001 From: Denis H Date: Thu, 16 Jul 2026 00:45:57 +0200 Subject: [PATCH 2/2] fix(memory): retain shared connections across live instances --- mnemosyne/core/beam.py | 13 ++++- mnemosyne/core/memory.py | 100 +++++++++++++++++++++++++++++++-------- tests/test_beam.py | 35 +++++++++++++- 3 files changed, 124 insertions(+), 24 deletions(-) diff --git a/mnemosyne/core/beam.py b/mnemosyne/core/beam.py index 48bd2148..4043af2f 100644 --- a/mnemosyne/core/beam.py +++ b/mnemosyne/core/beam.py @@ -487,15 +487,24 @@ def _get_connection(db_path: Path = None) -> sqlite3.Connection: return _thread_local.conn -def _close_beam_connection() -> None: +def _close_beam_connection(db_path: Optional[Path] = None) -> None: """Close the thread-local BEAM connection and checkpoint the WAL. Thread-local connections under WAL mode can block checkpoints after the owning thread exits. Explicitly closing the connection and running ``PRAGMA wal_checkpoint(TRUNCATE)`` prevents the ``database is locked`` cascade documented in #382. + + When ``db_path`` is provided, leave a thread-local connection for a + different database alone. This lets Mnemosyne release a closed bank + without invalidating a newer instance that selected another bank. """ - if hasattr(_thread_local, 'conn') and _thread_local.conn is not None: + expected_path = str(Path(db_path)) if db_path is not None else None + if ( + hasattr(_thread_local, 'conn') + and _thread_local.conn is not None + and (expected_path is None or getattr(_thread_local, 'db_path', None) == expected_path) + ): try: _thread_local.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception: diff --git a/mnemosyne/core/memory.py b/mnemosyne/core/memory.py index 0d4f9f31..2a5130ad 100644 --- a/mnemosyne/core/memory.py +++ b/mnemosyne/core/memory.py @@ -27,6 +27,8 @@ from mnemosyne.core import embeddings as _embeddings from mnemosyne.core.beam import BeamMemory, init_beam _thread_local = threading.local() +_instance_connection_lock = threading.Lock() +_instance_connection_refcounts: Dict[tuple, int] = {} # Default data directory # NOTE: On Fly.io and ephemeral VMs, only ~/.hermes is persisted. @@ -77,15 +79,25 @@ def _get_connection(db_path = None) -> sqlite3.Connection: return _thread_local.conn -def _close_connection() -> None: +def _close_connection(db_path: Optional[Path] = None) -> None: """Close the thread-local connection and checkpoint the WAL. Thread-local connections under WAL mode can block checkpoints after the owning thread exits. Explicitly closing the connection and running ``PRAGMA wal_checkpoint(TRUNCATE)`` prevents the ``database is locked`` cascade documented in #382. + + When ``db_path`` is provided, leave a thread-local connection for a + different database alone. A thread can switch banks between live + Mnemosyne instances, so closing an unrelated cached connection would + invalidate the newer instance. """ - if hasattr(_thread_local, 'conn') and _thread_local.conn is not None: + expected_path = str(Path(db_path)) if db_path is not None else None + if ( + hasattr(_thread_local, 'conn') + and _thread_local.conn is not None + and (expected_path is None or getattr(_thread_local, 'db_path', None) == expected_path) + ): try: _thread_local.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception: @@ -98,6 +110,30 @@ def _close_connection() -> None: _thread_local.db_path = None +def _connection_owner_key(db_path: Path) -> tuple: + """Return the per-thread key for Mnemosyne-owned shared connections.""" + return threading.get_ident(), str(Path(db_path)) + + +def _acquire_instance_connections(db_path: Path) -> tuple: + """Register one live Mnemosyne owner for the current thread and database.""" + key = _connection_owner_key(db_path) + with _instance_connection_lock: + _instance_connection_refcounts[key] = _instance_connection_refcounts.get(key, 0) + 1 + return key + + +def _release_instance_connections(key: tuple) -> bool: + """Release an owner and return whether it was the last one.""" + with _instance_connection_lock: + remaining = _instance_connection_refcounts.get(key, 0) - 1 + if remaining <= 0: + _instance_connection_refcounts.pop(key, None) + return True + _instance_connection_refcounts[key] = remaining + return False + + def wal_checkpoint(db_path=None) -> dict: """Run a WAL checkpoint on the thread-local or specified connection. @@ -177,6 +213,11 @@ class Mnemosyne: def __init__(self, session_id: str = "default", db_path: Path = None, bank: str = None, author_id: str = None, author_type: str = None, channel_id: str = None): + # Set lifecycle state before allocating BEAM resources so a partially + # constructed instance can be cleaned up safely by __del__. + self._closed = False + self._instance_connection_owner_key = None + # Auto-seed config.yaml on first Mnemosyne init from mnemosyne.core.config import get_config get_config() # triggers _seed() if config.yaml doesn't exist @@ -196,23 +237,35 @@ def __init__(self, session_id: str = "default", db_path: Path = None, bank: str else: self.db_path = _default_db_path() - self.conn = _get_connection(self.db_path) - init_db(self.db_path) - - # Phase 8: Streaming + Patterns + Plugins (lazy init) - self._stream = None - self._compressor = None - self._pattern_detector = None - self._delta_sync = None - self._plugin_manager = None - - # Create beam with streaming emitter wired - self.beam = BeamMemory(session_id=session_id, db_path=self.db_path, - author_id=author_id, author_type=author_type, - channel_id=channel_id, - event_emitter=self._stream_emit) - - self._closed = False + # Mnemosyne instances in one thread share both module-level SQLite + # connections. Keep a small ownership count so closing one wrapper + # cannot close the BeamMemory connection another live wrapper needs. + self._instance_connection_owner_key = _acquire_instance_connections(self.db_path) + try: + self.conn = _get_connection(self.db_path) + init_db(self.db_path) + + # Phase 8: Streaming + Patterns + Plugins (lazy init) + self._stream = None + self._compressor = None + self._pattern_detector = None + self._delta_sync = None + self._plugin_manager = None + + # Create beam with streaming emitter wired + self.beam = BeamMemory(session_id=session_id, db_path=self.db_path, + author_id=author_id, author_type=author_type, + channel_id=channel_id, + event_emitter=self._stream_emit) + except Exception: + # BeamMemory can open its own thread-local connection before a + # later constructor step fails. Release this provisional owner and + # close both resources only if no live Mnemosyne shares them. + try: + self.close() + except Exception: + pass + raise def close(self) -> None: """Close the database connection and checkpoint the WAL. @@ -223,9 +276,14 @@ def close(self) -> None: if self._closed: return self._closed = True - _close_connection() + owner_key = self._instance_connection_owner_key + self._instance_connection_owner_key = None + if owner_key is None or not _release_instance_connections(owner_key): + return + + _close_connection(self.db_path) from mnemosyne.core.beam import _close_beam_connection - _close_beam_connection() + _close_beam_connection(self.db_path) def __del__(self): """Best-effort cleanup on garbage collection.""" diff --git a/tests/test_beam.py b/tests/test_beam.py index 22856dfc..65830ec4 100644 --- a/tests/test_beam.py +++ b/tests/test_beam.py @@ -952,12 +952,20 @@ def test_sleep_consolidated_content_is_recallable(self, temp_db, monkeypatch): class TestMnemosyneIntegration: - def test_constructor_wires_beam_before_close(self, temp_db): + def test_constructor_wires_beam_before_close(self, temp_db, monkeypatch): """A close lifecycle must not move BEAM initialization into __del__.""" mem = Mnemosyne(session_id="lifecycle", db_path=temp_db) assert isinstance(mem.beam, BeamMemory) assert mem.beam._event_emitter == mem._stream_emit + delivered = [] + monkeypatch.setattr(mem.stream, "emit", delivered.append) + mem.enable_streaming() + event = object() + assert mem.beam._event_emitter is not None + mem.beam._event_emitter(event) + assert delivered == [event] + mem.close() mem.close() # Idempotent cleanup is safe for callers and __del__. @@ -965,6 +973,31 @@ def test_constructor_wires_beam_before_close(self, temp_db): assert isinstance(successor.beam, BeamMemory) successor.close() + def test_close_keeps_shared_beam_connection_for_live_instance(self, temp_db): + first = Mnemosyne(session_id="first", db_path=temp_db) + second = Mnemosyne(session_id="second", db_path=temp_db) + + first.close() + assert second.beam.conn.execute("SELECT 1").fetchone()[0] == 1 + second.close() + + def test_constructor_failure_releases_provisional_connection(self, temp_db, monkeypatch): + import mnemosyne.core.memory as memory_module + + real_beam = memory_module.BeamMemory + + def fail_beam(*args, **kwargs): + raise RuntimeError("beam construction failed") + + monkeypatch.setattr(memory_module, "BeamMemory", fail_beam) + with pytest.raises(RuntimeError, match="beam construction failed"): + Mnemosyne(session_id="broken", db_path=temp_db) + + monkeypatch.setattr(memory_module, "BeamMemory", real_beam) + recovered = Mnemosyne(session_id="recovered", db_path=temp_db) + assert isinstance(recovered.beam, BeamMemory) + recovered.close() + def test_legacy_and_beam_dual_write(self, temp_db): mem = Mnemosyne(session_id="s2", db_path=temp_db) mid = mem.remember("Likes pizza", source="preference", importance=0.8)