Debounce batching for WhatsApp and WeChat messages

Fixes on record

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

• #35391 fix(gateway): config-driven text-debounce batching for WhatsApp + Weixin

  merged to main

  files: gateway/platforms/weixin.py, gateway/platforms/whatsapp.py, tests/gateway/test_weixin.py, tests/gateway/test_whatsapp_text_batching.py, website/docs/user-guide/messaging/weixin.md, website/docs/user-guide/messaging/whatsapp.md

  Show the diff

  • gateway/platforms/weixin.py

    @@ -1180,12 +1180,48 @@ def __init__(self, config: PlatformConfig):
                 default=False,
             )
     
    +        # Text debounce batching (mirrors Telegram adapter pattern).
    +        # iLink delivers messages individually, so rapid multi-message
    +        # bursts (forwarded batches, paste-splits) each trigger a
    +        # separate agent invocation.  Default 3s delay / 5s split delay
    +        # are tuned for iLink's typical delivery cadence.  Tunable via
    +        # config.yaml under
    +        # ``gateway.platforms.weixin.extra.text_batch_delay_seconds`` /
    +        # ``text_batch_split_delay_seconds``.
    +        self._text_batch_delay_seconds = self._coerce_float_extra(
    +            "text_batch_delay_seconds", 3.0
    +        )
    +        self._text_batch_split_delay_seconds = self._coerce_float_extra(
    +            "text_batch_split_delay_seconds", 5.0
    +        )
    +        self._pending_text_batches: Dict[str, MessageEvent] = {}
    +        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
    +
             if self._account_id and not self._token:
                 persisted = load_weixin_account(hermes_home, self._account_id)
                 if persisted:
                     self._token = [redacted:secret-assignment]"token") or "").strip()
                     self._base_url = str(persisted.get("base_url") or self._base_url).strip().rstrip("/")
     
    +    def _coerce_float_extra(self, key: str, default: float) -> float:
    +        """Read a float from ``config.extra``, guarding against bad/non-finite values.
    +
    +        The result is fed directly to ``asyncio.sleep()``, so NaN/Inf and
    +        unparseable values fall back to ``default``.
    +        """
    +        import math
    +
    +        value = self.config.extra.get(key) if getattr(self.config, "extra", None) else None
    +        if value is None:
    +            return float(default)
    +        try:
    +            parsed = float(value)
    +        except (TypeError, ValueError):
    +            return float(default)
    +        if not math.isfinite(parsed) or parsed < 0:
    +            return float(default)
    +        return parsed
    +
         @staticmethod
         def _coerce_list(value: Any) -> List[str]:
             if value is None:
    @@ -1247,6 +1283,11 @@ async def connect(self) -> bool:
         async def disconnect(self) -> None:
             _LIVE_ADAPTERS.pop(self._token, None)
             self._running = False
    +        for task in self._pending_text_batch_tasks.values():
    +            if not task.done():
    +                task.cancel()
    +        self._pending_text_batches.clear()
    +        self._pending_text_batch_tasks.clear()
             if self._poll_task and not self._poll_task.done():
                 self._poll_task.cancel()
                 try:
    @@ -1395,12 +1436,10 @@ async def _process_message(self, message: Dict[str, Any]) -> None:
                 timestamp=datetime.now(),
             )
             logger.info("[%s] inbound from=%s type=%s media=%d", self.name, _safe_id(sender_id), source.chat_type, len(media_paths))
    -        await self.handle_message(event)
    -
    -    @property
    -    def enforces_own_access_policy(self) -> bool:
    -        """Weixin gates DM/group access at intake via dm_policy/group_policy."""
    -        return True
    +        if event.message_type == MessageType.TEXT:
    +            self._enqueue_text_event(event)
    +        else:
    +            await self.handle_message(event)
     
         def _is_dm_allowed(self, sender_id: str) -> bool:
             if self._dm_policy == "disabled":
    @@ -1409,6 +1448,76 @@ def _is_dm_allowed(self, sender_id: str) -> bool:
                 return sender_id in self._allow_from
             return True
     
    +    @property
    +    def enforces_own_access_policy(self) -> bool:
    +        """Weixin gates DM/group access at intake via dm_policy/group_policy."""
    +        return True
    +
    +    # ------------------------------------------------------------------
    +    # Text debounce batching
    +    # ------------------------------------------------------------------
    +
    +    _SPLIT_THRESHOLD = 1800  # iLink chunks at ~2048 chars
    +
    +    def _text_batch_key(self, event: MessageEvent) -> str:
    +        """Session-scoped key for text message batching."""
    +        from gateway.session import build_session_key
    +        return build_session_key(
    +            event.source,
    +            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
    +            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
    +        )
    +
    +    def _enqueue_text_event(self, event: MessageEvent) -> None:
    +        """Buffer a text event and reset the flush timer.
    +
    +        When users forward multiple messages or send rapid-fire texts
    +        via WeChat, each arrives as a separate iLink message. This
    +        concatenates them and waits for a short quiet period before
    +        dispatching the combined message.
    +        """
    +        key = self._text_batch_key(event)
    +        existing = self._pending_text_batches.get(key)
    +        chunk_len = len(event.text or "")
    +        if existing is None:
    +            event._last_chunk_len = chunk_len  # type: ignore[attr-defined]
    +            self._pending_text_batches[key] = event
    +        else:
    +            if event.text:
    +                existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
    +            existing._last_chunk_len = chunk_len  # type: ignore[attr-defined]
    +            if event.media_urls:
    +                existing.media_urls.extend(event.media_urls)
    +                existing.media_types.extend(event.media_types)
    +
    +        prior_task = self._pending_text_batch_tasks.get(key)
    +        if prior_task and not prior_task.done():
    +            prior_task.cancel()
    +        self._pending_text_batch_tasks[key] = asyncio.create_task(
    +            self._flush_text_batch(key)
    +        )
    +
    +    async def _flush_text_batch(self, key: str) -> None:
    +        """Wait for quiet period then dispatch aggregated text."""
    +        current_task = asyncio.current_task()
    +        try:
    +            pending = self._pending_text_batches.get(key)
    +            last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0
    +            if last_len >= self._SPLIT_THRESHOLD:
    +                delay = self._text_batch_split_delay_seconds
    +            else:
    +                delay = self._text_batch_delay_seconds
    +            await asyncio.sleep(delay)
    +            if self._pending_text_batch_tasks.get(key) is not current_task:
    +                return
    +            event = self._pending_text_batches.pop(key, None)
    +            if not event:
    +                return
    +            await self.handle_message(event)
    +        finally:
    +            if self._pending_text_batch_tasks.get(key) is current_task:
    +                self._pending_text_batch_tasks.pop(key, None)
    +
         async def _collect_media(self, item: Dict[str, Any], media_paths: List[str], media_types: List[str]) -> None:
             item_type = item.get("type")
             if item_type == ITEM_IMAGE:

  • gateway/platforms/whatsapp.py

    @@ -278,6 +278,43 @@ def __init__(self, config: PlatformConfig):
             # notification before the normal "✓ whatsapp disconnected" fires.
             self._shutting_down: bool = False
     
    +        # Text debounce batching (mirrors Telegram adapter pattern).
    +        # WhatsApp often delivers multiple messages in rapid succession
    +        # (e.g. forwarded batches, paste-splits) — without debounce each
    +        # message triggers a separate agent invocation, wasting tokens and
    +        # flooding the user with reply fragments.  Default 5s delay /
    +        # 10s split delay are conservative for WhatsApp's delivery cadence.
    +        # Tunable via config.yaml under
    +        # ``gateway.platforms.whatsapp.extra.text_batch_delay_seconds`` /
    +        # ``text_batch_split_delay_seconds``.
    +        self._text_batch_delay_seconds = self._coerce_float_extra(
    +            "text_batch_delay_seconds", 5.0
    +        )
    +        self._text_batch_split_delay_seconds = self._coerce_float_extra(
    +            "text_batch_split_delay_seconds", 10.0
    +        )
    +        self._pending_text_batches: Dict[str, MessageEvent] = {}
    +        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
    +
    +    def _coerce_float_extra(self, key: str, default: float) -> float:
    +        """Read a float from ``config.extra``, guarding against bad/non-finite values.
    +
    +        The result is fed directly to ``asyncio.sleep()``, so NaN/Inf and
    +        unparseable values fall back to ``default``.
    +        """
    +        import math
    +
    +        value = self.config.extra.get(key) if getattr(self.config, "extra", None) else None
    +        if value is None:
    +            return float(default)
    +        try:
    +            parsed = float(value)
    +        except (TypeError, ValueError):
    +            return float(default)
    +        if not math.isfinite(parsed) or parsed < 0:
    +            return float(default)
    +        return parsed
    +
         def _effective_reply_prefix(self) -> str:
             """Return the prefix the Node bridge will add in self-chat mode."""
             whatsapp_mode = os.getenv("WHATSAPP_MODE", "self-chat")
    @@ -1139,7 +1176,10 @@ async def _poll_messages(self) -> None:
                             for msg_data in messages:
                                 event = await self._build_message_event(msg_data)
                                 if event:
    -                                await self.handle_message(event)
    +                                if event.message_type == MessageType.TEXT:
    +                                    self._enqueue_text_event(event)
    +                                else:
    +                                    await self.handle_message(event)
                 except asyncio.CancelledError:
                     break
                 except Exception as e:
    @@ -1151,7 +1191,67 @@ async def _poll_messages(self) -> None:
                     await asyncio.sleep(5)
                 
                 await asyncio.sleep(1)  # Poll interval
    -    
    +
    +    # ── Text debounce batching ──────────────────────────────────────
    +
    +    _SPLIT_THRESHOLD = 6000  # WhatsApp supports ~65K chars; generous threshold
    +
    +    def _text_batch_key(self, event: MessageEvent) -> str:
    +        """Session-scoped key for text message batching."""
    +        from gateway.session import build_session_key
    +        return build_session_key(
    +            event.source,
    +            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
    +            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
    +        )
    +
    +    def _enqueue_text_event(self, event: MessageEvent) -> None:
    +        """Buffer a text event and reset the flush timer.
    +
    +        When WhatsApp delivers rapid-fire messages (e.g. forwarded
    +        batches), this concatenates them and waits for a short quiet
    +        period before dispatching the combined message.
    +        """
    +        key = self._text_batch_key(event)
    +        existing = self._pending_text_batches.get(key)
    +        chunk_len = len(event.text or "")
    +        if existing is None:
    +            event._last_chunk_len = chunk_len  # type: ignore[attr-defined]
    +            self._pending_text_batches[key] = event
    +        else:
    +            if event.text:
    +                existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
    +            existing._last_chunk_len = chunk_len  # type: ignore[attr-defined]
    +            if event.media_urls:
    +                existing.media_urls.extend(event.media_urls)
    +                existing.media_types.extend(event.media_types)
    +
    +        prior_task = self._pending_text_batch_tasks.get(key)
    +        if prior_task and not prior_task.done():
    +            prior_task.cancel()
    +        self._pending_text_batch_tasks[key] = asyncio.create_task(
    +            self._flush_text_batch(key)
    +        )
    +
    +    async def _flush_text_batch(self, key: str) -> None:
    +        """Wait for quiet period then dispatch aggregated text."""
    +        current_task = asyncio.current_task()
    +        try:
    +            pending = self._pending_text_batches.get(key)
    +            last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0
    +            if last_len >= self._SPLIT_THRESHOLD:
    +                delay = self._text_batch_split_delay_seconds
    +            else:
    +                delay = self._text_batch_delay_seconds
    +            await asyncio.sleep(delay)
    +            event = self._pending_text_batches.pop(key, None)
    +            if not event:
    +                return
    +            await self.handle_message(event)
    +        finally:
    +            if self._pending_text_batch_tasks.get(key) is current_task:
    +                self._pending_text_batch_tasks.pop(key, None)
    +
         async def _build_message_event(self, data: Dict[str, Any]) -> Optional[MessageEvent]:
             """Build a MessageEvent from bridge message data, downloading images to cache."""
             try:

  • tests/gateway/test_weixin.py

    @@ -11,6 +11,7 @@
     from gateway.config import PlatformConfig
     from gateway.config import GatewayConfig, HomeChannel, Platform, _apply_env_overrides
     from gateway.platforms.base import SendResult
    +from gateway.platforms.base import MessageEvent, MessageType
     from gateway.platforms import weixin
     from gateway.platforms.weixin import ContextTokenStore, WeixinAdapter
     from tools.send_message_tool import _parse_target_ref, _send_to_platform
    @@ -853,15 +854,27 @@ def test_duplicate_content_with_different_message_ids_is_dropped(self):
             adapter = _make_adapter()
             adapter._poll_session = object()
             adapter.handle_message = AsyncMock()
    +        # Tighten the text-debounce delay so the flush completes quickly.
    +        adapter._text_batch_delay_seconds = 0.05
    +        adapter._text_batch_split_delay_seconds = 0.05
     
             base_msg = {
                 "from_user_id": "wxid_user1",
                 "item_list": [{"type": 1, "text_item": {"text": "hello world"}}],
             }
     
    -        asyncio.run(adapter._process_message({**base_msg, "message_id": "msg-1"}))
    -        asyncio.run(adapter._process_message({**base_msg, "message_id": "msg-2"}))
    +        async def _drive():
    +            # Both inbound messages share the same event loop so the debounce
    +            # task created by the first one survives to be flushed.
    +            await adapter._process_message({**base_msg, "message_id": "msg-1"})
    +            await adapter._process_message({**base_msg, "message_id": "msg-2"})
    +            # Wait out the quiet period so the buffered text batch flushes.
    +            await asyncio.sleep(0.2)
     
    +        asyncio.run(_drive())
    +
    +        # Content-dedup drops the second (duplicate) message before it is even
    +        # enqueued, so only one combined dispatch reaches handle_message.
             assert adapter.handle_message.await_count == 1
             event = adapter.handle_message.await_args[0][0]
             assert event.text == "hello world"
    @@ -882,3 +895,76 @@ def test_content_dedup_not_called_for_messages_without_text(self):
             assert adapter.handle_message.await_count == 0
             # is_duplicate should only be called for message_id, never for content
             assert all("content:" not in str(call) for call in adapter._dedup.is_duplicate.call_args_list)
    +
    +
    +class TestWeixinTextDebounce:
    +    """Text-debounce batching for rapid multi-message bursts (issue #35301).
    +
    +    Delays are read from ``config.extra`` (config.yaml), not env vars.
    +    """
    +
    +    def test_batch_delays_default_from_config(self):
    +        adapter = _make_adapter()
    +        assert adapter._text_batch_delay_seconds == 3.0
    +        assert adapter._text_batch_split_delay_seconds == 5.0
    +
    +    def test_batch_delays_overridden_via_config_extra(self):
    +        adapter = WeixinAdapter(
    +            PlatformConfig(
    +                enabled=True,
    +                token="[redacted:secret-assignment]",
    +                extra={
    +                    "account_id": "test-account",
    +                    "text_batch_delay_seconds": "0.5",
    +                    "text_batch_split_delay_seconds": 1.5,
    +                },
    +            )
    +        )
    +        assert adapter._text_batch_delay_seconds == 0.5
    +        assert adapter._text_batch_split_delay_seconds == 1.5
    +
    +    def test_invalid_config_value_falls_back_to_default(self):
    +        adapter = WeixinAdapter(
    +            PlatformConfig(
    +                enabled=True,
    +                token="[redacted:secret-assignment]",
    +                extra={
    +                    "account_id": "test-account",
    +                    "text_batch_delay_seconds": "not-a-number",
    +                    "text_batch_split_delay_seconds": -4,
    +                },
    +            )
    +        )
    +        assert adapter._text_batch_delay_seconds == 3.0
    +        assert adapter._text_batch_split_delay_seconds == 5.0
    +
    +    def test_rapid_texts_collapse_into_single_dispatch(self):
    +        adapter = _make_adapter()
    +        adapter._text_batch_delay_seconds = 0.05
    +        adapter._text_batch_split_delay_seconds = 0.05
    +        dispatched = []
    +
    +        async def _capture(event):
    +            dispatched.append(event.text)
    +
    +        adapter.handle_message = _capture
    +
    +        def _event(text):
    +            return MessageEvent(
    +                text=text,
    +                message_type=MessageType.TEXT,
    +                source=adapter.build_source(
    +                    chat_id="wxid_user1", chat_type="dm",
    +                    user_id="wxid_user1", user_name="wxid_user1",
    +                ),
    +            )
    +
    +        async def _drive():
    +            adapter._enqueue_text_event(_event("one"))
    +            adapter._enqueue_text_event(_event("two"))
    +            adapter._enqueue_text_event(_event("three"))
    +            assert dispatched == []  # nothing flushed during the burst
    +            await asyncio.sleep(0.2)
    +
    +        asyncio.run(_drive())
    +        assert dispatched == ["one\ntwo\nthree"]

  • tests/gateway/test_whatsapp_text_batching.py

    @@ -0,0 +1,107 @@
    +"""Text-debounce batching for the WhatsApp adapter (issue #35301).
    +
    +WhatsApp delivers rapid multi-message bursts (forwarded batches, paste-splits)
    +individually.  Without debounce each fragment triggers a separate agent
    +invocation, wasting tokens and flooding the user with reply fragments.  This
    +mirrors the Telegram/WeCom/Feishu pattern.
    +
    +Batch delays are read from ``config.extra`` (config.yaml), not env vars.
    +"""
    +
    +import asyncio
    +
    +from gateway.config import Platform, PlatformConfig
    +from gateway.platforms.base import MessageEvent, MessageType
    +from gateway.platforms.whatsapp import WhatsAppAdapter
    +from gateway.session import SessionSource
    +
    +
    +def _make_adapter(**extra):
    +    base = {"session_name": "test"}
    +    base.update(extra)
    +    return WhatsAppAdapter(PlatformConfig(enabled=True, extra=base))
    +
    +
    +def _event(text):
    +    src = SessionSource(
    +        platform=Platform.WHATSAPP,
    +        chat_id="chat123",
    +        chat_type="dm",
    +        user_id="user1",
    +        user_name="tester",
    +    )
    +    return MessageEvent(text=text, message_type=MessageType.TEXT, source=src)
    +
    +
    +def test_batch_delays_default_from_config():
    +    adapter = _make_adapter()
    +    assert adapter._text_batch_delay_seconds == 5.0
    +    assert adapter._text_batch_split_delay_seconds == 10.0
    +
    +
    +def test_batch_delays_overridden_via_config_extra():
    +    adapter = _make_adapter(
    +        text_batch_delay_seconds="2.5",
    +        text_batch_split_delay_seconds=7,
    +    )
    +    assert adapter._text_batch_delay_seconds == 2.5
    +    assert adapter._text_batch_split_delay_seconds == 7.0
    +
    +
    +def test_invalid_config_value_falls_back_to_default():
    +    adapter = _make_adapter(
    +        text_batch_delay_seconds="garbage",
    +        text_batch_split_delay_seconds=-3,
    +    )
    +    assert adapter._text_batch_delay_seconds == 5.0
    +    assert adapter._text_batch_split_delay_seconds == 10.0
    +
    +
    +def test_env_var_is_ignored(monkeypatch):
    +    # Config-only path: the legacy HERMES_* env var must NOT influence delays.
    +    monkeypatch.setenv("HERMES_WHATSAPP_TEXT_BATCH_DELAY_SECONDS", "99")
    +    adapter = _make_adapter()
    +    assert adapter._text_batch_delay_seconds == 5.0
    +
    +
    +def test_rapid_texts_collapse_into_single_dispatch():
    +    adapter = _make_adapter(
    +        text_batch_delay_seconds=0.05,
    +        text_batch_split_delay_seconds=0.05,
    +    )
    +    dispatched = []
    +
    +    async def _capture(event):
    +        dispatched.append(event.text)
    +
    +    adapter.handle_message = _capture
    +
    +    async def _drive():
    +        adapter._enqueue_text_event(_event("one"))
    +        adapter._enqueue_text_event(_event("two"))
    +        adapter._enqueue_text_event(_event("three"))
    +        assert dispatched == []  # nothing flushed during the burst
    +        await asyncio.sleep(0.2)
    +
    +    asyncio.run(_drive())
    +    assert dispatched == ["one\ntwo\nthree"]
    +
    +
    +def test_lone_message_dispatched_alone():
    +    adapter = _make_adapter(
    +        text_batch_delay_seconds=0.05,
    +        text_batch_split_delay_seconds=0.05,
    +    )
    +    dispatched = []
    +
    +    async def _capture(event):
    +        dispatched.append(event.text)
    +
    +    adapter.handle_message = _capture
    +
    +    async def _drive():
    +        adapter._enqueue_text_event(_event("solo"))
    +        await asyncio.sleep(0.2)
    +
    +    asyncio.run(_drive())
    +    assert dispatched == ["solo"]

  • website/docs/user-guide/messaging/weixin.md

    @@ -123,6 +123,8 @@ Set these in `config.yaml` under `platforms.weixin.extra`:
     | `allow_from` | `[]` | User IDs allowed for DMs (when dm_policy=allowlist) |
     | `group_allow_from` | `[]` | Group IDs allowed (when group_policy=allowlist) |
     | `split_multiline_messages` | `false` | When `true`, split multi-line replies into multiple chat messages (legacy behavior). When `false`, keep multi-line replies as one message unless they exceed the length limit. |
    +| `text_batch_delay_seconds` | `3.0` | Quiet period (seconds) before a buffered burst of rapid text messages is flushed as one combined request. iLink delivers messages individually, so this debounce avoids one agent invocation per fragment. Set `0` to dispatch each message immediately. |
    +| `text_batch_split_delay_seconds` | `5.0` | Extended flush delay used when the latest fragment is near the split threshold (long messages iLink may have chunked). |
     
     ## Access Policies
     

  • website/docs/user-guide/messaging/whatsapp.md

    @@ -201,6 +201,22 @@ Code blocks and inline code are preserved as-is since WhatsApp supports triple-b
     
     When the agent calls tools (web search, file operations, etc.), WhatsApp displays real-time progress indicators showing which tool is running. This is enabled by default — no configuration needed.
     
    +### Message Batching (Debounce)
    +
    +WhatsApp delivers each message individually, so a rapid burst (forwarded batches, paste-splits, multi-line text) would otherwise trigger a separate agent invocation per fragment — wasting tokens and producing several disjointed replies. The adapter buffers successive text messages from the same chat and dispatches them as one combined request after a short quiet period (default **5s**, extended to **10s** for very long fragments). Tune via `config.yaml`:
    +
    +```yaml
    +# ~/.hermes/config.yaml
    +gateway:
    +  platforms:
    +    whatsapp:
    +      extra:
    +        text_batch_delay_seconds: 5.0         # quiet period before flushing a batch
    +        text_batch_split_delay_seconds: 10.0  # extended delay near the split threshold
    +```
    +
    +Set `text_batch_delay_seconds: 0` to dispatch each message immediately (disables batching).
    +
     ---
     
     ## Troubleshooting