983 lines
37 KiB
Python
983 lines
37 KiB
Python
"""``MarkdownFrontend`` — chat-via-markdown-files frontend.
|
|
|
|
Wires:
|
|
|
|
* ``POST /chat {filename, content?, agent?}`` — bearer-authenticated
|
|
trigger. The plugin in Obsidian fires this after the user edits a
|
|
``.md`` and the file gets synced (or with ``content`` to short-circuit
|
|
the sync delay). We parse the file, check the last turn — assistant
|
|
→ no-op, user → run the agent and append.
|
|
* ``GET /healthz`` — liveness.
|
|
|
|
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 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. 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
|
|
frontend's completed turns also land in the vault. The handler logic
|
|
lives in :mod:`.crossfront` so this module stays focused on the HTTP
|
|
shape.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import time
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
from anthropic.types import RawContentBlockStopEvent
|
|
from fastapi import FastAPI, HTTPException, Request, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse, StreamingResponse
|
|
|
|
from beaver_gateway.core import audit
|
|
from beaver_gateway.core.conversation_store import (
|
|
diff_and_fork,
|
|
load_messages,
|
|
rewrite_messages,
|
|
)
|
|
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
|
|
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.core.kinds import Kind
|
|
from beaver_gateway.frontends.base import GatewayRuntime
|
|
from beaver_gateway.storage.models import Conversation, ConversationBinding
|
|
|
|
|
|
_log = logging.getLogger("beaver_gateway.frontends.markdown")
|
|
|
|
|
|
__all__ = ["MarkdownFrontend"]
|
|
|
|
|
|
# How often we re-render the assistant turn into the .md file while the
|
|
# backend stream is still open. Trades responsiveness (faster updates to
|
|
# Obsidian sync / Raycast tailers) against write amplification. Each
|
|
# ``RawContentBlockStopEvent`` also forces a flush regardless of the
|
|
# timer, so block boundaries always land in the file.
|
|
_STREAM_FLUSH_DEBOUNCE = 0.4
|
|
|
|
# Debounce for the SSE ``/chat/stream`` path. Network IO is cheaper than
|
|
# atomic file rewrites, so we send updates more frequently — the client
|
|
# wants the lowest possible latency and we control the renderer on the
|
|
# other end (the Obsidian plugin splices deltas into the editor, no
|
|
# disk round-trip).
|
|
_SSE_FLUSH_DEBOUNCE = 0.1
|
|
|
|
|
|
class MarkdownFrontend(Frontend):
|
|
"""FastAPI app behind ``POST /chat`` driven by Obsidian-vault files."""
|
|
|
|
name = FRONTEND
|
|
kinds = ("deep",)
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
vault_path: Path | str,
|
|
host: str = "0.0.0.0", # noqa: S104
|
|
port: int = 8003,
|
|
default_agent: str | None = None,
|
|
log_all_chats: bool = False,
|
|
logged_subdir: str = "_logs",
|
|
chat_path: Callable[[str, str, Path], Path] | None = None,
|
|
public_base_url: str | None = None,
|
|
) -> None:
|
|
self.vault_path = Path(vault_path).expanduser().resolve()
|
|
self.host = host
|
|
self.port = port
|
|
self.default_agent = default_agent
|
|
self.log_all_chats = log_all_chats
|
|
self.logged_subdir = logged_subdir
|
|
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
|
|
# raw host:port".
|
|
self.public_base_url = public_base_url.rstrip("/") if public_base_url else None
|
|
self._runtime: GatewayRuntime | None = None
|
|
self._app: FastAPI | None = None
|
|
# Files currently being processed by an in-flight ``POST /chat``.
|
|
# Checked-and-added atomically in the request handler (no
|
|
# ``await`` between the check and the insert) so a concurrent
|
|
# 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:
|
|
msg = "MarkdownFrontend needs runtime.conversations"
|
|
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,
|
|
chat_path=self.chat_path,
|
|
)
|
|
# Scan the existing logged files synchronously here so the
|
|
# fingerprint→path map is populated before the first
|
|
# cross-frontend turn arrives. Cheap: frontmatter-only read.
|
|
self._crossfront.warm_index()
|
|
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: Kind) -> 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
|
|
|
|
if self._app is None:
|
|
msg = "configure() must be called before serve()"
|
|
raise RuntimeError(msg)
|
|
config = uvicorn.Config(
|
|
self._app, host=self.host, port=self.port, log_level="info"
|
|
)
|
|
server = uvicorn.Server(config)
|
|
mirror = asyncio.create_task(self.mirror.run())
|
|
try:
|
|
await server.serve()
|
|
finally:
|
|
mirror.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await mirror
|
|
|
|
# ---- app builder ---------------------------------------------------
|
|
|
|
def _build_app(self, runtime: GatewayRuntime) -> FastAPI:
|
|
app = FastAPI(title="beaver-gateway / Markdown")
|
|
|
|
# ``/chat/stream`` is consumed via ``fetch`` from the Obsidian
|
|
# plugin (``requestUrl`` can't read a body incrementally), and
|
|
# ``fetch`` is subject to CORS. Auth is bearer-token so we don't
|
|
# need credentialed mode; allow any origin and the standard
|
|
# methods/headers. The other endpoints are happy to ride along.
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/agents")
|
|
async def list_agents(request: Request) -> dict[str, Any]:
|
|
await require_token(request, runtime, scope="messages")
|
|
return {"agents": [{"name": a.name} for a in runtime.agents]}
|
|
|
|
@app.post("/chat")
|
|
async def chat(request: Request) -> Any:
|
|
token_name = await require_token(request, runtime, scope="messages")
|
|
try:
|
|
body = await request.json()
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
|
|
) from exc
|
|
|
|
filename = body.get("filename")
|
|
if not isinstance(filename, str) or not filename.strip():
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "missing or non-string `filename`"
|
|
)
|
|
content_override = body.get("content")
|
|
agent_override = body.get("agent")
|
|
if agent_override is not None and not isinstance(agent_override, str):
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "`agent` must be a string"
|
|
)
|
|
|
|
file_path = self._resolve_path(filename)
|
|
|
|
# Atomic check-and-claim: both ops run between awaits, so a
|
|
# second request can't slip into the same file slot.
|
|
if file_path in self._busy:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
content={"status": "in_progress", "filename": filename},
|
|
)
|
|
self._busy.add(file_path)
|
|
try:
|
|
return await self._handle_chat(
|
|
runtime=runtime,
|
|
token_name=token_name,
|
|
filename=filename,
|
|
file_path=file_path,
|
|
content_override=content_override,
|
|
agent_override=agent_override,
|
|
)
|
|
finally:
|
|
self._busy.discard(file_path)
|
|
|
|
@app.post("/chat/stream")
|
|
async def chat_stream(request: Request) -> Any:
|
|
# Same contract as ``/chat`` (bearer auth, identical body),
|
|
# but the response is ``text/event-stream`` and intermediate
|
|
# rendered states are pushed as ``delta`` events. The
|
|
# gateway-side disk write only happens once, at end of turn,
|
|
# so streaming consumers (Obsidian plugin) and Obsidian Sync
|
|
# don't fight over the same file mid-stream.
|
|
token_name = await require_token(request, runtime, scope="messages")
|
|
try:
|
|
body = await request.json()
|
|
except json.JSONDecodeError as exc:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, f"invalid JSON: {exc}"
|
|
) from exc
|
|
|
|
filename = body.get("filename")
|
|
if not isinstance(filename, str) or not filename.strip():
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "missing or non-string `filename`"
|
|
)
|
|
content_override = body.get("content")
|
|
agent_override = body.get("agent")
|
|
if agent_override is not None and not isinstance(agent_override, str):
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "`agent` must be a string"
|
|
)
|
|
|
|
file_path = self._resolve_path(filename)
|
|
|
|
# 409 path stays JSON — the stream hasn't started yet, so
|
|
# the caller can read it the same way as on ``/chat``.
|
|
if file_path in self._busy:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
content={"status": "in_progress", "filename": filename},
|
|
)
|
|
self._busy.add(file_path)
|
|
|
|
async def gen() -> AsyncIterator[bytes]:
|
|
try:
|
|
async for chunk in self._handle_chat_streaming(
|
|
runtime=runtime,
|
|
token_name=token_name,
|
|
filename=filename,
|
|
file_path=file_path,
|
|
content_override=content_override,
|
|
agent_override=agent_override,
|
|
):
|
|
yield chunk
|
|
finally:
|
|
self._busy.discard(file_path)
|
|
|
|
return StreamingResponse(
|
|
gen(), media_type="text/event-stream", headers=SSE_HEADERS
|
|
)
|
|
|
|
return app
|
|
|
|
# ---- dispatch ------------------------------------------------------
|
|
|
|
async def _handle_chat(
|
|
self,
|
|
*,
|
|
runtime: GatewayRuntime,
|
|
token_name: str,
|
|
filename: str,
|
|
file_path: Path,
|
|
content_override: Any,
|
|
agent_override: str | None,
|
|
) -> Any:
|
|
write_disk = content_override is None
|
|
if isinstance(content_override, str):
|
|
file_text = content_override
|
|
elif content_override is None:
|
|
file_text = await read_or_empty(file_path)
|
|
else:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "`content` must be a string when present"
|
|
)
|
|
|
|
parsed = parser.parse(file_text)
|
|
agent_name = parser.resolve_agent(
|
|
metadata=parsed.metadata,
|
|
request_override=agent_override,
|
|
default=self.default_agent,
|
|
)
|
|
if not agent_name:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST,
|
|
"no agent specified: pass `agent`, set frontmatter, "
|
|
"or configure `default_agent`",
|
|
)
|
|
|
|
# When the parser produced no messages (file empty / only
|
|
# frontmatter), there's nothing to dispatch.
|
|
if not parsed.messages:
|
|
return {
|
|
"status": "nothing_to_do",
|
|
"reason": "empty file",
|
|
"new_content": file_text,
|
|
}
|
|
|
|
if parser.last_role(parsed.messages) == "assistant":
|
|
return {
|
|
"status": "nothing_to_do",
|
|
"reason": "last turn is assistant",
|
|
"new_content": file_text,
|
|
}
|
|
|
|
agent = runtime.agents.get(agent_name)
|
|
if agent is None:
|
|
raise HTTPException(
|
|
status.HTTP_404_NOT_FOUND, f"unknown agent: {agent_name!r}"
|
|
)
|
|
backend = runtime.backends.get(agent.name)
|
|
if backend is None:
|
|
raise HTTPException(
|
|
status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
f"no backend configured for agent {agent.name!r}",
|
|
)
|
|
|
|
_log.info(
|
|
"chat: actor=%s agent=%s file=%s msgs=%d",
|
|
token_name,
|
|
agent.name,
|
|
filename,
|
|
len(parsed.messages),
|
|
)
|
|
await audit.log(
|
|
runtime,
|
|
actor=f"token:{token_name}",
|
|
kind="markdown_chat",
|
|
agent_name=agent.name,
|
|
filename=filename,
|
|
msgs=len(parsed.messages),
|
|
)
|
|
|
|
# Resolve / mint the conversation row, align incoming against
|
|
# 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,
|
|
file_path=file_path,
|
|
)
|
|
outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns)
|
|
capture = TurnCapture()
|
|
events = runtime.conversations.turn(
|
|
conv,
|
|
messages=outcome.messages,
|
|
origin="user",
|
|
capture=capture,
|
|
use_session=outcome.divergence_index is None,
|
|
)
|
|
try:
|
|
message = await self._stream_to_file(
|
|
events=events,
|
|
file_path=file_path,
|
|
parsed=parsed,
|
|
model=agent.model or agent.name,
|
|
filename=filename,
|
|
write_disk=write_disk,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(
|
|
status.HTTP_500_INTERNAL_SERVER_ERROR, f"backend error: {exc}"
|
|
) from exc
|
|
|
|
new_content = await self._write_assistant_reply(
|
|
file_path=file_path,
|
|
parsed=parsed,
|
|
message=message,
|
|
agent_name=agent.name,
|
|
conv_external_id=conv_external_id,
|
|
write_disk=write_disk,
|
|
)
|
|
|
|
await self._persist_canonical_history(
|
|
runtime=runtime,
|
|
conversation_id=cast("int", conv.id),
|
|
persist_messages=outcome.persist_messages,
|
|
new_user_text=parsed.turns[-1].text,
|
|
capture=capture,
|
|
message=message,
|
|
)
|
|
|
|
# Broadcast our own turn so other handlers (none today, but the
|
|
# symmetry is worth keeping) see what happened. ``source`` marks
|
|
# the origin so ``CrossFrontendLogger`` can skip its own files.
|
|
record = TurnRecord(
|
|
agent_name=agent.name,
|
|
input_messages=list(parsed.messages),
|
|
output_message=message,
|
|
system=None,
|
|
source="markdown",
|
|
)
|
|
for handler in runtime.turn_log_handlers:
|
|
try:
|
|
await handler(record)
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("turn_log_handler raised; continuing")
|
|
|
|
return {
|
|
"status": "ok",
|
|
"turns_appended": 1,
|
|
"agent": agent.name,
|
|
"new_content": new_content,
|
|
}
|
|
|
|
# ---- streaming dispatch (SSE) --------------------------------------
|
|
|
|
async def _handle_chat_streaming( # noqa: PLR0915 — mirrors _handle_chat, splitting only doubles read cost
|
|
self,
|
|
*,
|
|
runtime: GatewayRuntime,
|
|
token_name: str,
|
|
filename: str,
|
|
file_path: Path,
|
|
content_override: Any,
|
|
agent_override: str | None,
|
|
) -> AsyncIterator[bytes]:
|
|
"""SSE counterpart of :meth:`_handle_chat`.
|
|
|
|
Mirrors the same pipeline (resolve file → parse → resolve agent →
|
|
run backend → persist), but emits ``event: delta`` frames as the
|
|
rendered turn grows and a single terminal ``event: done`` /
|
|
``event: error``. Errors that ``_handle_chat`` would surface as
|
|
``HTTPException`` go out as ``error`` frames here (the HTTP
|
|
envelope is already 200 by the time the stream starts).
|
|
|
|
Intermediate disk writes are deliberately skipped — only the
|
|
post-stream :meth:`_write_assistant_reply` lands on disk, so the
|
|
gateway-side vault and the plugin-side editor are the only
|
|
writers in their respective halves of Obsidian Sync. Final
|
|
content is identical on both sides, so Sync no-ops.
|
|
"""
|
|
# With ``content`` the plugin is the only writer of the file
|
|
# (§3.10): the gateway never touches disk in that case.
|
|
write_disk = content_override is None
|
|
if isinstance(content_override, str):
|
|
file_text = content_override
|
|
elif content_override is None:
|
|
file_text = await read_or_empty(file_path)
|
|
else:
|
|
yield sse_pack(
|
|
"error",
|
|
{
|
|
"status_code": status.HTTP_400_BAD_REQUEST,
|
|
"detail": "`content` must be a string when present",
|
|
},
|
|
)
|
|
return
|
|
|
|
parsed = parser.parse(file_text)
|
|
agent_name = parser.resolve_agent(
|
|
metadata=parsed.metadata,
|
|
request_override=agent_override,
|
|
default=self.default_agent,
|
|
)
|
|
if not agent_name:
|
|
yield sse_pack(
|
|
"error",
|
|
{
|
|
"status_code": status.HTTP_400_BAD_REQUEST,
|
|
"detail": (
|
|
"no agent specified: pass `agent`, set frontmatter, "
|
|
"or configure `default_agent`"
|
|
),
|
|
},
|
|
)
|
|
return
|
|
|
|
if not parsed.messages:
|
|
yield sse_pack(
|
|
"done",
|
|
{
|
|
"status": "nothing_to_do",
|
|
"reason": "empty file",
|
|
"new_content": file_text,
|
|
},
|
|
)
|
|
return
|
|
|
|
if parser.last_role(parsed.messages) == "assistant":
|
|
yield sse_pack(
|
|
"done",
|
|
{
|
|
"status": "nothing_to_do",
|
|
"reason": "last turn is assistant",
|
|
"new_content": file_text,
|
|
},
|
|
)
|
|
return
|
|
|
|
agent = runtime.agents.get(agent_name)
|
|
if agent is None:
|
|
yield sse_pack(
|
|
"error",
|
|
{
|
|
"status_code": status.HTTP_404_NOT_FOUND,
|
|
"detail": f"unknown agent: {agent_name!r}",
|
|
},
|
|
)
|
|
return
|
|
backend = runtime.backends.get(agent.name)
|
|
if backend is None:
|
|
yield sse_pack(
|
|
"error",
|
|
{
|
|
"status_code": status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
"detail": f"no backend configured for agent {agent.name!r}",
|
|
},
|
|
)
|
|
return
|
|
|
|
_log.info(
|
|
"chat/stream: actor=%s agent=%s file=%s msgs=%d",
|
|
token_name,
|
|
agent.name,
|
|
filename,
|
|
len(parsed.messages),
|
|
)
|
|
await audit.log(
|
|
runtime,
|
|
actor=f"token:{token_name}",
|
|
kind="markdown_chat_stream",
|
|
agent_name=agent.name,
|
|
filename=filename,
|
|
msgs=len(parsed.messages),
|
|
)
|
|
|
|
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",
|
|
filename,
|
|
conv_external_id,
|
|
conv.id or -1,
|
|
len(stored_msgs),
|
|
len(parsed.turns),
|
|
)
|
|
outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns)
|
|
_log.info(
|
|
"chat/stream: file=%s diff_and_fork divergence=%s "
|
|
"backend_msgs=%d persist_msgs=%d",
|
|
filename,
|
|
outcome.divergence_index,
|
|
len(outcome.messages),
|
|
len(outcome.persist_messages),
|
|
)
|
|
capture = TurnCapture()
|
|
_log.info(
|
|
"chat/stream: file=%s calling backend.complete agent=%s session=%s",
|
|
filename,
|
|
agent.name,
|
|
conv.session_id,
|
|
)
|
|
events = runtime.conversations.turn(
|
|
conv,
|
|
messages=outcome.messages,
|
|
origin="user",
|
|
capture=capture,
|
|
use_session=outcome.divergence_index is None,
|
|
)
|
|
|
|
acc = StreamAccumulator()
|
|
model = agent.model or agent.name
|
|
last_flush = time.monotonic()
|
|
last_payload: str | None = None
|
|
|
|
def snapshot() -> str | None:
|
|
partial = acc.finalize(model=model)
|
|
if not partial.content:
|
|
return None
|
|
rendered = renderer.render_assistant_message(partial)
|
|
new_body = renderer.append_to_body(parsed.body, rendered)
|
|
return reattach_frontmatter(parsed.metadata, new_body)
|
|
|
|
try:
|
|
async for ev in events_with_heartbeat(events):
|
|
if ev is None:
|
|
yield KEEPALIVE
|
|
continue
|
|
acc.feed(ev)
|
|
now = time.monotonic()
|
|
if (
|
|
isinstance(ev, RawContentBlockStopEvent)
|
|
or (now - last_flush) >= _SSE_FLUSH_DEBOUNCE
|
|
):
|
|
payload = snapshot()
|
|
# Skip duplicate snapshots — e.g. tool_use blocks
|
|
# 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})
|
|
last_payload = payload
|
|
last_flush = now
|
|
except Exception as exc: # noqa: BLE001 — wire any backend failure as an SSE error frame
|
|
_log.exception("backend failed for %s", filename)
|
|
# Mirror the legacy path: write the last partial + an error
|
|
# callout to disk so other consumers (logs, file watchers)
|
|
# see what arrived. The client gets a clean SSE ``error``.
|
|
partial = acc.finalize(model=model)
|
|
new_body = parsed.body
|
|
if partial.content:
|
|
new_body = renderer.append_to_body(
|
|
new_body, renderer.render_assistant_message(partial)
|
|
)
|
|
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)
|
|
)
|
|
yield sse_pack(
|
|
"error",
|
|
{
|
|
"status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
"detail": f"backend error: {exc}",
|
|
},
|
|
)
|
|
return
|
|
|
|
message = acc.finalize(model=model)
|
|
|
|
new_content = await self._write_assistant_reply(
|
|
file_path=file_path,
|
|
parsed=parsed,
|
|
message=message,
|
|
agent_name=agent.name,
|
|
conv_external_id=conv_external_id,
|
|
write_disk=write_disk,
|
|
)
|
|
|
|
await self._persist_canonical_history(
|
|
runtime=runtime,
|
|
conversation_id=cast("int", conv.id),
|
|
persist_messages=outcome.persist_messages,
|
|
new_user_text=parsed.turns[-1].text,
|
|
capture=capture,
|
|
message=message,
|
|
)
|
|
|
|
record = TurnRecord(
|
|
agent_name=agent.name,
|
|
input_messages=list(parsed.messages),
|
|
output_message=message,
|
|
system=None,
|
|
source="markdown",
|
|
)
|
|
for handler in runtime.turn_log_handlers:
|
|
try:
|
|
await handler(record)
|
|
except Exception: # noqa: BLE001
|
|
_log.exception("turn_log_handler raised; continuing")
|
|
|
|
yield sse_pack(
|
|
"done",
|
|
{
|
|
"status": "ok",
|
|
"turns_appended": 1,
|
|
"agent": agent.name,
|
|
"new_content": new_content,
|
|
},
|
|
)
|
|
|
|
# ---- helpers -------------------------------------------------------
|
|
|
|
async def _stream_to_file(
|
|
self,
|
|
*,
|
|
events: Any,
|
|
file_path: Path,
|
|
parsed: parser.ParsedFile,
|
|
model: str,
|
|
filename: str,
|
|
write_disk: bool = True,
|
|
) -> Any:
|
|
"""Drain ``events`` into a ``Message``, flushing partials to disk.
|
|
|
|
Flushes happen on each ``RawContentBlockStopEvent`` (natural
|
|
block boundary, content is markdown-consistent) and on the
|
|
``_STREAM_FLUSH_DEBOUNCE`` timer between events. The partial
|
|
write keeps the as-parsed frontmatter; the post-stream final
|
|
write in ``_write_assistant_reply`` is what stamps the refreshed
|
|
fingerprint / agent / conversation_id.
|
|
|
|
On backend exception we still flush the last partial and append
|
|
an error callout, so the human sees both what arrived and why it
|
|
stopped. The exception propagates so ``_handle_chat`` can map it
|
|
to a 500.
|
|
"""
|
|
acc = StreamAccumulator()
|
|
|
|
async def flush_partial() -> None:
|
|
if not write_disk:
|
|
return
|
|
partial = acc.finalize(model=model)
|
|
if not partial.content:
|
|
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)
|
|
)
|
|
|
|
try:
|
|
last_flush = time.monotonic()
|
|
async for ev in events:
|
|
acc.feed(ev)
|
|
now = time.monotonic()
|
|
if (
|
|
isinstance(ev, RawContentBlockStopEvent)
|
|
or (now - last_flush) >= _STREAM_FLUSH_DEBOUNCE
|
|
):
|
|
await flush_partial()
|
|
last_flush = now
|
|
except Exception as exc:
|
|
_log.exception("backend failed for %s", filename)
|
|
partial = acc.finalize(model=model)
|
|
new_body = parsed.body
|
|
if partial.content:
|
|
new_body = renderer.append_to_body(
|
|
new_body, renderer.render_assistant_message(partial)
|
|
)
|
|
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)
|
|
)
|
|
raise
|
|
return acc.finalize(model=model)
|
|
|
|
async def _write_assistant_reply(
|
|
self,
|
|
*,
|
|
file_path: Path,
|
|
parsed: parser.ParsedFile,
|
|
message: Any,
|
|
agent_name: str,
|
|
conv_external_id: str,
|
|
write_disk: bool = True,
|
|
) -> str:
|
|
"""Render the assistant turn, refresh frontmatter, write if we own the file."""
|
|
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)
|
|
updated_metadata = dict(parsed.metadata)
|
|
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)
|
|
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,
|
|
file_path: Path,
|
|
) -> tuple[Conversation, str, list[dict[str, Any]]]:
|
|
"""Resolve the ``deep`` conversation for this file + its stored messages.
|
|
|
|
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")
|
|
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:
|
|
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
|
|
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:
|
|
stored = await load_messages(session, conversation_id=conv.id)
|
|
return conv, conv.external_id, stored
|
|
|
|
async def _persist_canonical_history(
|
|
self,
|
|
*,
|
|
runtime: GatewayRuntime,
|
|
conversation_id: int,
|
|
persist_messages: list[dict[str, Any]],
|
|
new_user_text: str,
|
|
capture: TurnCapture,
|
|
message: Any,
|
|
) -> None:
|
|
"""Stamp the DB with the post-turn canonical Anthropic-shape history.
|
|
|
|
Combines the matched/spliced prior state, the new user prompt,
|
|
and the synthesized assistant/tool cycle from the backend (or a
|
|
text-only fallback for backends that left ``capture`` empty).
|
|
"""
|
|
new_user_msg = {"role": "user", "content": new_user_text}
|
|
synthesized = capture.synthesized_messages or _fallback_synthesized(message)
|
|
canonical = [*persist_messages, new_user_msg, *synthesized]
|
|
_log.info(
|
|
"_persist_canonical_history: conv_id=%d writing %d msgs "
|
|
"(prior=%d + new_user + synth=%d)",
|
|
conversation_id,
|
|
len(canonical),
|
|
len(persist_messages),
|
|
len(synthesized),
|
|
)
|
|
async with runtime.db.session() as session:
|
|
await rewrite_messages(
|
|
session, conversation_id=conversation_id, messages=canonical
|
|
)
|
|
_log.info(
|
|
"_persist_canonical_history: conv_id=%d DB committed", conversation_id
|
|
)
|
|
|
|
def _resolve_path(self, filename: str) -> Path:
|
|
"""Resolve ``filename`` under the vault; reject escapes."""
|
|
# ``filename`` may be relative or absolute; we always anchor
|
|
# under ``vault_path`` so absolute paths from outside the vault
|
|
# don't sneak through. ``Path("/foo/bar")`` combined with a
|
|
# vault path keeps the absolute side; we strip leading slashes
|
|
# to coerce the rooted form into a relative path before joining.
|
|
rel = filename.lstrip("/")
|
|
if not rel.endswith(".md"):
|
|
rel = rel + ".md"
|
|
candidate = (self.vault_path / rel).resolve()
|
|
try:
|
|
candidate.relative_to(self.vault_path)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, f"filename escapes vault: {filename!r}"
|
|
) from exc
|
|
return candidate
|
|
|
|
|
|
def _fallback_synthesized(message: Any) -> list[dict[str, Any]]:
|
|
"""Build a single-assistant ``synthesized_messages`` list from a raw ``Message``.
|
|
|
|
For backends that don't populate a :class:`TurnCapture` (anthropic
|
|
HTTP, raycast, …) we don't have access to per-tool-cycle
|
|
granularity, so the assistant reply lands in the DB as one
|
|
canonical-block message. Tool memory across cache misses would
|
|
degrade in that case, but those backends don't have the cache-miss
|
|
re-seed problem to begin with — they manage history client-side.
|
|
"""
|
|
content: list[dict[str, Any]] = []
|
|
for block in getattr(message, "content", ()):
|
|
btype = getattr(block, "type", None)
|
|
if btype == "text":
|
|
content.append({"type": "text", "text": getattr(block, "text", "") or ""})
|
|
elif btype == "tool_use":
|
|
content.append(
|
|
{
|
|
"type": "tool_use",
|
|
"id": getattr(block, "id", ""),
|
|
"name": getattr(block, "name", ""),
|
|
"input": getattr(block, "input", {}),
|
|
}
|
|
)
|
|
elif btype == "thinking":
|
|
content.append(
|
|
{
|
|
"type": "thinking",
|
|
"thinking": getattr(block, "thinking", "") or "",
|
|
"signature": getattr(block, "signature", "") or "",
|
|
}
|
|
)
|
|
if not content:
|
|
return []
|
|
return [{"role": "assistant", "content": content}]
|
|
|
|
|
|
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__
|
|
safe = msg.replace("\n", " ").strip()
|
|
return f"### Assistant:\n\n> [!error]-\n> {safe}\n"
|