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 9642175b..97a6f81d 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,53 @@ 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). + + 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 + + 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 +2211,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 +2907,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..115cbc46 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -1134,3 +1134,521 @@ 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}) + + +# 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) + 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 _bounded(client.embed(["query"])) + elapsed = asyncio.get_running_loop().time() - t0 + assert elapsed < _ARM_CEILING, 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 _bounded(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_provider_install_is_still_bounded(self, tmp_path): + """T5: a no-provider install must be bounded, not skipped. + + 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) + + 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): + """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) + + 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. 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 = _FastEmbedClient() + 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 + + 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" + + # 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")) + 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 = {} + originals = {} + + def _by_profile(profile): + if profile == "embedding": + raise RuntimeError("embedding profile exploded") + # 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 + + 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 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" + + @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") == [] + + @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.app.crud import CRUDMixin + from memu.app.service import MemoryService + 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 + + # 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. + + 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"