From 2cea39da4f9f536e81122bda0ec299e0fd5bc0e7 Mon Sep 17 00:00:00 2001 From: feitianxiang Date: Wed, 21 Oct 2026 12:15:26 +0800 Subject: [PATCH] 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 Co-authored-by: Cursor --- .../core/src/one_memory/hy_memory_bridge.py | 43 ++++++++++--------- .../src/one_memory/hy_store_concurrency.py | 41 ------------------ .../core/src/one_memory/semantic_deadline.py | 2 - .../core/tests/test_context_hy_primary.py | 7 --- .../apps/core/tests/test_hy_memory_bridge.py | 35 --------------- 5 files changed, 22 insertions(+), 106 deletions(-) delete mode 100644 memhub/apps/core/src/one_memory/hy_store_concurrency.py diff --git a/memhub/apps/core/src/one_memory/hy_memory_bridge.py b/memhub/apps/core/src/one_memory/hy_memory_bridge.py index e1d34ea..ab600b4 100644 --- a/memhub/apps/core/src/one_memory/hy_memory_bridge.py +++ b/memhub/apps/core/src/one_memory/hy_memory_bridge.py @@ -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() diff --git a/memhub/apps/core/src/one_memory/hy_store_concurrency.py b/memhub/apps/core/src/one_memory/hy_store_concurrency.py deleted file mode 100644 index d996926..0000000 --- a/memhub/apps/core/src/one_memory/hy_store_concurrency.py +++ /dev/null @@ -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() diff --git a/memhub/apps/core/src/one_memory/semantic_deadline.py b/memhub/apps/core/src/one_memory/semantic_deadline.py index 6532706..1b2f883 100644 --- a/memhub/apps/core/src/one_memory/semantic_deadline.py +++ b/memhub/apps/core/src/one_memory/semantic_deadline.py @@ -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) diff --git a/memhub/apps/core/tests/test_context_hy_primary.py b/memhub/apps/core/tests/test_context_hy_primary.py index 56371ed..3df9bd4 100644 --- a/memhub/apps/core/tests/test_context_hy_primary.py +++ b/memhub/apps/core/tests/test_context_hy_primary.py @@ -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 diff --git a/memhub/apps/core/tests/test_hy_memory_bridge.py b/memhub/apps/core/tests/test_hy_memory_bridge.py index 5c1213c..647372d 100644 --- a/memhub/apps/core/tests/test_hy_memory_bridge.py +++ b/memhub/apps/core/tests/test_hy_memory_bridge.py @@ -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.