From 1d8d65b69a49d3ad1e4a681f26f33ff7b6b480ef Mon Sep 17 00:00:00 2001 From: h Date: Fri, 28 Aug 2026 03:50:33 +0200 Subject: [PATCH] feat(core,frontends,agents): agent kinds, frontend routing, anthropic on conversations, vault mirror --- examples/config.py | 37 ++- src/beaver_gateway/agents/claude.py | 18 +- src/beaver_gateway/backends/claude_sdk.py | 42 +--- src/beaver_gateway/cli.py | 1 + src/beaver_gateway/core/conversations.py | 49 +++- src/beaver_gateway/core/gateway_tools.py | 28 ++- src/beaver_gateway/core/transcript.py | 40 ++- src/beaver_gateway/frontends/anthropic.py | 236 +++++++++++------- src/beaver_gateway/frontends/api/frontend.py | 67 +++-- src/beaver_gateway/frontends/base.py | 23 +- .../frontends/markdown/crossfront.py | 34 +-- .../frontends/markdown/files.py | 50 ++++ .../frontends/markdown/frontend.py | 162 ++++++------ .../frontends/markdown/mirror.py | 169 +++++++++++++ .../frontends/markdown/renderer.py | 5 + tests/test_conversations.py | 112 ++++++++- tests/test_routing.py | 228 +++++++++++++++++ 17 files changed, 1009 insertions(+), 292 deletions(-) create mode 100644 src/beaver_gateway/frontends/markdown/files.py create mode 100644 src/beaver_gateway/frontends/markdown/mirror.py create mode 100644 tests/test_routing.py diff --git a/examples/config.py b/examples/config.py index 0eb0717..3664bb5 100644 --- a/examples/config.py +++ b/examples/config.py @@ -12,7 +12,7 @@ from beaver_gateway.agents.base import ExposedMcp from beaver_gateway.agents.claude import ClaudeAgent from beaver_gateway.agents.raycast import RaycastAgent, RemoteTool, UserPreferences from beaver_gateway.core.registry import Gateway -from beaver_gateway.core.turn_record import TurnRecord, slugify +from beaver_gateway.core.turn_record import slugify from beaver_gateway.frontends.admin import AdminFrontend from beaver_gateway.frontends.anthropic import AnthropicMessagesFrontend from beaver_gateway.frontends.markdown import MarkdownFrontend @@ -20,20 +20,19 @@ from beaver_gateway.frontends.mcp_server import McpServerFrontend from beaver_gateway.mcp.types import McpServer -def chat_log_path(record: TurnRecord, vault: Path) -> Path: - """Decide where a logged chat from another frontend lands in the vault. +def chat_path(title: str, agent: str, vault: Path) -> Path: # noqa: ARG001 + """Where a new chat file lands in the vault. - Called by ``MarkdownFrontend`` (with ``log_all_chats=True``) the first - time a conversation needs a file — continuation turns are matched by - fingerprint and stick to the file picked here. Return value can be + Called by ``MarkdownFrontend`` for every conversation that needs a + file: a ``deep`` chat spawned by the dispatcher, a ``/v1/messages`` + chat (``title`` = first user message), or with ``log_all_chats=True`` + the archive of a stateless agent's turns. Return value can be absolute or relative; relative paths are anchored under ``vault``. - Layout below: ``//_.md`` where - ``topic`` is a slug of the very first user message in the chat. + Layout below: ``//_.md``. """ today = date.today() - topic = slugify(record.first_user_text, maxlen=40) - return vault / f"{today:%Y-%m}" / f"{today:%Y-%m-%d}_{topic}.md" + return vault / f"{today:%Y-%m}" / f"{today:%Y-%m-%d}_{slugify(title, maxlen=40)}.md" def current_time() -> str: @@ -208,17 +207,13 @@ gateway = Gateway( vault_path=Path(tempfile.mkdtemp(prefix="beaver-vault-")).resolve(), default_agent="research", log_all_chats=True, - # ``log_path`` (optional) overrides the default - # ``{vault}/_logs//_.md`` layout for chats - # logged from OTHER frontends (Anthropic Messages, admin - # in-browser chat). Defined as a top-of-file function so the - # types are explicit and the IDE can hover them; a lambda - # works too, but a real ``def`` keeps the signature visible - # and lets you docstring it. Heads up: any custom path - # forces ``warm_index`` to scan the entire vault on startup - # so the fingerprint→file map survives a restart no matter - # where you put files. - log_path=chat_log_path, + # ``chat_path`` (optional) overrides the default + # ``{vault}/_logs//_.md`` layout. Heads up: + # any custom path forces ``warm_index`` to scan the entire + # vault on startup so the fingerprint→file map of archived + # stateless chats survives a restart no matter where you put + # files. + chat_path=chat_path, ), ], ) diff --git a/src/beaver_gateway/agents/claude.py b/src/beaver_gateway/agents/claude.py index 9604a46..5302c82 100644 --- a/src/beaver_gateway/agents/claude.py +++ b/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 diff --git a/src/beaver_gateway/backends/claude_sdk.py b/src/beaver_gateway/backends/claude_sdk.py index 97227c1..a481d4f 100644 --- a/src/beaver_gateway/backends/claude_sdk.py +++ b/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) diff --git a/src/beaver_gateway/cli.py b/src/beaver_gateway/cli.py index e89ef24..4ac95c9 100644 --- a/src/beaver_gateway/cli.py +++ b/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 diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index 8e046a0..89ed2f5 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/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, ) diff --git a/src/beaver_gateway/core/gateway_tools.py b/src/beaver_gateway/core/gateway_tools.py index 6e45c37..f83286d 100644 --- a/src/beaver_gateway/core/gateway_tools.py +++ b/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( diff --git a/src/beaver_gateway/core/transcript.py b/src/beaver_gateway/core/transcript.py index 1529417..609842f 100644 --- a/src/beaver_gateway/core/transcript.py +++ b/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 "" diff --git a/src/beaver_gateway/frontends/anthropic.py b/src/beaver_gateway/frontends/anthropic.py index ff9bd7d..ace6739 100644 --- a/src/beaver_gateway/frontends/anthropic.py +++ b/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: \ndata: \n\n`` — the shape + Each event becomes ``event: \ndata: \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: diff --git a/src/beaver_gateway/frontends/api/frontend.py b/src/beaver_gateway/frontends/api/frontend.py index e21c9a5..f1fe446 100644 --- a/src/beaver_gateway/frontends/api/frontend.py +++ b/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") diff --git a/src/beaver_gateway/frontends/base.py b/src/beaver_gateway/frontends/base.py index c1686fd..04fb39c 100644 --- a/src/beaver_gateway/frontends/base.py +++ b/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 diff --git a/src/beaver_gateway/frontends/markdown/crossfront.py b/src/beaver_gateway/frontends/markdown/crossfront.py index 7013124..a5887e5 100644 --- a/src/beaver_gateway/frontends/markdown/crossfront.py +++ b/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 diff --git a/src/beaver_gateway/frontends/markdown/files.py b/src/beaver_gateway/frontends/markdown/files.py new file mode 100644 index 0000000..4f9ee49 --- /dev/null +++ b/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" diff --git a/src/beaver_gateway/frontends/markdown/frontend.py b/src/beaver_gateway/frontends/markdown/frontend.py index 501acb5..db05558 100644 --- a/src/beaver_gateway/frontends/markdown/frontend.py +++ b/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, )``; 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``. diff --git a/src/beaver_gateway/frontends/markdown/mirror.py b/src/beaver_gateway/frontends/markdown/mirror.py new file mode 100644 index 0000000..3ffcdfb --- /dev/null +++ b/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"]) diff --git a/src/beaver_gateway/frontends/markdown/renderer.py b/src/beaver_gateway/frontends/markdown/renderer.py index 983eb0c..eae1d30 100644 --- a/src/beaver_gateway/frontends/markdown/renderer.py +++ b/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``.""" + 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. diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 4cf86d5..a3b8d6f 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -29,8 +29,47 @@ from beaver_gateway.core.transcript import ( strip_tool_entries, window_entries, ) +from beaver_gateway.frontends.base import Frontend from beaver_gateway.storage import Database -from beaver_gateway.storage.models import InjectQueueItem +from beaver_gateway.storage.models import ( + Conversation, + ConversationBinding, + InjectQueueItem, +) + + +class StubFrontend(Frontend): + def __init__( + self, + name: str, + kinds: tuple[str, ...], + agents: dict[str, str] | None = None, + *, + home: bool = False, + ) -> None: + self.name = name + self.kinds = kinds + self.agents = agents or {} + self.home = home + self.materialized: list[str] = [] + self.conversations = None + + def configure(self, runtime) -> None: + pass + + async def serve(self) -> None: + pass + + def agent_for(self, kind: str) -> str | None: + return self.agents.get(kind) + + async def materialize(self, conv: Conversation) -> ConversationBinding | None: + if not self.home: + return None + self.materialized.append(conv.external_id) + return await self.conversations.bind( + conv, frontend=self.name, external_id=f"{self.name}:{conv.external_id}" + ) class ScriptedClient: @@ -116,29 +155,41 @@ class World: model="m", system_prompt="hi", cwd=root, + kinds=("master", "branch", "job", "fork"), gateway_tools=("say",), options=ClaudeOptions(effort="low"), ) + self.deep_agent = ClaudeAgent( + name="d", model="m", system_prompt="deep", cwd=root + ) + self.markdown = StubFrontend("markdown", ("deep",), {"deep": "d"}, home=True) + self.api = StubFrontend("api", ("master", "branch", "deep", "job")) self.pool = SessionPool(rss_limit=1 << 40, max_live=100) - self.backend = ClaudeSdkBackend( - agent=self.agent, - mcp_internal_urls={}, - session_store=self.store, - client_factory=ScriptedClient, - work_dir=root / "work", - pool=self.pool, + self.backend, self.deep_backend = ( + ClaudeSdkBackend( + agent=agent, + mcp_internal_urls={}, + session_store=self.store, + client_factory=ScriptedClient, + work_dir=root / "work", + pool=self.pool, + ) + for agent in (self.agent, self.deep_agent) ) self.bus = EventBus() self.conversations = Conversations( db=self.db, - agents=AgentRegistry([self.agent]), - backends={"a": self.backend}, + agents=AgentRegistry([self.agent, self.deep_agent]), + backends={"a": self.backend, "d": self.deep_backend}, bus=self.bus, pool=self.pool, store=self.store, texts=ConversationTexts(), + frontends=[self.api, self.markdown], idle_interval=3600, ) + self.markdown.conversations = self.conversations + self.api.conversations = self.conversations async def setup(self) -> "World": await self.db.create_all() @@ -460,7 +511,7 @@ async def test_read_and_bindings(world: World) -> None: build_entries(history, session_id=sid, cwd=str(world.root), model="m"), ) conv = await world.conversations.create( - kind="deep", agent="a", origin="test", session_id=sid + kind="deep", agent="d", origin="test", session_id=sid ) assert ( await world.conversations.read(conv) @@ -544,3 +595,42 @@ def test_transcript_helpers() -> None: ) == "user:\nq\n\nassistant:\na" ) + + +async def test_agent_kinds_gate_create(world: World) -> None: + with pytest.raises(ValueError, match="does not serve kind 'deep'"): + await world.conversations.create(kind="deep", agent="a", origin="test") + with pytest.raises(ValueError, match="does not serve kind 'master'"): + await world.conversations.create(kind="master", agent="d", origin="test") + with pytest.raises(ValueError, match="'d' does not serve kind 'master'"): + await world.conversations.spawn(kind="master", seed="clean", agent="d") + conv = await world.conversations.create(kind="deep", agent="d", origin="test") + assert conv.agent_name == "d" + + +async def test_frontend_kinds_gate_bind(world: World) -> None: + master = await world.conversations.create(kind="master", agent="a", origin="test") + with pytest.raises(ValueError, match="'markdown' does not show kind 'master'"): + await world.conversations.bind(master, frontend="markdown", external_id="m.md") + with pytest.raises(ValueError, match="unknown frontend 'telegram'"): + await world.conversations.bind(master, frontend="telegram", external_id="1") + await world.conversations.bind(master, frontend="api", external_id="x") + + +async def test_spawn_defaults_agent_and_materializes(world: World) -> None: + master = await world.conversations.create(kind="master", agent="a", origin="test") + deep = await world.conversations.spawn( + kind="deep", seed="brief", text="dig", title="t", parent=master, origin="mcp" + ) + assert deep.agent_name == "d" + assert world.markdown.materialized == [deep.external_id] + bindings = await world.conversations.bindings(deep) + assert [(b.frontend, b.external_id) for b in bindings] == [ + ("markdown", f"markdown:{deep.external_id}") + ] + branch = await world.conversations.spawn(kind="branch", seed="clean", parent=master) + assert branch.agent_name == "a" + assert world.markdown.materialized == [deep.external_id] + with pytest.raises(ValueError, match="no default agent for kind 'job'"): + await world.conversations.spawn(kind="job", seed="clean") + await world.settle(deep, 1) diff --git a/tests/test_routing.py b/tests/test_routing.py new file mode 100644 index 0000000..1cc867c --- /dev/null +++ b/tests/test_routing.py @@ -0,0 +1,228 @@ +import asyncio +import tempfile +from pathlib import Path + +import frontmatter +import httpx +import pytest + +from beaver_gateway.core.auth import TokenStore +from beaver_gateway.core.conversation_store import load_messages +from beaver_gateway.core.gateway_tools import _tools +from beaver_gateway.core.registry import McpRegistry +from beaver_gateway.frontends.anthropic import AnthropicMessagesFrontend +from beaver_gateway.frontends.api import ApiFrontend +from beaver_gateway.frontends.base import GatewayRuntime +from beaver_gateway.frontends.markdown import MarkdownFrontend +from test_conversations import ScriptedClient, World + +AUTH = {"Authorization": "Bearer tok"} + + +class Stack: + def __init__(self, world: World) -> None: + self.world = world + self.vault = world.root / "vault" + self.api = ApiFrontend(default_agents={"master": "a"}) + self.markdown = MarkdownFrontend(vault_path=self.vault, default_agent="d") + self.anthropic = AnthropicMessagesFrontend() + frontends = [self.api, self.anthropic, self.markdown] + world.conversations._frontends = frontends + self.runtime = GatewayRuntime( + agents=world.conversations._agents, + mcps=McpRegistry([]), + backends=world.conversations._backends, + token_store=TokenStore(bootstrap={"t": "tok"}), + db=world.db, + conversations=world.conversations, + bus=world.bus, + pool=world.pool, + frontends=tuple(frontends), + ) + for fe in frontends: + fe.configure(self.runtime) + self.mirror = asyncio.create_task(self.markdown.mirror.run()) + + def client(self, fe) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=fe._app), base_url="http://t" + ) + + async def close(self) -> None: + self.mirror.cancel() + + async def file_of(self, conv_id: str) -> Path: + conv = await self.world.conversations.get(conv_id) + path = await self.markdown.mirror.bound_path(conv) + assert path is not None + return path + + async def wait_file(self, path: Path, needle: str, timeout: float = 5.0) -> str: + deadline = asyncio.get_running_loop().time() + timeout + while asyncio.get_running_loop().time() < deadline: + if path.exists() and needle in path.read_text(encoding="utf-8"): + return path.read_text(encoding="utf-8") + await asyncio.sleep(0.02) + msg = f"{needle!r} never showed up in {path}" + raise AssertionError(msg) + + +@pytest.fixture +async def stack() -> Stack: + root = Path(tempfile.mkdtemp(prefix="beaver-routing-")) + world = await World(root).setup() + await world.conversations.start() + s = Stack(world) + yield s + await s.close() + await world.conversations.stop() + await world.pool.close_all() + await world.db.dispose() + + +async def test_api_rejects_deep_with_dispatcher(stack: Stack) -> None: + async with stack.client(stack.api) as c: + r = await c.post( + "/api/conversations", json={"kind": "deep", "agent": "a"}, headers=AUTH + ) + assert r.status_code == 400 + assert "does not serve kind 'deep'" in r.json()["error"] + r = await c.post("/api/conversations", json={"kind": "job"}, headers=AUTH) + assert r.status_code == 400 + assert "no default agent" in r.json()["error"] + r = await c.get("/api/agents", headers=AUTH) + agents = {a["name"]: a["kinds"] for a in r.json()["agents"]} + assert agents == {"a": ["master", "branch", "job", "fork"], "d": ["deep"]} + homes = {f["name"]: f["default_agents"] for f in r.json()["frontends"]} + assert homes == { + "api": {"master": "a"}, + "anthropic": {}, + "markdown": {"deep": "d"}, + } + + +async def test_api_spawn_deep_lands_in_vault(stack: Stack) -> None: + async with stack.client(stack.api) as c: + r = await c.post("/api/conversations", json={"kind": "master"}, headers=AUTH) + assert r.status_code == 201 and r.json()["agent"] == "a" + master = r.json()["id"] + r = await c.post( + "/api/conversations", + json={"kind": "deep", "seed": "brief", "text": "dig", "title": "Тема"}, + headers=AUTH, + ) + assert r.status_code == 201, r.text + body = r.json() + assert body["agent"] == "d" + assert [b["frontend"] for b in body["bindings"]] == ["markdown"] + rel = body["bindings"][0]["external_id"] + assert rel.endswith("_Тема.md") and rel.startswith("_logs/d/") + r = await c.post( + f"/api/conversations/{master}/bind", + json={"frontend": "markdown", "external_id": "x.md"}, + headers=AUTH, + ) + assert r.status_code == 400 + path = stack.vault / rel + text = await stack.wait_file(path, "ok:[сид: brief]") + post = frontmatter.loads(text) + assert post.metadata == {"agent": "d", "conversation_id": body["id"]} + assert post.content.startswith("### User:\n\n[сид: brief] deep «Тема», ") + assert post.content.rstrip().endswith("### User:") + + +async def test_markdown_chat_rejects_dispatcher(stack: Stack) -> None: + async with stack.client(stack.markdown) as c: + r = await c.post( + "/chat", + json={"filename": "x.md", "agent": "a", "content": "### User:\n\nhi\n"}, + headers=AUTH, + ) + assert r.status_code == 400 + assert "does not serve kind 'deep'" in r.json()["detail"] + + +async def test_spawn_tool_reports_bad_pair(stack: Stack) -> None: + master = await stack.world.conversations.create( + kind="master", agent="a", origin="test" + ) + spawn = next( + t + for t in _tools(stack.world.conversations, master.external_id) + if t.name == "spawn" + ) + out = await spawn.handler({"kind": "deep", "agent": "a"}) + assert ( + out.get("is_error") + and "does not serve kind 'deep'" in out["content"][0]["text"] + ) + out = await spawn.handler({"kind": "deep", "seed": "brief", "text": "go"}) + assert not out.get("is_error") + conv_id = out["content"][0]["text"].split()[-1] + path = await stack.file_of(conv_id) + assert path.exists() + + +async def test_anthropic_turns_become_one_deep_conversation(stack: Stack) -> None: + async with stack.client(stack.anthropic) as c: + r = await c.post( + "/v1/messages", + json={"model": "a", "messages": [{"role": "user", "content": "hi"}]}, + headers=AUTH, + ) + assert r.status_code == 400 + first = [{"role": "user", "content": "first question"}] + r = await c.post( + "/v1/messages", json={"model": "d", "messages": first}, headers=AUTH + ) + assert r.status_code == 200, r.text + reply = r.json()["content"][0]["text"] + assert reply == "ok:first question" + convs = await stack.world.conversations.find(kind="deep") + assert len(convs) == 1 + conv = convs[0] + assert conv.title == "first question" and conv.frontend == "anthropic" + path = await stack.file_of(conv.external_id) + text = await stack.wait_file(path, "ok:first question") + assert frontmatter.loads(text).metadata == { + "agent": "d", + "conversation_id": conv.external_id, + } + + second = [ + *first, + {"role": "assistant", "content": reply}, + {"role": "user", "content": "second"}, + ] + r = await c.post( + "/v1/messages", + json={"model": "d", "messages": second, "stream": True}, + headers=AUTH, + ) + assert r.status_code == 200 + assert "ok:second" in r.text + assert len(await stack.world.conversations.find(kind="deep")) == 1 + assert len(ScriptedClient.instances) == 1 + assert ScriptedClient.instances[0].prompts == ["first question", "second"] + text = await stack.wait_file(path, "ok:second") + assert text.count("### User:") == 3 and text.count("### Assistant:") == 2 + bindings = await stack.world.conversations.bindings(conv) + assert [b.frontend for b in bindings if b.visible] == ["markdown", "anthropic"] + async with stack.world.db.session() as session: + stored = await load_messages(session, conversation_id=conv.id) + assert [m["role"] for m in stored] == ["user", "assistant", "user", "assistant"] + assert stored[-1]["content"] == [{"type": "text", "text": "ok:second"}] + + r = await c.post( + "/v1/messages", + json={ + "model": "a", + "messages": [ + *second, + {"role": "assistant", "content": "ok:second"}, + {"role": "user", "content": "x"}, + ], + }, + headers=AUTH, + ) + assert r.status_code == 400 and "runs on 'd'" in r.json()["detail"]