網關 (gateway) 核心

源码版本v2026.7.20

職責

網關是 agent 與外部世界的橋梁:接收各平台(Telegram/Discord/Slack/Signal/微信…)訊息,轉成統一的 MessageEvent 餵給 agent;把 agent 的串流 (stream) 輸出轉成各平台的發送動作。gateway/run.py 是這層的主程式(GatewayRunner),協調轉接器 (adapter)、會話、串流消費、投遞帳本 (delivery ledger)、hooks、cron 心跳等多條子系統。

設計動機

把「跟平台打交道」和「跑 agent 迴圈」切開,agent 才不必管 Telegram 長度上限、Discord 雪花 ID、Signal signal-cli 的 RPC schema。統一 MessageEvent + 抽象 BasePlatformAdapter 讓一個 agent 程序同時掛十幾個平台,新增平台只動轉接器。再疊 delivery_ledger,是因為網關長時上線,程序隨時可能崩——agent 已生成的最終回覆不能跟著丟,所以這層歸網關而不是 agent。

關鍵檔案

• class GatewayRunner:3029 — 網關主類別,混入授權/Kanban/Slash 能力
• _handle_message:9947 — 入站訊息分發 (dispatch) 入口
• _handle_message_with_agent:11956 — 把訊息交給 agent 跑一輪
• _start_one_profile_adapter / _start_secondary:9389-9947 — 啟動各 profile 的平台轉接器
• _start_stream_consumer:21218 — 啟動串流消費器
• _start_gateway_housekeeping:22246 — 週期性巡檢(60s)
• _start_cron_ticker:22335 — cron 心跳(60s)
• def main:22965 — 網關程式入口
• 例外與脫敏:276-340 — _gateway_loop_exception_handler / _redact_gateway_user_facing_secrets
• PlatformEntry:39-162 — 平台註冊項資料類別
• class PlatformRegistry:162-260 — 註冊表(register / register_deferred / unregister)

資料流

1. main()(gateway/run.py:22965)啟動 GatewayRunner。
2. _start_one_profile_adapter:9389 逐個拉起平台轉接器(見平台轉接器)。
3. 轉接器收到平台原始事件,轉成統一 MessageEvent(見 gateway/platforms/base.py:1759)。
4. _handle_message(gateway/run.py:9947)分發:授權檢查、會話解析、slash 命令路由,或交給 _handle_message_with_agent(gateway/run.py:11956)。
5. 後者呼叫 agent 的 run_conversation:588,並把串流回調接到 GatewayStreamConsumer:83。
6. 串流輸出經消費器漸進式編輯平台訊息;最終態由 delivery_ledger 記帳,防止崩潰丟失最終回覆。
7. 巡檢與 cron 心跳在背景 60s 週期跑(gateway/run.py:22246 / gateway/run.py:22335)。

GatewayRunner(gateway/run.py:3029)是個多頭混入類別:

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] = {}

授權/Kanban watcher/Slash 命令分別由三個 mixin 提供,主類別只管生命週期與路由——「瘦主類別 + 能力 mixin」讓單個特性能獨立測試、按需裁剪。

入站訊息走 _handle_message,管道開頭先做幾件不能跳過的事——重置跨會話 ContextVar、判斷 startup-restore 期、蓋 scale-to-zero 時鐘、跑 pre_gateway_dispatch 外掛 (plugin) 鉤子:

async def _handle_message(self, event: MessageEvent) -> Optional[str]:
    source = event.source
    # Cross-session leak guard: 新 task 可能繼承兄弟會話的
    # HERMES_SESSION_* ContextVar,先重置成 _UNSET 再走正常流程。
    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() 會快照 spawning context,並發訊息若在父任務裡 set_session_vars() 過,新任務會繼承 sibling 身份,subprocess-env 橋就讀到錯誤 session。

_handle_message_with_agent 找/建 session + 把訊息塞進 agent,並防「非法復活已死會話」:

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): spawning session 可能已 /new-reset,
        # 不能盲切回去復活使用者主動關掉的會話。
        ...

非同步子代理回填時若 spawning session 已被結束,網關寧可丟這次注入也不復活它。

邊界與失敗

• 跨會話 ContextVar 洩漏:_handle_message 跑在 create_task() 裡,會繼承父任務快照。若不先 reset_session_vars(),並發訊息會讀到 sibling 的 session 身份,subprocess 橋會把錯誤身份寫到環境裡。
• 非同步回填注入已死會話:子代理回來時 spawning session 可能已 /new-reset,盲切會復活使用者主動關掉的會話。fail closed 把結果留在 delegation records。
• scale-to-zero 時鐘污染:內部事件(背景程序完成、startup-restore replay)不應算流量,否則閒置網關被保活。程式碼用 is_internal 顯式分流。

小結

網關核心是「事件匯流排 + 串流橋接」:統一 MessageEvent 抽象屏蔽平台差異,GatewayStreamConsumer 把同步 agent 回調橋到非同步平台投遞,delivery_ledger 保證最終態不丟。它和 agent 解耦——agent 只管跑迴圈,投遞細節全在網關。