"""Conversations, threads and the one place a turn runs (§3.1, §3.4, §8.2-8.3). The service owns the rows (:class:`Conversation`, :class:`ConversationBinding`), the per-conversation queue workers, the fork/spawn/read/inject/say/schedule API that frontends, jobs and the in-process MCP tools all call, and the restart recovery: a conversation with ``running_turn`` set at startup gets its open ``tool_use`` closed in the transcript and an inject saying the turn was cut. Routing is deterministic (§0.1): a turn started by a user message streams back to whoever asked; a turn started by an inject streams nowhere - its events still hit the bus flagged ``origin="inject"`` so the panel can show activity, but the only way it speaks is ``say``. """ from __future__ import annotations import asyncio import contextlib import inspect import json import logging import re import uuid from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta, tzinfo from typing import TYPE_CHECKING, Any, cast from claude_agent_sdk import ( AssistantMessage, RateLimitEvent, ResultMessage, StreamEvent, ToolResultBlock, ToolUseBlock, UserMessage, fork_session_via_store, project_key_for_directory, ) from sqlmodel import col, select from beaver_gateway.core import injects 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 from beaver_gateway.core.kinds import KINDS, Kind, as_kind from beaver_gateway.core.transcript import ( messages_from_entries, render_messages, strip_tool_entries, text_of, window_entries, ) from beaver_gateway.core.turn_capture import TurnCapture from beaver_gateway.frontends._accumulate import StreamAccumulator from beaver_gateway.storage.models import ( Conversation, ConversationBinding, ConversationMessage, InjectQueueItem, RateLimit, Usage, ) if TYPE_CHECKING: from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence from claude_agent_sdk import SessionStore from beaver_gateway.agents.claude import ClaudeAgent from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend from beaver_gateway.core.bus import EventBus from beaver_gateway.core.envelope import Envelope from beaver_gateway.core.events import MessageStreamEvent from beaver_gateway.core.injects import Priority from beaver_gateway.core.registry import AgentRegistry from beaver_gateway.core.rotation import HandoutContext from beaver_gateway.core.scheduler import Scheduler from beaver_gateway.core.sessions import SessionPool from beaver_gateway.frontends.base import Frontend from beaver_gateway.storage.db import Database __all__ = [ "KINDS", "SEEDS", "ConversationTexts", "Conversations", "DistillResult", "ForkResult", "SeedContext", "UserSaid", ] _log = logging.getLogger("beaver_gateway.core.conversations") MASTER_ALIAS = "master" PARENT_ALIAS = "parent" SEEDS = ("clean", "morning", "copy", "brief") _STATUSES = ("open", "merged", "closed", "archived") _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) class SeedContext: kind: Kind seed: str agent: str parent: Conversation | None text: str | None title: str | None @dataclass(frozen=True, slots=True) class NewDayContext: """What the new master hears first. ``reason`` is ``ночь`` / ``возраст`` / ``транскрипт`` - only the first one is actually a new day. """ day: date reason: str moved: int @dataclass(frozen=True, slots=True) class UserSaid: """One message from the user as it enters a turn (``origin=user``).""" conversation_id: str kind: str title: str | None text: str at: datetime @dataclass(frozen=True, slots=True) class ConversationTexts: """Texts the gateway cannot invent for a setup. What a merge asks for and what a seed says; ``seed`` may return the body for any seed mode (the morning handout lives in the vault the gateway knows nothing about). """ merge_prompt: str = _DEFAULT_MERGE_PROMPT inject_header: Callable[[injects.InjectContext], str] = injects.inject_header """Framing line(s) above each inject: who it is from, whether it cut a turn.""" interrupted: str = "прервано" answered: str = "Пользователь ответил: {answer}" unanswered: str = ( "Пользователь не ответил за {minutes} мин. Вопрос ему показан текстом; " "заверши тёрн сейчас, ответ придёт следующим сообщением." ) seed: Callable[[SeedContext], Awaitable[str | None] | str | None] | None = None handout: Callable[[HandoutContext], Awaitable[str] | str] | str = ( "Этот мастер закрывается ({reason}). Напиши хендаут за {day}: справку " "на утро, не задание - прошедшее время, без повелительного наклонения." ) new_day: Callable[[NewDayContext], Awaitable[str] | str] | str = ( "Мастер сменился ({reason}), хендаут за {day} записан." ) """First inject of the new master; a callable sees the rotation reason.""" 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) class ForkResult: conversation: Conversation text: str 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) wake: asyncio.Event = field(default_factory=asyncio.Event) task: asyncio.Task[None] | None = None turn_id: str | None = None origin: str | None = None text: str | None = None started_at: datetime | None = None tools: dict[str, dict[str, Any]] = field(default_factory=dict) """Tool calls of the running turn, in order; ``describe`` hands them to a panel that subscribed mid-turn.""" def snapshot(self) -> dict[str, Any] | None: if self.turn_id is None: return None return { "id": self.turn_id, "origin": self.origin, "text": self.text, "started_at": _iso(self.started_at), "tools": list(self.tools.values()), } @dataclass(frozen=True, slots=True) class _Question: conversation_id: str turn_id: str | None questions: list[dict[str, Any]] answer: asyncio.Future[str] class Conversations: def __init__( self, *, db: Database, agents: AgentRegistry, backends: dict[str, Any], bus: EventBus, 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, question_timeout: float = 600.0, envelope: Envelope | None = None, distiller: Distiller | None = None, user_sink: Callable[[UserSaid], Awaitable[None] | None] | None = None, ) -> None: self._db = db self._user_sink = user_sink self._distiller = distiller self._agents = agents self._backends = backends self._bus = bus 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 self._question_timeout = question_timeout self._envelope = envelope self.scheduler: Scheduler | None = None self._questions: dict[str, _Question] = {} self._queue = InjectQueue(db) self._runners: dict[int, _Runner] = {} self._tasks: set[asyncio.Task[None]] = set() self._idle_task: asyncio.Task[None] | None = None @property def db(self) -> Database: return self._db @property def queue(self) -> InjectQueue: return self._queue @property def bus(self) -> EventBus: return self._bus @property def pool(self) -> SessionPool: return self._pool # ---- rows ---------------------------------------------------------- async def create( self, *, kind: Kind, agent: str, parent: Conversation | None = None, title: str | None = None, origin: str = "api", session_id: str | None = None, flags: dict[str, Any] | None = None, ) -> Conversation: if kind not in KINDS: msg = f"unknown conversation kind {kind!r}" raise ValueError(msg) 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, external_id=str(uuid.uuid4()), agent_name=agent, kind=kind, parent_id=parent.id if parent is not None else None, title=title, session_id=session_id, flags=dict(flags or {}), last_activity_at=now, ) async with self._db.session() as session: session.add(row) await session.commit() await session.refresh(row) self._bus.publish("conversation.created", **self.public(row)) return row async def get(self, public_id: str) -> Conversation | None: async with self._db.session() as session: result = await session.exec( select(Conversation).where(Conversation.external_id == public_id) ) return result.first() async def get_row(self, row_id: int) -> Conversation | None: async with self._db.session() as session: return await session.get(Conversation, row_id) async def resolve( self, key: str, *, origin: Conversation | None = None ) -> Conversation | None: """A conversation by public id or alias. ``master`` is the open master, ``parent`` the parent of ``origin`` - the names a job or a branch can use without knowing today's ids. """ key = key.strip() if key == MASTER_ALIAS: return await self._open_master() if key == PARENT_ALIAS: if origin is None or origin.parent_id is None: return None return await self.get_row(origin.parent_id) return await self.get(key) async def find( self, *, status: str | None = None, kind: str | None = None, parent: Conversation | None = None, limit: int = 200, ) -> list[Conversation]: stmt = select(Conversation).order_by(col(Conversation.id).desc()).limit(limit) if status is not None: stmt = stmt.where(Conversation.status == status) if kind is not None: stmt = stmt.where(Conversation.kind == kind) if parent is not None: stmt = stmt.where(Conversation.parent_id == parent.id) async with self._db.session() as session: return list((await session.exec(stmt)).all()) async def bindings(self, conv: Conversation) -> list[ConversationBinding]: async with self._db.session() as session: result = await session.exec( select(ConversationBinding) .where(ConversationBinding.conversation_id == conv.id) .order_by(col(ConversationBinding.id)) ) return list(result.all()) async def bind( self, conv: Conversation, *, frontend: str, 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( ( await session.exec( select(ConversationBinding).where( ConversationBinding.conversation_id == conv.id, ConversationBinding.frontend == frontend, ) ) ).all() ) row = next((b for b in existing if b.external_id == external_id), None) if visible: for other in existing: if other is not row and other.visible: other.visible = False session.add(other) same_window = await session.exec( select(ConversationBinding).where( ConversationBinding.frontend == frontend, ConversationBinding.external_id == external_id, ConversationBinding.conversation_id != conv.id, col(ConversationBinding.visible).is_(True), ) ) for other in same_window.all(): other.visible = False session.add(other) if row is None: row = ConversationBinding( conversation_id=cast("int", conv.id), frontend=frontend, external_id=external_id, visible=visible, ) else: row.visible = visible session.add(row) await session.commit() await session.refresh(row) self._bus.publish( "conversation.bound", conversation_id=conv.external_id, frontend=frontend, external_id=external_id, visible=visible, ) return row async def find_bound( self, *, frontend: str, external_id: str ) -> Conversation | None: async with self._db.session() as session: result = await session.exec( select(Conversation) .join( ConversationBinding, col(ConversationBinding.conversation_id) == col(Conversation.id), ) .where( ConversationBinding.frontend == frontend, ConversationBinding.external_id == external_id, col(ConversationBinding.visible).is_(True), ) .order_by(col(Conversation.id).desc()) ) return result.first() async def last_binding( self, *, frontend: str, kind: str ) -> ConversationBinding | None: """The window ``frontend`` last used for a conversation of ``kind``. A frontend whose window for the master outlives the master itself (the Telegram General topic) finds it here after a rotation. """ async with self._db.session() as session: result = await session.exec( select(ConversationBinding) .join( Conversation, col(Conversation.id) == col(ConversationBinding.conversation_id), ) .where( ConversationBinding.frontend == frontend, Conversation.kind == kind ) .order_by(col(ConversationBinding.id).desc()) ) return result.first() async def set_flags( self, conv: Conversation, flags: dict[str, Any] ) -> Conversation: async def apply(row: Conversation) -> None: row.flags = {**row.flags, **flags} return await self._update(conv, apply) async def set_status(self, conv: Conversation, status: str) -> Conversation: if status not in _STATUSES: msg = f"unknown status {status!r}" raise ValueError(msg) async def apply(row: Conversation) -> None: row.status = status return await self._update(conv, apply) async def set_title(self, conv: Conversation, title: str) -> Conversation: async def apply(row: Conversation) -> None: row.title = title return await self._update(conv, apply) async def touch_user(self, conv: Conversation) -> Conversation: async def apply(row: Conversation) -> None: row.last_user_activity_at = datetime.now(UTC) return await self._update(conv, apply) async def _update( self, conv: Conversation, apply: Callable[[Conversation], Awaitable[None]] ) -> Conversation: async with self._db.session() as session: row = await session.get(Conversation, conv.id) if row is None: msg = f"conversation {conv.external_id} vanished" raise LookupError(msg) await apply(row) row.updated_at = datetime.now(UTC) session.add(row) await session.commit() await session.refresh(row) self._bus.publish("conversation.updated", **self.public(row)) return row def public(self, conv: Conversation) -> dict[str, Any]: return { "id": conv.external_id, "kind": conv.kind, "agent": conv.agent_name, "title": conv.title, "status": conv.status, "parent_row": conv.parent_id, "session_id": conv.session_id, "running_turn": conv.running_turn, "pending_question": conv.pending_question, "flags": conv.flags, "origin": conv.frontend, "created_at": _iso(conv.created_at), "last_user_activity_at": _iso(conv.last_user_activity_at), "last_activity_at": _iso(conv.last_activity_at), } 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"] = [ {"frontend": b.frontend, "external_id": b.external_id, "visible": b.visible} for b in await self.bindings(conv) ] live = self._pool.get(conv.external_id) out["live"] = live is not None out["busy"] = live.busy if live is not None else False runner = self._runners.get(cast("int", conv.id)) out["turn"] = runner.snapshot() if runner is not None else None pending = self.pending_question(conv.external_id) out["question"] = ( {"id": pending[0], "questions": pending[1]} if pending else None ) return out async def rate_limits(self, *, limit: int = 100) -> list[RateLimit]: async with self._db.session() as session: result = await session.exec( select(RateLimit).order_by(col(RateLimit.id).desc()).limit(limit) ) return list(result.all()) async def context_tokens(self, conv: Conversation) -> int: """Size of the context the last turn ran with, from its usage row.""" async with self._db.session() as session: row = ( await session.exec( select(Usage) .where(Usage.conversation_id == conv.external_id) .order_by(col(Usage.id).desc()) .limit(1) ) ).first() return context_of(row) async def usage_tokens(self, since: datetime) -> int: async with self._db.session() as session: rows = ( await session.exec( select(Usage).where( col(Usage.ts) >= since.astimezone(UTC).replace(tzinfo=None) ) ) ).all() return sum( r.input_tokens + r.output_tokens + r.cache_creation_tokens for r in rows ) async def busy(self, conv: Conversation) -> bool: """A turn is running, a question is open or a message waits to run.""" row = await self.get_row(cast("int", conv.id)) or conv if row.running_turn or row.pending_question: return True live = self._pool.get(row.external_id) if live is not None and live.busy: return True pending = await self._queue.pending(cast("int", row.id)) return any(i.priority in ("user", "urgent", "wake") for i in pending) # ---- 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: Kind) -> 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: Kind, agent: str | None = None, seed: str = "clean", parent: Conversation | None = None, text: str | None = None, title: str | None = None, 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). ``binding`` = ``(frontend, external_id)`` puts it into a window that already exists (a topic the user created) instead of asking the home frontend to ``materialize`` one. ``text`` rides with the seed as the first thing the user said, whatever the seed mode; without it the seed waits in ``flags`` and opens the first turn, so a fresh window stays silent and costs nothing until someone speaks. """ if seed not in SEEDS: msg = f"unknown seed {seed!r}" raise ValueError(msg) 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: msg = "seed=copy needs a parent with a session" raise ValueError(msg) session_id = await self._copy_session( parent, window=window, strip_tools=False ) conv = await self.create( kind=kind, agent=agent, parent=parent, 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]) else: await self.materialize(conv) ctx = SeedContext( kind=kind, seed=seed, agent=agent, parent=parent, text=text, title=title ) if text is None: return await self.set_flags(conv, {"seed": seed, "seed_window": window}) await self._queue.push( conversation_id=cast("int", conv.id), priority="user", origin=f"сид:{seed}" if seed == "brief" else origin, text=await self._seed_text(ctx, window=window), ) self._ensure_worker(cast("int", conv.id)) return conv async def _pending_seed(self, conv: Conversation) -> str | None: """A seed nobody has spoken after yet: rendered now, spent once.""" seed = conv.flags.get("seed") if not seed: return None parent = await self.get_row(conv.parent_id) if conv.parent_id else None ctx = SeedContext( kind=as_kind(conv.kind), seed=str(seed), agent=conv.agent_name, parent=parent, text=None, title=conv.title, ) window = conv.flags.get("seed_window") text = await self._seed_text( ctx, window=window if isinstance(window, int) else None ) await self.set_flags(conv, {"seed": None, "seed_window": None}) return text async def fork( self, conv: Conversation, prompt: str, *, 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, agent=agent ) child = await self.create( kind="fork", agent=agent, parent=conv, title=title or f"fork: {conv.title or conv.external_id}", origin="system", session_id=session_id, ) try: text, capture = await self.run_text_turn( child, prompt, origin="fork", tools=False ) finally: await self._backend(agent).close(child.external_id) child = await self.set_status(child, "closed") return ForkResult(conversation=child, text=text, capture=capture) async def read(self, conv: Conversation, *, window: int | None = None) -> str: 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 [] key = {**self._store_key(conv), "subpath": subpath} return list(await self._store.load(cast("Any", key)) or []) async def subpaths(self, conv: Conversation) -> list[str]: if conv.session_id is None: return [] return list(await self._store.list_subkeys(cast("Any", self._store_key(conv)))) async def inject( self, conv: Conversation, text: str, *, urgency: Priority = "normal", origin: str = "system", interrupt: bool = True, ) -> InjectQueueItem: if conv.kind == "master" and conv.status != "open": live = await self._open_master() if live is None: _log.error( "inject (%s) for closed master %s: no open master, it stays there", origin, conv.external_id, ) else: _log.info( "inject (%s) for closed master %s goes to %s", origin, conv.external_id, live.external_id, ) conv = live item = await self._queue.push( conversation_id=cast("int", conv.id), priority=urgency, origin=origin, text=text, ) self._bus.publish( "inject.queued", conversation_id=conv.external_id, item=item.id, priority=urgency, origin=origin, ) if urgency == "urgent" and interrupt: backend = self._backend(conv.agent_name) if await backend.interrupt(conv.external_id): _log.info( "conversation %s: interrupted for urgent inject", conv.external_id ) await self._queue.mark_interrupting(item) self._ensure_worker(cast("int", conv.id)) return item async def post( self, conv: Conversation, text: str, *, origin: str = "user" ) -> InjectQueueItem: item = await self._queue.push( conversation_id=cast("int", conv.id), priority="user", origin=origin, text=text, ) await self.touch_user(conv) self._bus.publish( "message.queued", conversation_id=conv.external_id, item=item.id, origin=origin, ) self._ensure_worker(cast("int", conv.id)) return item def turn_origin(self, conv: Conversation) -> str | None: """Origin of the running turn (``user``, ``inject``, ...); None when idle.""" runner = self._runners.get(cast("int", conv.id)) return runner.origin if runner is not None and runner.turn_id else None async def say(self, conv: Conversation, text: str) -> dict[str, Any]: runner = self._runners.get(cast("int", conv.id)) _log.info("say[%s]: %s", conv.external_id, text[:200]) return self._bus.publish( "say", conversation_id=conv.external_id, text=text, turn_id=runner.turn_id if runner is not None else None, ) async def merge(self, conv: Conversation) -> ForkResult: if conv.parent_id is None: msg = "merge needs a parent conversation" raise ValueError(msg) parent = await self.get_row(conv.parent_id) if parent is None: msg = "parent conversation vanished" raise LookupError(msg) result = await self.fork( conv, self._texts.merge_prompt, title=f"слив: {conv.title or conv.external_id}", ) if result.text.strip(): await self.inject(parent, result.text, urgency="normal", origin="слив") await self.set_status(conv, "merged") await self.mark_closed(conv) self._bus.publish( "conversation.merged", conversation_id=conv.external_id, parent=parent.external_id, fork=result.conversation.external_id, ) return result async def schedule( self, conv: Conversation, at: str, text: str, *, urgency: Priority = "wake", dedupe_key: str | None = None, ) -> tuple[int | None, datetime]: if self.scheduler is None: msg = "no scheduler; `schedule` is unavailable" raise RuntimeError(msg) return await self.scheduler.schedule( conv, at, text, urgency=urgency, dedupe_key=dedupe_key ) async def schedules(self, conv: Conversation | None = None) -> list[dict[str, Any]]: return await self.scheduler.scheduled(conv) if self.scheduler else [] # ---- §4.5 rotation ------------------------------------------------- async def handout(self, conv: Conversation, ctx: HandoutContext) -> str: """The closing master's last turn: the handout prompt from the config.""" source = self._texts.handout if isinstance(source, str): prompt = source.format(day=ctx.day.isoformat(), reason=ctx.reason) else: produced: Any = source(ctx) prompt = await produced if inspect.isawaitable(produced) else produced self._bus.publish( "handout.start", conversation_id=conv.external_id, day=ctx.day.isoformat() ) try: text, _ = await self.run_text_turn(conv, prompt, origin="handout") except Exception: # noqa: BLE001 _log.exception("handout turn on %s failed", conv.external_id) text = "" self._bus.publish( "handout.end", conversation_id=conv.external_id, day=ctx.day.isoformat(), text=text[:2000], ) return text async def close(self, conv: Conversation) -> Conversation: row = await self.set_status(conv, "closed") with contextlib.suppress(LookupError): 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 return await self._update(conv, apply) async def mark_closed(self, conv: Conversation) -> bool: marked = False for fe in self._frontends: if conv.kind in fe.kinds: try: marked = await fe.mark_closed(conv) or marked except Exception: # noqa: BLE001 _log.exception("%s could not mark %s", fe.name, conv.external_id) return marked async def new_day( self, conv: Conversation, *, reason: str = "ночь", moved: int = 0 ) -> InjectQueueItem: ctx = NewDayContext( day=datetime.now(UTC).astimezone().date(), reason=reason, moved=moved ) source = self._texts.new_day if isinstance(source, str): text = source.format(day=ctx.day.isoformat(), reason=ctx.reason) else: produced: Any = source(ctx) text = await produced if inspect.isawaitable(produced) else produced if moved: text += f" Инжектов переехало из старого мастера: {moved}." return await self.inject( conv, text, urgency="urgent", origin="ротация", interrupt=False ) # ---- §3.7 questions ------------------------------------------------ async def ask(self, key: str, payload: dict[str, Any]) -> str | None: """``AskUserQuestion`` reached ``can_use_tool``: show it, wait for the answer. Returns the answer text, or ``None`` when nobody answered within ``question_timeout`` - the caller then tells the model to finish the turn, the frontend has already rendered the question as text. """ conv = await self.get(key) if conv is None: return None runner = self._runners.get(cast("int", conv.id)) question_id = f"q_{uuid.uuid4().hex[:10]}" pending = _Question( conversation_id=key, turn_id=runner.turn_id if runner is not None else None, questions=list(payload.get("questions") or []), answer=asyncio.get_running_loop().create_future(), ) self._questions[question_id] = pending await self._set_pending_question(conv, value=True) self._bus.publish( "question", conversation_id=key, turn_id=pending.turn_id, question_id=question_id, questions=pending.questions, timeout=self._question_timeout, ) try: async with asyncio.timeout(self._question_timeout): answer = await pending.answer except TimeoutError: self._bus.publish( "question.timeout", conversation_id=key, turn_id=pending.turn_id, question_id=question_id, ) return None finally: self._questions.pop(question_id, None) await self._set_pending_question(conv, value=False) self._bus.publish( "question.answered", conversation_id=key, turn_id=pending.turn_id, question_id=question_id, answer=answer, ) return answer def answer(self, question_id: str, answer: str) -> bool: pending = self._questions.get(question_id) if pending is None or pending.answer.done(): return False pending.answer.set_result(answer) return True def pending_question(self, key: str) -> tuple[str, list[dict[str, Any]]] | None: for question_id, pending in self._questions.items(): if pending.conversation_id == key and not pending.answer.done(): return question_id, pending.questions return None def answer_text(self, answer: str | None) -> str: if answer is None: return self._texts.unanswered.format( minutes=round(self._question_timeout / 60) ) return self._texts.answered.format(answer=answer) async def _set_pending_question(self, conv: Conversation, *, value: bool) -> None: async def apply(row: Conversation) -> None: row.pending_question = value with contextlib.suppress(LookupError): await self._update(conv, apply) # ---- turns --------------------------------------------------------- async def turn( self, conv: Conversation, *, messages: Sequence[Any], origin: str, capture: TurnCapture | None = None, session_id: str | None = None, use_session: bool = True, tools: bool = True, turn_id: str | None = None, item_origin: str | None = None, ) -> AsyncIterator[MessageStreamEvent]: """Run one turn on ``conv`` under its lock; the only path to the backend. ``use_session=False`` withholds the stored ``session_id`` so the backend reseeds from ``messages`` (markdown file diverged). """ row_id = cast("int", conv.id) runner = self._runner(row_id) backend = self._backend(conv.agent_name) turn_id = turn_id or f"turn_{uuid.uuid4().hex[:12]}" capture = capture or TurnCapture() resume = session_id if session_id is not None else conv.session_id async with runner.lock: runner.turn_id = turn_id runner.origin = origin runner.text = _prompt_preview(messages) 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, turn_id=turn_id, origin=origin, item_origin=item_origin, text=runner.text, ) stop = "error" cut = False try: events = backend.complete( agent=self._claude_agent(conv.agent_name), messages=messages, conversation_id=conv.external_id, session_id=resume if use_session else None, reseed=not use_session, capture=capture, kind=conv.kind, pinned=conv.kind == "master", tools=tools, observer=self._observer(conv, runner, turn_id, origin), turn_id=turn_id, ) async for event in events: yield event stop = "interrupted" if capture.interrupted else "end_turn" except asyncio.CancelledError: cut = True raise 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, turn_id=turn_id, origin=origin, item_origin=item_origin, stop=stop, usage=_usage_dict(capture), ) async def run_text_turn( self, conv: Conversation, text: str, *, origin: str, tools: bool = True, turn_id: str | None = None, item_origin: str | None = None, ) -> tuple[str, TurnCapture]: capture = TurnCapture() acc = StreamAccumulator() agent = self._claude_agent(conv.agent_name) async for event in self.turn( conv, messages=[{"role": "user", "content": text}], origin=origin, capture=capture, tools=tools, turn_id=turn_id, item_origin=item_origin, ): acc.feed(event) message = acc.finalize(model=agent.model) reply = "\n\n".join( getattr(b, "text", "") for b in message.content if getattr(b, "type", "") == "text" ).strip() return reply, capture # ---- lifecycle ----------------------------------------------------- async def start(self) -> None: await self.recover() for row_id in await self._queue.conversations_with_pending(): self._ensure_worker(row_id) self._idle_task = asyncio.create_task(self._idle_loop()) async def stop(self) -> None: tasks = list(self._tasks) if self._idle_task is not None: tasks.append(self._idle_task) for task in tasks: task.cancel() for task in tasks: with contextlib.suppress(BaseException): await task self._tasks.clear() self._idle_task = None async def recover(self) -> list[Conversation]: """Restart path (§8.1): repair transcripts of turns cut mid-flight.""" async with self._db.session() as session: result = await session.exec( select(Conversation).where(col(Conversation.running_turn).is_not(None)) ) cut = list(result.all()) for conv in cut: fixed = 0 if conv.session_id is not None: backend = self._backend(conv.agent_name) try: fixed = await backend.repair_session(conv.session_id) except Exception: # noqa: BLE001 _log.exception("repair of %s failed", conv.session_id) turn_id = conv.running_turn async def clear(row: Conversation) -> None: row.running_turn = None row.pending_question = False await self._update(conv, clear) note = f"тёрн {turn_id} оборван рестартом gateway" if fixed: note += ( f"; {fixed} незакрытых тул-коллов получили tool_result " f"«{self._texts.interrupted}»" ) await self.inject(conv, note, urgency="normal", origin="система") _log.warning("conversation %s: %s", conv.external_id, note) interrupted = await self._queue.interrupted() for item in interrupted: _log.warning( "queue item #%s (%s) was running at restart; marked interrupted", item.id, item.priority, ) return cut # ---- internals ----------------------------------------------------- def _backend(self, agent: str) -> ClaudeSdkBackend: backend = self._backends.get(agent) if backend is None or not hasattr(backend, "repair_session"): msg = f"agent {agent!r} has no Claude SDK backend" raise LookupError(msg) return cast("ClaudeSdkBackend", backend) def _claude_agent(self, name: str) -> ClaudeAgent: agent = self._agents.get(name) if agent is None or not hasattr(agent, "cwd"): msg = f"unknown Claude agent {name!r}" raise LookupError(msg) return cast("ClaudeAgent", agent) def _store_key(self, conv: Conversation) -> dict[str, str]: agent = self._claude_agent(conv.agent_name) return { "project_key": project_key_for_directory(str(agent.cwd)), "session_id": cast("str", conv.session_id), } async def _copy_session( 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" raise ValueError(msg) live = self._pool.get(conv.external_id) if live is not None and live.dirty: msg = f"conversation {conv.external_id} has a mirror gap; not forking" raise RuntimeError(msg) 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(source.cwd) ) 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", 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, forked.session_id, window, strip_tools, ) return forked.session_id async def _seed_text(self, ctx: SeedContext, *, window: int | None) -> str: stamp = datetime.now(UTC).astimezone().strftime("%Y-%m-%d %H:%M") title = f" «{ctx.title}»" if ctx.title else "" head = f"[сид: {ctx.seed}] {ctx.kind}{title}, {stamp}." body: str | None = None if self._texts.seed is not None: produced: Any = self._texts.seed(ctx) if inspect.isawaitable(produced): produced = await produced body = cast("str | None", produced) if body is None: if ctx.seed == "brief": body = ctx.text elif ctx.seed == "copy": scope = f"последние {window} тёрнов" if window else "вся история" body = f"История родителя скопирована ({scope}); продолжай в ней." elif ctx.seed == "morning": body = "Хендаут не приехал." parts = [head, body, ctx.text if ctx.seed != "brief" else None] return "\n\n".join(p for p in parts if p) def _runner(self, row_id: int) -> _Runner: runner = self._runners.get(row_id) if runner is None: runner = _Runner() self._runners[row_id] = runner return runner def _ensure_worker(self, row_id: int) -> None: runner = self._runner(row_id) runner.wake.set() if runner.task is None or runner.task.done(): runner.task = asyncio.create_task(self._worker(row_id)) self._tasks.add(runner.task) runner.task.add_done_callback(self._tasks.discard) async def _worker(self, row_id: int) -> None: runner = self._runner(row_id) while True: items = await self._queue.pending(row_id) batch, wait = self._pick(items) if batch is None: runner.wake.clear() if wait is None: return with contextlib.suppress(TimeoutError): await asyncio.wait_for(runner.wake.wait(), timeout=wait) continue conv = await self.get_row(row_id) if conv is None: await self._queue.finish(batch, status="failed") return await self._run_batch(conv, batch) def _pick( self, items: list[InjectQueueItem] ) -> tuple[list[InjectQueueItem] | None, float | None]: if not items: return None, None head = items[0] if head.priority == "urgent": return [head], None tail = [i for i in items if i is not head and i.priority in ("wake", "normal")] if head.priority in ("user", "wake"): return [head, *tail], None age = (datetime.now(UTC) - _aware(head.created_at)).total_seconds() if age >= self._normal_window: return [head, *tail], None return None, max(self._normal_window - age, 1.0) async def _run_batch( self, conv: Conversation, batch: list[InjectQueueItem] ) -> None: head = batch[0] turn_id = f"turn_{uuid.uuid4().hex[:12]}" await self._queue.start(batch, turn_id) if head.priority == "user": origin = "user" prompt = head.text await self._note_user(conv, head.text) envelope = self._envelope_for(conv, head.text) if envelope: prompt += "\n\n" + envelope if len(batch) > 1: prompt += "\n\n" + _bundle(batch[1:]) else: origin = "inject" prompt = "\n\n".join( f"{self._texts.inject_header(injects.context_of(i))}\n{i.text}" for i in batch ) seed = await self._pending_seed(conv) if seed: prompt = f"{seed}\n\n{prompt}" try: text, capture = await self.run_text_turn( conv, prompt, origin=origin, turn_id=turn_id, item_origin=head.origin ) except Exception: # noqa: BLE001 _log.exception("turn %s on %s failed", turn_id, conv.external_id) await self._queue.finish(batch, status="failed") return await self._queue.finish( batch, status="interrupted" if capture.interrupted else "done" ) if origin == "user": self._bus.publish( "reply", conversation_id=conv.external_id, turn_id=turn_id, item=head.id, item_origin=head.origin, source="queue", prompt=prompt, user_text=head.text, text=text, ) def _envelope_for(self, conv: Conversation, text: str = "") -> str | None: """Master gets the whole envelope, a branch only the recall lines (§3.3).""" if self._envelope is None: return None if conv.kind == "master": return self._envelope.build(text=text, kind="master") if conv.kind == "branch": return self._envelope.recall_only(text=text, kind="branch") return None async def _note_user(self, conv: Conversation, text: str) -> None: """Hand the user's text to the setup's sink. The setup keeps its own log of what Бобёр said, outside the transcript; a failure there never blocks the turn. """ if self._user_sink is None: return message = UserSaid( conversation_id=conv.external_id, kind=conv.kind, title=conv.title, text=text, at=datetime.now(UTC), ) try: result = self._user_sink(message) if inspect.isawaitable(result): await result except Exception: # noqa: BLE001 _log.exception("user sink failed for %s", conv.external_id) def _observer( self, conv: Conversation, runner: _Runner, turn_id: str, origin: str ) -> Callable[[Any], None]: conversation_id = conv.external_id def observe(message: Any) -> None: parent = getattr(message, "parent_tool_use_id", None) if isinstance(message, RateLimitEvent): self._observe_rate_limit(conv, message) elif isinstance(message, StreamEvent): self._bus.publish( "stream", conversation_id=conversation_id, turn_id=turn_id, origin=origin, parent_tool_use_id=parent, event=message.event, ) elif isinstance(message, AssistantMessage): for block in message.content: if isinstance(block, ToolUseBlock): event = self._bus.publish( "tool", conversation_id=conversation_id, turn_id=turn_id, origin=origin, parent_tool_use_id=parent, tool_use_id=block.id, name=block.name, input=block.input, ) runner.tools[block.id] = { "tool_use_id": block.id, "name": block.name, "input": block.input, "parent_tool_use_id": parent, "started_at": event["ts"], "ended_at": None, "is_error": None, "content": None, } elif isinstance(message, UserMessage): blocks = message.content if isinstance(message.content, list) else () for block in blocks: if isinstance(block, ToolResultBlock): event = self._bus.publish( "tool.result", conversation_id=conversation_id, turn_id=turn_id, origin=origin, parent_tool_use_id=parent, tool_use_id=block.tool_use_id, is_error=bool(block.is_error), content=_result_preview(block.content), ) node = runner.tools.get(block.tool_use_id) if node is not None: node["ended_at"] = event["ts"] node["is_error"] = event["is_error"] node["content"] = event["content"] elif isinstance(message, ResultMessage) and parent is None: self._bus.publish( "result", conversation_id=conversation_id, turn_id=turn_id, origin=origin, subtype=message.subtype, is_error=message.is_error, num_turns=message.num_turns, ) return observe def _observe_rate_limit(self, conv: Conversation, message: RateLimitEvent) -> None: info = message.rate_limit_info row = RateLimit( window=info.rate_limit_type or "unknown", status=info.status, utilization=info.utilization, resets_at=_from_unix(info.resets_at), overage_status=info.overage_status, overage_resets_at=_from_unix(info.overage_resets_at), agent_name=conv.agent_name, session_id=message.session_id, raw=dict(info.raw), ) self._bus.publish( "rate_limit", conversation_id=conv.external_id, window=row.window, status=row.status, utilization=row.utilization, resets_at=_iso(row.resets_at), overage_status=row.overage_status, ) task = asyncio.create_task(self._record_rate_limit(row)) self._tasks.add(task) task.add_done_callback(self._tasks.discard) async def _record_rate_limit(self, row: RateLimit) -> None: try: async with self._db.session() as session: session.add(row) await session.commit() except Exception: # noqa: BLE001 _log.exception("rate limit write failed") async def _mark_running(self, conv: Conversation, turn_id: str) -> None: async def apply(row: Conversation) -> None: row.running_turn = turn_id row.last_activity_at = datetime.now(UTC) await self._update(conv, apply) async def _mark_done( self, conv: Conversation, capture: TurnCapture, *, cut: bool = False ) -> None: """Close the turn; a cancelled one keeps ``running_turn`` for ``recover``.""" async def apply(row: Conversation) -> None: if not cut: row.running_turn = None row.last_activity_at = datetime.now(UTC) if capture.session_id is not None: row.session_id = capture.session_id await self._update(conv, apply) async def _idle_loop(self) -> None: while True: try: await self._emit_idle() except Exception: # noqa: BLE001 _log.exception("idle watcher failed") await asyncio.sleep(self._idle_interval) async def _emit_idle(self) -> None: if not self._idle_days: return now = datetime.now(UTC) for conv in await self.find(status="open", limit=10_000): last = _aware(conv.last_activity_at or conv.created_at) days = int((now - last).total_seconds() // 86400) due = [d for d in self._idle_days if days >= d] if not due: continue notified = int(conv.flags.get("idle_notified", 0) or 0) if due[-1] <= notified: continue await self.set_flags(conv, {"idle_notified": due[-1]}) bindings = await self.bindings(conv) self._bus.publish( "conversation.idle", conversation_id=conv.external_id, kind=conv.kind, agent=conv.agent_name, days=due[-1], bindings=[ {"frontend": b.frontend, "external_id": b.external_id} for b in bindings if b.visible ], ) def parse_at(at: str, tz: tzinfo = UTC) -> datetime: raw = at.strip() match = _RELATIVE.match(raw.replace(" ", "")) if match: amount, unit = match.groups() return datetime.now(UTC) + timedelta(seconds=int(amount) * _UNITS[unit]) parsed = datetime.fromisoformat(raw) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=tz) return parsed.astimezone(UTC) def _bundle(items: Sequence[InjectQueueItem]) -> str: lines = [f"[инжекты, накопившиеся с {_iso(items[0].created_at)}; это не Бобёр]"] # noqa: RUF001 lines.extend(f"- [{i.origin}] {i.text}" for i in items) return "\n".join(lines) def _aware(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=UTC) 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 text = text_of(messages[-1].get("content")) return text[:limit] if text else None def _from_unix(value: int | None) -> datetime | None: return datetime.fromtimestamp(value, tz=UTC) if value is not None else None def _result_preview( content: str | list[dict[str, Any]] | None, limit: int = 400 ) -> str: """Short text of a tool result for the panel: the transcript keeps the whole.""" if content is None: return "" text = ( content if isinstance(content, str) else "\n".join( str(part.get("text", "")) for part in content if isinstance(part, dict) and part.get("type") == "text" ) ) return text if len(text) <= limit else text[:limit] + "…" def _usage_dict(capture: TurnCapture) -> dict[str, Any] | None: usage = capture.usage if usage is None: return None return { "input": usage.input_tokens, "output": usage.output_tokens, "cache_read": usage.cache_read_tokens, "cache_creation": usage.cache_creation_tokens, "cost_usd": usage.cost_usd, "duration_ms": usage.duration_ms, } def context_of(row: Usage | None) -> int: """Context size of a turn. The last API call's input, or, for rows written before it was recorded, the per-call average of the turn's input sums. """ if row is None: return 0 if row.context_tokens: return row.context_tokens total = row.input_tokens + row.cache_read_tokens + row.cache_creation_tokens return round(total / max(row.num_turns or 1, 1))