feat(core,backends,frontends,storage): conversations, inject queue, session pool, gateway mcp tools, api frontend

2026-08-28 03:08:30 +02:00
parent ab52fdc2b8
commit e3074c266a
28 changed files with 3543 additions and 345 deletions
@@ -0,0 +1,63 @@
+"""Server-sent events helpers shared by the markdown and API frontends."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+__all__ = ["HEARTBEAT_INTERVAL", "SSE_HEADERS", "events_with_heartbeat", "sse_pack"]
+
+HEARTBEAT_INTERVAL = 15.0
+"""Seconds of backend silence before a comment frame keeps the socket warm."""
+
+SSE_HEADERS = {
+    "Cache-Control": "no-cache, no-transform",
+    "Connection": "keep-alive",
+    "X-Accel-Buffering": "no",
+}
+
+KEEPALIVE = b": keepalive\n\n"
+
+
+async def events_with_heartbeat(
+    events: AsyncIterator[Any], interval: float = HEARTBEAT_INTERVAL
+) -> AsyncIterator[Any]:
+    """Pass ``events`` through, yielding ``None`` after ``interval`` seconds of silence.
+
+    One in-flight ``__anext__`` task is reused across timeouts: a second
+    consumer on the same async generator raises ``RuntimeError``.
+    Cancellation of the outer scope cancels that task instead of leaving
+    it dangling.
+    """
+    src = events.__aiter__()
+    next_task: asyncio.Task[Any] | None = None
+    try:
+        while True:
+            if next_task is None:
+                next_task = asyncio.ensure_future(src.__anext__())
+            done, _pending = await asyncio.wait({next_task}, timeout=interval)
+            if not done:
+                yield None
+                continue
+            task = next_task
+            next_task = None
+            try:
+                result = task.result()
+            except StopAsyncIteration:
+                return
+            yield result
+    finally:
+        if next_task is not None and not next_task.done():
+            next_task.cancel()
+            with contextlib.suppress(BaseException):
+                await next_task
+
+
+def sse_pack(event: str, data: dict[str, Any]) -> bytes:
+    body = json.dumps(data, ensure_ascii=False)
+    return f"event: {event}\ndata: {body}\n\n".encode()
@@ -0,0 +1,5 @@
+"""``ApiFrontend`` - the conversations API and event stream (§3.9)."""
+
+from beaver_gateway.frontends.api.frontend import ApiFrontend
+
+__all__ = ["ApiFrontend"]
@@ -0,0 +1,491 @@
+"""``ApiFrontend`` - ``/api/conversations``, SSE events, sessions, usage (§3.9).
+
+Bearer scope ``api``. Every write goes through ``core/conversations``; the
+frontend only shapes JSON. ``/api/events`` and
+``/api/conversations/{id}/events`` replay the gateway bus as SSE with the
+same keepalive the markdown frontend uses, so a proxy never sees a
+silent socket.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING, Any, cast
+
+from fastapi import FastAPI, HTTPException, Request, status
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
+from sqlalchemy import func
+from sqlalchemy import select as sa_select
+from sqlmodel import col
+
+from beaver_gateway.core import audit
+from beaver_gateway.core.conversations import KINDS, SEEDS
+from beaver_gateway.frontends._auth import require_token
+from beaver_gateway.frontends._sse import (
+    KEEPALIVE,
+    SSE_HEADERS,
+    events_with_heartbeat,
+    sse_pack,
+)
+from beaver_gateway.frontends.base import Frontend
+from beaver_gateway.storage.models import Usage
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from beaver_gateway.core.conversations import Conversations
+    from beaver_gateway.frontends.base import GatewayRuntime
+    from beaver_gateway.storage.models import Conversation
+
+_log = logging.getLogger("beaver_gateway.frontends.api")
+
+__all__ = ["ApiFrontend"]
+
+SCOPE = "api"
+
+
+class ApiFrontend(Frontend):
+    def __init__(
+        self,
+        *,
+        host: str = "0.0.0.0",  # noqa: S104
+        port: int = 8004,
+        public_base_url: 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._app: FastAPI | None = None
+
+    def configure(self, runtime: GatewayRuntime) -> None:
+        if runtime.conversations is None or runtime.bus is None:
+            msg = "ApiFrontend needs runtime.conversations and runtime.bus"
+            raise RuntimeError(msg)
+        self._app = _build_app(runtime)
+
+    async def serve(self) -> None:
+        import uvicorn
+
+        if self._app is None:
+            msg = "configure() must be called before serve()"
+            raise RuntimeError(msg)
+        server = uvicorn.Server(
+            uvicorn.Config(self._app, host=self.host, port=self.port, log_level="info")
+        )
+        await server.serve()
+
+
+def _build_app(runtime: GatewayRuntime) -> FastAPI:  # noqa: PLR0915
+    app = FastAPI(title="beaver-gateway / API")
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=False,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+    conversations = cast("Conversations", runtime.conversations)
+
+    async def body_of(request: Request) -> dict[str, Any]:
+        if not await request.body():
+            return {}
+        try:
+            data = await request.json()
+        except json.JSONDecodeError as exc:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
+            ) from exc
+        if not isinstance(data, dict):
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, "body must be an object")
+        return data
+
+    async def conv_of(public_id: str) -> Conversation:
+        conv = await conversations.get(public_id)
+        if conv is None:
+            raise HTTPException(
+                status.HTTP_404_NOT_FOUND, f"unknown conversation {public_id}"
+            )
+        return conv
+
+    def text_of(data: dict[str, Any], key: str = "text") -> str:
+        text = data.get(key)
+        if not isinstance(text, str) or not text.strip():
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"missing `{key}`")
+        return text
+
+    def int_or_none(data: dict[str, Any], key: str) -> int | None:
+        value = data.get(key)
+        if value is None:
+            return None
+        if not isinstance(value, int) or value < 1:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, f"`{key}` must be a positive int"
+            )
+        return value
+
+    @app.get("/healthz")
+    async def healthz() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get("/api/conversations")
+    async def list_conversations(request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        q = request.query_params
+        rows = await conversations.find(
+            status=q.get("status"), kind=q.get("kind"), limit=int(q.get("limit", "200"))
+        )
+        return {"conversations": [conversations.public(r) for r in rows]}
+
+    @app.post("/api/conversations", status_code=status.HTTP_201_CREATED)
+    async def create_conversation(request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        data = await body_of(request)
+        kind = str(data.get("kind") or "deep")
+        agent = data.get("agent")
+        if kind not in KINDS or kind == "fork":
+            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`"
+            )
+        seed = str(data.get("seed") or "clean")
+        if seed not in SEEDS:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, f"seed must be one of {SEEDS}"
+            )
+        parent = await conv_of(str(data["parent"])) if data.get("parent") else None
+        try:
+            conv = await conversations.spawn(
+                kind=kind,
+                agent=agent,
+                seed=seed,
+                parent=parent,
+                text=data.get("text"),
+                title=data.get("title"),
+                window=int_or_none(data, "window"),
+                origin="api",
+            )
+        except ValueError 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,
+            conversation=conv.external_id,
+            seed=seed,
+        )
+        return await conversations.describe(conv)
+
+    @app.get("/api/conversations/{public_id}")
+    async def get_conversation(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        out = await conversations.describe(conv)
+        out["queue"] = [
+            {
+                "id": i.id,
+                "priority": i.priority,
+                "origin": i.origin,
+                "status": i.status,
+                "created_at": i.created_at.isoformat(),
+                "text": i.text[:200],
+            }
+            for i in await conversations.queue.recent(cast("int", conv.id), limit=20)
+        ]
+        return out
+
+    @app.get("/api/conversations/{public_id}/messages")
+    async def get_messages(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        raw = request.query_params.get("window")
+        window = int(raw) if raw and raw.isdigit() else None
+        return {
+            "id": conv.external_id,
+            "text": await conversations.read(conv, window=window),
+        }
+
+    @app.post(
+        "/api/conversations/{public_id}/messages", status_code=status.HTTP_202_ACCEPTED
+    )
+    async def post_message(public_id: str, request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        data = await body_of(request)
+        item = await conversations.post(
+            conv, text_of(data), origin=str(data.get("origin") or "user")
+        )
+        await audit.log(
+            runtime,
+            actor=f"token:{token}",
+            kind="api_message",
+            agent_name=conv.agent_name,
+            conversation=conv.external_id,
+        )
+        return {"id": conv.external_id, "item": item.id, "status": item.status}
+
+    @app.post(
+        "/api/conversations/{public_id}/inject", status_code=status.HTTP_202_ACCEPTED
+    )
+    async def post_inject(public_id: str, request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        data = await body_of(request)
+        urgency = str(data.get("urgency") or "normal")
+        if urgency not in ("normal", "urgent"):
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, "urgency must be normal|urgent"
+            )
+        item = await conversations.inject(
+            conv,
+            text_of(data),
+            urgency=cast("Any", urgency),
+            origin=str(data.get("origin") or "api"),
+        )
+        await audit.log(
+            runtime,
+            actor=f"token:{token}",
+            kind="api_inject",
+            agent_name=conv.agent_name,
+            conversation=conv.external_id,
+            urgency=urgency,
+        )
+        return {"id": conv.external_id, "item": item.id, "priority": item.priority}
+
+    @app.post("/api/conversations/{public_id}/say")
+    async def post_say(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        return await conversations.say(conv, text_of(await body_of(request)))
+
+    @app.post(
+        "/api/conversations/{public_id}/branch", status_code=status.HTTP_201_CREATED
+    )
+    async def post_branch(public_id: str, request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        parent = await conv_of(public_id)
+        data = await body_of(request)
+        seed = str(data.get("seed_mode") or data.get("seed") or "morning")
+        if seed not in SEEDS:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, f"seed must be one of {SEEDS}"
+            )
+        agent = str(data.get("agent") or parent.agent_name)
+        try:
+            child = await conversations.spawn(
+                kind="branch",
+                agent=agent,
+                seed=seed,
+                parent=parent,
+                text=data.get("text"),
+                title=data.get("title"),
+                window=int_or_none(data, "window"),
+                origin="api",
+            )
+        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_branch",
+            agent_name=agent,
+            conversation=child.external_id,
+            parent=parent.external_id,
+            seed=seed,
+        )
+        return await conversations.describe(child)
+
+    @app.post("/api/conversations/{public_id}/merge")
+    async def post_merge(public_id: str, request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        try:
+            result = await conversations.merge(conv)
+        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_merge",
+            agent_name=conv.agent_name,
+            conversation=conv.external_id,
+        )
+        return {
+            "id": conv.external_id,
+            "status": "merged",
+            "fork": result.conversation.external_id,
+            "text": result.text,
+        }
+
+    @app.post("/api/conversations/{public_id}/fork")
+    async def post_fork(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        data = await body_of(request)
+        try:
+            result = await conversations.fork(
+                conv,
+                text_of(data, "prompt"),
+                window=int_or_none(data, "window"),
+                strip_tools=bool(data.get("strip_tools", False)),
+            )
+        except (ValueError, RuntimeError) as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+        return {
+            "id": conv.external_id,
+            "fork": result.conversation.external_id,
+            "text": result.text,
+        }
+
+    @app.post("/api/conversations/{public_id}/bind")
+    async def post_bind(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        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)),
+        )
+        return await conversations.describe(conv)
+
+    @app.patch("/api/conversations/{public_id}/flags")
+    async def patch_flags(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        data = await body_of(request)
+        return conversations.public(await conversations.set_flags(conv, data))
+
+    @app.patch("/api/conversations/{public_id}")
+    async def patch_conversation(public_id: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        data = await body_of(request)
+        try:
+            if isinstance(data.get("status"), str):
+                conv = await conversations.set_status(conv, data["status"])
+            if isinstance(data.get("title"), str):
+                conv = await conversations.set_title(conv, data["title"])
+        except ValueError as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+        return conversations.public(conv)
+
+    @app.get("/api/conversations/{public_id}/events")
+    async def conversation_events(public_id: str, request: Request) -> Any:
+        await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        return _sse(runtime, conversation_id=conv.external_id)
+
+    @app.get("/api/events")
+    async def all_events(request: Request) -> Any:
+        await require_token(request, runtime, scope=SCOPE)
+        return _sse(runtime, conversation_id=None)
+
+    @app.get("/api/sessions")
+    async def sessions(request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        pool = runtime.pool
+        return {
+            "rss": pool.rss() if pool is not None else None,
+            "rss_limit": pool.rss_limit if pool is not None else None,
+            "sessions": pool.snapshot() if pool is not None else [],
+        }
+
+    @app.get("/api/schedules")
+    async def schedules(request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        raw = request.query_params.get("conversation")
+        conv = await conv_of(raw) if raw else None
+        return {
+            "schedules": [
+                {
+                    "id": s.id,
+                    "conversation_row": s.conversation_id,
+                    "execute_at": s.execute_at.isoformat(),
+                    "text": s.text,
+                    "delivered_at": s.delivered_at.isoformat()
+                    if s.delivered_at
+                    else None,
+                }
+                for s in await conversations.schedules(conv)
+            ]
+        }
+
+    @app.get("/api/usage")
+    async def usage(request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        hours = float(request.query_params.get("hours", "24"))
+        since = datetime.now(UTC) - timedelta(hours=hours)
+        columns = (
+            func.count(),
+            func.coalesce(func.sum(Usage.input_tokens), 0),
+            func.coalesce(func.sum(Usage.output_tokens), 0),
+            func.coalesce(func.sum(Usage.cache_read_tokens), 0),
+            func.coalesce(func.sum(Usage.cache_creation_tokens), 0),
+            func.coalesce(func.sum(Usage.cost_usd), 0.0),
+        )
+        async with runtime.db.session() as session:
+            by_agent = (
+                await session.execute(  # ty: ignore[deprecated]
+                    sa_select(col(Usage.agent_name), *columns)
+                    .where(col(Usage.ts) >= since.replace(tzinfo=None))
+                    .group_by(col(Usage.agent_name))
+                )
+            ).all()
+            by_conversation = (
+                await session.execute(  # ty: ignore[deprecated]
+                    sa_select(col(Usage.conversation_id), *columns)
+                    .where(col(Usage.ts) >= since.replace(tzinfo=None))
+                    .group_by(col(Usage.conversation_id))
+                )
+            ).all()
+        return {
+            "since": since.isoformat(timespec="seconds"),
+            "by_agent": [_usage_row("agent", r) for r in by_agent],
+            "by_conversation": [_usage_row("conversation", r) for r in by_conversation],
+        }
+
+    @app.exception_handler(HTTPException)
+    async def http_error(_request: Request, exc: HTTPException) -> JSONResponse:
+        return JSONResponse(
+            status_code=exc.status_code,
+            content={"error": exc.detail},
+            headers=exc.headers,
+        )
+
+    return app
+
+
+def _usage_row(label: str, row: Any) -> dict[str, Any]:
+    key, turns, inp, out, cache_read, cache_creation, cost = row
+    return {
+        label: key,
+        "turns": turns,
+        "input": inp,
+        "output": out,
+        "cache_read": cache_read,
+        "cache_creation": cache_creation,
+        "cost_usd": round(float(cost or 0.0), 4),
+    }
+
+
+def _sse(runtime: GatewayRuntime, *, conversation_id: str | None) -> StreamingResponse:
+    async def gen() -> AsyncIterator[bytes]:
+        stream = runtime.bus.stream(conversation_id=conversation_id)
+        yield sse_pack("hello", {"type": "hello", "conversation_id": conversation_id})
+        async for event in events_with_heartbeat(stream):
+            if event is None:
+                yield KEEPALIVE
+                continue
+            yield sse_pack(str(event["type"]), event)
+
+    return StreamingResponse(gen(), media_type="text/event-stream", headers=SSE_HEADERS)
@@ -12,7 +12,7 @@ from __future__ import annotations

 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable, Mapping, Sequence
@@ -77,6 +77,12 @@ class GatewayRuntime:
    # to keep ``anthropic.types`` out of the runtime import graph for
    # this base module.
    turn_log_handlers: list[TurnLogHandler] = field(default_factory=list)
+    # M1b: conversations service, event bus and the shared session pool.
+    # ``Any`` for the same import-graph reason as above; ``None`` only in
+    # tests that build a runtime without them.
+    conversations: Any = None
+    bus: Any = None
+    pool: Any = None


 class Frontend(ABC):
@@ -13,7 +13,13 @@ Concurrency model: an in-memory ``set[Path]`` of files currently in
 flight. Two concurrent requests for the same file → the second gets
 409. The set is single-process (one gateway instance) — that's by
 design; the markdown frontend is the only writer in its vault from
-the gateway side.
+the gateway side. The turn itself runs through ``core/conversations``
+(one turn per conversation, ``running_turn`` in the DB), so a message
+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.

 Cross-frontend logging: when ``log_all_chats=True``, ``configure()``
 registers a handler on ``runtime.turn_log_handlers`` so every other
@@ -24,7 +30,6 @@ shape.

 from __future__ import annotations

-import asyncio
 import contextlib
 import json
 import logging
@@ -33,7 +38,7 @@ import tempfile
 import time
 from collections.abc import AsyncIterator
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast

 import aiofile
 from anthropic.types import RawContentBlockStopEvent
@@ -44,29 +49,28 @@ from fastapi.responses import JSONResponse, StreamingResponse
 from beaver_gateway.core import audit
 from beaver_gateway.core.conversation_store import (
    diff_and_fork,
-    load_conversation,
    load_messages,
-    mint_conversation,
    rewrite_messages,
-    set_session_id,
 )
 from beaver_gateway.core.turn_capture import TurnCapture
 from beaver_gateway.core.turn_record import TurnRecord
 from beaver_gateway.frontends._accumulate import StreamAccumulator
 from beaver_gateway.frontends._auth import require_token
+from beaver_gateway.frontends._sse import (
+    KEEPALIVE,
+    SSE_HEADERS,
+    events_with_heartbeat,
+    sse_pack,
+)
 from beaver_gateway.frontends.base import Frontend
 from beaver_gateway.frontends.markdown import parser, renderer
-from beaver_gateway.frontends.markdown.crossfront import (
-    CrossFrontendLogger,
-    fingerprint_messages,
-)
+from beaver_gateway.frontends.markdown.crossfront import CrossFrontendLogger

 if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Callable

-    from anthropic.types import MessageParam
-
    from beaver_gateway.frontends.base import GatewayRuntime
+    from beaver_gateway.storage.models import Conversation


 _log = logging.getLogger("beaver_gateway.frontends.markdown")
@@ -89,12 +93,7 @@ _STREAM_FLUSH_DEBOUNCE = 0.4
 # disk round-trip).
 _SSE_FLUSH_DEBOUNCE = 0.1

-# Interval between SSE comment-frames sent when the backend is silent
-# (e.g. claude is mid-thinking on a large context). The Obsidian plugin
-# and any intermediate proxies will hold the connection open as long as
-# bytes keep flowing; a comment-frame is the cheapest legal SSE keepalive.
-# Set well under typical proxy/client idle timeouts (60s).
-_SSE_HEARTBEAT_INTERVAL = 15.0
+FRONTEND = "markdown"


 class MarkdownFrontend(Frontend):
@@ -134,6 +133,9 @@ class MarkdownFrontend(Frontend):
        self._crossfront: CrossFrontendLogger | None = None

    def configure(self, runtime: GatewayRuntime) -> None:
+        if runtime.conversations is None:
+            msg = "MarkdownFrontend needs runtime.conversations"
+            raise RuntimeError(msg)
        self._runtime = runtime
        self.vault_path.mkdir(parents=True, exist_ok=True)
        if self.log_all_chats:
@@ -286,17 +288,7 @@ class MarkdownFrontend(Frontend):
                    self._busy.discard(file_path)

            return StreamingResponse(
-                gen(),
-                media_type="text/event-stream",
-                headers={
-                    "Cache-Control": "no-cache, no-transform",
-                    "Connection": "keep-alive",
-                    # nginx default-buffers SSE bodies; this header tells
-                    # both nginx and uvicorn-behind-proxy to flush as we
-                    # write. Harmless if the deployment has no reverse
-                    # proxy in front.
-                    "X-Accel-Buffering": "no",
-                },
+                gen(), media_type="text/event-stream", headers=SSE_HEADERS
            )

        return app
@@ -384,16 +376,19 @@ class MarkdownFrontend(Frontend):
        # stored history, and feed the aligned messages to the backend
        # - see ``core/conversation_store.py`` for the full rationale.
        conv, conv_external_id, stored_msgs = await self._resolve_conversation(
-            runtime=runtime, metadata=parsed.metadata, agent_name=agent.name
+            runtime=runtime,
+            metadata=parsed.metadata,
+            agent_name=agent.name,
+            file_path=file_path,
        )
        outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns)
        capture = TurnCapture()
-        events = backend.complete(
-            agent=agent,
+        events = runtime.conversations.turn(
+            conv,
            messages=outcome.messages,
-            system=None,
+            origin="user",
            capture=capture,
-            **_session_options(conv, outcome.divergence_index),
+            use_session=outcome.divergence_index is None,
        )
        try:
            message = await self._stream_to_file(
@@ -422,7 +417,7 @@ class MarkdownFrontend(Frontend):

        await self._persist_canonical_history(
            runtime=runtime,
-            conversation_id=conv.id,
+            conversation_id=cast("int", conv.id),
            persist_messages=outcome.persist_messages,
            new_user_text=parsed.turns[-1].text,
            capture=capture,
@@ -487,7 +482,7 @@ class MarkdownFrontend(Frontend):
        elif content_override is None:
            file_text = await _read_or_empty(file_path)
        else:
-            yield _sse_pack(
+            yield sse_pack(
                "error",
                {
                    "status_code": status.HTTP_400_BAD_REQUEST,
@@ -503,7 +498,7 @@ class MarkdownFrontend(Frontend):
            default=self.default_agent,
        )
        if not agent_name:
-            yield _sse_pack(
+            yield sse_pack(
                "error",
                {
                    "status_code": status.HTTP_400_BAD_REQUEST,
@@ -516,7 +511,7 @@ class MarkdownFrontend(Frontend):
            return

        if not parsed.messages:
-            yield _sse_pack(
+            yield sse_pack(
                "done",
                {
                    "status": "nothing_to_do",
@@ -527,7 +522,7 @@ class MarkdownFrontend(Frontend):
            return

        if parser.last_role(parsed.messages) == "assistant":
-            yield _sse_pack(
+            yield sse_pack(
                "done",
                {
                    "status": "nothing_to_do",
@@ -539,7 +534,7 @@ class MarkdownFrontend(Frontend):

        agent = runtime.agents.get(agent_name)
        if agent is None:
-            yield _sse_pack(
+            yield sse_pack(
                "error",
                {
                    "status_code": status.HTTP_404_NOT_FOUND,
@@ -549,7 +544,7 @@ class MarkdownFrontend(Frontend):
            return
        backend = runtime.backends.get(agent.name)
        if backend is None:
-            yield _sse_pack(
+            yield sse_pack(
                "error",
                {
                    "status_code": status.HTTP_503_SERVICE_UNAVAILABLE,
@@ -575,7 +570,10 @@ class MarkdownFrontend(Frontend):
        )

        conv, conv_external_id, stored_msgs = await self._resolve_conversation(
-            runtime=runtime, metadata=parsed.metadata, agent_name=agent.name
+            runtime=runtime,
+            metadata=parsed.metadata,
+            agent_name=agent.name,
+            file_path=file_path,
        )
        _log.info(
            "chat/stream: file=%s conv_external_id=%s conv_id=%d "
@@ -602,12 +600,12 @@ class MarkdownFrontend(Frontend):
            agent.name,
            conv.session_id,
        )
-        events = backend.complete(
-            agent=agent,
+        events = runtime.conversations.turn(
+            conv,
            messages=outcome.messages,
-            system=None,
+            origin="user",
            capture=capture,
-            **_session_options(conv, outcome.divergence_index),
+            use_session=outcome.divergence_index is None,
        )

        acc = StreamAccumulator()
@@ -624,13 +622,9 @@ class MarkdownFrontend(Frontend):
            return _reattach_frontmatter(parsed.metadata, new_body)

        try:
-            async for ev in _events_with_heartbeat(events):
+            async for ev in events_with_heartbeat(events):
                if ev is None:
-                    # Backend is quiet (claude mid-thinking, MCP slow,
-                    # whatever). SSE comment-frame keeps the TCP socket
-                    # warm so the plugin / uvicorn / any reverse proxy
-                    # doesn't time the request out before we finish.
-                    yield b": keepalive\n\n"
+                    yield KEEPALIVE
                    continue
                acc.feed(ev)
                now = time.monotonic()
@@ -643,7 +637,7 @@ class MarkdownFrontend(Frontend):
                    # render to the same prefix as before they closed
                    # (we don't surface the tool-call args in markdown).
                    if payload is not None and payload != last_payload:
-                        yield _sse_pack("delta", {"new_content": payload})
+                        yield sse_pack("delta", {"new_content": payload})
                        last_payload = payload
                    last_flush = now
        except Exception as exc:  # noqa: BLE001 — wire any backend failure as an SSE error frame
@@ -662,7 +656,7 @@ class MarkdownFrontend(Frontend):
                await _write_atomic(
                    file_path, _reattach_frontmatter(parsed.metadata, new_body)
                )
-            yield _sse_pack(
+            yield sse_pack(
                "error",
                {
                    "status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
@@ -684,7 +678,7 @@ class MarkdownFrontend(Frontend):

        await self._persist_canonical_history(
            runtime=runtime,
-            conversation_id=conv.id,
+            conversation_id=cast("int", conv.id),
            persist_messages=outcome.persist_messages,
            new_user_text=parsed.turns[-1].text,
            capture=capture,
@@ -704,7 +698,7 @@ class MarkdownFrontend(Frontend):
            except Exception:  # noqa: BLE001
                _log.exception("turn_log_handler raised; continuing")

-        yield _sse_pack(
+        yield sse_pack(
            "done",
            {
                "status": "ok",
@@ -795,71 +789,53 @@ class MarkdownFrontend(Frontend):
        rendered = renderer.render_assistant_message(message)
        new_body = renderer.append_to_body(parsed.body, rendered)
        new_body = renderer.append_to_body(new_body, renderer.USER_SCAFFOLD)
-        # Recompute fingerprint so a future cross-frontend hit on this
-        # same conversation can find it. Stored as hex string in
-        # frontmatter — only the markdown frontend reads it.
-        assistant_param: MessageParam = {
-            "role": "assistant",
-            "content": _flatten_assistant_text(message),
-        }
-        updated_messages: list[MessageParam] = [*parsed.messages, assistant_param]
        updated_metadata = dict(parsed.metadata)
+        updated_metadata.pop("fingerprint", None)
        updated_metadata["agent"] = agent_name
        updated_metadata["conversation_id"] = conv_external_id
-        updated_metadata["fingerprint"] = fingerprint_messages(updated_messages)
        new_content = _reattach_frontmatter(updated_metadata, new_body)
        if write_disk:
            await _write_atomic(file_path, new_content)
        return new_content

    async def _resolve_conversation(
-        self, *, runtime: GatewayRuntime, metadata: dict[str, Any], agent_name: str
-    ) -> tuple[Any, str, list[dict[str, Any]]]:
-        """Resolve the conversation row + stored messages for this request.
+        self,
+        *,
+        runtime: GatewayRuntime,
+        metadata: dict[str, Any],
+        agent_name: str,
+        file_path: Path,
+    ) -> tuple[Conversation, str, list[dict[str, Any]]]:
+        """Resolve the ``deep`` conversation for this file + its stored messages.

-        Looks up by frontmatter ``conversation_id``, mints a new row if
-        missing, and returns ``(conv, external_id, stored_messages)``.
-        ``conv.id`` is guaranteed non-None because both
-        ``load_conversation`` (after refresh on a committed row) and
-        ``mint_conversation`` (post-commit refresh) populate it. We
-        coerce with a runtime check so the rest of the handler can
-        treat it as ``int``.
+        Frontmatter ``conversation_id`` wins; a file that lost it is found
+        by its visible ``(markdown, path)`` binding; otherwise a new
+        conversation is created. The binding follows the file: a moved
+        chat re-binds to its new path on the next turn.
        """
+        conversations = runtime.conversations
+        rel = file_path.relative_to(self.vault_path).as_posix()
        raw = metadata.get("conversation_id")
-        lookup_id = raw if isinstance(raw, str) and raw else None
+        conv = await conversations.get(raw) if isinstance(raw, str) and raw else None
+        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
+            )
+            _log.info("minted conversation %s for %s", conv.external_id, rel)
+        bound = [
+            b
+            for b in await conversations.bindings(conv)
+            if b.frontend == FRONTEND and b.visible and b.external_id == rel
+        ]
+        if not bound:
+            await conversations.bind(conv, frontend=FRONTEND, external_id=rel)
+        await conversations.touch_user(conv)
+        if conv.id is None:
+            msg = "conversation row missing primary key after commit"
+            raise RuntimeError(msg)
        async with runtime.db.session() as session:
-            conv = None
-            if lookup_id is not None:
-                conv = await load_conversation(
-                    session, frontend="markdown", external_id=lookup_id
-                )
-                if conv is None:
-                    _log.info(
-                        "_resolve_conversation: frontmatter conv_id=%s "
-                        "not found in DB, will mint new",
-                        lookup_id,
-                    )
-                else:
-                    _log.info(
-                        "_resolve_conversation: LOADED existing conv "
-                        "id=%d external_id=%s",
-                        conv.id or -1,
-                        conv.external_id,
-                    )
-            if conv is None:
-                conv = await mint_conversation(
-                    session, frontend="markdown", agent_name=agent_name
-                )
-                _log.info(
-                    "_resolve_conversation: MINTED new conv "
-                    "id=%d external_id=%s agent=%s",
-                    conv.id or -1,
-                    conv.external_id,
-                    agent_name,
-                )
-            if conv.id is None:
-                msg = "conversation row missing primary key after commit"
-                raise RuntimeError(msg)
            stored = await load_messages(session, conversation_id=conv.id)
        return conv, conv.external_id, stored

@@ -894,12 +870,6 @@ class MarkdownFrontend(Frontend):
            await rewrite_messages(
                session, conversation_id=conversation_id, messages=canonical
            )
-            if capture.session_id is not None:
-                await set_session_id(
-                    session,
-                    conversation_id=conversation_id,
-                    session_id=capture.session_id,
-                )
        _log.info(
            "_persist_canonical_history: conv_id=%d DB committed", conversation_id
        )
@@ -927,71 +897,6 @@ class MarkdownFrontend(Frontend):
 # ---- module-level utilities ----------------------------------------------


-def _session_options(conv: Any, divergence_index: int | None) -> dict[str, Any]:
-    """Backend options that pin the turn to the conversation's live session.
-
-    A divergence means the file's history no longer matches what the
-    session saw, so the stored ``session_id`` is withheld and the backend
-    seeds a fresh one from the aligned messages.
-    """
-    return {
-        "conversation_id": conv.external_id,
-        "session_id": conv.session_id if divergence_index is None else None,
-    }
-
-
-async def _events_with_heartbeat(
-    events: AsyncIterator[Any], interval: float = _SSE_HEARTBEAT_INTERVAL
-) -> AsyncIterator[Any]:
-    """Wrap an async event stream with idle-time heartbeat markers.
-
-    Yields ``None`` every ``interval`` seconds during silence; real
-    events pass through unchanged. When the wrapped iterator is
-    exhausted, this generator returns. Cancellation propagates: if the
-    outer scope is cancelled we cancel the pending ``__anext__`` task
-    instead of leaving it dangling.
-    """
-    src = events.__aiter__()
-    next_task: asyncio.Task[Any] | None = None
-    try:
-        while True:
-            # Reuse the in-flight task across timeouts. Spawning a fresh
-            # ``__anext__()`` while the previous one is still pending
-            # puts two consumers on the same async generator — that
-            # raises ``RuntimeError: anext(): asynchronous generator is
-            # already running``.
-            if next_task is None:
-                next_task = asyncio.ensure_future(src.__anext__())
-            done, _pending = await asyncio.wait({next_task}, timeout=interval)
-            if not done:
-                yield None
-                continue
-            task = next_task
-            next_task = None
-            try:
-                result = task.result()
-            except StopAsyncIteration:
-                return
-            yield result
-    finally:
-        if next_task is not None and not next_task.done():
-            next_task.cancel()
-            with contextlib.suppress(BaseException):
-                await next_task
-
-
-def _sse_pack(event: str, data: dict[str, Any]) -> bytes:
-    r"""Format one Server-Sent Event frame.
-
-    Uses named events (``event: <name>``) so the plugin can dispatch on
-    type without parsing JSON discriminators. ``ensure_ascii=False`` so
-    multibyte content rides through verbatim instead of becoming
-    ``\uXXXX`` blobs that bloat the wire.
-    """
-    body = json.dumps(data, ensure_ascii=False)
-    return f"event: {event}\ndata: {body}\n\n".encode()
-
-
 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
@@ -1083,21 +988,6 @@ def _fallback_synthesized(message: Any) -> list[dict[str, Any]]:
    return [{"role": "assistant", "content": content}]


-def _flatten_assistant_text(message: Any) -> str:
-    """Pull all text blocks from an assistant ``Message`` and join them.
-
-    Used when we need the assistant content as a plain string for
-    fingerprinting / equality with a parser-shaped history (parser
-    already drops thinking + tool_use from assistant turns).
-    """
-    chunks = [
-        getattr(block, "text", "") or ""
-        for block in getattr(message, "content", ())
-        if getattr(block, "type", None) == "text"
-    ]
-    return "\n\n".join(c for c in chunks if c)
-
-
 def _render_error_block(exc: BaseException) -> str:
    """Render a backend failure as an Assistant turn with a ``[!error]-`` callout."""
    msg = str(exc) or exc.__class__.__name__
@@ -9,11 +9,10 @@ from other frontends).

 from __future__ import annotations

-import json
 import re
 from typing import TYPE_CHECKING, Any, cast

-from anthropic.types import Message, TextBlock, ThinkingBlock, ToolUseBlock
+from anthropic.types import Message, TextBlock, ThinkingBlock

 if TYPE_CHECKING:
    from collections.abc import Iterable
@@ -61,9 +60,8 @@ def render_assistant_message(message: Message) -> str:

    * ``ThinkingBlock`` → ``> [!thinking]-`` collapsed callout
    * ``TextBlock`` → plain text (the spoken answer)
-    * ``ToolUseBlock`` → ``> [!tool]- <name>`` callout with the ``input``
-      JSON quoted inside. Tool *results* are not persisted — see
-      module docstring on ``parser.py`` for why.
+    * ``ToolUseBlock`` → nothing (§3.10: tool calls never reach the file;
+      "what the agent is doing" is the activity panel fed by SSE)

    Blank lines separate adjacent blocks; trailing newline guarantees
    the next ``---`` / ``### User:`` marker lands on its own line.
@@ -147,10 +145,7 @@ def _render_block(block: object) -> Iterable[str]:
    if isinstance(block, ThinkingBlock):
        yield from _render_thinking(block.thinking or "")
        return
-    if isinstance(block, ToolUseBlock):
-        yield from _render_tool_use(block)
-        return
-    # Unknown block type — skip silently rather than corrupting the file.
+    # Tool-use blocks and unknown block types never reach the file.


 def _render_thinking(text: str) -> Iterable[str]:
@@ -159,17 +154,6 @@ def _render_thinking(text: str) -> Iterable[str]:
        yield f"> {line}" if line else ">"


-def _render_tool_use(block: ToolUseBlock) -> Iterable[str]:
-    title = summarize_tool_input(block.name, block.input)
-    yield f"> [!tool]- {title}"
-    yield "> **input:**"
-    yield "> ```json"
-    pretty = json.dumps(block.input, indent=2, ensure_ascii=False, sort_keys=True)
-    for line in pretty.splitlines():
-        yield f"> {line}" if line else ">"
-    yield "> ```"
-
-
 def adaptive_fence(content: str) -> str:
    """Return a backtick fence at least one longer than the longest run in ``content``.