fix(sdk): serialize sqlite connection lifecycle

Co-authored-by: Cursor <cursoragent@cursor.com>

CHANGELOG.md

@@ -14,6 +14,8 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- Serialized each SDK SQLite store's shared connection lifecycle so concurrent
+  history/cache operations cannot race with `close()` and crash the process.
 - Clarified that `add(metadata=...)` carries legacy SDK processing options,
   while persistent caller business context belongs in `meta`.
 - Added an opt-in `get(..., include_raw=True)` path for reading `L1_RAW`

hy_memory/data/cache_sqlite.py

@@ -7,7 +7,7 @@ Agent Memory V2 - SqliteCache
 设计模式跟随 history_store.py:
 - 标准库 sqlite3 + _run_in_sqlite_pool()
 - WAL 模式 + check_same_thread=False
-- threading.Lock 保护写操作
+- 每实例 threading.RLock 保护共享连接的全部操作与关闭
 
 三张表:
 - memory_operations / pipeline_logs / system_metrics: 审计与观测
@@ -122,7 +122,10 @@ class SqliteCache(CacheBase):
     def __init__(self, config: MemoryConfig):
         self.config = config
         self._conn: Optional[sqlite3.Connection] = None
-        self._lock = threading.Lock()
+        # One connection is shared by the executor workers. Use the same lock
+        # for reads, writes, initialization and close so close() cannot free
+        # the connection while another worker is inside sqlite3.
+        self._lock = threading.RLock()
 
         # 读取 db_path
         cache_cfg = getattr(config, "cache", None)
@@ -144,6 +147,7 @@ class SqliteCache(CacheBase):
         logger.debug(f"SqliteCache initialized: {self._db_path}")
 
     def _init_sync(self) -> None:
+        with self._lock:
             db_dir = os.path.dirname(self._db_path)
             if db_dir:
                 os.makedirs(db_dir, exist_ok=True)
@@ -156,6 +160,7 @@ class SqliteCache(CacheBase):
 
     def _migrate_pipeline_logs_columns(self) -> None:
         """给旧库的 pipeline_logs_v2 补 namespace/session/event time 列。"""
+        with self._lock:
             try:
                 cols = {
                     row["name"]
@@ -180,6 +185,7 @@ class SqliteCache(CacheBase):
         logger.info("SqliteCache closed")
 
     def _close_sync(self) -> None:
+        with self._lock:
             if self._conn:
                 self._conn.close()
                 self._conn = None
@@ -255,6 +261,7 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
+                with self._lock:
                     conditions = []
                     params = []
                     if request_id:
@@ -369,6 +376,7 @@ class SqliteCache(CacheBase):
         """
         try:
             def _query():
+                with self._lock:
                     conditions = []
                     params = []
                     if request_id:
@@ -414,6 +422,7 @@ class SqliteCache(CacheBase):
         """存储一个分钟桶的增量指标数据（UPSERT: 同一分钟多次写入会合并）"""
         try:
             def _store():
+                with self._lock:
                     now = datetime.now().isoformat()
                     # UPSERT: 如果同一分钟已存在，合并数据
                     existing = self._conn.execute(
@@ -442,6 +451,7 @@ class SqliteCache(CacheBase):
         """读取指定时间范围内的分钟指标数据"""
         try:
             def _load():
+                with self._lock:
                     rows = self._conn.execute(
                         "SELECT data FROM system_metrics WHERE minute_ts >= ? AND minute_ts <= ? ORDER BY minute_ts",
                         (start_ts, end_ts),
@@ -462,6 +472,7 @@ class SqliteCache(CacheBase):
         """删除指定时间之前的 metrics 数据"""
         try:
             def _cleanup():
+                with self._lock:
                     self._conn.execute(
                         "DELETE FROM system_metrics WHERE minute_ts < ?",
                         (before_ts,),

hy_memory/data/history_store.py

@@ -16,6 +16,7 @@ import json
 import logging
 import os
 import sqlite3
+import threading
 from datetime import datetime, timezone
 from typing import Dict, List, Any, Optional
 
@@ -89,6 +90,10 @@ class HistoryStore:
         """
         self._db_path = config.history.db_path
         self._conn: Optional[sqlite3.Connection] = None
+        # sqlite3.Connection.close() must never race with an execute() on the
+        # same connection. The lock also serializes operations on this shared
+        # connection; separate HistoryStore instances remain independent.
+        self._conn_lock = threading.RLock()
 
     async def initialize(self) -> None:
         """创建表和索引（如果不存在）"""
@@ -97,6 +102,7 @@ class HistoryStore:
 
     def _init_sync(self) -> None:
         """同步初始化：创建目录、连接数据库、建表建索引"""
+        with self._conn_lock:
             # 确保目录存在
             db_dir = os.path.dirname(self._db_path)
             if db_dir:
@@ -218,6 +224,7 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._get_history_sync, memory_id)
 
     def _get_history_sync(self, memory_id: str) -> List[Dict[str, Any]]:
+        with self._conn_lock:
             if self._conn is None:
                 return []
             cursor = self._conn.execute(
@@ -244,6 +251,7 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._get_recent_sync, isolation_key, limit)
 
     def _get_recent_sync(self, isolation_key: str, limit: int) -> List[Dict[str, Any]]:
+        with self._conn_lock:
             if self._conn is None:
                 return []
             cursor = self._conn.execute(
@@ -266,6 +274,7 @@ class HistoryStore:
         return await _run_in_sqlite_pool(self._count_sync, isolation_key)
 
     def _count_sync(self, isolation_key: str) -> int:
+        with self._conn_lock:
             if self._conn is None:
                 return 0
             if isolation_key:
@@ -284,6 +293,7 @@ class HistoryStore:
             logger.info("[HistoryStore] closed")
 
     def _close_sync(self) -> None:
+        with self._conn_lock:
             if self._conn:
                 self._conn.close()
                 self._conn = None
@@ -339,6 +349,7 @@ class HistoryStore:
         role: str = "",
         extra: Optional[Dict[str, Any]] = None,
     ) -> int:
+        with self._conn_lock:
             if self._conn is None:
                 return -1
             now = _now_iso()

tests/test_sqlite_connection_lifecycle.py

@@ -0,0 +1,136 @@
+import asyncio
+import threading
+
+import pytest
+
+from hy_memory.config import MemoryConfig
+from hy_memory.data.cache_sqlite import SqliteCache
+from hy_memory.data.history_store import HistoryStore
+
+
+class _BlockingCursor:
+    def fetchone(self):
+        return (1,)
+
+    def fetchall(self):
+        return []
+
+
+class _BlockingConnection:
+    """Hold execute() open so a concurrent close() can be observed."""
+
+    def __init__(self):
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.closed = threading.Event()
+
+    def execute(self, *_args, **_kwargs):
+        self.entered.set()
+        if not self.release.wait(timeout=2):
+            raise TimeoutError("test did not release the blocking query")
+        if self.closed.is_set():
+            raise AssertionError("connection was closed while execute() was active")
+        return _BlockingCursor()
+
+    def close(self):
+        self.closed.set()
+
+
+def _config(tmp_path) -> MemoryConfig:
+    config = MemoryConfig()
+    config.history.db_path = str(tmp_path / "history.db")
+    config.cache.db_path = str(tmp_path / "cache.db")
+    return config
+
+
+async def _wait_for(event: threading.Event) -> None:
+    assert await asyncio.wait_for(asyncio.to_thread(event.wait, 1), timeout=2)
+
+
+@pytest.mark.asyncio
+async def test_history_close_waits_for_active_query(tmp_path):
+    store = HistoryStore(_config(tmp_path))
+    connection = _BlockingConnection()
+    store._conn = connection
+
+    query = asyncio.create_task(store.count())
+    await _wait_for(connection.entered)
+    closing = asyncio.create_task(store.close())
+    await asyncio.sleep(0.05)
+
+    assert not connection.closed.is_set()
+    connection.release.set()
+    assert await query == 1
+    await closing
+    assert connection.closed.is_set()
+
+
+@pytest.mark.asyncio
+async def test_cache_close_waits_for_active_query(tmp_path):
+    store = SqliteCache(_config(tmp_path))
+    connection = _BlockingConnection()
+    store._conn = connection
+
+    query = asyncio.create_task(store.load_metrics_range("0", "9"))
+    await _wait_for(connection.entered)
+    closing = asyncio.create_task(store.close())
+    await asyncio.sleep(0.05)
+
+    assert not connection.closed.is_set()
+    connection.release.set()
+    assert await query == []
+    await closing
+    assert connection.closed.is_set()
+
+
+@pytest.mark.asyncio
+async def test_history_concurrent_reads_and_writes_are_serialized(tmp_path):
+    store = HistoryStore(_config(tmp_path))
+    await store.initialize()
+    try:
+        inserted = await asyncio.gather(
+            *(
+                store.record_add(
+                    memory_id=f"memory-{index}",
+                    content="content",
+                    layer="l2_fact",
+                    isolation_key="user",
+                )
+                for index in range(200)
+            )
+        )
+        assert all(row_id > 0 for row_id in inserted)
+        assert len(set(inserted)) == 200
+
+        counts = await asyncio.gather(*(store.count() for _ in range(100)))
+        assert counts == [200] * 100
+    finally:
+        await store.close()
+
+
+@pytest.mark.asyncio
+async def test_cache_concurrent_pipeline_log_reads_and_writes_are_serialized(tmp_path):
+    store = SqliteCache(_config(tmp_path))
+    await store.initialize()
+    try:
+        written = await asyncio.gather(
+            *(
+                store.store_pipeline_log(
+                    request_id="request",
+                    user_id="user",
+                    agent_id="agent",
+                    step="EXTRACT",
+                    prompt=f"prompt-{index}",
+                    response="response",
+                )
+                for index in range(200)
+            )
+        )
+        assert written == [True] * 200
+
+        pages = await asyncio.gather(
+            *(store.get_pipeline_logs(request_id="request", limit=200) for _ in range(50))
+        )
+        assert all(len(page) == 200 for page in pages)
+    finally:
+        await store.close()