fix(gateway): preserve reply routing for streamed media sends

gateway/platforms/feishu.py

@@ -1727,6 +1727,7 @@ async def send_exec_approval(
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        reply_to: Optional[str] = None,
     ) -> SendResult:
         """Send an interactive card with approval buttons.
 
@@ -1777,7 +1778,7 @@ def _btn(label: str, action_name: str, btn_type: str = "default") -> dict:
                 chat_id=chat_id,
                 msg_type="interactive",
                 payload=payload,
-                reply_to=None,
+                reply_to=reply_to,
                 metadata=metadata,
             )
 

gateway/run.py

@@ -15,6 +15,7 @@
 
 import asyncio
 import dataclasses
+import inspect
 import json
 import logging
 import os
@@ -590,6 +591,23 @@ def _parse_session_key(session_key: str) -> "dict | None":
     return None
 
 
+def _build_stream_reply_routing(
+    source: SessionSource,
+    event_message_id: Optional[str] = None,
+) -> "tuple[Optional[Dict[str, Any]], Optional[str]]":
+    """Build thread metadata + reply target for mid-turn gateway sends.
+
+    Slack DMs need the originating message id as a thread fallback. Other
+    platforms should only use explicit source.thread_id metadata.
+    """
+    if source.platform == Platform.SLACK:
+        thread_id = source.thread_id or event_message_id
+    else:
+        thread_id = source.thread_id
+    metadata = {"thread_id": thread_id} if thread_id else None
+    return metadata, event_message_id
+
+
 def _format_gateway_process_notification(evt: dict) -> "str | None":
     """Format a watch pattern event from completion_queue into a [IMPORTANT:] message."""
     evt_type = evt.get("type", "completion")
@@ -6232,6 +6250,8 @@ async def _handle_retry_command(self, event: MessageEvent) -> str:
             message_type=MessageType.TEXT,
             source=source,
             raw_message=event.raw_message,
+            message_id=event.message_id,
+            platform_update_id=event.platform_update_id,
             channel_prompt=event.channel_prompt,
         )
 
@@ -6657,7 +6677,10 @@ async def _deliver_media_from_response(
             _, cleaned = adapter.extract_images(response)
             local_files, _ = adapter.extract_local_files(cleaned)
 
-            _thread_meta = {"thread_id": event.source.thread_id} if event.source.thread_id else None
+            _thread_meta, _reply_to = _build_stream_reply_routing(
+                event.source,
+                event.message_id,
+            )
 
             _AUDIO_EXTS = {'.ogg', '.opus', '.mp3', '.wav', '.m4a'}
             _VIDEO_EXTS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'}
@@ -6670,24 +6693,28 @@ async def _deliver_media_from_response(
                         await adapter.send_voice(
                             chat_id=event.source.chat_id,
                             audio_path=media_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                     elif ext in _VIDEO_EXTS:
                         await adapter.send_video(
                             chat_id=event.source.chat_id,
                             video_path=media_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                     elif ext in _IMAGE_EXTS:
                         await adapter.send_image_file(
                             chat_id=event.source.chat_id,
                             image_path=media_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                     else:
                         await adapter.send_document(
                             chat_id=event.source.chat_id,
                             file_path=media_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                 except Exception as e:
@@ -6700,12 +6727,14 @@ async def _deliver_media_from_response(
                         await adapter.send_image_file(
                             chat_id=event.source.chat_id,
                             image_path=file_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                     else:
                         await adapter.send_document(
                             chat_id=event.source.chat_id,
                             file_path=file_path,
+                            reply_to=_reply_to,
                             metadata=_thread_meta,
                         )
                 except Exception as e:
@@ -9289,10 +9318,10 @@ def _run_still_current() -> bool:
             else bool(_plat_streaming)
         )
 
-        if source.thread_id:
-            _thread_metadata: Optional[Dict[str, Any]] = {"thread_id": source.thread_id}
-        else:
-            _thread_metadata = None
+        _thread_metadata, _stream_reply_to = _build_stream_reply_routing(
+            source,
+            event_message_id,
+        )
 
         if _streaming_enabled:
             try:
@@ -9326,6 +9355,7 @@ def _run_still_current() -> bool:
                         chat_id=source.chat_id,
                         config=_consumer_cfg,
                         metadata=_thread_metadata,
+                        reply_to=_stream_reply_to,
                     )
             except Exception as _sc_err:
                 logger.debug("Proxy: could not set up stream consumer: %s", _sc_err)
@@ -9539,13 +9569,31 @@ def _run_still_current() -> bool:
         except Exception:
             pass
 
-        # Tool progress mode — resolved per-platform with env var fallback
-        _resolved_tp = resolve_display_setting(user_config, platform_key, "tool_progress")
-        progress_mode = (
-            _resolved_tp
-            or os.getenv("HERMES_TOOL_PROGRESS_MODE")
-            or "all"
+        # Tool progress mode — explicit config wins, then env override, then
+        # built-in per-platform defaults.
+        _platform_display = display_config.get("platforms") or {}
+        _platform_progress_cfg = None
+        if isinstance(_platform_display, dict):
+            _platform_cfg = _platform_display.get(platform_key)
+            if isinstance(_platform_cfg, dict):
+                _platform_progress_cfg = _platform_cfg.get("tool_progress")
+        _legacy_progress_cfg = None
+        _legacy_progress = display_config.get("tool_progress_overrides")
+        if isinstance(_legacy_progress, dict):
+            _legacy_progress_cfg = _legacy_progress.get(platform_key)
+        _global_progress_cfg = display_config.get("tool_progress")
+        _has_explicit_progress_cfg = any(
+            value is not None
+            for value in (_platform_progress_cfg, _legacy_progress_cfg, _global_progress_cfg)
         )
+        if _has_explicit_progress_cfg:
+            progress_mode = resolve_display_setting(user_config, platform_key, "tool_progress")
+        else:
+            progress_mode = (
+                os.getenv("HERMES_TOOL_PROGRESS_MODE")
+                or resolve_display_setting(user_config, platform_key, "tool_progress")
+                or "all"
+            )
         # Disable tool progress for webhooks - they don't support message editing,
         # so each progress line would be sent as a separate message.
         from gateway.config import Platform
@@ -9680,16 +9728,10 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non
         # Background task to send progress messages
         # Accumulates tool lines into a single message that gets edited.
         #
-        # Threading metadata is platform-specific:
-        # - Slack DM threading needs event_message_id fallback (reply thread)
-        # - Telegram uses message_thread_id only for forum topics; passing a
-        #   normal DM/group message id as thread_id causes send failures
-        # - Other platforms should use explicit source.thread_id only
-        if source.platform == Platform.SLACK:
-            _progress_thread_id = source.thread_id or event_message_id
-        else:
-            _progress_thread_id = source.thread_id
-        _progress_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None
+        _progress_metadata, _progress_reply_to = _build_stream_reply_routing(
+            source,
+            event_message_id,
+        )
 
         async def send_progress_messages():
             if not progress_queue:
@@ -9789,15 +9831,30 @@ async def send_progress_messages():
                                     adapter.name,
                                 )
                             can_edit = False
-                            await adapter.send(chat_id=source.chat_id, content=msg, metadata=_progress_metadata)
+                            await adapter.send(
+                                chat_id=source.chat_id,
+                                content=msg,
+                                reply_to=_progress_reply_to,
+                                metadata=_progress_metadata,
+                            )
                     else:
                         if can_edit:
                             # First tool: send all accumulated text as new message
                             full_text = "\n".join(progress_lines)
-                            result = await adapter.send(chat_id=source.chat_id, content=full_text, metadata=_progress_metadata)
+                            result = await adapter.send(
+                                chat_id=source.chat_id,
+                                content=full_text,
+                                reply_to=_progress_reply_to,
+                                metadata=_progress_metadata,
+                            )
                         else:
                             # Editing unsupported: send just this line
-                            result = await adapter.send(chat_id=source.chat_id, content=msg, metadata=_progress_metadata)
+                            result = await adapter.send(
+                                chat_id=source.chat_id,
+                                content=msg,
+                                reply_to=_progress_reply_to,
+                                metadata=_progress_metadata,
+                            )
                         if result.success and result.message_id:
                             progress_msg_id = result.message_id
 
@@ -9879,7 +9936,8 @@ def _step_callback_sync(iteration: int, prev_tools: list) -> None:
         # Bridge sync status_callback → async adapter.send for context pressure
         _status_adapter = self.adapters.get(source.platform)
         _status_chat_id = source.chat_id
-        _status_thread_metadata = {"thread_id": _progress_thread_id} if _progress_thread_id else None
+        _status_thread_metadata = _progress_metadata
+        _status_reply_to = _progress_reply_to
 
         def _status_callback_sync(event_type: str, message: str) -> None:
             if not _status_adapter or not _run_still_current():
@@ -9889,6 +9947,7 @@ def _status_callback_sync(event_type: str, message: str) -> None:
                     _status_adapter.send(
                         _status_chat_id,
                         message,
+                        reply_to=_status_reply_to,
                         metadata=_status_thread_metadata,
                     ),
                     _loop_for_step,
@@ -10023,7 +10082,8 @@ def run_sync():
                             adapter=_adapter,
                             chat_id=source.chat_id,
                             config=_consumer_cfg,
-                            metadata={"thread_id": _progress_thread_id} if _progress_thread_id else None,
+                            metadata=_progress_metadata,
+                            reply_to=_progress_reply_to,
                         )
                         if _want_stream_deltas:
                             def _stream_delta_cb(text: str) -> None:
@@ -10049,6 +10109,7 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                         _status_adapter.send(
                             _status_chat_id,
                             text,
+                            reply_to=_status_reply_to,
                             metadata=_status_thread_metadata,
                         ),
                         _loop_for_step,
@@ -10146,6 +10207,7 @@ def _deliver_bg_review_message(message: str) -> None:
                         _status_adapter.send(
                             _status_chat_id,
                             message,
+                            reply_to=_status_reply_to,
                             metadata=_status_thread_metadata,
                         ),
                         _loop_for_step,
@@ -10294,14 +10356,22 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 # false positives from MagicMock auto-attribute creation in tests.
                 if getattr(type(_status_adapter), "send_exec_approval", None) is not None:
                     try:
+                        _approval_kwargs = {
+                            "chat_id": _status_chat_id,
+                            "command": cmd,
+                            "session_key": _approval_session_key,
+                            "description": desc,
+                            "metadata": _status_thread_metadata,
+                        }
+                        try:
+                            if "reply_to" in inspect.signature(
+                                _status_adapter.send_exec_approval
+                            ).parameters:
+                                _approval_kwargs["reply_to"] = _status_reply_to
+                        except Exception:
+                            pass
                         _approval_result = asyncio.run_coroutine_threadsafe(
-                            _status_adapter.send_exec_approval(
-                                chat_id=_status_chat_id,
-                                command=cmd,
-                                session_key=_approval_session_key,
-                                description=desc,
-                                metadata=_status_thread_metadata,
-                            ),
+                            _status_adapter.send_exec_approval(**_approval_kwargs),
                             _loop_for_step,
                         ).result(timeout=15)
                         if _approval_result.success:
@@ -10329,6 +10399,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
                         _status_adapter.send(
                             _status_chat_id,
                             msg,
+                            reply_to=_status_reply_to,
                             metadata=_status_thread_metadata,
                         ),
                         _loop_for_step,
@@ -10663,6 +10734,7 @@ async def _notify_long_running():
                     await _notify_adapter.send(
                         source.chat_id,
                         f"⏳ Still working... ({_elapsed_mins} min elapsed{_status_detail})",
+                        reply_to=_status_reply_to,
                         metadata=_status_thread_metadata,
                     )
                 except Exception as _ne:
@@ -10757,6 +10829,7 @@ async def _notify_long_running():
                                     f"If the agent does not respond soon, it will "
                                     f"be timed out in {_remaining_mins} min. "
                                     f"You can continue waiting or use /reset.",
+                                    reply_to=_status_reply_to,
                                     metadata=_status_thread_metadata,
                                 )
                             except Exception as _warn_err:
@@ -10991,6 +11064,7 @@ async def _notify_long_running():
                             await adapter.send(
                                 source.chat_id,
                                 first_response,
+                                reply_to=_status_reply_to,
                                 metadata=_status_thread_metadata,
                             )
                         except Exception as e:

gateway/stream_consumer.py

@@ -91,11 +91,13 @@ def __init__(
         chat_id: str,
         config: Optional[StreamConsumerConfig] = None,
         metadata: Optional[dict] = None,
+        reply_to: Optional[str] = None,
     ):
         self.adapter = adapter
         self.chat_id = chat_id
         self.cfg = config or StreamConsumerConfig()
         self.metadata = metadata
+        self.reply_to = reply_to
         self._queue: queue.Queue = queue.Queue()
         self._accumulated = ""
         self._message_id: Optional[str] = None
@@ -519,10 +521,11 @@ async def _send_new_chunk(self, text: str, reply_to_id: Optional[str]) -> Option
             return reply_to_id
         try:
             meta = dict(self.metadata) if self.metadata else {}
+            effective_reply_to = reply_to_id or self.reply_to
             result = await self.adapter.send(
                 chat_id=self.chat_id,
                 content=text,
-                reply_to=reply_to_id,
+                reply_to=effective_reply_to,
                 metadata=meta,
             )
             if result.success and result.message_id:
@@ -628,6 +631,7 @@ async def _send_fallback_final(self, text: str) -> None:
                 result = await self.adapter.send(
                     chat_id=self.chat_id,
                     content=chunk,
+                    reply_to=last_message_id,
                     metadata=self.metadata,
                 )
                 if result.success:
@@ -737,6 +741,7 @@ async def _send_commentary(self, text: str) -> bool:
             result = await self.adapter.send(
                 chat_id=self.chat_id,
                 content=text,
+                reply_to=self.reply_to,
                 metadata=self.metadata,
             )
             # Note: do NOT set _already_sent = True here.
@@ -953,6 +958,7 @@ async def _send_or_edit(self, text: str, *, finalize: bool = False) -> bool:
                 result = await self.adapter.send(
                     chat_id=self.chat_id,
                     content=text,
+                    reply_to=self.reply_to,
                     metadata=self.metadata,
                 )
                 if result.success: