fix(memory): restore beam initialization lifecycle

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:

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,10 +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)
-
-        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.
@@ -210,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."""
@@ -221,19 +292,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

tests/test_beam.py

@@ -952,6 +952,52 @@ def test_sleep_consolidated_content_is_recallable(self, temp_db, monkeypatch):
 
 
 class TestMnemosyneIntegration:
+    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__.
+
+        successor = Mnemosyne(session_id="successor", db_path=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)