gateway — Multi-Platform Messaging Gateway

🎯 Why This Module Exists

Without the gateway module, every scheduled job, every agent response, and every notification would need its own bespoke integration with each messaging platform. Teams running Hermes across Telegram, Discord, and Slack would maintain three separate codebases with three separate authentication flows, three separate message-formatting pipelines, and three separate session stores. Bug fixes would be multiplied by the number of platforms, and adding a new platform like Signal or Matrix would require changes in every component that ever sends a message.

The gateway solves this by placing a single adapter abstraction between the Hermes agent core and the outside world. Platform adapters translate platform-specific events (Telegram updates, Discord interactions, Slack slash commands) into a normalized InboundMessage envelope and translate outbound AgentResponse objects back into platform-native formats. The session store ensures conversational context persists across messages without leaking between users or platforms. The delivery router decides which adapter receives the output of cron jobs without the scheduler needing to know anything about messaging protocols. This separation means a new platform can be added by implementing a single Python class — roughly 150–300 lines — with zero changes to the agent, scheduler, or plugin system.

💡 Real-world analogy

Think of the gateway as a multilingual hotel concierge desk. Guests arrive speaking different languages (Telegram, Discord, Slack, etc.), but behind the desk every request is logged in the same ledger (session store), routed to the same staff (agent core), and answered in the guest's own language. The concierge doesn't need to understand how the kitchen (plugins) works — they just relay orders and deliver plates. Adding a new language is hiring one new translator, not retraining the entire hotel staff.

flowchart LR
  subgraph External["External Platforms"]
    TG["Telegram"]
    DC["Discord"]
    SL["Slack"]
    WA["WhatsApp"]
    SG["Signal"]
    MX["Matrix"]
    EM["Email"]
    SMS["SMS"]
    OT["Other Adapters"]
  end

  subgraph Gateway["Gateway Core"]
    GR["GatewayRunner\n(run.py)"]
    GC["GatewayConfig\n(config.py)"]
    HR["HookRegistry\n(hooks.py)"]
  end

  subgraph Session["Session Layer"]
    SS["SessionStore\n(session.py)"]
    PS["PairingStore\n(pairing.py)"]
  end

  subgraph Output["Output Pipeline"]
    SC["GatewayStreamConsumer\n(stream_consumer.py)"]
    DR["DeliveryRouter\n(delivery.py)"]
  end

  subgraph Core["Hermes Core"]
    AG["Agent Engine"]
  end

  TG & DC & SL & WA & SG & MX & EM & SMS & OT -->|InboundMessage| GR
  GR --> SS
  GR --> PS
  GR --> HR
  GC -.->|config| GR
  GR -->|invoke| AG
  AG -->|stream| SC
  SC -->|render| TG & DC & SL & WA & SG & MX & EM & SMS & OT
  DR -->|route cron output| GR

sequenceDiagram
    participant P as Platform (Telegram/Discord/...)
    participant A as PlatformAdapter
    participant GR as GatewayRunner
    participant PS as PairingStore
    participant SS as SessionStore
    participant AG as Agent Engine
    participant SC as StreamConsumer
    participant HR as HookRegistry

    P->>A: Platform-native event
    A->>A: Parse into InboundMessage
    A->>GR: dispatch(InboundMessage)
    GR->>HR: fire(on_message_received)
    GR->>PS: check_authorization(user_id)
    PS-->>GR: authorized / denied
    GR->>SS: get_or_create(session_key)
    SS-->>GR: SessionContext (history, metadata)
    GR->>SS: check_lifecycle(idle_timeout, daily_reset)
    GR->>AG: invoke(prompt, context)
    AG-->>SC: stream chunks
    SC->>SC: batch & format chunks
    SC->>A: send_message(formatted_chunk)
    A->>P: Platform-native response
    GR->>SS: persist(updated_context)
    GR->>HR: fire(on_message_complete)

🚶 Step-by-Step Narrated Walkthrough

Follow a single Telegram message from a user as it flows through the entire gateway pipeline:

1. Webhook Arrival — Telegram sends an HTTPS POST to the bot's webhook endpoint. The TelegramAdapter receives the raw JSON payload containing the chat ID, user ID, message text, and any attachments. Latency: ~5 ms to parse.
2. Envelope Construction — The adapter calls parse_update() to transform the Telegram-specific JSON into a normalized InboundMessage dataclass with fields like platform, user_id, channel_id, text, and attachments. Platform quirks (e.g., Telegram's chat.type field) are abstracted away here.
3. Dispatch to GatewayRunner — The adapter calls GatewayRunner.dispatch(msg). The runner is the central event loop — it doesn't know or care that this message came from Telegram. The HookRegistry fires on_message_received hooks (logging, rate-limit checks, analytics).
4. Authorization Check — The runner queries PairingStore.check_authorization(user_id, channel_id). In DM mode, the user must have previously entered a valid pairing code. In group mode, the channel itself is the authorization scope. Unauthorized messages receive a polite rejection and the pipeline short-circuits.
5. Session Key Construction — The runner builds a composite session key: f"{platform}:{channel_id}:{user_id}" for DMs or f"{platform}:{channel_id}" for group chats. This key determines which conversation history is loaded. The algorithm is deliberately simple to make debugging straightforward.
6. Session Retrieval & Lifecycle — SessionStore.get_or_create(key) loads the session from disk (JSON files) or creates a fresh one. Before returning it, the store checks lifecycle policies: if the session has been idle longer than idle_timeout_minutes (default: 120) or a new calendar day has started since the last message, the session is auto-reset to prevent context drift. The old session is archived, not deleted.
7. Agent Invocation — The runner passes the user's message plus the loaded SessionContext (conversation history, system prompt overrides, tool permissions) to the agent engine via agent.invoke(). The agent runs in a subprocess for crash isolation — if the agent segfaults or hangs, only that subprocess dies.
8. Stream Processing — As the agent produces output tokens, GatewayStreamConsumer receives them in real time. It batches tokens into display-friendly chunks (typically 50–100 tokens) to avoid flooding the platform's API with single-character messages. For platforms that support message editing (Telegram, Discord), it sends an initial message and then edits it in place — giving the user a "typing" effect.
9. Platform-Native Formatting — The stream consumer calls adapter.send_message(chunk). Each adapter translates the generic Markdown output into platform-native formatting: Telegram uses HTML tags, Discord uses its own Markdown dialect, Slack uses Block Kit JSON, Email uses MIME multipart. Code blocks, images, and links are all platform-adapted.
10. Session Persistence — After the agent finishes, the runner calls SessionStore.persist(key, updated_context). The updated context includes the new user message, the agent's full response, token counts, and a timestamp. The session file is written atomically (write to temp, then rename) to prevent corruption on crash.
11. Hook Finalization — The runner fires on_message_complete hooks. These typically handle analytics (response latency, token usage), quota tracking, and optional logging to an external observability service.
12. Cleanup & Readiness — The adapter's event loop returns to listening for the next Telegram update. Total end-to-end latency for a simple query: 200–800 ms (dominated by LLM inference time). The gateway is immediately ready for the next message on any platform.

Session Key Construction & Message Batching

Session Key Construction

The Problem: The gateway must route incoming messages to the correct conversation context. A user can talk to the bot in a Telegram DM, a Discord server channel, and a Slack thread simultaneously — each must have its own session. But in group chats, all users share a single context.

The Approach: Build a deterministic composite key from three fields: platform, channel_id, and optionally user_id. For DMs, the key is {platform}:{channel_id}:{user_id}. For group chats, the key is {platform}:{channel_id}. The platform prefix prevents collisions when the same numeric channel ID appears on different platforms (Telegram and Discord both use snowflake-like IDs).

Why This Algorithm: It's O(1) to construct, O(1) to look up in a dict, and — critically — human-readable. An operator debugging a session issue can ls sessions/telegram:* to find all Telegram sessions instantly. Hash-based keys (SHA-256 of the inputs) would be collision-free but opaque, requiring a secondary index for debugging.

Message Batching

The Problem: LLM streaming produces tokens one-at-a-time (5–50 ms apart). Sending each token as a separate platform message would trigger rate limits (Telegram: 30 msgs/sec, Discord: 5 msgs/sec) and create a terrible user experience.

The Approach: The GatewayStreamConsumer accumulates tokens in a buffer. It flushes when: (a) the buffer reaches a threshold (50–100 tokens), (b) a natural break point is detected (double newline, code fence boundary), or (c) a time deadline passes (500 ms since last flush). On platforms supporting edit-in-place (Telegram, Discord), it sends one initial message and edits it repeatedly. On send-only platforms (SMS, Email), it waits for the full response.

Why This Algorithm: The multi-trigger flush strategy balances responsiveness against API rate limits. A pure token-count threshold would delay short responses. A pure timer would flush mid-word. The natural-break-point detection ensures chunks end at readable boundaries.

graph TD
    A["Token arrives from LLM"] --> B{"Buffer full?\n(50+ tokens)"}
    B -->|Yes| F["Flush buffer to platform"]
    B -->|No| C{"Natural break?\n(newline/fence)"}
    C -->|Yes| F
    C -->|No| D{"Timer expired?\n(500ms)"}
    D -->|Yes| F
    D -->|No| E["Append to buffer,\nwait for next token"]
    F --> G{"Platform supports\nedit-in-place?"}
    G -->|Yes| H["Edit existing message"]
    G -->|No| I["Append new message\n(or buffer until done)"]
    H --> E
    I --> E

Trace through an example

Given: A Telegram user asks "Explain quicksort." The LLM streams 200 tokens over 3 seconds.

Step 1: Tokens 1–12 arrive in 180 ms: "Quicksort is a". Buffer has 12 tokens, no break point, timer not expired → keep buffering.

Step 2: Token 13 arrives at 200 ms, it's a newline after the first sentence. Natural break detected → flush. adapter.send_message("Quicksort is a divide-and-conquer...") creates message ID 42.

Step 3: Tokens 14–80 arrive over 1.2 seconds. At token 65, a double-newline is detected → flush via adapter.edit_message(42, accumulated_text). The user sees the message grow in real time.

Step 4: Tokens 81–200 continue with periodic flushes at natural breaks and timer deadlines. Final flush on stream-end sends the complete response.

Result: The user sees 4–5 progressive edits to a single message over 3 seconds, instead of either (a) 200 individual messages or (b) a 3-second delay followed by a wall of text.

Hot Paths

The most performance-sensitive code paths in the gateway module:

• Message Dispatch (GatewayRunner.dispatch): O(1) adapter lookup by platform name, O(1) session key construction, O(1) session dict lookup. Total overhead before agent invocation: <5 ms. This path is hot because every single inbound message traverses it.
• Stream Consumer Flush (GatewayStreamConsumer.flush): O(n) where n = buffered tokens (typically 50–100). Formatting is string concatenation, no regex. Called 3–10 times per agent response. The real bottleneck is the platform API call latency (50–200 ms), not the flush logic itself.
• Session Serialization (SessionStore.persist): O(m) where m = session turns. JSON serialization of 100+ turns can take 5–15 ms. Mitigated by writing asynchronously in a background task — the user-facing response is sent before persistence completes.

Resource Usage

• Memory: Each active session consumes 10–500 KB depending on conversation length. With 1,000 concurrent sessions, baseline memory is ~50–500 MB. Sessions are loaded on-demand and evicted from the in-memory cache after cache_ttl_minutes (default: 30). Agent subprocesses are forked per request and cleaned up after completion.
• I/O: Session reads/writes hit local disk (SSD). Each message triggers 1 read + 1 write. Platform API calls are the dominant I/O: 2–8 HTTPS requests per message (initial send + edits during streaming). Webhook ingestion is passive — platforms push to us, no polling.
• CPU: Minimal — the gateway is I/O-bound. Token batching and message formatting use basic string operations. The LLM inference (the actual CPU/GPU workload) happens in the agent subprocess or on a remote API server, not in the gateway process.

gateway/
├── run.py                         ← GatewayRunner: main event loop & dispatch
├── config.py                      ← GatewayConfig: YAML/env config loader
├── session.py                     ← SessionStore: persistent context w/ lifecycle
├── pairing.py                     ← PairingStore: DM authorization codes
├── hooks.py                       ← HookRegistry: lifecycle event system
├── delivery.py                    ← DeliveryRouter: cron output routing
├── stream_consumer.py             ← GatewayStreamConsumer: token batching
├── models.py                      ← InboundMessage, AgentResponse dataclasses
├── platforms/
│   ├── __init__.py                ← Adapter registry & auto-discovery
│   ├── base.py                    ← BasePlatformAdapter (abstract)
│   ├── telegram.py                ← Telegram Bot API adapter
│   ├── discord.py                 ← Discord bot adapter
│   ├── slack.py                   ← Slack Bolt adapter
│   ├── whatsapp.py                ← WhatsApp Business API adapter
│   ├── signal.py                  ← Signal-cli adapter
│   ├── matrix.py                  ← Matrix (matrix-nio) adapter
│   ├── email.py                   ← IMAP/SMTP email adapter
│   ├── sms.py                     ← Twilio SMS adapter
│   ├── irc.py                     ← IRC adapter
│   ├── mattermost.py              ← Mattermost adapter
│   ├── teams.py                   ← Microsoft Teams adapter
│   ├── line.py                    ← LINE Messaging adapter
│   ├── wechat.py                  ← WeChat adapter
│   ├── viber.py                   ← Viber adapter
│   ├── rocket_chat.py             ← Rocket.Chat adapter
│   ├── zulip.py                   ← Zulip adapter
│   └── mumble.py                  ← Mumble adapter
├── tests/
│   ├── test_runner.py             ← GatewayRunner unit tests
│   ├── test_session.py            ← SessionStore tests
│   ├── test_delivery.py           ← DeliveryRouter tests
│   ├── test_stream_consumer.py    ← StreamConsumer tests
│   └── conftest.py                ← Shared fixtures & mock adapters
└── README.md                      ← Module documentation