Weixin messages from cron/send_message dropped while gateway runs

Fixes on record

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

• #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=self._token,
                         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=self._token,
                     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=self._token,
                         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=self._token,
                     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=self._token,
                     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=self._token,
                 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=self._token,
                     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": [redacted:secret-assignment]
    +        }
    +
         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 = os.getenv("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