fix(core): drop native store write gate on HY search (#126)

MemHub expects concurrent native reads during precipitation. The
process-local read slot mostly existed for a Chroma/GIL fail-fast and
was never held by writers on main; activating it (as in #98) would only
push recall onto weaker fallbacks. Remove the lock module and the
hy_memory_store_busy path instead.

Co-authored-by: alvinfei <alvinfei@tencent.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
authored and GitHub committed 2026-10-21 12:15:26 +08:00
1 parent 059e3a4903
commit 2cea39da4f
5 files changed
+22 -106

No files matched your search

@@ -31,7 +31,6 @@ from one_memory.hy_runtime_config import (
diagnostic_log_level,
)
from one_memory.hy_source_context import HySourceContext, HyWriteBindings
from one_memory.hy_store_concurrency import native_store_read_slot
from one_memory.model_profiles import ModelProfileSnapshot, ProviderKind, attach_disable_thinking
from one_memory.model_request_audit import audit_precipitation_request
from one_memory.semantic_deadline import SemanticMemoryHit
@@ -984,27 +983,29 @@ class HyMemoryBridge:
caller could only rank by position — which forced semantic hits to be
placed as a block ahead of everything the keyword lane found. The
score is what lets the two lanes be ranked against each other.
Native search runs without a process-local write gate. MemHub expects
concurrent read during precipitation; store work is a short burst after
network-bound embed/LLM steps, and refusing reads with
``hy_memory_store_busy`` only pushed callers onto weaker fallbacks.
"""
async with native_store_read_slot() as admitted:
if not admitted:
raise HyMemoryBridgeError("hy_memory_store_busy")
client = await self._get_client()
try:
result = await _joined_thread(
client.search,
query,
user_ids=[user_id],
agent_ids=[] if agent_id == "*" else [agent_id],
limit=limit,
min_score=hy_relevance_floor(),
schema_min_score=hy_relevance_floor(),
schema_limit=limit,
intention_limit=limit,
scene="normal",
reader=HY_MEMORY_READER,
)
except Exception as error:
raise HyMemoryBridgeError("hy_memory_search_failed") from error
client = await self._get_client()
try:
result = await _joined_thread(
client.search,
query,
user_ids=[user_id],
agent_ids=[] if agent_id == "*" else [agent_id],
limit=limit,
min_score=hy_relevance_floor(),
schema_min_score=hy_relevance_floor(),
schema_limit=limit,
intention_limit=limit,
scene="normal",
reader=HY_MEMORY_READER,
)
except Exception as error:
raise HyMemoryBridgeError("hy_memory_search_failed") from error
hits = _replied_memories(result, "hy_memory_search_shape_unknown")
found: list[SemanticMemoryHit] = []
seen: set[str] = set()
@@ -1,41 +0,0 @@
"""Process-local coordination for HY's native stores."""
from __future__ import annotations
import asyncio
import weakref
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
_WRITE_LOCKS: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock] = (
weakref.WeakKeyDictionary()
)
def native_store_write_lock() -> asyncio.Lock:
"""Return the write lock for the current event loop."""
loop = asyncio.get_running_loop()
lock = _WRITE_LOCKS.get(loop)
if lock is None:
lock = asyncio.Lock()
_WRITE_LOCKS[loop] = lock
return lock
@asynccontextmanager
async def native_store_read_slot() -> AsyncIterator[bool]:
"""Admit a native HY read only while no write is active.
Chroma's native reader may retain the GIL after an asyncio timeout. Reads
therefore share the physical-store lock, but never wait behind a writer:
host recall can immediately fall back to the governance ledger.
"""
lock = native_store_write_lock()
if lock.locked():
yield False
return
await lock.acquire()
try:
yield True
finally:
lock.release()
@@ -140,8 +140,6 @@ async def search_within(
return (), "failed"
error = pending.exception()
if error is not None:
if str(error) == "hy_memory_store_busy":
return (), "timed_out"
return (), "failed"
found = pending.result()
normalized = normalize_semantic_hits(found)
@@ -436,13 +436,6 @@ def test_unavailable_semantic_result_cannot_discard_lexical_memory(app, monkeypa
assert body["paths"]["memory"]["reason"] == "lexical_only"
def test_busy_hy_store_is_reported_as_a_timed_out_semantic_lane() -> None:
async def busy(_query: str) -> tuple[str, ...]:
raise RuntimeError("hy_memory_store_busy")
assert asyncio.run(search_within(busy, "q", 1.0)) == ((), "timed_out")
def test_context_diagnostics_report_each_lane_without_changing_v1(app, monkeypatch): # noqa: F811
from one_memory import context_routes
@@ -6,7 +6,6 @@ import sqlite3
import threading
import time
from collections.abc import Mapping, Sequence
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
@@ -1199,40 +1198,6 @@ def test_bridge_preserves_sdk_formatted_search_content(tmp_path) -> None:
assert hits[0].formatting_complete is True
def test_semantic_search_skips_native_store_while_a_write_is_active(tmp_path, monkeypatch) -> None:
client = FakeHyMemoryClient(
{"success": True},
search_result={"memories": [{"memory_id": "must-not-be-read"}]},
)
bridge = HyMemoryBridge(
profile=snapshot(
ProviderKind.OLLAMA,
base_url="http://127.0.0.1:11434",
has_api_key=False,
),
api_key=None,
data_dir=tmp_path,
client=client,
)
@asynccontextmanager
async def busy_store():
yield False
monkeypatch.setattr("one_memory.hy_memory_bridge.native_store_read_slot", busy_store)
with pytest.raises(HyMemoryBridgeError, match="hy_memory_store_busy"):
asyncio.run(
bridge.search_ids(
"导入期间查询",
user_id="local",
agent_id="*",
)
)
assert client.search_calls == []
def test_search_ids_does_not_create_nodes_tasks_or_collections(tmp_path) -> None:
"""Search is a read. Creating store objects here would invent a graph.