deer-flow/backend/app/channels/telegram.py

"""Telegram channel — connects via long-polling (no public IP needed)."""

from __future__ import annotations

import asyncio
import logging
import re
import threading
import time
from collections.abc import Coroutine
from concurrent.futures import Future
from typing import Any

from app.channels.allowed_users import parse_allowed_users
from app.channels.base import Channel, ChannelStopTimeout
from app.channels.connection_identity import attach_connection_identity
from app.channels.message_bus import (
    INBOUND_FILE_CONTENT_KEY,
    InboundMessage,
    InboundMessageType,
    InboundReservation,
    MessageBus,
    OutboundMessage,
    ResolvedAttachment,
)
from deerflow.uploads.manager import is_upload_staging_file, normalize_filename

logger = logging.getLogger(__name__)

TELEGRAM_MAX_MESSAGE_LENGTH = 4096
TELEGRAM_MAX_RICH_MESSAGE_LENGTH = 32768
# Telegram's hosted Bot API documents this as 20 MB (decimal bytes).
TELEGRAM_MAX_INBOUND_FILE_BYTES = 20_000_000
# Keep Telegram cleanup inside the Gateway's five-second shutdown-hook budget.
TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS = 4.0
TELEGRAM_BRIDGE_DRAIN_TIMEOUT_SECONDS = 1.0
STREAM_EDIT_MIN_INTERVAL_SECONDS = 1.0
# Groups (negative chat_id) are capped at 20 messages/minute by Telegram,
# so stream edits there must pace well below the private-chat 1 msg/s guideline.
STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS = 3.0
# Bound on tracked in-flight streamed messages; entries normally clear on the
# final update, this only guards against leaks when a final never arrives.
MAX_TRACKED_STREAM_MESSAGES = 256

# Indirection so tests can patch the clock without touching the global time module.
_monotonic = time.monotonic

# Rich Messages only earn their keep when the text carries a construct that
# renders natively: a fenced code block, emphasis, a table, a task list,
# <details>, block math, or a link. Structured command/error replies are plain
# text with none of these, so they stay plain and their newlines and
# <placeholder> tokens survive verbatim instead of being collapsed by the
# rich parser.
#
# The patterns are deliberately *well-formed*, not "contains this character":
# a table must be a line that leads with a pipe, a link must be [text](url),
# so a command line like "/goal [condition|clear]" (brackets + a mid-line pipe)
# never trips the detector.
_TELEGRAM_RICH_CONSTRUCT_RE = re.compile(
    r"(?m)"
    r"^\s*(`{3,}|~{3,})"  # fenced code block
    r"|^\s*[-*+]\s+\[[ xX]\]"  # task list item
    r"|^\s*\|.*\|"  # table row (line leads with a pipe)
    r"|^\s*\|?\s*:?-{2,}\s*\|"  # table separator row (needs a pipe, so "--flag" stays plain)
    r"|\[[^\]]*\]\("  # markdown link [text](url)
    r"|\*\*[^*\n]+\*\*"  # bold
    r"|\*(?!\s)[^*\n]+?(?<!\s)\*"  # italic (tight delimiters, so "2 * 3" stays plain)
    r"|<details"  # collapsible details
    r"|\$\$"  # block math
)


def _has_rich_constructs(text: str) -> bool:
    """Whether *text* contains a construct that needs native rich rendering."""
    return _TELEGRAM_RICH_CONSTRUCT_RE.search(text) is not None


def _utf16_width(char: str) -> int:
    """UTF-16 code units taken by *char*, the unit Telegram measures message limits in."""
    return 2 if ord(char) > 0xFFFF else 1


def _first_utf16_chunk(text: str, limit: int) -> tuple[str, bool]:
    """The longest prefix of *text* fitting in *limit* UTF-16 code units, and whether anything was left out.

    Stops at the first character that does not fit, so previewing a bounded
    number of units out of a growing reply costs the limit, not the reply.
    """
    units = 0
    for index, char in enumerate(text):
        units += _utf16_width(char)
        if units > limit:
            return text[:index], True
    return text, False


def _load_telegram_input_file(path, filename: str):
    from telegram import InputFile

    return InputFile(path.read_bytes(), filename=filename)


def _parse_telegram_user_id(entry: Any) -> int | None:
    """A positive Telegram user ID from an ``int`` or a digit string, else ``None``."""
    if isinstance(entry, bool):
        # bool is an int subclass: ``true`` must not become user 1.
        return None
    if isinstance(entry, int):
        return entry if entry > 0 else None
    if isinstance(entry, str):
        text = entry.strip()
        if text.isascii() and text.isdigit():
            return int(text) or None
    return None


def _parse_allowed_users(allowed_users: Any) -> frozenset[int] | None:
    """Parse positive numeric Telegram IDs with shared fail-closed semantics."""
    return parse_allowed_users(
        allowed_users,
        parse_user_id=_parse_telegram_user_id,
        logger=logger,
        channel_name="Telegram",
        expected_id="a positive numeric Telegram user ID (not an @username)",
        valid_id_name="numeric user ID",
    )


class TelegramChannel(Channel):
    """Telegram bot channel using long-polling.

    Configuration keys (in ``config.yaml`` under ``channels.telegram``):
        - ``bot_token``: Telegram Bot API token (from @BotFather).
        - ``allowed_users``: (optional) List of numeric Telegram user IDs (not
          @usernames), or a single ID. Empty = allow all; a non-empty list with
          no valid ID denies everyone.
    """

    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
        super().__init__(name="telegram", bus=bus, config=config)
        self._application = None
        self._thread: threading.Thread | None = None
        self._tg_loop: asyncio.AbstractEventLoop | None = None
        self._main_loop: asyncio.AbstractEventLoop | None = None
        self._bootstrap_task: asyncio.Task[None] | None = None
        self._polling_cleanup_started = False
        # Dedicated Bot for inbound file downloads. The Application's Bot is
        # built on the manager/main loop (see start()), so its httpx
        # connection pool is bound to that loop; opening a fresh download
        # connection from the Telegram loop then trips "Event bound to a
        # different event loop". A Bot initialized on the Telegram loop keeps
        # its pool on the loop that performs the download.
        self._download_bot: Any = None
        self._download_bot_lock = asyncio.Lock()
        # The first stop() owns this Telegram-loop cleanup. A bounded or
        # cancelled caller leaves it running so a retry can join the same work.
        self._shutdown_resources_future: Future[None] | None = None
        self._shutdown_resources_lock = threading.Lock()
        # Tasks submitted from the main dispatcher loop back to PTB's loop.
        # Only the Telegram loop mutates this set.
        self._tg_bridge_tasks: set[asyncio.Task[Any]] = set()
        self._allowed_users = _parse_allowed_users(config.get("allowed_users"))
        # chat_id -> last sent message_id for threaded replies
        self._last_bot_message: dict[str, int] = {}
        # stream_key ("chat_id:thread_ts") -> state of the in-flight streamed
        # bot message being edited in place: {"message_id", "last_edit_at", "last_text"}
        self._stream_messages: dict[str, dict[str, Any]] = {}

    @property
    def supports_streaming(self) -> bool:
        return True

    async def start(self) -> None:
        if self._running:
            return
        if self._thread is not None and self._thread.is_alive():
            raise ChannelStopTimeout("Telegram polling worker has not finished shutting down")
        with self._shutdown_resources_lock:
            if self._shutdown_resources_future is not None and not self._shutdown_resources_future.done():
                raise ChannelStopTimeout("Telegram resource cleanup has not finished")
            self._shutdown_resources_future = None
        self._polling_cleanup_started = False

        try:
            from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters
        except ImportError:
            logger.error("python-telegram-bot is not installed. Install it with: uv add python-telegram-bot")
            return

        bot_token = self.config.get("bot_token", "")
        if not bot_token:
            logger.error("Telegram channel requires bot_token")
            return

        self._main_loop = asyncio.get_event_loop()
        self._open_threadsafe_future_intake()
        self._running = True
        self.bus.subscribe_outbound(self._on_outbound)

        # Build the application
        app = ApplicationBuilder().token(bot_token).build()

        # Command handlers
        app.add_handler(CommandHandler("start", self._cmd_start))
        app.add_handler(CommandHandler("bootstrap", self._cmd_generic))
        app.add_handler(CommandHandler("new", self._cmd_generic))
        app.add_handler(CommandHandler("status", self._cmd_generic))
        app.add_handler(CommandHandler("models", self._cmd_generic))
        app.add_handler(CommandHandler("model", self._cmd_generic))
        app.add_handler(CommandHandler("memory", self._cmd_generic))
        app.add_handler(CommandHandler("agent", self._cmd_generic))
        app.add_handler(CommandHandler("goal", self._cmd_generic))
        app.add_handler(CommandHandler("help", self._cmd_generic))

        # Slash skill commands are dynamic and cannot all be pre-registered
        # with Telegram, so route unknown slash commands through chat handling.
        app.add_handler(MessageHandler(filters.TEXT & filters.COMMAND, self._on_text))

        # General message handler
        app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self._on_text))

        # Telegram keeps attachment captions separate from message.text, and
        # photo/document updates do not match filters.TEXT.
        app.add_handler(MessageHandler(filters.PHOTO | filters.Document.ALL, self._on_text))

        self._application = app

        # Run polling in a dedicated thread with its own event loop
        self._thread = threading.Thread(target=self._run_polling, daemon=True)
        self._thread.start()
        logger.info("Telegram channel started")

    async def stop(self) -> None:
        self._running = False
        self.bus.unsubscribe_outbound(self._on_outbound)
        await self._close_and_drain_threadsafe_futures()
        shutdown_loop = asyncio.get_running_loop()
        deadline = shutdown_loop.time() + TELEGRAM_SHUTDOWN_TIMEOUT_SECONDS
        telegram_loop = self._tg_loop
        worker_thread = self._thread
        telegram_loop_running = telegram_loop is not None and telegram_loop.is_running()

        # Bridge tasks and the download Bot's HTTPX clients may only be released
        # on the Telegram loop, so that loop has to outlive the cleanup. Exactly
        # one cleanup generation is scheduled per channel: a bounded or
        # cancelled caller leaves it running, and a retried stop() joins the
        # same work instead of scheduling a second shutdown next to it.
        cleanup_future = self._shutdown_resources_future
        if cleanup_future is None and telegram_loop_running:
            cleanup_future = self._request_telegram_cleanup(telegram_loop)
        if cleanup_future is not None and not telegram_loop_running and not cleanup_future.done():
            # Its owner loop is gone, so nothing can finish this cleanup.
            logger.warning("[Telegram] Telegram loop stopped before its resource cleanup finished; the download bot's HTTPX clients may not have been closed")
            with self._shutdown_resources_lock:
                if self._shutdown_resources_future is cleanup_future:
                    self._shutdown_resources_future = None
            cleanup_future = None
        if cleanup_future is not None:
            try:
                remaining = max(0.0, deadline - shutdown_loop.time())
                # shield(): the deadline and caller cancellation must interrupt
                # only this wait, never the cleanup the Telegram loop owns.
                await asyncio.wait_for(asyncio.shield(asyncio.wrap_future(cleanup_future)), timeout=remaining)
            except asyncio.CancelledError:
                # Retain the loop and its cleanup: ChannelService keeps this
                # instance, and its next stop() converges on the same generation.
                raise
            except TimeoutError:
                logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown")
                raise ChannelStopTimeout("Telegram loop cleanup is still running after stop timeout") from None
            except Exception as exc:
                logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__)

        if telegram_loop_running:
            try:
                telegram_loop.call_soon_threadsafe(self._stop_telegram_loop, telegram_loop)
            except RuntimeError:
                pass
        if worker_thread is not None and worker_thread.is_alive():
            remaining = max(0.0, deadline - shutdown_loop.time())
            if remaining:
                try:
                    await asyncio.wait_for(
                        asyncio.to_thread(worker_thread.join, remaining),
                        timeout=remaining,
                    )
                except TimeoutError:
                    logger.warning("[Telegram] polling thread did not stop within the shutdown budget")
        if worker_thread is not None and worker_thread.is_alive():
            raise ChannelStopTimeout("Telegram polling thread is still running after stop timeout")
        if self._thread is worker_thread:
            self._thread = None
        self._application = None
        with self._shutdown_resources_lock:
            if self._shutdown_resources_future is cleanup_future:
                self._shutdown_resources_future = None
        logger.info("Telegram channel stopped")

    def _request_telegram_cleanup(self, loop: asyncio.AbstractEventLoop) -> Future[None]:
        """Share one owner-loop cleanup across stop callers and startup failure."""
        with self._shutdown_resources_lock:
            if self._shutdown_resources_future is None:
                self._shutdown_resources_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), loop)
            return self._shutdown_resources_future

    def _stop_telegram_loop(self, loop: asyncio.AbstractEventLoop) -> None:
        # A stop retry must not interrupt run_until_complete while the polling
        # thread is joining its cleanup after bootstrap cancellation/failure.
        if not self._polling_cleanup_started:
            loop.stop()

    async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
        if not self._application:
            return

        try:
            chat_id = int(msg.chat_id)
        except (ValueError, TypeError):
            logger.error("Invalid Telegram chat_id: %s", msg.chat_id)
            return

        key = self._stream_key(msg.chat_id, msg.thread_ts)

        if not msg.is_final:
            await self._send_stream_update(chat_id, key, msg.text, reply_to=self._parse_message_id(msg.thread_ts))
            return

        state = self._stream_messages.pop(key, None)
        if state is not None:
            if self._can_send_rich(msg.text) and await self._edit_rich_message(chat_id, state["message_id"], msg.text):
                self._last_bot_message[msg.chat_id] = state["message_id"]
                return
            await self._finalize_stream_message(chat_id, msg.chat_id, state, msg.text)
            return

        if self._can_send_rich(msg.text):
            try:
                message_id = await self._send_new_rich_message(chat_id, msg.chat_id, msg.text, _max_retries=_max_retries)
            except Exception as exc:
                logger.warning("[Telegram] Rich Message send failed in chat=%s; falling back to plain text: %s", chat_id, exc)
                message_id = None
            if message_id is not None:
                return

        for chunk in self._split_message(msg.text):
            await self._send_new_message(chat_id, msg.chat_id, chunk, _max_retries=_max_retries)

    async def _send_stream_update(self, chat_id: int, key: str, text: str, reply_to: int | None = None) -> None:
        """Edit the in-flight streamed message with accumulated text.

        Updates are best-effort: throttled, rate-limit drops are silent.  The
        manager always publishes a final message afterwards, which guarantees
        delivery of the complete text.
        """
        if not text:
            return

        # The manager republishes the whole cumulative reply as it grows, so the
        # preview is clipped by a bounded scan instead of measuring and splitting
        # the full text on every update.
        display, over_limit = _first_utf16_chunk(text, TELEGRAM_MAX_MESSAGE_LENGTH)
        if over_limit:
            # Clip again with one unit free, so the ellipsis stays inside the limit.
            display, _ = _first_utf16_chunk(display, TELEGRAM_MAX_MESSAGE_LENGTH - 1)
            display += "…"

        bot = self._application.bot
        state = self._stream_messages.get(key)

        send_kwargs: dict[str, Any] = {"chat_id": chat_id, "text": display}
        if reply_to:
            send_kwargs["reply_to_message_id"] = reply_to
            if chat_id > 0:
                send_kwargs["allow_sending_without_reply"] = True

        if state is None:
            try:
                sent = await bot.send_message(**send_kwargs)
            except Exception:
                logger.exception("[Telegram] failed to start stream message in chat=%s", chat_id)
                return
            self._register_stream_message(key, message_id=sent.message_id, last_text=display, last_edit_at=_monotonic())
            return

        now = _monotonic()
        min_interval = STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS if chat_id < 0 else STREAM_EDIT_MIN_INTERVAL_SECONDS
        if now - state["last_edit_at"] < min_interval:
            return
        if display == state["last_text"]:
            return

        try:
            await bot.edit_message_text(chat_id=chat_id, message_id=state["message_id"], text=display)
        except Exception as exc:
            if self._is_not_modified(exc):
                state["last_text"] = display
                return
            if self._is_retry_after(exc):
                logger.debug("[Telegram] stream edit rate-limited in chat=%s, dropping update", chat_id)
                return
            logger.warning("[Telegram] stream edit failed in chat=%s, sending new message: %s", chat_id, exc)
            try:
                sent = await bot.send_message(**send_kwargs)
            except Exception:
                logger.exception("[Telegram] failed to send fallback stream message in chat=%s", chat_id)
                return
            state["message_id"] = sent.message_id

        state["last_edit_at"] = _monotonic()
        state["last_text"] = display

    async def _finalize_stream_message(self, chat_id: int, chat_key: str, state: dict[str, Any], text: str) -> None:
        """Apply the final text: edit the streamed message, splitting overflow into follow-ups."""
        bot = self._application.bot
        chunks = self._split_message(text or "")

        edited = True
        if chunks[0] != state["last_text"]:
            edited = await self._edit_final_chunk(bot, chat_id, state["message_id"], chunks[0])

        if edited:
            self._last_bot_message[chat_key] = state["message_id"]
        else:
            # Edit could not be applied (e.g. message deleted) — deliver the
            # first chunk as a fresh message with the standard retry policy.
            await self._send_new_message(chat_id, chat_key, chunks[0])

        for chunk in chunks[1:]:
            await self._send_new_message(chat_id, chat_key, chunk)

    async def _edit_final_chunk(self, bot, chat_id: int, message_id: int, text: str) -> bool:
        """Edit with one rate-limit retry. Returns False if the edit could not be applied."""
        for attempt in range(2):
            try:
                await bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text)
                return True
            except Exception as exc:
                if self._is_not_modified(exc):
                    return True
                if self._is_retry_after(exc) and attempt == 0:
                    await asyncio.sleep(self._retry_after_seconds(exc))
                    continue
                logger.warning("[Telegram] final edit failed in chat=%s: %s", chat_id, exc)
                return False
        return False

    def _can_send_rich(self, text: str) -> bool:
        # Rich Messages are used only when rich_messages is on and the text
        # actually contains a rich construct. Structured command/error replies
        # are plain text with none, so they stay plain and their newlines and
        # <placeholder> tokens are not collapsed into one line.
        if not self.config.get("rich_messages"):
            return False
        # Telegram measures this cap in UTF-16 code units like its other limits, while
        # `len()` counts code points, so reuse the bounded scan the plain-text path
        # uses: an emoji-heavy reply that fits by code points is still rejected.
        _, over_limit = _first_utf16_chunk(text, TELEGRAM_MAX_RICH_MESSAGE_LENGTH)
        return not over_limit and _has_rich_constructs(text)

    async def _edit_rich_message(self, chat_id: int, message_id: int, text: str) -> bool:
        """Replace a streamed preview with a persistent Telegram Rich Message."""
        from telegram.error import BadRequest, EndPointNotFound

        bot = self._application.bot
        data = {"chat_id": chat_id, "message_id": message_id, "rich_message": {"markdown": text}}
        for attempt in range(2):
            try:
                await bot.do_api_request("editMessageText", api_kwargs=data)
                return True
            except Exception as exc:
                if self._is_retry_after(exc) and attempt == 0:
                    await asyncio.sleep(self._retry_after_seconds(exc))
                    continue
                if isinstance(exc, (BadRequest, EndPointNotFound)):
                    logger.warning("[Telegram] Rich Message rejected in chat=%s; falling back to plain text: %s", chat_id, exc)
                    return False
                logger.warning("[Telegram] final rich edit failed in chat=%s: %s", chat_id, exc)
                return False
        return False

    async def _send_new_rich_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int) -> int | None:
        """Send raw agent Markdown through Bot API 10.1 Rich Messages."""
        from telegram.error import BadRequest, EndPointNotFound

        bot = self._application.bot

        async def send_message() -> int | None:
            try:
                result = await bot.do_api_request(
                    "sendRichMessage",
                    api_kwargs={"chat_id": chat_id, "rich_message": {"markdown": text}},
                )
            except (BadRequest, EndPointNotFound) as exc:
                logger.warning("[Telegram] Rich Message rejected in chat=%s; falling back to plain text: %s", chat_id, exc)
                return None
            message_id = int(result["message_id"])
            self._last_bot_message[chat_key] = message_id
            return message_id

        return await self._send_with_retry(send_message, max_retries=_max_retries, log_prefix="[Telegram]")

    async def _send_new_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int = 3) -> int | None:
        """Send a fresh message with retry/backoff. Returns the sent message_id."""
        kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text}

        # Reply to the last bot message in this chat for threading
        reply_to = self._last_bot_message.get(chat_key)
        if reply_to:
            kwargs["reply_to_message_id"] = reply_to
            # A deleted private-chat reply must not block delivery. Keep group
            # replies strict: dropping their target could change forum routing.
            if chat_id > 0:
                kwargs["allow_sending_without_reply"] = True

        bot = self._application.bot

        async def send_message() -> int:
            sent = await bot.send_message(**kwargs)
            self._last_bot_message[chat_key] = sent.message_id
            return sent.message_id

        return await self._send_with_retry(
            send_message,
            max_retries=_max_retries,
            log_prefix="[Telegram]",
        )

    async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
        if not self._application:
            return False

        try:
            chat_id = int(msg.chat_id)
        except (ValueError, TypeError):
            logger.error("[Telegram] Invalid chat_id: %s", msg.chat_id)
            return False

        # Telegram limits: 10MB for photos, 50MB for documents
        if attachment.size > 50 * 1024 * 1024:
            logger.warning("[Telegram] file too large (%d bytes), skipping: %s", attachment.size, attachment.filename)
            return False

        bot = self._application.bot
        reply_to = self._last_bot_message.get(msg.chat_id)

        try:
            input_file = await asyncio.to_thread(_load_telegram_input_file, attachment.actual_path, attachment.filename)
            if attachment.is_image and attachment.size <= 10 * 1024 * 1024:
                kwargs: dict[str, Any] = {"chat_id": chat_id, "photo": input_file}
                if reply_to:
                    kwargs["reply_to_message_id"] = reply_to
                    if chat_id > 0:
                        kwargs["allow_sending_without_reply"] = True
                sent = await bot.send_photo(**kwargs)
            else:
                kwargs = {"chat_id": chat_id, "document": input_file}
                if reply_to:
                    kwargs["reply_to_message_id"] = reply_to
                    if chat_id > 0:
                        kwargs["allow_sending_without_reply"] = True
                sent = await bot.send_document(**kwargs)

            self._last_bot_message[msg.chat_id] = sent.message_id
            logger.info("[Telegram] file sent: %s to chat=%s", attachment.filename, msg.chat_id)
            return True
        except Exception:
            logger.exception("[Telegram] failed to send file: %s", attachment.filename)
            return False

    async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage:
        """Download inbound Telegram attachments for the shared upload pipeline.

        The Bot API download URL contains the bot token, so this adapter never
        exposes that URL to ``InboundMessage`` or logs. Instead it hands the
        downloaded bytes to ``ChannelManager`` through a short-lived private
        field that the manager consumes before persisting safe upload metadata.
        """
        # Owner/thread scoping is intentionally applied by ChannelManager when
        # it persists these bytes through _ingest_inbound_files().
        del thread_id, user_id
        if not msg.files:
            return msg

        materialized: list[dict[str, Any]] = []
        unavailable: list[str] = []

        for file_info in msg.files:
            if not isinstance(file_info, dict):
                continue

            filename = self._safe_inbound_filename(file_info.get("filename"), "attachment")
            file_id = file_info.get("file_id") if isinstance(file_info.get("file_id"), str) else ""
            declared_size = self._file_size(file_info.get("size"))
            if declared_size is not None and declared_size > TELEGRAM_MAX_INBOUND_FILE_BYTES:
                logger.warning("[Telegram] inbound file exceeds 20 MB download limit, skipping: %s", filename)
                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
                continue

            # Resolve the loop-bound download Bot. It is created on the
            # Telegram loop so its httpx pool is bound to the loop that
            # performs the download (the Application's Bot is main-loop-bound).
            # First-time resolution performs a network getMe, so it runs inside
            # its own guard: an init failure must mark only this attachment
            # unavailable, never abort the whole message.
            try:
                bot = await self._get_download_bot()
            except Exception as exc:
                logger.error(
                    "[Telegram] failed to initialize download bot for inbound file %s: %s%s",
                    filename,
                    type(exc).__name__,
                    self._describe_download_cause(exc),
                )
                unavailable.append(f"{filename} (download failed)")
                continue

            if bot is None or not file_id:
                logger.error("[Telegram] cannot download inbound file: %s", filename)
                unavailable.append(f"{filename} (download unavailable)")
                continue

            try:
                resolved_size, content = await self._run_on_telegram_loop(self._download_inbound_file(bot, file_id))
            except Exception as exc:
                # Exception strings from HTTP clients can contain request URLs.
                # Log only class names so a Bot API token can never leak, but
                # include the cause chain to distinguish timeout / reset / TLS.
                logger.error(
                    "[Telegram] failed to download inbound file %s: %s%s",
                    filename,
                    type(exc).__name__,
                    self._describe_download_cause(exc),
                )
                unavailable.append(f"{filename} (download failed)")
                continue

            if content is None:
                logger.warning("[Telegram] resolved inbound file exceeds 20 MB download limit, skipping: %s", filename)
                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
                continue

            if len(content) > TELEGRAM_MAX_INBOUND_FILE_BYTES:
                logger.warning("[Telegram] downloaded inbound file exceeds 20 MB limit, skipping: %s", filename)
                unavailable.append(f"{filename} (exceeds the 20 MB download limit)")
                continue

            materialized.append(
                {
                    "type": "image" if file_info.get("type") == "image" else "file",
                    "filename": filename,
                    "mime_type": file_info.get("mime_type") if isinstance(file_info.get("mime_type"), str) else "application/octet-stream",
                    "size": len(content),
                    INBOUND_FILE_CONTENT_KEY: content,
                }
            )

        msg.files = materialized
        if unavailable:
            notice = f"[Telegram attachment unavailable: {', '.join(unavailable)}]"
            msg.text = f"{msg.text}\n\n{notice}" if msg.text else notice
        return msg

    # -- helpers -----------------------------------------------------------

    def _describe_download_cause(self, exc: BaseException) -> str:
        """Build a token-safe suffix describing a download failure's cause.

        HTTP client exception strings can embed request URLs. Both Bot API URL
        forms carry the token in the path — the file download URL
        (``/file/bot<token>/…``) and the method URLs (``/bot<token>/getMe``,
        ``/bot<token>/getFile``) — so the configured token is removed first and
        any remaining Bot API URL is collapsed before the cause chain is logged.
        The cause class and message are kept to distinguish timeout / reset / TLS.
        """
        cause = exc.__cause__ or exc.__context__
        if cause is None:
            return ""
        token = self.config.get("bot_token", "")
        cause_msg = str(cause)
        if token:
            cause_msg = cause_msg.replace(token, "[redacted]")
        cause_msg = re.sub(r"api\.telegram\.org/bot[^/\s]+/", "api.telegram.org/bot[redacted]/", cause_msg)
        cause_msg = re.sub(r"api\.telegram\.org/file/\S+", "api.telegram.org/file/[redacted]", cause_msg)
        cause_msg = cause_msg[:200]
        return f" caused_by={type(cause).__name__}:{cause_msg}"

    async def _get_download_bot(self) -> Any:
        """Return a PTB Bot whose httpx client is bound to the Telegram loop.

        The Application's Bot is constructed on the manager/main loop, so its
        shared connection pool is bound there; opening a fresh download
        connection from the Telegram loop fails with "bound to a different
        event loop". This dedicated Bot is created and initialized on the
        Telegram loop, keeping its pool on the loop that performs downloads.
        """
        if self._download_bot is not None:
            return self._download_bot
        telegram_loop = self._tg_loop
        if telegram_loop is None:
            return None

        async def _init() -> Any:
            if self._download_bot is not None:
                return self._download_bot
            async with self._download_bot_lock:
                if self._download_bot is not None:
                    return self._download_bot
                from telegram import Bot

                bot = Bot(token=self.config.get("bot_token", ""))
                try:
                    await bot.initialize()
                except BaseException:
                    # initialize() performs a network getMe and may have left
                    # partially-open HTTPX clients. Close them (no-op if the
                    # request objects never initialized) before the error
                    # propagates, and never cache a bot that failed to start.
                    try:
                        await bot.shutdown()
                    except BaseException:
                        logger.debug("[Telegram] failed to close partially-initialized download bot", exc_info=True)
                    raise
                self._download_bot = bot
                return bot

        return await self._run_on_telegram_loop(_init())

    async def _download_inbound_file(self, bot: Any, file_id: str) -> tuple[int | None, bytearray | None]:
        """Fetch one file entirely on the event loop that owns PTB's HTTP client."""
        telegram_file = await bot.get_file(file_id)
        resolved_size = self._file_size(getattr(telegram_file, "file_size", None))
        if resolved_size is not None and resolved_size > TELEGRAM_MAX_INBOUND_FILE_BYTES:
            return resolved_size, None
        return resolved_size, await telegram_file.download_as_bytearray()

    async def _track_telegram_bridge_task(self, coroutine: Coroutine[Any, Any, Any]) -> Any:
        """Keep a main-loop submission visible to Telegram-loop shutdown."""
        task = asyncio.current_task()
        if task is None:
            coroutine.close()
            raise RuntimeError("Telegram bridge task is unavailable")
        self._tg_bridge_tasks.add(task)
        try:
            return await coroutine
        finally:
            self._tg_bridge_tasks.discard(task)

    async def _cancel_telegram_bridge_tasks(self) -> None:
        """Cancel and drain cross-loop PTB work before its event loop exits."""
        current = asyncio.current_task()
        pending = [task for task in self._tg_bridge_tasks if task is not current and not task.done()]
        for task in pending:
            task.cancel()
        if pending:
            done, still_pending = await asyncio.wait(pending, timeout=TELEGRAM_BRIDGE_DRAIN_TIMEOUT_SECONDS)
            if done:
                await asyncio.gather(*done, return_exceptions=True)
            if still_pending:
                logger.warning("[Telegram] %d inbound file download task(s) did not cancel promptly", len(still_pending))

    async def _shutdown_telegram_resources(self) -> None:
        """Drain startup and PTB work, then close both Bots on their owner loop.

        The application may not have reached running or even initialized: PTB
        Application.shutdown() then skips Bot.shutdown(), despite initialized
        HTTP clients from an interrupted/failed getMe request.
        """
        bootstrap = self._bootstrap_task
        if bootstrap is not None and bootstrap is not asyncio.current_task():
            if not bootstrap.done():
                bootstrap.cancel()
            await asyncio.gather(bootstrap, return_exceptions=True)
        await self._cancel_telegram_bridge_tasks()
        await self._shutdown_download_bot()
        application = self._application
        if application is not None:
            application_cleanup_failed = False
            try:
                if application.updater.running:
                    await application.updater.stop()
                if application.running:
                    await application.stop()
                await application.shutdown()
            except BaseException:
                application_cleanup_failed = True
                raise
            finally:
                # Idempotent after full shutdown, necessary after partial init.
                try:
                    await application.bot.shutdown()
                except Exception:
                    if not application_cleanup_failed:
                        raise
                    logger.warning("[Telegram] failed to shut down the application bot", exc_info=True)

    async def _shutdown_download_bot(self) -> None:
        """Close the download Bot's HTTPX clients on the Telegram loop.

        python-telegram-bot 22.7 exposes no ``Bot.session``; the HTTPX clients
        are closed by ``Bot.shutdown()``, which is a no-op if request resources
        were never initialized or were already shut down. Run on their owner loop.
        """
        bot = self._download_bot
        if bot is None:
            return
        try:
            await bot.shutdown()
        except Exception:
            logger.debug("[Telegram] failed to shut down download bot", exc_info=True)
        self._download_bot = None

    async def _run_on_telegram_loop(self, coroutine: Coroutine[Any, Any, Any]) -> Any:
        """Await a PTB coroutine without using its HTTP client across event loops."""
        telegram_loop = self._tg_loop
        current_loop = asyncio.get_running_loop()
        if telegram_loop is None or telegram_loop is current_loop:
            return await coroutine

        if not telegram_loop.is_running():
            coroutine.close()
            raise RuntimeError("Telegram event loop is not running")

        tracked_coroutine = self._track_telegram_bridge_task(coroutine)
        try:
            future = asyncio.run_coroutine_threadsafe(tracked_coroutine, telegram_loop)
        except BaseException:
            # A concurrent shutdown can close the loop between the running
            # check and scheduling. Close the unscheduled coroutine explicitly.
            tracked_coroutine.close()
            coroutine.close()
            raise
        try:
            return await asyncio.wrap_future(future)
        except asyncio.CancelledError:
            future.cancel()
            raise

    @staticmethod
    def _stream_key(chat_id: str, thread_ts: str | None) -> str:
        return f"{chat_id}:{thread_ts or ''}"

    @staticmethod
    def _parse_message_id(value: str | None) -> int | None:
        try:
            return int(value) if value else None
        except (OverflowError, TypeError, ValueError):
            return None

    @staticmethod
    def _file_size(value: Any) -> int | None:
        if isinstance(value, bool):
            return None
        try:
            size = int(value)
        except (TypeError, ValueError):
            return None
        return size if size >= 0 else None

    @staticmethod
    def _safe_inbound_filename(value: Any, fallback: str) -> str:
        if not isinstance(value, str):
            return fallback
        try:
            candidate = normalize_filename(value.strip())
        except (UnicodeError, ValueError):
            return fallback
        if is_upload_staging_file(candidate) or any(ord(char) < 32 for char in candidate):
            return fallback
        return candidate

    @classmethod
    def _extract_inbound_files(cls, message: Any) -> list[dict[str, Any]]:
        files: list[dict[str, Any]] = []
        message_id = str(getattr(message, "message_id", "message"))

        # Materialize the PTB sequence once. Test doubles and partially shaped
        # update objects can expose a truthy but empty iterable here.
        photo_sizes = tuple(getattr(message, "photo", None) or ())
        if photo_sizes:
            photo = max(
                photo_sizes,
                key=lambda item: (
                    (cls._file_size(getattr(item, "width", None)) or 0) * (cls._file_size(getattr(item, "height", None)) or 0),
                    cls._file_size(getattr(item, "file_size", None)) or 0,
                ),
            )
            file_id = getattr(photo, "file_id", None)
            if isinstance(file_id, str) and file_id:
                file_unique_id = getattr(photo, "file_unique_id", None)
                files.append(
                    {
                        "type": "image",
                        "file_id": file_id,
                        "file_unique_id": file_unique_id if isinstance(file_unique_id, str) else None,
                        "filename": f"telegram-photo-{message_id}.jpg",
                        "mime_type": "image/jpeg",
                        "size": cls._file_size(getattr(photo, "file_size", None)),
                    }
                )

        document = getattr(message, "document", None)
        document_id = getattr(document, "file_id", None) if document is not None else None
        if isinstance(document_id, str) and document_id:
            file_unique_id = getattr(document, "file_unique_id", None)
            mime_type = getattr(document, "mime_type", None)
            if not isinstance(mime_type, str) or not mime_type:
                mime_type = "application/octet-stream"
            # Avoid the lazy system MIME database lookup on the Telegram event
            # loop. The original name is preferred; an opaque extension is safe.
            fallback = f"telegram-document-{message_id}.bin"
            files.append(
                {
                    "type": "file",
                    "file_id": document_id,
                    "file_unique_id": file_unique_id if isinstance(file_unique_id, str) else None,
                    "filename": cls._safe_inbound_filename(getattr(document, "file_name", None), fallback),
                    "mime_type": mime_type,
                    "size": cls._file_size(getattr(document, "file_size", None)),
                }
            )

        return files

    def _register_stream_message(self, key: str, *, message_id: int, last_text: str, last_edit_at: float) -> None:
        self._stream_messages.pop(key, None)
        while len(self._stream_messages) >= MAX_TRACKED_STREAM_MESSAGES:
            self._stream_messages.pop(next(iter(self._stream_messages)))
        self._stream_messages[key] = {
            "message_id": message_id,
            "last_edit_at": last_edit_at,
            "last_text": last_text,
        }

    @staticmethod
    def _is_retry_after(exc: Exception) -> bool:
        return getattr(exc, "retry_after", None) is not None

    @staticmethod
    def _retry_after_seconds(exc: Exception) -> float:
        value = getattr(exc, "retry_after", 0)
        if hasattr(value, "total_seconds"):
            return float(value.total_seconds())
        return float(value)

    @staticmethod
    def _is_not_modified(exc: Exception) -> bool:
        return "message is not modified" in str(exc).lower()

    @staticmethod
    def _split_message(text: str) -> list[str]:
        """Split *text* into chunks that fit Telegram's 4096 UTF-16 code unit limit.

        Slicing by code points hands sendMessage up to twice the limit when the
        reply is emoji-heavy (a non-BMP character costs 2 units), the retry policy
        repeats the rejected 400 "Message is too long" and the reply is dropped.
        A split can still fall inside one user-perceived character (a combining
        mark, a ZWJ emoji sequence, the second half of a flag emoji); every chunk
        stays within budget, so a seam is a rendering artifact, not a failed send.
        """
        chunks: list[str] = []
        current: list[str] = []
        units = 0
        for char in text:
            width = _utf16_width(char)
            if units + width > TELEGRAM_MAX_MESSAGE_LENGTH:
                chunks.append("".join(current))
                current = []
                units = 0
            current.append(char)
            units += width
        chunks.append("".join(current))
        return chunks

    async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None:
        """Send a 'Working on it...' reply and register it as the stream target."""
        if not self._application:
            return
        try:
            bot = self._application.bot
            numeric_chat_id = int(chat_id)
            kwargs: dict[str, Any] = {"chat_id": numeric_chat_id, "text": "Working on it...", "reply_to_message_id": reply_to_message_id}
            if numeric_chat_id > 0:
                kwargs["allow_sending_without_reply"] = True
            sent = await bot.send_message(**kwargs)
            self._register_stream_message(
                self._stream_key(chat_id, str(reply_to_message_id)),
                message_id=sent.message_id,
                last_text="Working on it...",
                last_edit_at=0.0,
            )
            logger.info("[Telegram] 'Working on it...' reply sent in chat=%s", chat_id)
        except Exception:
            logger.exception("[Telegram] failed to send running reply in chat=%s", chat_id)

    async def _bootstrap_telegram(self, application: Any) -> None:
        if self._running:
            await application.initialize()
        if self._running:
            await application.start()
        if self._running:
            await application.updater.start_polling()

    def _run_polling(self) -> None:
        """Run telegram polling in a dedicated thread."""
        application = self._application
        if application is None:
            return
        loop = asyncio.new_event_loop()
        self._tg_loop = loop
        asyncio.set_event_loop(loop)
        try:
            # Cannot use run_polling() because it calls add_signal_handler(),
            # which only works in the main thread.  Instead, manually
            # initialize the application and start the updater.
            self._bootstrap_task = loop.create_task(self._bootstrap_telegram(application))
            loop.run_until_complete(self._bootstrap_task)
            if self._running:
                loop.run_forever()
        except asyncio.CancelledError:
            # stop() cancelled and awaits bootstrap on this same loop.
            pass
        except Exception:
            if self._running:
                logger.exception("Telegram polling error")
        finally:
            self._running = False
            self._polling_cleanup_started = True
            try:
                cleanup = self._request_telegram_cleanup(loop)
                loop.run_until_complete(asyncio.wrap_future(cleanup, loop=loop))
            except Exception:
                logger.exception("Error during Telegram shutdown")
            finally:
                try:
                    loop.run_until_complete(loop.shutdown_asyncgens())
                except Exception:
                    logger.exception("Error shutting down Telegram async generators")
                finally:
                    self._bootstrap_task = None
                    asyncio.set_event_loop(None)
                    loop.close()

    def _check_user(self, user_id: int) -> bool:
        if self._allowed_users is None:
            return True
        return user_id in self._allowed_users

    @staticmethod
    def _telegram_display_name(user) -> str:
        full_name = getattr(user, "full_name", None)
        if isinstance(full_name, str) and full_name:
            return full_name
        username = getattr(user, "username", None)
        if isinstance(username, str) and username:
            return username
        return str(getattr(user, "id", ""))

    async def _bind_connection_from_start_token_on_main(
        self,
        update,
        state_token: str,
    ) -> bool:
        """Run the bind flow on the main event loop (SQLAlchemy session is bound there)."""
        if self._connection_repo is None or not state_token:
            return False

        state = await self._connection_repo.consume_oauth_state(provider="telegram", state=state_token)
        if state is None:
            await self._run_on_telegram_loop(update.message.reply_text("Telegram connection link is invalid or expired."))
            return True

        owner_user_id = state["owner_user_id"]
        user_id = str(update.effective_user.id)
        chat_id = str(update.effective_chat.id)
        connection = await self._connection_repo.upsert_connection(
            owner_user_id=owner_user_id,
            provider="telegram",
            external_account_id=user_id,
            external_account_name=self._telegram_display_name(update.effective_user),
            workspace_id=chat_id,
            workspace_name=None,
            metadata={
                "chat_id": chat_id,
                "chat_type": update.effective_chat.type,
                "telegram_username": getattr(update.effective_user, "username", None),
            },
            status="connected",
        )
        logger.info("[Telegram] bound chat=%s user=%s to DeerFlow user=%s connection=%s", chat_id, user_id, owner_user_id, connection["id"])
        await self._run_on_telegram_loop(update.message.reply_text("Telegram connected to DeerFlow."))
        return True

    async def _bind_connection_from_start_token(self, update, state_token: str) -> bool:
        """Dispatch the bind flow to the main loop if repo is configured."""
        if self._connection_repo is None or not state_token:
            return False

        if self._main_loop and self._main_loop.is_running():
            msg_id = getattr(getattr(update, "message", None), "message_id", None)
            scheduled = self._submit_threadsafe_coroutine(
                self._bind_connection_from_start_token_on_main(update, state_token),
                self._main_loop,
                name="bind_connection",
                msg_id=msg_id,
            )
            if not scheduled:
                logger.info("[Telegram] main loop stopped before channel connection bind could be scheduled")
            # Schedule counts as handled so /start does not fall through to help
            # while SQL runs on the main loop.
            return scheduled
        logger.warning("[Telegram] main loop not running, cannot bind channel connection")
        return False

    async def _attach_connection_identity(self, inbound: InboundMessage) -> InboundMessage:
        return await attach_connection_identity(
            inbound,
            repo=self._connection_repo,
            provider="telegram",
            workspace_id=inbound.chat_id,
        )

    def _get_bot_username(self, context) -> str | None:
        bot = getattr(context, "bot", None)
        username = getattr(bot, "username", None)
        if not username and self._application is not None:
            username = getattr(getattr(self._application, "bot", None), "username", None)
        return str(username) if username else None

    @staticmethod
    def _strip_bot_username_from_leading_command(text: str, bot_username: str | None) -> str:
        username = (bot_username or "").lstrip("@").lower()
        if not username or not text.startswith("/"):
            return text

        parts = text.split(maxsplit=1)
        command_token = parts[0]
        if "@" not in command_token:
            return text

        command_name, addressed_username = command_token[1:].rsplit("@", 1)
        if not command_name or addressed_username.lower() != username:
            return text

        normalized = f"/{command_name}"
        if len(parts) > 1:
            normalized = f"{normalized} {parts[1]}"
        return normalized

    async def _cmd_start(self, update, context) -> None:
        """Handle /start command."""
        args = getattr(context, "args", []) if context is not None else []
        if args:
            # Handle the deep-link bind token before applying allowed_users so a
            # browser-initiated bind can bootstrap a new external identity.
            handled = await self._bind_connection_from_start_token(update, str(args[0]))
            if handled:
                return
        if not self._check_user(update.effective_user.id):
            return
        await update.message.reply_text("Welcome to DeerFlow! Send me a message to start a conversation.\nType /help for available commands.")

    async def _process_incoming_with_reply(
        self,
        chat_id: str,
        msg_id: int,
        inbound: InboundMessage,
        *,
        reservation: InboundReservation | None = None,
    ) -> None:
        try:
            await self._send_running_reply(chat_id, msg_id)
            if reservation is None:
                await self.bus.publish_inbound(inbound)
            else:
                self._commit_reserved_inbound(reservation, inbound)
        finally:
            if reservation is not None:
                reservation.release()

    async def _process_incoming_with_identity(
        self,
        chat_id: str,
        msg_id: int,
        inbound: InboundMessage,
        *,
        reservation: InboundReservation | None = None,
    ) -> None:
        """Attach connection identity and dispatch on the main event loop.

        This must run on the main loop because _attach_connection_identity
        uses a SQLAlchemy session factory bound to that loop.  Calling it
        from the Telegram polling thread's own loop raises
        ``RuntimeError: got Future attached to a different loop``.
        """
        inbound = await self._attach_connection_identity(inbound)
        await self._process_incoming_with_reply(chat_id, msg_id, inbound, reservation=reservation)

    async def _cmd_generic(self, update, context) -> None:
        """Forward slash commands to the channel manager."""
        if not self._check_user(update.effective_user.id):
            return

        text = self._strip_bot_username_from_leading_command(update.message.text.strip(), self._get_bot_username(context))
        chat_id = str(update.effective_chat.id)
        user_id = str(update.effective_user.id)
        msg_id = str(update.message.message_id)

        # Use the same topic_id logic as _on_text so that commands
        # like /new target the correct thread mapping.
        if update.effective_chat.type == "private":
            topic_id = None
        else:
            reply_to = update.message.reply_to_message
            if reply_to:
                topic_id = str(reply_to.message_id)
            else:
                topic_id = msg_id

        inbound = self._make_inbound(
            chat_id=chat_id,
            user_id=user_id,
            text=text,
            msg_type=InboundMessageType.COMMAND,
            thread_ts=msg_id,
            metadata={"message_id": msg_id},
        )
        inbound.topic_id = topic_id

        if self._main_loop and self._main_loop.is_running():
            reservation = self._reserve_inbound(inbound)
            if reservation is None:
                return
            try:
                scheduled = self._submit_threadsafe_coroutine(
                    self._process_incoming_with_identity(
                        chat_id,
                        update.message.message_id,
                        inbound,
                        reservation=reservation,
                    ),
                    self._main_loop,
                    name="process_incoming_with_identity",
                    msg_id=update.message.message_id,
                    reservation=reservation,
                )
                if not scheduled:
                    logger.info("[Telegram] main loop stopped before reserved command could be scheduled")
            except Exception:
                reservation.release()
                raise
        else:
            logger.warning("[Telegram] Main loop not running. Cannot publish inbound message.")

    async def _on_text(self, update, context) -> None:
        """Handle regular text, photo, and document messages."""
        if not self._check_user(update.effective_user.id):
            return

        message = update.message
        message_text = getattr(message, "text", None)
        caption = getattr(message, "caption", None)
        raw_text = message_text if isinstance(message_text, str) else caption if isinstance(caption, str) else ""
        text = raw_text.strip()
        if text:
            text = self._strip_bot_username_from_leading_command(text, self._get_bot_username(context))
        files = self._extract_inbound_files(message)
        if not text and not files:
            return

        chat_id = str(update.effective_chat.id)
        user_id = str(update.effective_user.id)
        msg_id = str(update.message.message_id)

        # topic_id determines which DeerFlow thread the message maps to.
        # In private chats, use None so that all messages share a single
        # thread (the store key becomes "channel:chat_id").
        # In group chats, use the reply-to message id or the current
        # message id to keep separate conversation threads.
        if update.effective_chat.type == "private":
            topic_id = None
        else:
            reply_to = update.message.reply_to_message
            if reply_to:
                topic_id = str(reply_to.message_id)
            else:
                topic_id = msg_id

        inbound = self._make_inbound(
            chat_id=chat_id,
            user_id=user_id,
            text=text,
            msg_type=InboundMessageType.CHAT,
            thread_ts=msg_id,
            files=files,
            metadata={"message_id": msg_id},
        )
        inbound.topic_id = topic_id

        if self._main_loop and self._main_loop.is_running():
            reservation = self._reserve_inbound(inbound)
            if reservation is None:
                return
            try:
                scheduled = self._submit_threadsafe_coroutine(
                    self._process_incoming_with_identity(
                        chat_id,
                        update.message.message_id,
                        inbound,
                        reservation=reservation,
                    ),
                    self._main_loop,
                    name="process_incoming_with_identity",
                    msg_id=update.message.message_id,
                    reservation=reservation,
                )
                if not scheduled:
                    logger.info("[Telegram] main loop stopped before reserved inbound could be scheduled")
            except Exception:
                reservation.release()
                raise
        else:
            logger.warning("[Telegram] Main loop not running. Cannot publish inbound message.")