feat(core,frontends,agents): agent kinds, frontend routing, anthropic on conversations, vault mirror

src/beaver_gateway/agents/claude.py

@@ -13,8 +13,9 @@ from __future__ import annotations
 
 from collections.abc import Mapping  # noqa: TC003 - pydantic runtime
 from pathlib import Path  # noqa: TC003 - pydantic runtime
+from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from beaver_gateway.agents.base import BaseAgent
 from beaver_gateway.core.prompt import PromptSource  # noqa: TC001 - pydantic runtime
@@ -57,6 +58,10 @@ class ClaudeAgent(BaseAgent):
     """Per conversation kind (``master``/``branch``/``deep``/``job``/``fork``)
     assembly; falls back to ``prompt_sources``. Constant per kind (§3.12)."""
 
+    kinds: tuple[str, ...] = ()
+    """Conversation kinds this agent serves; ``create``/``spawn`` reject the
+    rest. Defaults to the keys of ``prompt_sources_by_kind`` or ``("deep",)``."""
+
     skill_sets: tuple[Path, ...] = ()
     gateway_tools: tuple[str, ...] = ()
     """Gateway tools exposed in-process (``read_conversation``, ``spawn``,
@@ -64,5 +69,16 @@ class ClaudeAgent(BaseAgent):
 
     options: ClaudeOptions = Field(default_factory=ClaudeOptions)
 
+    @model_validator(mode="before")
+    @classmethod
+    def _default_kinds(cls, data: Any) -> Any:
+        if isinstance(data, dict) and not data.get("kinds"):
+            by_kind = data.get("prompt_sources_by_kind") or {}
+            data = {**data, "kinds": tuple(by_kind) or ("deep",)}
+        return data
+
     def prompt_for(self, kind: str) -> tuple[PromptSource, ...]:
         return self.prompt_sources_by_kind.get(kind, self.prompt_sources)
+
+    def serves(self, kind: str) -> bool:
+        return kind in self.kinds

src/beaver_gateway/backends/claude_sdk.py

@@ -78,7 +78,12 @@ from beaver_gateway.core.events import (
     build_tool_use_block_start,
 )
 from beaver_gateway.core.sessions import Session, SessionClient, SessionPool
-from beaver_gateway.core.transcript import build_entries, close_open_tool_uses
+from beaver_gateway.core.transcript import (
+    build_entries,
+    close_open_tool_uses,
+    fingerprint,
+    text_of,
+)
 from beaver_gateway.core.turn_capture import TurnCapture, TurnUsage
 
 if TYPE_CHECKING:
@@ -700,41 +705,8 @@ def _mcp_disallowed(
     return out
 
 
-def fingerprint(messages: Iterable[Mapping[str, Any]]) -> str:
-    turns: list[tuple[str, str]] = []
-    for message in messages:
-        text = _text_of(message.get("content"))
-        if not text:
-            continue
-        role = str(message.get("role", ""))
-        if turns and turns[-1][0] == role:
-            turns[-1] = (role, turns[-1][1] + "\n" + text)
-        else:
-            turns.append((role, text))
-    digest = hashlib.sha1(usedforsecurity=False)
-    for role, text in turns:
-        digest.update(role.encode("utf-8"))
-        digest.update(b"\x00")
-        digest.update(text.strip().encode("utf-8"))
-        digest.update(b"\x01")
-    return digest.hexdigest()
-
-
-def _text_of(content: Any) -> str:
-    if isinstance(content, str):
-        return content
-    if isinstance(content, list):
-        parts = [
-            str(b.get("text", ""))
-            for b in content
-            if isinstance(b, Mapping) and b.get("type") == "text"
-        ]
-        return "\n".join(p for p in parts if p)
-    return ""
-
-
 def _prompt_text(content: Any) -> str:
-    text = _text_of(content)
+    text = text_of(content)
     if not text:
         msg = "user message has no text content"
         raise ValueError(msg)

src/beaver_gateway/cli.py

@@ -170,6 +170,7 @@ async def _async_main() -> None:
             pool=pool,
             store=session_store,
             texts=gateway.texts,
+            frontends=gateway.frontends,
         )
         late.conversations = conversations
 

src/beaver_gateway/core/conversations.py

@@ -63,6 +63,7 @@ if TYPE_CHECKING:
     from beaver_gateway.core.injects import Priority
     from beaver_gateway.core.registry import AgentRegistry
     from beaver_gateway.core.sessions import SessionPool
+    from beaver_gateway.frontends.base import Frontend
     from beaver_gateway.storage.db import Database
 
 __all__ = [
@@ -138,6 +139,7 @@ class Conversations:
         pool: SessionPool,
         store: SessionStore,
         texts: ConversationTexts | None = None,
+        frontends: Sequence[Frontend] = (),
         normal_window: float = 3600.0,
         idle_days: Sequence[int] = (2,),
         idle_interval: float = 3600.0,
@@ -149,6 +151,7 @@ class Conversations:
         self._pool = pool
         self._store = store
         self._texts = texts or ConversationTexts()
+        self._frontends = [f for f in frontends if f.name]
         self._normal_window = normal_window
         self._idle_days = tuple(sorted(idle_days))
         self._idle_interval = idle_interval
@@ -185,7 +188,9 @@ class Conversations:
         if kind not in KINDS:
             msg = f"unknown conversation kind {kind!r}"
             raise ValueError(msg)
-        self._claude_agent(agent)
+        if not self._claude_agent(agent).serves(kind):
+            msg = f"agent {agent!r} does not serve kind {kind!r}"
+            raise ValueError(msg)
         now = datetime.now(UTC)
         row = Conversation(
             frontend=origin,
@@ -251,6 +256,9 @@ class Conversations:
         external_id: str,
         visible: bool = True,
     ) -> ConversationBinding:
+        if conv.kind not in self.frontend(frontend).kinds:
+            msg = f"frontend {frontend!r} does not show kind {conv.kind!r}"
+            raise ValueError(msg)
         async with self._db.session() as session:
             existing = list(
                 (
@@ -385,13 +393,41 @@ class Conversations:
         out["busy"] = live.busy if live is not None else False
         return out
 
+    # ---- routing -------------------------------------------------------
+
+    @property
+    def frontends(self) -> list[Frontend]:
+        return list(self._frontends)
+
+    def frontend(self, name: str) -> Frontend:
+        for fe in self._frontends:
+            if fe.name == name:
+                return fe
+        msg = f"unknown frontend {name!r}"
+        raise ValueError(msg)
+
+    def default_agent(self, kind: str) -> str | None:
+        for fe in self._frontends:
+            if kind in fe.kinds and (agent := fe.agent_for(kind)):
+                return agent
+        return None
+
+    async def materialize(self, conv: Conversation) -> ConversationBinding | None:
+        for fe in self._frontends:
+            if conv.kind not in fe.kinds:
+                continue
+            binding = await fe.materialize(conv)
+            if binding is not None:
+                return binding
+        return None
+
     # ---- §3.1 api ------------------------------------------------------
 
     async def spawn(
         self,
         *,
         kind: str,
-        agent: str,
+        agent: str | None = None,
         seed: str = "clean",
         parent: Conversation | None = None,
         text: str | None = None,
@@ -405,6 +441,12 @@ class Conversations:
         if seed == "brief" and not text:
             msg = "seed=brief needs text"
             raise ValueError(msg)
+        if agent is None and kind == "branch" and parent is not None:
+            agent = parent.agent_name
+        agent = agent or self.default_agent(kind)
+        if agent is None:
+            msg = f"no default agent for kind {kind!r}; pass `agent`"
+            raise ValueError(msg)
         session_id: str | None = None
         if seed == "copy":
             if parent is None or parent.session_id is None:
@@ -421,6 +463,7 @@ class Conversations:
             origin=origin,
             session_id=session_id,
         )
+        await self.materialize(conv)
         prompt = await self._seed_text(
             SeedContext(
                 kind=kind, seed=seed, agent=agent, parent=parent, text=text, title=title
@@ -888,6 +931,8 @@ class Conversations:
                 conversation_id=conv.external_id,
                 turn_id=turn_id,
                 item=head.id,
+                source="queue",
+                prompt=prompt,
                 text=text,
             )
 

src/beaver_gateway/core/gateway_tools.py

@@ -75,11 +75,14 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]:
         "Open a new conversation of the given kind (branch = your own thread, "
         "deep = a long research chat, job = a headless task). `seed` is how it "
         "starts: clean (nothing), morning (handout), copy (copy of this "
-        "conversation, last `window` turns), brief (your `text`). Returns the id.",
+        "conversation, last `window` turns), brief (your `text`). A branch "
+        "keeps your agent, other kinds get their frontend's default unless "
+        "`agent` says otherwise. Returns the id.",
         {
             "type": "object",
             "properties": {
                 "kind": {"type": "string", "enum": ["branch", "deep", "job"]},
+                "agent": {"type": "string", "description": "agent name"},
                 "seed": {
                     "type": "string",
                     "enum": ["clean", "morning", "copy", "brief"],
@@ -94,16 +97,19 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]:
     )
     async def spawn(args: dict[str, Any]) -> dict[str, Any]:
         parent = await current()
-        child = await conversations.spawn(
-            kind=str(args["kind"]),
-            agent=parent.agent_name,
-            seed=str(args.get("seed") or "clean"),
-            parent=parent,
-            text=args.get("text"),
-            title=args.get("title"),
-            window=args.get("window"),
-            origin="mcp",
-        )
+        try:
+            child = await conversations.spawn(
+                kind=str(args["kind"]),
+                agent=args.get("agent"),
+                seed=str(args.get("seed") or "clean"),
+                parent=parent,
+                text=args.get("text"),
+                title=args.get("title"),
+                window=args.get("window"),
+                origin="mcp",
+            )
+        except (ValueError, LookupError) as exc:
+            return _error(str(exc))
         return _text(f"spawned {child.kind} {child.external_id}")
 
     @tool(

src/beaver_gateway/core/transcript.py

@@ -15,7 +15,9 @@ anything that needs Anthropic-shape history out of a mirrored transcript.
 
 from __future__ import annotations
 
+import hashlib
 import uuid as _uuid
+from collections.abc import Mapping
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
@@ -25,17 +27,19 @@ except ImportError:  # pragma: no cover
     _cli_version = "2.1.248"
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable, Mapping
+    from collections.abc import Iterable
 
 __all__ = [
     "CLI_VERSION",
     "build_entries",
     "close_open_tool_uses",
+    "fingerprint",
     "messages_from_entries",
     "open_tool_uses",
     "prompt_count",
     "render_messages",
     "strip_tool_entries",
+    "text_of",
     "window_entries",
 ]
 
@@ -503,3 +507,37 @@ def _text_of_content(content: Any) -> str:
             if isinstance(b, dict) and b.get("type") == "text" and b.get("text")
         )
     return ""
+
+
+def fingerprint(messages: Iterable[Mapping[str, Any]]) -> str:
+    """Text-only hash of a history; stateless callers are keyed by it."""
+    turns: list[tuple[str, str]] = []
+    for message in messages:
+        text = text_of(message.get("content"))
+        if not text:
+            continue
+        role = str(message.get("role", ""))
+        if turns and turns[-1][0] == role:
+            turns[-1] = (role, turns[-1][1] + "\n" + text)
+        else:
+            turns.append((role, text))
+    digest = hashlib.sha1(usedforsecurity=False)
+    for role, text in turns:
+        digest.update(role.encode("utf-8"))
+        digest.update(b"\x00")
+        digest.update(text.strip().encode("utf-8"))
+        digest.update(b"\x01")
+    return digest.hexdigest()
+
+
+def text_of(content: Any) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = [
+            str(b.get("text", ""))
+            for b in content
+            if isinstance(b, Mapping) and b.get("type") == "text"
+        ]
+        return "\n".join(p for p in parts if p)
+    return ""

src/beaver_gateway/frontends/anthropic.py

@@ -4,39 +4,44 @@ Exposes the gateway as an Anthropic-compatible Messages endpoint, so any
 client that already speaks Anthropic (Cursor, Cline, the official SDK,
 ``curl``) can hit a configured agent by passing its name as ``model``.
 
-The frontend is intentionally thin: it authenticates the bearer token,
-resolves ``body.model`` to an agent + its backend, and then either
-streams the backend's events straight to SSE or accumulates them into a
-single ``Message`` for ``stream=false`` callers. All provider quirks
-already live in the backend adapters; we don't translate here.
-
-Phase 1.4 wires only ``RaycastAgent`` through ``RaycastBackend``;
-``ClaudeAgent`` lands in Phase 2 and will plug into the same dispatch
-table without changes to this module.
+A Claude agent behind this endpoint is a ``deep`` conversation: the client
+knows nothing about our ids, so the text fingerprint of the history it
+sends is the ``(anthropic, fingerprint)`` binding of the conversation,
+rebound after every turn to the fingerprint the next request will carry.
+A history nobody has seen becomes a new conversation, materialized by the
+home frontend of ``deep`` (the vault file), and every reply is published
+on the bus so that file follows the chat. Other agents (Raycast) stay
+stateless and are only archived through ``turn_log_handlers``.
 """
 
 from __future__ import annotations
 
 import json
 import logging
+import uuid
 from typing import TYPE_CHECKING, Any
 
 from fastapi import FastAPI, HTTPException, Request, status
 from fastapi.responses import JSONResponse, StreamingResponse
 
+from beaver_gateway.agents.claude import ClaudeAgent
 from beaver_gateway.core import audit
+from beaver_gateway.core.transcript import fingerprint, text_of
+from beaver_gateway.core.turn_capture import TurnCapture
 from beaver_gateway.core.turn_record import TurnRecord
-from beaver_gateway.frontends._accumulate import StreamAccumulator, accumulate
+from beaver_gateway.frontends._accumulate import StreamAccumulator
 from beaver_gateway.frontends._auth import require_token
 from beaver_gateway.frontends.base import Frontend
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator
+    from collections.abc import AsyncIterator, Awaitable, Callable
 
     from anthropic.types import Message, MessageParam
 
+    from beaver_gateway.core.conversations import Conversations
     from beaver_gateway.core.events import MessageStreamEvent
     from beaver_gateway.frontends.base import GatewayRuntime
+    from beaver_gateway.storage.models import Conversation
 
 
 _log = logging.getLogger("beaver_gateway.frontends.anthropic")
@@ -44,10 +49,16 @@ _log = logging.getLogger("beaver_gateway.frontends.anthropic")
 
 __all__ = ["AnthropicMessagesFrontend"]
 
+FRONTEND = "anthropic"
+_TITLE_LEN = 80
+
 
 class AnthropicMessagesFrontend(Frontend):
     """FastAPI app behind ``POST /v1/messages`` + ``GET /v1/models``."""
 
+    name = FRONTEND
+    kinds = ("deep",)
+
     def __init__(
         self,
         *,
@@ -57,14 +68,6 @@ class AnthropicMessagesFrontend(Frontend):
     ) -> None:
         self.host = host
         self.port = port
-        # External URL prefix the reverse proxy (Caddy/nginx/Cloudflare/…)
-        # uses to reach this frontend, e.g. ``https://api.example.com/ai``.
-        # The frontend's internal paths (``/v1/messages``, ``/v1/models``)
-        # are appended to it when the admin dashboard renders copy-pastable
-        # URLs. Trailing slash is stripped so the concatenation is
-        # idempotent. ``None`` means "advertise raw ``host:port``" (dev /
-        # no proxy) — the dashboard then derives the base from the
-        # browser's own request hostname.
         self.public_base_url = public_base_url.rstrip("/") if public_base_url else None
         self._runtime: GatewayRuntime | None = None
         self._app: FastAPI | None = None
@@ -74,9 +77,6 @@ class AnthropicMessagesFrontend(Frontend):
         self._app = self._build_app(runtime)
 
     async def serve(self) -> None:
-        # Local import: uvicorn pulls in a lot, no reason to load it when
-        # something else (a test, a script) imports this module just for
-        # the FastAPI factory.
         import uvicorn
 
         if self._app is None:
@@ -134,9 +134,6 @@ class AnthropicMessagesFrontend(Frontend):
 
             backend = runtime.backends.get(agent.name)
             if backend is None:
-                # Agent exists in config but its backend isn't wired in
-                # this phase (e.g. ClaudeAgent before Phase 2, or a
-                # RaycastAgent without RAYCAST_BEARER set at startup).
                 raise HTTPException(
                     status.HTTP_503_SERVICE_UNAVAILABLE,
                     f"no backend configured for agent {agent.name!r}",
@@ -144,7 +141,12 @@ class AnthropicMessagesFrontend(Frontend):
 
             messages = body.get("messages") or []
             system = body.get("system")
+            system_str = system if isinstance(system, str) else None
             stream_flag = bool(body.get("stream", False))
+            if not messages or messages[-1].get("role") != "user":
+                raise HTTPException(
+                    status.HTTP_400_BAD_REQUEST, "the last message must be a user turn"
+                )
 
             _log.info(
                 "messages: actor=%s agent=%s stream=%s msgs=%d",
@@ -153,10 +155,6 @@ class AnthropicMessagesFrontend(Frontend):
                 stream_flag,
                 len(messages),
             )
-            # Record at request acceptance, not at stream completion:
-            # a long streaming response can be aborted mid-flight by
-            # the client, and we still want the row in the audit trail.
-            # Detail stays small — no message bodies, no system prompt.
             await audit.log(
                 runtime,
                 actor=f"token:{token_name}",
@@ -166,74 +164,139 @@ class AnthropicMessagesFrontend(Frontend):
                 msgs=len(messages),
             )
 
-            # Forward per-request knobs the Anthropic body may carry —
-            # backend adapters layer these over per-agent defaults. Only
-            # values explicitly present (not Anthropic-defaulted ones we
-            # never received) are forwarded, so the agent's default still
-            # wins when the caller omits the field.
-            options: dict[str, Any] = {}
-            if isinstance(body.get("temperature"), int | float):
-                options["temperature"] = body["temperature"]
+            if isinstance(agent, ClaudeAgent) and runtime.conversations is not None:
+                conv, history = await _resolve(runtime.conversations, agent, messages)
+                turn_id = f"turn_{uuid.uuid4().hex[:12]}"
+                capture = TurnCapture()
+                events = runtime.conversations.turn(
+                    conv,
+                    messages=messages,
+                    origin="user",
+                    capture=capture,
+                    turn_id=turn_id,
+                )
 
-            events = backend.complete(
-                agent=agent,
-                messages=messages,
-                system=system if isinstance(system, str) else None,
-                **options,
-            )
+                async def after(message: Message) -> None:
+                    await _finish(
+                        runtime,
+                        conv,
+                        messages=messages,
+                        message=message,
+                        capture=capture,
+                        turn_id=turn_id,
+                        history=history,
+                    )
+            else:
+                options: dict[str, Any] = {}
+                if isinstance(body.get("temperature"), int | float):
+                    options["temperature"] = body["temperature"]
+                events = backend.complete(
+                    agent=agent, messages=messages, system=system_str, **options
+                )
 
-            system_str = system if isinstance(system, str) else None
-
-            if stream_flag:
-                # Side-accumulate while streaming so we can still emit a
-                # ``TurnRecord`` after the response closes. The buffered
-                # ``Message`` lives only in this coroutine's frame; SSE
-                # bytes still flow to the client unchanged.
-                return StreamingResponse(
-                    _sse_and_broadcast(
-                        events,
-                        runtime=runtime,
+                async def after(message: Message) -> None:
+                    await _broadcast_turn(
+                        runtime,
                         agent_name=agent.name,
                         input_messages=messages,
+                        output_message=message,
                         system=system_str,
-                        model=model,
-                    ),
+                    )
+
+            if stream_flag:
+                return StreamingResponse(
+                    _sse(events, model=model, after=after),
                     media_type="text/event-stream",
                 )
-            message = await accumulate(events, model=model)
-            await _broadcast_turn(
-                runtime,
-                agent_name=agent.name,
-                input_messages=messages,
-                output_message=message,
-                system=system_str,
-            )
+            acc = StreamAccumulator()
+            async for ev in events:
+                acc.feed(ev)
+            message = acc.finalize(model=model)
+            await after(message)
             return JSONResponse(content=message.model_dump(mode="json"))
 
         return app
 
 
-async def _sse_and_broadcast(
+async def _resolve(
+    conversations: Conversations, agent: ClaudeAgent, messages: list[MessageParam]
+) -> tuple[Conversation, list[dict[str, str]] | None]:
+    prior = [dict(m) for m in messages[:-1]]
+    if prior:
+        conv = await conversations.find_bound(
+            frontend=FRONTEND, external_id=fingerprint(prior)
+        )
+        if conv is not None:
+            if conv.agent_name != agent.name:
+                raise HTTPException(
+                    status.HTTP_400_BAD_REQUEST,
+                    f"this chat runs on {conv.agent_name!r}, not {agent.name!r}",
+                )
+            return conv, None
+    first = next(
+        (text_of(m.get("content")) for m in messages if m.get("role") == "user"), ""
+    )
+    try:
+        conv = await conversations.create(
+            kind="deep",
+            agent=agent.name,
+            origin=FRONTEND,
+            title=" ".join(first.split())[:_TITLE_LEN] or None,
+        )
+    except ValueError as exc:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+    await conversations.materialize(conv)
+    history = [
+        {"role": str(m.get("role", "")), "text": text_of(m.get("content"))}
+        for m in prior
+    ]
+    return conv, history
+
+
+async def _finish(
+    runtime: GatewayRuntime,
+    conv: Conversation,
+    *,
+    messages: list[MessageParam],
+    message: Message,
+    capture: TurnCapture,
+    turn_id: str,
+    history: list[dict[str, str]] | None,
+) -> None:
+    reply = "\n\n".join(
+        getattr(b, "text", "")
+        for b in message.content
+        if getattr(b, "type", "") == "text"
+    ).strip()
+    tail = capture.synthesized_messages or [
+        {"role": "assistant", "content": [{"type": "text", "text": reply}]}
+    ]
+    await runtime.conversations.bind(
+        conv, frontend=FRONTEND, external_id=fingerprint([*messages, *tail])
+    )
+    runtime.bus.publish(
+        "reply",
+        conversation_id=conv.external_id,
+        turn_id=turn_id,
+        source=FRONTEND,
+        prompt=text_of(messages[-1].get("content")),
+        text=reply,
+        history=history,
+    )
+
+
+async def _sse(
     events: AsyncIterator[MessageStreamEvent],
     *,
-    runtime: GatewayRuntime,
-    agent_name: str,
-    input_messages: list[MessageParam],
-    system: str | None,
     model: str,
+    after: Callable[[Message], Awaitable[None]],
 ) -> AsyncIterator[bytes]:
-    r"""Serialize an event stream to SSE; broadcast a :class:`TurnRecord` after.
+    r"""Serialize an event stream to SSE, then hand the assembled ``Message`` on.
 
-    Each event becomes ``event: <type>\ndata: <json>\n\n`` — the shape
+    Each event becomes ``event: <type>\ndata: <json>\n\n`` - the shape
     the Anthropic SDK's SSE decoder expects. Errors mid-stream are
     swallowed into a synthetic ``error`` event so the client sees the
     failure rather than a hung connection.
-
-    The same events feed a :class:`StreamAccumulator` on the side so that
-    once the SSE response closes we can hand a fully-assembled
-    ``Message`` to every ``runtime.turn_log_handlers`` entry (the
-    markdown frontend's archive logger lives in there). Broadcast
-    failures are caught — they must never bubble up to the client.
     """
     acc = StreamAccumulator()
     try:
@@ -248,14 +311,10 @@ async def _sse_and_broadcast(
         )
         yield f"event: error\ndata: {err}\n\n".encode()
         return
-    message = acc.finalize(model=model)
-    await _broadcast_turn(
-        runtime,
-        agent_name=agent_name,
-        input_messages=input_messages,
-        output_message=message,
-        system=system,
-    )
+    try:
+        await after(acc.finalize(model=model))
+    except Exception:  # noqa: BLE001
+        _log.exception("post-turn hook failed")
 
 
 async def _broadcast_turn(
@@ -266,11 +325,6 @@ async def _broadcast_turn(
     output_message: Message,
     system: str | None,
 ) -> None:
-    """Fire each ``turn_log_handlers`` entry with a fresh :class:`TurnRecord`.
-
-    Handler exceptions are caught and logged — they're observability
-    plumbing, not part of the user-visible request path.
-    """
     if not runtime.turn_log_handlers:
         return
     record = TurnRecord(
@@ -278,7 +332,7 @@ async def _broadcast_turn(
         input_messages=list(input_messages),
         output_message=output_message,
         system=system,
-        source="anthropic",
+        source=FRONTEND,
     )
     for handler in runtime.turn_log_handlers:
         try:

src/beaver_gateway/frontends/api/frontend.py

@@ -34,7 +34,7 @@ from beaver_gateway.frontends.base import Frontend
 from beaver_gateway.storage.models import Usage
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator
+    from collections.abc import AsyncIterator, Mapping
 
     from beaver_gateway.core.conversations import Conversations
     from beaver_gateway.frontends.base import GatewayRuntime
@@ -48,18 +48,26 @@ SCOPE = "api"
 
 
 class ApiFrontend(Frontend):
+    name = "api"
+    kinds = ("master", "branch", "deep", "job")
+
     def __init__(
         self,
         *,
         host: str = "0.0.0.0",  # noqa: S104
         port: int = 8004,
         public_base_url: str | None = None,
+        default_agents: Mapping[str, str] | None = None,
     ) -> None:
         self.host = host
         self.port = port
         self.public_base_url = public_base_url.rstrip("/") if public_base_url else None
+        self.default_agents = dict(default_agents or {})
         self._app: FastAPI | None = None
 
+    def agent_for(self, kind: str) -> str | None:
+        return self.default_agents.get(kind)
+
     def configure(self, runtime: GatewayRuntime) -> None:
         if runtime.conversations is None or runtime.bus is None:
             msg = "ApiFrontend needs runtime.conversations and runtime.bus"
@@ -130,6 +138,30 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
     async def healthz() -> dict[str, str]:
         return {"status": "ok"}
 
+    @app.get("/api/agents")
+    async def list_agents(request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        return {
+            "agents": [
+                {
+                    "name": a.name,
+                    "model": a.model,
+                    "kinds": list(getattr(a, "kinds", ())),
+                }
+                for a in runtime.agents
+            ],
+            "frontends": [
+                {
+                    "name": fe.name,
+                    "kinds": list(fe.kinds),
+                    "default_agents": {
+                        k: fe.agent_for(k) for k in fe.kinds if fe.agent_for(k)
+                    },
+                }
+                for fe in conversations.frontends
+            ],
+        }
+
     @app.get("/api/conversations")
     async def list_conversations(request: Request) -> dict[str, Any]:
         await require_token(request, runtime, scope=SCOPE)
@@ -149,10 +181,10 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
             raise HTTPException(
                 status.HTTP_400_BAD_REQUEST, f"kind must be one of {KINDS[:-1]}"
             )
-        if not isinstance(agent, str) or agent not in runtime.agents:
-            raise HTTPException(
-                status.HTTP_400_BAD_REQUEST, "unknown or missing `agent`"
-            )
+        if agent is not None and (
+            not isinstance(agent, str) or agent not in runtime.agents
+        ):
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, "unknown `agent`")
         seed = str(data.get("seed") or "clean")
         if seed not in SEEDS:
             raise HTTPException(
@@ -170,13 +202,13 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
                 window=int_or_none(data, "window"),
                 origin="api",
             )
-        except ValueError as exc:
+        except (ValueError, LookupError) as exc:
             raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
         await audit.log(
             runtime,
             actor=f"token:{token}",
             kind="api_spawn",
-            agent_name=agent,
+            agent_name=conv.agent_name,
             conversation=conv.external_id,
             seed=seed,
         )
@@ -276,7 +308,9 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
             raise HTTPException(
                 status.HTTP_400_BAD_REQUEST, f"seed must be one of {SEEDS}"
             )
-        agent = str(data.get("agent") or parent.agent_name)
+        agent = data.get("agent")
+        if agent is not None and not isinstance(agent, str):
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, "`agent` must be a string")
         try:
             child = await conversations.spawn(
                 kind="branch",
@@ -294,7 +328,7 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
             runtime,
             actor=f"token:{token}",
             kind="api_branch",
-            agent_name=agent,
+            agent_name=child.agent_name,
             conversation=child.external_id,
             parent=parent.external_id,
             seed=seed,
@@ -350,12 +384,15 @@ def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
         data = await body_of(request)
         frontend = text_of(data, "frontend")
         external_id = text_of(data, "external_id")
-        await conversations.bind(
-            conv,
-            frontend=frontend,
-            external_id=external_id,
-            visible=bool(data.get("visible", True)),
-        )
+        try:
+            await conversations.bind(
+                conv,
+                frontend=frontend,
+                external_id=external_id,
+                visible=bool(data.get("visible", True)),
+            )
+        except ValueError as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
         return await conversations.describe(conv)
 
     @app.patch("/api/conversations/{public_id}/flags")

src/beaver_gateway/frontends/base.py

@@ -22,6 +22,7 @@ if TYPE_CHECKING:
     from beaver_gateway.core.registry import AgentRegistry, McpRegistry
     from beaver_gateway.core.turn_record import TurnRecord
     from beaver_gateway.storage import Database
+    from beaver_gateway.storage.models import Conversation, ConversationBinding
 
     TurnLogHandler = Callable[[TurnRecord], Awaitable[None]]
 
@@ -86,10 +87,30 @@ class GatewayRuntime:
 
 
 class Frontend(ABC):
-    """Listens on a port, dispatches into the gateway."""
+    """Listens on a port, dispatches into the gateway.
+
+    A frontend that shows conversations declares ``name`` (the binding
+    key) and ``kinds`` (which conversation kinds it shows);
+    ``core/conversations`` refuses to bind a conversation to a frontend
+    outside its declaration. The first frontend in declaration order
+    whose ``materialize`` returns a binding is the *home* of that kind:
+    ``spawn`` calls it so a new conversation gets a window (a vault file,
+    a Telegram topic). ``agent_for`` names the default agent for a kind
+    so callers may omit ``agent``. Stateless frontends (MCP, admin) keep
+    the defaults and stay outside the routing.
+    """
+
+    name: str = ""
+    kinds: tuple[str, ...] = ()
 
     @abstractmethod
     def configure(self, runtime: GatewayRuntime) -> None: ...
 
     @abstractmethod
     async def serve(self) -> None: ...
+
+    def agent_for(self, kind: str) -> str | None:  # noqa: ARG002
+        return None
+
+    async def materialize(self, conv: Conversation) -> ConversationBinding | None:  # noqa: ARG002
+        return None

src/beaver_gateway/frontends/markdown/crossfront.py

@@ -38,14 +38,16 @@ if TYPE_CHECKING:
 _log = logging.getLogger("beaver_gateway.frontends.markdown.crossfront")
 
 
-# User hook: take a turn + vault root, return where the new file should
-# live. Returning a relative ``Path`` is treated as relative to the
-# vault. ``None`` (the default) keeps the built-in
-# ``{vault}/{logged_subdir}/{agent}/{YYYY-MM-DD}_{hex8}.md`` layout.
-LogPathFn = "Callable[[TurnRecord, Path], Path]"
+__all__ = [
+    "ChatPathFn",
+    "CrossFrontendLogger",
+    "fingerprint_messages",
+    "strip_trailing_user_scaffold",
+]
 
-
-__all__ = ["CrossFrontendLogger", "LogPathFn", "fingerprint_messages"]
+ChatPathFn = "Callable[[str, str, Path], Path]"
+"""``(title, agent, vault) -> path`` of a new chat file; relative = under the
+vault. ``None`` keeps ``{vault}/{logged_subdir}/{agent}/{YYYY-MM-DD}_{slug}.md``."""
 
 
 def fingerprint_messages(messages: Iterable[MessageParam]) -> str:
@@ -96,18 +98,18 @@ class CrossFrontendLogger:
         *,
         vault_path: Path,
         logged_subdir: str,
-        log_path: Callable[[TurnRecord, Path], Path] | None = None,
+        chat_path: Callable[[str, str, Path], Path] | None = None,
     ) -> None:
         self._vault = vault_path
         self._root = vault_path / logged_subdir
         self._index: dict[str, Path] = {}
         self._lock = asyncio.Lock()
-        self._log_path_fn = log_path
+        self._chat_path_fn = chat_path
         # When the user supplies a custom path function, files can land
         # anywhere in the vault — so we have to scan the whole vault on
         # startup to rebuild the fingerprint→path map. With the default
         # layout we can bound the scan to ``_logs/``.
-        self._scan_root = vault_path if log_path is not None else self._root
+        self._scan_root = vault_path if chat_path is not None else self._root
 
     def warm_index(self) -> None:
         """Scan logged files synchronously, populating the fingerprint map.
@@ -171,7 +173,7 @@ class CrossFrontendLogger:
             if target.exists():
                 existing = target.read_text(encoding="utf-8")
                 parsed = frontmatter.loads(existing)
-                body = _strip_trailing_user_scaffold(parsed.content)
+                body = strip_trailing_user_scaffold(parsed.content)
                 # We append only the *new* user turn (the last one in
                 # input_messages, since prior turns are already on disk)
                 # plus the assistant reply.
@@ -206,13 +208,15 @@ class CrossFrontendLogger:
     def _new_file_path(self, record: TurnRecord) -> Path:
         """Pick a fresh filename for a brand-new conversation.
 
-        With a user-supplied ``log_path`` we delegate to it (joining a
+        With a user-supplied ``chat_path`` we delegate to it (joining a
         relative result with the vault root). Without one, we fall back
         to ``{logged_subdir}/{agent}/{date}_{hex8}.md`` and ensure the
         ``.md`` suffix in case the user picks a non-md extension by hand.
         """
-        if self._log_path_fn is not None:
-            result = self._log_path_fn(record, self._vault)
+        if self._chat_path_fn is not None:
+            result = self._chat_path_fn(
+                record.first_user_text, record.agent_name, self._vault
+            )
             if not result.is_absolute():
                 result = self._vault / result
             if result.suffix != ".md":
@@ -270,7 +274,7 @@ def _render_full_history(messages: list[MessageParam], assistant: Any) -> str:
     return body
 
 
-def _strip_trailing_user_scaffold(body: str) -> str:
+def strip_trailing_user_scaffold(body: str) -> str:
     """Drop a trailing empty ``### User:`` block if present.
 
     Cross-frontend turns aren't typed into the file by the human — they

src/beaver_gateway/frontends/markdown/files.py

@@ -0,0 +1,50 @@
+"""Vault file IO shared by the markdown frontend and its mirror."""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import tempfile
+from typing import TYPE_CHECKING, Any
+
+import aiofile
+import frontmatter
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+__all__ = ["read_or_empty", "reattach_frontmatter", "write_atomic"]
+
+
+async def read_or_empty(path: Path) -> str:
+    if not path.exists():  # noqa: ASYNC240
+        return ""
+    async with aiofile.async_open(path, "r", encoding="utf-8") as f:
+        return await f.read()
+
+
+async def write_atomic(path: Path, content: str) -> None:
+    """Write via tmp + ``os.replace`` in the same directory: no torn reads."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(
+        prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
+    )
+    try:
+        async with aiofile.async_open(tmp_path, "w", encoding="utf-8") as f:
+            await f.write(content)
+        os.close(fd)
+        os.replace(tmp_path, path)  # noqa: PTH105
+    except BaseException:
+        with contextlib.suppress(OSError):
+            os.close(fd)
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_path)  # noqa: PTH108
+        raise
+
+
+def reattach_frontmatter(metadata: dict[str, Any], body: str) -> str:
+    """Re-emit a ``.md`` with YAML frontmatter; no block at all for empty metadata."""
+    if not metadata:
+        return body if body.endswith("\n") else body + "\n"
+    post = frontmatter.Post(content=body, **metadata)
+    return frontmatter.dumps(post) + "\n"

src/beaver_gateway/frontends/markdown/frontend.py

@@ -19,7 +19,10 @@ posted to the same conversation via ``/api`` waits its turn.
 
 A chat file is a ``deep`` conversation bound as
 ``(markdown, <vault-relative path>)``; frontmatter carries only ``agent``
-and ``conversation_id`` (§3.10), tool calls are never rendered.
+and ``conversation_id`` (§3.10), tool calls are never rendered. The
+frontend is the home of ``deep``: ``materialize`` gives a conversation
+spawned elsewhere its file, and :class:`.mirror.ChatMirror` keeps that
+file in step with replies produced outside ``/chat``.
 
 Cross-frontend logging: when ``log_all_chats=True``, ``configure()``
 registers a handler on ``runtime.turn_log_handlers`` so every other
@@ -30,17 +33,15 @@ shape.
 
 from __future__ import annotations
 
+import asyncio
 import contextlib
 import json
 import logging
-import os
-import tempfile
 import time
 from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 
-import aiofile
 from anthropic.types import RawContentBlockStopEvent
 from fastapi import FastAPI, HTTPException, Request, status
 from fastapi.middleware.cors import CORSMiddleware
@@ -65,12 +66,18 @@ from beaver_gateway.frontends._sse import (
 from beaver_gateway.frontends.base import Frontend
 from beaver_gateway.frontends.markdown import parser, renderer
 from beaver_gateway.frontends.markdown.crossfront import CrossFrontendLogger
+from beaver_gateway.frontends.markdown.files import (
+    read_or_empty,
+    reattach_frontmatter,
+    write_atomic,
+)
+from beaver_gateway.frontends.markdown.mirror import FRONTEND, ChatMirror
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Callable
 
     from beaver_gateway.frontends.base import GatewayRuntime
-    from beaver_gateway.storage.models import Conversation
+    from beaver_gateway.storage.models import Conversation, ConversationBinding
 
 
 _log = logging.getLogger("beaver_gateway.frontends.markdown")
@@ -93,12 +100,13 @@ _STREAM_FLUSH_DEBOUNCE = 0.4
 # disk round-trip).
 _SSE_FLUSH_DEBOUNCE = 0.1
 
-FRONTEND = "markdown"
-
 
 class MarkdownFrontend(Frontend):
     """FastAPI app behind ``POST /chat`` driven by Obsidian-vault files."""
 
+    name = FRONTEND
+    kinds = ("deep",)
+
     def __init__(
         self,
         *,
@@ -108,7 +116,7 @@ class MarkdownFrontend(Frontend):
         default_agent: str | None = None,
         log_all_chats: bool = False,
         logged_subdir: str = "_logs",
-        log_path: Callable[[TurnRecord, Path], Path] | None = None,
+        chat_path: Callable[[str, str, Path], Path] | None = None,
         public_base_url: str | None = None,
     ) -> None:
         self.vault_path = Path(vault_path).expanduser().resolve()
@@ -117,7 +125,7 @@ class MarkdownFrontend(Frontend):
         self.default_agent = default_agent
         self.log_all_chats = log_all_chats
         self.logged_subdir = logged_subdir
-        self.log_path = log_path
+        self.chat_path = chat_path
         # External URL prefix when behind a reverse proxy — same role as
         # on the other bearer frontends. Trailing slash trimmed for
         # idempotent concatenation; ``None`` means "no proxy / advertise
@@ -131,6 +139,7 @@ class MarkdownFrontend(Frontend):
         # request reliably loses the race to 409.
         self._busy: set[Path] = set()
         self._crossfront: CrossFrontendLogger | None = None
+        self._mirror: ChatMirror | None = None
 
     def configure(self, runtime: GatewayRuntime) -> None:
         if runtime.conversations is None:
@@ -138,11 +147,17 @@ class MarkdownFrontend(Frontend):
             raise RuntimeError(msg)
         self._runtime = runtime
         self.vault_path.mkdir(parents=True, exist_ok=True)
+        self._mirror = ChatMirror(
+            vault_path=self.vault_path,
+            runtime=runtime,
+            logged_subdir=self.logged_subdir,
+            chat_path=self.chat_path,
+        )
         if self.log_all_chats:
             self._crossfront = CrossFrontendLogger(
                 vault_path=self.vault_path,
                 logged_subdir=self.logged_subdir,
-                log_path=self.log_path,
+                chat_path=self.chat_path,
             )
             # Scan the existing logged files synchronously here so the
             # fingerprint→path map is populated before the first
@@ -151,6 +166,19 @@ class MarkdownFrontend(Frontend):
             runtime.turn_log_handlers.append(self._crossfront.handle)
         self._app = self._build_app(runtime)
 
+    @property
+    def mirror(self) -> ChatMirror:
+        if self._mirror is None:
+            msg = "configure() must be called first"
+            raise RuntimeError(msg)
+        return self._mirror
+
+    def agent_for(self, kind: str) -> str | None:
+        return self.default_agent if kind == "deep" else None
+
+    async def materialize(self, conv: Conversation) -> ConversationBinding | None:
+        return await self.mirror.materialize(conv)
+
     async def serve(self) -> None:
         import uvicorn
 
@@ -161,7 +189,13 @@ class MarkdownFrontend(Frontend):
             self._app, host=self.host, port=self.port, log_level="info"
         )
         server = uvicorn.Server(config)
-        await server.serve()
+        mirror = asyncio.create_task(self.mirror.run())
+        try:
+            await server.serve()
+        finally:
+            mirror.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await mirror
 
     # ---- app builder ---------------------------------------------------
 
@@ -309,7 +343,7 @@ class MarkdownFrontend(Frontend):
         if isinstance(content_override, str):
             file_text = content_override
         elif content_override is None:
-            file_text = await _read_or_empty(file_path)
+            file_text = await read_or_empty(file_path)
         else:
             raise HTTPException(
                 status.HTTP_400_BAD_REQUEST, "`content` must be a string when present"
@@ -480,7 +514,7 @@ class MarkdownFrontend(Frontend):
         if isinstance(content_override, str):
             file_text = content_override
         elif content_override is None:
-            file_text = await _read_or_empty(file_path)
+            file_text = await read_or_empty(file_path)
         else:
             yield sse_pack(
                 "error",
@@ -569,12 +603,18 @@ class MarkdownFrontend(Frontend):
             msgs=len(parsed.messages),
         )
 
-        conv, conv_external_id, stored_msgs = await self._resolve_conversation(
-            runtime=runtime,
-            metadata=parsed.metadata,
-            agent_name=agent.name,
-            file_path=file_path,
-        )
+        try:
+            conv, conv_external_id, stored_msgs = await self._resolve_conversation(
+                runtime=runtime,
+                metadata=parsed.metadata,
+                agent_name=agent.name,
+                file_path=file_path,
+            )
+        except HTTPException as exc:
+            yield sse_pack(
+                "error", {"status_code": exc.status_code, "detail": exc.detail}
+            )
+            return
         _log.info(
             "chat/stream: file=%s conv_external_id=%s conv_id=%d "
             "stored_msgs=%d incoming_turns=%d",
@@ -619,7 +659,7 @@ class MarkdownFrontend(Frontend):
                 return None
             rendered = renderer.render_assistant_message(partial)
             new_body = renderer.append_to_body(parsed.body, rendered)
-            return _reattach_frontmatter(parsed.metadata, new_body)
+            return reattach_frontmatter(parsed.metadata, new_body)
 
         try:
             async for ev in events_with_heartbeat(events):
@@ -653,8 +693,8 @@ class MarkdownFrontend(Frontend):
                 )
             new_body = renderer.append_to_body(new_body, _render_error_block(exc))
             if write_disk:
-                await _write_atomic(
-                    file_path, _reattach_frontmatter(parsed.metadata, new_body)
+                await write_atomic(
+                    file_path, reattach_frontmatter(parsed.metadata, new_body)
                 )
             yield sse_pack(
                 "error",
@@ -744,8 +784,8 @@ class MarkdownFrontend(Frontend):
                 return
             rendered = renderer.render_assistant_message(partial)
             new_body = renderer.append_to_body(parsed.body, rendered)
-            await _write_atomic(
-                file_path, _reattach_frontmatter(parsed.metadata, new_body)
+            await write_atomic(
+                file_path, reattach_frontmatter(parsed.metadata, new_body)
             )
 
         try:
@@ -769,8 +809,8 @@ class MarkdownFrontend(Frontend):
                 )
             new_body = renderer.append_to_body(new_body, _render_error_block(exc))
             if write_disk:
-                await _write_atomic(
-                    file_path, _reattach_frontmatter(parsed.metadata, new_body)
+                await write_atomic(
+                    file_path, reattach_frontmatter(parsed.metadata, new_body)
                 )
             raise
         return acc.finalize(model=model)
@@ -793,9 +833,9 @@ class MarkdownFrontend(Frontend):
         updated_metadata.pop("fingerprint", None)
         updated_metadata["agent"] = agent_name
         updated_metadata["conversation_id"] = conv_external_id
-        new_content = _reattach_frontmatter(updated_metadata, new_body)
+        new_content = reattach_frontmatter(updated_metadata, new_body)
         if write_disk:
-            await _write_atomic(file_path, new_content)
+            await write_atomic(file_path, new_content)
         return new_content
 
     async def _resolve_conversation(
@@ -820,9 +860,12 @@ class MarkdownFrontend(Frontend):
         if conv is None:
             conv = await conversations.find_bound(frontend=FRONTEND, external_id=rel)
         if conv is None:
-            conv = await conversations.create(
-                kind="deep", agent=agent_name, origin=FRONTEND, title=file_path.stem
-            )
+            try:
+                conv = await conversations.create(
+                    kind="deep", agent=agent_name, origin=FRONTEND, title=file_path.stem
+                )
+            except ValueError as exc:
+                raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
             _log.info("minted conversation %s for %s", conv.external_id, rel)
         bound = [
             b
@@ -894,63 +937,6 @@ class MarkdownFrontend(Frontend):
         return candidate
 
 
-# ---- module-level utilities ----------------------------------------------
-
-
-async def _read_or_empty(path: Path) -> str:
-    """Return file contents, or empty string if the file doesn't exist."""
-    # ``path.exists()`` here is a metadata stat — microseconds — and
-    # gating an async read on whether the file is there is exactly the
-    # check we want. Switching to anyio.Path / aiofiles.os just to
-    # silence the async-pathlib lint would cost a dep edge for no
-    # practical win.
-    if not path.exists():  # noqa: ASYNC240
-        return ""
-    async with aiofile.async_open(path, "r", encoding="utf-8") as f:
-        return await f.read()
-
-
-async def _write_atomic(path: Path, content: str) -> None:
-    """Write ``content`` to ``path`` via tmp + ``os.replace`` (atomic)."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    # ``NamedTemporaryFile`` keeps the file open which complicates
-    # ``os.replace`` on some platforms. Build the tmp name manually,
-    # write+fsync, then rename. Same-directory so the rename is atomic.
-    tmp_name = tempfile.mkstemp(
-        prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent)
-    )
-    fd, tmp_path = tmp_name
-    try:
-        async with aiofile.async_open(tmp_path, "w", encoding="utf-8") as f:
-            await f.write(content)
-        os.close(fd)
-        # ``os.replace`` is the atomic primitive — ``Path.replace`` is a
-        # thin wrapper around the same syscall; either works, ``os.`` is
-        # the one Linux/POSIX docs reach for.
-        os.replace(tmp_path, path)  # noqa: PTH105
-    except BaseException:
-        # Cleanup on failure: close fd, remove tmp.
-        with contextlib.suppress(OSError):
-            os.close(fd)
-        with contextlib.suppress(OSError):
-            os.unlink(tmp_path)  # noqa: PTH108
-        raise
-
-
-def _reattach_frontmatter(metadata: dict[str, Any], body: str) -> str:
-    r"""Re-emit a ``.md`` file with YAML frontmatter at the top.
-
-    Empty metadata → no frontmatter block (avoid littering every file
-    with a hollow ``---\n---``).
-    """
-    if not metadata:
-        return body if body.endswith("\n") else body + "\n"
-    import frontmatter as _fm
-
-    post = _fm.Post(content=body, **metadata)
-    return _fm.dumps(post) + "\n"
-
-
 def _fallback_synthesized(message: Any) -> list[dict[str, Any]]:
     """Build a single-assistant ``synthesized_messages`` list from a raw ``Message``.
 

src/beaver_gateway/frontends/markdown/mirror.py

@@ -0,0 +1,169 @@
+"""Vault files for ``deep`` conversations that were not typed into a file (§3.10).
+
+``materialize`` is the markdown frontend's answer to ``spawn(kind=deep)``:
+a new file in the vault with ``agent`` + ``conversation_id`` frontmatter
+and the ``(markdown, path)`` binding. ``run`` tails the gateway bus and
+appends every ``reply`` of a markdown-bound conversation to its file -
+the seed turn of a spawn, a message posted through ``/api``, a turn
+that came in over ``/v1/messages`` - and stamps the same exchange into
+the canonical history, so a continuation typed in Obsidian aligns
+against the store and resumes the same SDK session instead of reseeding.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+import frontmatter
+
+from beaver_gateway.core.conversation_store import load_messages, rewrite_messages
+from beaver_gateway.core.turn_record import slugify
+from beaver_gateway.frontends.markdown import renderer
+from beaver_gateway.frontends.markdown.crossfront import strip_trailing_user_scaffold
+from beaver_gateway.frontends.markdown.files import (
+    read_or_empty,
+    reattach_frontmatter,
+    write_atomic,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sequence
+    from pathlib import Path
+
+    from beaver_gateway.core.bus import Event
+    from beaver_gateway.frontends.base import GatewayRuntime
+    from beaver_gateway.storage.models import Conversation, ConversationBinding
+
+__all__ = ["FRONTEND", "ChatMirror"]
+
+_log = logging.getLogger("beaver_gateway.frontends.markdown.mirror")
+
+FRONTEND = "markdown"
+
+
+class ChatMirror:
+    def __init__(
+        self,
+        *,
+        vault_path: Path,
+        runtime: GatewayRuntime,
+        logged_subdir: str,
+        chat_path: Callable[[str, str, Path], Path] | None = None,
+    ) -> None:
+        self._vault = vault_path
+        self._runtime = runtime
+        self._root = vault_path / logged_subdir
+        self._chat_path_fn = chat_path
+        self._lock = asyncio.Lock()
+
+    def path_for(self, title: str, agent: str) -> Path:
+        if self._chat_path_fn is not None:
+            path = self._chat_path_fn(title, agent, self._vault)
+            if not path.is_absolute():
+                path = self._vault / path
+        else:
+            day = datetime.now(UTC).strftime("%Y-%m-%d")
+            path = self._root / agent / f"{day}_{slugify(title, maxlen=60)}.md"
+        if path.suffix != ".md":
+            path = path.with_suffix(".md")
+        path = path.resolve()
+        path.relative_to(self._vault)
+        candidate, n = path, 1
+        while candidate.exists():
+            n += 1
+            candidate = path.with_name(f"{path.stem} ({n}){path.suffix}")
+        return candidate
+
+    async def materialize(self, conv: Conversation) -> ConversationBinding:
+        path = self.path_for(conv.title or conv.external_id, conv.agent_name)
+        rel = path.relative_to(self._vault).as_posix()
+        async with self._lock:
+            await write_atomic(path, reattach_frontmatter(_frontmatter(conv), ""))
+        binding = await self._runtime.conversations.bind(
+            conv, frontend=FRONTEND, external_id=rel
+        )
+        _log.info("materialized %s as %s", conv.external_id, rel)
+        return binding
+
+    async def bound_path(self, conv: Conversation) -> Path | None:
+        for b in await self._runtime.conversations.bindings(conv):
+            if b.frontend == FRONTEND and b.visible:
+                return self._vault / b.external_id
+        return None
+
+    async def run(self) -> None:
+        async for event in self._runtime.bus.stream():
+            if event.get("type") != "reply":
+                continue
+            try:
+                await self._on_reply(event)
+            except Exception:  # noqa: BLE001
+                _log.exception("mirror of reply %s failed", event.get("turn_id"))
+
+    async def _on_reply(self, event: Event) -> None:
+        conv = await self._runtime.conversations.get(str(event["conversation_id"]))
+        if conv is None:
+            return
+        await self.append(
+            conv,
+            prompt=str(event.get("prompt") or ""),
+            text=str(event.get("text") or ""),
+            history=event.get("history"),
+        )
+
+    async def append(
+        self,
+        conv: Conversation,
+        *,
+        prompt: str,
+        text: str,
+        history: Sequence[dict[str, Any]] | None = None,
+    ) -> None:
+        path = await self.bound_path(conv)
+        if path is None:
+            return
+        prior = [
+            {"role": str(h["role"]), "content": str(h["text"])}
+            for h in history or ()
+            if h.get("text")
+        ]
+        async with self._lock:
+            parsed = frontmatter.loads(await read_or_empty(path))
+            body = strip_trailing_user_scaffold(parsed.content)
+            if prior and "### " not in body:
+                for m in prior:
+                    body = renderer.append_to_body(body, _render(m))
+            body = renderer.append_to_body(body, renderer.render_user_text(prompt))
+            body = renderer.append_to_body(body, renderer.render_assistant_text(text))
+            body = renderer.append_to_body(body, renderer.USER_SCAFFOLD)
+            metadata = {**parsed.metadata, **_frontmatter(conv)}
+            metadata.pop("fingerprint", None)
+            await write_atomic(path, reattach_frontmatter(metadata, body))
+        await self._persist(conv, prior=prior, prompt=prompt, text=text)
+
+    async def _persist(
+        self, conv: Conversation, *, prior: list[dict[str, Any]], prompt: str, text: str
+    ) -> None:
+        if conv.id is None:
+            return
+        async with self._runtime.db.session() as session:
+            stored = await load_messages(session, conversation_id=conv.id)
+            canonical = [
+                *(stored or prior),
+                {"role": "user", "content": prompt},
+                {"role": "assistant", "content": [{"type": "text", "text": text}]},
+            ]
+            await rewrite_messages(session, conversation_id=conv.id, messages=canonical)
+
+
+def _frontmatter(conv: Conversation) -> dict[str, Any]:
+    return {"agent": conv.agent_name, "conversation_id": conv.external_id}
+
+
+def _render(message: dict[str, Any]) -> str:
+    if message["role"] == "user":
+        return renderer.render_user_text(message["content"])
+    return renderer.render_assistant_text(message["content"])

src/beaver_gateway/frontends/markdown/renderer.py

@@ -53,6 +53,11 @@ def render_user_text(content: str) -> str:
     return f"### User:\n\n{content.strip()}\n"
 
 
+def render_assistant_text(text: str) -> str:
+    r"""Render a text-only assistant turn as ``### Assistant:\n\n<text>``."""
+    return f"### Assistant:\n\n{text.strip()}\n"
+
+
 def render_assistant_message(message: Message) -> str:
     """Render an assistant ``Message`` (with content blocks) into a turn block.