diff --git a/tests/tools/test_send_message_cross_loop.py b/tests/tools/test_send_message_cross_loop.py new file mode 100644 index 0000000000..3fcca70e50 --- /dev/null +++ b/tests/tools/test_send_message_cross_loop.py @@ -0,0 +1,201 @@ +"""Regression tests for the cross-event-loop deadlock fix in send_message. + +When the agent's tool worker thread calls _send_via_adapter() while the +adapter's queues live on the gateway's main event loop, the send must be +dispatched via run_coroutine_threadsafe to the gateway loop — NOT awaited +directly on the worker loop (which would deadlock due to the selector never +being woken by cross-thread future.set_result). +""" + +import asyncio +import sys +import threading +from types import ModuleType, SimpleNamespace + +import pytest + +from gateway.config import Platform + + +class TestSendViaAdapterCrossLoopDispatch: + + @pytest.mark.asyncio + async def test_cross_loop_dispatches_to_gateway_loop(self, monkeypatch): + """adapter.send() runs on gateway loop, not the caller's loop.""" + from tools.send_message_tool import _send_via_adapter + + send_loop_id = {} + platform = Platform("wecom") + + class FakeAdapter: + async def send(self, *, chat_id, content, metadata=None): + send_loop_id["loop"] = id(asyncio.get_running_loop()) + return SimpleNamespace(success=True, message_id="cross-ok") + + gateway_loop = asyncio.new_event_loop() + started = threading.Event() + + def run_gateway(): + asyncio.set_event_loop(gateway_loop) + started.set() + gateway_loop.run_forever() + + t = threading.Thread(target=run_gateway, daemon=True) + t.start() + started.wait(timeout=2) + + try: + runner = SimpleNamespace( + adapters={platform: FakeAdapter()}, + _gateway_loop=gateway_loop, + ) + fake_gateway_run = ModuleType("gateway.run") + fake_gateway_run._gateway_runner_ref = lambda: runner + monkeypatch.setitem(sys.modules, "gateway.run", fake_gateway_run) + + result = await _send_via_adapter( + platform, + SimpleNamespace(extra={}), + "wr_group_123", + "hello from worker", + ) + + assert result == {"success": True, "message_id": "cross-ok"} + # Verify send() ran on the gateway loop, not our current loop + assert send_loop_id["loop"] == id(gateway_loop) + finally: + gateway_loop.call_soon_threadsafe(gateway_loop.stop) + t.join(timeout=2) + gateway_loop.close() + + @pytest.mark.asyncio + async def test_same_loop_uses_direct_await(self, monkeypatch): + """When current loop IS the gateway loop, adapter.send() is awaited + directly — no run_coroutine_threadsafe (which would self-lock).""" + from tools.send_message_tool import _send_via_adapter + + current_loop = asyncio.get_running_loop() + platform = Platform("wecom") + called_directly = {} + + class FakeAdapter: + async def send(self, *, chat_id, content, metadata=None): + called_directly["loop"] = id(asyncio.get_running_loop()) + return SimpleNamespace(success=True, message_id="direct-ok") + + runner = SimpleNamespace( + adapters={platform: FakeAdapter()}, + _gateway_loop=current_loop, + ) + fake_gateway_run = ModuleType("gateway.run") + fake_gateway_run._gateway_runner_ref = lambda: runner + monkeypatch.setitem(sys.modules, "gateway.run", fake_gateway_run) + + result = await _send_via_adapter( + platform, + SimpleNamespace(extra={}), + "wr_group_456", + "direct send", + ) + + assert result == {"success": True, "message_id": "direct-ok"} + assert called_directly["loop"] == id(current_loop) + + @pytest.mark.asyncio + async def test_gateway_loop_not_running_returns_error(self, monkeypatch): + """When gateway loop exists but is stopped, return an error rather + than attempting direct await on a loop-bound adapter.""" + from tools.send_message_tool import _send_via_adapter + + stopped_loop = asyncio.new_event_loop() + stopped_loop.close() + platform = Platform("wecom") + + class FakeAdapter: + async def send(self, *, chat_id, content, metadata=None): + raise AssertionError("should not be called") + + runner = SimpleNamespace( + adapters={platform: FakeAdapter()}, + _gateway_loop=stopped_loop, + ) + fake_gateway_run = ModuleType("gateway.run") + fake_gateway_run._gateway_runner_ref = lambda: runner + monkeypatch.setitem(sys.modules, "gateway.run", fake_gateway_run) + + result = await _send_via_adapter( + platform, + SimpleNamespace(extra={}), + "wr_group_789", + "should fail", + ) + + assert "error" in result + assert "not running" in result["error"] + + @pytest.mark.asyncio + async def test_shield_prevents_cancel_of_enqueued_send(self, monkeypatch): + """asyncio.shield ensures that cancelling the caller does NOT cancel + the already-dispatched send on the gateway loop.""" + from tools.send_message_tool import _send_via_adapter + + send_completed = asyncio.Event() + send_result_holder = {} + platform = Platform("wecom") + + class FakeAdapter: + async def send(self, *, chat_id, content, metadata=None): + # Simulate a slow send (token bucket wait) + await asyncio.sleep(0.3) + send_result_holder["sent"] = True + send_completed.set() + return SimpleNamespace(success=True, message_id="shielded") + + gateway_loop = asyncio.new_event_loop() + started = threading.Event() + + def run_gateway(): + asyncio.set_event_loop(gateway_loop) + started.set() + gateway_loop.run_forever() + + t = threading.Thread(target=run_gateway, daemon=True) + t.start() + started.wait(timeout=2) + + try: + runner = SimpleNamespace( + adapters={platform: FakeAdapter()}, + _gateway_loop=gateway_loop, + ) + fake_gateway_run = ModuleType("gateway.run") + fake_gateway_run._gateway_runner_ref = lambda: runner + monkeypatch.setitem(sys.modules, "gateway.run", fake_gateway_run) + + # Start the send, then cancel the caller task after a short delay + async def do_send(): + return await _send_via_adapter( + platform, + SimpleNamespace(extra={}), + "wr_group_shield", + "shielded msg", + ) + + task = asyncio.create_task(do_send()) + await asyncio.sleep(0.1) # let it dispatch to gateway loop + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + # The send on the gateway loop should still complete despite cancel + fut = asyncio.run_coroutine_threadsafe( + asyncio.wait_for(send_completed.wait(), timeout=1.0), + gateway_loop, + ) + fut.result(timeout=2) + assert send_result_holder.get("sent") is True + finally: + gateway_loop.call_soon_threadsafe(gateway_loop.stop) + t.join(timeout=2) + gateway_loop.close() diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 08f72d9509..9d64123909 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -600,6 +600,14 @@ def _parse_target_ref(platform_name: str, target_ref: str): if group_id: return f"group:{group_id}", None, True return None, None, False + # WeCom: group IDs start with "wr" or "wc", user IDs start with "wo" or + # are bare alphanumeric strings. Treat any non-empty WeCom target_ref as + # an explicit chat_id — the adapter resolves whether to use APP_CMD_RESPONSE + # (groups) or APP_CMD_SEND (DMs) internally. + if platform_name == "wecom": + stripped = target_ref.strip() + if stripped: + return stripped, None, True if platform_name in _PHONE_PLATFORMS: match = _E164_TARGET_RE.fullmatch(target_ref) if match: @@ -870,7 +878,48 @@ async def _send_via_adapter( metadata["publish_topic"] = chat_id if not metadata: metadata = None - result = await adapter.send(chat_id=chat_id, content=chunk, metadata=metadata) + + # The adapter's send() uses asyncio.Queue + worker tasks bound + # to the gateway's main event loop. Calling send() from a + # different thread/loop (the agent's tool worker thread) causes + # a cross-loop Future deadlock: the worker loop's selector never + # gets woken when the gateway loop resolves the future. + # When on a different loop, dispatch onto the gateway loop via + # run_coroutine_threadsafe and await the wrapped future. + gateway_loop = getattr(runner, "_gateway_loop", None) + try: + _current_loop = asyncio.get_running_loop() + except RuntimeError: + _current_loop = None + + _need_cross_loop = ( + gateway_loop is not None + and _current_loop is not gateway_loop + ) + + if _need_cross_loop: + if not gateway_loop.is_running(): + return {"error": "Gateway loop is not running; cannot dispatch adapter send"} + from agent.async_utils import safe_schedule_threadsafe + fut = safe_schedule_threadsafe( + adapter.send(chat_id=chat_id, content=chunk, metadata=metadata), + gateway_loop, + logger=logger, + log_message="send_message: failed to schedule on gateway loop", + ) + if fut is None: + return {"error": "Gateway loop unavailable for send dispatch"} + # Use shield so that if the caller's task is cancelled (e.g. + # agent interrupt), the already-enqueued send on the gateway + # loop is NOT cancelled — preventing "tool failed but message + # still sent later" followed by agent retry causing duplicates. + # No explicit timeout here: the adapter's internal request + # timeout (15s) and the upper-layer _run_async 300s timeout + # provide sufficient protection against hangs. + result = await asyncio.shield(asyncio.wrap_future(fut)) + else: + # Same loop or no gateway loop (CLI, tests) — direct await. + result = await adapter.send(chat_id=chat_id, content=chunk, metadata=metadata) except asyncio.CancelledError: raise except Exception as e: @@ -1259,6 +1308,25 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, last_result = result return last_result + # --- WeCom: native media attachment support via live gateway adapter --- + if platform == Platform.WECOM and media_files: + last_result = None + for i, chunk in enumerate(chunks): + is_last = (i == len(chunks) - 1) + result = await _send_via_adapter( + platform, + pconfig, + chat_id, + chunk, + thread_id=thread_id, + media_files=media_files if is_last else None, + force_document=force_document, + ) + if isinstance(result, dict) and result.get("error"): + return result + last_result = result + return last_result + # --- Non-media platforms --- if media_files and not message.strip(): return {