From 47d15117b8277966e0c9445acefd8f4d703c33cd Mon Sep 17 00:00:00 2001 From: Snow <272681648+snowctl@users.noreply.github.com> Date: Sat, 11 Jul 2026 22:43:22 -0400 Subject: [PATCH] fix(cron): avoid live adapter reuse across event loops --- cron/scheduler.py | 11 ++-- tests/tools/test_send_message_tool.py | 72 +++++++++++++++++++++++++++ tools/send_message_tool.py | 14 +++++- 3 files changed, 93 insertions(+), 4 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 7cf688f3734a8..59076428f0059 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1536,9 +1536,14 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option delivery_errors.append(msg) continue - # Prefer the live adapter when the gateway is running — this supports E2EE - # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. - runtime_adapter = (adapters or {}).get(platform) + # Cron delivery runs outside the gateway adapter's ownership context. + # Use the standalone sender here; otherwise the DeliveryRouter can reuse + # a live aiohttp-backed adapter across event loops. + runtime_adapter = ( + None + if os.getenv("HERMES_CRON_SESSION") + else (adapters or {}).get(platform) + ) delivered = False target_errors = [] diff --git a/tests/tools/test_send_message_tool.py b/tests/tools/test_send_message_tool.py index 55c5165901b6c..9fee2034da03e 100644 --- a/tests/tools/test_send_message_tool.py +++ b/tests/tools/test_send_message_tool.py @@ -933,6 +933,78 @@ def test_matrix_media_uses_native_adapter_helper(self, tmp_path): finally: doc_path.unlink(missing_ok=True) + + def test_matrix_send_to_platform_cron_skips_live_adapter(self, monkeypatch): + live_adapter = SimpleNamespace( + _client=SimpleNamespace(api=SimpleNamespace(session=SimpleNamespace(_loop=asyncio.get_event_loop()))) + ) + ephemeral = SimpleNamespace(connect=AsyncMock(return_value=True), disconnect=AsyncMock()) + core = AsyncMock(return_value={"success": True, "platform": "matrix", "message_id": "$cron"}) + runner = SimpleNamespace(adapters={Platform.MATRIX: live_adapter}) + pconfig = SimpleNamespace(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.com"}) + monkeypatch.setenv("HERMES_CRON_SESSION", "1") + + with ( + patch("gateway.run._gateway_runner_ref", return_value=runner), + patch("plugins.platforms.matrix.adapter.MatrixAdapter", return_value=ephemeral), + patch("tools.send_message_tool._matrix_send_core", core), + ): + result = asyncio.run( + _send_to_platform(Platform.MATRIX, pconfig, "!room:example.com", "cron text") + ) + + assert result == {"success": True, "platform": "matrix", "message_id": "$cron"} + core.assert_awaited_once_with(ephemeral, "!room:example.com", "cron text", [], None) + ephemeral.connect.assert_awaited_once() + ephemeral.disconnect.assert_awaited_once() + + def test_matrix_send_to_platform_falls_back_for_foreign_loop(self): + foreign_loop = asyncio.new_event_loop() + try: + live_adapter = SimpleNamespace( + _client=SimpleNamespace(api=SimpleNamespace(session=SimpleNamespace(_loop=foreign_loop))) + ) + ephemeral = SimpleNamespace(connect=AsyncMock(return_value=True), disconnect=AsyncMock()) + core = AsyncMock(return_value={"success": True, "platform": "matrix", "message_id": "$fallback"}) + runner = SimpleNamespace(adapters={Platform.MATRIX: live_adapter}) + pconfig = SimpleNamespace(enabled=True, token="tok", extra={}) + with ( + patch("gateway.run._gateway_runner_ref", return_value=runner), + patch("plugins.platforms.matrix.adapter.MatrixAdapter", return_value=ephemeral), + patch("tools.send_message_tool._matrix_send_core", core), + ): + result = asyncio.run( + _send_to_platform( + Platform.MATRIX, pconfig, "!room:example.com", "fallback text" + ) + ) + assert result == {"success": True, "platform": "matrix", "message_id": "$fallback"} + core.assert_awaited_once_with(ephemeral, "!room:example.com", "fallback text", [], None) + ephemeral.connect.assert_awaited_once() + ephemeral.disconnect.assert_awaited_once() + finally: + foreign_loop.close() + + def test_matrix_send_to_platform_reuses_same_loop_live_adapter(self): + async def run(): + live_adapter = SimpleNamespace( + _client=SimpleNamespace(api=SimpleNamespace(session=SimpleNamespace(_loop=asyncio.get_running_loop()))) + ) + core = AsyncMock(return_value={"success": True, "platform": "matrix", "message_id": "$live"}) + runner = SimpleNamespace(adapters={Platform.MATRIX: live_adapter}) + pconfig = SimpleNamespace(enabled=True, token="tok", extra={}) + with ( + patch("gateway.run._gateway_runner_ref", return_value=runner), + patch("tools.send_message_tool._matrix_send_core", core), + ): + result = await _send_to_platform( + Platform.MATRIX, pconfig, "!room:example.com", "live text" + ) + assert result == {"success": True, "platform": "matrix", "message_id": "$live"} + core.assert_awaited_once_with(live_adapter, "!room:example.com", "live text", [], None) + + asyncio.run(run()) + def test_matrix_text_only_uses_adapter_path(self): """Text-only Matrix sends must go through the E2EE-capable adapter. diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index c143ea8064f16..52570f0ce1058 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -1678,7 +1678,7 @@ async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None, runner = _gateway_runner_ref() except Exception: runner = None - if runner is not None: + if runner is not None and not os.getenv("HERMES_CRON_SESSION"): try: from gateway.config import Platform live_adapter = runner.adapters.get(Platform.MATRIX) @@ -1690,6 +1690,18 @@ async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None, ) live_adapter = None + if live_adapter is not None: + # aiohttp sessions are bound to the loop that created them. Cron and + # tool execution can run on a different loop from the gateway, so do + # not reuse a live Matrix adapter across loop ownership boundaries. + try: + adapter_loop = live_adapter._client.api.session._loop + except AttributeError: + adapter_loop = None + if adapter_loop is not None and adapter_loop is not asyncio.get_running_loop(): + logger.debug("Matrix: live adapter belongs to another event loop; using ephemeral adapter") + live_adapter = None + if live_adapter is not None: # NOTE: the live adapter is owned by the gateway — we must NOT # disconnect it. Correctness here depends on this branch returning