Async event loop and timeout handling fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #2190 fix: event loop lifecycle + session search lineage exclusion

  merged to main

  files: agent/auxiliary_client.py, model_tools.py, tests/test_model_tools_async_bridge.py

  Show the diff

  • agent/auxiliary_client.py

    @@ -1191,8 +1191,18 @@ def _get_cached_client(
         cache_key = (provider, async_mode, base_url or "", api_key or "")
         with _client_cache_lock:
             if cache_key in _client_cache:
    -            cached_client, cached_default = _client_cache[cache_key]
    -            return cached_client, model or cached_default
    +            cached_client, cached_default, cached_loop = _client_cache[cache_key]
    +            if async_mode:
    +                # Async clients are bound to the event loop that created them.
    +                # A cached async client whose loop has been closed will raise
    +                # "Event loop is closed" when httpx tries to clean up its
    +                # transport.  Discard the stale client and create a fresh one.
    +                if cached_loop is not None and cached_loop.is_closed():
    +                    del _client_cache[cache_key]
    +                else:
    +                    return cached_client, model or cached_default
    +            else:
    +                return cached_client, model or cached_default
         # Build outside the lock
         client, default_model = resolve_provider_client(
             provider,
    @@ -1202,11 +1212,20 @@ def _get_cached_client(
             explicit_api_key=[redacted:secret-assignment]
         )
         if client is not None:
    +        # For async clients, remember which loop they were created on so we
    +        # can detect stale entries later.
    +        bound_loop = None
    +        if async_mode:
    +            try:
    +                import asyncio as _aio
    +                bound_loop = _aio.get_event_loop()
    +            except RuntimeError:
    +                pass
             with _client_cache_lock:
                 if cache_key not in _client_cache:
    -                _client_cache[cache_key] = (client, default_model)
    +                _client_cache[cache_key] = (client, default_model, bound_loop)
                 else:
    -                client, default_model = _client_cache[cache_key]
    +                client, default_model, _ = _client_cache[cache_key]
         return client, model or default_model
     
     

  • model_tools.py

    @@ -24,6 +24,7 @@
     import asyncio
     import os
     import logging
    +import threading
     from typing import Dict, Any, List, Optional, Tuple
     
     from tools.registry import registry
    @@ -36,6 +37,25 @@
     # Async Bridging  (single source of truth -- used by registry.dispatch too)
     # =============================================================================
     
    +_tool_loop = None          # persistent loop for the main (CLI) thread
    +_tool_loop_lock = threading.Lock()
    +
    +
    +def _get_tool_loop():
    +    """Return a long-lived event loop for running async tool handlers.
    +
    +    Using a persistent loop (instead of asyncio.run() which creates and
    +    *closes* a fresh loop every time) prevents "Event loop is closed"
    +    errors that occur when cached httpx/AsyncOpenAI clients attempt to
    +    close their transport on a dead loop during garbage collection.
    +    """
    +    global _tool_loop
    +    with _tool_loop_lock:
    +        if _tool_loop is None or _tool_loop.is_closed():
    +            _tool_loop = asyncio.new_event_loop()
    +        return _tool_loop
    +
    +
     def _run_async(coro):
         """Run an async coroutine from a sync context.
     
    @@ -44,6 +64,10 @@ def _run_async(coro):
         disposable thread so asyncio.run() can create its own loop without
         conflicting.
     
    +    For the common CLI path (no running loop), we use a persistent event
    +    loop so that cached async clients (httpx / AsyncOpenAI) remain bound
    +    to a live loop and don't trigger "Event loop is closed" on GC.
    +
         This is the single source of truth for sync->async bridging in tool
         handlers. The RL paths (agent_loop.py, tool_context.py) also provide
         outer thread-pool wrapping as defense-in-depth, but each handler is
    @@ -59,7 +83,9 @@ def _run_async(coro):
             with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                 future = pool.submit(asyncio.run, coro)
                 return future.result(timeout=300)
    -    return asyncio.run(coro)
    +
    +    tool_loop = _get_tool_loop()
    +    return tool_loop.run_until_complete(coro)
     
     
     # =============================================================================

  • tests/test_model_tools_async_bridge.py

    @@ -0,0 +1,211 @@
    +"""Regression tests for the _run_async() event-loop lifecycle.
    +
    +These tests verify the fix for GitHub issue #2104:
    +  "Event loop is closed" after vision_analyze used as first call in session.
    +
    +Root cause: asyncio.run() creates and *closes* a fresh event loop on every
    +call.  Cached httpx/AsyncOpenAI clients that were bound to the now-dead loop
    +would crash with RuntimeError("Event loop is closed") when garbage-collected.
    +
    +The fix replaces asyncio.run() with a persistent event loop in _run_async().
    +"""
    +
    +import asyncio
    +import json
    +import threading
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +
    +# ---------------------------------------------------------------------------
    +# Helpers
    +# ---------------------------------------------------------------------------
    +
    +async def _get_current_loop():
    +    """Return the running event loop from inside a coroutine."""
    +    return asyncio.get_event_loop()
    +
    +
    +async def _create_and_return_transport():
    +    """Simulate an async client creating a transport on the current loop.
    +
    +    Returns a simple asyncio.Future bound to the running loop so we can
    +    later check whether the loop is still alive.
    +    """
    +    loop = asyncio.get_event_loop()
    +    fut = loop.create_future()
    +    fut.set_result("ok")
    +    return loop, fut
    +
    +
    +# ---------------------------------------------------------------------------
    +# Tests
    +# ---------------------------------------------------------------------------
    +
    +class TestRunAsyncLoopLifecycle:
    +    """Verify _run_async() keeps the event loop alive after returning."""
    +
    +    def test_loop_not_closed_after_run_async(self):
    +        """The loop used by _run_async must still be open after the call."""
    +        from model_tools import _run_async
    +
    +        loop = _run_async(_get_current_loop())
    +
    +        assert not loop.is_closed(), (
    +            "_run_async() closed the event loop — cached async clients will "
    +            "crash with 'Event loop is closed' on GC (issue #2104)"
    +        )
    +
    +    def test_same_loop_reused_across_calls(self):
    +        """Consecutive _run_async calls should reuse the same loop."""
    +        from model_tools import _run_async
    +
    +        loop1 = _run_async(_get_current_loop())
    +        loop2 = _run_async(_get_current_loop())
    +
    +        assert loop1 is loop2, (
    +            "_run_async() created a new loop on the second call — cached "
    +            "async clients from the first call would be orphaned"
    +        )
    +
    +    def test_cached_transport_survives_between_calls(self):
    +        """A transport/future created in call 1 must be valid in call 2."""
    +        from model_tools import _run_async
    +
    +        loop, fut = _run_async(_create_and_return_transport())
    +
    +        assert not loop.is_closed()
    +        assert fut.result() == "ok"
    +
    +        loop2 = _run_async(_get_current_loop())
    +        assert loop2 is loop, "Loop changed between calls"
    +        assert not loop.is_closed(), "Loop closed before second call"
    +
    +
    +class TestRunAsyncWithRunningLoop:
    +    """When a loop is already running, _run_async falls back to a thread."""
    +
    +    @pytest.mark.asyncio
    +    async def test_run_async_from_async_context(self):
    +        """_run_async should still work when called from inside an
    +        already-running event loop (gateway / Atropos path)."""
    +        from model_tools import _run_async
    +
    +        async def _simple():
    +            return 42
    +
    +        result = await asyncio.get_event_loop().run_in_executor(
    +            None, _run_async, _simple()
    +        )
    +        assert result == 42
    +
    +
    +# ---------------------------------------------------------------------------
    +# Integration: full vision_analyze dispatch chain
    +# ---------------------------------------------------------------------------
    +
    +def _mock_vision_response():
    +    """Build a fake LLM response matching async_call_llm's return shape."""
    +    message = SimpleNamespace(content="A cat sitting on a chair.")
    +    choice = SimpleNamespace(index=0, message=message, finish_reason="stop")
    +    return SimpleNamespace(choices=[choice], model="test/vision", usage=None)
    +
    +
    +class TestVisionDispatchLoopSafety:
    +    """Simulate the full registry.dispatch('vision_analyze') chain and
    +    verify the event loop stays alive afterwards — the exact scenario
    +    from issue #2104."""
    +
    +    def test_vision_dispatch_keeps_loop_alive(self, tmp_path):
    +        """After dispatching vision_analyze via the registry, the event
    +        loop must remain open so cached async clients don't crash on GC."""
    +        from model_tools import _run_async, _get_tool_loop
    +        from tools.registry import registry
    +
    +        fake_response = _mock_vision_response()
    +
    +        with (
    +            patch(
    +                "tools.vision_tools.async_call_llm",
    +                new_callable=AsyncMock,
    +                return_value=fake_response,
    +            ),
    +            patch(
    +                "tools.vision_tools._download_image",
    +                new_callable=AsyncMock,
    +                side_effect=lambda url, dest, **kw: _write_fake_image(dest),
    +            ),
    +            patch(
    +                "tools.vision_tools._validate_image_url",
    +                return_value=True,
    +            ),
    +            patch(
    +                "tools.vision_tools._image_to_base64_data_url",
    +                return_value="data:image/jpeg;base64,abc",
    +            ),
    +        ):
    +            result_json = registry.dispatch(
    +                "vision_analyze",
    +                {"image_url": "https://example.com/cat.png", "question": "What is this?"},
    +            )
    +
    +        result = json.loads(result_json)
    +        assert result.get("success") is True, f"dispatch failed: {result}"
    +        assert "cat" in result.get("analysis", "").lower()
    +
    +        loop = _get_tool_loop()
    +        assert not loop.is_closed(), (
    +            "Event loop closed after vision_analyze dispatch — cached async "
    +            "clients will crash with 'Event loop is closed' (issue #2104)"
    +        )
    +
    +    def test_two_consecutive_vision_dispatches(self, tmp_path):
    +        """Two back-to-back vision_analyze dispatches must both succeed
    +        and share the same loop (simulates 'first call fails, second
    +        works' from the issue report)."""
    +        from model_tools import _get_tool_loop
    +        from tools.registry import registry
    +
    +        fake_response = _mock_vision_response()
    +
    +        with (
    +            patch(
    +                "tools.vision_tools.async_call_llm",
    +                new_callable=AsyncMock,
    +                return_value=fake_response,
    +            ),
    +            patch(
    +                "tools.vision_tools._download_image",
    +                new_callable=AsyncMock,
    +                side_effect=lambda url, dest, **kw: _write_fake_image(dest),
    +            ),
    +            patch(
    +                "tools.vision_tools._validate_image_url",
    +                return_value=True,
    +            ),
    +            patch(
    +                "tools.vision_tools._image_to_base64_data_url",
    +                return_value="data:image/jpeg;base64,abc",
    +            ),
    +        ):
    +            args = {"image_url": "https://example.com/cat.png", "question": "Describe"}
    +
    +            r1 = json.loads(registry.dispatch("vision_analyze", args))
    +            loop_after_first = _get_tool_loop()
    +
    +            r2 = json.loads(registry.dispatch("vision_analyze", args))
    +            loop_after_second = _get_tool_loop()
    +
    +        assert r1.get("success") is True
    +        assert r2.get("success") is True
    +        assert loop_after_first is loop_after_second, "Loop changed between dispatches"
    +        assert not loop_after_second.is_closed()
    +
    +
    +def _write_fake_image(dest):
    +    """Write minimal bytes so vision_analyze_tool thinks download succeeded."""
    +    dest.parent.mkdir(parents=True, exist_ok=True)
    +    dest.write_bytes(b"\xff\xd8\xff" + b"\x00" * 16)
    +    return dest

• #2701 fix(agent): prevent AsyncOpenAI/httpx cross-loop deadlock in gateway mode

  merged to main

  files: agent/auxiliary_client.py, tests/test_crossloop_client_cache.py, tools/session_search_tool.py

• #3398 fix: eliminate 'Event loop is closed' / 'Press ENTER to continue' during idle

  merged to main

  files: agent/auxiliary_client.py, cli.py, tests/test_async_httpx_del_neuter.py

• #10470 fix: bound auxiliary client cache to prevent fd exhaustion in long-running gateways

  merged to main

  files: agent/auxiliary_client.py, tests/run_agent/test_async_httpx_del_neuter.py

  Show the diff

  • agent/auxiliary_client.py

    @@ -1835,9 +1835,15 @@ def auxiliary_max_tokens_param(value: int) -> dict:
     # Every auxiliary LLM consumer should use these instead of manually
     # constructing clients and calling .chat.completions.create().
     
    -# Client cache: (provider, async_mode, base_url, api_key) -> (client, default_model)
    +# Client cache: (provider, async_mode, base_url, api_key, api_mode, runtime_key) -> (client, default_model, loop)
    +# NOTE: loop identity is NOT part of the key.  On async cache hits we check
    +# whether the cached loop is the *current* loop; if not, the stale entry is
    +# replaced in-place.  This bounds cache growth to one entry per unique
    +# provider config rather than one per (config × event-loop), which previously
    +# caused unbounded fd accumulation in long-running gateway processes (#10200).
     _client_cache: Dict[tuple, tuple] = {}
     _client_cache_lock = threading.Lock()
    +_CLIENT_CACHE_MAX_SIZE = 64  # safety belt — evict oldest when exceeded
     
     
     def neuter_async_httpx_del() -> None:
    @@ -1970,39 +1976,49 @@ def _get_cached_client(
         Async clients (AsyncOpenAI) use httpx.AsyncClient internally, which
         binds to the event loop that was current when the client was created.
         Using such a client on a *different* loop causes deadlocks or
    -    RuntimeError.  To prevent cross-loop issues (especially in gateway
    -    mode where _run_async() may spawn fresh loops in worker threads), the
    -    cache key for async clients includes the current event loop's identity
    -    so each loop gets its own client instance.
    +    RuntimeError.  To prevent cross-loop issues, the cache validates on
    +    every async hit that the cached loop is the *current, open* loop.
    +    If the loop changed (e.g. a new gateway worker-thread loop), the stale
    +    entry is replaced in-place rather than creating an additional entry.
    +
    +    This keeps cache size bounded to one entry per unique provider config,
    +    preventing the fd-exhaustion that previously occurred in long-running
    +    gateways where recycled worker threads created unbounded entries (#10200).
         """
    -    # Include loop identity for async clients to prevent cross-loop reuse.
    -    # httpx.AsyncClient (inside AsyncOpenAI) is bound to the loop where it
    -    # was created — reusing it on a different loop causes deadlocks (#2681).
    -    loop_id = 0
    +    # Resolve the current event loop for async clients so we can validate
    +    # cached entries.  Loop identity is NOT in the cache key — instead we
    +    # check at hit time whether the cached loop is still current and open.
    +    # This prevents unbounded cache growth from recycled worker-thread loops
    +    # while still guaranteeing we never reuse a client on the wrong loop
    +    # (which causes deadlocks, see #2681).
         current_loop = None
         if async_mode:
             try:
                 import asyncio as _aio
                 current_loop = _aio.get_event_loop()
    -            loop_id = id(current_loop)
             except RuntimeError:
                 pass
         runtime = _normalize_main_runtime(main_runtime)
         runtime_key = tuple(runtime.get(field, "") for field in _MAIN_RUNTIME_FIELDS) if provider == "auto" else ()
    -    cache_key = (provider, async_mode, base_url or "", api_key or "", api_mode or "", loop_id, runtime_key)
    +    cache_key = (provider, async_mode, base_url or "", api_key or "", api_mode or "", runtime_key)
         with _client_cache_lock:
             if cache_key in _client_cache:
                 cached_client, cached_default, cached_loop = _client_cache[cache_key]
                 if async_mode:
    -                # A cached async client whose loop has been closed will raise
    -                # "Event loop is closed" when httpx tries to clean up its
    -                # transport.  Discard the stale client and create a fresh one.
    -                if cached_loop is not None and cached_loop.is_closed():
    -                    _force_close_async_httpx(cached_client)
    -                    del _client_cache[cache_key]
    -                else:
    +                # Validate: the cached client must be bound to the CURRENT,
    +                # OPEN loop.  If the loop changed or was closed, the httpx
    +                # transport inside is dead — force-close and replace.
    +                loop_ok = (
    +                    cached_loop is not None
    +                    and cached_loop is current_loop
    +                    and not cached_loop.is_closed()
    +                )
    +                if loop_ok:
                         effective = _compat_model(cached_client, model, cached_default)
                         return cached_client, effective
    +                # Stale — evict and fall through to create a new client.
    +                _force_close_async_httpx(cached_client)
    +                del _client_cache[cache_key]
                 else:
                     effective = _compat_model(cached_client, model, cached_default)
                     return cached_client, effective
    @@ -2022,6 +2038,12 @@ def _get_cached_client(
             bound_loop = current_loop
             with _client_cache_lock:
                 if cache_key not in _client_cache:
    +                # Safety belt: if the cache has grown beyond the max, evict
    +                # the oldest entries (FIFO — dict preserves insertion order).
    +                while len(_client_cache) >= _CLIENT_CACHE_MAX_SIZE:
    +                    evict_key, evict_entry = next(iter(_client_cache.items()))
    +                    _force_close_async_httpx(evict_entry[0])
    +                    del _client_cache[evict_key]
                     _client_cache[cache_key] = (client, default_model, bound_loop)
                 else:
                     client, default_model, _ = _client_cache[cache_key]

  • tests/run_agent/test_async_httpx_del_neuter.py

    @@ -103,7 +103,7 @@ def test_removes_stale_entries(self):
             mock_client._client = MagicMock()
             mock_client._client.is_closed = False
     
    -        key = ("test_stale", True, "", "", id(loop))
    +        key = ("test_stale", True, "", "", "", ())
             with _client_cache_lock:
                 _client_cache[key] = (mock_client, "test-model", loop)
     
    @@ -127,7 +127,7 @@ def test_keeps_live_entries(self):
             loop = asyncio.new_event_loop()  # NOT closed
     
             mock_client = MagicMock()
    -        key = ("test_live", True, "", "", id(loop))
    +        key = ("test_live", True, "", "", "", ())
             with _client_cache_lock:
                 _client_cache[key] = (mock_client, "test-model", loop)
     
    @@ -149,7 +149,7 @@ def test_keeps_entries_without_loop(self):
             )
     
             mock_client = MagicMock()
    -        key = ("test_sync", False, "", "", 0)
    +        key = ("test_sync", False, "", "", "", ())
             with _client_cache_lock:
                 _client_cache[key] = (mock_client, "test-model", None)
     
    @@ -160,3 +160,131 @@ def test_keeps_entries_without_loop(self):
             finally:
                 with _client_cache_lock:
                     _client_cache.pop(key, None)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Cache bounded growth (#10200)
    +# ---------------------------------------------------------------------------
    +
    +class TestClientCacheBoundedGrowth:
    +    """Verify the cache stays bounded when loops change (fix for #10200).
    +
    +    Previously, loop_id was part of the cache key, so every new event loop
    +    created a new entry for the same provider config.  Now loop identity is
    +    validated at hit time and stale entries are replaced in-place.
    +    """
    +
    +    def test_same_key_replaces_stale_loop_entry(self):
    +        """When the loop changes, the old entry should be replaced, not duplicated."""
    +        from agent.auxiliary_client import (
    +            _client_cache,
    +            _client_cache_lock,
    +            _get_cached_client,
    +        )
    +
    +        key = ("test_replace", True, "", "", "", ())
    +
    +        # Simulate a stale entry from a closed loop
    +        old_loop = asyncio.new_event_loop()
    +        old_loop.close()
    +        old_client = MagicMock()
    +        old_client._client = MagicMock()
    +        old_client._client.is_closed = False
    +
    +        with _client_cache_lock:
    +            _client_cache[key] = (old_client, "old-model", old_loop)
    +
    +        try:
    +            # Now call _get_cached_client — should detect stale loop and evict
    +            with patch("agent.auxiliary_client.resolve_provider_client") as mock_resolve:
    +                mock_resolve.return_value = (MagicMock(), "new-model")
    +                client, model = _get_cached_client(
    +                    "test_replace", async_mode=True,
    +                )
    +            # The old entry should have been replaced
    +            with _client_cache_lock:
    +                assert key in _client_cache, "Key should still exist (replaced)"
    +                entry = _client_cache[key]
    +                assert entry[1] == "new-model", "Should have the new model"
    +        finally:
    +            with _client_cache_lock:
    +                _client_cache.pop(key, None)
    +
    +    def test_different_loops_do_not_grow_cache(self):
    +        """Multiple event loops for the same provider should NOT create multiple entries."""
    +        from agent.auxiliary_client import (
    +            _client_cache,
    +            _client_cache_lock,
    +        )
    +
    +        key = ("test_no_grow", True, "", "", "", ())
    +
    +        loops = []
    +        try:
    +            for i in range(5):
    +                loop = asyncio.new_event_loop()
    +                loops.append(loop)
    +                mock_client = MagicMock()
    +                mock_client._client = MagicMock()
    +                mock_client._client.is_closed = False
    +
    +                # Close previous loop entries (simulating worker thread recycling)
    +                if i > 0:
    +                    loops[i - 1].close()
    +
    +                with _client_cache_lock:
    +                    # Simulate what _get_cached_client does: replace on loop mismatch
    +                    if key in _client_cache:
    +                        old_entry = _client_cache[key]
    +                        del _client_cache[key]
    +                    _client_cache[key] = (mock_client, f"model-{i}", loop)
    +
    +            # Only one entry should exist for this key
    +            with _client_cache_lock:
    +                count = sum(1 for k in _client_cache if k == key)
    +                assert count == 1, f"Expected 1 entry, got {count}"
    +        finally:
    +            for loop in loops:
    +                if not loop.is_closed():
    +                    loop.close()
    +            with _client_cache_lock:
    +                _client_cache.pop(key, None)
    +
    +    def test_max_cache_size_eviction(self):
    +        """Cache should not exceed _CLIENT_CACHE_MAX_SIZE."""
    +        from agent.auxiliary_client import (
    +            _client_cache,
    +            _client_cache_lock,
    +            _CLIENT_CACHE_MAX_SIZE,
    +        )
    +
    +        # Save existing cache state
    +        with _client_cache_lock:
    +            saved = dict(_client_cache)
    +            _client_cache.clear()
    +
    +        try:
    +            # Fill to max + 5
    +            for i in range(_CLIENT_CACHE_MAX_SIZE + 5):
    +                mock_client = MagicMock()
    +                mock_client._client = MagicMock()
    +                mock_client._client.is_closed = False
    +                key = (f"evict_test_{i}", False, "", "", "", ())
    +                with _client_cache_lock:
    +                    # Inline the eviction logic (same as _get_cached_client)
    +                    while len(_client_cache) >= _CLIENT_CACHE_MAX_SIZE:
    +                        evict_key = next(iter(_client_cache))
    +                        del _client_cache[evict_key]
    +                    _client_cache[key] = (mock_client, f"model-{i}", None)
    +
    +            with _client_cache_lock:
    +                assert len(_client_cache) <= _CLIENT_CACHE_MAX_SIZE, \
    +                    f"Cache size {len(_client_cache)} exceeds max {_CLIENT_CACHE_MAX_SIZE}"
    +                # The earliest entries should have been evicted
    +                assert ("evict_test_0", False, "", "", "", ()) not in _client_cache
    +                # The latest entries should be present
    +                assert (f"evict_test_{_CLIENT_CACHE_MAX_SIZE + 4}", False, "", "", "", ()) in _client_cache
    +        finally:
    +            with _client_cache_lock:
    +                _client_cache.clear()
    +                _client_cache.update(saved)

• #11594 fix(cron+weixin): Batch-4 salvage — origin fallback dict + Weixin session split + multi-target deliver

  merged to main

  files: cron/scheduler.py, gateway/platforms/weixin.py, scripts/release.py, tests/cron/test_scheduler.py, tests/gateway/test_weixin.py, tools/send_message_tool.py

  Show the diff

  • cron/scheduler.py

    @@ -27,7 +27,7 @@
         except ImportError:
             msvcrt = None
     from pathlib import Path
    -from typing import Optional
    +from typing import List, Optional
     
     # Add parent directory to path for imports BEFORE repo-level imports.
     # Without this, standalone invocations (e.g. after `hermes update` reloads
    @@ -49,6 +49,25 @@
         "qqbot",
     })
     
    +# Platforms that support a configured cron/notification home target, mapped to
    +# the environment variable used by gateway setup/runtime config.
    +_HOME_TARGET_ENV_VARS = {
    +    "matrix": "MATRIX_HOME_ROOM",
    +    "telegram": "TELEGRAM_HOME_CHANNEL",
    +    "discord": "DISCORD_HOME_CHANNEL",
    +    "slack": "SLACK_HOME_CHANNEL",
    +    "signal": "SIGNAL_HOME_CHANNEL",
    +    "mattermost": "MATTERMOST_HOME_CHANNEL",
    +    "sms": "SMS_HOME_CHANNEL",
    +    "email": "EMAIL_HOME_ADDRESS",
    +    "dingtalk": "DINGTALK_HOME_CHANNEL",
    +    "feishu": "FEISHU_HOME_CHANNEL",
    +    "wecom": "WECOM_HOME_CHANNEL",
    +    "weixin": "WEIXIN_HOME_CHANNEL",
    +    "bluebubbles": "BLUEBUBBLES_HOME_CHANNEL",
    +    "qqbot": "QQ_HOME_CHANNEL",
    +}
    +
     from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run
     
     # Sentinel: when a cron agent has nothing new to report, it can start its
    @@ -76,15 +95,23 @@ def _resolve_origin(job: dict) -> Optional[dict]:
         return None
     
     
    -def _resolve_delivery_target(job: dict) -> Optional[dict]:
    -    """Resolve the concrete auto-delivery target for a cron job, if any."""
    -    deliver = job.get("deliver", "local")
    +def _get_home_target_chat_id(platform_name: str) -> str:
    +    """Return the configured home target chat/room ID for a delivery platform."""
    +    env_var = _HOME_TARGET_ENV_VARS.get(platform_name.lower())
    +    if not env_var:
    +        return ""
    +    return os.getenv(env_var, "")
    +
    +
    +def _resolve_single_delivery_target(job: dict, deliver_value: str) -> Optional[dict]:
    +    """Resolve one concrete auto-delivery target for a cron job."""
    +
         origin = _resolve_origin(job)
     
    -    if deliver == "local":
    +    if deliver_value == "local":
             return None
     
    -    if deliver == "origin":
    +    if deliver_value == "origin":
             if origin:
                 return {
                     "platform": origin["platform"],
    @@ -93,8 +120,8 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
                 }
             # Origin missing (e.g. job created via API/script) — try each
             # platform's home channel as a fallback instead of silently dropping.
    -        for platform_name in ("matrix", "telegram", "discord", "slack", "bluebubbles"):
    -            chat_id = os.getenv(f"{platform_name.upper()}_HOME_CHANNEL", "")
    +        for platform_name in _HOME_TARGET_ENV_VARS:
    +            chat_id = _get_home_target_chat_id(platform_name)
                 if chat_id:
                     logger.info(
                         "Job '%s' has deliver=origin but no origin; falling back to %s home channel",
    @@ -108,8 +135,8 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
                     }
             return None
     
    -    if ":" in deliver:
    -        platform_name, rest = deliver.split(":", 1)
    +    if ":" in deliver_value:
    +        platform_name, rest = deliver_value.split(":", 1)
             platform_key = platform_name.lower()
     
             from tools.send_message_tool import _parse_target_ref
    @@ -139,7 +166,7 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
                 "thread_id": thread_id,
             }
     
    -    platform_name = deliver
    +    platform_name = deliver_value
         if origin and origin.get("platform") == platform_name:
             return {
                 "platform": platform_name,
    @@ -149,7 +176,7 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
     
         if platform_name.lower() not in _KNOWN_DELIVERY_PLATFORMS:
             return None
    -    chat_id = os.getenv(f"{platform_name.upper()}_HOME_CHANNEL", "")
    +    chat_id = _get_home_target_chat_id(platform_name)
         if not chat_id:
             return None
     
    @@ -160,6 +187,30 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]:
         }
     
     
    +def _resolve_delivery_targets(job: dict) -> List[dict]:
    +    """Resolve all concrete auto-delivery targets for a cron job (supports comma-separated deliver)."""
    +    deliver = job.get("deliver", "local")
    +    if deliver == "local":
    +        return []
    +    parts = [p.strip() for p in str(deliver).split(",") if p.strip()]
    +    seen = set()
    +    targets = []
    +    for part in parts:
    +        target = _resolve_single_delivery_target(job, part)
    +        if target:
    +            key = (target["platform"].lower(), str(target["chat_id"]), target.get("thread_id"))
    +            if key not in seen:
    +                seen.add(key)
    +                targets.append(target)
    +    return targets
    +
    +
    +def _resolve_delivery_target(job: dict) -> Optional[dict]:
    +    """Resolve the concrete auto-delivery target for a cron job, if any."""
    +    targets = _resolve_delivery_targets(job)
    +    return targets[0] if targets else None
    +
    +
     # Media extension sets — keep in sync with gateway/platforms/base.py:_process_message_background
     _AUDIO_EXTS = frozenset({'.ogg', '.opus', '.mp3', '.wav', '.m4a'})
     _VIDEO_EXTS = frozenset({'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'})
    @@ -200,7 +251,7 @@ def _send_media_via_adapter(adapter, chat_id: str, media_files: list, metadata:
     
     def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]:
         """
    -    Deliver job output to the configured target (origin chat, specific platform, etc.).
    +    Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
     
         When ``adapters`` and ``loop`` are provided (gateway is running), tries to
         use the live adapter first — this supports E2EE rooms (e.g. Matrix) where
    @@ -209,33 +260,14 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     
         Returns None on success, or an error string on failure.
         """
    -    target = _resolve_delivery_target(job)
    -    if not target:
    +    targets = _resolve_delivery_targets(job)
    +    if not targets:
             if job.get("deliver", "local") != "local":
                 msg = f"no delivery target resolved for deliver={job.get('deliver', 'local')}"
                 logger.warning("Job '%s': %s", job["id"], msg)
                 return msg
             return None  # local-only jobs don't deliver — not a failure
     
    -    platform_name = target["platform"]
    -    chat_id = target["chat_id"]
    -    thread_id = target.get("thread_id")
    -
    -    # Diagnostic: log thread_id for topic-aware delivery debugging
    -    origin = job.get("origin") or {}
    -    origin_thread = origin.get("thread_id")
    -    if origin_thread and not thread_id:
    -        logger.warning(
    -            "Job '%s': origin has thread_id=%s but delivery target lost it "
    -            "(deliver=%s, target=%s)",
    -            job["id"], origin_thread, job.get("deliver", "local"), target,
    -        )
    -    elif thread_id:
    -        logger.debug(
    -            "Job '%s': delivering to %s:%s thread_id=%s",
    -            job["id"], platform_name, chat_id, thread_id,
    -        )
    -
         from tools.send_message_tool import _send_to_platform
         from gateway.config import load_gateway_config, Platform
     
    @@ -258,24 +290,6 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
             "bluebubbles": Platform.BLUEBUBBLES,
             "qqbot": Platform.QQBOT,
         }
    -    platform = platform_map.get(platform_name.lower())
    -    if not platform:
    -        msg = f"unknown platform '{platform_name}'"
    -        logger.warning("Job '%s': %s", job["id"], msg)
    -        return msg
    -
    -    try:
    -        config = load_gateway_config()
    -    except Exception as e:
    -        msg = f"failed to load gateway config: {e}"
    -        logger.error("Job '%s': %s", job["id"], msg)
    -        return msg
    -
    -    pconfig = config.platforms.get(platform)
    -    if not pconfig or not pconfig.enabled:
    -        msg = f"platform '{platform_name}' not configured/enabled"
    -        logger.warning("Job '%s': %s", job["id"], msg)
    -        return msg
     
         # Optionally wrap the content with a header/footer so the user knows this
         # is a cron delivery.  Wrapping is on by default; set cron.wrap_response: false
    @@ -304,67 +318,117 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         from gateway.platforms.base import BasePlatformAdapter
         media_files, cleaned_delivery_content = BasePlatformAdapter.extract_media(delivery_content)
     
    -    # 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)
    -    if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
    -        send_metadata = {"thread_id": thread_id} if thread_id else None
    -        try:
    -            # Send cleaned text (MEDIA tags stripped) — not the raw content
    -            text_to_send = cleaned_delivery_content.strip()
    -            adapter_ok = True
    -            if text_to_send:
    -                future = asyncio.run_coroutine_threadsafe(
    -                    runtime_adapter.send(chat_id, text_to_send, metadata=send_metadata),
    -                    loop,
    -                )
    -                send_result = future.result(timeout=60)
    -                if send_result and not getattr(send_result, "success", True):
    -                    err = getattr(send_result, "error", "unknown")
    -                    logger.warning(
    -                        "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
    -                        job["id"], platform_name, chat_id, err,
    -                    )
    -                    adapter_ok = False  # fall through to standalone path
    +    try:
    +        config = load_gateway_config()
    +    except Exception as e:
    +        msg = f"failed to load gateway config: {e}"
    +        logger.error("Job '%s': %s", job["id"], msg)
    +        return msg
     
    -            # Send extracted media files as native attachments via the live adapter
    -            if adapter_ok and media_files:
    -                _send_media_via_adapter(runtime_adapter, chat_id, media_files, send_metadata, loop, job)
    +    delivery_errors = []
     
    -            if adapter_ok:
    -                logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
    -                return None
    -        except Exception as e:
    +    for target in targets:
    +        platform_name = target["platform"]
    +        chat_id = target["chat_id"]
    +        thread_id = target.get("thread_id")
    +
    +        # Diagnostic: log thread_id for topic-aware delivery debugging
    +        origin = job.get("origin") or {}
    +        origin_thread = origin.get("thread_id")
    +        if origin_thread and not thread_id:
                 logger.warning(
    -                "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
    -                job["id"], platform_name, chat_id, e,
    +                "Job '%s': origin has thread_id=%s but delivery target lost it "
    +                "(deliver=%s, target=%s)",
    +                job["id"], origin_thread, job.get("deliver", "local"), target,
    +            )
    +        elif thread_id:
    +            logger.debug(
    +                "Job '%s': delivering to %s:%s thread_id=%s",
    +                job["id"], platform_name, chat_id, thread_id,
                 )
     
    -    # Standalone path: run the async send in a fresh event loop (safe from any thread)
    -    coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
    -    try:
    -        result = asyncio.run(coro)
    -    except RuntimeError:
    -        # asyncio.run() checks for a running loop before awaiting the coroutine;
    -        # when it raises, the original coro was never started — close it to
    -        # prevent "coroutine was never awaited" RuntimeWarning, then retry in a
    -        # fresh thread that has no running loop.
    -        coro.close()
    -        import concurrent.futures
    -        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
    -            future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files))
    -            result = future.result(timeout=30)
    -    except Exception as e:
    -        msg = f"delivery to {platform_name}:{chat_id} failed: {e}"
    -        logger.error("Job '%s': %s", job["id"], msg)
    -        return msg
    +        platform = platform_map.get(platform_name.lower())
    +        if not platform:
    +            msg = f"unknown platform '{platform_name}'"
    +            logger.warning("Job '%s': %s", job["id"], msg)
    +            delivery_errors.append(msg)
    +            continue
     
    -    if result and result.get("error"):
    -        msg = f"delivery error: {result['error']}"
    -        logger.error("Job '%s': %s", job["id"], msg)
    -        return msg
    +        # 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)
    +        delivered = False
    +        if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
    +            send_metadata = {"thread_id": thread_id} if thread_id else None
    +            try:
    +                # Send cleaned text (MEDIA tags stripped) — not the raw content
    +                text_to_send = cleaned_delivery_content.strip()
    +                adapter_ok = True
    +                if text_to_send:
    +                    future = asyncio.run_coroutine_threadsafe(
    +                        runtime_adapter.send(chat_id, text_to_send, metadata=send_metadata),
    +                        loop,
    +                    )
    +                    send_result = future.result(timeout=60)
    +                    if send_result and not getattr(send_result, "success", True):
    +                        err = getattr(send_result, "error", "unknown")
    +                        logger.warning(
    +                            "Job '%s': live adapter send to %s:%s failed (%s), falling back to standalone",
    +                            job["id"], platform_name, chat_id, err,
    +                        )
    +                        adapter_ok = False  # fall through to standalone path
    +
    +                # Send extracted media files as native attachments via the live adapter
    +                if adapter_ok and media_files:
    +                    _send_media_via_adapter(runtime_adapter, chat_id, media_files, send_metadata, loop, job)
    +
    +                if adapter_ok:
    +                    logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
    +                    delivered = True
    +            except Exception as e:
    +                logger.warning(
    +                    "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
    +                    job["id"], platform_name, chat_id, e,
    +                )
    +
    +        if not delivered:
    +            pconfig = config.platforms.get(platform)
    +            if not pconfig or not pconfig.enabled:
    +                msg = f"platform '{platform_name}' not configured/enabled"
    +                logger.warning("Job '%s': %s", job["id"], msg)
    +                delivery_errors.append(msg)
    +                continue
    +
    +            # Standalone path: run the async send in a fresh event loop (safe from any thread)
    +            coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
    +            try:
    +                result = asyncio.run(coro)
    +            except RuntimeError:
    +                # asyncio.run() checks for a running loop before awaiting the coroutine;
    +                # when it raises, the original coro was never started — close it to
    +                # prevent "coroutine was never awaited" RuntimeWarning, then retry in a
    +                # fresh thread that has no running loop.
    +                coro.close()
    +                import concurrent.futures
    +                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
    +                    future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files))
    +                    result = future.result(timeout=30)
    +            except Exception as e:
    +                msg = f"delivery to {platform_name}:{chat_id} failed: {e}"
    +                logger.error("Job '%s': %s", job["id"], msg)
    +                delivery_errors.append(msg)
    +                continue
    +
    +            if result and result.get("error"):
    +                msg = f"delivery error: {result['error']}"
    +                logger.error("Job '%s': %s", job["id"], msg)
    +                delivery_errors.append(msg)
    +                continue
    +
    +            logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)
     
    -    logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)
    +    if delivery_errors:
    +        return "; ".join(delivery_errors)
         return None
     
     

  • gateway/platforms/weixin.py

    @@ -96,6 +96,8 @@
     MEDIA_FILE = 3
     MEDIA_VOICE = 4
     
    +_LIVE_ADAPTERS: Dict[str, Any] = {}
    +
     ITEM_TEXT = 1
     ITEM_IMAGE = 2
     ITEM_VOICE = 3
    @@ -1052,7 +1054,8 @@ def __init__(self, config: PlatformConfig):
             self._hermes_home = hermes_home
             self._token_store = [redacted:secret-assignment]
             self._typing_cache = TypingTicketCache()
    -        self._session: Optional[aiohttp.ClientSession] = None
    +        self._poll_session: Optional[aiohttp.ClientSession] = None
    +        self._send_session: Optional[aiohttp.ClientSession] = None
             self._poll_task: Optional[asyncio.Task] = None
             self._dedup = MessageDeduplicator(ttl_seconds=MESSAGE_DEDUP_TTL_SECONDS)
     
    @@ -1127,14 +1130,17 @@ async def connect(self) -> bool:
             except Exception as exc:
                 logger.debug("[%s] Token lock unavailable (non-fatal): %s", self.name, exc)
     
    -        self._session = aiohttp.ClientSession(trust_env=True)
    +        self._poll_session = aiohttp.ClientSession(trust_env=True)
    +        self._send_session = aiohttp.ClientSession(trust_env=True)
             self._token_store.restore(self._account_id)
             self._poll_task = asyncio.create_task(self._poll_loop(), name="weixin-poll")
             self._mark_connected()
    +        _LIVE_ADAPTERS[self._token] = self
             logger.info("[%s] Connected account=%s base=%s", self.name, _safe_id(self._account_id), self._base_url)
             return True
     
         async def disconnect(self) -> None:
    +        _LIVE_ADAPTERS.pop(self._token, None)
             self._running = False
             if self._poll_task and not self._poll_task.done():
                 self._poll_task.cancel()
    @@ -1143,15 +1149,18 @@ async def disconnect(self) -> None:
                 except asyncio.CancelledError:
                     pass
             self._poll_task = None
    -        if self._session and not self._session.closed:
    -            await self._session.close()
    -        self._session = None
    +        if self._poll_session and not self._poll_session.closed:
    +            await self._poll_session.close()
    +        self._poll_session = None
    +        if self._send_session and not self._send_session.closed:
    +            await self._send_session.close()
    +        self._send_session = None
             self._release_platform_lock()
             self._mark_disconnected()
             logger.info("[%s] Disconnected", self.name)
     
         async def _poll_loop(self) -> None:
    -        assert self._session is not None
    +        assert self._poll_session is not None
             sync_buf = _load_sync_buf(self._hermes_home, self._account_id)
             timeout_ms = LONG_POLL_TIMEOUT_MS
             consecutive_failures = 0
    @@ -1159,7 +1168,7 @@ async def _poll_loop(self) -> None:
             while self._running:
                 try:
                     response = await _get_updates(
    -                    self._session,
    +                    self._poll_session,
                         base_url=self._base_url,
                         token=[redacted:secret-assignment]
                         sync_buf=sync_buf,
    @@ -1216,7 +1225,7 @@ async def _process_message_safe(self, message: Dict[str, Any]) -> None:
                 logger.error("[%s] unhandled inbound error from=%s: %s", self.name, _safe_id(message.get("from_user_id")), exc, exc_info=True)
     
         async def _process_message(self, message: Dict[str, Any]) -> None:
    -        assert self._session is not None
    +        assert self._poll_session is not None
             sender_id = str(message.get("from_user_id") or "").strip()
             if not sender_id:
                 return
    @@ -1309,7 +1318,7 @@ async def _download_image(self, item: Dict[str, Any]) -> Optional[str]:
             media = _media_reference(item, "image_item")
             try:
                 data = await _download_and_decrypt_media(
    -                self._session,
    +                self._poll_session,
                     cdn_base_url=self._cdn_base_url,
                     encrypted_query_param=media.get("encrypt_query_param"),
                     aes_key_b64=(item.get("image_item") or {}).get("aeskey")
    @@ -1327,7 +1336,7 @@ async def _download_video(self, item: Dict[str, Any]) -> Optional[str]:
             media = _media_reference(item, "video_item")
             try:
                 data = await _download_and_decrypt_media(
    -                self._session,
    +                self._poll_session,
                     cdn_base_url=self._cdn_base_url,
                     encrypted_query_param=media.get("encrypt_query_param"),
                     aes_key_b64=media.get("aes_key"),
    @@ -1346,7 +1355,7 @@ async def _download_file(self, item: Dict[str, Any]) -> Tuple[Optional[str], str
             mime = _mime_from_filename(filename)
             try:
                 data = await _download_and_decrypt_media(
    -                self._session,
    +                self._poll_session,
                     cdn_base_url=self._cdn_base_url,
                     encrypted_query_param=media.get("encrypt_query_param"),
                     aes_key_b64=media.get("aes_key"),
    @@ -1365,7 +1374,7 @@ async def _download_voice(self, item: Dict[str, Any]) -> Optional[str]:
                 return None
             try:
                 data = await _download_and_decrypt_media(
    -                self._session,
    +                self._poll_session,
                     cdn_base_url=self._cdn_base_url,
                     encrypted_query_param=media.get("encrypt_query_param"),
                     aes_key_b64=media.get("aes_key"),
    @@ -1378,13 +1387,13 @@ async def _download_voice(self, item: Dict[str, Any]) -> Optional[str]:
                 return None
     
         async def _maybe_fetch_typing_ticket(self, user_id: str, context_token: [redacted:secret-assignment] -> None:
    -        if not self._session or not self._token:
    +        if not self._poll_session or not self._token:
                 return
             if self._typing_cache.get(user_id):
                 return
             try:
                 response = await _get_config(
    -                self._session,
    +                self._poll_session,
                     base_url=self._base_url,
                     token=[redacted:secret-assignment]
                     user_id=user_id,
    @@ -1414,7 +1423,7 @@ async def _send_text_chunk(
             for attempt in range(self._send_chunk_retries + 1):
                 try:
                     await _send_message(
    -                    self._session,
    +                    self._send_session,
                         base_url=self._base_url,
                         token=[redacted:secret-assignment]
                         to=chat_id,
    @@ -1449,7 +1458,7 @@ async def send(
             reply_to: Optional[str] = None,
             metadata: Optional[Dict[str, Any]] = None,
         ) -> SendResult:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return SendResult(success=False, error="Not connected")
             context_token = [redacted:secret-assignment] chat_id)
             last_message_id: Optional[str] = None
    @@ -1508,14 +1517,14 @@ async def _deliver_media(path: str, is_voice: bool = False) -> None:
                 return SendResult(success=False, error=str(exc))
     
         async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return
             typing_ticket = self._typing_cache.get(chat_id)
             if not typing_ticket:
                 return
             try:
                 await _send_typing(
    -                self._session,
    +                self._send_session,
                     base_url=self._base_url,
                     token=[redacted:secret-assignment]
                     to_user_id=chat_id,
    @@ -1526,14 +1535,14 @@ async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = N
                 logger.debug("[%s] typing start failed for %s: %s", self.name, _safe_id(chat_id), exc)
     
         async def stop_typing(self, chat_id: str) -> None:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return
             typing_ticket = self._typing_cache.get(chat_id)
             if not typing_ticket:
                 return
             try:
                 await _send_typing(
    -                self._session,
    +                self._send_session,
                     base_url=self._base_url,
                     token=[redacted:secret-assignment]
                     to_user_id=chat_id,
    @@ -1585,7 +1594,7 @@ async def send_document(
             caption: str = "",
             metadata: Optional[Dict[str, Any]] = None,
         ) -> SendResult:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return SendResult(success=False, error="Not connected")
             try:
                 message_id = await self._send_file(chat_id, file_path, caption)
    @@ -1602,7 +1611,7 @@ async def send_video(
             reply_to: Optional[str] = None,
             metadata: Optional[Dict[str, Any]] = None,
         ) -> SendResult:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return SendResult(success=False, error="Not connected")
             try:
                 message_id = await self._send_file(chat_id, video_path, caption or "")
    @@ -1619,7 +1628,7 @@ async def send_voice(
             reply_to: Optional[str] = None,
             metadata: Optional[Dict[str, Any]] = None,
         ) -> SendResult:
    -        if not self._session or not self._token:
    +        if not self._send_session or not self._token:
                 return SendResult(success=False, error="Not connected")
     
             # Native outbound Weixin voice bubbles are not proven-working in the
    @@ -1644,8 +1653,8 @@ async def _download_remote_media(self, url: str) -> str:
             if not is_safe_url(url):
                 raise ValueError(f"Blocked unsafe URL (SSRF protection): {url}")
     
    -        assert self._session is not None
    -        async with self._session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
    +        assert self._send_session is not None
    +        async with self._send_session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response:
                 response.raise_for_status()
                 data = await response.read()
                 suffix = Path(url.split("?", 1)[0]).suffix or ".bin"
    @@ -1660,7 +1669,7 @@ async def _send_file(
             caption: str,
             force_file_attachment: bool = False,
         ) -> str:
    -        assert self._session is not None and self._token is not None
    +        assert self._send_session is not None and self._token is not None
             plaintext = Path(path).read_bytes()
             media_type, item_builder = self._outbound_media_builder(path, force_file_attachment=force_file_attachment)
             filekey = secrets.token_hex(16)
    @@ -1668,7 +1677,7 @@ async def _send_file(
             rawsize = len(plaintext)
             rawfilemd5 = hashlib.md5(plaintext).hexdigest()
             upload_response = await _get_upload_url(
    -            self._session,
    +            self._send_session,
                 base_url=self._base_url,
                 token=[redacted:secret-assignment]
                 to_user_id=chat_id,
    @@ -1694,7 +1703,7 @@ async def _send_file(
                 raise RuntimeError(f"getUploadUrl returned neither upload_param nor upload_full_url: {upload_response}")
     
             encrypted_query_param = await _upload_ciphertext(
    -            self._session,
    +            self._send_session,
                 ciphertext=ciphertext,
                 upload_url=upload_url,
             )
    @@ -1722,7 +1731,7 @@ async def _send_file(
             if caption:
                 last_message_id = f"hermes-weixin-{uuid.uuid4().hex}"
                 await _send_message(
    -                self._session,
    +                self._send_session,
                     base_url=self._base_url,
                     token=[redacted:secret-assignment]
                     to=chat_id,
    @@ -1733,7 +1742,7 @@ async def _send_file(
     
             last_message_id = f"hermes-weixin-{uuid.uuid4().hex}"
             await _api_post(
    -            self._session,
    +            self._send_session,
                 base_url=self._base_url,
                 endpoint=EP_SEND_MESSAGE,
                 payload={
    @@ -1857,6 +1866,33 @@ async def send_weixin_direct(
         token_store.restore(account_id)
         context_token = [redacted:secret-assignment] chat_id)
     
    +    live_adapter = _LIVE_ADAPTERS.get(resolved_token)
    +    send_session = getattr(live_adapter, '_send_session', None)
    +    if live_adapter is not None and send_session is not None and not send_session.closed:
    +        last_result: Optional[SendResult] = None
    +        cleaned = live_adapter.format_message(message)
    +        if cleaned:
    +            last_result = await live_adapter.send(chat_id, cleaned)
    +            if not last_result.success:
    +                return {"error": f"Weixin send failed: {last_result.error}"}
    +
    +        for media_path, _is_voice in media_files or []:
    +            ext = Path(media_path).suffix.lower()
    +            if ext in {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}:
    +                last_result = await live_adapter.send_image_file(chat_id, media_path)
    +            else:
    +                last_result = await live_adapter.send_document(chat_id, media_path)
    +            if not last_result.success:
    +                return {"error": f"Weixin media send failed: {last_result.error}"}
    +
    +        return {
    +            "success": True,
    +            "platform": "weixin",
    +            "chat_id": chat_id,
    +            "message_id": last_result.message_id if last_result else None,
    +            "context_token_used": bool(context_token),
    +        }
    +
         async with aiohttp.ClientSession(trust_env=True) as session:
             adapter = WeixinAdapter(
                 PlatformConfig(
    @@ -1870,6 +1906,7 @@ async def send_weixin_direct(
                     },
                 )
             )
    +        adapter._send_session = session
             adapter._session = session
             adapter._token = [redacted:secret-assignment]
             adapter._account_id = account_id

  • scripts/release.py

    @@ -242,6 +242,8 @@
         "[redacted:email]": "imink",
         "[redacted:email]": "hedgeho9X",
         "[redacted:email]": "bernylinville",
    +    "[redacted:email]": "briandevans",
    +    "[redacted:email]": "LLQWQ",
     }
     
     

  • tests/cron/test_scheduler.py

    @@ -64,6 +64,60 @@ def test_origin_delivery_preserves_thread_id(self):
                 "thread_id": "17585",
             }
     
    +    @pytest.mark.parametrize(
    +        ("platform", "env_var", "chat_id"),
    +        [
    +            ("matrix", "MATRIX_HOME_ROOM", "!bot-room:example.org"),
    +            ("signal", "SIGNAL_HOME_CHANNEL", "+15551234567"),
    +            ("mattermost", "MATTERMOST_HOME_CHANNEL", "team-town-square"),
    +            ("sms", "SMS_HOME_CHANNEL", "+15557654321"),
    +            ("email", "EMAIL_HOME_ADDRESS", "[redacted:email]"),
    +            ("dingtalk", "DINGTALK_HOME_CHANNEL", "cidNNN"),
    +            ("feishu", "FEISHU_HOME_CHANNEL", "oc_home"),
    +            ("wecom", "WECOM_HOME_CHANNEL", "wecom-home"),
    +            ("weixin", "WEIXIN_HOME_CHANNEL", "wxid_home"),
    +            ("qqbot", "QQ_HOME_CHANNEL", "group-openid-home"),
    +        ],
    +    )
    +    def test_origin_delivery_without_origin_falls_back_to_supported_home_channels(
    +        self, monkeypatch, platform, env_var, chat_id
    +    ):
    +        for fallback_env in (
    +            "MATRIX_HOME_ROOM",
    +            "MATRIX_HOME_CHANNEL",
    +            "TELEGRAM_HOME_CHANNEL",
    +            "DISCORD_HOME_CHANNEL",
    +            "SLACK_HOME_CHANNEL",
    +            "SIGNAL_HOME_CHANNEL",
    +            "MATTERMOST_HOME_CHANNEL",
    +            "SMS_HOME_CHANNEL",
    +            "EMAIL_HOME_ADDRESS",
    +            "DINGTALK_HOME_CHANNEL",
    +            "BLUEBUBBLES_HOME_CHANNEL",
    +            "FEISHU_HOME_CHANNEL",
    +            "WECOM_HOME_CHANNEL",
    +            "WEIXIN_HOME_CHANNEL",
    +            "QQ_HOME_CHANNEL",
    +        ):
    +            monkeypatch.delenv(fallback_env, raising=False)
    +        monkeypatch.setenv(env_var, chat_id)
    +
    +        assert _resolve_delivery_target({"deliver": "origin"}) == {
    +            "platform": platform,
    +            "chat_id": chat_id,
    +            "thread_id": None,
    +        }
    +
    +    def test_bare_matrix_delivery_uses_matrix_home_room(self, monkeypatch):
    +        monkeypatch.delenv("MATRIX_HOME_CHANNEL", raising=False)
    +        monkeypatch.setenv("MATRIX_HOME_ROOM", "!room123:example.org")
    +
    +        assert _resolve_delivery_target({"deliver": "matrix"}) == {
    +            "platform": "matrix",
    +            "chat_id": "!room123:example.org",
    +            "thread_id": None,
    +        }
    +
         def test_explicit_telegram_topic_target_with_thread_id(self):
             """deliver: 'telegram:chat_id:thread_id' parses correctly."""
             job = {

  • tests/gateway/test_weixin.py

    @@ -311,6 +311,7 @@ class TestWeixinChunkDelivery:
         def _connected_adapter(self) -> WeixinAdapter:
             adapter = _make_adapter()
             adapter._session = object()
    +        adapter._send_session = adapter._session
             adapter._token = "[redacted:secret-assignment]"
             adapter._base_url = "https://weixin.example.com"
             adapter._token_store.get = lambda account_id, chat_id: "ctx-token"
    @@ -420,6 +421,7 @@ def test_split_text_returns_empty_list_for_empty_string_split_per_line(self):
         def test_send_empty_content_does_not_call_send_message(self, send_message_mock):
             adapter = _make_adapter()
             adapter._session = object()
    +        adapter._send_session = adapter._session
             adapter._token = "[redacted:secret-assignment]"
             adapter._base_url = "https://weixin.example.com"
             adapter._token_store.get = lambda account_id, chat_id: "ctx-token"
    @@ -525,6 +527,7 @@ def test_send_image_file_uses_image_path_parameter(self, send_document_mock):
             """Verify send_image_file accepts image_path and forwards to send_document."""
             adapter = _make_adapter()
             adapter._session = object()
    +        adapter._send_session = adapter._session
             adapter._token = "[redacted:secret-assignment]"
     
             send_document_mock.return_value = weixin.SendResult(success=True, message_id="test-id")
    @@ -552,6 +555,7 @@ def test_send_image_file_works_without_optional_params(self, send_document_mock)
             """Verify send_image_file works with minimal required params."""
             adapter = _make_adapter()
             adapter._session = object()
    +        adapter._send_session = adapter._session
             adapter._token = "[redacted:secret-assignment]"
     
             send_document_mock.return_value = weixin.SendResult(success=True, message_id="test-id")
    @@ -576,6 +580,7 @@ class TestWeixinVoiceSending:
         def _connected_adapter(self) -> WeixinAdapter:
             adapter = _make_adapter()
             adapter._session = object()
    +        adapter._send_session = adapter._session
             adapter._token = "[redacted:secret-assignment]"
             adapter._base_url = "https://weixin.example.com"
             adapter._token_store.get = lambda account_id, chat_id: "ctx-token"

  • tools/send_message_tool.py

    @@ -215,7 +215,27 @@ def _handle_send(args):
     
         pconfig = config.platforms.get(platform)
         if not pconfig or not pconfig.enabled:
    -        return tool_error(f"Platform '{platform_name}' is not configured. Set up credentials in ~/.hermes/config.yaml or environment variables.")
    +        # Weixin can be configured purely via .env; synthesize a pconfig so
    +        # send_message and cron delivery work without a gateway.yaml entry.
    +        if platform_name == "weixin":
    +            import os
    +            wx_token = [redacted:secret-assignment]"WEIXIN_TOKEN", "").strip()
    +            wx_account = os.getenv("WEIXIN_ACCOUNT_ID", "").strip()
    +            if wx_token and wx_account:
    +                from gateway.config import PlatformConfig
    +                pconfig = PlatformConfig(
    +                    enabled=True,
    +                    token=[redacted:secret-assignment]
    +                    extra={
    +                        "account_id": wx_account,
    +                        "base_url": os.getenv("WEIXIN_BASE_URL", "").strip(),
    +                        "cdn_base_url": os.getenv("WEIXIN_CDN_BASE_URL", "").strip(),
    +                    },
    +                )
    +            else:
    +                return tool_error(f"Platform '{platform_name}' is not configured. Set up credentials in ~/.hermes/config.yaml or environment variables.")
    +        else:
    +            return tool_error(f"Platform '{platform_name}' is not configured. Set up credentials in ~/.hermes/config.yaml or environment variables.")
     
         from gateway.platforms.base import BasePlatformAdapter
     
    @@ -225,6 +245,12 @@ def _handle_send(args):
         used_home_channel = False
         if not chat_id:
             home = config.get_home_channel(platform)
    +        if not home and platform_name == "weixin":
    +            import os
    +            wx_home = os.getenv("WEIXIN_HOME_CHANNEL", "").strip()
    +            if wx_home:
    +                from gateway.config import HomeChannel
    +                home = HomeChannel(platform=platform, chat_id=wx_home, name="Weixin Home")
             if home:
                 chat_id = home.chat_id
                 used_home_channel = True

• #21196 fix(weixin): replace aiohttp ClientTimeout with asyncio.wait_for (salvage #19050)

  merged to mainSuggested link · files and diff not shown

• #80807 fix: dispatch cronjob(action='run') to the background like delegate_task

  merged to main

  files: cron/scheduler.py, tests/tools/test_cronjob_run_background.py, tools/cronjob_tools.py

• #80838 fix(cron): gateway-loop delivery, per-run prompt context, Matrix standalone send — run-action sweep salvage

  merged to main

  files: cron/scheduler.py, plugins/platforms/matrix/adapter.py, tests/cron/test_cron_kanban_env_isolation.py, tests/cron/test_cron_workdir.py, tests/cron/test_execution_ledger.py, tests/cron/test_parallel_pool.py, tests/cron/test_run_one_job.py, tests/cron/test_scheduler.py, tests/tools/test_cronjob_run_background.py, tests/tools/test_cronjob_run_immediate.py, tools/cronjob_tools.py