diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index 718f01e9954d3..e74a176188173 100644 --- a/gateway/platforms/feishu.py +++ b/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, ) diff --git a/gateway/run.py b/gateway/run.py index 01eb529693784..232fbda0519c9 100644 --- a/gateway/run.py +++ b/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: diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 1adbdd3a69413..e34805ca41306 100644 --- a/gateway/stream_consumer.py +++ b/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: diff --git a/tests/gateway/test_feishu_approval_buttons.py b/tests/gateway/test_feishu_approval_buttons.py index 954e9c06104fa..f2f86b09b5daf 100644 --- a/tests/gateway/test_feishu_approval_buttons.py +++ b/tests/gateway/test_feishu_approval_buttons.py @@ -127,6 +127,32 @@ async def test_sends_interactive_card(self): assert action_names == [ "approve_once", "approve_session", "approve_always", "deny" ] + assert kwargs["reply_to"] is None + + @pytest.mark.asyncio + async def test_forwards_reply_target(self): + adapter = _make_adapter() + + mock_response = SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="msg_reply"), + ) + with patch.object( + adapter, "_feishu_send_with_retry", new_callable=AsyncMock, + return_value=mock_response, + ) as mock_send: + result = await adapter.send_exec_approval( + chat_id="oc_12345", + command="rm -rf /important", + session_key="agent:main:feishu:group:oc_12345", + reply_to="om_parent_1", + metadata={"thread_id": "omt_thread_1"}, + ) + + assert result.success is True + kwargs = mock_send.call_args.kwargs + assert kwargs["reply_to"] == "om_parent_1" + assert kwargs["metadata"] == {"thread_id": "omt_thread_1"} @pytest.mark.asyncio async def test_stores_approval_state(self): diff --git a/tests/gateway/test_feishu_reply_routing.py b/tests/gateway/test_feishu_reply_routing.py new file mode 100644 index 0000000000000..33f37dade8f91 --- /dev/null +++ b/tests/gateway/test_feishu_reply_routing.py @@ -0,0 +1,52 @@ +import asyncio +import os +from types import SimpleNamespace +from unittest.mock import patch + + +class TestFeishuReplyRouting: + @patch.dict(os.environ, {}, clear=True) + def test_send_with_thread_metadata_only_uses_create_not_reply(self): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + + adapter = FeishuAdapter(PlatformConfig()) + captured = {"reply_calls": 0, "create_calls": 0} + + class _MessageAPI: + def reply(self, request): + captured["reply_calls"] += 1 + raise AssertionError("reply() should not be used without reply_to") + + def create(self, request): + captured["create_calls"] += 1 + captured["request"] = request + return SimpleNamespace( + success=lambda: True, + data=SimpleNamespace(message_id="om_create"), + ) + + adapter._client = SimpleNamespace( + im=SimpleNamespace( + v1=SimpleNamespace( + message=_MessageAPI(), + ) + ) + ) + + async def _direct(func, *args, **kwargs): + return func(*args, **kwargs) + + with patch("gateway.platforms.feishu.asyncio.to_thread", side_effect=_direct): + result = asyncio.run( + adapter.send( + chat_id="oc_chat", + content="hello", + metadata={"thread_id": "omt-thread"}, + ) + ) + + assert result.success + assert result.message_id == "om_create" + assert captured["reply_calls"] == 0 + assert captured["create_calls"] == 1 diff --git a/tests/gateway/test_proxy_mode.py b/tests/gateway/test_proxy_mode.py index 7ed6a19cb222d..640c963dd1963 100644 --- a/tests/gateway/test_proxy_mode.py +++ b/tests/gateway/test_proxy_mode.py @@ -230,6 +230,55 @@ async def test_run_agent_skips_proxy_when_not_configured(self, monkeypatch): class TestRunAgentViaProxy: """Test the actual proxy HTTP forwarding logic.""" + @pytest.mark.asyncio + async def test_streaming_consumer_replies_to_originating_message(self, monkeypatch): + monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") + runner = _make_runner() + runner.config.streaming.enabled = True + source = _make_source(platform=Platform.FEISHU) + source.thread_id = "omt_thread_1" + + resp = _FakeSSEResponse( + status=200, + sse_chunks=[ + 'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n', + "data: [DONE]\n\n", + ], + ) + session = _FakeSession(resp) + + adapter = MagicMock() + adapter.SUPPORTS_MESSAGE_EDITING = True + adapter.send_typing = AsyncMock() + runner.adapters[source.platform] = adapter + + consumer_ctor = MagicMock() + consumer = MagicMock() + consumer.run = AsyncMock() + consumer.on_delta = MagicMock() + consumer.finish = MagicMock() + consumer.final_response_sent = False + consumer_ctor.return_value = consumer + + with ( + patch("gateway.run._load_gateway_config", return_value={}), + _patch_aiohttp(session), + patch("aiohttp.ClientTimeout"), + patch("gateway.stream_consumer.GatewayStreamConsumer", consumer_ctor), + ): + await runner._run_agent_via_proxy( + message="hello", + context_prompt="", + history=[], + source=source, + session_id="sess-1", + event_message_id="om_parent_1", + ) + + consumer_ctor.assert_called_once() + assert consumer_ctor.call_args.kwargs["metadata"] == {"thread_id": "omt_thread_1"} + assert consumer_ctor.call_args.kwargs["reply_to"] == "om_parent_1" + @pytest.mark.asyncio async def test_builds_correct_request(self, monkeypatch): monkeypatch.setenv("GATEWAY_PROXY_URL", "http://host:8642") diff --git a/tests/gateway/test_retry_response.py b/tests/gateway/test_retry_response.py index 34a98015e0ac4..4fb765362cca7 100644 --- a/tests/gateway/test_retry_response.py +++ b/tests/gateway/test_retry_response.py @@ -58,3 +58,46 @@ async def test_retry_no_previous_message(gateway): ) result = await gateway._handle_retry_command(event) assert result == "No previous message to retry." + + +@pytest.mark.asyncio +async def test_retry_preserves_current_message_routing_context(gateway): + """Synthetic /retry events must keep reply/thread routing from the command message.""" + source = MagicMock() + source.thread_id = "topic-42" + gateway.session_store.get_or_create_session.return_value = MagicMock( + session_id="test-session" + ) + gateway.session_store.load_transcript.return_value = [ + {"role": "user", "content": "Hello again"}, + {"role": "assistant", "content": "Old answer"}, + ] + gateway.session_store.rewrite_transcript = MagicMock() + + captured = {} + + async def fake_handle_message(event): + captured["message_id"] = event.message_id + captured["platform_update_id"] = event.platform_update_id + captured["thread_id"] = event.source.thread_id + return "retried" + + gateway._handle_message = AsyncMock(side_effect=fake_handle_message) + + event = MessageEvent( + text="/retry", + message_type=MessageType.TEXT, + source=source, + raw_message=MagicMock(), + message_id="msg-123", + platform_update_id=987, + ) + + result = await gateway._handle_retry_command(event) + + assert result == "retried" + assert captured == { + "message_id": "msg-123", + "platform_update_id": 987, + "thread_id": "topic-42", + } diff --git a/tests/gateway/test_run_progress_topics.py b/tests/gateway/test_run_progress_topics.py index 49fb91d449dd4..efc9cceab48e8 100644 --- a/tests/gateway/test_run_progress_topics.py +++ b/tests/gateway/test_run_progress_topics.py @@ -65,6 +65,62 @@ async def edit_message(self, chat_id, message_id, content) -> SendResult: raise AssertionError("non-editable adapters should not receive edit_message calls") +class ApprovalCaptureAdapter(ProgressCaptureAdapter): + def __init__(self, platform=Platform.TELEGRAM): + super().__init__(platform=platform) + self.approval_calls = [] + self.paused_typing = [] + + def pause_typing_for_chat(self, chat_id) -> None: + self.paused_typing.append(chat_id) + + async def send_exec_approval( + self, + chat_id, + command, + session_key, + description="dangerous command", + metadata=None, + reply_to=None, + ) -> SendResult: + self.approval_calls.append( + { + "chat_id": chat_id, + "command": command, + "session_key": session_key, + "description": description, + "metadata": metadata, + "reply_to": reply_to, + } + ) + return SendResult(success=True, message_id="approval-1") + + +class MediaCaptureAdapter(ProgressCaptureAdapter): + async def send_document( + self, + chat_id, + file_path, + caption=None, + file_name=None, + reply_to=None, + metadata=None, + **kwargs, + ) -> SendResult: + text = f"📎 File: {file_path}" + if caption: + text = f"{caption}\n{text}" + self.sent.append( + { + "chat_id": chat_id, + "content": text, + "reply_to": reply_to, + "metadata": metadata, + } + ) + return SendResult(success=True, message_id="media-1") + + class FakeAgent: def __init__(self, **kwargs): self.tool_progress_callback = kwargs.get("tool_progress_callback") @@ -134,6 +190,19 @@ def run_conversation(self, message, conversation_history=None, task_id=None): } +class LongRunningAgent: + def __init__(self, **kwargs): + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + time.sleep(0.25) + return { + "final_response": "done", + "messages": [], + "api_calls": 1, + } + + def _make_runner(adapter): gateway_run = importlib.import_module("gateway.run") GatewayRunner = gateway_run.GatewayRunner @@ -205,6 +274,52 @@ async def test_run_agent_progress_stays_in_originating_topic(monkeypatch, tmp_pa assert all(call["metadata"] == {"thread_id": "17585"} for call in adapter.typing) +@pytest.mark.asyncio +async def test_run_agent_progress_replies_to_originating_feishu_message(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all") + + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = FakeAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + import tools.terminal_tool # noqa: F401 - register terminal emoji for this fake-agent test + + adapter = ProgressCaptureAdapter(platform=Platform.FEISHU) + runner = _make_runner(adapter) + gateway_run = importlib.import_module("gateway.run") + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}) + source = SessionSource( + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + ) + + result = await runner._run_agent( + message="hello", + context_prompt="", + history=[], + source=source, + session_id="sess-feishu-1", + session_key="agent:main:feishu:group:oc_chat_1:omt_thread_1", + event_message_id="om_parent_1", + ) + + assert result["final_response"] == "done" + assert adapter.sent == [ + { + "chat_id": "oc_chat_1", + "content": '💻 terminal: "pwd"', + "reply_to": "om_parent_1", + "metadata": {"thread_id": "omt_thread_1"}, + } + ] + + @pytest.mark.asyncio async def test_run_agent_progress_does_not_use_event_message_id_for_telegram_dm(monkeypatch, tmp_path): """Telegram DM progress must not reuse event message id as thread metadata.""" @@ -476,6 +591,27 @@ def run_conversation(self, message, conversation_history=None, task_id=None): } +class ApprovalCallbackAgent: + def __init__(self, **kwargs): + self.gateway_session_key = kwargs.get("gateway_session_key") + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + import tools.approval as approval_mod + + approval_mod._gateway_notify_cbs[self.gateway_session_key]( + { + "command": "rm -rf /tmp/demo", + "description": "dangerous command", + } + ) + return { + "final_response": "done", + "messages": [], + "api_calls": 1, + } + + class VerboseAgent: """Agent that emits a tool call with args whose JSON exceeds 200 chars.""" LONG_CODE = "x" * 300 @@ -509,6 +645,7 @@ async def _run_with_agent( chat_id="-1001", chat_type="group", thread_id="17585", + event_message_id=None, adapter_cls=ProgressCaptureAdapter, ): if config_data: @@ -555,6 +692,7 @@ async def _run_with_agent( source=source, session_id=session_id, session_key=session_key, + event_message_id=event_message_id, ) return adapter, result @@ -573,6 +711,33 @@ async def test_run_agent_surfaces_real_interim_commentary(monkeypatch, tmp_path) assert any(call["content"] == "I'll inspect the repo first." for call in adapter.sent) +@pytest.mark.asyncio +async def test_run_agent_interim_commentary_replies_to_originating_feishu_message(monkeypatch, tmp_path): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + CommentaryAgent, + session_id="sess-commentary-feishu-reply", + config_data={"display": {"interim_assistant_messages": True}}, + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + event_message_id="om_parent_1", + ) + + assert result.get("already_sent") is not True + commentary_calls = [call for call in adapter.sent if call["content"] == "I'll inspect the repo first."] + assert commentary_calls == [ + { + "chat_id": "oc_chat_1", + "content": "I'll inspect the repo first.", + "reply_to": "om_parent_1", + "metadata": {"thread_id": "omt_thread_1"}, + } + ] + + @pytest.mark.asyncio async def test_run_agent_surfaces_interim_commentary_by_default(monkeypatch, tmp_path): adapter, result = await _run_with_agent( @@ -708,6 +873,38 @@ async def test_run_agent_previewed_final_marks_already_sent(monkeypatch, tmp_pat assert [call["content"] for call in adapter.sent] == ["You're welcome."] +@pytest.mark.asyncio +async def test_post_stream_media_file_replies_to_originating_feishu_message(): + adapter = MediaCaptureAdapter(platform=Platform.FEISHU) + runner = _make_runner(adapter) + event = MessageEvent( + text="报告已生成。", + message_type=MessageType.TEXT, + source=SessionSource( + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + ), + message_id="om_parent_1", + ) + + await runner._deliver_media_from_response( + "报告已生成。\nMEDIA:/tmp/report.pdf", + event, + adapter, + ) + + assert adapter.sent == [ + { + "chat_id": "oc_chat_1", + "content": "📎 File: /tmp/report.pdf", + "reply_to": "om_parent_1", + "metadata": {"thread_id": "omt_thread_1"}, + } + ] + + @pytest.mark.asyncio async def test_run_agent_matrix_streaming_omits_cursor(monkeypatch, tmp_path): adapter, result = await _run_with_agent( @@ -764,6 +961,90 @@ async def test_run_agent_defers_background_review_notification_until_release(mon assert adapter.sent == [] +@pytest.mark.asyncio +async def test_run_agent_background_review_replies_to_originating_message_after_release(monkeypatch, tmp_path): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + BackgroundReviewAgent, + session_id="sess-bg-review-reply", + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + event_message_id="om_parent_1", + ) + + assert result["final_response"] == "done" + assert adapter.sent == [] + assert len(adapter._post_delivery_callbacks) == 1 + + release_cb = next(iter(adapter._post_delivery_callbacks.values())) + release_cb() + await asyncio.sleep(0.05) + + assert adapter.sent == [ + { + "chat_id": "oc_chat_1", + "content": "💾 Skill 'prospect-scanner' created.", + "reply_to": "om_parent_1", + "metadata": {"thread_id": "omt_thread_1"}, + } + ] + + +@pytest.mark.asyncio +async def test_run_agent_approval_buttons_reply_to_originating_feishu_message(monkeypatch, tmp_path): + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + ApprovalCallbackAgent, + session_id="sess-approval-feishu-reply", + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + event_message_id="om_parent_1", + adapter_cls=ApprovalCaptureAdapter, + ) + + assert result["final_response"] == "done" + assert adapter.paused_typing == ["oc_chat_1"] + assert adapter.approval_calls == [ + { + "chat_id": "oc_chat_1", + "command": "rm -rf /tmp/demo", + "session_key": "agent:main:feishu:group:oc_chat_1:omt_thread_1", + "description": "dangerous command", + "metadata": {"thread_id": "omt_thread_1"}, + "reply_to": "om_parent_1", + } + ] + + +@pytest.mark.asyncio +async def test_run_agent_long_running_notification_replies_to_originating_message(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_AGENT_NOTIFY_INTERVAL", "0.05") + + adapter, result = await _run_with_agent( + monkeypatch, + tmp_path, + LongRunningAgent, + session_id="sess-long-running-reply", + platform=Platform.FEISHU, + chat_id="oc_chat_1", + chat_type="group", + thread_id="omt_thread_1", + event_message_id="om_parent_1", + ) + + assert result["final_response"] == "done" + heartbeat_calls = [call for call in adapter.sent if call["content"].startswith("⏳ Still working...")] + assert heartbeat_calls + assert heartbeat_calls[0]["reply_to"] == "om_parent_1" + assert heartbeat_calls[0]["metadata"] == {"thread_id": "omt_thread_1"} + + @pytest.mark.asyncio async def test_base_processing_releases_post_delivery_callback_after_main_send(): """Post-delivery callbacks on the adapter fire after the main response.""" diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py index 7ae587dadd728..2f9f96fe4e4b5 100644 --- a/tests/gateway/test_stream_consumer.py +++ b/tests/gateway/test_stream_consumer.py @@ -189,6 +189,26 @@ async def test_first_send_strips_media(self): assert "MEDIA:" not in sent_text assert "Here is your image" in sent_text + @pytest.mark.asyncio + async def test_first_send_preserves_initial_reply_to(self): + """Initial streamed send should reply to the originating user message.""" + adapter = MagicMock() + send_result = SimpleNamespace(success=True, message_id="msg_1") + adapter.send = AsyncMock(return_value=send_result) + adapter.MAX_MESSAGE_LENGTH = 4096 + + consumer = GatewayStreamConsumer( + adapter, + "chat_123", + metadata={"thread_id": "omt_thread_1"}, + reply_to="om_parent_1", + ) + await consumer._send_or_edit("streaming hello") + + adapter.send.assert_called_once() + assert adapter.send.call_args.kwargs["reply_to"] == "om_parent_1" + assert adapter.send.call_args.kwargs["metadata"] == {"thread_id": "omt_thread_1"} + @pytest.mark.asyncio async def test_edit_strips_media(self): """Edit call removes MEDIA: tags from visible text."""