From 23cfb162286e57d3aba7eeb8ee4f59dcb43e8ce4 Mon Sep 17 00:00:00 2001 From: h Date: Sat, 29 Aug 2026 03:15:39 +0200 Subject: [PATCH] feat(distill,conversations,scheduler,api,ui): close deep chats with a distiller fork, digest and index, idle and weekly state jobs --- src/beaver_gateway/cli.py | 1 + src/beaver_gateway/core/conversations.py | 359 +++++++++++++++- src/beaver_gateway/core/distill.py | 197 +++++++++ src/beaver_gateway/core/gateway_tools.py | 20 +- src/beaver_gateway/core/registry.py | 3 + src/beaver_gateway/core/scheduler.py | 40 +- src/beaver_gateway/frontends/api/frontend.py | 30 ++ tests/test_api.py | 28 ++ tests/test_distill.py | 419 +++++++++++++++++++ ui/src/lib/api/client.ts | 11 + ui/src/lib/panel/conversation-header.svelte | 14 + 11 files changed, 1103 insertions(+), 19 deletions(-) create mode 100644 src/beaver_gateway/core/distill.py create mode 100644 tests/test_distill.py diff --git a/src/beaver_gateway/cli.py b/src/beaver_gateway/cli.py index 47ad554..3621c25 100644 --- a/src/beaver_gateway/cli.py +++ b/src/beaver_gateway/cli.py @@ -181,6 +181,7 @@ async def _async_main() -> None: texts=gateway.texts, frontends=gateway.frontends, envelope=Envelope(watch=gateway.watch, tz=gateway.tz), + distiller=gateway.distiller, ) late.conversations = conversations scheduler = Scheduler( diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index d2603c3..e2c6d2a 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -40,6 +40,18 @@ from claude_agent_sdk import ( from sqlmodel import col, select from beaver_gateway.core.conversation_store import load_messages +from beaver_gateway.core.distill import ( + Digest, + DistillContext, + Distiller, + LineCap, + append_index, + check_digest, + find_digest, + index_line, + trim_summary, + written_paths, +) 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 ( @@ -83,6 +95,7 @@ __all__ = [ "SEEDS", "ConversationTexts", "Conversations", + "DistillResult", "ForkResult", "SeedContext", ] @@ -96,8 +109,19 @@ _DEFAULT_MERGE_PROMPT = ( "что не сделано и почему, открытые вопросы. Идентификаторы и ссылки - " "дословно. Коротко, прошедшее время." ) +_DEFAULT_DISTILL_PROMPT = ( + "Глубокий чат «{chat}» закрыт ({reason}), сегодня {day}. Напиши выжимку " + "файлом и слив текстом ответа: до 5 строк, третье лицо." +) +_DEFAULT_DISTILL_PROMPT_NO_MEMORY = ( + "Глубокий чат «{chat}» закрыт ({reason}), сегодня {day}. Память для него " + "выключена: файл не пиши, только слив текстом ответа - до 5 строк, третье лицо." +) _RELATIVE = re.compile(r"^\+(\d+)\s*([smhd])$") _UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400} +_CLOSE_WAIT = 0.25 +_CLOSE_TRIES = 40 +_CAP_TRIES = 3 @dataclass(frozen=True, slots=True) @@ -132,6 +156,18 @@ class ConversationTexts: "на утро, не задание - прошедшее время, без повелительного наклонения." ) new_day: str = "Новый день: мастер сменился, хендаут за {day} записан." + distill: Callable[[DistillContext], Awaitable[str] | str] | None = None + """The distiller fork's first message (§8.4): which chat, what day, where + the digest goes; ``None`` uses a path-less default.""" + + closed: str = "Закрыт глубокий чат [[{chat}]]{digest}.\n{text}" + """What the master hears about a closed deep chat; ``{digest}`` is + ``, выжимка [[…]]`` or nothing, ``{text}`` the distiller's merge.""" + + too_long: str = ( + "`{name}`: {lines} строк при потолке {max_lines}. Запись отбита, файл " + "возвращён к прежней версии. Сократи и перепиши." + ) @dataclass(frozen=True, slots=True) @@ -141,6 +177,18 @@ class ForkResult: capture: TurnCapture +@dataclass(frozen=True, slots=True) +class DistillResult: + """A closed deep chat: the merge text and, if memory was on, the digest.""" + + conversation: Conversation + fork: Conversation + text: str + digest: Digest | None + error: str | None + trimmed: bool + + @dataclass class _Runner: lock: asyncio.Lock = field(default_factory=asyncio.Lock) @@ -191,8 +239,10 @@ class Conversations: idle_interval: float = 3600.0, question_timeout: float = 600.0, envelope: Envelope | None = None, + distiller: Distiller | None = None, ) -> None: self._db = db + self._distiller = distiller self._agents = agents self._backends = backends self._bus = bus @@ -573,6 +623,7 @@ class Conversations: window: int | None = None, origin: str = "api", binding: tuple[str, str] | None = None, + flags: dict[str, Any] | None = None, ) -> Conversation: """Create a conversation and queue its seed turn (§8.2). @@ -610,6 +661,7 @@ class Conversations: title=title, origin=origin, session_id=session_id, + flags=flags, ) if binding is not None: await self.bind(conv, frontend=binding[0], external_id=binding[1]) @@ -658,13 +710,21 @@ class Conversations: window: int | None = None, strip_tools: bool = False, title: str | None = None, + agent: str | None = None, ) -> ForkResult: + """Copy the history into a one-off session and run ``prompt`` on it. + + ``agent`` runs the fork under another agent's prompt (the distiller + closing a deep chat, §8.4); the copy then lives under that agent's + project key. Forks get no MCP (§3.2). + """ + agent = agent or conv.agent_name session_id = await self._copy_session( - conv, window=window, strip_tools=strip_tools + conv, window=window, strip_tools=strip_tools, agent=agent ) child = await self.create( kind="fork", - agent=conv.agent_name, + agent=agent, parent=conv, title=title or f"fork: {conv.title or conv.external_id}", origin="system", @@ -675,7 +735,7 @@ class Conversations: child, prompt, origin="fork", tools=False ) finally: - await self._backend(conv.agent_name).close(child.external_id) + await self._backend(agent).close(child.external_id) child = await self.set_status(child, "closed") return ForkResult(conversation=child, text=text, capture=capture) @@ -882,6 +942,261 @@ class Conversations: await self._backend(conv.agent_name).close(conv.external_id) return row + # ---- §8.4 closing a deep chat --------------------------------------- + + async def request_close(self, conv: Conversation) -> Conversation: + """``close_chat`` from inside a turn: the chat closes once the turn ends.""" + if conv.kind != "deep": + msg = f"only deep chats close this way, {conv.external_id} is {conv.kind}" + raise ValueError(msg) + return await self.set_flags(conv, {"close_requested": True}) + + async def idle( + self, + *, + kind: str, + days: int, + since: datetime | None = None, + limit: int | None = None, + ) -> list[Conversation]: + """Open conversations of ``kind`` quiet for ``days``, oldest first. + + Only those with a session (something to fork) and, with ``since``, + with activity after it - the pile from before the system went live + is left alone (§4.5). + """ + now = datetime.now(UTC) + cutoff = now - timedelta(days=days) + out: list[tuple[datetime, Conversation]] = [] + for conv in await self.find(status="open", kind=kind, limit=10_000): + if conv.session_id is None: + continue + last = _aware(conv.last_activity_at or conv.created_at) + if last > cutoff or (since is not None and last < since): + continue + out.append((last, conv)) + out.sort(key=lambda pair: pair[0]) + rows = [conv for _, conv in out] + return rows[:limit] if limit is not None else rows + + async def distill( + self, conv: Conversation, *, reason: str = "api" + ) -> DistillResult: + """Close a deep chat (§8.4): fork it under the distiller, two channels. + + The digest is a file the fork writes under ``Distiller.dir`` - the + gateway checks it appeared with a valid frontmatter and puts a line + into the index; the merge is the fork's text, at most five lines, + injected into the open master. ``flags.memory=False`` skips the + file. The chat's own file is untouched; ``status=closed``. + """ + if self._distiller is None: + msg = "no distiller configured (Gateway(distiller=...))" + raise RuntimeError(msg) + if conv.kind != "deep": + msg = f"only deep chats are distilled, {conv.external_id} is {conv.kind}" + raise ValueError(msg) + row = await self.get_row(cast("int", conv.id)) or conv + if row.status != "open": + msg = f"conversation {row.external_id} is {row.status}" + raise ValueError(msg) + if await self.busy(row): + msg = f"conversation {row.external_id} is busy" + raise RuntimeError(msg) + memory = bool(row.flags.get("memory", True)) + chat_name = await self._chat_name(row) + ctx = DistillContext( + conversation=row, + title=await self.implied_title(row), + source=await self._window_of(row), + chat_name=chat_name, + memory=memory, + reason=reason, + day=datetime.now(UTC).astimezone().date(), + ) + prompt = await self._distill_prompt(ctx) + started = datetime.now(UTC) + self._bus.publish( + "distill.start", + conversation_id=row.external_id, + reason=reason, + memory=memory, + ) + result = await self.fork( + row, + prompt, + strip_tools=True, + agent=self._distiller.agent, + title=f"выжимка: {chat_name}", + ) + text, trimmed = trim_summary(result.text) + digest: Digest | None = None + error: str | None = None + if memory: + written = written_paths(result.capture.synthesized_messages) + path = find_digest(self._distiller, since=started, written=written) + if path is None: + error = "файл выжимки не появился" + else: + checked = check_digest(path, self._distiller) + if isinstance(checked, str): + error = f"{path.name}: {checked}" + else: + digest = checked + append_index(self._distiller, index_line(digest, chat_name)) + if error is not None: + _log.warning("distill of %s: %s", row.external_id, error) + master = await self._open_master() + if master is not None and text: + note = self._texts.closed.format( + chat=chat_name, + digest=f", выжимка [[{digest.path.stem}]]" if digest else "", + text=text, + ) + await self.inject(master, note, urgency="normal", origin="выжимка") + await self.close(row) + row = await self.set_flags( + row, + { + "close_requested": None, + "closed_reason": reason, + "digest": str(digest.path) if digest else None, + "digest_error": error, + }, + ) + self._bus.publish( + "conversation.distilled", + conversation_id=row.external_id, + fork=result.conversation.external_id, + reason=reason, + memory=memory, + digest=str(digest.path) if digest else None, + error=error, + text=text, + trimmed=trimmed, + master=master.external_id if master is not None else None, + ) + return DistillResult( + conversation=row, + fork=result.conversation, + text=text, + digest=digest, + error=error, + trimmed=trimmed, + ) + + async def _distill_prompt(self, ctx: DistillContext) -> str: + source = self._texts.distill + if source is None: + template = ( + _DEFAULT_DISTILL_PROMPT + if ctx.memory + else _DEFAULT_DISTILL_PROMPT_NO_MEMORY + ) + return template.format( + chat=ctx.chat_name, reason=ctx.reason, day=ctx.day.isoformat() + ) + produced: Any = source(ctx) + return await produced if inspect.isawaitable(produced) else produced + + async def _open_master(self) -> Conversation | None: + masters = await self.find(kind="master", status="open", limit=1) + return masters[0] if masters else None + + async def _window_of(self, conv: Conversation) -> str | None: + for binding in await self.bindings(conv): + if binding.visible: + return binding.external_id + return None + + async def _chat_name(self, conv: Conversation) -> str: + """What ``[[…]]`` to the chat says: the file's stem when it has one.""" + window = await self._window_of(conv) + if window and window.endswith(".md"): + return window.rsplit("/", 1)[-1][: -len(".md")] + return await self.implied_title(conv) or conv.external_id + + async def _close_after_turn(self, conv: Conversation) -> None: + for _ in range(_CLOSE_TRIES): + if await self.busy(conv): + await asyncio.sleep(_CLOSE_WAIT) + continue + try: + await self.distill(conv, reason="close_chat") + except RuntimeError as exc: + _log.info("closing %s: %s, retrying", conv.external_id, exc) + await asyncio.sleep(_CLOSE_WAIT) + continue + except Exception: # noqa: BLE001 + _log.exception("closing %s after its turn failed", conv.external_id) + return + _log.warning("closing %s: still busy, giving up", conv.external_id) + + # ---- turn hooks ------------------------------------------------------ + + async def _before_turn(self, conv: Conversation) -> str | None: + """Snapshot the capped file so a too-long rewrite can be bounced.""" + cap = LineCap.from_flags(conv.flags.get("line_cap")) + if cap is None: + return None + try: + return cap.path.read_text(encoding="utf-8") if cap.path.exists() else "" + except OSError: + _log.exception("line cap: cannot read %s", cap.path) + return None + + async def _after_turn(self, conv: Conversation, before: str | None) -> None: + row = await self.get_row(cast("int", conv.id)) + if row is None: + return + if row.kind == "deep" and row.flags.get("close_requested"): + task = asyncio.create_task(self._close_after_turn(row)) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + cap = LineCap.from_flags(row.flags.get("line_cap")) + if cap is not None and before is not None: + await self._enforce_cap(row, cap, before) + + async def _enforce_cap(self, conv: Conversation, cap: LineCap, before: str) -> None: + if not cap.path.exists(): + return + after = cap.path.read_text(encoding="utf-8") + lines = sum(1 for line in after.splitlines() if line.strip()) + if lines <= cap.max_lines: + return + if before: + cap.path.write_text(before, encoding="utf-8") + else: + cap.path.unlink() + attempts = int(conv.flags.get("line_cap_attempts", 0) or 0) + 1 + await self.set_flags(conv, {"line_cap_attempts": attempts}) + self._bus.publish( + "line_cap.bounced", + conversation_id=conv.external_id, + path=str(cap.path), + lines=lines, + max_lines=cap.max_lines, + attempt=attempts, + ) + _log.warning( + "line cap: %s came back with %d lines (cap %d), restored; attempt %d", + cap.path, + lines, + cap.max_lines, + attempts, + ) + if attempts > _CAP_TRIES: + return + await self.inject( + conv, + self._texts.too_long.format( + name=cap.path.name, lines=lines, max_lines=cap.max_lines + ), + urgency="urgent", + origin="потолок", + interrupt=False, + ) + async def reparent(self, conv: Conversation, parent: Conversation) -> Conversation: async def apply(row: Conversation) -> None: row.parent_id = parent.id @@ -1020,6 +1335,7 @@ class Conversations: runner.started_at = datetime.now(UTC) runner.tools = {} await self._mark_running(conv, turn_id) + before = await self._before_turn(conv) self._bus.publish( "turn.start", conversation_id=conv.external_id, @@ -1053,6 +1369,11 @@ class Conversations: finally: runner.turn_id = None await self._mark_done(conv, capture, cut=cut) + if stop != "error": + try: + await self._after_turn(conv, before) + except Exception: # noqa: BLE001 + _log.exception("after-turn hook on %s failed", conv.external_id) self._bus.publish( "turn.end", conversation_id=conv.external_id, @@ -1177,7 +1498,12 @@ class Conversations: } async def _copy_session( - self, conv: Conversation, *, window: int | None, strip_tools: bool + self, + conv: Conversation, + *, + window: int | None, + strip_tools: bool, + agent: str | None = None, ) -> str: if conv.session_id is None: msg = f"conversation {conv.external_id} has no session to copy" @@ -1186,21 +1512,26 @@ class Conversations: if live is not None and live.dirty: msg = f"conversation {conv.external_id} has a mirror gap; not forking" raise RuntimeError(msg) - agent = self._claude_agent(conv.agent_name) + source = self._claude_agent(conv.agent_name) + target = self._claude_agent(agent) if agent else source forked = await fork_session_via_store( - self._store, conv.session_id, directory=str(agent.cwd) + self._store, conv.session_id, directory=str(source.cwd) ) - if window is not None or strip_tools: - key = { - "project_key": project_key_for_directory(str(agent.cwd)), - "session_id": forked.session_id, - } - entries = await self._store.load(cast("Any", key)) or [] + source_key = { + "project_key": project_key_for_directory(str(source.cwd)), + "session_id": forked.session_id, + } + target_key = { + "project_key": project_key_for_directory(str(target.cwd)), + "session_id": forked.session_id, + } + if window is not None or strip_tools or target_key != source_key: + entries = await self._store.load(cast("Any", source_key)) or [] trimmed = window_entries(cast("Any", entries), window=window) if strip_tools: trimmed = strip_tool_entries(trimmed) - await self._store.delete(cast("Any", key)) - await self._store.append(cast("Any", key), cast("Any", trimmed)) + await self._store.delete(cast("Any", source_key)) + await self._store.append(cast("Any", target_key), cast("Any", trimmed)) _log.info( "forked session %s -> %s (window=%s, strip_tools=%s)", conv.session_id, diff --git a/src/beaver_gateway/core/distill.py b/src/beaver_gateway/core/distill.py new file mode 100644 index 0000000..8e19a6e --- /dev/null +++ b/src/beaver_gateway/core/distill.py @@ -0,0 +1,197 @@ +"""Closing a deep chat: the digest, the index, the file line cap (§6.4, §8.4). + +The gateway knows no path by itself (§0.8): ``Distiller`` from ``config.py`` +names the agent, says where digests land and where the index lives, and +the distiller writes the file on its own. What the gateway does is check that a file +with a valid frontmatter appeared under ``Distiller.dir`` during the fork +turn, put one line into the index, and cap the merge text at +``SUMMARY_LINES``. ``LineCap`` is the same idea for a file a job rewrites +(``состояние.md``): a result longer than the cap is bounced - the file +goes back to what it was and the job is told to shorten. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import frontmatter + +if TYPE_CHECKING: + from collections.abc import Iterable, Sequence + + from beaver_gateway.storage.models import Conversation + +__all__ = [ + "SUMMARY_LINES", + "Digest", + "DistillContext", + "Distiller", + "LineCap", + "append_index", + "check_digest", + "find_digest", + "index_line", + "trim_summary", + "written_paths", +] + +SUMMARY_LINES = 5 +"""A merge into the master is at most this many lines (§6.4).""" + + +@dataclass(frozen=True, slots=True) +class Distiller: + """Who closes deep chats, where digests land, where the index lives.""" + + agent: str + dir: Path + index: Path + type: str = "выжимка" + """Value the ``type`` frontmatter key must carry.""" + + index_header: str = "# индекс\n\nстрока на выжимку: чат → его выжимка.\n" # noqa: RUF001 + + +@dataclass(frozen=True, slots=True) +class DistillContext: + """What the setup's distill prompt is built from.""" + + conversation: Conversation + title: str | None + source: str | None + """The window the chat lives in (a vault-relative path for markdown).""" + + chat_name: str + """What a ``[[wikilink]]`` to the chat is called.""" + + memory: bool + reason: str + day: date + + +@dataclass(frozen=True, slots=True) +class Digest: + path: Path + source: str + date: date + + +@dataclass(frozen=True, slots=True) +class LineCap: + """A file a job rewrites may not exceed ``max_lines`` after its turn.""" + + path: Path + max_lines: int + + def as_flags(self) -> dict[str, Any]: + return {"path": str(self.path), "max_lines": self.max_lines} + + @classmethod + def from_flags(cls, value: Any) -> LineCap | None: + if not isinstance(value, dict): + return None + path, max_lines = value.get("path"), value.get("max_lines") + if not isinstance(path, str) or not isinstance(max_lines, int): + return None + return cls(path=Path(path), max_lines=max_lines) + + +def written_paths(messages: Iterable[dict[str, Any]]) -> list[Path]: + """Paths the turn's ``Write``/``Edit`` calls targeted, in order.""" + out: list[Path] = [] + for message in messages: + content = message.get("content") + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_use": + continue + if block.get("name") not in ("Write", "Edit", "MultiEdit"): + continue + target = (block.get("input") or {}).get("file_path") + if isinstance(target, str) and target: + out.append(Path(target)) + return out + + +def find_digest( + digests: Distiller, *, since: datetime, written: Sequence[Path] = () +) -> Path | None: + """The digest the fork turn produced. + + A written path under ``dir`` first, otherwise the newest file there + modified since ``since``. + """ + root = digests.dir.resolve() + for path in reversed(list(written)): + candidate = path if path.is_absolute() else digests.dir / path + try: + inside = candidate.resolve().relative_to(root) + except (OSError, ValueError): + continue + if (digests.dir / inside).is_file(): + return digests.dir / inside + if not digests.dir.is_dir(): + return None + stamp = since.timestamp() - 1 + fresh = [ + p + for p in digests.dir.glob("*.md") + if p.is_file() and p.stat().st_mtime >= stamp + ] + return max(fresh, key=lambda p: p.stat().st_mtime) if fresh else None + + +def check_digest(path: Path, digests: Distiller) -> Digest | str: + """The digest's frontmatter parsed, or why it is not a digest.""" + try: + post = frontmatter.load(str(path)) + except (OSError, ValueError) as exc: + return f"не читается: {exc}" + meta = post.metadata + if meta.get("type") != digests.type: + return f"`type` должен быть `{digests.type}`, не {meta.get('type')!r}" + source = meta.get("source") + if not isinstance(source, str) or not source.strip(): + return "`source` пустой" + when = meta.get("date") + if isinstance(when, datetime): + when = when.date() + elif isinstance(when, str): + try: + when = date.fromisoformat(when.strip()) + except ValueError: + return f"`date` не дата: {when!r}" + if not isinstance(when, date): + return "`date` отсутствует" + if not post.content.strip(): + return "тело пустое" + return Digest(path=path, source=source.strip(), date=when) + + +def index_line(digest: Digest, chat_name: str) -> str: + return f"- {digest.date.isoformat()} [[{chat_name}]] → [[{digest.path.stem}]]" + + +def append_index(digests: Distiller, line: str) -> None: + index = digests.index + text = index.read_text(encoding="utf-8") if index.exists() else "" + if line in text.splitlines(): + return + if not text: + text = digests.index_header + if not text.endswith("\n"): + text += "\n" + index.parent.mkdir(parents=True, exist_ok=True) + index.write_text(text + line + "\n", encoding="utf-8") + + +def trim_summary(text: str, limit: int = SUMMARY_LINES) -> tuple[str, bool]: + """The merge text cut to ``limit`` non-empty lines; ``True`` if it was.""" + lines = [line.rstrip() for line in text.strip().splitlines() if line.strip()] + if len(lines) <= limit: + return "\n".join(lines), False + return "\n".join(lines[:limit]), True diff --git a/src/beaver_gateway/core/gateway_tools.py b/src/beaver_gateway/core/gateway_tools.py index ed76f86..da52fd6 100644 --- a/src/beaver_gateway/core/gateway_tools.py +++ b/src/beaver_gateway/core/gateway_tools.py @@ -26,7 +26,7 @@ __all__ = ["SERVER_NAME", "TOOL_NAMES", "build_tool_server"] _log = logging.getLogger("beaver_gateway.core.gateway_tools") SERVER_NAME = "gateway" -TOOL_NAMES = ("read_conversation", "spawn", "say", "schedule", "inject") +TOOL_NAMES = ("read_conversation", "spawn", "say", "schedule", "inject", "close_chat") def build_tool_server( @@ -171,7 +171,23 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]: ) return _text(f"queued #{item.id}") - return [read_conversation, spawn, say, schedule, inject] + @tool( + "close_chat", + "Close this deep chat once the current reply is finished: the " + "distiller forks it, writes the digest and hands a short merge to the " + "master. Call it when the human says the discussion is over; the chat " + "file stays as it is.", + {"type": "object", "properties": {}}, + ) + async def close_chat(_args: dict[str, Any]) -> dict[str, Any]: + conv = await current() + try: + await conversations.request_close(conv) + except ValueError as exc: + return _error(str(exc)) + return _text("ok: the chat closes after this reply") + + return [read_conversation, spawn, say, schedule, inject, close_chat] def _text(text: str) -> dict[str, Any]: diff --git a/src/beaver_gateway/core/registry.py b/src/beaver_gateway/core/registry.py index acdeedc..f9b3361 100644 --- a/src/beaver_gateway/core/registry.py +++ b/src/beaver_gateway/core/registry.py @@ -17,6 +17,7 @@ if TYPE_CHECKING: from beaver_gateway.agents.base import BaseAgent from beaver_gateway.core.conversations import ConversationTexts + from beaver_gateway.core.distill import Distiller from beaver_gateway.core.rotation import RotationPolicy from beaver_gateway.core.scheduler import Budget, Job from beaver_gateway.core.watch import VaultWatch @@ -95,6 +96,8 @@ class Gateway: """Vault watcher feeding the envelope (§3.5, §4.6); ``None`` = no vault block.""" budget: Budget | None = None """Subscription window past which non-critical jobs wait (§4.5).""" + distiller: Distiller | None = None + """Who closes deep chats and where the digests and the index live (§8.4).""" tz: str = "UTC" """Local zone for the envelope clock and the rotation hour.""" host: str = "0.0.0.0" # noqa: S104 diff --git a/src/beaver_gateway/core/scheduler.py b/src/beaver_gateway/core/scheduler.py index 8627b16..4a23765 100644 --- a/src/beaver_gateway/core/scheduler.py +++ b/src/beaver_gateway/core/scheduler.py @@ -36,7 +36,8 @@ if TYPE_CHECKING: from pgqueuer.ports.driver import Driver from starlette.requests import Request - from beaver_gateway.core.conversations import Conversations + from beaver_gateway.core.conversations import Conversations, DistillResult + from beaver_gateway.core.distill import LineCap from beaver_gateway.core.injects import Priority from beaver_gateway.core.rotation import Rotation from beaver_gateway.storage.models import Conversation @@ -98,12 +99,45 @@ class JobRun: return True async def spawn_job( - self, *, agent: str, text: str, title: str | None = None + self, + *, + agent: str, + text: str, + title: str | None = None, + line_cap: LineCap | None = None, ) -> Conversation: + """A headless job turn; ``line_cap`` bounces a rewrite past the cap.""" return await self.conversations.spawn( - kind="job", agent=agent, seed="brief", text=text, title=title, origin="job" + kind="job", + agent=agent, + seed="brief", + text=text, + title=title, + origin="job", + flags={"line_cap": line_cap.as_flags()} if line_cap else None, ) + async def close_idle( + self, + *, + kind: str = "deep", + days: int = 2, + limit: int = 3, + since: datetime | None = None, + ) -> list[DistillResult]: + """§4.5: close chats quiet for ``days``, at most ``limit`` per run.""" + out: list[DistillResult] = [] + for conv in await self.conversations.idle( + kind=kind, days=days, since=since, limit=limit + ): + try: + out.append( + await self.conversations.distill(conv, reason=f"idle {days}d") + ) + except Exception: # noqa: BLE001 + _log.exception("closing idle %s failed", conv.external_id) + return out + async def retry_in(self, delay: timedelta) -> None: await self.scheduler.trigger( self.job, self.payload, delay=delay, trigger=self.trigger diff --git a/src/beaver_gateway/frontends/api/frontend.py b/src/beaver_gateway/frontends/api/frontend.py index adf4f35..43dedbf 100644 --- a/src/beaver_gateway/frontends/api/frontend.py +++ b/src/beaver_gateway/frontends/api/frontend.py @@ -441,6 +441,36 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa "text": result.text, } + @app.post("/conversations/{public_id}/close") + async def post_close(public_id: str, request: Request) -> dict[str, Any]: + token = await require_token(request, runtime, scope=SCOPE) + conv = await conv_of(public_id) + if conv.kind != "deep": + closed = await conversations.close(conv) + return {"id": closed.external_id, "status": closed.status} + try: + result = await conversations.distill(conv, reason="api") + except (ValueError, LookupError) as exc: + raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc + except RuntimeError as exc: + raise HTTPException(status.HTTP_409_CONFLICT, str(exc)) from exc + await audit.log( + runtime, + actor=f"token:{token}", + kind="api_close", + agent_name=conv.agent_name, + conversation=conv.external_id, + digest=str(result.digest.path) if result.digest else None, + ) + return { + "id": conv.external_id, + "status": "closed", + "fork": result.fork.external_id, + "text": result.text, + "digest": str(result.digest.path) if result.digest else None, + "error": result.error, + } + @app.post("/conversations/{public_id}/fork") async def post_fork(public_id: str, request: Request) -> dict[str, Any]: await require_token(request, runtime, scope=SCOPE) diff --git a/tests/test_api.py b/tests/test_api.py index cf8ddbb..4036562 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -420,3 +420,31 @@ async def test_pre_sdk_rows_show_canonical_history_and_a_title(world: World) -> 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"] + + +async def test_close_distills_a_deep_chat(world: World) -> None: + from test_distill import DistillerClient, deep_chat, distiller + + config = distiller(world, DistillerClient) + api = Api(world) + master = await world.conversations.create(kind="master", agent="a", origin="test") + chat = await deep_chat(world) + res = await api.http.post( + f"/conversations/{chat.external_id}/close", headers=HEADERS + ) + assert res.status_code == 200, res.text + body = res.json() + assert body["status"] == "closed" and body["error"] is None + assert body["digest"] == str(config.dir / "2026-08-29 - тема.md") + assert body["text"].count("\n") == 2 + assert (await world.conversations.get(chat.external_id)).status == "closed" + assert (await world.conversations.queue.recent(master.id))[0].origin == "выжимка" + again = await api.http.post( + f"/conversations/{chat.external_id}/close", headers=HEADERS + ) + assert again.status_code == 400 + branch = await world.conversations.create(kind="branch", agent="a", origin="test") + plain = await api.http.post( + f"/conversations/{branch.external_id}/close", headers=HEADERS + ) + assert plain.status_code == 200 and plain.json()["status"] == "closed" diff --git a/tests/test_distill.py b/tests/test_distill.py new file mode 100644 index 0000000..9c6deea --- /dev/null +++ b/tests/test_distill.py @@ -0,0 +1,419 @@ +import asyncio +import uuid +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from claude_agent_sdk import ( + AssistantMessage, + ResultMessage, + TextBlock, + ToolResultBlock, + ToolUseBlock, + UserMessage, + project_key_for_directory, +) +from test_conversations import ScriptedClient, World, world + +from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions +from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend +from beaver_gateway.core.conversations import ConversationTexts +from beaver_gateway.core.distill import ( + Distiller, + DistillContext, + LineCap, + check_digest, + trim_summary, +) +from beaver_gateway.core.gateway_tools import _tools, build_tool_server +from beaver_gateway.core.registry import AgentRegistry +from beaver_gateway.core.scheduler import Job, JobRun, Scheduler +from beaver_gateway.core.transcript import build_entries +from beaver_gateway.storage.models import Conversation + +__all__ = ["world"] + +DIGEST = """--- +type: выжимка +source: "[[{chat}]]" +date: 2026-08-29 +--- +# тема + +## контекст +зачем открывали + +## решили +- одно +""" + + +class DistillerClient(ScriptedClient): + """The distiller: writes a digest file when told to, answers in N lines.""" + + digest_dir: Path | None = None + lines = 3 + write = True + frontmatter = DIGEST + + async def receive_response(self): + prompt = self.prompts[-1] + written: list[ToolUseBlock] = [] + if self.write and self.digest_dir is not None and "файл не пиши" not in prompt: + chat = prompt.split("«", 1)[1].split("»", 1)[0] if "«" in prompt else "чат" + path = self.digest_dir / "2026-08-29 - тема.md" + path.write_text(self.frontmatter.format(chat=chat), encoding="utf-8") + written.append( + ToolUseBlock(id="w1", name="Write", input={"file_path": str(path)}) + ) + for block in written: + yield AssistantMessage(content=[block], model="m") + yield UserMessage(content=[ToolResultBlock(tool_use_id=block.id)]) + text = "\n".join(f"строка {i} про {prompt[:20]!r}" for i in range(self.lines)) + yield AssistantMessage(content=[TextBlock(text=text)], model="m") + yield ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=False, + num_turns=1, + session_id=self.session_id, + stop_reason="end_turn", + total_cost_usd=0.0, + usage={"input_tokens": 1, "output_tokens": 1}, + ) + + +class ClosingClient(ScriptedClient): + """A deep chat that calls ``close_chat`` inside its reply.""" + + conversations: Any = None + key = "" + + async def receive_response(self): + prompt = self.prompts[-1] + if "обсудили" in prompt: + assert "gateway" in self.options.mcp_servers + tools = _tools(ClosingClient.conversations, ClosingClient.key) + close = next(t for t in tools if t.name == "close_chat") + result = await close.handler({}) + assert "closes after this reply" in result["content"][0]["text"] + yield AssistantMessage( + content=[ + ToolUseBlock(id="c1", name="mcp__gateway__close_chat", input={}) + ], + model="m", + ) + yield AssistantMessage(content=[TextBlock(text=f"ok:{prompt}")], model="m") + yield ResultMessage( + subtype="success", + duration_ms=1, + duration_api_ms=1, + is_error=False, + num_turns=1, + session_id=self.session_id, + stop_reason="end_turn", + total_cost_usd=0.0, + usage={"input_tokens": 1, "output_tokens": 1}, + ) + + +class CapClient(ScriptedClient): + """A job that rewrites the capped file with ``lines`` lines.""" + + target: Path | None = None + lines = 70 + + async def receive_response(self): + assert self.target is not None + self.target.write_text( + "\n".join(f"- строка {i}" for i in range(self.lines)), encoding="utf-8" + ) + CapClient.lines = 10 + async for event in super().receive_response(): + yield event + + +def distiller(world: World, client: type[ScriptedClient]) -> Distiller: + vault = world.root / "vault" + digests = vault / "мета" / "бобер" / "выжимки" + digests.mkdir(parents=True) + agent = ClaudeAgent( + name="x", + model="m", + system_prompt="distill", + cwd=vault, + kinds=("fork", "job"), + options=ClaudeOptions( + effort="medium", tools=("Read", "Write"), include_partial_messages=False + ), + ) + backend = ClaudeSdkBackend( + agent=agent, + mcp_internal_urls={}, + session_store=world.store, + client_factory=client, + work_dir=world.root / "work", + pool=world.pool, + ) + world.conversations._agents = AgentRegistry( # noqa: SLF001 + [world.agent, world.deep_agent, agent] + ) + world.conversations._backends["x"] = backend # noqa: SLF001 + config = Distiller(agent="x", dir=digests, index=digests.parent / "индекс.md") + world.conversations._distiller = config # noqa: SLF001 + DistillerClient.digest_dir = digests + DistillerClient.lines = 3 + DistillerClient.write = True + DistillerClient.frontmatter = DIGEST + return config + + +async def deep_chat(world: World, name: str = "2026-08-20 - тема чата") -> Conversation: + sid = str(uuid.uuid4()) + await world.store.append( + world.key(sid), + build_entries( + [ + {"role": "user", "content": "q"}, + { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "t1", "name": "Read", "input": {}} + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "t1", "content": "x"} + ], + }, + {"role": "assistant", "content": "a"}, + ], + session_id=sid, + cwd=str(world.root), + model="m", + ), + ) + conv = await world.conversations.create( + kind="deep", agent="d", origin="markdown", session_id=sid + ) + await world.conversations.bind( + conv, frontend="markdown", external_id=f"2026-08/{name}.md" + ) + return conv + + +async def test_distill_writes_the_digest_indexes_it_and_merges_short( + world: World, +) -> None: + config = distiller(world, DistillerClient) + prompts: list[DistillContext] = [] + + def distill_prompt(ctx: DistillContext) -> str: + prompts.append(ctx) + return f"Чат «{ctx.chat_name}» закрыт ({ctx.reason})." + + world.conversations._texts = ConversationTexts(distill=distill_prompt) # noqa: SLF001 + master = await world.conversations.create(kind="master", agent="a", origin="test") + chat = await deep_chat(world) + + result = await world.conversations.distill(chat, reason="api") + + assert prompts[0].chat_name == "2026-08-20 - тема чата" + assert prompts[0].memory is True + assert result.digest is not None and result.error is None + assert result.digest.path == config.dir / "2026-08-29 - тема.md" + assert result.digest.source == "[[2026-08-20 - тема чата]]" + index = config.index.read_text(encoding="utf-8") + assert index.startswith("# индекс") + assert "- 2026-08-29 [[2026-08-20 - тема чата]] → [[2026-08-29 - тема]]" in index + assert result.text.count("\n") == 2 and not result.trimmed + closed = await world.conversations.get(chat.external_id) + assert closed.status == "closed" + assert closed.flags["digest"] == str(result.digest.path) + assert closed.flags["closed_reason"] == "api" + fork = await world.conversations.get(result.fork.external_id) + assert fork.kind == "fork" and fork.agent_name == "x" and fork.status == "closed" + items = await world.conversations.queue.recent(master.id) + assert items[0].origin == "выжимка" + assert items[0].text.startswith( + "Закрыт глубокий чат [[2026-08-20 - тема чата]], выжимка [[2026-08-29 - тема]]." + ) + assert items[0].text.endswith(result.text) + forked = ScriptedClient.instances[-1] + assert forked.options.resume is not None + vault_key = { + "project_key": project_key_for_directory(str(config.dir.parents[2])), + "session_id": forked.options.resume, + } + entries = await world.store.load(vault_key) + assert entries + assert all( + not isinstance(e.get("message", {}).get("content"), list) + or all(b.get("type") == "text" for b in e["message"]["content"]) + for e in entries + ) + + +async def test_memory_off_merges_without_a_file(world: World) -> None: + config = distiller(world, DistillerClient) + master = await world.conversations.create(kind="master", agent="a", origin="test") + chat = await deep_chat(world) + await world.conversations.set_flags(chat, {"memory": False}) + + result = await world.conversations.distill(chat, reason="api") + + assert result.digest is None and result.error is None + assert list(config.dir.iterdir()) == [] + assert not config.index.exists() + assert (await world.conversations.get(chat.external_id)).status == "closed" + items = await world.conversations.queue.recent(master.id) + assert len(items) == 1 and items[0].text.endswith(result.text) + assert ", выжимка" not in items[0].text + assert "файл не пиши" in ScriptedClient.instances[-1].prompts[0] + + +async def test_bad_frontmatter_and_long_merge_are_reported(world: World) -> None: + distiller(world, DistillerClient) + DistillerClient.frontmatter = "---\ntype: заметка\n---\n# тема\n" + DistillerClient.lines = 9 + await world.conversations.create(kind="master", agent="a", origin="test") + chat = await deep_chat(world) + + result = await world.conversations.distill(chat, reason="idle 2d") + + assert result.digest is None + assert result.error is not None and "`type`" in result.error + assert result.trimmed and result.text.count("\n") == 4 + closed = await world.conversations.get(chat.external_id) + assert closed.status == "closed" + assert closed.flags["digest_error"] == result.error + + +def test_check_digest_rejects_what_is_not_a_digest(tmp_path: Path) -> None: + config = Distiller(agent="x", dir=tmp_path, index=tmp_path / "i.md") + path = tmp_path / "d.md" + path.write_text("---\ntype: выжимка\nsource: ''\ndate: 2026-08-29\n---\nx\n") + assert check_digest(path, config) == "`source` пустой" + path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: вчера\n---\nx\n") + assert "`date`" in check_digest(path, config) + path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: 2026-08-29\n---\n\n") + assert check_digest(path, config) == "тело пустое" + path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: 2026-08-29\n---\nx\n") + digest = check_digest(path, config) + assert not isinstance(digest, str) and digest.date.isoformat() == "2026-08-29" + assert trim_summary("a\n\nb\nc\nd\ne\nf") == ("a\nb\nc\nd\ne", True) + + +async def test_close_chat_tool_closes_after_the_reply(world: World) -> None: + distiller(world, DistillerClient) + world.deep_agent = ClaudeAgent( + name="d", + model="m", + system_prompt="deep", + cwd=world.root, + gateway_tools=("close_chat",), + ) + world.conversations._agents = AgentRegistry( # noqa: SLF001 + [world.agent, world.deep_agent, world.conversations._agents.get("x")] # noqa: SLF001 + ) + world.conversations._backends["d"] = ClaudeSdkBackend( # noqa: SLF001 + agent=world.deep_agent, + mcp_internal_urls={}, + session_store=world.store, + client_factory=ClosingClient, + work_dir=world.root / "work", + pool=world.pool, + tool_server=lambda key, _kind: build_tool_server( + world.conversations, conversation_key=key, names=("close_chat",) + ), + ) + master = await world.conversations.create(kind="master", agent="a", origin="test") + chat = await deep_chat(world) + ClosingClient.conversations = world.conversations + ClosingClient.key = chat.external_id + + await world.conversations.post(chat, "ок, обсудили") + await world.settle(chat, 1) + for _ in range(100): + row = await world.conversations.get(chat.external_id) + if row.status == "closed": + break + await asyncio.sleep(0.05) + else: + raise AssertionError("chat did not close") + + assert row.flags["closed_reason"] == "close_chat" + assert row.flags["digest"] is not None + items = await world.conversations.queue.recent(master.id) + assert items[0].origin == "выжимка" + + +async def test_idle_picks_quiet_chats_after_launch_at_most_limit(world: World) -> None: + distiller(world, DistillerClient) + now = datetime.now(UTC) + launch = now - timedelta(days=10) + + async def aged(name: str, days: float) -> Conversation: + conv = await deep_chat(world, name) + + async def apply(row: Conversation) -> None: + row.last_activity_at = now - timedelta(days=days) + + return await world.conversations._update(conv, apply) # noqa: SLF001 + + old = await aged("old", 30) + a = await aged("a", 5) + b = await aged("b", 4) + c = await aged("c", 3) + d = await aged("d", 2.5) + fresh = await aged("fresh", 1) + no_session = await world.conversations.create(kind="deep", agent="d", origin="t") + + idle = await world.conversations.idle(kind="deep", days=2, since=launch) + assert [x.external_id for x in idle] == [ + a.external_id, + b.external_id, + c.external_id, + d.external_id, + ] + assert old.external_id not in {x.external_id for x in idle} + assert fresh.external_id not in {x.external_id for x in idle} + assert no_session.external_id not in {x.external_id for x in idle} + + scheduler = Scheduler(conversations=world.conversations) + run = JobRun(Job("закрытие", lambda _run: asyncio.sleep(0)), "cron", {}, scheduler) + closed = await run.close_idle(kind="deep", days=2, limit=3, since=launch) + assert [r.conversation.external_id for r in closed] == [ + a.external_id, + b.external_id, + c.external_id, + ] + assert (await world.conversations.get(d.external_id)).status == "open" + assert len(await world.conversations.idle(kind="deep", days=2, since=launch)) == 1 + + +async def test_line_cap_bounces_a_long_rewrite_and_asks_to_shorten( + world: World, +) -> None: + distiller(world, CapClient) + state = world.root / "vault" / "состояние.md" + state.write_text("# состояние\n- было так\n", encoding="utf-8") + CapClient.target = state + CapClient.lines = 70 + scheduler = Scheduler(conversations=world.conversations) + run = JobRun(Job("память", lambda _run: asyncio.sleep(0)), "cron", {}, scheduler) + + job = await run.spawn_job( + agent="x", text="перепиши", line_cap=LineCap(state, max_lines=60) + ) + await world.settle(job, 2) + + client = ScriptedClient.instances[-1] + assert len(client.prompts) == 2 + assert "70 строк при потолке 60" in client.prompts[1] + assert "[инжект: потолок" in client.prompts[1] + assert state.read_text(encoding="utf-8").count("\n") == 10 - 1 + row = await world.conversations.get(job.external_id) + assert row.flags["line_cap_attempts"] == 1 diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index bbdc7ab..40053bd 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -234,6 +234,17 @@ export class ApiClient { return this.post(`/api/conversations/${id}/merge`); } + close(id: string): Promise<{ + id: string; + status: string; + fork?: string; + text?: string; + digest?: string | null; + error?: string | null; + }> { + return this.post(`/api/conversations/${id}/close`); + } + bind( id: string, frontend: string, diff --git a/ui/src/lib/panel/conversation-header.svelte b/ui/src/lib/panel/conversation-header.svelte index 7c74849..ab2c5c3 100644 --- a/ui/src/lib/panel/conversation-header.svelte +++ b/ui/src/lib/panel/conversation-header.svelte @@ -87,6 +87,15 @@ } } + async function distill() { + const result = await client.close(info.id); + if (result.error) { + toast.warning(`Digest: ${result.error}`); + } else if (result.digest) { + toast.message(`Digest: ${result.digest}`); + } + } + function copyId() { navigator.clipboard .writeText(info.id) @@ -153,6 +162,11 @@ Merge into parent {/if} + {#if info.kind === "deep" && info.status === "open"} + run(distill, "Chat closed")}> + Close with digest + + {/if} {#if info.status === "open"}