diff --git a/.env.example b/.env.example
index a37a803..e67469d 100644
--- a/.env.example
+++ b/.env.example
@@ -23,3 +23,8 @@ CLAUDE_CODE_OAUTH_TOKEN=
RAYCAST_CONFIG_PATH=../raycast-api/config.json
RAYCAST_DEVICE_ID=
RAYCAST_BEARER=
+
+# Telegram (нужно только если в config.py есть TelegramFrontend):
+# токен от @BotFather, свой user id - @userinfobot
+TELEGRAM_BOT_TOKEN=
+TELEGRAM_USER_ID=
diff --git a/pyproject.toml b/pyproject.toml
index f77068d..acd44ea 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,6 +8,7 @@ authors = [
requires-python = ">=3.13"
dependencies = [
"aiofile>=3.11.1",
+ "aiogram>=3.31.0",
"aiohttp>=3.13.5",
"aiosqlite>=0.22.1",
"anthropic>=0.103.0",
diff --git a/src/beaver_gateway/backends/claude_sdk.py b/src/beaver_gateway/backends/claude_sdk.py
index e031437..66125f6 100644
--- a/src/beaver_gateway/backends/claude_sdk.py
+++ b/src/beaver_gateway/backends/claude_sdk.py
@@ -41,6 +41,7 @@ import sys
import tempfile
import time
import uuid
+import warnings
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
@@ -49,9 +50,12 @@ from typing import TYPE_CHECKING, Any, Self, cast
import claude_agent_sdk
from claude_agent_sdk import (
AssistantMessage,
+ CanUseToolShadowedWarning,
ClaudeAgentOptions,
ClaudeSDKClient,
MirrorErrorMessage,
+ PermissionResultAllow,
+ PermissionResultDeny,
ResultMessage,
StreamEvent,
TextBlock,
@@ -91,7 +95,12 @@ if TYPE_CHECKING:
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
from anthropic.types import MessageParam
- from claude_agent_sdk import McpSdkServerConfig, SessionStore
+ from claude_agent_sdk import (
+ McpSdkServerConfig,
+ PermissionResult,
+ SessionStore,
+ ToolPermissionContext,
+ )
from beaver_gateway.agents.base import BaseAgent
from beaver_gateway.agents.claude import ClaudeAgent
@@ -100,6 +109,12 @@ if TYPE_CHECKING:
_log = logging.getLogger("beaver_gateway.backends.claude_sdk")
+# §3.7: in bypass the callback only ever sees AskUserQuestion, and that is
+# exactly the one we want - the SDK's warning about the rest is noise here.
+warnings.filterwarnings("ignore", category=CanUseToolShadowedWarning)
+
+ASK_TOOL = "AskUserQuestion"
+
__all__ = [
"ClaudeSdkBackend",
"RunnerConfig",
@@ -151,6 +166,11 @@ ClientFactory = "Callable[[ClaudeAgentOptions], SessionClient]"
UsageSink = "Callable[[UsageEvent], Awaitable[None]]"
ToolServerFactory = "Callable[[str, str], McpSdkServerConfig | None]"
"""``(conversation_key, kind) -> in-process MCP server config`` or ``None``."""
+Asker = "Callable[[str, dict[str, Any]], Awaitable[str]]"
+"""``(conversation_key, AskUserQuestion input) -> text the model reads as the
+tool result``. The only channel an answer has in bypass mode is
+``PermissionResultDeny.message`` (spike S1, s05): ``updated_input`` never
+reaches the model."""
@dataclass(frozen=True, slots=True)
@@ -197,6 +217,7 @@ class ClaudeSdkBackend:
work_dir: Path | None = None,
pool: SessionPool | None = None,
tool_server: Callable[[str, str], McpSdkServerConfig | None] | None = None,
+ asker: Callable[[str, dict[str, Any]], Awaitable[str]] | None = None,
) -> None:
self._agent = agent
self._store = session_store
@@ -208,6 +229,7 @@ class ClaudeSdkBackend:
self._mcp_disallowed = _mcp_disallowed(agent, mcp_tool_names or {})
self._pool = pool if pool is not None else SessionPool()
self._tool_server = tool_server
+ self._asker = asker if ASK_TOOL not in agent.options.disallowed_tools else None
self._uid, self._gid = _resolve_ids(self._runner.user)
self._wrapper: Path | None = None
@@ -553,6 +575,7 @@ class ClaudeSdkBackend:
env=env,
cli_path=str(self._exec_wrapper(extra_keep=tuple(env))),
include_partial_messages=opt.include_partial_messages,
+ can_use_tool=self._can_use_tool(key) if self._asker else None,
session_store=self._store,
session_store_flush=cast("Any", opt.session_store_flush),
resume=resume,
@@ -564,6 +587,30 @@ class ClaudeSdkBackend:
),
)
+ def _can_use_tool(
+ self, key: str
+ ) -> Callable[
+ [str, dict[str, Any], ToolPermissionContext], Awaitable[PermissionResult]
+ ]:
+ asker = self._asker
+
+ async def can_use_tool(
+ name: str, tool_input: dict[str, Any], _ctx: ToolPermissionContext
+ ) -> PermissionResult:
+ if name != ASK_TOOL or asker is None:
+ return PermissionResultAllow()
+ live = self._pool.get(key)
+ if live is not None:
+ live.pending_question = True
+ try:
+ message = await asker(key, tool_input)
+ finally:
+ if live is not None:
+ live.pending_question = False
+ return PermissionResultDeny(message=message)
+
+ return can_use_tool
+
def _plugins(self) -> list[dict[str, str]]:
plugins: list[dict[str, str]] = []
root = self._work_dir / "plugins" / self._agent.name
diff --git a/src/beaver_gateway/cli.py b/src/beaver_gateway/cli.py
index 4ac95c9..bd19015 100644
--- a/src/beaver_gateway/cli.py
+++ b/src/beaver_gateway/cli.py
@@ -26,7 +26,7 @@ import functools
import logging
import signal
from contextlib import AsyncExitStack
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
import uvicorn
import uvloop
@@ -294,6 +294,13 @@ class _LateConversations:
return None
return build_tool_server(self.conversations, conversation_key=key, names=names)
+ async def ask(self, key: str, payload: dict[str, Any]) -> str:
+ if self.conversations is None:
+ msg = "conversations service is not up yet"
+ raise RuntimeError(msg)
+ answer = await self.conversations.ask(key, payload)
+ return self.conversations.answer_text(answer)
+
async def _build_backends(
*,
@@ -368,6 +375,7 @@ async def _build_backends(
usage_sink=record_usage,
pool=pool,
tool_server=functools.partial(late.server, names=a.gateway_tools),
+ asker=late.ask,
)
await stack.enter_async_context(adapter)
backends[a.name] = adapter
diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py
index 6803fc3..27d793a 100644
--- a/src/beaver_gateway/core/conversations.py
+++ b/src/beaver_gateway/core/conversations.py
@@ -112,6 +112,11 @@ class ConversationTexts:
merge_prompt: str = _DEFAULT_MERGE_PROMPT
interrupted: str = "прервано"
+ answered: str = "Пользователь ответил: {answer}"
+ unanswered: str = (
+ "Пользователь не ответил за {minutes} мин. Вопрос ему показан текстом; "
+ "заверши тёрн сейчас, ответ придёт следующим сообщением."
+ )
seed: Callable[[SeedContext], Awaitable[str | None] | str | None] | None = None
@@ -130,6 +135,14 @@ class _Runner:
turn_id: str | None = None
+@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,
@@ -145,6 +158,7 @@ class Conversations:
normal_window: float = 3600.0,
idle_days: Sequence[int] = (2,),
idle_interval: float = 3600.0,
+ question_timeout: float = 600.0,
) -> None:
self._db = db
self._agents = agents
@@ -157,6 +171,8 @@ class Conversations:
self._normal_window = normal_window
self._idle_days = tuple(sorted(idle_days))
self._idle_interval = idle_interval
+ self._question_timeout = question_timeout
+ self._questions: dict[str, _Question] = {}
self._queue = InjectQueue(db)
self._runners: dict[int, _Runner] = {}
self._tasks: set[asyncio.Task[None]] = set()
@@ -278,6 +294,17 @@ class Conversations:
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),
@@ -436,7 +463,15 @@ class Conversations:
title: str | None = None,
window: int | None = None,
origin: str = "api",
+ binding: tuple[str, str] | 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.
+ """
if seed not in SEEDS:
msg = f"unknown seed {seed!r}"
raise ValueError(msg)
@@ -465,7 +500,10 @@ class Conversations:
origin=origin,
session_id=session_id,
)
- await self.materialize(conv)
+ if binding is not None:
+ await self.bind(conv, frontend=binding[0], external_id=binding[1])
+ else:
+ await self.materialize(conv)
prompt = await self._seed_text(
SeedContext(
kind=kind, seed=seed, agent=agent, parent=parent, text=text, title=title
@@ -619,6 +657,86 @@ class Conversations:
)
return row
+ # ---- §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)
+
async def schedules(self, conv: Conversation | None = None) -> list[Schedule]:
stmt = select(Schedule).order_by(col(Schedule.execute_at))
if conv is not None:
@@ -757,6 +875,7 @@ class Conversations:
async def clear(row: Conversation) -> None:
row.running_turn = None
+ row.pending_question = False
await self._update(conv, clear)
note = f"тёрн {turn_id} оборван рестартом gateway"
@@ -851,7 +970,8 @@ class Conversations:
body = f"История родителя скопирована ({scope}); продолжай в ней."
elif ctx.seed == "morning":
body = "Хендаут не приехал."
- return f"{head}\n\n{body}" if body else head
+ 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)
@@ -933,8 +1053,10 @@ class Conversations:
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,
)
diff --git a/src/beaver_gateway/frontends/telegram/__init__.py b/src/beaver_gateway/frontends/telegram/__init__.py
new file mode 100644
index 0000000..687778a
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/__init__.py
@@ -0,0 +1,5 @@
+"""Telegram frontend (§3.8): General = master, topic = branch, drafts, inbox/outbox."""
+
+from beaver_gateway.frontends.telegram.frontend import Attachments, TelegramFrontend
+
+__all__ = ["Attachments", "TelegramFrontend"]
diff --git a/src/beaver_gateway/frontends/telegram/drafts.py b/src/beaver_gateway/frontends/telegram/drafts.py
new file mode 100644
index 0000000..7f7944b
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/drafts.py
@@ -0,0 +1,101 @@
+"""One ``sendMessageDraft`` stream per running turn (§3.8).
+
+A draft is ephemeral and lives 30 s, Telegram throttles edits to about one
+per second per chat, and thinking or a tool call would otherwise look like a
+hang - so the draft opens with a status line straight away, is refreshed on
+a timer rather than on every delta, and is kept alive while nothing changes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import time
+import zlib
+from typing import TYPE_CHECKING
+
+from aiogram.exceptions import TelegramAPIError
+
+if TYPE_CHECKING:
+ from aiogram import Bot
+
+__all__ = ["Draft"]
+
+_log = logging.getLogger("beaver_gateway.frontends.telegram.drafts")
+
+_TAIL = 3500
+_KEEPALIVE = 20.0
+
+
+class Draft:
+ def __init__(
+ self,
+ bot: Bot,
+ *,
+ chat_id: int,
+ thread_id: int | None,
+ turn_id: str,
+ interval: float = 0.7,
+ status: str = "⏳ думаю…",
+ ) -> None:
+ self._bot = bot
+ self._chat_id = chat_id
+ self._thread_id = thread_id
+ self._draft_id = (zlib.crc32(turn_id.encode()) & 0x7FFFFFFF) or 1
+ self._interval = interval
+ self.status = status
+ self.text = ""
+ self._dirty = True
+ self._broken = False
+ self._last_sent = 0.0
+ self._task: asyncio.Task[None] | None = None
+
+ def start(self) -> None:
+ if self._task is None:
+ self._task = asyncio.create_task(self._run())
+
+ def set_status(self, status: str) -> None:
+ if status != self.status:
+ self.status = status
+ self._dirty = True
+
+ def append(self, text: str) -> None:
+ if text:
+ self.text += text
+ self._dirty = True
+
+ async def stop(self) -> None:
+ if self._task is None:
+ return
+ self._task.cancel()
+ with contextlib.suppress(asyncio.CancelledError):
+ await self._task
+ self._task = None
+
+ async def _run(self) -> None:
+ while not self._broken:
+ if self._dirty or time.monotonic() - self._last_sent > _KEEPALIVE:
+ await self._push()
+ await asyncio.sleep(self._interval)
+
+ async def _push(self) -> None:
+ self._dirty = False
+ self._last_sent = time.monotonic()
+ try:
+ await self._bot.send_message_draft(
+ chat_id=self._chat_id,
+ draft_id=self._draft_id,
+ message_thread_id=self._thread_id,
+ text=self._render(),
+ parse_mode=None,
+ )
+ except TelegramAPIError as exc:
+ self._broken = True
+ _log.warning(
+ "draft to %s/%s stopped: %s", self._chat_id, self._thread_id, exc
+ )
+
+ def _render(self) -> str:
+ tail = self.text[-_TAIL:]
+ return f"{self.status}\n\n{tail}" if tail.strip() else self.status
diff --git a/src/beaver_gateway/frontends/telegram/frontend.py b/src/beaver_gateway/frontends/telegram/frontend.py
new file mode 100644
index 0000000..37c896c
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/frontend.py
@@ -0,0 +1,845 @@
+"""``TelegramFrontend`` - the private chat with the bot as the window (§3.8).
+
+General is the master, a topic is a branch. The user makes a topic and the
+first message in it spawns the branch (``seed=morning``); a message into a
+topic whose branch is merged or closed spawns a new branch on the same
+topic. Replies stream as drafts and land through the outbox; turns that
+came from other windows are mirrored with a marker; ``origin=system`` is
+never shown. ``AskUserQuestion`` becomes inline buttons (§3.7).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import html
+import logging
+import tempfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, cast
+
+from aiogram import Bot
+from aiogram.client.default import DefaultBotProperties
+from aiogram.exceptions import TelegramAPIError
+from aiogram.types import (
+ CallbackQuery,
+ InlineKeyboardButton,
+ InlineKeyboardMarkup,
+ Message,
+ ReactionTypeEmoji,
+ Update,
+)
+
+from beaver_gateway.frontends.base import Frontend
+from beaver_gateway.frontends.telegram.drafts import Draft
+from beaver_gateway.frontends.telegram.inbox import Inbox
+from beaver_gateway.frontends.telegram.outbox import Outbox
+from beaver_gateway.frontends.telegram.render import status_label
+
+if TYPE_CHECKING:
+ from beaver_gateway.core.bus import Event, EventBus
+ from beaver_gateway.core.conversations import Conversations
+ from beaver_gateway.core.kinds import Kind
+ from beaver_gateway.frontends.base import GatewayRuntime
+ from beaver_gateway.storage.models import Conversation, ConversationBinding
+
+__all__ = ["FRONTEND", "Attachments", "TelegramFrontend"]
+
+_log = logging.getLogger("beaver_gateway.frontends.telegram")
+
+FRONTEND = "telegram"
+_DONE = "done"
+_COMMANDS = ("merge", "new", "chat", "status", "start", "help")
+_HELP = (
+ "General - мастер, топик - ветка. Создай топик и пиши в него.\n"
+ "/merge - слить ветку в мастер\n"
+ "/new [название] - новая ветка (в General - новый топик)\n"
+ "/chat <тема> - открыть глубокий чат в vault\n"
+ "/status - что с этим разговором" # noqa: RUF001
+)
+
+
+@dataclass(frozen=True, slots=True)
+class Attachments:
+ """Where files from Telegram go.
+
+ ``ephemeral`` - under the gateway's data dir, swept after ``keep_days``;
+ ``vault`` - into a directory the agent owns. Pulling what is worth
+ keeping into the vault is the agent's job either way.
+ """
+
+ mode: Literal["ephemeral", "vault"] = "ephemeral"
+ dir: Path | None = None
+ keep_days: int = 7
+
+ @property
+ def root(self) -> Path:
+ if self.dir is not None:
+ return self.dir
+ if self.mode == "vault":
+ msg = "Attachments(mode='vault') needs `dir`"
+ raise ValueError(msg)
+ return Path(tempfile.gettempdir()) / "beaver-attachments"
+
+
+EPHEMERAL = Attachments()
+
+
+@dataclass
+class _Ask:
+ conversation_id: str
+ chat_id: int
+ thread_id: int | None
+ questions: list[dict[str, Any]]
+ messages: list[int] = field(default_factory=list)
+ picked: dict[int, list[str]] = field(default_factory=dict)
+ done: set[int] = field(default_factory=set)
+
+
+class TelegramFrontend(Frontend):
+ name = FRONTEND
+ kinds = ("master", "branch")
+
+ def __init__(
+ self,
+ *,
+ token: str,
+ user_id: int,
+ master_agent: str | None = None,
+ branch_agent: str | None = None,
+ chat_id: int | None = None,
+ attachments: Attachments = EPHEMERAL,
+ draft_interval: float = 0.7,
+ queued_reaction: str = "👀",
+ poll_timeout: int = 30,
+ outbox_backoff: float = 2.0,
+ ) -> None:
+ self._token = token
+ self.user_id = user_id
+ self.chat_id = chat_id if chat_id is not None else user_id
+ self.master_agent = master_agent
+ self.branch_agent = branch_agent
+ self.attachments = attachments
+ self.draft_interval = draft_interval
+ self.queued_reaction = queued_reaction
+ self.poll_timeout = poll_timeout
+ self.outbox_backoff = outbox_backoff
+ self._runtime: GatewayRuntime | None = None
+ self._bot: Bot | None = None
+ self._inbox: Inbox | None = None
+ self._outbox: Outbox | None = None
+ self._targets: dict[str, tuple[int, int | None] | None] = {}
+ self._topic_names: dict[int, str] = {}
+ self._drafts: dict[str, Draft] = {}
+ self._asks: dict[str, _Ask] = {}
+ self._reactions: dict[int, tuple[int, int]] = {}
+ self._tasks: set[asyncio.Task[None]] = set()
+
+ # ---- Frontend --------------------------------------------------------
+
+ def agent_for(self, kind: Kind) -> str | None:
+ return {"master": self.master_agent, "branch": self.branch_agent}.get(kind)
+
+ def configure(self, runtime: GatewayRuntime) -> None:
+ if runtime.conversations is None or runtime.bus is None:
+ msg = "TelegramFrontend needs runtime.conversations and runtime.bus"
+ raise RuntimeError(msg)
+ self._runtime = runtime
+ if self._bot is None:
+ self._bot = Bot(self._token, default=DefaultBotProperties(parse_mode=None))
+ self._inbox = Inbox(
+ runtime.db, self._bot, handler=self._handle, poll_timeout=self.poll_timeout
+ )
+ self._outbox = Outbox(
+ runtime.db, self._bot, bus=runtime.bus, backoff=self.outbox_backoff
+ )
+
+ async def serve(self) -> None:
+ me = await self.bot.get_me()
+ _log.info(
+ "telegram: @%s, user %s, chat %s, topics=%s",
+ me.username,
+ self.user_id,
+ self.chat_id,
+ getattr(me, "has_topics_enabled", None),
+ )
+ self._sweep_attachments()
+ try:
+ async with asyncio.TaskGroup() as tg:
+ tg.create_task(self.inbox.run())
+ tg.create_task(self.outbox.run())
+ tg.create_task(self._events())
+ finally:
+ for draft in list(self._drafts.values()):
+ await draft.stop()
+ await self.bot.session.close()
+
+ async def materialize(self, conv: Conversation) -> ConversationBinding | None:
+ if conv.kind == "master":
+ return await self.conversations.bind(
+ conv, frontend=FRONTEND, external_id=self._ext(None)
+ )
+ if conv.kind != "branch":
+ return None
+ topic = await self.bot.create_forum_topic(
+ self.chat_id, name=(conv.title or "ветка")[:128]
+ )
+ self._topic_names[topic.message_thread_id] = topic.name
+ return await self.conversations.bind(
+ conv, frontend=FRONTEND, external_id=self._ext(topic.message_thread_id)
+ )
+
+ async def mark_topic(self, conv: Conversation, prefix: str = "✅ ") -> bool:
+ """Rotation hook for M3.
+
+ ``closeForumTopic`` does not exist in private chats; the state of a
+ merged or closed branch lives in its name.
+ """
+ target = await self._target_of(conv)
+ if target is None or target[1] is None:
+ return False
+ name = self._topic_names.get(target[1]) or conv.title or "ветка"
+ if name.startswith(prefix):
+ return True
+ await self.bot.edit_forum_topic(
+ target[0], target[1], name=f"{prefix}{name}"[:128]
+ )
+ self._topic_names[target[1]] = f"{prefix}{name}"
+ return True
+
+ # ---- plumbing --------------------------------------------------------
+
+ @property
+ def bot(self) -> Bot:
+ if self._bot is None:
+ msg = "configure() must be called before use"
+ raise RuntimeError(msg)
+ return self._bot
+
+ @property
+ def inbox(self) -> Inbox:
+ return cast("Inbox", self._inbox)
+
+ @property
+ def outbox(self) -> Outbox:
+ return cast("Outbox", self._outbox)
+
+ @property
+ def conversations(self) -> Conversations:
+ return cast(
+ "Conversations", cast("GatewayRuntime", self._runtime).conversations
+ )
+
+ @property
+ def bus(self) -> EventBus:
+ return cast("EventBus", cast("GatewayRuntime", self._runtime).bus)
+
+ def _ext(self, thread_id: int | None) -> str:
+ return f"{self.chat_id}/{thread_id}" if thread_id else str(self.chat_id)
+
+ @staticmethod
+ def _parse_ext(ext: str) -> tuple[int, int | None]:
+ chat, _, thread = ext.partition("/")
+ return int(chat), int(thread) if thread else None
+
+ async def _target_of(self, conv: Conversation) -> tuple[int, int | None] | None:
+ key = conv.external_id
+ if key not in self._targets:
+ bound = next(
+ (
+ b
+ for b in await self.conversations.bindings(conv)
+ if b.frontend == FRONTEND and b.visible
+ ),
+ None,
+ )
+ self._targets[key] = self._parse_ext(bound.external_id) if bound else None
+ return self._targets[key]
+
+ async def _deliver(
+ self,
+ conv: Conversation,
+ text: str,
+ *,
+ turn_id: str | None = None,
+ key: str | None = None,
+ ) -> None:
+ target = await self._target_of(conv)
+ if target is None or not text.strip():
+ return
+ await self.outbox.enqueue(
+ chat_id=target[0],
+ thread_id=target[1],
+ text=text,
+ conversation_id=conv.id,
+ turn_id=turn_id,
+ dedupe_key=key,
+ )
+
+ async def _master(self) -> Conversation:
+ ext = self._ext(None)
+ conv = await self.conversations.find_bound(frontend=FRONTEND, external_id=ext)
+ if conv is not None and conv.status == "open":
+ return conv
+ masters = await self.conversations.find(kind="master", status="open", limit=1)
+ if masters:
+ await self.conversations.bind(
+ masters[0], frontend=FRONTEND, external_id=ext
+ )
+ return masters[0]
+ return await self.conversations.spawn(
+ kind="master", seed="clean", origin=FRONTEND, binding=(FRONTEND, ext)
+ )
+
+ async def _branch(
+ self, thread_id: int, *, title: str | None, text: str | None
+ ) -> Conversation:
+ master = await self._master()
+ return await self.conversations.spawn(
+ kind="branch",
+ seed="morning",
+ parent=master,
+ title=(title or self._topic_names.get(thread_id) or "ветка")[:128],
+ text=text,
+ origin=FRONTEND,
+ binding=(FRONTEND, self._ext(thread_id)),
+ )
+
+ # ---- inbox -----------------------------------------------------------
+
+ async def _handle(self, update: Update) -> None:
+ if update.message is not None:
+ await self._on_message(update.message)
+ elif update.callback_query is not None:
+ await self._on_callback(update.callback_query)
+
+ async def _on_message(self, message: Message) -> None:
+ if message.forum_topic_created is not None and message.message_thread_id:
+ self._topic_names[message.message_thread_id] = (
+ message.forum_topic_created.name
+ )
+ if message.from_user is None or message.from_user.id != self.user_id:
+ _log.warning(
+ "ignoring message from %s", getattr(message.from_user, "id", None)
+ )
+ return
+ if message.chat.id != self.chat_id:
+ return
+ in_topic = message.is_topic_message or message.forum_topic_created is not None
+ thread_id = message.message_thread_id if in_topic else None
+ if message.forum_topic_created is not None and thread_id is not None:
+ if await self._live(thread_id) is None:
+ await self._branch(
+ thread_id, title=message.forum_topic_created.name, text=None
+ )
+ return
+ if message.forum_topic_edited is not None and thread_id is not None:
+ if message.forum_topic_edited.name:
+ self._topic_names[thread_id] = message.forum_topic_edited.name
+ return
+ text = (message.text or message.caption or "").strip()
+ attachment = await self._save_attachment(message, thread_id)
+ if attachment:
+ text = f"{text}\n\n{attachment}".strip()
+ if not text:
+ return
+ if message.text and message.text.startswith("/"):
+ command, _, args = message.text[1:].partition(" ")
+ command = command.partition("@")[0].lower()
+ if command in _COMMANDS:
+ await self._command(command, args.strip(), message, thread_id)
+ return
+ if thread_id is None:
+ conv = await self._master()
+ else:
+ conv = await self._live(thread_id)
+ if conv is None:
+ await self._branch(thread_id, title=self._title_from(text), text=text)
+ return
+ pending = self.conversations.pending_question(conv.external_id)
+ if pending is not None and self.conversations.answer(pending[0], text):
+ await self._close_ask(pending[0], f"✍️ {text}")
+ return
+ item = await self.conversations.post(conv, text, origin=FRONTEND)
+ if conv.running_turn or conv.pending_question:
+ await self._react(message, item.id)
+
+ async def _live(self, thread_id: int) -> Conversation | None:
+ conv = await self.conversations.find_bound(
+ frontend=FRONTEND, external_id=self._ext(thread_id)
+ )
+ return conv if conv is not None and conv.status == "open" else None
+
+ @staticmethod
+ def _title_from(text: str) -> str:
+ line = text.strip().splitlines()[0]
+ return line if len(line) <= 60 else line[:57] + "…"
+
+ async def _react(self, message: Message, item_id: int | None) -> None:
+ if not self.queued_reaction or item_id is None:
+ return
+ try:
+ await self.bot.set_message_reaction(
+ message.chat.id,
+ message.message_id,
+ reaction=[ReactionTypeEmoji(emoji=self.queued_reaction)],
+ )
+ except TelegramAPIError as exc:
+ _log.debug("reaction failed: %s", exc)
+ return
+ self._reactions[item_id] = (message.chat.id, message.message_id)
+
+ async def _unreact(self, item_id: int | None) -> None:
+ target = self._reactions.pop(cast("int", item_id), None) if item_id else None
+ if target is None:
+ return
+ with contextlib.suppress(TelegramAPIError):
+ await self.bot.set_message_reaction(target[0], target[1], reaction=[])
+
+ async def _save_attachment(
+ self, message: Message, thread_id: int | None
+ ) -> str | None:
+ file_id: str | None = None
+ name: str | None = None
+ kind = ""
+ size: int | None = None
+ if message.photo:
+ photo = message.photo[-1]
+ file_id, kind, size = photo.file_id, "фото", photo.file_size
+ name = f"{photo.file_unique_id}.jpg"
+ elif message.document:
+ doc = message.document
+ file_id, kind, size = doc.file_id, "файл", doc.file_size
+ name = doc.file_name or f"{doc.file_unique_id}.bin"
+ elif message.voice:
+ file_id, kind, size = (
+ message.voice.file_id,
+ "голосовое",
+ message.voice.file_size,
+ )
+ name = f"{message.voice.file_unique_id}.ogg"
+ elif message.audio:
+ file_id, kind, size = (
+ message.audio.file_id,
+ "аудио",
+ message.audio.file_size,
+ )
+ name = message.audio.file_name or f"{message.audio.file_unique_id}.mp3"
+ elif message.video:
+ file_id, kind, size = (
+ message.video.file_id,
+ "видео",
+ message.video.file_size,
+ )
+ name = message.video.file_name or f"{message.video.file_unique_id}.mp4"
+ elif message.video_note:
+ file_id, kind, size = (
+ message.video_note.file_id,
+ "кружок",
+ message.video_note.file_size,
+ )
+ name = f"{message.video_note.file_unique_id}.mp4"
+ if file_id is None or name is None:
+ return None
+ if size and size > 20 * 1024 * 1024:
+ return (
+ f"[вложение: {kind} {name}, {size // 1024 // 1024} МБ - "
+ "больше 20 МБ, Telegram не отдаёт ботам]"
+ )
+ folder = self.attachments.root / (self._ext(thread_id).replace("/", "_"))
+ folder.mkdir(parents=True, exist_ok=True)
+ with contextlib.suppress(OSError):
+ self.attachments.root.chmod(0o755)
+ folder.chmod(0o755)
+ path = folder / f"{int(time.time())}-{Path(name).name}"
+ try:
+ await self.bot.download(file_id, destination=path)
+ path.chmod(0o644)
+ except (TelegramAPIError, OSError) as exc:
+ _log.warning("attachment download failed: %s", exc)
+ return f"[вложение: {kind} {name} - не скачалось: {exc}]"
+ shown = f", {size // 1024} КБ" if size else ""
+ return f"[вложение: {kind} {path}{shown}]"
+
+ def _sweep_attachments(self) -> None:
+ if self.attachments.mode != "ephemeral":
+ return
+ root = self.attachments.root
+ if not root.exists():
+ return
+ cutoff = time.time() - self.attachments.keep_days * 86400
+ for path in root.rglob("*"):
+ with contextlib.suppress(OSError):
+ if path.is_file() and path.stat().st_mtime < cutoff:
+ path.unlink()
+
+ # ---- commands --------------------------------------------------------
+
+ async def _command(
+ self, command: str, args: str, message: Message, thread_id: int | None
+ ) -> None:
+ reply = await self._run_command(command, args, thread_id)
+ if reply:
+ await self.outbox.enqueue(
+ chat_id=message.chat.id, thread_id=thread_id, text=reply
+ )
+
+ async def _run_command(self, command: str, args: str, thread_id: int | None) -> str:
+ if command in ("start", "help"):
+ return _HELP
+ conv = (
+ await self._master() if thread_id is None else await self._live(thread_id)
+ )
+ if command == "status":
+ return await self._status(conv)
+ if command == "merge":
+ if conv is None or conv.kind != "branch":
+ return "сливать нечего: это не открытая ветка"
+ self._spawn_task(self._merge(conv))
+ return "🔀 сливаю в мастер…"
+ if command == "new":
+ if thread_id is None:
+ child = await self.conversations.spawn(
+ kind="branch",
+ seed="morning",
+ parent=conv,
+ title=args or None,
+ origin=FRONTEND,
+ )
+ return f"🌿 ветка «{child.title or child.external_id}» - в новом топике"
+ if conv is not None:
+ await self.conversations.set_status(conv, "closed")
+ await self._branch(thread_id, title=args or None, text=None)
+ return "🌿 новая ветка на этом топике"
+ if command == "chat":
+ if not args:
+ return "/chat <тема>"
+ try:
+ deep = await self.conversations.spawn(
+ kind="deep", seed="clean", title=args, origin=FRONTEND
+ )
+ except (ValueError, LookupError) as exc:
+ return f"не вышло: {exc}"
+ where = next(
+ (
+ b.external_id
+ for b in await self.conversations.bindings(deep)
+ if b.visible
+ ),
+ deep.external_id,
+ )
+ return f"💬 глубокий чат: {where}"
+ return _HELP
+
+ async def _status(self, conv: Conversation | None) -> str:
+ if conv is None:
+ return "этот топик ни к чему не привязан - напиши, и откроется ветка"
+ info = await self.conversations.describe(conv)
+ queued = sum(
+ 1
+ for i in await self.conversations.queue.recent(
+ cast("int", conv.id), limit=20
+ )
+ if i.status == "queued"
+ )
+ lines = [
+ f"{conv.kind} · {conv.status} · {conv.agent_name}",
+ f"сессия: {'живая' if info['live'] else 'нет'}"
+ f" · тёрн: {'идёт' if conv.running_turn else 'нет'}"
+ f" · в очереди: {queued}",
+ ]
+ if conv.pending_question:
+ lines.append("❓ ждёт ответа на вопрос")
+ if conv.kind == "master":
+ pool = self.conversations.pool
+ lines.append(f"пул: {len(pool)} сессий, rss {pool.rss() // (1 << 20)} МБ")
+ lines.append(f"outbox: {await self.outbox.pending()} в очереди")
+ lines.append(f"id: {conv.external_id}")
+ return "\n".join(lines)
+
+ async def _merge(self, conv: Conversation) -> None:
+ try:
+ await self.conversations.merge(conv)
+ except Exception as exc: # noqa: BLE001
+ _log.exception("merge of %s failed", conv.external_id)
+ await self._deliver(conv, f"⚠️ слив не удался: {exc}")
+
+ def _spawn_task(self, coro: Any) -> None:
+ task = asyncio.create_task(coro)
+ self._tasks.add(task)
+ task.add_done_callback(self._tasks.discard)
+
+ # ---- bus -------------------------------------------------------------
+
+ async def _events(self) -> None:
+ async for event in self.bus.stream():
+ try:
+ await self._on_event(event)
+ except Exception: # noqa: BLE001
+ _log.exception("event %s failed", event.get("type"))
+
+ async def _on_event(self, event: Event) -> None:
+ kind = event["type"]
+ if kind == "conversation.bound":
+ self._targets.pop(str(event.get("conversation_id")), None)
+ return
+ if kind in ("delivery.sent", "delivery.failed", "conversation.created"):
+ return
+ key = event.get("conversation_id")
+ if not isinstance(key, str):
+ return
+ conv = await self.conversations.get(key)
+ if conv is None:
+ return
+ target = await self._target_of(conv)
+ if target is None:
+ return
+ match kind:
+ case "turn.start":
+ if event.get("origin") == "user":
+ await self._open_draft(key, event, target)
+ case "stream":
+ self._feed_draft(key, event)
+ case "tool":
+ draft = self._drafts.get(key)
+ if draft is not None and event.get("parent_tool_use_id") is None:
+ draft.set_status(
+ f"⏳ {status_label(str(event['name']), event.get('input'))}"
+ )
+ case "turn.end":
+ await self._close_draft(key)
+ if event.get("stop") == "error" and event.get("origin") == "user":
+ await self._deliver(
+ conv,
+ "⚠️ тёрн упал, смотри логи gateway",
+ turn_id=event.get("turn_id"),
+ key=f"{event.get('turn_id')}:error",
+ )
+ case "reply":
+ await self._on_reply(conv, event)
+ case "say":
+ await self._deliver(
+ conv, str(event.get("text") or ""), turn_id=event.get("turn_id")
+ )
+ case "question":
+ await self._ask(conv, event, target)
+ case "question.answered":
+ await self._close_ask(
+ str(event["question_id"]), f"✅ {event.get('answer') or ''}"
+ )
+ case "question.timeout":
+ await self._close_ask(
+ str(event["question_id"]), "⌛ время вышло - ответь текстом"
+ )
+ case "conversation.merged":
+ await self._deliver(conv, "✅ слито в мастер", key=f"{key}:merged")
+
+ async def _on_reply(self, conv: Conversation, event: Event) -> None:
+ turn_id = str(event.get("turn_id") or "")
+ origin = str(event.get("item_origin") or "")
+ await self._unreact(event.get("item"))
+ if origin != FRONTEND and not origin.startswith("сид"):
+ user_text = str(event.get("user_text") or "")
+ if user_text:
+ await self._deliver(
+ conv,
+ f"📝 из панели:\n{user_text}",
+ turn_id=turn_id,
+ key=f"{turn_id}:mirror",
+ )
+ await self._deliver(
+ conv, str(event.get("text") or ""), turn_id=turn_id, key=f"{turn_id}:reply"
+ )
+
+ # ---- drafts ------------------------------------------------------------
+
+ async def _open_draft(
+ self, key: str, event: Event, target: tuple[int, int | None]
+ ) -> None:
+ if target[0] < 0:
+ return
+ await self._close_draft(key)
+ draft = Draft(
+ self.bot,
+ chat_id=target[0],
+ thread_id=target[1],
+ turn_id=str(event.get("turn_id") or key),
+ interval=self.draft_interval,
+ )
+ self._drafts[key] = draft
+ draft.start()
+
+ def _feed_draft(self, key: str, event: Event) -> None:
+ draft = self._drafts.get(key)
+ if draft is None or event.get("parent_tool_use_id") is not None:
+ return
+ raw = event.get("event") or {}
+ kind = raw.get("type")
+ if kind == "content_block_delta":
+ delta = raw.get("delta") or {}
+ if delta.get("type") == "text_delta":
+ draft.set_status("✍️ пишу…")
+ draft.append(str(delta.get("text") or ""))
+ elif delta.get("type") == "thinking_delta":
+ draft.set_status("🤔 думаю…")
+ elif kind == "content_block_start":
+ block = raw.get("content_block") or {}
+ if block.get("type") == "tool_use":
+ draft.set_status(
+ f"⏳ {status_label(str(block.get('name') or ''), None)}"
+ )
+
+ async def _close_draft(self, key: str) -> None:
+ draft = self._drafts.pop(key, None)
+ if draft is not None:
+ await draft.stop()
+
+ # ---- questions (§3.7) ---------------------------------------------------
+
+ async def _ask(
+ self, conv: Conversation, event: Event, target: tuple[int, int | None]
+ ) -> None:
+ question_id = str(event["question_id"])
+ questions = [q for q in event.get("questions") or [] if isinstance(q, dict)]
+ if not questions:
+ return
+ ask = _Ask(
+ conversation_id=conv.external_id,
+ chat_id=target[0],
+ thread_id=target[1],
+ questions=questions,
+ )
+ self._asks[question_id] = ask
+ for qi, question in enumerate(questions):
+ try:
+ sent = await self.bot.send_message(
+ target[0],
+ _question_html(question),
+ message_thread_id=target[1],
+ parse_mode="HTML",
+ reply_markup=_keyboard(question_id, qi, question, []),
+ )
+ except TelegramAPIError:
+ _log.exception("question %s could not be sent", question_id)
+ continue
+ ask.messages.append(sent.message_id)
+
+ async def _on_callback(self, query: CallbackQuery) -> None:
+ if query.from_user.id != self.user_id or not query.data:
+ return
+ parts = query.data.split(":")
+ if len(parts) != 4 or parts[0] != "q":
+ return
+ _, question_id, qi_raw, choice = parts
+ ask = self._asks.get(question_id)
+ if ask is None or not qi_raw.isdigit():
+ await self._callback_reply(query, "вопрос уже закрыт")
+ return
+ qi = int(qi_raw)
+ question = ask.questions[qi]
+ options = [str(o.get("label", "")) for o in question.get("options") or []]
+ picked = ask.picked.setdefault(qi, [])
+ multi = bool(question.get("multiSelect"))
+ if choice == _DONE:
+ ask.done.add(qi)
+ elif choice.isdigit() and int(choice) < len(options):
+ label = options[int(choice)]
+ if multi:
+ if label in picked:
+ picked.remove(label)
+ else:
+ picked.append(label)
+ else:
+ picked[:] = [label]
+ ask.done.add(qi)
+ message = query.message if isinstance(query.message, Message) else None
+ if message is not None:
+ with contextlib.suppress(TelegramAPIError):
+ if qi in ask.done:
+ await self.bot.edit_message_text(
+ f"{_question_html(question)}\n\n"
+ f"✅ {html.escape(', '.join(picked) or '-')}",
+ chat_id=message.chat.id,
+ message_id=message.message_id,
+ parse_mode="HTML",
+ )
+ else:
+ await self.bot.edit_message_reply_markup(
+ chat_id=message.chat.id,
+ message_id=message.message_id,
+ reply_markup=_keyboard(question_id, qi, question, picked),
+ )
+ await self._callback_reply(query, None)
+ if len(ask.done) == len(ask.questions):
+ answer = _answer_text(ask)
+ self._asks.pop(question_id, None)
+ if not self.conversations.answer(question_id, answer):
+ await self._edit_asks(ask, "⌛ время вышло - ответь текстом")
+
+ async def _callback_reply(self, query: CallbackQuery, text: str | None) -> None:
+ with contextlib.suppress(TelegramAPIError):
+ await self.bot.answer_callback_query(query.id, text=text)
+
+ async def _close_ask(self, question_id: str, note: str) -> None:
+ ask = self._asks.pop(question_id, None)
+ if ask is not None:
+ await self._edit_asks(ask, note)
+
+ async def _edit_asks(self, ask: _Ask, note: str) -> None:
+ for qi, message_id in enumerate(ask.messages):
+ if qi in ask.done and not note.startswith("⌛"):
+ continue
+ with contextlib.suppress(TelegramAPIError):
+ await self.bot.edit_message_text(
+ f"{_question_html(ask.questions[qi])}\n\n{html.escape(note)}",
+ chat_id=ask.chat_id,
+ message_id=message_id,
+ parse_mode="HTML",
+ )
+
+
+def _question_html(question: dict[str, Any]) -> str:
+ header = html.escape(str(question.get("header") or "").strip())
+ body = html.escape(str(question.get("question") or "").strip())
+ lines = [f"❓ {header}" if header else "❓", body]
+ for option in question.get("options") or []:
+ label = html.escape(str(option.get("label", "")))
+ description = html.escape(str(option.get("description") or "").strip())
+ lines.append(f"• {label}" + (f" - {description}" if description else ""))
+ return "\n".join(line for line in lines if line)
+
+
+def _keyboard(
+ question_id: str, qi: int, question: dict[str, Any], picked: list[str]
+) -> InlineKeyboardMarkup:
+ rows = [
+ [
+ InlineKeyboardButton(
+ text=("☑ " if str(o.get("label", "")) in picked else "")
+ + str(o.get("label", ""))[:60],
+ callback_data=f"q:{question_id}:{qi}:{oi}",
+ )
+ ]
+ for oi, o in enumerate(question.get("options") or [])
+ ]
+ if question.get("multiSelect"):
+ rows.append(
+ [
+ InlineKeyboardButton(
+ text="✅ готово", callback_data=f"q:{question_id}:{qi}:{_DONE}"
+ )
+ ]
+ )
+ return InlineKeyboardMarkup(inline_keyboard=rows)
+
+
+def _answer_text(ask: _Ask) -> str:
+ if len(ask.questions) == 1:
+ return ", ".join(ask.picked.get(0, [])) or "-"
+ return "; ".join(
+ f"{q.get('header') or q.get('question')}: "
+ f"{', '.join(ask.picked.get(i, [])) or '-'}"
+ for i, q in enumerate(ask.questions)
+ )
diff --git a/src/beaver_gateway/frontends/telegram/inbox.py b/src/beaver_gateway/frontends/telegram/inbox.py
new file mode 100644
index 0000000..2a1141a
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/inbox.py
@@ -0,0 +1,138 @@
+"""Long-polling inbox (§3.8).
+
+Every update lands in ``telegram_updates`` before the offset moves past it;
+a worker handles rows from the table, oldest first, and finishes whatever a
+previous process left unprocessed at startup.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from aiogram.exceptions import TelegramConflictError, TelegramNetworkError
+from aiogram.types import Update
+from sqlalchemy import func
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import col, select
+
+from beaver_gateway.storage.models import TelegramUpdate
+
+if TYPE_CHECKING:
+ from collections.abc import Awaitable, Callable
+
+ from aiogram import Bot
+
+ from beaver_gateway.storage.db import Database
+
+__all__ = ["Inbox"]
+
+_log = logging.getLogger("beaver_gateway.frontends.telegram.inbox")
+
+ALLOWED_UPDATES = ("message", "callback_query")
+
+
+class Inbox:
+ def __init__(
+ self,
+ db: Database,
+ bot: Bot,
+ *,
+ handler: Callable[[Update], Awaitable[None]],
+ poll_timeout: int = 30,
+ ) -> None:
+ self._db = db
+ self._bot = bot
+ self._handler = handler
+ self._poll_timeout = poll_timeout
+ self._wake = asyncio.Event()
+
+ async def run(self) -> None:
+ async with asyncio.TaskGroup() as tg:
+ tg.create_task(self._poll())
+ tg.create_task(self._work())
+
+ async def _poll(self) -> None:
+ offset = await self._next_offset()
+ backoff = 1.0
+ while True:
+ try:
+ updates = await self._bot.get_updates(
+ offset=offset,
+ timeout=self._poll_timeout,
+ allowed_updates=list(ALLOWED_UPDATES),
+ request_timeout=self._poll_timeout + 10,
+ )
+ except TelegramConflictError:
+ _log.error("another poller holds this bot token; retrying in 10s")
+ await asyncio.sleep(10)
+ continue
+ except (TimeoutError, TelegramNetworkError, OSError) as exc:
+ _log.warning("getUpdates failed (%s); retrying in %.0fs", exc, backoff)
+ await asyncio.sleep(backoff)
+ backoff = min(backoff * 2, 60.0)
+ continue
+ backoff = 1.0
+ for update in updates:
+ await self._store(update)
+ offset = update.update_id + 1
+ if updates:
+ self._wake.set()
+
+ async def _store(self, update: Update) -> None:
+ row = TelegramUpdate(
+ update_id=update.update_id,
+ payload=update.model_dump(mode="json", by_alias=True, exclude_none=True),
+ )
+ async with self._db.session() as session:
+ session.add(row)
+ try:
+ await session.commit()
+ except IntegrityError:
+ await session.rollback()
+
+ async def _next_offset(self) -> int | None:
+ async with self._db.session() as session:
+ latest = (
+ await session.exec(select(func.max(col(TelegramUpdate.update_id))))
+ ).one()
+ return int(latest) + 1 if latest is not None else None
+
+ async def _work(self) -> None:
+ while True:
+ rows = await self._pending()
+ if not rows:
+ self._wake.clear()
+ with contextlib.suppress(TimeoutError):
+ await asyncio.wait_for(self._wake.wait(), timeout=5.0)
+ continue
+ for row in rows:
+ await self._handle(row)
+
+ async def _pending(self, limit: int = 50) -> list[TelegramUpdate]:
+ async with self._db.session() as session:
+ result = await session.exec(
+ select(TelegramUpdate)
+ .where(col(TelegramUpdate.processed_at).is_(None))
+ .order_by(col(TelegramUpdate.update_id))
+ .limit(limit)
+ )
+ return list(result.all())
+
+ async def _handle(self, row: TelegramUpdate) -> None:
+ error: str | None = None
+ try:
+ await self._handler(Update.model_validate(row.payload))
+ except Exception as exc: # noqa: BLE001
+ error = f"{type(exc).__name__}: {exc}"[:500]
+ _log.exception("update %s failed", row.update_id)
+ async with self._db.session() as session:
+ stored = await session.get(TelegramUpdate, row.update_id)
+ if stored is not None:
+ stored.processed_at = datetime.now(UTC)
+ stored.error = error
+ session.add(stored)
+ await session.commit()
diff --git a/src/beaver_gateway/frontends/telegram/outbox.py b/src/beaver_gateway/frontends/telegram/outbox.py
new file mode 100644
index 0000000..d95cb04
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/outbox.py
@@ -0,0 +1,235 @@
+"""Outbox (§3.8): a reply is a ``deliveries`` row first, a message second.
+
+Rows are sent oldest first, retried with backoff on network errors and
+flood limits, resent as plain text when Telegram rejects our HTML, and
+given up only when Telegram says the window is gone.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
+from aiogram.exceptions import (
+ TelegramBadRequest,
+ TelegramForbiddenError,
+ TelegramNetworkError,
+ TelegramNotFound,
+ TelegramRetryAfter,
+ TelegramServerError,
+)
+from aiogram.types import LinkPreviewOptions
+from sqlalchemy import func
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import col, select
+
+from beaver_gateway.frontends.telegram.render import chunks, to_html
+from beaver_gateway.storage.models import Delivery
+
+if TYPE_CHECKING:
+ from aiogram import Bot
+
+ from beaver_gateway.core.bus import EventBus
+ from beaver_gateway.storage.db import Database
+
+__all__ = ["Outbox"]
+
+_log = logging.getLogger("beaver_gateway.frontends.telegram.outbox")
+
+_MAX_BACKOFF = 300.0
+_GONE = ("thread not found", "chat not found", "topic_deleted", "topic_closed")
+_NO_PREVIEW = LinkPreviewOptions(is_disabled=True)
+
+
+class Outbox:
+ def __init__(
+ self, db: Database, bot: Bot, *, bus: EventBus, backoff: float = 2.0
+ ) -> None:
+ self._db = db
+ self._bot = bot
+ self._bus = bus
+ self._backoff = backoff
+ self._wake = asyncio.Event()
+
+ async def enqueue(
+ self,
+ *,
+ chat_id: int,
+ thread_id: int | None,
+ text: str,
+ conversation_id: int | None = None,
+ turn_id: str | None = None,
+ dedupe_key: str | None = None,
+ ) -> list[Delivery]:
+ rows: list[Delivery] = []
+ for n, part in enumerate(chunks(text)):
+ row = Delivery(
+ conversation_id=conversation_id,
+ chat_id=chat_id,
+ thread_id=thread_id,
+ text=part,
+ turn_id=turn_id,
+ dedupe_key=f"{dedupe_key}:{n}" if dedupe_key else None,
+ )
+ async with self._db.session() as session:
+ session.add(row)
+ try:
+ await session.commit()
+ except IntegrityError:
+ await session.rollback()
+ continue
+ await session.refresh(row)
+ rows.append(row)
+ if rows:
+ self._wake.set()
+ return rows
+
+ async def run(self) -> None:
+ while True:
+ rows = await self._due()
+ if not rows:
+ self._wake.clear()
+ with contextlib.suppress(TimeoutError):
+ await asyncio.wait_for(
+ self._wake.wait(), timeout=await self._wait_for_next()
+ )
+ continue
+ for row in rows:
+ await self._send(row)
+
+ async def _wait_for_next(self, cap: float = 5.0) -> float:
+ async with self._db.session() as session:
+ earliest = (
+ await session.exec(
+ select(func.min(col(Delivery.next_attempt_at))).where(
+ Delivery.status == "queued"
+ )
+ )
+ ).one()
+ if earliest is None:
+ return cap
+ now = datetime.now(UTC).replace(tzinfo=None)
+ return max(0.05, min(cap, (earliest - now).total_seconds()))
+
+ async def _due(self, limit: int = 50) -> list[Delivery]:
+ now = datetime.now(UTC).replace(tzinfo=None)
+ async with self._db.session() as session:
+ result = await session.exec(
+ select(Delivery)
+ .where(
+ Delivery.status == "queued", col(Delivery.next_attempt_at) <= now
+ )
+ .order_by(col(Delivery.id))
+ .limit(limit)
+ )
+ return list(result.all())
+
+ async def _send(self, row: Delivery) -> None:
+ try:
+ message = await self._bot.send_message(
+ row.chat_id,
+ row.text if row.plain else to_html(row.text),
+ message_thread_id=row.thread_id,
+ parse_mode=None if row.plain else "HTML",
+ link_preview_options=_NO_PREVIEW,
+ )
+ except TelegramRetryAfter as exc:
+ await self._retry(row, str(exc), delay=float(exc.retry_after))
+ except TelegramBadRequest as exc:
+ text = str(exc).lower()
+ if "parse" in text and not row.plain:
+ await self._retry(row, str(exc), delay=0.0, plain=True)
+ elif any(marker in text for marker in _GONE):
+ await self._fail(row, str(exc))
+ else:
+ await self._fail(row, str(exc))
+ except (TelegramNotFound, TelegramForbiddenError) as exc:
+ await self._fail(row, str(exc))
+ except (TelegramNetworkError, TelegramServerError, OSError) as exc:
+ await self._retry(
+ row,
+ str(exc),
+ delay=min(self._backoff ** (row.attempts + 1), _MAX_BACKOFF),
+ )
+ else:
+ await self._mark(row, status="sent", message_id=message.message_id)
+ self._bus.publish(
+ "delivery.sent",
+ delivery=row.id,
+ conversation_row=row.conversation_id,
+ chat_id=row.chat_id,
+ thread_id=row.thread_id,
+ message_id=message.message_id,
+ turn_id=row.turn_id,
+ )
+
+ async def _retry(
+ self, row: Delivery, error: str, *, delay: float, plain: bool = False
+ ) -> None:
+ _log.warning(
+ "delivery #%s attempt %d failed: %s (retry in %.0fs)",
+ row.id,
+ row.attempts + 1,
+ error,
+ delay,
+ )
+ await self._mark(row, status="queued", error=error, delay=delay, plain=plain)
+ if delay < 5.0:
+ self._wake.set()
+
+ async def _fail(self, row: Delivery, error: str) -> None:
+ _log.error(
+ "delivery #%s to %s/%s given up: %s",
+ row.id,
+ row.chat_id,
+ row.thread_id,
+ error,
+ )
+ await self._mark(row, status="failed", error=error)
+ self._bus.publish(
+ "delivery.failed",
+ delivery=row.id,
+ conversation_row=row.conversation_id,
+ chat_id=row.chat_id,
+ thread_id=row.thread_id,
+ error=error,
+ )
+
+ async def _mark(
+ self,
+ row: Delivery,
+ *,
+ status: str,
+ error: str | None = None,
+ delay: float = 0.0,
+ plain: bool = False,
+ message_id: int | None = None,
+ ) -> None:
+ async with self._db.session() as session:
+ stored = await session.get(Delivery, row.id)
+ if stored is None:
+ return
+ stored.status = status
+ stored.attempts += 1
+ stored.last_error = error[:500] if error else None
+ stored.next_attempt_at = (
+ datetime.now(UTC) + timedelta(seconds=delay)
+ ).replace(tzinfo=None)
+ if plain:
+ stored.plain = True
+ if message_id is not None:
+ stored.message_id = message_id
+ if status == "sent":
+ stored.sent_at = datetime.now(UTC)
+ session.add(stored)
+ await session.commit()
+
+ async def pending(self) -> int:
+ async with self._db.session() as session:
+ result = await session.exec(
+ select(Delivery).where(Delivery.status == "queued")
+ )
+ return len(list(result.all()))
diff --git a/src/beaver_gateway/frontends/telegram/render.py b/src/beaver_gateway/frontends/telegram/render.py
new file mode 100644
index 0000000..954cafc
--- /dev/null
+++ b/src/beaver_gateway/frontends/telegram/render.py
@@ -0,0 +1,111 @@
+"""Model markdown → Telegram HTML, chunking, and the status line for drafts."""
+
+from __future__ import annotations
+
+import html
+import re
+from typing import Any
+
+__all__ = ["LIMIT", "chunks", "status_label", "to_html"]
+
+LIMIT = 4000
+_FENCE = re.compile(r"```[^\n]*\n(.*?)(?:```|$)", re.DOTALL)
+_INLINE_CODE = re.compile(r"(`[^`\n]+`)")
+_HEADING = re.compile(r"^#{1,6}\s+(.+)$", re.MULTILINE)
+_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__", re.DOTALL)
+_ITALIC = re.compile(r"(? str:
+ out: list[str] = []
+ pos = 0
+ for match in _FENCE.finditer(text):
+ out.append(_inline(text[pos : match.start()]))
+ out.append(f"{html.escape(match.group(1).rstrip())}")
+ pos = match.end()
+ out.append(_inline(text[pos:]))
+ return "".join(out).strip()
+
+
+def _inline(text: str) -> str:
+ parts = _INLINE_CODE.split(text)
+ for i, part in enumerate(parts):
+ if i % 2:
+ parts[i] = f"{html.escape(part[1:-1])}"
+ continue
+ s = html.escape(part, quote=False)
+ s = _HEADING.sub(r"\1", s)
+ s = _BOLD.sub(lambda m: f"{m.group(1) or m.group(2)}", s)
+ s = _ITALIC.sub(r"\1", s)
+ s = _ITALIC_U.sub(r"\1", s)
+ s = _STRIKE.sub(r"\1", s)
+ s = _LINK.sub(r'\1', s)
+ parts[i] = _BULLET.sub(r"\1• ", s)
+ return "".join(parts)
+
+
+def chunks(text: str, limit: int = LIMIT) -> list[str]:
+ text = text.strip()
+ if len(text) <= limit:
+ return [text] if text else []
+ out: list[str] = []
+ while len(text) > limit:
+ cut = _cut_point(text, limit)
+ out.append(text[:cut].rstrip())
+ text = text[cut:].lstrip()
+ if text:
+ out.append(text)
+ return out
+
+
+def _cut_point(text: str, limit: int) -> int:
+ for sep in ("\n\n", "\n", ". ", " "):
+ cut = text.rfind(sep, limit // 2, limit)
+ if cut > 0:
+ return cut + len(sep)
+ return limit
+
+
+def status_label(name: str, tool_input: dict[str, Any] | None = None) -> str:
+ label = _LABELS.get(name)
+ if label is not None:
+ return label
+ if name.startswith("mcp__"):
+ parts = name.split("__", 2)
+ server = parts[1]
+ tool = parts[2] if len(parts) == 3 else ""
+ return f"{server}: {tool}…" if tool else f"{server}…"
+ hint = ""
+ if tool_input:
+ for key in ("description", "command", "file_path", "pattern", "query"):
+ value = tool_input.get(key)
+ if isinstance(value, str) and value.strip():
+ hint = value.strip().splitlines()[0][:60]
+ break
+ return f"{name} {hint}…".strip() if hint else f"{name}…"
diff --git a/src/beaver_gateway/storage/__init__.py b/src/beaver_gateway/storage/__init__.py
index 8093662..751dd1e 100644
--- a/src/beaver_gateway/storage/__init__.py
+++ b/src/beaver_gateway/storage/__init__.py
@@ -20,8 +20,10 @@ from beaver_gateway.storage.models import (
AuditLog,
Conversation,
ConversationBinding,
+ Delivery,
InjectQueueItem,
Schedule,
+ TelegramUpdate,
Token,
TranscriptEntry,
Usage,
@@ -33,9 +35,11 @@ __all__ = [
"Conversation",
"ConversationBinding",
"Database",
+ "Delivery",
"InjectQueueItem",
"PostgresSessionStore",
"Schedule",
+ "TelegramUpdate",
"Token",
"TranscriptEntry",
"Usage",
diff --git a/src/beaver_gateway/storage/models.py b/src/beaver_gateway/storage/models.py
index b2c3a55..9227113 100644
--- a/src/beaver_gateway/storage/models.py
+++ b/src/beaver_gateway/storage/models.py
@@ -184,6 +184,52 @@ class Schedule(SQLModel, table=True):
delivered_at: datetime | None = Field(default=None)
+class TelegramUpdate(SQLModel, table=True):
+ """Inbox of the Telegram frontend (§3.8).
+
+ The update is stored before the poll offset moves past it and handled
+ from here, so a restart neither loses nor duplicates a message.
+ """
+
+ __tablename__ = "telegram_updates"
+
+ update_id: int = Field(primary_key=True, sa_column_kwargs={"autoincrement": False})
+ payload: dict[str, Any] = Field(
+ sa_column=Column(JSON().with_variant(JSONB(), "postgresql"), nullable=False)
+ )
+ received_at: datetime = Field(default_factory=_utcnow)
+ processed_at: datetime | None = Field(default=None, index=True)
+ error: str | None = Field(default=None)
+
+
+class Delivery(SQLModel, table=True):
+ """Outbox (§3.8): a reply is a row first and a Telegram message second.
+
+ ``status`` walks ``queued -> sent`` with retries on the way, ``failed``
+ only when Telegram rejects the row for good (unknown thread, blocked
+ bot). ``dedupe_key`` keeps one row per turn chunk across restarts.
+ """
+
+ __tablename__ = "deliveries"
+
+ id: int | None = Field(default=None, primary_key=True)
+ conversation_id: int | None = Field(default=None, index=True)
+ frontend: str = Field(default="telegram", index=True)
+ chat_id: int
+ thread_id: int | None = Field(default=None)
+ text: str
+ plain: bool = Field(default=False)
+ turn_id: str | None = Field(default=None)
+ dedupe_key: str | None = Field(default=None, unique=True)
+ status: str = Field(default="queued", index=True)
+ attempts: int = Field(default=0)
+ next_attempt_at: datetime = Field(default_factory=_utcnow, index=True)
+ last_error: str | None = Field(default=None)
+ message_id: int | None = Field(default=None)
+ created_at: datetime = Field(default_factory=_utcnow)
+ sent_at: datetime | None = Field(default=None)
+
+
class ConversationMessage(SQLModel, table=True):
"""One raw Anthropic-shape message in a conversation's transcript.
@@ -280,8 +326,10 @@ __all__ = [
"Conversation",
"ConversationBinding",
"ConversationMessage",
+ "Delivery",
"InjectQueueItem",
"Schedule",
+ "TelegramUpdate",
"Token",
"TranscriptEntry",
"Usage",
diff --git a/tests/test_telegram.py b/tests/test_telegram.py
new file mode 100644
index 0000000..0adef03
--- /dev/null
+++ b/tests/test_telegram.py
@@ -0,0 +1,533 @@
+import asyncio
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from aiogram.exceptions import TelegramBadRequest, TelegramNetworkError
+from aiogram.methods import SendMessage
+from aiogram.types import Update
+from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny
+
+from beaver_gateway.core.registry import McpRegistry
+from beaver_gateway.core.transcript import build_entries
+from beaver_gateway.frontends.base import GatewayRuntime
+from beaver_gateway.frontends.telegram import TelegramFrontend
+from beaver_gateway.frontends.telegram.render import chunks, status_label, to_html
+from beaver_gateway.storage.models import ConversationBinding, Delivery, TelegramUpdate
+from sqlmodel import select
+from test_conversations import ScriptedClient, StubFrontend, World
+
+USER = 42
+
+
+class FakeBot:
+ def __init__(self) -> None:
+ self.updates: list[dict[str, Any]] = []
+ self.sent: list[dict[str, Any]] = []
+ self.drafts: list[dict[str, Any]] = []
+ self.edits: list[dict[str, Any]] = []
+ self.topics: list[str] = []
+ self.reactions: list[tuple[int, list[Any]]] = []
+ self.fail_sends = 0
+ self.reject_html = False
+ self._message_id = 100
+ self.session = SimpleNamespace(close=self._close)
+
+ async def _close(self) -> None:
+ pass
+
+ async def get_me(self) -> Any:
+ return SimpleNamespace(username="bot", has_topics_enabled=True)
+
+ async def get_updates(self, offset=None, **_: Any) -> list[Update]:
+ await asyncio.sleep(0.01)
+ return [
+ Update.model_validate(u)
+ for u in self.updates
+ if offset is None or u["update_id"] >= offset
+ ]
+
+ async def send_message(
+ self, chat_id, text, message_thread_id=None, parse_mode=None, **kwargs: Any
+ ) -> Any:
+ if self.fail_sends:
+ self.fail_sends -= 1
+ raise TelegramNetworkError(
+ method=SendMessage(chat_id=0, text="x"), message="boom"
+ )
+ if self.reject_html and parse_mode == "HTML":
+ raise TelegramBadRequest(
+ method=SendMessage(chat_id=0, text="x"),
+ message="Bad Request: can't parse entities",
+ )
+ self._message_id += 1
+ self.sent.append(
+ {
+ "chat_id": chat_id,
+ "thread": message_thread_id,
+ "text": text,
+ "parse_mode": parse_mode,
+ "markup": kwargs.get("reply_markup"),
+ "message_id": self._message_id,
+ }
+ )
+ return SimpleNamespace(message_id=self._message_id)
+
+ async def send_message_draft(self, **kwargs: Any) -> bool:
+ self.drafts.append(kwargs)
+ return True
+
+ async def create_forum_topic(self, chat_id, name, **_: Any) -> Any:
+ self.topics.append(name)
+ return SimpleNamespace(message_thread_id=900 + len(self.topics), name=name)
+
+ async def edit_forum_topic(
+ self, chat_id, message_thread_id, name=None, **_: Any
+ ) -> bool:
+ self.topics.append(f"edit:{message_thread_id}:{name}")
+ return True
+
+ async def set_message_reaction(
+ self, chat_id, message_id, reaction=None, **_: Any
+ ) -> bool:
+ self.reactions.append((message_id, reaction or []))
+ return True
+
+ async def answer_callback_query(self, *_: Any, **__: Any) -> bool:
+ return True
+
+ async def edit_message_text(self, text, chat_id, message_id, **_: Any) -> Any:
+ self.edits.append({"message_id": message_id, "text": text})
+ return True
+
+ async def edit_message_reply_markup(
+ self, chat_id, message_id, reply_markup=None
+ ) -> Any:
+ self.edits.append({"message_id": message_id, "markup": reply_markup})
+ return True
+
+ async def download(self, file_id, destination=None) -> None:
+ Path(destination).write_bytes(b"data")
+
+ # helpers for tests
+ def push(self, payload: dict[str, Any]) -> None:
+ payload["update_id"] = 1000 + len(self.updates)
+ self.updates.append(payload)
+
+ def message(self, text: str, *, thread: int | None = None, uid: int = USER) -> None:
+ body: dict[str, Any] = {
+ "message_id": 10 + len(self.updates),
+ "date": 1700000000,
+ "chat": {"id": USER, "type": "private"},
+ "from": {"id": uid, "is_bot": False, "first_name": "h"},
+ "text": text,
+ }
+ if thread is not None:
+ body |= {"message_thread_id": thread, "is_topic_message": True}
+ self.push({"message": body})
+
+ def topic_created(self, name: str, thread: int) -> None:
+ self.push(
+ {
+ "message": {
+ "message_id": 10 + len(self.updates),
+ "date": 1700000000,
+ "chat": {"id": USER, "type": "private"},
+ "from": {"id": USER, "is_bot": False, "first_name": "h"},
+ "message_thread_id": thread,
+ "is_topic_message": True,
+ "forum_topic_created": {"name": name, "icon_color": 1},
+ }
+ }
+ )
+
+ def callback(self, data: str, message_id: int) -> None:
+ self.push(
+ {
+ "callback_query": {
+ "id": f"cb{len(self.updates)}",
+ "from": {"id": USER, "is_bot": False, "first_name": "h"},
+ "chat_instance": "ci",
+ "data": data,
+ "message": {
+ "message_id": message_id,
+ "date": 1700000000,
+ "chat": {"id": USER, "type": "private"},
+ "text": "?",
+ },
+ }
+ }
+ )
+
+
+class Stack:
+ def __init__(self, world: World, bot: FakeBot) -> None:
+ self.world = world
+ self.bot = bot
+ self.tg = TelegramFrontend(
+ token="t",
+ user_id=USER,
+ master_agent="a",
+ branch_agent="a",
+ draft_interval=0.05,
+ outbox_backoff=0.01,
+ )
+ self.tg._bot = bot # noqa: SLF001
+ markdown = StubFrontend("markdown", ("deep",), {"deep": "d"}, home=True)
+ markdown.conversations = world.conversations
+ frontends = [self.tg, markdown]
+ world.conversations._frontends = frontends # noqa: SLF001
+ runtime = GatewayRuntime(
+ agents=world.conversations._agents, # noqa: SLF001
+ mcps=McpRegistry([]),
+ backends=world.conversations._backends, # noqa: SLF001
+ token_store=None,
+ db=world.db,
+ conversations=world.conversations,
+ bus=world.bus,
+ pool=world.pool,
+ frontends=tuple(frontends),
+ )
+ self.tg.configure(runtime)
+ self.tasks = [
+ asyncio.create_task(self.tg.inbox.run()),
+ asyncio.create_task(self.tg.outbox.run()),
+ asyncio.create_task(self.tg._events()), # noqa: SLF001
+ ]
+
+ async def close(self) -> None:
+ for t in self.tasks:
+ t.cancel()
+ for t in self.tasks:
+ with pytest.raises(BaseException):
+ await t
+
+ async def until(self, pred, timeout: float = 5.0, what: str = "condition") -> Any:
+ deadline = asyncio.get_running_loop().time() + timeout
+ while asyncio.get_running_loop().time() < deadline:
+ value = pred()
+ if value:
+ return value
+ await asyncio.sleep(0.02)
+ msg = f"{what} never happened; sent={self.bot.sent}"
+ raise AssertionError(msg)
+
+ def sent_with(self, needle: str) -> dict[str, Any] | None:
+ return next((m for m in self.bot.sent if needle in m["text"]), None)
+
+
+@pytest.fixture
+async def stack() -> Stack:
+ root = Path(tempfile.mkdtemp(prefix="beaver-tg-"))
+ world = await World(root).setup()
+ await world.conversations.start()
+ s = Stack(world, FakeBot())
+ yield s
+ await s.close()
+ await world.conversations.stop()
+ await world.pool.close_all()
+ await world.db.dispose()
+
+
+async def test_general_is_master_and_reply_has_no_thread(stack: Stack) -> None:
+ stack.bot.message("hi")
+ reply = await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ assert reply["thread"] is None
+ assert reply["parse_mode"] == "HTML"
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ assert master is not None and master.kind == "master"
+ assert stack.bot.drafts and stack.bot.drafts[0]["chat_id"] == USER
+ rows = await stack.world.conversations.queue.recent(master.id)
+ assert {r.origin for r in rows} == {"сид:clean", "telegram"}
+
+
+async def test_new_topic_becomes_morning_branch_and_replies_in_thread(
+ stack: Stack,
+) -> None:
+ stack.bot.topic_created("план", 7)
+ stack.bot.message("hello topic", thread=7)
+ reply = await stack.until(lambda: stack.sent_with("ok:hello topic"), what="reply")
+ assert reply["thread"] == 7
+ branch = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/7"
+ )
+ assert branch is not None
+ assert branch.kind == "branch" and branch.title == "план"
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ assert branch.parent_id == master.id
+ prompts = [p for c in ScriptedClient.instances for p in c.prompts]
+ seed = next(p for p in prompts if p.startswith("[сид: morning] branch «план»"))
+ assert "Хендаут не приехал." in seed
+ assert stack.bot.drafts[-1]["message_thread_id"] == 7
+
+
+async def test_first_message_without_service_message_seeds_with_text(
+ stack: Stack,
+) -> None:
+ stack.bot.message("сразу текстом", thread=8)
+ await stack.until(lambda: stack.sent_with("сразу текстом"), what="reply")
+ branch = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/8"
+ )
+ assert branch.title == "сразу текстом"
+ client = next(
+ c for c in ScriptedClient.instances if c.prompts and "сразу" in c.prompts[0]
+ )
+ assert client.prompts[0].startswith("[сид: morning]")
+ assert client.prompts[0].endswith("сразу текстом")
+ assert len(client.prompts) == 1
+
+
+async def test_message_into_merged_branch_rebinds_the_topic(stack: Stack) -> None:
+ stack.bot.message("one", thread=9)
+ await stack.until(lambda: stack.sent_with("ok:"), what="first reply")
+ old = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/9"
+ )
+ await stack.world.conversations.set_status(old, "merged")
+ stack.bot.message("two", thread=9)
+ await stack.until(lambda: stack.sent_with("two"), what="second reply")
+ new = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/9"
+ )
+ assert new.id != old.id and new.status == "open"
+ async with stack.world.db.session() as session:
+ rows = list(
+ (
+ await session.exec(
+ select(ConversationBinding).where(
+ ConversationBinding.external_id == f"{USER}/9"
+ )
+ )
+ ).all()
+ )
+ assert {(r.conversation_id, r.visible) for r in rows} == {
+ (old.id, False),
+ (new.id, True),
+ }
+
+
+async def test_reply_from_another_window_is_mirrored_with_marker(stack: Stack) -> None:
+ stack.bot.message("hi")
+ await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ await stack.world.conversations.post(master, "from panel", origin="user")
+ await stack.until(lambda: stack.sent_with("ok:from panel"), what="mirrored reply")
+ texts = [m["text"] for m in stack.bot.sent]
+ marker = next(i for i, t in enumerate(texts) if t.startswith("📝 из панели:"))
+ assert "from panel" in texts[marker]
+ assert texts[marker + 1] == "ok:from panel"
+
+
+async def test_say_is_delivered_and_inject_turns_are_silent(stack: Stack) -> None:
+ stack.bot.message("hi")
+ await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ before = len(stack.bot.sent)
+ await stack.world.conversations.inject(
+ master, "cron tick", urgency="urgent", origin="крон"
+ )
+ await stack.world.settle(master, 3)
+ await asyncio.sleep(0.2)
+ assert len(stack.bot.sent) == before
+ await stack.world.conversations.say(master, "psst")
+ await stack.until(lambda: stack.sent_with("psst"), what="say")
+
+
+async def test_question_becomes_buttons_and_callback_answers(stack: Stack) -> None:
+ stack.bot.message("hi")
+ await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ payload = {
+ "questions": [
+ {
+ "header": "Цвет",
+ "question": "Какой цвет?",
+ "options": [
+ {"label": "Синий", "description": "как небо"},
+ {"label": "Красный", "description": ""},
+ ],
+ "multiSelect": False,
+ }
+ ]
+ }
+ asking = asyncio.create_task(
+ stack.world.conversations.ask(master.external_id, payload)
+ )
+ question = await stack.until(
+ lambda: stack.sent_with("Какой цвет?"), what="question"
+ )
+ assert question["markup"] is not None
+ labels = [b.text for row in question["markup"].inline_keyboard for b in row]
+ assert labels == ["Синий", "Красный"]
+ pending = stack.world.conversations.pending_question(master.external_id)
+ assert pending is not None
+ stack.bot.callback(f"q:{pending[0]}:0:1", question["message_id"])
+ assert await asyncio.wait_for(asking, 5) == "Красный"
+ assert stack.world.conversations.answer_text("Красный") == (
+ "Пользователь ответил: Красный"
+ )
+ await stack.until(
+ lambda: any("✅ Красный" in e.get("text", "") for e in stack.bot.edits),
+ what="edit",
+ )
+ row = await stack.world.conversations.get(master.external_id)
+ assert row.pending_question is False
+
+
+async def test_question_timeout_renders_text_and_free_text_answers(
+ stack: Stack,
+) -> None:
+ stack.bot.message("hi")
+ await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ master = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=str(USER)
+ )
+ stack.world.conversations._question_timeout = 0.3 # noqa: SLF001
+ payload = {"questions": [{"header": "Q", "question": "Сколько?", "options": []}]}
+ result = await stack.world.conversations.ask(master.external_id, payload)
+ assert result is None
+ assert "не ответил" in stack.world.conversations.answer_text(None)
+ await stack.until(
+ lambda: any("время вышло" in e.get("text", "") for e in stack.bot.edits),
+ what="timeout edit",
+ )
+ stack.world.conversations._question_timeout = 5.0 # noqa: SLF001
+ asking = asyncio.create_task(
+ stack.world.conversations.ask(master.external_id, payload)
+ )
+ await stack.until(
+ lambda: stack.world.conversations.pending_question(master.external_id),
+ what="pending",
+ )
+ stack.bot.message("семь")
+ assert await asyncio.wait_for(asking, 5) == "семь"
+
+
+async def test_commands_status_merge_and_new(stack: Stack) -> None:
+ stack.bot.message("/status")
+ status = await stack.until(lambda: stack.sent_with("master · open"), what="status")
+ assert "пул:" in status["text"]
+ stack.bot.message("work", thread=11)
+ await stack.until(lambda: stack.sent_with("work"), what="branch reply")
+ branch = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/11"
+ )
+ await stack.world.store.append(
+ stack.world.key(branch.session_id),
+ build_entries(
+ [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}],
+ session_id=branch.session_id,
+ cwd=str(stack.world.root),
+ model="m",
+ ),
+ )
+ stack.bot.message("/merge", thread=11)
+ await stack.until(lambda: stack.sent_with("слито в мастер"), what="merged")
+ branch = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/11"
+ )
+ assert branch.status == "merged"
+ stack.bot.message("/new отчёт")
+ await stack.until(lambda: stack.sent_with("в новом топике"), what="new topic")
+ assert stack.bot.topics == ["отчёт"]
+ child = await stack.world.conversations.find_bound(
+ frontend="telegram", external_id=f"{USER}/901"
+ )
+ assert child is not None and child.title == "отчёт"
+ assert await stack.tg.mark_topic(child)
+ assert stack.bot.topics[-1] == "edit:901:✅ отчёт"
+
+
+async def test_inbox_stores_first_and_replays_after_restart(stack: Stack) -> None:
+ stack.bot.message("hi")
+ await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
+ async with stack.world.db.session() as session:
+ rows = list((await session.exec(select(TelegramUpdate))).all())
+ assert [r.update_id for r in rows] == [1000]
+ assert rows[0].processed_at is not None and rows[0].error is None
+ async with stack.world.db.session() as session:
+ session.add(
+ TelegramUpdate(
+ update_id=999,
+ payload={
+ "update_id": 999,
+ "message": {
+ "message_id": 1,
+ "date": 1700000000,
+ "chat": {"id": USER, "type": "private"},
+ "from": {"id": USER, "is_bot": False, "first_name": "h"},
+ "text": "replayed",
+ },
+ },
+ )
+ )
+ await session.commit()
+ await stack.until(
+ lambda: stack.sent_with("ok:replayed"), timeout=8, what="replayed reply"
+ )
+ stack.bot.message("stranger", uid=1)
+ await asyncio.sleep(0.3)
+ assert stack.sent_with("stranger") is None
+
+
+async def test_outbox_retries_and_falls_back_to_plain(stack: Stack) -> None:
+ stack.bot.fail_sends = 2
+ stack.bot.reject_html = True
+ stack.bot.message("**bold**")
+ reply = await stack.until(lambda: stack.sent_with("ok:**bold**"), what="reply")
+ assert reply["parse_mode"] is None
+ async with stack.world.db.session() as session:
+ rows = list((await session.exec(select(Delivery))).all())
+ sent = [r for r in rows if r.status == "sent"]
+ assert sent and sent[0].plain and sent[0].attempts >= 3
+
+
+async def test_can_use_tool_answers_through_deny(stack: Stack) -> None:
+ seen: list[tuple[str, dict[str, Any]]] = []
+
+ async def asker(key: str, payload: dict[str, Any]) -> str:
+ seen.append((key, payload))
+ return "Пользователь ответил: да"
+
+ backend = stack.world.backend
+ backend._asker = asker # noqa: SLF001
+ can_use_tool = backend._can_use_tool("conv-1") # noqa: SLF001
+ allow = await can_use_tool("Bash", {"command": "ls"}, None)
+ assert isinstance(allow, PermissionResultAllow)
+ deny = await can_use_tool("AskUserQuestion", {"questions": []}, None)
+ assert isinstance(deny, PermissionResultDeny)
+ assert deny.message == "Пользователь ответил: да"
+ assert seen == [("conv-1", {"questions": []})]
+
+
+def test_render_helpers() -> None:
+ assert (
+ to_html("**жирно** и `code` ")
+ == "жирно и code <b>"
+ )
+ assert to_html("# Заголовок\n- пункт") == "Заголовок\n• пункт"
+ assert to_html("```py\nx = 1\n```") == "x = 1
"
+ assert (
+ to_html("[док](https://a.b/c?x=1&y=2)")
+ == 'док'
+ )
+ parts = chunks("абв\n\n" + "г" * 30 + "\n\n" + "д" * 30, limit=40)
+ assert parts == ["абв\n\n" + "г" * 30, "д" * 30]
+ assert status_label("Read", {"file_path": "/x"}) == "читаю vault…"
+ assert (
+ status_label("mcp__firefly__list_accounts", None) == "firefly: list_accounts…"
+ )
+ assert status_label("Foo", {"command": "ls -la"}) == "Foo ls -la…"
diff --git a/uv.lock b/uv.lock
index cc9fb2c..e60d895 100644
--- a/uv.lock
+++ b/uv.lock
@@ -22,6 +22,32 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/67/cd/0d76dfc5de72bde52f55f53e925c7d152d9c7906634ec1e0cbc7e8d4ad93/aiofile-3.11.1-py3-none-any.whl", hash = "sha256:ce77d14ac07f77bc2b757834a5c129321f3f705c474593deed5ab209079a52c9", size = 20446, upload-time = "2026-05-16T08:18:32.051Z" },
]
+[[package]]
+name = "aiofiles"
+version = "25.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/41/c3/534eac40372d8ee36ef40df62ec129bee4fdb5ad9706e58a29be53b2c970/aiofiles-25.1.0.tar.gz", hash = "sha256:a8d728f0a29de45dc521f18f07297428d56992a742f0cd2701ba86e44d23d5b2", size = 46354, upload-time = "2025-10-09T20:51:04.358Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/bc/8a/340a1555ae33d7354dbca4faa54948d76d89a27ceef032c8c3bc661d003e/aiofiles-25.1.0-py3-none-any.whl", hash = "sha256:abe311e527c862958650f9438e859c1fa7568a141b22abcd015e120e86a85695", size = 14668, upload-time = "2025-10-09T20:51:03.174Z" },
+]
+
+[[package]]
+name = "aiogram"
+version = "3.31.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "aiofiles" },
+ { name = "aiohttp" },
+ { name = "certifi" },
+ { name = "magic-filter" },
+ { name = "pydantic" },
+ { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ec/5f/b2ee094181bb578c0987eef92a51af1fc2caa2be3589da2e37543c0f2906/aiogram-3.31.0.tar.gz", hash = "sha256:f2c5064fe52d88898c86af62261c405440a014216e45dd99b2c6555a5602ce30", size = 2025095, upload-time = "2026-08-26T00:00:42.97Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/ab/dd/c1ea24eb687e564b6b64db3939b858bbcbd3f6998496508a168ce54c5770/aiogram-3.31.0-py3-none-any.whl", hash = "sha256:d889493c5917a867fc9da81fd6e6b3ae438e98bfe1b004fcdafe7400461a6c1b", size = 859100, upload-time = "2026-08-26T00:00:41.178Z" },
+]
+
[[package]]
name = "aiohappyeyeballs"
version = "2.6.1"
@@ -262,6 +288,7 @@ version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "aiofile" },
+ { name = "aiogram" },
{ name = "aiohttp" },
{ name = "aiosqlite" },
{ name = "anthropic" },
@@ -303,6 +330,7 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "aiofile", specifier = ">=3.11.1" },
+ { name = "aiogram", specifier = ">=3.31.0" },
{ name = "aiohttp", specifier = ">=3.13.5" },
{ name = "aiosqlite", specifier = ">=0.22.1" },
{ name = "anthropic", specifier = ">=0.103.0" },
@@ -1062,6 +1090,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/81/db/e655086b7f3a705df045bf0933bdd9c2f79bb3c97bfef1384598bb79a217/keyring-25.7.0-py3-none-any.whl", hash = "sha256:be4a0b195f149690c166e850609a477c532ddbfbaed96a404d4e43f8d5e2689f", size = 39160, upload-time = "2025-11-16T16:26:08.402Z" },
]
+[[package]]
+name = "magic-filter"
+version = "1.0.12"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/e6/08/da7c2cc7398cc0376e8da599d6330a437c01d3eace2f2365f300e0f3f758/magic_filter-1.0.12.tar.gz", hash = "sha256:4751d0b579a5045d1dc250625c4c508c18c3def5ea6afaf3957cb4530d03f7f9", size = 11071, upload-time = "2023-10-01T12:33:19.006Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/cc/75/f620449f0056eff0ec7c1b1e088f71068eb4e47a46eb54f6c065c6ad7675/magic_filter-1.0.12-py3-none-any.whl", hash = "sha256:e5929e544f310c2b1f154318db8c5cdf544dd658efa998172acd2e4ba0f6c6a6", size = 11335, upload-time = "2023-10-01T12:33:17.711Z" },
+]
+
[[package]]
name = "markdown-it-py"
version = "4.2.0"