# NousResearch/hermes-agent · Gateway media and voice delivery bugs and fixes ## Open now - Nothing open on GitHub. These issues and pull requests cluster around failures in the messaging gateway to correctly deliver media files, preserve queued voice and photo messages, and avoid echoing local file paths instead of uploading attachments. A secondary thread covers voice message transcription (STT) regressions during active agent runs. ## TL;DR - 16 member(s): 0 open issue(s), 5 closed issue(s), 0 open PR(s), 4 merged PR(s), 7 PR(s) closed without merging. - A fix was merged to main in PR #3919. Verify it addresses your case. - **Recommended:** read PR #3919 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-2339,pr-7948`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text. - Members outside that structural core may belong to a different underlying problem. - Pairwise link density `0.10`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #1803: MEDIA: tag fails to send files - document upload not working - #6249: Telegram MEDIA image delivery can echo local file path as text instead of uploading attachment - #7925: \[Bug\] Queued voice-message STT speech-to-text regression in gateway - #11538: bug: queued photo messages lose media during vision enrichment - #60845: bug(gateway): queued follow-up responses bypass MEDIA extraction — attachment delivered as plain text path ## Prior & competing PRs - **#3919: fix: background task media delivery + vision download timeout**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `cli-config.yaml.example`, `gateway/platforms/telegram.py`, `gateway/run.py`, `hermes_cli/config.py`, `tools/vision_tools.py`, `website/docs/user-guide/configuration.md` - fix hunks: - `cli-config.yaml.example` ```diff @@ -324,6 +324,9 @@ compression: # vision: # provider: "auto" # model: "" # e.g. "google/gemini-2.5-flash", "openai/gpt-4o" +# timeout: 30 # LLM API call timeout (seconds) +# download_timeout: 30 # Image HTTP download timeout (seconds) +# # Increase for slow connections or self-hosted image servers # # # Web page scraping / summarization + browser page text extraction # web_extract: ``` - `gateway/platforms/telegram.py` ```diff @@ -123,6 +123,7 @@ def __init__(self, config: PlatformConfig): super().__init__(config, Platform.TELEGRAM) self._app: Optional[Application] = None self._bot: Optional[Bot] = None + self._webhook_mode: bool = False self._mention_patterns = self._compile_mention_patterns() self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first' # Buffer rapid/album photo updates so Telegram image bursts are handled ``` ```diff @@ -458,7 +459,19 @@ async def _setup_dm_topics(self) -> None: self._persist_dm_topic_thread_id(int(chat_id), topic_name, thread_id) async def connect(self) -> bool: - """Connect to Telegram and start polling for updates.""" + """Connect to Telegram via polling or webhook. + + By default, uses long polling (outbound connection to Telegram). + If ``TELEGRAM_WEBHOOK_URL`` is set, starts an HTTP webhook server + instead. Webhook mode is useful for cloud deployments (Fly.io, + Railway) where inbound HTTP can wake a suspended machine. + + Env vars for webhook mode:: + + TELEGRAM_WEBHOOK_URL Public HTTPS URL (e.g. https://app.fly.dev/telegram) + TELEGRAM_WEBHOOK_PORT Local listen port (default 8443) + TELEGRAM_WEBHOOK_SECRET Secret token for update verification + """ if not TELEGRAM_AVAILABLE: logger.error( "[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot", ``` ```diff @@ -552,27 +565,57 @@ async def connect(self) -> bool: else: raise await self._app.start() - loop = asyncio.get_running_loop() - def _polling_error_callback(error: Exception) -> None: - if self._polling_error_task and not self._polling_error_task.done(): - return - if self._looks_like_polling_conflict(error): - self._polling_error_task = loop.create_task(self._handle_polling_conflict(error)) - elif self._looks_like_network_error(error): - logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error) - self._polling_error_task = loop.create_task(self._handle_polling_network_error(error)) - else: - logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True) + # Decide between webhook and polling mode + webhook_url = os.getenv("TELEGRAM_WEBHOOK_URL", "").strip() + + if webhook_url: + # ── Webhook mode ───────────────────────────────────── + # Telegram pushes updates to our HTTP endpoint. This + # enables cloud platforms (Fly.io, Railway) to auto-wake + # suspended machines on inbound HTTP traffic. + webhook_port = int(os.getenv("TELEGRAM_WEBHOOK_PORT", "8443")) + webhook_secret = [redacted:secret-assignment]"TELEGRAM_WEBHOOK_SECRET", "").strip() or None + from urllib.parse import urlparse + webhook_path = urlparse(webhook_url).path or "/telegram" + + await self._app.updater.start_webhook( + listen="0.0.0.0", + port=webhook_port, + url_path=webhook_path, + webhook_url=webhook_url, + secret_token=[redacted:secret-assignment] + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=True, + ) + self._webhook_mode = True + logger.info( + "[%s] Webhook server listening on 0.0.0.0:%d%s", + self.name, webhook_port, webhook_path, + ) + else: + # ── Polling mode (default) ─────────────────────────── + loop = asyncio.get_running_loop() + + def _polling_error_callback(error: Exception) -> None: + if self._polling_error_task and not self._polling_error_task.done(): + return + if self._looks_like_polling_conflict(error): + self._polling_error_task = loop.create_task(self._handle_polling_conflict(error)) + elif self._looks_like_network_error(error): + logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error) + self._polling_error_task = loop.create_task(self._handle_polling_network_error(error)) + else: + logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True) - # Store reference for retry use in _handle_polling_conflict - self._polling_error_callback_ref = _polling_error_callback + # Store reference for retry use in _handle_polling_conflict + self._polling_error_callback_ref = _polling_error_callback - await self._app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=True, - error_callback=_polling_error_callback, - ) + await self._app.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=True, + error_callback=_polling_error_callback, + ) # Register bot commands so Telegram shows a hint menu when users type / # List is derived from the central COMMAND_REGISTRY — adding a new ``` ```diff @@ -592,7 +635,8 @@ def _polling_error_callback(error: Exception) -> None: ) self._mark_connected() - logger.info("[%s] Connected and polling for Telegram updates", self.name) + mode = "webhook" if self._webhook_mode else "polling" + logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode) # Set up DM topics (Bot API 9.4 — Private Chat Topics) # Runs after connection is established so the bot can call createForumTopic. ``` ```diff @@ -620,7 +664,7 @@ def _polling_error_callback(error: Exception) -> None: return False async def disconnect(self) -> None: - """Stop polling, cancel pending album flushes, and disconnect.""" + """Stop polling/webhook, cancel pending album flushes, and disconnect.""" pending_media_group_tasks = list(self._media_group_tasks.values()) for task in pending_media_group_tasks: task.cancel() ``` - `gateway/run.py` ```diff @@ -3891,7 +3891,7 @@ def run_sync(): # Send media files for media_path in (media_files or []): try: - await adapter.send_file( + await adapter.send_document( chat_id=source.chat_id, file_path=media_path, ) ``` - `hermes_cli/config.py` ```diff @@ -223,7 +223,8 @@ def ensure_hermes_home(): "model": "", # e.g. "google/gemini-2.5-flash", "gpt-4o" "base_url": "", # direct OpenAI-compatible endpoint (takes precedence over provider) "api_key": "", # API key for base_url (falls back to OPENAI_API_KEY) - "timeout": 30, # seconds — increase for slow local vision models + "timeout": 30, # seconds — LLM API call timeout; increase for slow local vision models + "download_timeout": 30, # seconds — image HTTP download timeout; increase for slow connections }, "web_extract": { "provider": "auto", ``` - `tools/vision_tools.py` ```diff @@ -45,6 +45,28 @@ _debug = DebugSession("vision_tools", env_var="VISION_TOOLS_DEBUG") +# Configurable HTTP download timeout for _download_image(). +# Separate from auxiliary.vision.timeout which governs the LLM API call. +# Resolution: config.yaml auxiliary.vision.download_timeout → env var → 30s default. +def _resolve_download_timeout() -> float: + env_val = os.getenv("HERMES_VISION_DOWNLOAD_TIMEOUT", "").strip() + if env_val: + try: + return float(env_val) + except ValueError: + pass + try: + from hermes_cli.config import load_config + cfg = load_config() + val = cfg.get("auxiliary", {}).get("vision", {}).get("download_timeout") + if val is not None: + return float(val) + except Exception: + pass + return 30.0 + +_VISION_DOWNLOAD_TIMEOUT = _resolve_download_timeout() + def _validate_image_url(url: str) -> bool: """ ``` ```diff @@ -146,7 +168,7 @@ async def _ssrf_redirect_guard(response): # Enable follow_redirects to handle image CDNs that redirect (e.g., Imgur, Picsum) # SSRF: event_hooks validates each redirect target against private IP ranges async with httpx.AsyncClient( - timeout=30.0, + timeout=_VISION_DOWNLOAD_TIMEOUT, follow_redirects=True, event_hooks={"response": [_ssrf_redirect_guard]}, ) as client: ``` ```diff @@ -183,6 +205,10 @@ async def _ssrf_redirect_guard(response): exc_info=True, ) + if last_error is None: + raise RuntimeError( + f"_download_image exited retry loop without attempting (max_retries={max_retries})" + ) raise last_error ``` - `website/docs/user-guide/configuration.md` ```diff @@ -1018,7 +1018,8 @@ auxiliary: model: "" # e.g. "openai/gpt-4o", "google/gemini-2.5-flash" base_url: "" # Custom OpenAI-compatible endpoint (overrides provider) api_key: "" # API key for base_url (falls back to OPENAI_API_KEY) - timeout: 30 # seconds — increase for slow local vision models + timeout: 30 # seconds — LLM API call; increase for slow local vision models + download_timeout: 30 # seconds — image HTTP download; increase for slow connections # Web page summarization + browser page text extraction web_extract: ``` ```diff @@ -1042,7 +1043,7 @@ auxiliary: ``` :::tip -Each auxiliary task has a configurable `timeout` (in seconds). Defaults: vision 30s, web_extract 30s, approval 30s, compression 120s. Increase these if you use slow local models for auxiliary tasks. +Each auxiliary task has a configurable `timeout` (in seconds). Defaults: vision 30s, web_extract 30s, approval 30s, compression 120s. Increase these if you use slow local models for auxiliary tasks. Vision also has a separate `download_timeout` (default 30s) for the HTTP image download — increase this for slow connections or self-hosted image servers. ::: :::info ``` - **#7948: fix(gateway): preserve queued voice events for STT**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `gateway/run.py`, `tests/gateway/test_queue_consumption.py`, `tests/gateway/test_stt_config.py` - fix hunks: - `gateway/run.py` ```diff @@ -352,19 +352,14 @@ def _build_media_placeholder(event) -> str: return "\n".join(parts) -def _dequeue_pending_text(adapter, session_key: str) -> str | None: - """Consume and return the text of a pending queued message. +def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None: + """Consume and return the full pending event for a session. - Preserves media context for captionless photo/document events by - building a placeholder so the message isn't silently dropped. + Queued follow-ups must preserve their media metadata so they can re-enter + the normal image/STT/document preprocessing path instead of being reduced + to a placeholder string. """ - event = adapter.get_pending_message(session_key) - if not event: - return None - text = event.text - if not text and getattr(event, "media_urls", None): - text = _build_media_placeholder(event) - return text + return adapter.get_pending_message(session_key) def _check_unavailable_skill(command_name: str) -> str | None: ``` ```diff @@ -2775,6 +2770,162 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: del self._running_agents[_quick_key] self._running_agents_ts.pop(_quick_key, None) + async def _prepare_inbound_message_text( + self, + *, + event: MessageEvent, + source: SessionSource, + history: List[Dict[str, Any]], + ) -> Optional[str]: + """Prepare inbound event text for the agent. + + Keep the normal inbound path and the queued follow-up path on the same + preprocessing pipeline so sender attribution, image enrichment, STT, + document notes, reply context, and @ references all behave the same. + """ + history = history or [] + message_text = event.text or "" + + _is_shared_thread = ( + source.chat_type != "dm" + and source.thread_id + and not getattr(self.config, "thread_sessions_per_user", False) + ) + if _is_shared_thread and source.user_name: + message_text = f"[{source.user_name}] {message_text}" + + if event.media_urls: + image_paths = [] + audio_paths = [] + for i, path in enumerate(event.media_urls): + mtype = event.media_types[i] if i < len(event.media_types) else "" + if mtype.startswith("image/") or event.message_type == MessageType.PHOTO: + image_paths.append(path) + if mtype.startswith("audio/") or event.message_type in (MessageType.VOICE, MessageType.AUDIO): + audio_paths.append(path) + + if image_paths: + message_text = await self._enrich_message_with_vision( + message_text, + image_paths, + ) + + if audio_paths: + message_text = await self._enrich_message_with_transcription( + message_text, + audio_paths, + ) + _stt_fail_markers = ( + "No STT provider", + "STT is disabled", + "can't listen", + "VOICE_TOOLS_OPENAI_KEY", + ) + if any(marker in message_text for marker in _stt_fail_markers): + _stt_adapter = self.adapters.get(source.platform) + _stt_meta = {"thread_id": source.thread_id} if source.thread_id else None + if _stt_adapter: + try: + _stt_msg = ( + "🎤 I received your voice message but can't transcribe it — " + "no speech-to-text provider is configured.\n\n" + "To enable voice: install faster-whisper " + "(`pip install faster-whisper` in the Hermes venv) " + "and set `stt.enabled: true` in config.yaml, " + "then /restart the gateway." + ) + if self._has_setup_skill(): + _stt_msg += "\n\nFor full setup instructions, type: `/skill hermes-agent-setup`" + await _stt_adapter.send( + source.chat_id, + _stt_msg, + metadata=_stt_meta, + ) + except Exception: + pass + + if event.media_urls and event.message_type == MessageType.DOCUMENT: + import mimetypes as _mimetypes + + _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"} + for i, path in enumerate(event.media_urls): + mtype = event.media_types[i] if i < len(event.media_types) else "" + if mtype in ("", "application/octet-stream"): + import os as _os2 + + _ext = _os2.path.splitext(path)[1].lower() + if _ext in _TEXT_EXTENSIONS: + mtype = "text/plain" + else: + guessed, _ = _mimetypes.guess_type(path) + if guessed: + mtype = guessed + if not mtype.startswith(("application/", "text/")): + continue + + import os as _os + import re as _re + + basename = _os.path.basename(path) + parts = basename.split("_", 2) + display_name = parts[2] if len(parts) >= 3 else basename + display_name = _re.sub(r'[^\w.\- ]', '_', display_name) + + if mtype.startswith("text/"): + context_note = ( + f"[The user sent a text document: '{display_name}'. " + f"Its content has been included below. " + f"The file is also saved at: {path}]" + ) + else: + context_note = ( + f"[The user sent a document: '{display_name}'. " + f"The file is saved at: {path}. " + f"Ask the user what they'd like you to do with it.]" + ) + message_text = f"{context_note}\n\n{message_text}" + + if getattr(event, "reply_to_text", None) and event.reply_to_message_id: + reply_snippet = event.reply_to_text[:500] + found_in_history = any( + reply_snippet[:200] in (msg.get("content") or "") + for msg in history + if msg.get("role") in ("assistant", "user", "tool") + ) + if not found_in_history: + message_text = f'[Replying to: "{reply_snippet}"]\n\n{message_text}' + + if "@" in message_text: + try: + from agent.context_references import preprocess_context_references_async + from agent.model_metadata import get_model_context_length + + _msg_cwd = os.environ.get("MESSAGING_CWD", os.path.expanduser("~")) + _msg_ctx_len = get_model_context_length( + self._model, + base_url=self._base_url or "", + ) + _ctx_result = await preprocess_context_references_async( + message_text, + cwd=_msg_cwd, + context_length=_msg_ctx_len, + allowed_root=_msg_cwd, + ) + if _ctx_result.blocked: + _adapter = self.adapters.get(source.platform) + if _adapter: + await _adapter.send( + source.chat_id, + "\n".join(_ctx_result.warnings) or "Context injection refused.", + ) + return None + if _ctx_result.expanded: + message_text = _ctx_result.message + except Exception as exc: + logger.debug("@ context reference expansion failed: %s", exc) + + return message_text + async def _handle_message_with_agent(self, event, source, _quick_key: str): """Inner handler that runs under the _running_agents sentinel guard.""" _msg_start_time = time.time() ``` ```diff @@ -3215,149 +3366,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str): # attachments (documents, audio, etc.) are not sent to the vision # tool even when they appear in the same message. # ----------------------------------------------------------------- - message_text = event.text or "" - - # ----------------------------------------------------------------- - # Sender attribution for shared thread sessions. - # - # When multiple users share a single thread session (the default for - # threads), prefix each message with [sender name] so the agent can - # tell participants apart. Skip for DMs (single-user by nature) and - # when per-user thread isolation is explicitly enabled. - # ----------------------------------------------------------------- - _is_shared_thread = ( - source.chat_type != "dm" - and source.thread_id - and not getattr(self.config, "thread_sessions_per_user", False) + message_text = await self._prepare_inbound_message_text( + event=event, + source=source, + history=history, ) - if _is_shared_thread and source.user_name: - message_text = f"[{source.user_name}] {message_text}" - - if event.media_urls: - image_paths = [] - for i, path in enumerate(event.media_urls): - # Check media_types if available; otherwise infer from message type - mtype = event.media_types[i] if i < len(event.media_types) else "" - is_image = ( - mtype.startswith("image/") - or event.message_type == MessageType.PHOTO - ) - if is_image: - image_paths.append(path) - if image_paths: - message_text = await self._enrich_message_with_vision( - message_text, image_paths - ) - - # ----------------------------------------------------------------- - # Auto-transcribe voice/audio messages sent by the user - # ----------------------------------------------------------------- - if event.media_urls: - audio_paths = [] - for i, path in enumerate(event.media_urls): - mtype = event.media_types[i] if i < len(event.media_types) else "" - is_audio = ( - mtype.startswith("audio/") - or event.message_type in (MessageType.VOICE, MessageType.AUDIO) - ) - if is_audio: - audio_paths.append(path) - if audio_paths: - message_text = await self._enrich_message_with_transcription( - message_text, audio_paths - ) - # If STT failed, send a direct message to the user so they - # know voice isn't configured — don't rely on the agent to - # relay the error clearly. - _stt_fail_markers = ( - "No STT provider", - "STT is disabled", - "can't listen", - "VOICE_TOOLS_OPENAI_KEY", - ) - if any(m in message_text for m in _stt_fail_markers): - _stt_adapter = self.adapters.get(source.platform) - _stt_meta = {"thread_id": source.thread_id} if source.thread_id else None - if _stt_adapter: - try: - _stt_msg = ( - "🎤 I received your voice message but can't transcribe it — " - "no speech-to-text provider is configured.\n\n" - "To enable voice: install faster-whisper " - "(`pip install faster-whisper` in the Hermes venv) " - "and set `stt.enabled: true` in config.yaml, " - "then /restart the gateway." - ) - # Point to setup skill if it's installed - if self._has_setup_skill(): - _stt_msg += "\n\nFor full setup instructions, type: `/skill hermes-agent-setup`" - await _stt_adapter.send( - source.chat_id, _stt_msg, - metadata=_stt_meta, - ) - except Exception: - pass - - # ----------------------------------------------------------------- - # Enrich document messages with context notes for the agent - # ----------------------------------------------------------------- - if event.media_urls and event.message_type == MessageType.DOCUMENT: - import mimetypes as _mimetypes - _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"} - for i, path in enumerate(event.media_urls): - mtype = event.media_types[i] if i < len(event.media_types) else "" - # Fall back to extension-based detection when MIME type is unreliable. - if mtype in ("", "application/octet-stream"): - import os as _os2 - _ext = _os2.path.splitext(path)[1].lower() - if _ext in _TEXT_EXTENSIONS: - mtype = "text/plain" - else: - guessed, _ = _mimetypes.guess_type(path) - if guessed: - mtype = guessed - if not mtype.startswith(("application/", "text/")): - continue - # Extract display filename by stripping the doc_{uuid12}_ prefix - import os as _os - basename = _os.path.basename(path) - # Format: doc_<12hex>_ - parts = basename.split("_", 2) - display_name = parts[2] if len(parts) >= 3 else basename - # Sanitize to prevent prompt injection via filenames - import re as _re - display_name = _re.sub(r'[^\w.\- ]', '_', display_name) - - if mtype.startswith("text/"): - context_note = ( - f"[The user sent a text document: '{display_name}'. " - f"Its content has been included below. " - f"The file is also saved at: {path}]" - ) - else: - context_note = ( - f"[The user sent a document: '{display_name}'. " - f"The file is saved at: {path}. " - f"Ask the user what they'd like you to do with it.]" - ) - message_text = f"{context_note}\n\n{message_text}" - - # ----------------------------------------------------------------- - # Inject reply context when user replies to a message not in history. - # Telegram (and other platforms) let users reply to specific messages, - # but if the quoted message is from a previous session, cron delivery, - # or background task, the agent has no context about what's being - # referenced. Prepend the quoted text so the agent understands. (#1594) - # ----------------------------------------------------------------- - if getattr(event, 'reply_to_text', None) and event.reply_to_message_id: - reply_snippet = event.reply_to_text[:500] - found_in_history = any( - reply_snippet[:200] in (msg.get("content") or "") - for msg in history - if msg.get("role") in ("assistant", "user", "tool") - ) - if not found_in_history: - message_text = f'[Replying to: "{reply_snippet}"]\n\n{message_text}' + if message_text is None: + return try: # Emit agent:start hook ``` ```diff @@ -3369,30 +3384,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str): } await self.hooks.emit("agent:start", hook_ctx) - # Expand @ context references (@file:, @folder:, @diff, etc.) - if "@" in message_text: - try: - from agent.context_references import preprocess_context_references_async - from agent.model_metadata import get_model_context_length - _msg_cwd = os.environ.get("MESSAGING_CWD", os.path.expanduser("~")) - _msg_ctx_len = get_model_context_length( - self._model, base_url=self._base_url or "") - _ctx_result = await preprocess_context_references_async( - message_text, cwd=_msg_cwd, - context_length=_msg_ctx_len, allowed_root=_msg_cwd) - if _ctx_result.blocked: - _adapter = self.adapters.get(source.platform) - if _adapter: - await _adapter.send( - source.chat_id, - "\n".join(_ctx_result.warnings) or "Context injection refused.", - ) - return - if _ctx_result.expanded: - message_text = _ctx_result.message - except Exception as exc: - logger.debug("@ context reference expansion failed: %s", exc) - # Run the agent agent_result = await self._run_agent( message=message_text, ``` ```diff @@ -8057,17 +8048,16 @@ async def _notify_long_running(): # Get pending message from adapter. # Use session_key (not source.chat_id) to match adapter's storage keys. + pending_event = None pending = None if result and adapter and session_key: - if result.get("interrupted"): - pending = _dequeue_pending_text(adapter, session_key) - if not pending and result.get("interrupt_message"): - pending = result.get("interrupt_message") - else: - pending = _dequeue_pending_text(adapter, session_key) - if pending: - logger.debug("Processing queued message after agent completion: '%s...'", pending[:40]) - + pending_event = _dequeue_pending_event(adapter, session_key) + if result.get("interrupted") and not pending_event and result.get("interrupt_message"): + pending = result.get("interrupt_message") + elif pending_event: + pending = pending_event.text or _build_media_placeholder(pending_event) + logger.debug("Processing queued message after agent completion: '%s...'", pending[:40]) + # Safety net: if the pending text is a slash command (e.g. "/stop", # "/new"), discard it — commands should never be passed to the agent # as user input. The primary fix is in base.py (commands bypass the ``` ```diff @@ -8085,27 +8075,29 @@ async def _notify_long_running(): "commands must not be passed as agent input", _pending_cmd_word, ) + pending_event = None pending = None except Exception: pass - if self._draining and pending: + if self._draining and (pending_event or pending): logger.info( "Discarding pending follow-up for session %s during gateway %s", session_key[:20] if session_key else "?", self._status_action_label(), ) + pending_event = None pending = None - if pending: + if pending_event or pending: logger.debug("Processing pending message: '%s...'", pending[:40]) - + # Clear the adapter's interrupt event so the next _run_agent call # doesn't immediately re-trigger the interrupt before the new agent # even makes its first API call (this was causing an infinite loop). if adapter and hasattr(adapter, '_active_sessions') and session_key and session_key in adapter._active_sessions: adapter._active_sessions[session_key].clear() - + # Cap recursion depth to prevent resource exhaustion when the # user sends multiple messages while the agent keeps failing. (#816) if _interrupt_depth >= self._MAX_INTERRUPT_DEPTH: ``` ```diff @@ -8114,9 +8106,10 @@ async def _notify_long_running(): "queueing message instead of recursing.", _interrupt_depth, session_key, ) - # Queue the pending message for normal processing on next turn adapter = self.adapters.get(source.platform) - if adapter and hasattr(adapter, 'queue_message'): + if adapter and pending_event: + merge_pending_message_event(adapter._pending_messages, session_key, pending_event) + elif adapter and hasattr(adapter, 'queue_message'): adapter.queue_message(session_key, pending) return result_holder[0] or {"final_response": response, "messages": history} ``` ```diff @@ -8138,16 +8131,30 @@ async def _notify_long_running(): # interrupted." is just noise; the user already knows they sent a # new message). - # Process the pending message with updated history updated_history = result.get("messages", history) + next_source = source + next_message = pending + next_message_id = None + if pending_event is not None: + next_source = getattr(pending_event, "source", None) or source + next_message = await self._prepare_inbound_message_text( + event=pending_event, + source=next_source, + history=updated_history, + ) + if next_message is None: + return result + next_message_id = getattr(pending_event, "message_id", None) + return await self._run_agent( - message=pending, + message=next_message, context_prompt=context_prompt, history=updated_history, - source=source, + source=next_source, session_id=session_id, session_key=session_key, _interrupt_depth=_interrupt_depth + 1, + event_message_id=next_message_id, ) finally: # Stop progress sender, interrupt monitor, and notification task ``` - `tests/gateway/test_queue_consumption.py` ```diff @@ -10,6 +10,7 @@ import pytest +from gateway.run import _dequeue_pending_event from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, ``` ```diff @@ -79,6 +80,26 @@ def test_get_pending_message_consumes_and_clears(self): # Should be consumed (cleared) assert adapter.get_pending_message(session_key) is None + def test_dequeue_pending_event_preserves_voice_media_metadata(self): + adapter = _StubAdapter() + session_key = "telegram:user:voice" + event = MessageEvent( + text="", + message_type=MessageType.VOICE, + source=MagicMock(chat_id="123", platform=Platform.TELEGRAM), + message_id="voice-q1", + media_urls=["/tmp/voice.ogg"], + media_types=["audio/ogg"], + ) + adapter._pending_messages[session_key] = event + + retrieved = _dequeue_pending_event(adapter, session_key) + + assert retrieved is event + assert retrieved.media_urls == ["/tmp/voice.ogg"] + assert retrieved.media_types == ["audio/ogg"] + assert adapter.get_pending_message(session_key) is None + def test_queue_does_not_set_interrupt_event(self): """The whole point of /queue — no interrupt signal.""" adapter = _StubAdapter() ``` - `tests/gateway/test_stt_config.py` ```diff @@ -6,7 +6,9 @@ import pytest import yaml -from gateway.config import GatewayConfig, load_gateway_config +from gateway.config import GatewayConfig, Platform, load_gateway_config +from gateway.platforms.base import MessageEvent, MessageType +from gateway.session import SessionSource def test_gateway_config_stt_disabled_from_dict_nested(): ``` ```diff @@ -69,3 +71,46 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag assert "No STT provider is configured" not in result assert "trouble transcribing" in result assert "caption" in result + + [redacted:email] +async def test_prepare_inbound_message_text_transcribes_queued_voice_event(): + from gateway.run import GatewayRunner + + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig(stt_enabled=True) + runner.adapters = {} + runner._model = "test-model" + runner._base_url = "" + runner._has_setup_skill = lambda: False + + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="123", + chat_type="dm", + ) + event = MessageEvent( + text="", + message_type=MessageType.VOICE, + source=source, + media_urls=["/tmp/queued-voice.ogg"], + media_types=["audio/ogg"], + ) + + with patch( + "tools.transcription_tools.transcribe_audio", + return_value={ + "success": True, + "transcript": "queued voice transcript", + "provider": "local_command", + }, + ): + result = await runner._prepare_inbound_message_text( + event=event, + source=source, + history=[], + ) + + assert result is not None + assert "queued voice transcript" in result + assert "voice message" in result.lower() ``` - **#82162: fix(gateway): preserve MEDIA delivery on queued follow-up responses (salvage #71031)**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `gateway/run.py`, `tests/gateway/test_run_progress_topics.py`, `tests/gateway/test_tts_media_routing.py` - **#1809: fix(gateway): pass message_thread_id in send_image_file, send_document, send_video**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `#2339` - **#2269: fix: opencode-go auth config, vision download timeout, gateway send_document**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `#3919` - **#2339: fix(gateway): pass message_thread_id in send_image_file, send_document, send_video**. merged to main - maintainer direction: category `merged`; implication `superseded` - files: `gateway/platforms/telegram.py`, `tests/gateway/test_send_image_file.py`, `tests/gateway/test_telegram_documents.py` - **#6254: fix(gateway): stop echoing local file paths in media fallbacks**. closed without merge - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#15517`, `#15517` - **#7927: fix(gateway): preserve queued voice events for STT**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `#7948` - **#18546: fix(gateway): deliver MEDIA files for queued follow-up responses**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#82162`, `#71031` - **#38430: fix(gateway): remove default message for media without caption**. closed without merge - maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#11538` - **#71031: fix(gateway): preserve queued follow-up media continuity**. closed without merge - maintainer direction: category `merged_via`; implication `superseded`; refs `d220f1fdc0`, `a52dd17d93`, `0b17b691d6`, `#82162`, `#25119` ## Fix status - Merged to main in PR #3919 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-08-09`. - Subsystem: `gateway`