From edb5ab00e285017f288e2c415b9cf17c541d86fa Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 03:48:26 +1200 Subject: [PATCH 1/5] Bound memU embedding calls: a hung provider no longer stalls memory Every memU call that produces an embedding runs on the "embedding" LLM profile, and that profile was bounded by nothing: not by the bridge's asyncio.wait_for layer, not by an httpx transport timeout, not by any caller. _instrument_llm_timeouts iterates ("memorize", "fast", "default") only, so the embedding client fell through to the bare OpenAI SDK default (600s read, 2 retries), i.e. up to ~30 minutes per call. Every other network call in the file is bounded explicitly, and _LLM_CALL_TIMEOUT's own comment already claims to cover embed. Reachable carriers: category seeding during agent init (inside initialize(), whose return engine.py discards), recall() (the query embedding is the first thing the RAG retrieve pipeline does), and the runtime category create/update paths reached from the web UI and the category_update tool. The bound goes on the client rather than at the call sites, because that is where this file already puts per-call bounds and because it covers the 16 memU-internal embed() sites nerve does not own. Ordering matters: the seed site embeds ~185 lines before _instrument_llm_timeouts() runs, so the client is also instrumented right after the service is constructed, before availability is published. That call is deliberately uncaught - _initialize_impl's existing handler turns a failure into an unavailable bridge rather than one advertising itself as usable with an unbounded client. The reset-path call stays best-effort so an embedding failure cannot break chat re-instrumentation. SDK max_retries is left at its default here, unlike the chat profile: the chat path can disable retries because _timeout_chat supplies its own transient ladder, which is chat-only. With retries kept, the outer wait_for still fires at the bound (measured overshoot +0.00s). Two adjacent gaps the bound would otherwise have made worse: - recall() returned [] for a timeout, which renders as "Recalled 0 memories". _is_transient_llm_error keys on HTTP status and matches no timeout type, so bounding the call would have converted a stall into a confident wrong answer. recall now reports a timeout as backend-down. The base already had this hole for SDK-level embed timeouts. - _reset_llm_clients_impl evicted memorize/fast/default only, so an embed timeout recycled three unrelated chat clients and kept the offending transport. Pre-existing; this change makes that path materially more reachable, so "embedding" is added to the eviction tuple. The post-reset warmup tuple is deliberately NOT changed - warming an embedding client means a real billed request on every reset. Validation: 13 new arms in TestEmbeddingCallTimeout, no arm touching a real endpoint (hangs await an Event that is never set). Both directions: 11 failed / 2 passed against the unfixed source with the new tests present, 13 passed with the fix. Full suite 2933 -> 2946 passed with a byte-identical 7-name pre-existing failure set. 13-mutant matrix, all 13 killed by their predicted arms, unmutated control green at both ends. --- nerve/memory/memu_bridge.py | 83 ++++++++- tests/test_memu_bridge.py | 327 ++++++++++++++++++++++++++++++++++++ 2 files changed, 407 insertions(+), 3 deletions(-) diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 9642175b..be776537 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -1564,6 +1564,14 @@ async def _initialize_impl(self) -> bool: if not self.config.openai_api_key else {}), }, ) + # Bound the embedding client BEFORE publishing availability, and + # therefore before _ensure_categories() embeds below: the regular + # _instrument_llm_timeouts() call happens much later. Uncaught on + # purpose: a failure here rides the outer handler and leaves the + # bridge unavailable rather than advertising it as usable with an + # unbounded embedding client. + self._instrument_embedding_timeout() + self._available = True self._metrics.service_available = True self._metrics.initialized_at = datetime.now(timezone.utc).isoformat() @@ -2081,6 +2089,69 @@ async def _timeout_chat( except Exception as e: logger.warning("Could not configure LLM client %s: %s", profile, e) + # The loop above cannot cover "embedding": its body wraps .chat and + # sets max_retries=0, neither of which is right for embeddings. + # Best-effort here so an embedding failure can't break chat. + try: + self._instrument_embedding_timeout() + except Exception as e: + logger.warning("Could not configure embedding LLM client: %s", e) + + def _instrument_embedding_timeout(self) -> None: + """Bound calls on the "embedding" LLM profile (same two layers). + + The ``_instrument_llm_timeouts`` loop deliberately excludes this + profile: its Layer 2 wraps ``.chat`` (embeddings go through + ``.embed``) and its Layer 1 sets ``max_retries = 0``, which is only + safe for chat because ``_timeout_chat`` supplies its own transient + retry ladder. Without this method every ``embed()`` call runs under + the bare OpenAI SDK default (600s read, 2 retries), so a hung + embeddings endpoint stalls category seeding and ``recall`` for up to + ~30 minutes. + + SDK ``max_retries`` is left at its default on purpose: a transient + 429 on embeddings should still ride out, and the Layer 2 wait_for + fires at the bound regardless of retries in flight. + + Deliberately does NOT catch: per-site failure policy lives at the + call sites. The startup call must fail closed (leave the bridge + unavailable rather than publish it with an unbounded client), and + ``_initialize_impl``'s own handler already provides that. + """ + if not self._has_embeddings: + return # no embedding profile exists; nothing to bound + + import httpx as _httpx + + client = self._service._get_llm_base_client("embedding") + + # --- Layer 1: httpx transport timeout (no max_retries change) --- + inner = getattr(client, "client", None) # OpenAISDKClient.client = AsyncOpenAI + if inner is not None: + inner.timeout = _httpx.Timeout(self._LLM_CALL_TIMEOUT, connect=10.0) + + # --- Layer 2: asyncio.wait_for wrapper on .embed --- + # Needed in addition to Layer 1 because embed_batch_size defaults to 1, + # so one embed() call can be N sequential requests and a per-request + # timeout does not bound the call. + if not callable(getattr(client, "embed", None)): + return + if getattr(client.embed, "_nerve_timeout_wrapped", False): + return # already wrapped (idempotent across startup/reset calls) + original_embed = client.embed + + async def _timeout_embed(inputs, *args, _orig=original_embed, **kwargs): + return await asyncio.wait_for( + _orig(inputs, *args, **kwargs), + timeout=self._LLM_CALL_TIMEOUT, + ) + + _timeout_embed._nerve_timeout_wrapped = True # type: ignore[attr-defined] + client.embed = _timeout_embed # type: ignore[method-assign] + logger.info( + "Configured %ds timeout on embedding LLM client", self._LLM_CALL_TIMEOUT, + ) + @staticmethod def _is_llm_timeout(exc: Exception) -> bool: """Check if exception is an LLM-level timeout (not a logic error). @@ -2156,7 +2227,11 @@ async def _reset_llm_clients_impl(self) -> None: health = await self._probe_api_health("fast") logger.info("API health probe before reset: %s", health) - for profile in ("memorize", "fast", "default"): + # "embedding" included: an embed timeout reaches this path, and without + # it the offending transport was the one client NOT recycled. The + # post-reset warmup loop below deliberately still excludes it: warming + # an embedding client means a real (billed) embeddings request. + for profile in ("memorize", "fast", "default", "embedding"): try: client = self._service._llm_clients.get(profile) if client is None: @@ -2848,8 +2923,10 @@ async def recall( self._metrics.end_op(op_id, success=False, error=str(e)) # A transient backend outage must not masquerade as "no results" — # surface it so the caller can flag amnesia instead of trusting the - # empty list. - if _is_transient_llm_error(e): + # empty list. Timeouts count: the query embedding is bounded now, + # so a hung embeddings endpoint fails here instead of stalling, and + # _is_transient_llm_error keys on HTTP status only (never a timeout). + if isinstance(e, TimeoutError) or self._is_llm_timeout(e) or _is_transient_llm_error(e): raise MemoryBackendUnavailable(f"memory backend unavailable: {e}") from e return [] diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 92037666..4545271c 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1134,3 +1134,330 @@ async def test_transient_llm_error_still_raises_backend_unavailable(self, tmp_pa await bridge.memorize_file(str(target)) assert bridge._service.memorize.await_count == 1 + + +# --------------------------------------------------------------------------- +# Embedding-call boundedness (_instrument_embedding_timeout) +# --------------------------------------------------------------------------- + + +class _HangingEmbedClient: + """Base-client stand-in whose .embed() never completes. + + Awaits an Event that is never set, so it hangs without touching a real + endpoint and without a sleep the test has to outwait. + """ + + def __init__(self): + self.client = MagicMock() # stands in for AsyncOpenAI + self.client.timeout = "untouched" + self.client.max_retries = 7 # sentinel: must survive (T7) + self.gate = asyncio.Event() + self.embed_calls = 0 + + async def embed(self, inputs, *args, **kwargs): + self.embed_calls += 1 + await self.gate.wait() # never set + return ([[0.0]], None) # pragma: no cover - unreachable + + async def chat(self, prompt, **kwargs): # pragma: no cover - not exercised + return "ok" + + +class _FastEmbedClient(_HangingEmbedClient): + """Same shape, but .embed() returns memU's real 2-tuple immediately.""" + + async def embed(self, inputs, *args, **kwargs): + self.embed_calls += 1 + return ([[0.1, 0.2]], {"raw": 1}) + + +def _make_embed_bridge(tmp_path, client, *, has_embeddings=True): + """Bridge with a mocked service handing out ``client`` for "embedding".""" + config = _make_config(tmp_path) + if has_embeddings: + config.openai_api_key = "test-embed-key" + bridge = MemUBridge(config) + bridge._available = True + bridge._service = MagicMock() + bridge._service._get_llm_base_client = MagicMock(return_value=client) + bridge._LLM_CALL_TIMEOUT = 0.25 # keep the arms fast + return bridge + + +class TestEmbeddingCallTimeout: + """The "embedding" LLM profile must be bounded like every other call.""" + + @pytest.mark.asyncio + async def test_t1_hanging_embed_is_bounded(self, tmp_path): + """T1: a hung embed raises TimeoutError instead of stalling.""" + client = _HangingEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + + bridge._instrument_embedding_timeout() + + t0 = asyncio.get_running_loop().time() + with pytest.raises(asyncio.TimeoutError): + await client.embed(["query"]) + elapsed = asyncio.get_running_loop().time() - t0 + assert elapsed < 5.0, f"took {elapsed:.2f}s - not bounded" + assert client.embed_calls == 1 + + @pytest.mark.asyncio + async def test_t2_bound_fires_through_the_real_memu_wrapper(self, tmp_path): + """T2: the bound survives memU's LLMClientWrapper delegation. + + This pins the load-bearing mechanism: the wrapper resolves + ``self._client.embed`` at CALL time, so patching the base client's + instance attribute is observed even by a wrapper built earlier. + """ + from memu.llm.wrapper import LLMClientWrapper, LLMInterceptorRegistry + + client = _HangingEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + # built BEFORE patching - the delegation is resolved at call time + wrapper = LLMClientWrapper(client, registry=LLMInterceptorRegistry()) + + bridge._instrument_embedding_timeout() + + with pytest.raises(asyncio.TimeoutError): + await wrapper.embed(["query"]) + assert client.embed_calls == 1 + + @pytest.mark.asyncio + async def test_t3_return_shape_is_preserved(self, tmp_path): + """T3: the 2-tuple embed contract is untouched by the wrapper.""" + from memu.llm.wrapper import LLMClientWrapper, LLMInterceptorRegistry + + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + bridge._instrument_embedding_timeout() + + result = await client.embed(["q"]) + assert isinstance(result, tuple) and len(result) == 2 + vecs, raw = result + assert vecs[0] == [0.1, 0.2] + assert raw is not None + + # ...and through the wrapper, which unpacks to the vectors. + wrapper = LLMClientWrapper(client, registry=LLMInterceptorRegistry()) + via_wrapper = await wrapper.embed(["q"]) + assert via_wrapper[0] == [0.1, 0.2] + + @pytest.mark.asyncio + async def test_t4_instrumenting_twice_does_not_double_wrap(self, tmp_path): + """T4: the sentinel makes repeat instrumentation idempotent.""" + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + + bridge._instrument_embedding_timeout() + wrapped_once = client.embed + bridge._instrument_embedding_timeout() + + assert client.embed is wrapped_once, "embed was wrapped twice" + await client.embed(["q"]) + assert client.embed_calls == 1, "original reached more than once per call" + + @pytest.mark.asyncio + async def test_t5_no_embeddings_install_is_a_noop(self, tmp_path): + """T5: without an embedding provider the method does nothing. + + The ``_has_embeddings`` gate (not a swallowed exception) is what makes + this safe, so the service must never be consulted at all. + """ + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client, has_embeddings=False) + bridge._service._get_llm_base_client = MagicMock(side_effect=KeyError("embedding")) + + assert bridge._has_embeddings is False + bridge._instrument_embedding_timeout() # must not raise + bridge._service._get_llm_base_client.assert_not_called() + + @pytest.mark.asyncio + async def test_t6_instrument_llm_timeouts_covers_the_embedding_profile(self, tmp_path): + """T6: regression arm for the enumeration that silently rots. + + Asserted behaviourally (the method IS invoked), not by grepping the + ("memorize","fast","default") tuple. + """ + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + bridge._service._get_llm_base_client = MagicMock(side_effect=KeyError("chat")) + + with patch.object(bridge, "_instrument_embedding_timeout") as spy: + bridge._instrument_llm_timeouts() + + spy.assert_called_once() + + @pytest.mark.asyncio + async def test_t7_sdk_max_retries_is_left_alone(self, tmp_path): + """T7: Layer 1 sets timeout but must NOT copy chat's max_retries=0. + + Embeddings have no compensating retry ladder, so dropping SDK retries + would make a transient 429 fail immediately. + """ + client = _HangingEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + before = client.client.max_retries + + bridge._instrument_embedding_timeout() + + assert client.client.max_retries == before == 7 + assert client.client.timeout != "untouched", "Layer 1 timeout was not set" + + def test_t8_startup_call_precedes_availability_and_seeding(self): + """T8: ordering + availability-window guard, by source order. + + Cannot run ``_initialize_impl`` (it builds a real MemoryService, and + only one may exist per process), so assert on its source text. + """ + import inspect + + src = inspect.getsource(MemUBridge._initialize_impl) + i_bound = src.index("self._instrument_embedding_timeout()") + i_avail = src.index("self._available = True") + i_seed = src.index("await self._ensure_categories()") + + assert i_bound < i_avail, "availability is published before the bound is installed" + assert i_avail < i_seed, "unexpected ordering: seeding moved above availability" + assert i_bound < i_seed, "category seeding embeds before the bound is installed" + + @pytest.mark.asyncio + async def test_t9_reset_evicts_the_embedding_client_but_never_warms_it(self, tmp_path): + """T9: (d) recycles the transport that actually timed out. + + Also pins the negative: the post-reset warmup loop must NOT touch + "embedding", since warming it means a real billed request per reset. + """ + config = _make_config(tmp_path) + config.openai_api_key = "test-embed-key" + bridge = MemUBridge(config) + bridge._available = True + bridge._service = MagicMock() + + clients = {p: MagicMock() for p in ("memorize", "fast", "default", "embedding")} + for c in clients.values(): + c.client._client.aclose = AsyncMock() + + # One ordered timeline, so "was it evicted", "was re-instrumentation + # after the eviction" and "did the WARMUP loop touch it" are three + # independent reads rather than one conflated lookup count. A bare + # lookup list cannot separate the warmup loop from the + # re-instrumentation call, which legitimately resolves "embedding". + timeline: list[tuple[str, str]] = [] + + class _RecordingClients(dict): + def __delitem__(self, key): + timeline.append(("evict", key)) + super().__delitem__(key) + + bridge._service._llm_clients = _RecordingClients(clients) + bridge._service._get_llm_base_client = MagicMock( + side_effect=lambda p: (timeline.append(("lookup", p)), clients[p])[1], + ) + bridge._probe_api_health = AsyncMock(return_value="ok") + + real_instrument = bridge._instrument_llm_timeouts + + def _tracking_instrument(): + timeline.append(("instrument", "begin")) + try: + return real_instrument() + finally: + timeline.append(("instrument", "end")) + + bridge._instrument_llm_timeouts = _tracking_instrument + + await bridge._reset_llm_clients_impl() + + assert ("evict", "embedding") in timeline, \ + "the embedding client was not evicted on reset" + assert "embedding" not in bridge._service._llm_clients + + i_begin = timeline.index(("instrument", "begin")) + i_end = timeline.index(("instrument", "end")) + last_evict = max(i for i, ev in enumerate(timeline) if ev[0] == "evict") + assert last_evict < i_begin, \ + "re-instrumentation ran before eviction - the fresh client would not be wrapped" + + # Everything after instrumentation ends is the warmup loop. + warmed = [p for kind, p in timeline[i_end + 1:] if kind == "lookup"] + assert warmed == ["memorize", "fast"], f"unexpected warmup set: {warmed}" + assert "embedding" not in warmed, \ + "the post-reset warmup issued an embeddings request" + + @pytest.mark.asyncio + async def test_t10_startup_instrumentation_fails_closed(self, tmp_path): + """T10: the method must NOT swallow its own failure. + + The startup call site relies on propagation to leave the bridge + unavailable; an internal try/except would silently reintroduce an + unbounded client behind ``_available = True``. + """ + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + bridge._service._get_llm_base_client = MagicMock( + side_effect=RuntimeError("profile exploded"), + ) + + with pytest.raises(RuntimeError, match="profile exploded"): + bridge._instrument_embedding_timeout() + + @pytest.mark.asyncio + async def test_t11_reset_path_stays_best_effort(self, tmp_path): + """T11: an embedding failure cannot break chat re-instrumentation.""" + client = _FastEmbedClient() + bridge = _make_embed_bridge(tmp_path, client) + + chat_clients = {} + + def _by_profile(profile): + if profile == "embedding": + raise RuntimeError("embedding profile exploded") + c = MagicMock() + c.chat = AsyncMock(return_value="ok") + chat_clients[profile] = c + return c + + bridge._service._get_llm_base_client = MagicMock(side_effect=_by_profile) + + bridge._instrument_llm_timeouts() # must NOT raise + + assert set(chat_clients) == {"memorize", "fast", "default"} + for profile, c in chat_clients.items(): + assert getattr(c.chat, "_nerve_timeout_wrapped", False), \ + f"chat profile {profile} was left un-instrumented" + + @pytest.mark.asyncio + async def test_t12_recall_timeout_is_backend_down_not_empty(self, tmp_path): + """T12: a bounded-but-failed recall must not answer "no memories". + + Both layers: asyncio.TimeoutError (our wait_for) and + openai.APITimeoutError (the httpx transport). + """ + from openai import APITimeoutError + import httpx + + for exc in ( + asyncio.TimeoutError(), + APITimeoutError(request=httpx.Request("POST", "http://embeddings.invalid")), + ): + config = _make_config(tmp_path) + bridge = MemUBridge(config) + bridge._available = True + bridge._service = MagicMock() + bridge._service.retrieve = MagicMock(side_effect=exc) + + with pytest.raises(MemoryBackendUnavailable, match="unavailable"): + await bridge.recall("anything") + + @pytest.mark.asyncio + async def test_t13_recall_still_returns_empty_for_logic_errors(self, tmp_path): + """T13: (e) must not over-reach - a genuine error still yields [].""" + config = _make_config(tmp_path) + bridge = MemUBridge(config) + bridge._available = True + bridge._service = MagicMock() + bridge._service.retrieve = MagicMock(side_effect=ValueError("bad payload")) + + assert await bridge.recall("anything") == [] From 153643e6083bdb2ea951364547199035464755e8 Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 04:09:36 +1200 Subject: [PATCH 2/5] Make the hang arms fail instead of hanging when the bound is absent T1/T2 awaited the wrapped embed() directly, so a tree without the wait_for (the M1 mutation, or a future round deleting it) hung the arm forever rather than failing it - the matrix could not report a kill. The _bounded() helper escapes with AssertionError, never TimeoutError, so it cannot be mistaken for the bound firing and cannot make either arm pass vacuously. Verified in both directions: with the fix 13 pass; with the wait_for deleted T1/T2 fail in 10.3s on "the call is NOT bounded". --- tests/test_memu_bridge.py | 29 ++++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 4545271c..ede33ffe 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1172,6 +1172,29 @@ async def embed(self, inputs, *args, **kwargs): return ([[0.1, 0.2]], {"raw": 1}) +# Wall-clock ceiling for the hang arms. Well above the 0.25s bound they +# assert, so a passing arm never waits on it. +_ARM_CEILING = 5.0 + + +async def _bounded(awaitable): + """Await ``awaitable``, converting a HANG into a distinguishable failure. + + Without this an un-bounded embed (e.g. a mutant that deletes the wait_for) + would hang the arm forever instead of failing it. The escape raises + ``AssertionError``, never ``TimeoutError``, so it can never be mistaken + for the bound firing - i.e. it cannot make T1/T2 pass vacuously. + """ + task = asyncio.ensure_future(awaitable) + done, _ = await asyncio.wait({task}, timeout=_ARM_CEILING) + if not done: + task.cancel() + raise AssertionError( + f"embed did not complete within {_ARM_CEILING}s - the call is NOT bounded", + ) + return task.result() + + def _make_embed_bridge(tmp_path, client, *, has_embeddings=True): """Bridge with a mocked service handing out ``client`` for "embedding".""" config = _make_config(tmp_path) @@ -1198,9 +1221,9 @@ async def test_t1_hanging_embed_is_bounded(self, tmp_path): t0 = asyncio.get_running_loop().time() with pytest.raises(asyncio.TimeoutError): - await client.embed(["query"]) + await _bounded(client.embed(["query"])) elapsed = asyncio.get_running_loop().time() - t0 - assert elapsed < 5.0, f"took {elapsed:.2f}s - not bounded" + assert elapsed < _ARM_CEILING, f"took {elapsed:.2f}s - not bounded" assert client.embed_calls == 1 @pytest.mark.asyncio @@ -1221,7 +1244,7 @@ async def test_t2_bound_fires_through_the_real_memu_wrapper(self, tmp_path): bridge._instrument_embedding_timeout() with pytest.raises(asyncio.TimeoutError): - await wrapper.embed(["query"]) + await _bounded(wrapper.embed(["query"])) assert client.embed_calls == 1 @pytest.mark.asyncio From 38ad135906134988a3faf86b46a249d652523eb0 Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 07:39:51 +1200 Subject: [PATCH 3/5] Address review round 1: bound the embedding profile unconditionally Removes the `_has_embeddings` gate from `_instrument_embedding_timeout`, because the premise behind it is false. memU's `LLMProfilesConfig.ensure_default` synthesizes an "embedding" profile from "default" whenever the caller omits one (`memu/app/settings.py:286`), so the profile exists even with no `openai_api_key`. Measured on a real no-key `_initialize_impl`: `_get_llm_base_client("embedding")` returns an `OpenAISDKClient` whose inner `AsyncOpenAI` carries the SDK default `Timeout(read=600)` and `max_retries=2`, and `embed` is unwrapped. That path is reachable: memU's update workflow declares `embed_llm_profile: "embedding"` (`memu/app/crud.py:431`), reached from nerve via `update_item`, and nothing along it imposes a timeout. So the gate left a no-provider install able to make a 600s-by-2-retries embedding call, which is the stall this change exists to remove. With the gate gone, a no-key startup still returns True, the inner timeout drops to the 120s bound, SDK retries are untouched, and a hung embed raises at exactly the bound. Tests. T5 is repurposed rather than deleted: its premise ("a no-provider install is a no-op") is now false, so it asserts the opposite, that such an install is still bounded. T14 drives the no-key update-path shape through memU's real `LLMClientWrapper`. T15 covers the fail-closed contract that no arm previously observed: edit (c) is deliberately uncaught so a failure leaves the bridge unavailable, and a call-site `try/except` left the suite fully green. It runs the real `_initialize_impl` in a fresh interpreter, since only one `MemoryService` may exist per process, and asserts a control run first so a later False cannot be an unrelated fixture failure. T9 stopped asserting a falsehood about production: the reset evicts the embedding client and the following re-instrumentation re-creates and re-wraps it, so the cache is repopulated rather than left empty. The old assertion only passed because the fake getter closed over a stale copy of the client mapping. Also widens the backend-down tool message to name timeouts, which now route into that block, and condenses the method docstring from 19 lines to 6 while keeping the clauses that pinned mutants; the removed prose is above. The test and mutant figures in the previous commit message are superseded by this round's, which are re-derived rather than carried. --- nerve/agent/tools/handlers/memory.py | 3 +- nerve/memory/memu_bridge.py | 26 +---- tests/test_memu_bridge.py | 161 ++++++++++++++++++++++++--- 3 files changed, 156 insertions(+), 34 deletions(-) diff --git a/nerve/agent/tools/handlers/memory.py b/nerve/agent/tools/handlers/memory.py index fe36b3b9..6da982bf 100644 --- a/nerve/agent/tools/handlers/memory.py +++ b/nerve/agent/tools/handlers/memory.py @@ -107,7 +107,8 @@ async def memory_recall_handler(ctx: ToolContext, args: dict) -> ToolResult: except MemoryBackendUnavailable as e: logger.warning("Memory recall: backend unavailable: %s", e) memu_block = ( - "⚠️ MEMORY BACKEND DOWN (transient proxy/auth error) — recall is UNAVAILABLE right now, " + "⚠️ MEMORY BACKEND DOWN (transient proxy/auth error or timeout) — recall is UNAVAILABLE " + "right now, " "NOT empty. Do NOT conclude 'no relevant memories'. Reconstruct context from other " "sources (git history, your own notes) and alert the operator that memory is down." ) diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index be776537..730b7cb5 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -2100,27 +2100,13 @@ async def _timeout_chat( def _instrument_embedding_timeout(self) -> None: """Bound calls on the "embedding" LLM profile (same two layers). - The ``_instrument_llm_timeouts`` loop deliberately excludes this - profile: its Layer 2 wraps ``.chat`` (embeddings go through - ``.embed``) and its Layer 1 sets ``max_retries = 0``, which is only - safe for chat because ``_timeout_chat`` supplies its own transient - retry ladder. Without this method every ``embed()`` call runs under - the bare OpenAI SDK default (600s read, 2 retries), so a hung - embeddings endpoint stalls category seeding and ``recall`` for up to - ~30 minutes. - - SDK ``max_retries`` is left at its default on purpose: a transient - 429 on embeddings should still ride out, and the Layer 2 wait_for - fires at the bound regardless of retries in flight. - - Deliberately does NOT catch: per-site failure policy lives at the - call sites. The startup call must fail closed (leave the bridge - unavailable rather than publish it with an unbounded client), and - ``_initialize_impl``'s own handler already provides that. + ``_instrument_llm_timeouts`` cannot cover it: that loop wraps ``.chat`` + and zeroes ``max_retries``, safe only for chat's own retry ladder, so + SDK retries stay here. Ungated because memU synthesizes "embedding" + from "default" when omitted (``memu/app/settings.py:286``), so even a + no-provider install reaches ``embed`` via memU's update workflow. + Uncaught, so startup fails closed via ``_initialize_impl``'s handler. """ - if not self._has_embeddings: - return # no embedding profile exists; nothing to bound - import httpx as _httpx client = self._service._get_llm_base_client("embedding") diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index ede33ffe..2d62a25c 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1282,19 +1282,25 @@ async def test_t4_instrumenting_twice_does_not_double_wrap(self, tmp_path): assert client.embed_calls == 1, "original reached more than once per call" @pytest.mark.asyncio - async def test_t5_no_embeddings_install_is_a_noop(self, tmp_path): - """T5: without an embedding provider the method does nothing. + async def test_t5_no_provider_install_is_still_bounded(self, tmp_path): + """T5: a no-provider install must be bounded, not skipped. - The ``_has_embeddings`` gate (not a swallowed exception) is what makes - this safe, so the service must never be consulted at all. + memU synthesizes "embedding" from "default" whenever the caller omits + it (memu/app/settings.py:286), so the profile exists even with no + openai_api_key and its embed calls are reachable through memU's update + workflow. Gating on ``_has_embeddings`` would leave them unbounded. """ client = _FastEmbedClient() bridge = _make_embed_bridge(tmp_path, client, has_embeddings=False) - bridge._service._get_llm_base_client = MagicMock(side_effect=KeyError("embedding")) - assert bridge._has_embeddings is False - bridge._instrument_embedding_timeout() # must not raise - bridge._service._get_llm_base_client.assert_not_called() + assert bridge._has_embeddings is False, "fixture is not a no-provider install" + + bridge._instrument_embedding_timeout() + + bridge._service._get_llm_base_client.assert_called_once_with("embedding") + assert getattr(client.embed, "_nerve_timeout_wrapped", False), \ + "a no-provider install was left with an unbounded embed" + assert client.client.timeout != "untouched", "Layer 1 timeout was not set" @pytest.mark.asyncio async def test_t6_instrument_llm_timeouts_covers_the_embedding_profile(self, tmp_path): @@ -1374,11 +1380,27 @@ def __delitem__(self, key): timeline.append(("evict", key)) super().__delitem__(key) - bridge._service._llm_clients = _RecordingClients(clients) - bridge._service._get_llm_base_client = MagicMock( - side_effect=lambda p: (timeline.append(("lookup", p)), clients[p])[1], - ) + cache = _RecordingClients(clients) + bridge._service._llm_clients = cache + + def _factory(profile): + """Stand in for memU's _get_llm_base_client: cache, else create. + + Writing the fresh object back into ``cache`` (not the original + ``clients`` dict) is what makes the post-reset assertions observe + production behaviour rather than a stale copy. + """ + timeline.append(("lookup", profile)) + if profile in cache: + return cache[profile] + fresh = MagicMock() + fresh.client._client.aclose = AsyncMock() + cache[profile] = fresh + return fresh + + bridge._service._get_llm_base_client = MagicMock(side_effect=_factory) bridge._probe_api_health = AsyncMock(return_value="ok") + before_embedding = clients["embedding"] real_instrument = bridge._instrument_llm_timeouts @@ -1395,7 +1417,17 @@ def _tracking_instrument(): assert ("evict", "embedding") in timeline, \ "the embedding client was not evicted on reset" - assert "embedding" not in bridge._service._llm_clients + + # Production truth: the re-instrumentation that follows the eviction + # re-resolves the profile through the factory, so the cache is + # repopulated with a FRESH, wrapped client - it does not stay empty. + assert "embedding" in bridge._service._llm_clients, \ + "the embedding client was not re-created after the eviction" + after_embedding = bridge._service._llm_clients["embedding"] + assert after_embedding is not before_embedding, \ + "the evicted client was reused instead of re-created" + assert getattr(after_embedding.embed, "_nerve_timeout_wrapped", False), \ + "the fresh embedding client was left unbounded" i_begin = timeline.index(("instrument", "begin")) i_end = timeline.index(("instrument", "end")) @@ -1484,3 +1516,106 @@ async def test_t13_recall_still_returns_empty_for_logic_errors(self, tmp_path): bridge._service.retrieve = MagicMock(side_effect=ValueError("bad payload")) assert await bridge.recall("anything") == [] + + @pytest.mark.asyncio + async def test_t14_no_key_update_path_is_bounded(self, tmp_path): + """T14: the no-provider update workflow is bounded end to end. + + memU's update workflow embeds changed content through the "embedding" + profile (memu/app/crud.py:431 declares embed_llm_profile), reached from + nerve via update_item. Nothing on that route imposes a timeout, so the + instrumentation is the only bound - and it must apply with no + openai_api_key, which is the install shape that used to be skipped. + """ + from memu.llm.wrapper import LLMClientWrapper, LLMInterceptorRegistry + + client = _HangingEmbedClient() + bridge = _make_embed_bridge(tmp_path, client, has_embeddings=False) + # built BEFORE instrumentation, exactly as memU builds it at startup + wrapper = LLMClientWrapper(client, registry=LLMInterceptorRegistry()) + + bridge._instrument_embedding_timeout() + + with pytest.raises(asyncio.TimeoutError): + await _bounded(wrapper.embed(["a changed fact"])) + assert client.embed_calls == 1 + + def test_t15_startup_instrumentation_failure_leaves_the_bridge_unavailable(self, tmp_path): + """T15: edit (c) is uncaught, so a failure there must fail closed. + + Runs the REAL _initialize_impl in a fresh interpreter: only one + MemoryService may exist per process (a second raises ArgumentError on + a re-declared column), which is why T8 settles for source order. A + call-site try/except would leave the bridge available with an + unbounded client, and no in-process arm can observe that. + """ + import os + import subprocess + import sys + + script = ( + "import asyncio, json, sqlite3, sys\n" + "from unittest.mock import AsyncMock\n" + "from nerve.config import MemoryCategoryConfig, MemoryConfig, NerveConfig\n" + "from nerve.memory import memu_bridge as mb\n" + "db = sys.argv[1]\n" + "if sys.argv[2] == 'break':\n" + " def _boom(self):\n" + " raise RuntimeError('instrumentation exploded')\n" + " mb.MemUBridge._instrument_embedding_timeout = _boom\n" + "async def main():\n" + " c = NerveConfig()\n" + " c.memory = MemoryConfig(sqlite_dsn='sqlite:///' + db)\n" + " c.memory.categories = [MemoryCategoryConfig(name='probes', description='d')]\n" + " c.anthropic_api_key = 'k'\n" + # unroutable local port, so the warmup never reaches a real endpoint + " c.proxy.enabled = True\n" + " c.proxy.host = '127.0.0.1'\n" + " c.proxy.port = 1\n" + " b = mb.MemUBridge(c)\n" + " b._audit = AsyncMock()\n" + " rc = await b.initialize()\n" + " rows = -1\n" + " try:\n" + " con = sqlite3.connect(db)\n" + " rows = list(con.execute(" + "'select count(*) from memu_memory_categories'))[0][0]\n" + " except sqlite3.OperationalError:\n" + " rows = 0\n" # table never created at all + " print('NERVE_RESULT ' + json.dumps(" + "{'rc': bool(rc), 'available': bool(b.available), 'rows': rows}))\n" + " await b.shutdown()\n" + "asyncio.run(main())\n" + ) + + def run(mode): + db = tmp_path / f"memu-{mode}.sqlite" + proc = subprocess.run( + [sys.executable, "-c", script, str(db), mode], + capture_output=True, text=True, timeout=300, + cwd=str(Path(__file__).resolve().parent.parent), + env={**os.environ, "NERVE_HOME": str(tmp_path / f"home-{mode}")}, + ) + marker = [ln for ln in proc.stdout.splitlines() + if ln.startswith("NERVE_RESULT ")] + assert marker, ( + f"no result marker from the {mode} run\n" + f"stdout tail:\n{proc.stdout[-2000:]}\n" + f"stderr tail:\n{proc.stderr[-2000:]}" + ) + return json.loads(marker[-1][len("NERVE_RESULT "):]) + + # Control first: a later False must not be the fixture failing for an + # unrelated reason. + ok = run("intact") + assert ok["rc"] is True, f"control startup failed: {ok}" + assert ok["available"] is True + assert ok["rows"] >= 1, "control persisted no category, so rows==0 proves nothing" + + broken = run("break") + assert broken["rc"] is False, \ + "startup succeeded despite the instrumentation raising - it is caught somewhere" + assert broken["available"] is False, \ + "the bridge advertised itself as usable with an unbounded embedding client" + assert broken["rows"] == 0, \ + "a category row was persisted after the bound failed to install" From 515eaaaa7f8cb3faacfc56186a3ef4e6e0190418 Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 07:48:43 +1200 Subject: [PATCH 4/5] Make T9's wrapped-client assertion non-vacuous The rewritten T9 asserted that the post-reset embedding client carries `_nerve_timeout_wrapped`, but its factory handed out a `MagicMock`, which auto-creates that attribute as a truthy child mock. The assertion could therefore never fail. Caught by a mutant that repopulates the cache with a fresh client and never wraps it: it survived, while the same mutant against a real client stand-in fails on exactly that assertion. The factory now returns `_FastEmbedClient`, which has a real `embed`. A `MagicMock` can only carry a "does this attribute exist" assertion when the attribute's absence is asserted; for presence it is always vacuous. --- tests/test_memu_bridge.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 2d62a25c..5b2130a8 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1388,12 +1388,15 @@ def _factory(profile): Writing the fresh object back into ``cache`` (not the original ``clients`` dict) is what makes the post-reset assertions observe - production behaviour rather than a stale copy. + production behaviour rather than a stale copy. The fresh object is + a real client stand-in, NOT a MagicMock: a MagicMock auto-creates + ``embed._nerve_timeout_wrapped`` as a truthy child mock, so the + "is it wrapped" assertion below could never fail against one. """ timeline.append(("lookup", profile)) if profile in cache: return cache[profile] - fresh = MagicMock() + fresh = _FastEmbedClient() fresh.client._client.aclose = AsyncMock() cache[profile] = fresh return fresh From 4e4e8e45434709cd215cb9d6e0dcc906b6946e21 Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 4 Aug 2026 09:31:57 +1200 Subject: [PATCH 5/5] Make the T11 and T14 arms observe what their names claim Two arms were green for reasons unrelated to the property they exist to pin, both instances of the same MagicMock rule the previous commit stated. T11 handed out `MagicMock()` clients, so `chat._nerve_timeout_wrapped` existed as a truthy child mock before any instrumentation. That made production's own sentinel guard skip all three chat profiles, and made the arm's assertion read the same truthy child and pass anyway. Measured: deleting `client.chat = _timeout_chat` outright, so chat is never wrapped anywhere, left T11 at 1 passed. The fixture now hands out `_FastEmbedClient`, captures the original bound methods, and asserts first that the fixture does not pre-carry the sentinel - that assertion is what stops a future round regressing to a mock. T14 was T2's body with `has_embeddings=False`: same hanging client, same manually built wrapper, same two assertions. It never entered `update_item` and never touched a memU workflow, so it stayed green if either link in its own docstring broke. It keeps the bound-fires assertion, which is what makes it the no-key regression arm, and adds the two links directly: nerve still forwards the changed content as `memory_content`, and memU's real update workflow still resolves the "embedding" profile for that step. A real in-memory `MemoryService` was rejected as the vehicle - only one may exist per process, which is the same constraint that pushed T8 to source-order assertions and T15 into a subprocess, and a second subprocess arm is disproportionate here. Link 2 therefore builds the declaration with `object.__new__` and runs memU's own resolver over it, asserting the literal profile string rather than truthiness, since the resolver returns None for any other key shape and the call site would fall back to an "embedding" default. The method docstring drops from 9 source lines to 7, keeping the four clauses that pin mutants. No mutant anchors on docstring text, so the coverage cost is zero. The client-layer bound covers the 17 memU-internal `embed()` sites nerve does not own, 14 of them reachable through `MemoryService`'s MRO (`PatchMixin` is not in it). The first commit of this branch said 16. Validation, all re-derived this round and superseding the previous message's figures: 15 arms, 15 passed. 19-mutant matrix, all 19 killed, unmutated control green at both ends, tree restored. M17 (chat never wrapped) and M18 (sentinel guard always skips) are killed by T11; M19 (`update_item` stops forwarding the content) by T14. Link 2's discriminator is memU's own source, which a shipped matrix cannot mutate. Full suite 7 failed / 2948 passed, with the failed-name set byte-identical to the 7 pre-existing names and the pass count unchanged, since both arms were rewritten in place rather than added. `ruff` findings byte-identical between base and head (79 each, over the same 339 tracked files). --- nerve/memory/memu_bridge.py | 10 ++++------ tests/test_memu_bridge.py | 34 ++++++++++++++++++++++++++++++++-- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 730b7cb5..97a6f81d 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -2100,12 +2100,10 @@ async def _timeout_chat( def _instrument_embedding_timeout(self) -> None: """Bound calls on the "embedding" LLM profile (same two layers). - ``_instrument_llm_timeouts`` cannot cover it: that loop wraps ``.chat`` - and zeroes ``max_retries``, safe only for chat's own retry ladder, so - SDK retries stay here. Ungated because memU synthesizes "embedding" - from "default" when omitted (``memu/app/settings.py:286``), so even a - no-provider install reaches ``embed`` via memU's update workflow. - Uncaught, so startup fails closed via ``_initialize_impl``'s handler. + Not covered by ``_instrument_llm_timeouts``' loop (that wraps ``.chat`` + and zeroes ``max_retries``; embeddings have no retry ladder, so SDK + retries stay). Ungated: memU synthesizes "embedding" from "default" + when omitted (``memu/app/settings.py:286``). Uncaught by design. """ import httpx as _httpx diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 5b2130a8..115cbc46 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1468,12 +1468,17 @@ async def test_t11_reset_path_stays_best_effort(self, tmp_path): bridge = _make_embed_bridge(tmp_path, client) chat_clients = {} + originals = {} def _by_profile(profile): if profile == "embedding": raise RuntimeError("embedding profile exploded") - c = MagicMock() - c.chat = AsyncMock(return_value="ok") + # A concrete fake, not a MagicMock: a mock auto-creates + # ``chat._nerve_timeout_wrapped`` as a truthy child, which makes + # production skip wrapping at its sentinel guard AND makes the + # assertion below pass anyway. + c = _FastEmbedClient() + originals[profile] = c.chat chat_clients[profile] = c return c @@ -1483,6 +1488,10 @@ def _by_profile(profile): assert set(chat_clients) == {"memorize", "fast", "default"} for profile, c in chat_clients.items(): + assert not getattr(originals[profile], "_nerve_timeout_wrapped", False), \ + "fixture pre-carried the sentinel, so the assertion below is vacuous" + assert c.chat is not originals[profile], \ + f"chat profile {profile} method was never replaced" assert getattr(c.chat, "_nerve_timeout_wrapped", False), \ f"chat profile {profile} was left un-instrumented" @@ -1530,6 +1539,8 @@ async def test_t14_no_key_update_path_is_bounded(self, tmp_path): instrumentation is the only bound - and it must apply with no openai_api_key, which is the install shape that used to be skipped. """ + from memu.app.crud import CRUDMixin + from memu.app.service import MemoryService from memu.llm.wrapper import LLMClientWrapper, LLMInterceptorRegistry client = _HangingEmbedClient() @@ -1543,6 +1554,25 @@ async def test_t14_no_key_update_path_is_bounded(self, tmp_path): await _bounded(wrapper.embed(["a changed fact"])) assert client.embed_calls == 1 + # Link 1: nerve still forwards the content memU embeds. + bridge._service.update_memory_item = AsyncMock(return_value={}) + bridge._audit = AsyncMock() + assert await bridge.update_item("id-1", content="a changed fact") is True + kwargs = bridge._service.update_memory_item.await_args.kwargs + assert kwargs["memory_content"] == "a changed fact", \ + "update_item no longer forwards the content memU would embed" + + # Link 2: memU's update step still embeds on the bounded profile. + # Declaration only, so no MemoryService.__init__ and no per-process cost. + svc = object.__new__(MemoryService) + steps = CRUDMixin._build_update_memory_item_workflow(svc) + step = next(s for s in steps if s.step_id == "update_memory_item") + profile = MemoryService._llm_profile_from_context( + {"step_config": step.config}, task="embedding", + ) + assert profile == "embedding", \ + f"memU's update step no longer embeds on the bounded profile: {profile}" + def test_t15_startup_instrumentation_failure_leaves_the_bridge_unavailable(self, tmp_path): """T15: edit (c) is uncaught, so a failure there must fail closed.