fix(telegram): handle channel senders and fix guest reply ordering

gateway/platforms/telegram.py

@@ -476,6 +476,12 @@ def __init__(self, config: PlatformConfig):
         self._forum_command_registered: set[int] = set()
         # Lock per la registrazione sicura dei comandi nei forum supergroup
         self._forum_lock = asyncio.Lock()
+        # chat_id → guest_query_id for Bot API 10.0 guest replies
+        self._pending_guest_queries: Dict[str, str] = {}
+        # chat IDs that are guest-mode only (bot not a member)
+        self._guest_only_chats: set = set()
+        # accumulated send() content for guest chats; flushed via answerGuestQuery in on_processing_complete
+        self._guest_reply_buffer: Dict[str, str] = {}
         # DM Topics config from extra.dm_topics
         self._dm_topics_config: List[Dict[str, Any]] = self.config.extra.get("dm_topics", [])
         # Precomputed chat_ids that have DM topics configured (for O(1) root-DM ignore check)
@@ -1591,7 +1597,7 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
 
             try:
                 await self._app.updater.start_polling(
-                    allowed_updates=Update.ALL_TYPES,
+                    allowed_updates=[*Update.ALL_TYPES, "guest_message"],
                     drop_pending_updates=False,
                     error_callback=self._polling_error_callback_ref,
                 )
@@ -2098,6 +2104,15 @@ def _env_float(name: str, default: float) -> float:
             ))
             # Handle inline keyboard button callbacks (update prompts)
             self._app.add_handler(CallbackQueryHandler(self._handle_callback_query))
+            # Handle guest_message updates (Bot API 10.0 — not yet in PTB typed layer;
+            # the raw payload arrives in update.api_kwargs["guest_message"]).
+            try:
+                from telegram.ext import TypeHandler as _TypeHandler
+                self._app.add_handler(
+                    _TypeHandler(Update, self._handle_guest_message_update), group=1
+                )
+            except Exception as _th_err:
+                logger.warning("[%s] Could not register guest_message TypeHandler: %s", self.name, _th_err)
 
             # Start polling — retry initialize() for transient TLS resets
             try:
@@ -2160,7 +2175,7 @@ def _env_float(name: str, default: float) -> float:
                     url_path=webhook_path,
                     webhook_url=webhook_url,
                     secret_token=webhook_secret,
-                    allowed_updates=Update.ALL_TYPES,
+                    allowed_updates=[*Update.ALL_TYPES, "guest_message"],
                     drop_pending_updates=True,
                 )
                 self._webhook_mode = True
@@ -2193,7 +2208,7 @@ def _polling_error_callback(error: Exception) -> None:
                 self._polling_error_callback_ref = _polling_error_callback
 
                 await self._app.updater.start_polling(
-                    allowed_updates=Update.ALL_TYPES,
+                    allowed_updates=[*Update.ALL_TYPES, "guest_message"],
                     drop_pending_updates=True,
                     error_callback=_polling_error_callback,
                 )
@@ -2338,6 +2353,13 @@ async def send(
             return SendResult(success=True, message_id=None)
 
         try:
+            # Bot API 10.0 guest reply: buffer content and return immediately.
+            # Must run before the rich/legacy send paths — both use sendMessage
+            # which Telegram rejects with Forbidden when the bot is not a member.
+            if self._pending_guest_queries.get(chat_id) is not None or chat_id in self._guest_only_chats:
+                self._guest_reply_buffer[chat_id] = content
+                return SendResult(success=True, message_id=None)
+
             # Bot API 10.1 rich fast-path: send the raw agent markdown via
             # sendRichMessage so tables/task lists/etc. render natively. Falls
             # through to the legacy MarkdownV2 path on permanent/capability
@@ -2628,6 +2650,10 @@ async def send_or_update_status(
         message in place. If the edit fails (message deleted, too old, etc.)
         we drop the cached id and send fresh.
         """
+        # Guest chats have no existing message to edit and status messages would
+        # consume the one-shot query_id before the real answer is ready.
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=True, message_id=None)
         key = (str(chat_id), str(status_key))
         cached_id = self._status_message_ids.get(key)
         if cached_id is not None:
@@ -2667,6 +2693,11 @@ async def edit_message(
         if not self._bot:
             return SendResult(success=False, error="Not connected")
 
+        # "__no_edit__" is a stream_consumer sentinel meaning "no preview message
+        # was created"; treating it as a real message_id would crash int() below.
+        if message_id == "__no_edit__":
+            return SendResult(success=True, message_id=message_id)
+
         # Rich finalize (Bot API 10.1): when the completed content has
         # constructs the legacy MarkdownV2 edit degrades (tables → bullet
         # lists, task lists, <details>, block math) and rich is available,
@@ -3062,6 +3093,10 @@ async def send_draft(
         final ``sendMessage``/``sendRichMessage`` is what the user receives in
         their history).
         """
+        # Guest chats: draft streaming requires an existing message to animate;
+        # the bot is not a member, so suppress silently.
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=True, message_id=None)
         if not self._bot:
             return SendResult(success=False, error="not_connected")
 
@@ -4350,6 +4385,29 @@ def _missing_media_path_error(self, label: str, path: str) -> str:
             )
         return error
 
+    def _resolve_workspace_path(self, path: str) -> str:
+        """Translate /workspace/ Docker container paths to host equivalents.
+
+        When the terminal backend is Docker, tool-generated media lives at
+        /workspace/<file> inside the container. The gateway runs on the host
+        where that path does not exist. This looks up the active Docker env's
+        workspace host directory and rewrites the path so os.path.exists() works.
+        """
+        if not path.startswith("/workspace"):
+            return path
+        try:
+            from tools.terminal_tool import _active_environments  # type: ignore[import]
+            for env in list(_active_environments.values()):
+                wd = getattr(env, "_workspace_dir", None)
+                if wd:
+                    if path == "/workspace":
+                        return wd
+                    if path.startswith("/workspace/"):
+                        return wd + path[len("/workspace"):]
+        except Exception:
+            pass
+        return path
+
     def _telegram_media_too_large_note(self, label: str, file_size: Any, max_bytes: int) -> str:
         limit_mb = max(1, max_bytes // (1024 * 1024))
         try:
@@ -4389,7 +4447,11 @@ async def send_voice(
         """Send audio as a native Telegram voice message or audio file."""
         if not self._bot:
             return SendResult(success=False, error="Not connected")
-
+
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=False, error="guest_chat_no_media: bot is not a member of this group so Telegram blocks file uploads. Send the file to the user's private DMs and tell them in the group that the file is in their DMs.")
+
+        audio_path = self._resolve_workspace_path(audio_path)
         try:
             if not os.path.exists(audio_path):
                 return SendResult(success=False, error=self._missing_media_path_error("Audio", audio_path))
@@ -4616,6 +4678,10 @@ async def send_image_file(
         if not self._bot:
             return SendResult(success=False, error="Not connected")
 
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=False, error="guest_chat_no_media: bot is not a member of this group so Telegram blocks file uploads. Send the file to the user's private DMs and tell them in the group that the file is in their DMs.")
+
+        image_path = self._resolve_workspace_path(image_path)
         try:
             if not os.path.exists(image_path):
                 return SendResult(success=False, error=self._missing_media_path_error("Image", image_path))
@@ -4710,6 +4776,10 @@ async def send_document(
         if not self._bot:
             return SendResult(success=False, error="Not connected")
 
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=False, error="guest_chat_no_media: bot is not a member of this group so Telegram blocks file uploads. Send the file to the user's private DMs and tell them in the group that the file is in their DMs.")
+
+        file_path = self._resolve_workspace_path(file_path)
         try:
             if not os.path.exists(file_path):
                 return SendResult(success=False, error=self._missing_media_path_error("File", file_path))
@@ -4760,6 +4830,10 @@ async def send_video(
         if not self._bot:
             return SendResult(success=False, error="Not connected")
 
+        if self._pending_guest_queries.get(str(chat_id)) is not None or str(chat_id) in self._guest_only_chats:
+            return SendResult(success=False, error="guest_chat_no_media: bot is not a member of this group so Telegram blocks file uploads. Send the file to the user's private DMs and tell them in the group that the file is in their DMs.")
+
+        video_path = self._resolve_workspace_path(video_path)
         try:
             if not os.path.exists(video_path):
                 return SendResult(success=False, error=self._missing_media_path_error("Video", video_path))
@@ -5897,6 +5971,57 @@ def _effective_update_message(self, update: Update) -> Optional[Message]:
         """
         return getattr(update, "effective_message", None) or getattr(update, "message", None)
 
+    async def _handle_guest_message_update(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle guest_message updates (Bot API 10.0 guest bot feature).
+
+        Telegram delivers @mentions from chats the bot hasn't joined via the
+        ``guest_message`` update field.  PTB doesn't know this field yet so it
+        lands in ``update.api_kwargs``.  We parse the raw payload, store the
+        ``guest_query_id`` so ``send()`` can call ``answerGuestQuery``, then
+        route the message through the normal text-processing pipeline.
+        """
+        if not self._telegram_guest_mode():
+            return
+        raw_gm = update.api_kwargs.get("guest_message") if update.api_kwargs else None
+        if not raw_gm or not isinstance(raw_gm, dict):
+            return
+        logger.info("[%s] guest_message update received (update_id=%s)", self.name, update.update_id)
+
+        guest_query_id = raw_gm.get("guest_query_id")
+        if not guest_query_id:
+            logger.warning("[%s] guest_message missing guest_query_id, skipping", self.name)
+            return
+
+        try:
+            msg = Message.de_json(raw_gm, self._bot)
+        except Exception as exc:
+            logger.warning("[%s] Failed to parse guest_message payload: %s", self.name, exc)
+            return
+        if not msg:
+            return
+
+        text = msg.text or getattr(msg, "caption", None) or ""
+        if not text.strip():
+            return
+
+        chat_id_str = str(msg.chat.id) if msg.chat else ""
+        if not chat_id_str:
+            return
+
+        # Store guest_query_id so send() uses answerGuestQuery for the reply.
+        self._pending_guest_queries[chat_id_str] = guest_query_id
+        self._guest_only_chats.add(chat_id_str)
+
+        if not self._should_process_message(msg):
+            self._pending_guest_queries.pop(chat_id_str, None)
+            self._guest_only_chats.discard(chat_id_str)
+            return
+
+        event = self._build_message_event(msg, MessageType.TEXT, update_id=update.update_id)
+        event.text = self._clean_bot_trigger_text(event.text)
+        event = self._apply_telegram_group_observe_attribution(event)
+        self._enqueue_text_event(event)
+
     async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle incoming text messages.
 
@@ -6663,22 +6788,34 @@ def _build_message_event(
                     break
 
         # Build source
+        # When a user posts "as a channel" in a group, from_user is None but
+        # sender_chat carries the channel identity. Use sender_chat.id so the
+        # auth check can match it and the session key is stable across messages.
+        sender_chat = getattr(message, "sender_chat", None)
         source = self.build_source(
             chat_id=str(chat.id),
             chat_name=chat.title or (chat.full_name if hasattr(chat, "full_name") else None),
             chat_type=chat_type,
             user_id=(
                 str(user.id)
                 if user
-                else (str(chat.id) if chat_type in {"dm", "channel"} else None)
+                else (
+                    str(sender_chat.id)
+                    if sender_chat
+                    else (str(chat.id) if chat_type in {"dm", "channel"} else None)
+                )
             ),
             user_name=(
                 user.full_name
                 if user
                 else (
-                    chat.full_name
-                    if hasattr(chat, "full_name") and chat_type == "dm"
-                    else (chat.title if chat_type == "channel" else None)
+                    getattr(sender_chat, "title", None) or getattr(sender_chat, "username", None)
+                    if sender_chat
+                    else (
+                        chat.full_name
+                        if hasattr(chat, "full_name") and chat_type == "dm"
+                        else (chat.title if chat_type == "channel" else None)
+                    )
                 )
             ),
             thread_id=thread_id_str,
@@ -6809,6 +6946,43 @@ async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingO
         another agent run to swap it to 👍/👎 — which never happens if the
         cancellation was the last activity in the chat.
         """
+        # Flush buffered guest reply via answerGuestQuery (Bot API 10.0).
+        # All send() / send_draft() / send_or_update_status() calls during processing
+        # were silently buffered; we fire a single answerGuestQuery here with the
+        # complete response so the user sees the full answer, not a status fragment.
+        #
+        # Private-chat routing note: in groups the reply appears in the group chat as
+        # expected.  In P2P private chats between two regular users, Telegram cannot
+        # post a bot message into the conversation, so it surfaces the reply in the
+        # bot's own DM thread with the mentioning user instead.  This is Telegram API
+        # behaviour — our call is identical in both cases; the difference is how
+        # Telegram routes the answerGuestQuery result on its end.
+        _gc_id = str(getattr(event.source, "chat_id", None) or "")
+        if _gc_id:
+            _guest_qid = self._pending_guest_queries.pop(_gc_id, None)
+            _buffered = self._guest_reply_buffer.pop(_gc_id, "")
+            self._guest_only_chats.discard(_gc_id)
+            if _guest_qid and self._bot:
+                _plain = _strip_mdv2(self.format_message(_buffered)).strip() if _buffered else ""
+                _plain = _plain[:4096] or "​"  # zero-width space if somehow empty
+                _gq_result = {
+                    "type": "article",
+                    "id": "reply",
+                    "title": "Reply",
+                    "input_message_content": {"message_text": _plain},
+                }
+                try:
+                    await self._bot.do_api_request(
+                        "answerGuestQuery",
+                        api_kwargs={"guest_query_id": _guest_qid, "result": _gq_result},
+                    )
+                    logger.info("[%s] answerGuestQuery flushed (chat=%s)", self.name, _gc_id)
+                except Exception as _flush_err:
+                    logger.warning(
+                        "[%s] answerGuestQuery flush failed (chat=%s): %s",
+                        self.name, _gc_id, _flush_err,
+                    )
+
         if not self._reactions_enabled():
             return
         chat_id = getattr(event.source, "chat_id", None)