From 71689142bcf21101bdb9434fc99c72c72ebd5ba8 Mon Sep 17 00:00:00 2001
From: Vuong Hoang <vh@phasefinal.com>
Date: Sat, 1 Aug 2026 18:55:58 -0700
Subject: [PATCH] =?UTF-8?q?feat:=20Donut=20voiced-interview=20slice-3=20?=
 =?UTF-8?q?=E2=80=94=20retire-ready=20KB-recall=20bridge?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Grounds the interview character in the ingested corpus while she stays in-voice.
Tier-3 agents are tool-less by design in v1, so this is the consumer-side
workaround (DEC-6, worldtree-dev ruling): per opted-in interview turn, ratatoskr
consults Mimir out-of-band, extracts the passages, and pins them as memory_context
on the character's turn. She frames the pinned corpus as her own memory.

- src/ratatoskr/kb_bridge.py (new, RETIRE-READY): pin_kb_context — THE single seam
  (INV-KB-1). Allowlist-gated (INV-KB-4: ratatoskr:donut only), hard-timeout-bounded,
  degrades to [] on any failure/timeout/empty (INV-KB-3, never raises; CancelledError
  propagates). Imports nothing from the SDK-adapter / TTS core. aclosing() closes the
  SDK stream deterministically on the DoneEvent break.
- wt.stream_turn: memory_context passthrough (defaults None — inert for every other
  caller and for the bridge's own retirement). Seam-review catch: the contract's
  original touch list undercounted wt.py by one file (recorded in the contract).
- web/server.py: TurnHandle.agent_id + the single pin_kb_context call-site on the
  turn path; the browser now sends agent_id so the allowlist can gate.
- web/static/index.html: the turn POST carries agent_id.

Consult prompt tuned live: "search_library EXACTLY ONCE, no read_note" converges
Mimir in ~3-15s (the softer "do one search" phrasing looped past 25s on
conversational questions).

TDD: 12 kb_bridge unit tests + wt memory_context forwarding + 2 server wiring tests
(531 green). Live-smoked on :8081/b128: pin_kb_context grounds in the DCC corpus
(real excerpts, <20s) and Donut answers in-voice; degrades cleanly on a slow consult.
KNOWN LIMIT surfaced (not a bridge defect): DCC's fiction index is weak (failed
backfill, a worldtree-dev item), so grounding is opportunistic — the bridge's real
payoff is a corpus the model does not already know.

Per docs/contracts/donut_voiced_interview.contract.md (slice 3 of 3).
---
 .../donut_voiced_interview.contract.md        |  26 +++-
 src/ratatoskr/kb_bridge.py                    | 113 ++++++++++++++
 src/ratatoskr/web/server.py                   |  24 ++-
 src/ratatoskr/web/static/index.html           |   2 +-
 src/ratatoskr/wt.py                           |  15 +-
 tests/test_kb_bridge.py                       | 143 ++++++++++++++++++
 tests/test_web_server.py                      |  55 +++++++
 tests/test_wt.py                              |  10 +-
 8 files changed, 375 insertions(+), 13 deletions(-)
 create mode 100644 src/ratatoskr/kb_bridge.py
 create mode 100644 tests/test_kb_bridge.py

diff --git a/docs/contracts/donut_voiced_interview.contract.md b/docs/contracts/donut_voiced_interview.contract.md
index 650e52c..5fd506f 100644
--- a/docs/contracts/donut_voiced_interview.contract.md
+++ b/docs/contracts/donut_voiced_interview.contract.md
@@ -19,9 +19,11 @@ scope: >
   consult (an existing agent turn).
 touches:
   - src/ratatoskr/web/server.py          # /api/tts route + the retrieval-pinning seam on the turn path
-  - src/ratatoskr/web/static/index.html   # speak-on-done playback, 🔊 toggle, <audio> sink
+  - src/ratatoskr/web/static/index.html   # speak-on-done playback, 🔊 toggle, <audio> sink; turn POST carries agent_id
+  - src/ratatoskr/web/entrypoint.py       # RATATOSKR_TTS_URL override (the tts swap seam)
   - src/ratatoskr/tts.py                  # NEW — Zonos gateway client + PAD->emotion-dial mapping
   - src/ratatoskr/kb_bridge.py            # NEW, RETIRE-READY — consumer-side retrieval + memory_context pinning
+  - src/ratatoskr/wt.py                   # stream_turn gains a memory_context passthrough (seam-review: the contract's original touch list undercounted this by one file; the param defaults None so the bridge's RETIREMENT stays inert — deleting kb_bridge.py + the one call-site leaves wt.stream_turn's SDK-parity param harmless)
   - docs/characters/donut.md              # NEW — Princess Donut persona (content; the tier3 define source)
 depends_on:
   - "Zonos gateway: POST http://10.100.79.3:8890/v1/audio/speech (infra-ops; WG-internal, no auth; wav; verified 2026-08-02)"
@@ -138,16 +140,26 @@ POST /api/tts  {text, agent_id?}  -> audio/wav
 
 ### FN pin_kb_context  (kb_bridge.py — RETIRE-READY, INV-KB-1)
 ```
-pin_kb_context(question: str, agent_id: str, *, client) -> list[dict]   # memory_context items, or []
+pin_kb_context(question: str, agent_id: str | None, *, client) -> list[dict]   # memory_context items, or []
   # The bridge. Consumer-side retrieval + pinning (DEC-6).
   steps:
-    - gate on the interview-character allowlist (INV-KB-4); not listed -> [].
-    - out-of-band Mimir consult: a one-shot turn "search the well for <question topic> (<char's corpus>)".
-    - extract cited evidence text (bounded length).
+    - gate on the interview-character allowlist (INV-KB-4); not listed / blank question -> [].
+    - out-of-band Mimir consult (a throwaway session + one turn), HARD-bounded by a timeout.
+    - extract the answer text (prefer DoneEvent.response; fall back to text deltas), bounded length.
     - return [{"kind":"corpus_reference","text":<extract>}].
-  error/empty: any failure or no hits -> [] (INV-KB-3; never raises to the turn path).
+  error/empty/timeout: any failure or no hits -> [] (INV-KB-3; never raises to the turn path).
+                       CancelledError (browser disconnect) is NOT caught — it propagates.
+  CONSULT PROMPT (foot-gun mitigation, tuned live 2026-08-02): force "search_library EXACTLY ONCE,
+    no read_note" — converges Mimir in ~3-15s. The softer "do one search" phrasing let Mimir loop
+    read_note<->search past a 25s ceiling on conversational (non-keyword) questions (live-observed).
+  KNOWN LIMIT (surfaced by the live smoke, NOT a bridge defect): the bridge's GROUNDING VALUE is gated
+    by Mimir's retrieval quality on the target corpus. DCC's fiction-wing index is currently weak
+    (scores ~0.02, failed backfill — a standing worldtree-dev item), so hits are noisy/partial; the
+    model's own DCC training knowledge already grounds Donut well, so the bridge is opportunistic here.
+    Its real payoff is a corpus the model does NOT know AND that indexes cleanly.
   RETIREMENT: when Worldtree #361 reference_knowledge reaches Tier-3, delete this module + the single
-              server.py call-site; Donut then searches in-voice natively.
+              server.py call-site (wt.stream_turn's memory_context param stays, inert); Donut then
+              searches in-voice natively.
 ```
 
 ### FN client: speakOnDone  (index.html)
diff --git a/src/ratatoskr/kb_bridge.py b/src/ratatoskr/kb_bridge.py
new file mode 100644
index 0000000..6fd8b26
--- /dev/null
+++ b/src/ratatoskr/kb_bridge.py
@@ -0,0 +1,113 @@
+"""Retire-ready KB-recall bridge (slice 3 of the Donut voiced-interview contract).
+
+Tier-3 agents are DELIBERATELY tool-less in Worldtree v1 (build_tier3_agent_context
+hard-codes tool_schemas=[]); a character cannot search the corpus herself. This is
+the consumer-side workaround (DEC-6, worldtree-dev ruling, wyrd-proven): per interview
+turn, ratatoskr consults Mimir (the Well-of-Knowledge agent) out-of-band, extracts the
+grounded passages, and pins them as `memory_context` on the character's turn. She
+answers in-voice, framing the pinned corpus as her own memory (the persona prompt
+instructs exactly this).
+
+RETIRE-READY (INV-KB-1): isolated behind ONE seam — `pin_kb_context` — called from
+exactly one call-site (web/server.py's turn path). Retiring the bridge = delete this
+module + that call-site; `wt.stream_turn`'s `memory_context` param stays (SDK-parity,
+inert). This module imports nothing from the TTS or SDK-adapter core
+(`ratatoskr.wt` / `ratatoskr.tts`); it drives the passed-in WorldtreeClient directly.
+Delete when Worldtree #361 `reference_knowledge` extends native retrieval to Tier-3.
+
+Foot-gun (verified live 2026-08-02): an UNCONSTRAINED Mimir consult loops
+read_note↔search_library and can run >60s without converging. The CONSTRAINED
+"one search, passages verbatim, no iterative reading" prompt converges in ~10s. The
+consult is HARD-bounded by a timeout and degrades to [] (INV-KB-3), so a slow/looping
+consult never hangs the character's turn.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import aclosing
+
+from worldtree_sdk import DoneEvent, TextEvent, WorldtreeClient
+
+# INV-KB-4: only interview characters that need corpus recall pay the retrieval
+# round-trip. Maps each opted-in agent_id → the corpus name that scopes the search.
+_KB_INTERVIEW_CORPORA = {"ratatoskr:donut": "Dungeon Crawler Carl"}
+
+# The KB agent driven for retrieval (Worldtree's Well-of-Knowledge custodian).
+_MIMIR_AGENT_ID = "mimir"
+# A fixed partition for the out-of-band consult, isolated from the character's own
+# end-user partition. The retrieval channel is never persisted (INV-KB-2): the
+# consult is throwaway and its result rides only the character's turn POST body.
+_KB_BRIDGE_END_USER = "ratatoskr-kb-bridge"
+# Hard ceiling on the consult (degrade to [] past it). The single-search prompt
+# below converges in ~3-4s live; the headroom absorbs gateway variance. An
+# unconstrained consult loops read_note↔search_library and blows well past this.
+_CONSULT_TIMEOUT_S = 20.0
+# Bound the pinned extract; a whole-note dump would bloat the character's prompt.
+_MAX_EXTRACT_CHARS = 2000
+
+
+def _consult_prompt(question: str, corpus: str) -> str:
+    """The CONSTRAINED consult (the foot-gun mitigation, tuned live 2026-08-02):
+    force EXACTLY ONE search_library call and forbid read_note. This is what makes
+    Mimir converge in ~3-4s — the softer "do one search" phrasing still let it loop
+    read_note↔search past the 25s ceiling on conversational (non-keyword) questions."""
+    return (
+        f"You may call search_library EXACTLY ONCE and call no other tool "
+        f"(do NOT call read_note). Query: {question} (corpus: {corpus}). "
+        f"After the single search returns, immediately output the most relevant "
+        f"result excerpts verbatim and stop."
+    )
+
+
+async def _run_consult(client: WorldtreeClient, prompt: str) -> str:
+    """Drive one throwaway Mimir turn and return its grounded answer text. Prefers
+    the DoneEvent's authoritative full `response`, falling back to the accumulated
+    text deltas if the terminal event carried none."""
+    session = await client.sessions.create(
+        {"agent_id": _MIMIR_AGENT_ID, "end_user_id": _KB_BRIDGE_END_USER}
+    )
+    session_id = session["session_id"]
+    parts: list[str] = []
+    final = ""
+    # aclosing: we break on DoneEvent without exhausting the stream; deterministically
+    # close the SDK generator so it can't finalize mid-flight against a closing
+    # transport (the "async generator already running" race the live smoke surfaced).
+    async with aclosing(client.sessions.stream_turn(session_id, prompt)) as stream:
+        async for ev in stream:
+            if isinstance(ev, TextEvent) and isinstance(ev.content, str):
+                parts.append(ev.content)
+            elif isinstance(ev, DoneEvent):
+                if isinstance(ev.response, str) and ev.response.strip():
+                    final = ev.response
+                break
+    return (final or "".join(parts)).strip()
+
+
+async def pin_kb_context(
+    question: str, agent_id: str | None, *, client: WorldtreeClient
+) -> list[dict]:
+    """Consumer-side retrieval + memory_context pinning (FN pin_kb_context / DEC-6).
+
+    Returns memory_context items for the character's turn, or [] on any miss. Total:
+    an allowlist miss, blank question, empty/failed/timed-out consult, or ANY
+    exception → [] — never raises to the turn path (INV-KB-3), never blocks the turn.
+    CancelledError (a browser disconnect) is intentionally NOT caught — it propagates.
+    """
+    corpus = _KB_INTERVIEW_CORPORA.get(agent_id or "")
+    if corpus is None:  # INV-KB-4: not an opted-in interview character
+        return []
+    if not (question and question.strip()):
+        return []
+    try:
+        extract = await asyncio.wait_for(
+            _run_consult(client, _consult_prompt(question, corpus)),
+            timeout=_CONSULT_TIMEOUT_S,
+        )
+    except Exception:
+        # timeout / transport / protocol / any consult failure → degrade silently
+        # (INV-KB-3). CancelledError is a BaseException, so it is NOT swallowed here.
+        return []
+    if not extract:
+        return []
+    return [{"kind": "corpus_reference", "text": extract[:_MAX_EXTRACT_CHARS]}]
diff --git a/src/ratatoskr/web/server.py b/src/ratatoskr/web/server.py
index 3c3adf7..a7c3d1c 100644
--- a/src/ratatoskr/web/server.py
+++ b/src/ratatoskr/web/server.py
@@ -37,6 +37,7 @@ from worldtree_sdk import (
 from ratatoskr import local_agents as _local_agents
 from ratatoskr import wt
 from ratatoskr.first_message import seed_preset_first_message
+from ratatoskr.kb_bridge import pin_kb_context
 from ratatoskr.sessions import (
     AgentNotAvailable,
     AgentNotFound,
@@ -271,6 +272,10 @@ class TurnHandle:
     content: str
     status: str = "queued"  # queued | streaming | done | error | cancelled
     upstream_turn_id: int | None = None
+    # The browser-selected agent for this session — carried so the turn path can gate
+    # the retire-ready KB-recall bridge on the interview-character allowlist (INV-KB-4).
+    # Optional: absent (older client / non-character turn) simply means no KB consult.
+    agent_id: str | None = None
 
 
 # Process-local monotonic turn_id counter. Per FN submit_turn_endpoint
@@ -291,9 +296,14 @@ async def _submit_turn_endpoint(request: Request) -> JSONResponse:
     if not content:
         return JSONResponse({"error_code": "missing_content"}, status_code=400)
     session_id = request.path_params["session_id"]
+    # The browser may include agent_id so the stream path can gate the KB-recall
+    # bridge (INV-KB-4); optional — absent → no consult (a non-str is ignored).
+    agent_id = body.get("agent_id") if isinstance(body, dict) else None
+    if not isinstance(agent_id, str):
+        agent_id = None
     turn_id = next(_TURN_COUNTER)
     request.app.state.turn_registry[(session_id, turn_id)] = TurnHandle(
-        session_id=session_id, turn_id=turn_id, content=content,
+        session_id=session_id, turn_id=turn_id, content=content, agent_id=agent_id,
     )
     return JSONResponse({"turn_id": turn_id}, status_code=200)
 
@@ -364,8 +374,18 @@ async def _stream_turn_endpoint(request: Request) -> StreamingResponse:
         wt_client = _wt_client(client)
         try:
             handle.status = "streaming"
+            # Retire-ready KB-recall bridge (INV-KB-1: THE single call-site). Consumer-side
+            # Mimir retrieval for opted-in interview characters (INV-KB-4); [] for everyone
+            # else. Bounded + degrade-to-[] internally (INV-KB-3), so it never blocks or
+            # fails the turn — a miss just streams the turn with no pinned corpus context.
+            memory_context = await pin_kb_context(
+                handle.content, handle.agent_id, client=wt_client
+            )
             try:
-                async for event in wt.stream_turn(wt_client, session_id, handle.content):
+                async for event in wt.stream_turn(
+                    wt_client, session_id, handle.content,
+                    memory_context=memory_context or None,
+                ):
                     # v0.16.0: capture the upstream (Worldtree-assigned) turn_id from
                     # the first event so cancel paths target the real upstream turn,
                     # not our local counter — parsed from the composite sse_id.
diff --git a/src/ratatoskr/web/static/index.html b/src/ratatoskr/web/static/index.html
index b17f1b2..4a3bc75 100644
--- a/src/ratatoskr/web/static/index.html
+++ b/src/ratatoskr/web/static/index.html
@@ -1715,7 +1715,7 @@ async function submitPrompt() {
   try {
     r = await fetch("/api/turns/" + encodeURIComponent(state.sessionId), {
       method: "POST", headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({ content }),
+      body: JSON.stringify({ content, agent_id: state.agentId || null }),
     });
   } catch (e) { tickerAdd("err", "submit failed", "network_error"); return; }
   if (r.status !== 200) { tickerAdd("err", "submit failed", "status=" + r.status); return; }
diff --git a/src/ratatoskr/wt.py b/src/ratatoskr/wt.py
index 3a0e802..2cd41df 100644
--- a/src/ratatoskr/wt.py
+++ b/src/ratatoskr/wt.py
@@ -278,13 +278,22 @@ async def get_session_tools(
 
 
 async def stream_turn(
-    client: WorldtreeClient, session_id: str, content: str
+    client: WorldtreeClient,
+    session_id: str,
+    content: str,
+    *,
+    memory_context: list[dict] | None = None,
 ) -> AsyncGenerator[wtsdk.TurnEvent, None]:
     """Drive the resilient turn stream (auto-resume; absorbs the old `reconnect_turn`)
     and yield the SDK's `TurnEvent`s, re-wrapping the stream's TERMINAL SDK errors
     into ratatoskr's caller-semantic exceptions (INV-CUT-2 / DEC-2 — the presenter
     keeps catching ratatoskr's types).
 
+    `memory_context` (optional) is forwarded verbatim into the turn POST body (the SDK
+    passes it through). Ratatoskr uses it for the retire-ready KB-recall bridge (the
+    Donut voiced-interview contract); defaulting to None keeps every other caller and
+    the bridge's own retirement inert.
+
     The SDK's `stream_turn` retries only the transport-drop class internally; a
     resume failure / protocol violation / connect failure surfaces unchanged
     (B-RES-6), and a drop that exhausts the reconnect budget surfaces as
@@ -293,7 +302,9 @@ async def stream_turn(
     SDK's `ConnectFailed`, so they are caught before the generic `ConnectFailed`.
     """
     try:
-        async for event in client.sessions.stream_turn(session_id, content):
+        async for event in client.sessions.stream_turn(
+            session_id, content, memory_context=memory_context
+        ):
             yield event
     except wtsdk.SessionRetired as exc:
         # Fresh-mode 410 → the session is gone server-side; a generic API failure.
diff --git a/tests/test_kb_bridge.py b/tests/test_kb_bridge.py
new file mode 100644
index 0000000..ccdaa60
--- /dev/null
+++ b/tests/test_kb_bridge.py
@@ -0,0 +1,143 @@
+"""Tests for ratatoskr.kb_bridge per docs/contracts/donut_voiced_interview.contract.md
+(slice 3 — the retire-ready KB-recall bridge, FN pin_kb_context).
+
+The bridge drives a passed-in WorldtreeClient; unit tests use a fake client + real
+SDK event instances. No live network.
+
+Invariants exercised:
+  - INV-KB-3: any miss / failure / timeout / exception → [], never raises to the turn.
+  - INV-KB-4: only the opted-in interview character(s) consult; others short-circuit.
+  - The consult prompt is the CONSTRAINED shape (the >60s-loop foot-gun mitigation).
+"""
+
+import asyncio
+
+import pytest
+from worldtree_sdk import DoneEvent, TextEvent
+
+from ratatoskr import kb_bridge
+from ratatoskr.kb_bridge import pin_kb_context
+
+_DONUT = "ratatoskr:donut"
+
+
+def _text(s: str) -> TextEvent:
+    return TextEvent(type="text", sse_id="2:1", turn_id=2, raw={"content": s}, content=s)
+
+
+def _done(response: str = "") -> DoneEvent:
+    return DoneEvent(
+        type="done", sse_id="2:9", turn_id=2, raw={}, phase="Finishing",
+        response=response, model="gen", duration_ms=1, usage={},
+    )
+
+
+class _FakeSessions:
+    def __init__(self, *, events=None, create_result=None, stream_error=None,
+                 create_error=None, stream_delay=0.0):
+        self.events = events or []
+        self.create_result = create_result or {"session_id": "kb-sess-1"}
+        self.stream_error = stream_error
+        self.create_error = create_error
+        self.stream_delay = stream_delay
+        self.calls: list = []
+
+    async def create(self, body=None, **kw):
+        self.calls.append(("create", body, kw))
+        if self.create_error is not None:
+            raise self.create_error
+        return self.create_result
+
+    def stream_turn(self, *args, **kw):
+        self.calls.append(("stream_turn", args, kw))
+        return self._gen()
+
+    async def _gen(self):
+        if self.stream_delay:
+            await asyncio.sleep(self.stream_delay)
+        for e in self.events:
+            yield e
+        if self.stream_error is not None:
+            raise self.stream_error
+
+
+class _FakeClient:
+    def __init__(self, sessions: _FakeSessions):
+        self.sessions = sessions
+
+
+class TestAllowlistGate:
+    async def test_unlisted_agent_returns_empty_without_consulting(self) -> None:
+        fake = _FakeSessions(events=[_text("x"), _done("x")])
+        out = await pin_kb_context("who is Carl?", "mimir", client=_FakeClient(fake))
+        assert out == []
+        assert fake.calls == []  # INV-KB-4: no consult for a non-interview agent
+
+    async def test_none_agent_returns_empty(self) -> None:
+        fake = _FakeSessions()
+        out = await pin_kb_context("q", None, client=_FakeClient(fake))
+        assert out == [] and fake.calls == []
+
+    async def test_blank_question_returns_empty(self) -> None:
+        fake = _FakeSessions()
+        out = await pin_kb_context("   ", _DONUT, client=_FakeClient(fake))
+        assert out == [] and fake.calls == []
+
+
+class TestConsult:
+    async def test_happy_pins_done_response_as_corpus_reference(self) -> None:
+        fake = _FakeSessions(events=[_text("partial "), _done("The full grounded passage.")])
+        out = await pin_kb_context("what happened?", _DONUT, client=_FakeClient(fake))
+        assert out == [{"kind": "corpus_reference", "text": "The full grounded passage."}]
+
+    async def test_falls_back_to_text_deltas_when_no_done_response(self) -> None:
+        fake = _FakeSessions(events=[_text("alpha "), _text("beta"), _done("")])
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert out == [{"kind": "corpus_reference", "text": "alpha beta"}]
+
+    async def test_constrained_consult_prompt_and_mimir_target(self) -> None:
+        fake = _FakeSessions(events=[_done("p")])
+        await pin_kb_context("who is Princess Donut?", _DONUT, client=_FakeClient(fake))
+        create_body = fake.calls[0][1]
+        assert create_body["agent_id"] == "mimir"
+        stream_prompt = fake.calls[1][1][1]  # ("stream_turn", (sid, prompt), {})
+        # The foot-gun mitigation: EXACTLY one search_library call, no read_note.
+        assert "EXACTLY ONCE" in stream_prompt
+        assert "read_note" in stream_prompt                  # forbids the loop tool
+        assert "Dungeon Crawler Carl" in stream_prompt      # corpus scoping
+        assert "who is Princess Donut?" in stream_prompt     # the question
+
+    async def test_empty_consult_returns_empty(self) -> None:
+        fake = _FakeSessions(events=[_text("   "), _done("   ")])
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert out == []
+
+    async def test_extract_bounded_to_max_chars(self) -> None:
+        long = "x" * 5000
+        fake = _FakeSessions(events=[_done(long)])
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert len(out[0]["text"]) == kb_bridge._MAX_EXTRACT_CHARS
+
+
+class TestDegradation:
+    async def test_stream_error_degrades_to_empty(self) -> None:
+        fake = _FakeSessions(events=[_text("x")], stream_error=RuntimeError("boom"))
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert out == []  # INV-KB-3: never raises to the turn path
+
+    async def test_create_error_degrades_to_empty(self) -> None:
+        fake = _FakeSessions(create_error=RuntimeError("no session"))
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert out == []
+
+    async def test_timeout_degrades_to_empty(self, monkeypatch) -> None:
+        monkeypatch.setattr(kb_bridge, "_CONSULT_TIMEOUT_S", 0.05)
+        fake = _FakeSessions(events=[_done("late")], stream_delay=5.0)
+        out = await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
+        assert out == []  # a slow/looping consult never hangs the turn
+
+    async def test_cancellation_is_not_swallowed(self) -> None:
+        # CancelledError (a browser disconnect) MUST propagate, not degrade to [].
+        fake = _FakeSessions(create_error=asyncio.CancelledError())
+        with pytest.raises(asyncio.CancelledError):
+            await pin_kb_context("q", _DONUT, client=_FakeClient(fake))
diff --git a/tests/test_web_server.py b/tests/test_web_server.py
index 63e2ca7..c1c6ba0 100644
--- a/tests/test_web_server.py
+++ b/tests/test_web_server.py
@@ -1526,3 +1526,58 @@ class TestTtsEndpoint:
         app = create_app(_mock_client_factory(), tts_url=self._TTS)
         resp = TestClient(app).post("/api/tts", json={"text": "hi"})
         assert resp.status_code == 503
+
+
+class TestKbBridgeWiring:
+    """The KB-recall bridge seam (slice 3, INV-KB-1): the turn path calls
+    pin_kb_context and forwards its result as memory_context on the turn POST.
+    Proves the ONE call-site is wired end-to-end (the consult itself is unit-tested
+    in test_kb_bridge.py; here we stub it to assert the plumbing)."""
+
+    @respx.mock
+    def test_interview_agent_pins_context_into_turn_post(self, monkeypatch) -> None:
+        from ratatoskr.web import server
+        from ratatoskr.web.server import create_app
+
+        seen = {}
+
+        async def _fake_pin(question, agent_id, *, client):
+            seen["question"] = question
+            seen["agent_id"] = agent_id
+            return [{"kind": "corpus_reference", "text": "grounded passage"}]
+
+        monkeypatch.setattr(server, "pin_kb_context", _fake_pin)
+        route = respx.post("https://w.example/sessions/s-1/messages").mock(
+            return_value=_sse_resp(_sse_chunk("42:1", _DONE_BODY))
+        )
+        c = TestClient(create_app(_mock_client_factory()))
+        tid = c.post(
+            "/api/turns/s-1",
+            json={"content": "who is Carl?", "agent_id": "ratatoskr:donut"},
+        ).json()["turn_id"]
+        with c.stream("GET", f"/api/turns/s-1/stream?turn_id={tid}") as resp:
+            b"".join(resp.iter_bytes())
+        # The browser-sent agent_id reached the bridge, and its result rode the POST.
+        assert seen == {"question": "who is Carl?", "agent_id": "ratatoskr:donut"}
+        body = json.loads(route.calls.last.request.content)
+        assert body["memory_context"] == [
+            {"kind": "corpus_reference", "text": "grounded passage"}
+        ]
+
+    @respx.mock
+    def test_non_interview_agent_pins_nothing(self) -> None:
+        # Real pin_kb_context: a non-allowlisted agent short-circuits to [] (no consult),
+        # so memory_context is absent from the turn POST body (SDK omits None).
+        from ratatoskr.web.server import create_app
+
+        route = respx.post("https://w.example/sessions/s-1/messages").mock(
+            return_value=_sse_resp(_sse_chunk("42:1", _DONE_BODY))
+        )
+        c = TestClient(create_app(_mock_client_factory()))
+        tid = c.post(
+            "/api/turns/s-1", json={"content": "hi", "agent_id": "mimir"}
+        ).json()["turn_id"]
+        with c.stream("GET", f"/api/turns/s-1/stream?turn_id={tid}") as resp:
+            b"".join(resp.iter_bytes())
+        body = json.loads(route.calls.last.request.content)
+        assert "memory_context" not in body or body["memory_context"] is None
diff --git a/tests/test_wt.py b/tests/test_wt.py
index c016927..99675b8 100644
--- a/tests/test_wt.py
+++ b/tests/test_wt.py
@@ -360,7 +360,15 @@ class TestStreamTurn:
         fake = _FakeSessions(events=[e1, e2])
         got = await _drain(stream_turn(_wt(fake), "s-1", "hello"))
         assert got == [e1, e2]
-        assert fake.calls[-1] == ("stream_turn", ("s-1", "hello"), {})
+        # memory_context defaults to None (forwarded verbatim; inert for every caller
+        # that doesn't pin corpus context via the KB-recall bridge).
+        assert fake.calls[-1] == ("stream_turn", ("s-1", "hello"), {"memory_context": None})
+
+    async def test_forwards_memory_context_verbatim(self) -> None:
+        mc = [{"kind": "corpus_reference", "text": "the pinned passage"}]
+        fake = _FakeSessions(events=[])
+        await _drain(stream_turn(_wt(fake), "s-1", "hi", memory_context=mc))
+        assert fake.calls[-1] == ("stream_turn", ("s-1", "hi"), {"memory_context": mc})
 
     async def test_session_retired_maps_to_session_api_failed(self) -> None:
         fake = _FakeSessions(