Inbound text batching fixes across gateway adapters

Fixes on record

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

• #6979 fix(gateway): add inbound text batching to Discord, Matrix, WeCom + adaptive delay for all

  merged to main

  files: gateway/platforms/discord.py, gateway/platforms/feishu.py, gateway/platforms/matrix.py, gateway/platforms/telegram.py, gateway/platforms/wecom.py, tests/gateway/test_discord_channel_controls.py, tests/gateway/test_discord_free_response.py, tests/gateway/test_discord_slash_commands.py, tests/gateway/test_matrix_mention.py, tests/gateway/test_text_batching.py, tests/gateway/test_wecom.py

  Show the diff

  • gateway/platforms/discord.py

    @@ -422,6 +422,7 @@ class DiscordAdapter(BasePlatformAdapter):
     
         # Discord message limits
         MAX_MESSAGE_LENGTH = 2000
    +    _SPLIT_THRESHOLD = 1900  # near the 2000-char split point
     
         # Auto-disconnect from voice channel after this many seconds of inactivity
         VOICE_TIMEOUT = 300
    @@ -433,6 +434,11 @@ def __init__(self, config: PlatformConfig):
             self._allowed_user_ids: set = set()  # For button approval authorization
             # Voice channel state (per-guild)
             self._voice_clients: Dict[int, Any] = {}  # guild_id -> VoiceClient
    +        # Text batching: merge rapid successive messages (Telegram-style)
    +        self._text_batch_delay_seconds = float(os.getenv("HERMES_DISCORD_TEXT_BATCH_DELAY_SECONDS", "0.6"))
    +        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_DISCORD_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0"))
    +        self._pending_text_batches: Dict[str, MessageEvent] = {}
    +        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
             self._voice_text_channels: Dict[int, int] = {}  # guild_id -> text_channel_id
             self._voice_timeout_tasks: Dict[int, asyncio.Task] = {}  # guild_id -> timeout task
             # Phase 2: voice listening
    @@ -2466,7 +2472,80 @@ async def _handle_message(self, message: DiscordMessage) -> None:
             if thread_id:
                 self._track_thread(thread_id)
     
    -        await self.handle_message(event)
    +        # Only batch plain text messages — commands, media, etc. dispatch
    +        # immediately since they won't be split by the Discord client.
    +        if msg_type == MessageType.TEXT:
    +            self._enqueue_text_event(event)
    +        else:
    +            await self.handle_message(event)
    +
    +    # ------------------------------------------------------------------
    +    # Text message aggregation (handles Discord client-side splits)
    +    # ------------------------------------------------------------------
    +
    +    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 Discord splits a long user message at 2000 chars, the chunks
    +        arrive within a few hundred milliseconds.  This merges them into
    +        a single event before dispatching.
    +        """
    +        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 the quiet period then dispatch the aggregated text.
    +
    +        Uses a longer delay when the latest chunk is near Discord's 2000-char
    +        split point, since a continuation chunk is almost certain.
    +        """
    +        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
    +            logger.info(
    +                "[Discord] Flushing text batch %s (%d chars)",
    +                key, len(event.text or ""),
    +            )
    +            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)
     
     
     # ---------------------------------------------------------------------------

  • gateway/platforms/feishu.py

    @@ -264,6 +264,7 @@ class FeishuAdapterSettings:
         bot_name: str
         dedup_cache_size: int
         text_batch_delay_seconds: float
    +    text_batch_split_delay_seconds: float
         text_batch_max_messages: int
         text_batch_max_chars: int
         media_batch_delay_seconds: float
    @@ -1014,6 +1015,10 @@ class FeishuAdapter(BasePlatformAdapter):
         """Feishu/Lark bot adapter."""
     
         MAX_MESSAGE_LENGTH = 8000
    +    # Threshold for detecting Feishu client-side message splits.
    +    # When a chunk is near the ~4096-char practical limit, a continuation
    +    # is almost certain.
    +    _SPLIT_THRESHOLD = 4000
     
         # =========================================================================
         # Lifecycle — init / settings / connect / disconnect
    @@ -1105,6 +1110,9 @@ def _load_settings(extra: Dict[str, Any]) -> FeishuAdapterSettings:
                 text_batch_delay_seconds=float(
                     os.getenv("HERMES_FEISHU_TEXT_BATCH_DELAY_SECONDS", str(_DEFAULT_TEXT_BATCH_DELAY_SECONDS))
                 ),
    +            text_batch_split_delay_seconds=float(
    +                os.getenv("HERMES_FEISHU_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0")
    +            ),
                 text_batch_max_messages=max(
                     1,
                     int(os.getenv("HERMES_FEISHU_TEXT_BATCH_MAX_MESSAGES", str(_DEFAULT_TEXT_BATCH_MAX_MESSAGES))),
    @@ -1152,6 +1160,7 @@ def _apply_settings(self, settings: FeishuAdapterSettings) -> None:
             self._bot_name = settings.bot_name
             self._dedup_cache_size = settings.dedup_cache_size
             self._text_batch_delay_seconds = settings.text_batch_delay_seconds
    +        self._text_batch_split_delay_seconds = settings.text_batch_split_delay_seconds
             self._text_batch_max_messages = settings.text_batch_max_messages
             self._text_batch_max_chars = settings.text_batch_max_chars
             self._media_batch_delay_seconds = settings.media_batch_delay_seconds
    @@ -2478,8 +2487,10 @@ def _text_batch_is_compatible(existing: MessageEvent, incoming: MessageEvent) ->
         async def _enqueue_text_event(self, event: MessageEvent) -> None:
             """Debounce rapid Feishu text bursts into a single MessageEvent."""
             key = self._text_batch_key(event)
    +        chunk_len = len(event.text or "")
             existing = self._pending_text_batches.get(key)
             if existing is None:
    +            event._last_chunk_len = chunk_len  # type: ignore[attr-defined]
                 self._pending_text_batches[key] = event
                 self._pending_text_batch_counts[key] = 1
                 self._schedule_text_batch_flush(key)
    @@ -2504,6 +2515,7 @@ async def _enqueue_text_event(self, event: MessageEvent) -> None:
                 return
     
             existing.text = next_text
    +        existing._last_chunk_len = chunk_len  # type: ignore[attr-defined]
             existing.timestamp = event.timestamp
             if event.message_id:
                 existing.message_id = event.message_id
    @@ -2530,10 +2542,22 @@ def _reschedule_batch_task(
             task_map[key] = asyncio.create_task(flush_fn(key))
     
         async def _flush_text_batch(self, key: str) -> None:
    -        """Flush a pending text batch after the quiet period."""
    +        """Flush a pending text batch after the quiet period.
    +
    +        Uses a longer delay when the latest chunk is near Feishu's ~4096-char
    +        split point, since a continuation chunk is almost certain.
    +        """
             current_task = asyncio.current_task()
             try:
    -            await asyncio.sleep(self._text_batch_delay_seconds)
    +            # Adaptive delay: if the latest chunk is near the split threshold,
    +            # a continuation is almost certain — wait longer.
    +            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)
                 await self._flush_text_batch_now(key)
             finally:
                 if self._pending_text_batch_tasks.get(key) is current_task:

  • gateway/platforms/matrix.py

    @@ -120,6 +120,11 @@ def check_matrix_requirements() -> bool:
     class MatrixAdapter(BasePlatformAdapter):
         """Gateway adapter for Matrix (any homeserver)."""
     
    +    # Threshold for detecting Matrix client-side message splits.
    +    # When a chunk is near the ~4000-char practical limit, a continuation
    +    # is almost certain.
    +    _SPLIT_THRESHOLD = 3900
    +
         def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.MATRIX)
     
    @@ -172,6 +177,13 @@ def __init__(self, config: PlatformConfig):
                 "MATRIX_REACTIONS", "true"
             ).lower() not in ("false", "0", "no")
     
    +        # Text batching: merge rapid successive messages (Telegram-style).
    +        # Matrix clients split long messages around 4000 chars.
    +        self._text_batch_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS", "0.6"))
    +        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0"))
    +        self._pending_text_batches: Dict[str, MessageEvent] = {}
    +        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
    +
         def _is_duplicate_event(self, event_id) -> bool:
             """Return True if this event was already processed. Tracks the ID otherwise."""
             if not event_id:
    @@ -1088,7 +1100,81 @@ async def _on_room_message(self, room: Any, event: Any) -> None:
             # Acknowledge receipt so the room shows as read (fire-and-forget).
             self._background_read_receipt(room.room_id, event.event_id)
     
    -        await self.handle_message(msg_event)
    +        # Only batch plain text messages — commands dispatch immediately.
    +        if msg_type == MessageType.TEXT:
    +            self._enqueue_text_event(msg_event)
    +        else:
    +            await self.handle_message(msg_event)
    +
    +    # ------------------------------------------------------------------
    +    # Text message aggregation (handles Matrix client-side splits)
    +    # ------------------------------------------------------------------
    +
    +    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 a Matrix client splits a long message, the chunks arrive within
    +        a few hundred milliseconds.  This merges them into a single event
    +        before dispatching.
    +        """
    +        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]
    +            # Merge any media that might be attached
    +            if event.media_urls:
    +                existing.media_urls.extend(event.media_urls)
    +                existing.media_types.extend(event.media_types)
    +
    +        # Cancel any pending flush and restart the timer
    +        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 the quiet period then dispatch the aggregated text.
    +
    +        Uses a longer delay when the latest chunk is near Matrix's ~4000-char
    +        split point, since a continuation chunk is almost certain.
    +        """
    +        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
    +            logger.info(
    +                "[Matrix] Flushing text batch %s (%d chars)",
    +                key, len(event.text or ""),
    +            )
    +            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 _on_room_message_media(self, room: Any, event: Any) -> None:
             """Handle incoming media messages (images, audio, video, files)."""

  • gateway/platforms/telegram.py

    @@ -121,6 +121,9 @@ class TelegramAdapter(BasePlatformAdapter):
         
         # Telegram message limits
         MAX_MESSAGE_LENGTH = 4096
    +    # Threshold for detecting Telegram client-side message splits.
    +    # When a chunk is near this limit, a continuation is almost certain.
    +    _SPLIT_THRESHOLD = 4000
         MEDIA_GROUP_WAIT_SECONDS = 0.8
         
         def __init__(self, config: PlatformConfig):
    @@ -140,6 +143,7 @@ def __init__(self, config: PlatformConfig):
             # Buffer rapid text messages so Telegram client-side splits of long
             # messages are aggregated into a single MessageEvent.
             self._text_batch_delay_seconds = float(os.getenv("HERMES_TELEGRAM_TEXT_BATCH_DELAY_SECONDS", "0.6"))
    +        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_TELEGRAM_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0"))
             self._pending_text_batches: Dict[str, MessageEvent] = {}
             self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
             self._token_lock_identity: [redacted:secret-assignment] = None
    @@ -2160,12 +2164,15 @@ def _enqueue_text_event(self, event: MessageEvent) -> None:
             """
             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:
                 # Append text from the follow-up chunk
                 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]
                 # Merge any media that might be attached
                 if event.media_urls:
                     existing.media_urls.extend(event.media_urls)
    @@ -2180,10 +2187,22 @@ def _enqueue_text_event(self, event: MessageEvent) -> None:
             )
     
         async def _flush_text_batch(self, key: str) -> None:
    -        """Wait for the quiet period then dispatch the aggregated text."""
    +        """Wait for the quiet period then dispatch the aggregated text.
    +
    +        Uses a longer delay when the latest chunk is near Telegram's 4096-char
    +        split point, since a continuation chunk is almost certain.
    +        """
             current_task = asyncio.current_task()
             try:
    -            await asyncio.sleep(self._text_batch_delay_seconds)
    +            # Adaptive delay: if the latest chunk is near Telegram's 4096-char
    +            # split point, a continuation is almost certain — wait longer.
    +            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

  • gateway/platforms/wecom.py

    @@ -143,6 +143,9 @@ class WeComAdapter(BasePlatformAdapter):
         """WeCom AI Bot adapter backed by a persistent WebSocket connection."""
     
         MAX_MESSAGE_LENGTH = MAX_MESSAGE_LENGTH
    +    # Threshold for detecting WeCom client-side message splits.
    +    # When a chunk is near the 4000-char limit, a continuation is almost certain.
    +    _SPLIT_THRESHOLD = 3900
     
         def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.WECOM)
    @@ -172,6 +175,13 @@ def __init__(self, config: PlatformConfig):
             self._seen_messages: Dict[str, float] = {}
             self._reply_req_ids: Dict[str, str] = {}
     
    +        # Text batching: merge rapid successive messages (Telegram-style).
    +        # WeCom clients split long messages around 4000 chars.
    +        self._text_batch_delay_seconds = float(os.getenv("HERMES_WECOM_TEXT_BATCH_DELAY_SECONDS", "0.6"))
    +        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_WECOM_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0"))
    +        self._pending_text_batches: Dict[str, MessageEvent] = {}
    +        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
    +
         # ------------------------------------------------------------------
         # Connection lifecycle
         # ------------------------------------------------------------------
    @@ -519,7 +529,82 @@ async def _on_message(self, payload: Dict[str, Any]) -> None:
                 timestamp=datetime.now(tz=timezone.utc),
             )
     
    -        await self.handle_message(event)
    +        # Only batch plain text messages — commands, media, etc. dispatch
    +        # immediately since they won't be split by the WeCom client.
    +        if message_type == MessageType.TEXT:
    +            self._enqueue_text_event(event)
    +        else:
    +            await self.handle_message(event)
    +
    +    # ------------------------------------------------------------------
    +    # Text message aggregation (handles WeCom client-side splits)
    +    # ------------------------------------------------------------------
    +
    +    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 WeCom splits a long user message at 4000 chars, the chunks
    +        arrive within a few hundred milliseconds.  This merges them into
    +        a single event before dispatching.
    +        """
    +        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]
    +            # Merge any media that might be attached
    +            if event.media_urls:
    +                existing.media_urls.extend(event.media_urls)
    +                existing.media_types.extend(event.media_types)
    +
    +        # Cancel any pending flush and restart the timer
    +        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 the quiet period then dispatch the aggregated text.
    +
    +        Uses a longer delay when the latest chunk is near WeCom's 4000-char
    +        split point, since a continuation chunk is almost certain.
    +        """
    +        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
    +            logger.info(
    +                "[WeCom] Flushing text batch %s (%d chars)",
    +                key, len(event.text or ""),
    +            )
    +            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)
     
         @staticmethod
         def _extract_text(body: Dict[str, Any]) -> Tuple[str, Optional[str]]:

  • tests/gateway/test_discord_channel_controls.py

    @@ -81,6 +81,7 @@ def adapter(monkeypatch):
         config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
         adapter = DiscordAdapter(config)
         adapter._client = SimpleNamespace(user=SimpleNamespace(id=999))
    +    adapter._text_batch_delay_seconds = 0  # disable batching for tests
         adapter.handle_message = AsyncMock()
         return adapter
     

  • tests/gateway/test_discord_free_response.py

    @@ -91,6 +91,7 @@ def adapter(monkeypatch):
         config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
         adapter = DiscordAdapter(config)
         adapter._client = SimpleNamespace(user=SimpleNamespace(id=999))
    +    adapter._text_batch_delay_seconds = 0  # disable batching for tests
         adapter.handle_message = AsyncMock()
         return adapter
     

  • tests/gateway/test_discord_slash_commands.py

    @@ -62,6 +62,7 @@ def adapter():
             fetch_channel=AsyncMock(),
             user=SimpleNamespace(id=99999, name="HermesBot"),
         )
    +    adapter._text_batch_delay_seconds = 0  # disable batching for tests
         return adapter
     
     

  • tests/gateway/test_matrix_mention.py

    @@ -44,6 +44,7 @@ def _make_adapter(tmp_path=None):
             },
         )
         adapter = MatrixAdapter(config)
    +    adapter._text_batch_delay_seconds = 0  # disable batching for tests
         adapter.handle_message = AsyncMock()
         adapter._startup_ts = time.time() - 10  # avoid startup grace filter
         return adapter

  • tests/gateway/test_text_batching.py

    @@ -0,0 +1,448 @@
    +"""Tests for text message batching across all gateway adapters.
    +
    +When a user sends a long message, the messaging client splits it at the
    +platform's character limit.  Each adapter should buffer rapid successive
    +text messages from the same session and aggregate them before dispatching.
    +
    +Covers: Discord, Matrix, WeCom, and the adaptive delay logic for
    +Telegram and Feishu.
    +"""
    +
    +import asyncio
    +import os
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +from gateway.config import Platform, PlatformConfig
    +from gateway.platforms.base import MessageEvent, MessageType, SessionSource
    +
    +
    +# =====================================================================
    +# Helpers
    +# =====================================================================
    +
    +def _make_event(
    +    text: str,
    +    platform: Platform,
    +    chat_id: str = "12345",
    +    msg_type: MessageType = MessageType.TEXT,
    +) -> MessageEvent:
    +    return MessageEvent(
    +        text=text,
    +        message_type=msg_type,
    +        source=SessionSource(platform=platform, chat_id=chat_id, chat_type="dm"),
    +    )
    +
    +
    +# =====================================================================
    +# Discord text batching
    +# =====================================================================
    +
    +def _make_discord_adapter():
    +    """Create a minimal DiscordAdapter for testing text batching."""
    +    from gateway.platforms.discord import DiscordAdapter
    +
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = object.__new__(DiscordAdapter)
    +    adapter._platform = Platform.DISCORD
    +    adapter.config = config
    +    adapter._pending_text_batches = {}
    +    adapter._pending_text_batch_tasks = {}
    +    adapter._text_batch_delay_seconds = 0.1  # fast for tests
    +    adapter._text_batch_split_delay_seconds = 0.3  # fast for tests
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._message_handler = AsyncMock()
    +    adapter.handle_message = AsyncMock()
    +    return adapter
    +
    +
    +class TestDiscordTextBatching:
    +    @pytest.mark.asyncio
    +    async def test_single_message_dispatched_after_delay(self):
    +        adapter = _make_discord_adapter()
    +        event = _make_event("hello world", Platform.DISCORD)
    +
    +        adapter._enqueue_text_event(event)
    +
    +        # Not dispatched yet
    +        adapter.handle_message.assert_not_called()
    +
    +        # Wait for flush
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        dispatched = adapter.handle_message.call_args[0][0]
    +        assert dispatched.text == "hello world"
    +
    +    @pytest.mark.asyncio
    +    async def test_split_messages_aggregated(self):
    +        """Two rapid messages from the same chat should be merged."""
    +        adapter = _make_discord_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("Part one of a long", Platform.DISCORD))
    +        await asyncio.sleep(0.02)
    +        adapter._enqueue_text_event(_make_event("message that was split.", Platform.DISCORD))
    +
    +        adapter.handle_message.assert_not_called()
    +
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        text = adapter.handle_message.call_args[0][0].text
    +        assert "Part one" in text
    +        assert "split" in text
    +
    +    @pytest.mark.asyncio
    +    async def test_three_way_split_aggregated(self):
    +        adapter = _make_discord_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("chunk 1", Platform.DISCORD))
    +        await asyncio.sleep(0.02)
    +        adapter._enqueue_text_event(_make_event("chunk 2", Platform.DISCORD))
    +        await asyncio.sleep(0.02)
    +        adapter._enqueue_text_event(_make_event("chunk 3", Platform.DISCORD))
    +
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        text = adapter.handle_message.call_args[0][0].text
    +        assert "chunk 1" in text
    +        assert "chunk 2" in text
    +        assert "chunk 3" in text
    +
    +    @pytest.mark.asyncio
    +    async def test_different_chats_not_merged(self):
    +        adapter = _make_discord_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("from A", Platform.DISCORD, chat_id="111"))
    +        adapter._enqueue_text_event(_make_event("from B", Platform.DISCORD, chat_id="222"))
    +
    +        await asyncio.sleep(0.2)
    +
    +        assert adapter.handle_message.call_count == 2
    +
    +    @pytest.mark.asyncio
    +    async def test_batch_cleans_up_after_flush(self):
    +        adapter = _make_discord_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("test", Platform.DISCORD))
    +        await asyncio.sleep(0.2)
    +
    +        assert len(adapter._pending_text_batches) == 0
    +
    +    @pytest.mark.asyncio
    +    async def test_adaptive_delay_for_near_limit_chunk(self):
    +        """Chunks near the 2000-char limit should trigger longer delay."""
    +        adapter = _make_discord_adapter()
    +        # Simulate a chunk near Discord's 2000-char split point
    +        long_text = "x" * 1950
    +        adapter._enqueue_text_event(_make_event(long_text, Platform.DISCORD))
    +
    +        # After the short delay (0.1s), should NOT have flushed yet (split delay is 0.3s)
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_not_called()
    +
    +        # After the split delay, should be flushed
    +        await asyncio.sleep(0.25)
    +        adapter.handle_message.assert_called_once()
    +
    +
    +# =====================================================================
    +# Matrix text batching
    +# =====================================================================
    +
    +def _make_matrix_adapter():
    +    """Create a minimal MatrixAdapter for testing text batching."""
    +    from gateway.platforms.matrix import MatrixAdapter
    +
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = object.__new__(MatrixAdapter)
    +    adapter._platform = Platform.MATRIX
    +    adapter.config = config
    +    adapter._pending_text_batches = {}
    +    adapter._pending_text_batch_tasks = {}
    +    adapter._text_batch_delay_seconds = 0.1
    +    adapter._text_batch_split_delay_seconds = 0.3
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._message_handler = AsyncMock()
    +    adapter.handle_message = AsyncMock()
    +    return adapter
    +
    +
    +class TestMatrixTextBatching:
    +    @pytest.mark.asyncio
    +    async def test_single_message_dispatched_after_delay(self):
    +        adapter = _make_matrix_adapter()
    +        event = _make_event("hello world", Platform.MATRIX)
    +
    +        adapter._enqueue_text_event(event)
    +
    +        adapter.handle_message.assert_not_called()
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        assert adapter.handle_message.call_args[0][0].text == "hello world"
    +
    +    @pytest.mark.asyncio
    +    async def test_split_messages_aggregated(self):
    +        adapter = _make_matrix_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("first part", Platform.MATRIX))
    +        await asyncio.sleep(0.02)
    +        adapter._enqueue_text_event(_make_event("second part", Platform.MATRIX))
    +
    +        adapter.handle_message.assert_not_called()
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        text = adapter.handle_message.call_args[0][0].text
    +        assert "first part" in text
    +        assert "second part" in text
    +
    +    @pytest.mark.asyncio
    +    async def test_different_rooms_not_merged(self):
    +        adapter = _make_matrix_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("room A", Platform.MATRIX, chat_id="!aaa:matrix.org"))
    +        adapter._enqueue_text_event(_make_event("room B", Platform.MATRIX, chat_id="!bbb:matrix.org"))
    +
    +        await asyncio.sleep(0.2)
    +
    +        assert adapter.handle_message.call_count == 2
    +
    +    @pytest.mark.asyncio
    +    async def test_adaptive_delay_for_near_limit_chunk(self):
    +        """Chunks near the 4000-char limit should trigger longer delay."""
    +        adapter = _make_matrix_adapter()
    +        long_text = "x" * 3950
    +        adapter._enqueue_text_event(_make_event(long_text, Platform.MATRIX))
    +
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_not_called()
    +
    +        await asyncio.sleep(0.25)
    +        adapter.handle_message.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_batch_cleans_up_after_flush(self):
    +        adapter = _make_matrix_adapter()
    +        adapter._enqueue_text_event(_make_event("test", Platform.MATRIX))
    +        await asyncio.sleep(0.2)
    +        assert len(adapter._pending_text_batches) == 0
    +
    +
    +# =====================================================================
    +# WeCom text batching
    +# =====================================================================
    +
    +def _make_wecom_adapter():
    +    """Create a minimal WeComAdapter for testing text batching."""
    +    from gateway.platforms.wecom import WeComAdapter
    +
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = object.__new__(WeComAdapter)
    +    adapter._platform = Platform.WECOM
    +    adapter.config = config
    +    adapter._pending_text_batches = {}
    +    adapter._pending_text_batch_tasks = {}
    +    adapter._text_batch_delay_seconds = 0.1
    +    adapter._text_batch_split_delay_seconds = 0.3
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._message_handler = AsyncMock()
    +    adapter.handle_message = AsyncMock()
    +    return adapter
    +
    +
    +class TestWeComTextBatching:
    +    @pytest.mark.asyncio
    +    async def test_single_message_dispatched_after_delay(self):
    +        adapter = _make_wecom_adapter()
    +        event = _make_event("hello world", Platform.WECOM)
    +
    +        adapter._enqueue_text_event(event)
    +
    +        adapter.handle_message.assert_not_called()
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        assert adapter.handle_message.call_args[0][0].text == "hello world"
    +
    +    @pytest.mark.asyncio
    +    async def test_split_messages_aggregated(self):
    +        adapter = _make_wecom_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("first part", Platform.WECOM))
    +        await asyncio.sleep(0.02)
    +        adapter._enqueue_text_event(_make_event("second part", Platform.WECOM))
    +
    +        adapter.handle_message.assert_not_called()
    +        await asyncio.sleep(0.2)
    +
    +        adapter.handle_message.assert_called_once()
    +        text = adapter.handle_message.call_args[0][0].text
    +        assert "first part" in text
    +        assert "second part" in text
    +
    +    @pytest.mark.asyncio
    +    async def test_different_chats_not_merged(self):
    +        adapter = _make_wecom_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("chat A", Platform.WECOM, chat_id="chat_a"))
    +        adapter._enqueue_text_event(_make_event("chat B", Platform.WECOM, chat_id="chat_b"))
    +
    +        await asyncio.sleep(0.2)
    +
    +        assert adapter.handle_message.call_count == 2
    +
    +    @pytest.mark.asyncio
    +    async def test_adaptive_delay_for_near_limit_chunk(self):
    +        """Chunks near the 4000-char limit should trigger longer delay."""
    +        adapter = _make_wecom_adapter()
    +        long_text = "x" * 3950
    +        adapter._enqueue_text_event(_make_event(long_text, Platform.WECOM))
    +
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_not_called()
    +
    +        await asyncio.sleep(0.25)
    +        adapter.handle_message.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_batch_cleans_up_after_flush(self):
    +        adapter = _make_wecom_adapter()
    +        adapter._enqueue_text_event(_make_event("test", Platform.WECOM))
    +        await asyncio.sleep(0.2)
    +        assert len(adapter._pending_text_batches) == 0
    +
    +
    +# =====================================================================
    +# Telegram adaptive delay (PR #6891)
    +# =====================================================================
    +
    +def _make_telegram_adapter():
    +    """Create a minimal TelegramAdapter for testing adaptive delay."""
    +    from gateway.platforms.telegram import TelegramAdapter
    +
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = object.__new__(TelegramAdapter)
    +    adapter._platform = Platform.TELEGRAM
    +    adapter.config = config
    +    adapter._pending_text_batches = {}
    +    adapter._pending_text_batch_tasks = {}
    +    adapter._text_batch_delay_seconds = 0.1
    +    adapter._text_batch_split_delay_seconds = 0.3
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._message_handler = AsyncMock()
    +    adapter.handle_message = AsyncMock()
    +    return adapter
    +
    +
    +class TestTelegramAdaptiveDelay:
    +    @pytest.mark.asyncio
    +    async def test_short_chunk_uses_normal_delay(self):
    +        adapter = _make_telegram_adapter()
    +        adapter._enqueue_text_event(_make_event("short msg", Platform.TELEGRAM))
    +
    +        # Should flush after the normal 0.1s delay
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_near_limit_chunk_uses_split_delay(self):
    +        """A chunk near the 4096-char limit should trigger longer delay."""
    +        adapter = _make_telegram_adapter()
    +        long_text = "x" * 4050  # near the 4096 limit
    +        adapter._enqueue_text_event(_make_event(long_text, Platform.TELEGRAM))
    +
    +        # After the short delay, should NOT have flushed yet
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_not_called()
    +
    +        # After the split delay, should be flushed
    +        await asyncio.sleep(0.25)
    +        adapter.handle_message.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_split_continuation_merged(self):
    +        """Two near-limit chunks should both be merged."""
    +        adapter = _make_telegram_adapter()
    +
    +        adapter._enqueue_text_event(_make_event("x" * 4050, Platform.TELEGRAM))
    +        await asyncio.sleep(0.05)
    +        adapter._enqueue_text_event(_make_event("continuation text", Platform.TELEGRAM))
    +
    +        # Short chunk arrived → should use normal delay now
    +        await asyncio.sleep(0.15)
    +        adapter.handle_message.assert_called_once()
    +        text = adapter.handle_message.call_args[0][0].text
    +        assert "continuation text" in text
    +
    +
    +# =====================================================================
    +# Feishu adaptive delay
    +# =====================================================================
    +
    +def _make_feishu_adapter():
    +    """Create a minimal FeishuAdapter for testing adaptive delay."""
    +    from gateway.platforms.feishu import FeishuAdapter, FeishuBatchState
    +
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = object.__new__(FeishuAdapter)
    +    adapter._platform = Platform.FEISHU
    +    adapter.config = config
    +    batch_state = FeishuBatchState()
    +    adapter._pending_text_batches = batch_state.events
    +    adapter._pending_text_batch_tasks = batch_state.tasks
    +    adapter._pending_text_batch_counts = batch_state.counts
    +    adapter._text_batch_delay_seconds = 0.1
    +    adapter._text_batch_split_delay_seconds = 0.3
    +    adapter._text_batch_max_messages = 20
    +    adapter._text_batch_max_chars = 50000
    +    adapter._active_sessions = {}
    +    adapter._pending_messages = {}
    +    adapter._message_handler = AsyncMock()
    +    adapter._handle_message_with_guards = AsyncMock()
    +    return adapter
    +
    +
    +class TestFeishuAdaptiveDelay:
    +    @pytest.mark.asyncio
    +    async def test_short_chunk_uses_normal_delay(self):
    +        adapter = _make_feishu_adapter()
    +        event = _make_event("short msg", Platform.FEISHU)
    +        await adapter._enqueue_text_event(event)
    +
    +        await asyncio.sleep(0.15)
    +        adapter._handle_message_with_guards.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_near_limit_chunk_uses_split_delay(self):
    +        """A chunk near the 4096-char limit should trigger longer delay."""
    +        adapter = _make_feishu_adapter()
    +        long_text = "x" * 4050
    +        event = _make_event(long_text, Platform.FEISHU)
    +        await adapter._enqueue_text_event(event)
    +
    +        await asyncio.sleep(0.15)
    +        adapter._handle_message_with_guards.assert_not_called()
    +
    +        await asyncio.sleep(0.25)
    +        adapter._handle_message_with_guards.assert_called_once()
    +
    +    @pytest.mark.asyncio
    +    async def test_split_continuation_merged(self):
    +        adapter = _make_feishu_adapter()
    +
    +        await adapter._enqueue_text_event(_make_event("x" * 4050, Platform.FEISHU))
    +        await asyncio.sleep(0.05)
    +        await adapter._enqueue_text_event(_make_event("continuation text", Platform.FEISHU))
    +
    +        await asyncio.sleep(0.15)
    +        adapter._handle_message_with_guards.assert_called_once()
    +        text = adapter._handle_message_with_guards.call_args[0][0].text
    +        assert "continuation text" in text

  • tests/gateway/test_wecom.py

    @@ -508,6 +508,7 @@ async def test_on_message_builds_event(self):
             from gateway.platforms.wecom import WeComAdapter
     
             adapter = WeComAdapter(PlatformConfig(enabled=True))
    +        adapter._text_batch_delay_seconds = 0  # disable batching for tests
             adapter.handle_message = AsyncMock()
             adapter._extract_media = AsyncMock(return_value=(["/tmp/test.png"], ["image/png"]))
     
    @@ -539,6 +540,7 @@ async def test_on_message_preserves_quote_context(self):
             from gateway.platforms.wecom import WeComAdapter
     
             adapter = WeComAdapter(PlatformConfig(enabled=True))
    +        adapter._text_batch_delay_seconds = 0  # disable batching for tests
             adapter.handle_message = AsyncMock()
             adapter._extract_media = AsyncMock(return_value=([], []))