Gateway Core

源码版本v2026.7.20

Responsibility

The gateway is the bridge between the agent and the outside world: it receives messages from each platform (Telegram/Discord/Slack/Signal/WeChat...) and converts them into a unified MessageEvent fed to the agent; it turns the agent's streaming output into per-platform send actions. gateway/run.py is the main process of this layer (GatewayRunner), coordinating adapters, sessions, stream consumption, the delivery ledger, hooks, the cron heartbeat, and other subsystems.

Design Motivation

Splitting "talking to platforms" from "running the agent loop" means the agent never has to care about Telegram length limits, Discord snowflake IDs, or the Signal signal-cli RPC schema. A unified MessageEvent plus an abstract BasePlatformAdapter lets one agent process host a dozen platforms at once; adding a new platform only touches the adapter. The delivery_ledger sits on top because the gateway is long-lived online and the process can crash at any time — the final reply the agent already produced must not vanish with it, so that responsibility belongs to the gateway, not the agent.

Key Files

• class GatewayRunner:3029 — gateway main class, mixes in authz/Kanban/Slash capabilities
• _handle_message:9947 — inbound message dispatch entry
• _handle_message_with_agent:11956 — hands the message to the agent for a turn
• _start_one_profile_adapter / _start_secondary:9389-9947 — start each profile's platform adapters
• _start_stream_consumer:21218 — start the stream consumer
• _start_gateway_housekeeping:22246 — periodic housekeeping (60s)
• _start_cron_ticker:22335 — cron heartbeat (60s)
• def main:22965 — gateway process entry
• exceptions & redaction:276-340 — _gateway_loop_exception_handler / _redact_gateway_user_facing_secrets
• PlatformEntry:39-162 — platform registration dataclass
• class PlatformRegistry:162-260 — registry (register / register_deferred / unregister)

Data Flow

1. main() (gateway/run.py:22965) starts GatewayRunner.
2. _start_one_profile_adapter:9389 brings up each platform adapter (see Platform Adapters).
3. Adapters receive raw platform events and convert them to unified MessageEvents (see gateway/platforms/base.py:1759).
4. _handle_message (gateway/run.py:9947) dispatches: authz check, session resolution, slash command routing, or handoff to _handle_message_with_agent (gateway/run.py:11956).
5. The latter calls the agent's run_conversation:588 and wires the streaming callback to GatewayStreamConsumer:83.
6. Streaming output progressively edits the platform message via the consumer; the final state is recorded by delivery_ledger to prevent loss of the final reply on crash.
7. Housekeeping and the cron heartbeat run in the background on a 60s cycle (gateway/run.py:22246 / gateway/run.py:22335).

GatewayRunner (gateway/run.py:3029) is a multi-head mixin class:

class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin):
    """Main gateway controller. Manages the lifecycle of all platform
    adapters and routes messages to/from the agent."""
    _busy_input_mode: str = "interrupt"
    _draining: bool = False
    def __init__(self, config: Optional[GatewayConfig] = None):
        self.adapters: Dict[Platform, BasePlatformAdapter] = {}

Authorization, Kanban watcher, and Slash commands come from three separate mixins; the main class only manages lifecycle and routing — "thin main class + capability mixins" lets each trait be tested in isolation and trimmed as needed.

Inbound messages go through _handle_message. The start of the pipeline runs a few non-skippable steps — reset cross-session ContextVars, detect startup-restore window, gate the scale-to-zero clock, and fire the pre_gateway_dispatch plugin hook:

async def _handle_message(self, event: MessageEvent) -> Optional[str]:
    source = event.source
    # Cross-session leak guard: a new task may inherit a sibling session's
    # HERMES_SESSION_* ContextVar, so reset to _UNSET before proceeding.
    try:
        from gateway.session_context import reset_session_vars
        reset_session_vars()
    except Exception:
        logger.debug("reset_session_vars failed at handler entry", exc_info=True)

    if getattr(self, "_startup_restore_in_progress", False) and not getattr(event, "internal", False):
        self._queue_startup_restore_event(event)
        return None

create_task() snapshots the spawning context. If a concurrent message's parent task has called set_session_vars(), the new task inherits the sibling identity, and the subprocess-env bridge would write the wrong session into the environment.

_handle_message_with_agent finds or creates a session, hands the message to the agent, and guards against "illegally reviving a dead session":

async def _handle_message_with_agent(self, event, source, _quick_key, run_generation):
    """Inner handler that runs under the _running_agents sentinel guard."""
    session_entry = await self.async_session_store.get_or_create_session(source)
    session_key = session_entry.session_key
    pinned_session_id = str((getattr(event, "metadata", None) or {}).get("gateway_session_id") or "").strip()
    if pinned_session_id and pinned_session_id != session_entry.session_id:
        # Fail closed (#55578): the spawning session may have already /new-reset,
        # don't blindly switch back and revive a session the user closed.
        ...

When an async sub-agent comes back to fill in results, if the spawning session has already been ended, the gateway drops this injection rather than reviving it.

Boundaries and Failure

• Cross-session ContextVar leak: _handle_message runs inside create_task() and inherits the parent task's snapshot. Without a reset_session_vars() call up front, concurrent messages would read a sibling's session identity and the subprocess bridge would write that wrong identity into the environment.
• Async backfill into a dead session: by the time a sub-agent returns, the spawning session may have already been /new-reset. Blindly switching back would revive a session the user closed on purpose. Fail-closed leaves the result in the delegation records instead.
• Scale-to-zero clock pollution: internal events (background process completion, startup-restore replay) must not count as traffic, or an idle gateway stays alive forever. The code branches these out explicitly with is_internal.

Summary

The gateway core is an "event bus + streaming bridge": the unified MessageEvent abstraction hides platform differences; GatewayStreamConsumer bridges the synchronous agent callback to asynchronous platform delivery; delivery_ledger ensures the final state isn't lost. It is decoupled from the agent — the agent only runs the loop; all delivery details live in the gateway.