From 0479d7cdda7e2af6bb4d73f84c97b5568229f846 Mon Sep 17 00:00:00 2001 From: h Date: Sat, 29 Aug 2026 01:44:05 +0200 Subject: [PATCH] feat(conversations): pre-SDK rows read from canonical messages, implied titles, markdown adopts them by first prompt --- src/beaver_gateway/core/conversations.py | 75 ++++++++++++++++++- src/beaver_gateway/frontends/api/frontend.py | 6 +- .../frontends/markdown/frontend.py | 16 +++- tests/test_api.py | 23 ++++++ tests/test_routing.py | 31 +++++++- 5 files changed, 145 insertions(+), 6 deletions(-) diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index 6cc5ec1..ba0189e 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -18,6 +18,7 @@ from __future__ import annotations import asyncio import contextlib import inspect +import json import logging import re import uuid @@ -38,6 +39,7 @@ from claude_agent_sdk import ( ) from sqlmodel import col, select +from beaver_gateway.core.conversation_store import load_messages from beaver_gateway.core.injects import InjectQueue, inject_header from beaver_gateway.core.kinds import KINDS, Kind, as_kind from beaver_gateway.core.transcript import ( @@ -52,13 +54,14 @@ from beaver_gateway.frontends._accumulate import StreamAccumulator from beaver_gateway.storage.models import ( Conversation, ConversationBinding, + ConversationMessage, InjectQueueItem, RateLimit, Schedule, ) if TYPE_CHECKING: - from collections.abc import AsyncIterator, Awaitable, Callable, Sequence + from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence from claude_agent_sdk import SessionStore @@ -453,6 +456,7 @@ class Conversations: async def describe(self, conv: Conversation) -> dict[str, Any]: out = self.public(conv) + out["title"] = await self.implied_title(conv) parent = await self.get_row(conv.parent_id) if conv.parent_id else None out["parent"] = parent.external_id if parent is not None else None out["bindings"] = [ @@ -629,8 +633,67 @@ class Conversations: return render_messages(await self.history(conv), window=window) async def history(self, conv: Conversation) -> list[dict[str, Any]]: + if conv.session_id is None: + async with self._db.session() as session: + return await load_messages( + session, conversation_id=cast("int", conv.id) + ) return messages_from_entries(cast("Any", await self.entries(conv))) + async def first_user_texts(self, ids: Iterable[int]) -> dict[int, str]: + wanted = list(ids) + if not wanted: + return {} + async with self._db.session() as session: + rows = ( + await session.exec( + select(ConversationMessage).where( + col(ConversationMessage.conversation_id).in_(wanted), + ConversationMessage.seq == 0, + ConversationMessage.role == "user", + ) + ) + ).all() + return { + r.conversation_id: text_of(json.loads(r.content_json)).strip() for r in rows + } + + async def implied_title(self, conv: Conversation) -> str | None: + if conv.title: + return conv.title + text = (await self.first_user_texts([cast("int", conv.id)])).get( + cast("int", conv.id) + ) + return implied_title(text) + + async def adopt(self, *, kind: Kind, first_user_text: str) -> Conversation | None: + """The one unbound, session-less conversation whose history starts here. + + Rows from before the SDK cut-over have canonical messages but no + window and no session; a vault file that begins with the same + prompt is that conversation continued. + """ + text = first_user_text.strip() + if not text: + return None + bound = select(ConversationBinding.conversation_id).where( + col(ConversationBinding.visible).is_(True) + ) + async with self._db.session() as session: + rows = ( + await session.exec( + select(Conversation).where( + Conversation.kind == kind, + Conversation.status == "open", + col(Conversation.session_id).is_(None), + col(Conversation.id).not_in(bound), + ) + ) + ).all() + firsts = await self.first_user_texts(cast("int", r.id) for r in rows) + hits = [r for r in rows if firsts.get(cast("int", r.id)) == text] + return hits[0] if len(hits) == 1 else None + async def entries(self, conv: Conversation, *, subpath: str = "") -> list[Any]: if conv.session_id is None: return [] @@ -1353,6 +1416,16 @@ def _iso(value: datetime | None) -> str | None: return _aware(value).isoformat(timespec="seconds") if value is not None else None +TITLE_MAX = 80 + + +def implied_title(text: str | None) -> str | None: + if not text: + return None + line = text.strip().splitlines()[0].strip() + return line if len(line) <= TITLE_MAX else line[: TITLE_MAX - 1] + "…" + + def _prompt_preview(messages: Sequence[Any], limit: int = 400) -> str | None: if not messages: return None diff --git a/src/beaver_gateway/frontends/api/frontend.py b/src/beaver_gateway/frontends/api/frontend.py index 1f0ec28..2207fc8 100644 --- a/src/beaver_gateway/frontends/api/frontend.py +++ b/src/beaver_gateway/frontends/api/frontend.py @@ -30,7 +30,7 @@ from sqlmodel import col, select from beaver_gateway.core import audit from beaver_gateway.core.auth import VALID_SCOPES, hash_token -from beaver_gateway.core.conversations import SEEDS +from beaver_gateway.core.conversations import SEEDS, implied_title from beaver_gateway.core.kinds import Kind, as_kind from beaver_gateway.frontends._auth import require_token from beaver_gateway.frontends._sse import ( @@ -215,10 +215,14 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa limit=query_int(request, "limit", 200), ) latest = await conversations.queue.latest(cast("int", r.id) for r in rows) + firsts = await conversations.first_user_texts( + cast("int", r.id) for r in rows if not r.title + ) return { "conversations": [ { **conversations.public(r), + "title": r.title or implied_title(firsts.get(cast("int", r.id))), "last_item": _queue_item(latest.get(cast("int", r.id))), } for r in rows diff --git a/src/beaver_gateway/frontends/markdown/frontend.py b/src/beaver_gateway/frontends/markdown/frontend.py index 8dce655..cc275f7 100644 --- a/src/beaver_gateway/frontends/markdown/frontend.py +++ b/src/beaver_gateway/frontends/markdown/frontend.py @@ -392,6 +392,7 @@ class MarkdownFrontend(Frontend): metadata=parsed.metadata, agent_name=agent.name, file_path=file_path, + first_user_text=parsed.turns[0].text, ) outcome = diff_and_fork(stored=stored_msgs, incoming=parsed.turns) capture = TurnCapture() @@ -587,6 +588,7 @@ class MarkdownFrontend(Frontend): metadata=parsed.metadata, agent_name=agent.name, file_path=file_path, + first_user_text=parsed.turns[0].text, ) except HTTPException as exc: yield sse_pack( @@ -823,13 +825,15 @@ class MarkdownFrontend(Frontend): metadata: dict[str, Any], agent_name: str, file_path: Path, + first_user_text: str = "", ) -> 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. + by its visible ``(markdown, path)`` binding, then by adopting the + one unbound pre-SDK conversation that starts with the same prompt; + 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() @@ -837,6 +841,12 @@ class MarkdownFrontend(Frontend): conv = await conversations.get(raw) if isinstance(raw, str) and raw else None if conv is None: conv = await conversations.find_bound(frontend=FRONTEND, external_id=rel) + if conv is None: + conv = await conversations.adopt( + kind="deep", first_user_text=first_user_text + ) + if conv is not None: + _log.info("adopted conversation %s for %s", conv.external_id, rel) if conv is None: try: conv = await conversations.create( diff --git a/tests/test_api.py b/tests/test_api.py index 3df958e..cf8ddbb 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -18,6 +18,7 @@ from claude_agent_sdk import ( from httpx import ASGITransport, AsyncClient from test_conversations import ScriptedClient, World +from beaver_gateway.core.conversation_store import rewrite_messages from beaver_gateway.core.auth import TokenStore from beaver_gateway.core.registry import McpRegistry from beaver_gateway.core.transcript import build_entries @@ -397,3 +398,25 @@ async def test_admin_login_hands_out_the_ui_bearer(world: World) -> None: assert me.status_code == 200 assert (await http.post("/admin/auth/logout")).status_code == 204 assert (await http.get("/admin/auth/session")).status_code == 401 + + +async def test_pre_sdk_rows_show_canonical_history_and_a_title(world: World) -> None: + api = Api(world) + conv = await world.conversations.create(kind="deep", agent="d", origin="markdown") + async with world.db.session() as session: + await rewrite_messages( + session, + conversation_id=cast("int", conv.id), + messages=[ + {"role": "user", "content": "так смотри, план на май\nвторая строка"}, + {"role": "assistant", "content": [{"type": "text", "text": "ок"}]}, + ], + ) + rows = (await api.get("/conversations"))["conversations"] + assert [r["title"] for r in rows if r["id"] == conv.external_id] == [ + "так смотри, план на май" + ] + shown = await api.get(f"/conversations/{conv.external_id}") + assert shown["title"] == "так смотри, план на май" and shown["session_id"] is None + history = await api.get(f"/conversations/{conv.external_id}/history") + assert [m["role"] for m in history["messages"]] == ["user", "assistant"] diff --git a/tests/test_routing.py b/tests/test_routing.py index 57d0fe9..c8db6b8 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -1,13 +1,14 @@ import asyncio import tempfile from pathlib import Path +from typing import cast 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.conversation_store import load_messages, rewrite_messages from beaver_gateway.core.gateway_tools import _tools from beaver_gateway.core.registry import McpRegistry from beaver_gateway.frontends.anthropic import AnthropicMessagesFrontend @@ -255,3 +256,31 @@ async def test_markdown_edited_reply_reseeds_the_session(stack: Stack) -> None: async with stack.world.db.session() as session: stored = await load_messages(session, conversation_id=1) assert stored[1]["content"] == [{"type": "text", "text": "penguin"}] + + +async def test_markdown_adopts_a_pre_sdk_conversation(stack: Stack) -> None: + world = stack.world + old = await world.conversations.create(kind="deep", agent="d", origin="markdown") + async with world.db.session() as session: + await rewrite_messages( + session, + conversation_id=cast("int", old.id), + messages=[ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "ok:hello"}, + ], + ) + content = "### User:\n\nhello\n\n---\n\n### Assistant:\n\nok:hello\n\n---\n\n### User:\n\nnext\n" + async with stack.client(stack.markdown) as c: + r = await c.post( + "/chat", + json={"filename": "old.md", "agent": "d", "content": content}, + headers=AUTH, + ) + assert r.status_code == 200, r.text + assert f"conversation_id: {old.external_id}" in r.json()["new_content"] + assert len(await world.conversations.find(kind="deep")) == 1 + bindings = await world.conversations.bindings(old) + assert [(b.frontend, b.external_id) for b in bindings] == [("markdown", "old.md")] + assert ScriptedClient.instances[-1].prompts == ["next"] + assert "ok:hello" in repr(vars(world.store))