fix(cron): avoid live adapter reuse across event loops

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 = []
 

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.
 

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