feat(telegram,core,backends,storage): telegram frontend with inbox, outbox, drafts and question buttons

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

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

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,
             )
 

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"]

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

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"❓ <b>{header}</b>" 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"• <b>{label}</b>" + (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)
+    )

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()

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()))

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"(?<![\w*])\*(?!\s)(.+?)(?<!\s)\*(?![\w*])")
+_ITALIC_U = re.compile(r"(?<![\w_])_(?!\s)(.+?)(?<!\s)_(?![\w_])")
+_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^)\s]+)\)")
+_BULLET = re.compile(r"^(\s*)[-*]\s+", re.MULTILINE)
+_STRIKE = re.compile(r"~~(.+?)~~")
+
+_LABELS: dict[str, str] = {
+    "Read": "читаю vault…",
+    "Glob": "ищу файлы…",
+    "Grep": "ищу в vault…",
+    "Edit": "правлю файл…",
+    "Write": "пишу файл…",
+    "MultiEdit": "правлю файлы…",
+    "Bash": "выполняю команду…",
+    "WebSearch": "ищу в сети…",
+    "WebFetch": "читаю страницу…",
+    "Task": "запустил сабагента…",
+    "Agent": "запустил сабагента…",
+    "AskUserQuestion": "спрашиваю…",
+    "TodoWrite": "планирую…",
+    "Skill": "открываю скилл…",
+    "mcp__gateway__spawn": "открываю разговор…",
+    "mcp__gateway__read_conversation": "читаю разговор…",
+    "mcp__gateway__say": "говорю…",
+    "mcp__gateway__schedule": "ставлю напоминание…",
+    "mcp__gateway__inject": "передаю в другой разговор…",
+}
+
+
+def to_html(text: str) -> str:
+    out: list[str] = []
+    pos = 0
+    for match in _FENCE.finditer(text):
+        out.append(_inline(text[pos : match.start()]))
+        out.append(f"<pre>{html.escape(match.group(1).rstrip())}</pre>")
+        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"<code>{html.escape(part[1:-1])}</code>"
+            continue
+        s = html.escape(part, quote=False)
+        s = _HEADING.sub(r"<b>\1</b>", s)
+        s = _BOLD.sub(lambda m: f"<b>{m.group(1) or m.group(2)}</b>", s)
+        s = _ITALIC.sub(r"<i>\1</i>", s)
+        s = _ITALIC_U.sub(r"<i>\1</i>", s)
+        s = _STRIKE.sub(r"<s>\1</s>", s)
+        s = _LINK.sub(r'<a href="\2">\1</a>', 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}…"

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",

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",