fix: share SQLite storage across LCM engine clones

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)))

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)))