fix: keep local Qdrant on an owner thread

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

hy_memory/data/vector_store_qdrant.py

@@ -16,11 +16,10 @@ Agent Memory V2 - VectorStore (Qdrant)
 - 简化运维，后续可按需拆分
 """
 
-import asyncio
 import concurrent.futures
 import os
 import threading
-from typing import Optional, List, Dict, Any, Tuple
+from typing import Optional, List, Dict, Any, Tuple, Callable
 from pathlib import Path
 import logging
 
@@ -30,12 +29,10 @@ from ..models.memory import (
 from ..config import MemoryConfig
 from ..utils.async_executor import run_in_executor_guarded
 from .vector_store_base import VectorStoreBase
+from .vector_store_chroma import _vdb_executor
 
 logger = logging.getLogger(__name__)
 
-# VDB 独立线程池（与 Chroma 共享同一策略）
-from .vector_store_chroma import _vdb_executor
-
 
 def _run_in_vdb_pool(func, *args, **kwargs):
     """在 VDB 独立线程池中执行同步函数"""
@@ -44,6 +41,70 @@ def _run_in_vdb_pool(func, *args, **kwargs):
     return run_in_executor_guarded(_vdb_executor, call)
 
 
+class _LocalQdrantOwner:
+    """Keep one local Qdrant client and its SQLite state on one owner thread.
+
+    Local Qdrant may enable SQLite's ``check_same_thread`` on environments
+    compiled with ``THREADSAFE=2``. The public async surface still uses the
+    shared VDB pool, but every actual client access is forwarded to this
+    instance's single owner thread.
+    """
+
+    def __init__(self, factory: Callable[[], Any]):
+        self._executor = concurrent.futures.ThreadPoolExecutor(
+            max_workers=1,
+            thread_name_prefix="qdrant-local-owner",
+        )
+        self._lifecycle_lock = threading.Lock()
+        self._closed = False
+        try:
+            self._client = self._executor.submit(factory).result()
+        except BaseException:
+            self._closed = True
+            self._executor.shutdown(wait=True, cancel_futures=True)
+            raise
+
+    def _submit(self, callback: Callable[[], Any]) -> Any:
+        with self._lifecycle_lock:
+            if self._closed:
+                raise RuntimeError("Local Qdrant client is closed")
+            future = self._executor.submit(callback)
+        return future.result()
+
+    def __getattr__(self, name: str) -> Any:
+        is_callable, value = self._submit(
+            lambda: (
+                callable(attribute := getattr(self._client, name)),
+                None if callable(attribute) else attribute,
+            )
+        )
+        if not is_callable:
+            return value
+
+        def invoke(*args: Any, **kwargs: Any) -> Any:
+            return self._submit(
+                lambda: getattr(self._client, name)(*args, **kwargs)
+            )
+
+        return invoke
+
+    def close(self) -> None:
+        def close_owned_client() -> None:
+            client, self._client = self._client, None
+            if client is not None:
+                client.close()
+
+        with self._lifecycle_lock:
+            if self._closed:
+                return
+            self._closed = True
+            future = self._executor.submit(close_owned_client)
+        try:
+            future.result()
+        finally:
+            self._executor.shutdown(wait=True, cancel_futures=True)
+
+
 class QdrantVectorStore(VectorStoreBase):
     """
     Qdrant 向量存储实现
@@ -113,8 +174,8 @@ class QdrantVectorStore(VectorStoreBase):
             self._is_local = True
             # 兼容性处理: 清理高版本 qdrant-client 写入的 meta.json 中的未知字段
             self._patch_local_meta(vs_config.persist_directory)
-            self._client = QdrantClient(
-                path=vs_config.persist_directory,
+            self._client = _LocalQdrantOwner(
+                lambda: QdrantClient(path=vs_config.persist_directory)
             )
 
         # 确保集合存在（同步调用，在 to_thread 中执行）
@@ -217,7 +278,13 @@ class QdrantVectorStore(VectorStoreBase):
                     has_sparse = False
             return has_sparse
 
-        has_sparse = await _run_in_vdb_pool(_ensure_collection)
+        try:
+            has_sparse = await _run_in_vdb_pool(_ensure_collection)
+        except BaseException:
+            client, self._client = self._client, None
+            if client is not None:
+                await _run_in_vdb_pool(client.close)
+            raise
         self._supports_fulltext = bool(has_sparse and bm25_fastembed.is_available())
         logger.info(
             f"[qdrant] fulltext(BM25 sparse)={'on' if self._supports_fulltext else 'off'}"
@@ -1043,7 +1110,10 @@ class QdrantVectorStore(VectorStoreBase):
                 FieldCondition(key="status", match=MatchAny(any=status_values))
             )
         if layers:
-            layer_values = [l.value if hasattr(l, 'value') else str(l) for l in layers]
+            layer_values = [
+                layer.value if hasattr(layer, "value") else str(layer)
+                for layer in layers
+            ]
             if not allow_raw_layer:
                 layer_values = [v for v in layer_values if v != MemoryLayer.L1_RAW.value]
             if layer_values:
@@ -1372,9 +1442,9 @@ class QdrantVectorStore(VectorStoreBase):
 
     async def close(self) -> None:
         """关闭客户端"""
-        if self._client:
-            self._client.close()
-            self._client = None
+        client, self._client = self._client, None
+        if client is not None:
+            await _run_in_vdb_pool(client.close)
 
     # ================================================================
     # Keyword Search (Full-text)
@@ -1412,8 +1482,6 @@ class QdrantVectorStore(VectorStoreBase):
             except Exception as e:
                 logger.debug(f"[qdrant] inspect search_text index failed (will try create anyway): {e}")
 
-            target_tokenizer_value = getattr(TokenizerType.WHITESPACE, "value", "whitespace")
-
             if existing_tokenizer is not None:
                 if str(existing_tokenizer).lower().endswith("whitespace") or str(existing_tokenizer).lower() == "whitespace":
                     return  # 已经是新格式，啥都不做
@@ -1436,7 +1504,9 @@ class QdrantVectorStore(VectorStoreBase):
                     max_token_len=20,
                 ),
             )
-            logger.info(f"[qdrant] Created search_text text index (whitespace) on existing collection")
+            logger.info(
+                "[qdrant] Created search_text text index (whitespace) on existing collection"
+            )
         except Exception as e:
             # 兜底：未支持或并发竞争，静默
             logger.debug(f"[qdrant] _ensure_text_index swallowed: {e}")
@@ -1510,7 +1580,10 @@ class QdrantVectorStore(VectorStoreBase):
             )
         if layers:
             must_conditions.append(
-                FieldCondition(key="layer", match=MatchAny(any=[l.value for l in layers]))
+                FieldCondition(
+                    key="layer",
+                    match=MatchAny(any=[layer.value for layer in layers]),
+                )
             )
         if status_filter:
             must_conditions.append(
@@ -1729,7 +1802,7 @@ class QdrantVectorStore(VectorStoreBase):
 
         if changed:
             meta_path.write_text(_json.dumps(meta), encoding="utf-8")
-            logger.info(f"Patched Qdrant meta.json: removed unsupported fields")
+            logger.info("Patched Qdrant meta.json: removed unsupported fields")
 
     # ================================================================
     # Tag Index（per-user tag embedding，供 reader_hybrid_tag 使用）

hy_memory/utils/async_executor.py

@@ -54,6 +54,8 @@ def concurrent_result_guarded(
         try:
             return future.result(timeout=interval)
         except concurrent.futures.TimeoutError:
+            if future.done():
+                return future.result()
             target_loop.call_soon_threadsafe(lambda: None)
 
 
@@ -70,4 +72,6 @@ async def await_concurrent_future_guarded(
         try:
             return await asyncio.wait_for(asyncio.shield(wrapped), timeout=interval)
         except asyncio.TimeoutError:
+            if wrapped.done():
+                return wrapped.result()
             target_loop.call_soon_threadsafe(lambda: None)

tests/test_async_executor.py

@@ -0,0 +1,100 @@
+"""Completed executor tasks must return or propagate their own exceptions."""
+
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize(
+    "outcome",
+    ["success", "value_error", "builtin_timeout", "asyncio_timeout",
+     "concurrent_timeout", "cancelled"],
+)
+def test_guarded_future_completion(mode, outcome):
+    # A regression can loop forever on an already-failed Future. Isolate each
+    # case so the parent can stop it without leaving a busy thread in pytest.
+    script = textwrap.dedent(
+        """
+        import asyncio
+        import concurrent.futures
+        import sys
+        import time
+
+        from hy_memory.utils.async_executor import (
+            await_concurrent_future_guarded,
+            concurrent_result_guarded,
+        )
+
+        mode, outcome = sys.argv[1:]
+        errors = {
+            "value_error": ValueError,
+            "builtin_timeout": TimeoutError,
+            "asyncio_timeout": asyncio.TimeoutError,
+            "concurrent_timeout": concurrent.futures.TimeoutError,
+        }
+
+        def work():
+            # Exceed several guard intervals: a wait timeout must not cancel
+            # or resubmit the underlying task.
+            time.sleep(0.08)
+            if outcome in errors:
+                raise errors[outcome]("task failed")
+            return "completed"
+
+        async def check():
+            loop = asyncio.get_running_loop()
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+                if outcome == "cancelled":
+                    future = concurrent.futures.Future()
+                    assert future.cancel()
+                else:
+                    future = pool.submit(work)
+                try:
+                    if mode == "sync":
+                        result = concurrent_result_guarded(
+                            future, loop, guard_seconds=0.01,
+                        )
+                    else:
+                        result = await await_concurrent_future_guarded(
+                            future, loop, guard_seconds=0.01,
+                        )
+                except BaseException as exc:
+                    if outcome == "cancelled":
+                        expected = (concurrent.futures.CancelledError
+                                    if mode == "sync" else asyncio.CancelledError)
+                        assert isinstance(exc, expected), repr(exc)
+                    else:
+                        assert outcome in errors, repr(exc)
+                        # wrap_future translates concurrent.futures.TimeoutError
+                        # to asyncio.TimeoutError (distinct before Python 3.11).
+                        expected = (asyncio.TimeoutError
+                                    if mode == "async" and outcome == "concurrent_timeout"
+                                    else errors[outcome])
+                        assert isinstance(exc, expected), repr(exc)
+                        assert str(exc) == "task failed"
+                else:
+                    assert outcome == "success"
+                    assert result == "completed"
+
+        asyncio.run(check())
+        """
+    )
+    process = subprocess.Popen(
+        [sys.executable, "-c", script, mode, outcome],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    try:
+        stdout, stderr = process.communicate(timeout=10)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        stdout, stderr = process.communicate()
+        pytest.fail(
+            f"{mode}/{outcome}: completed task did not reach caller; "
+            f"stdout={stdout!r}, stderr={stderr!r}"
+        )
+    assert process.returncode == 0, stderr

tests/test_qdrant_local_thread_owner.py

@@ -0,0 +1,83 @@
+"""Local Qdrant must keep its SQLite connection on one owner thread."""
+
+import asyncio
+
+import pytest
+
+qdrant_client = pytest.importorskip("qdrant_client")
+
+from qdrant_client import QdrantClient, models  # noqa: E402
+from qdrant_client.local.persistence import CollectionPersistence  # noqa: E402
+
+from hy_memory.config import MemoryConfig, VectorStoreConfig  # noqa: E402
+from hy_memory.data.vector_store import create_vector_store  # noqa: E402
+from hy_memory.models.memory import MemoryLayer, MemoryNode  # noqa: E402
+
+
+def _config(directory, name="thread_probe"):
+    config = MemoryConfig()
+    config.vector_store = VectorStoreConfig(
+        provider="qdrant",
+        persist_directory=str(directory),
+        collection_name=name,
+        embedding_dims=4,
+    )
+    return config
+
+
+def _node(index=0):
+    return MemoryNode(
+        node_id=f"test-node-{index}",
+        user_id="test-user",
+        agent_id="test-agent",
+        session_id="test-session",
+        layer=MemoryLayer.L2_FACT,
+        content=f"Offline owner-thread test {index}",
+        embedding=[1.0, 0.0, 0.0, float(index)],
+    )
+
+
+@pytest.fixture
+def same_thread_qdrant(monkeypatch):
+    monkeypatch.setattr(CollectionPersistence, "CHECK_SAME_THREAD", True)
+    monkeypatch.setattr(
+        "hy_memory.pipelines._retrieval.bm25_fastembed.is_available",
+        lambda: False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_reopened_local_collection_writes_with_same_thread_check(
+    tmp_path,
+    same_thread_qdrant,
+):
+    client = QdrantClient(path=str(tmp_path))
+    client.create_collection(
+        "thread_probe_4",
+        vectors_config=models.VectorParams(
+            size=4,
+            distance=models.Distance.COSINE,
+        ),
+    )
+    client.close()
+
+    store = create_vector_store(_config(tmp_path))
+    await store.initialize()
+    try:
+        assert await store.upsert(_node()) == "test-node-0"
+    finally:
+        await store.close()
+
+
+@pytest.mark.asyncio
+async def test_fresh_local_collection_serializes_concurrent_writes_and_close(
+    tmp_path,
+    same_thread_qdrant,
+):
+    store = create_vector_store(_config(tmp_path))
+    await store.initialize()
+    try:
+        ids = await asyncio.gather(*(store.upsert(_node(index)) for index in range(8)))
+        assert ids == [f"test-node-{index}" for index in range(8)]
+    finally:
+        await store.close()