Compare commits

...
Author SHA1 Message Date
alvinfeiandCursor 1821a31295 fix(core): drop native store write gate on HY search
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: Cursor <cursoragent@cursor.com>
2026-10-21 12:07:37 +08:00
5 changed files with 22 additions and 106 deletions

No files matched your search

@@ -31,7 +31,6 @@ from one_memory.hy_runtime_config import (
diagnostic_log_level, diagnostic_log_level,
) )
from one_memory.hy_source_context import HySourceContext, HyWriteBindings 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_profiles import ModelProfileSnapshot, ProviderKind, attach_disable_thinking
from one_memory.model_request_audit import audit_precipitation_request from one_memory.model_request_audit import audit_precipitation_request
from one_memory.semantic_deadline import SemanticMemoryHit 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 caller could only rank by position — which forced semantic hits to be
placed as a block ahead of everything the keyword lane found. The placed as a block ahead of everything the keyword lane found. The
score is what lets the two lanes be ranked against each other. 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: client = await self._get_client()
if not admitted: try:
raise HyMemoryBridgeError("hy_memory_store_busy") result = await _joined_thread(
client = await self._get_client() client.search,
try: query,
result = await _joined_thread( user_ids=[user_id],
client.search, agent_ids=[] if agent_id == "*" else [agent_id],
query, limit=limit,
user_ids=[user_id], min_score=hy_relevance_floor(),
agent_ids=[] if agent_id == "*" else [agent_id], schema_min_score=hy_relevance_floor(),
limit=limit, schema_limit=limit,
min_score=hy_relevance_floor(), intention_limit=limit,
schema_min_score=hy_relevance_floor(), scene="normal",
schema_limit=limit, reader=HY_MEMORY_READER,
intention_limit=limit, )
scene="normal", except Exception as error:
reader=HY_MEMORY_READER, raise HyMemoryBridgeError("hy_memory_search_failed") from error
)
except Exception as error:
raise HyMemoryBridgeError("hy_memory_search_failed") from error
hits = _replied_memories(result, "hy_memory_search_shape_unknown") hits = _replied_memories(result, "hy_memory_search_shape_unknown")
found: list[SemanticMemoryHit] = [] found: list[SemanticMemoryHit] = []
seen: set[str] = set() 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" return (), "failed"
error = pending.exception() error = pending.exception()
if error is not None: if error is not None:
if str(error) == "hy_memory_store_busy":
return (), "timed_out"
return (), "failed" return (), "failed"
found = pending.result() found = pending.result()
normalized = normalize_semantic_hits(found) 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" 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 def test_context_diagnostics_report_each_lane_without_changing_v1(app, monkeypatch): # noqa: F811
from one_memory import context_routes from one_memory import context_routes
@@ -6,7 +6,6 @@ import sqlite3
import threading import threading
import time import time
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
from contextlib import asynccontextmanager
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from types import SimpleNamespace 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 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: 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. """Search is a read. Creating store objects here would invent a graph.