Gateway & Platforms

Multi-Channel Message Gateway & Concurrent Message Deduplication Queue

Deep dive into gateway/run.py multi-platform adaptation, dedup_key duplicate interception, and progress message collapse algorithm
Multi-platform Gateway Deduplication
📊 Figure 5-1: Multi-tenant message gateway distribution architecture and message deduplication filtering schematic

🔌 1. Gateway Distribution Model & GatewayRunner Scheduling

The message gateway serves as the agent's always-on tentacle, connecting over 20 different chat platforms including Feishu, Telegram, DingTalk and more. The gateway's main control hub is gateway/run.py:

  • GatewayRunner Class: The core management process of the gateway. On service startup, it fetches platform configuration parameters from config.yaml and spins up a separate Adapter for each enabled communication platform concurrently within an independent asyncio event loop.
  • Dynamic Module Loading: The communication logic for each platform is isolated in separate sub-scripts under gateway/platforms/ (e.g., telegram.py, feishu.py). The system loads them dynamically to maintain low overhead.

🛡️ 2. Concurrent Debounce & dedup_key Message Deduplication

Under poor network conditions, communication platforms may automatically retry messages when Webhook response latency occurs (e.g., if the system's 3-second response window is exceeded). Without gateway-level filtering, the LLM could be called multiple times with the same request, generating massive token bills. Hermes implements in-memory and persistent message filtering in gateway/run.py to prevent re-entry:

  • Building dedup_key: When receiving a Webhook event payload, the gateway performs structural mapping and extracts three specific variables: 
    dedup_key = (platform_str, chat_id, thread_id)
  • notified Dedup Queue: The system maintains a concurrency-safe notified set. If a dedup_key has already been added, it means an AIAgent instance is already processing that conversation. New duplicate requests are immediately intercepted and discarded rather than pushed to the task queue, ensuring idempotent agent interaction at the foundation level.

📉 3. Progress Message Collapse (Progress Deduplication)

In tool-intensive tasks, the agent continuously outputs [Executing: read_file], [Executing: terminal] execution hints. If every step sent a separate message through the chat UI, the user's IM interface would be flooded with useless bubbles (screen spam). To purify the experience, the gateway collapses duplicate status messages into a progress merge mechanism when processing downstream progress notifications. The flow is shown in the diagram below:

Progress Message Collapse Sequence Diagram
📊 Figure 5-2: Progress message merge collapse with client-side dynamic bubble modification sequence diagram
# Progress merge logic in gateway/run.py
def handle_progress_stream(progress_queue):
    last_progress_msg = None
    repeat_count = 0
    
    while True:
        raw_msg = progress_queue.get()
        # Detect if it matches the last progress message
        if raw_msg == last_progress_msg:
            repeat_count += 1
            # Push a tuple with a special __dedup__ tag to notify the gateway
            progress_queue.put(("__dedup__", raw_msg, repeat_count))
        else:
            last_progress_msg = raw_msg
            repeat_count = 0
            progress_queue.put(raw_msg)

When the gateway sends a message to the platform, if it detects "__dedup__", it does not call send_message. Instead, it directly calls the platform's edit_message API to append a repeat count (e.g., (x3)) to the previous bubble. From the chat application's perspective, only a single line of progress text fluidly updates, delivering a premium user experience.

⏳ 4. Graceful Shutdown Signal Interception & Reverse Interrupt Broadcast

When the gateway runs as a daemon process, it may receive a SIGTERM forced shutdown signal from the OS due to systemd or K8s deployment updates. Abrupt termination could corrupt SQLite during an active write transaction. The gateway implements a sophisticated backoff broadcast mechanism:

  • Signal Hijacking & Context Output: On startup, the gateway binds a custom callback handler using signal.signal(signal.SIGTERM, ...). Upon receiving the signal, it immediately saves diagnostic logs including parent_pid, parent_cmdline, loadavg_1m (one-minute load average) within a Shutdown context.
  • Reverse Interrupt Flag: Before teardown, the gateway retrieves all currently active AIAgent instances and forcibly sets their _interrupt_requested property to True.
  • Graceful Persistence: The LLM main loop is immediately intercepted by this boolean flag (interrupting the current LLM thread), flushes its state to state.db, safely disconnects WebSocket communication, and gracefully terminates the process.

🔗 In-Depth Subtopic Analysis

For a deeper understanding of the IM message gateway mechanism, we recommend the following technical deep-dives: