Hermes Agent 源码专题【左扬精讲】—— Gateway 消息处理流程：GatewayRunner / BasePlatformAdapter / build_session_key
这是 Hermes Agent 源码专题【左扬精讲】系列 第 9 篇。
本篇是 Layer 4 消息网关层 —— 当一个 Telegram 私聊里输入 "把仓库 README 翻译成中文"，Hermes 是怎么从 0 到 1 走完这条链路的：Telegram getUpdates → BasePlatformAdapter.handle_message() → build_session_key(source) → GatewayRunner._handle_message() → AIAgent → stream edit_message → send() 回 Telegram。
第 5、6、7、8 篇讲了工具怎么注册、怎么编排、命令怎么执行、长期记忆怎么挂上。本篇切到消息网关层：Hermes 把 "20+ 平台 chat 转 agent turn" 放在 BasePlatformAdapter 后面，由 GatewayRunner 做唯一编排器。所有上层调用 (handle_message / send / edit_message / interrupt()) 拿到的都是同一个 MessageEvent 契约 —— 不需要知道存储是 SQLite、本地 JSON、还是 Telegram Bot API 长轮询。
Layer 框架视角：先看 窄腰 (BasePlatformAdapter + 3 个 @abstractmethod + MessageEvent 数据类)，再看 编排器 (GatewayRunner 怎么 auth/授权/intr 路由单消息)，最后看 边界 (build_session_key 的隔离规则 + 跨 profile 多路复用 + prompt caching 影响)。
Layer 4 ─ 消息网关层
════════════════════════════════════════════════════════════Layer 4 ─ 抽象接口（gateway/platforms/base.py）BasePlatformAdapter ABC ← 窄腰：3 abstract + 默认 send/edit/deleteconnect / disconnect / send ← 必实现 3 个 @abstractmethodedit_message / delete_message / send_voice ← 可选覆写（默认 success=False）handle_message / _send_with_retry ← 内置消息处理流水线MessageEvent / MessageType / SendResult ← 数据契约SessionSource dataclass ← 来源描述EphemeralReply str ← TTL 控制Layer 4 ─ 编排器（gateway/run.py）GatewayRunner ← GatewayAuthorizationMixin + Kanban + Slashstart_gateway / GatewayRunner.start ← 入口_handle_message / _handle_message_with_agent← 单条消息流水线_handle_update_command / _handle_active_session_busy_message← 路由分流DeliveryRouter ← Cron 投递shutdown_signal_handler ← 信号整合acquire_gateway_runtime_lock / write_takeover_marker← 重复实例防护Layer 4 ─ Session 抽象（gateway/session.py）SessionSource / SessionContext / SessionEntry ← 数据build_session_key / build_session_context ← 唯一来源SessionStore ← SQLite JSON 混合持久化is_shared_multi_user_session / _session_key_namespace← 隔离规则Layer 4 ─ 22 个内置 adapter（gateway/platforms/）telegram / discord / slack / matrix / signal / mattermostfeishu / wecom / weixin / dingtalk / qqbot / yuanbaowhatsapp / whatsapp_cloud / email / sms / webhookapi_server / msgraph_webhook / bluebubbles← 不同协议、同一接口Layer 4 ─ 插件发现（gateway/platforms/__init__.py）BasePlatformAdapter / MessageEvent / SendResult 直接 importQQAdapter / YuanbaoAdapter 通过 __getattr__ 懒加载 ← PEP 562Layer 4 GatewayRunner BasePlatformAdapter MessageEvent build_session_key SessionStore DeliveryRouter handle_message prompt caching
本篇学习重点
必须掌握
- 理解 gateway/platforms/base.py 中的 BasePlatformAdapter ABC 为什么定义 connect / disconnect / send 3 个核心方法 + 1 个数据契约 MessageEvent
- 理解 gateway/run.py 怎么用 GatewayRunner (第 2358 行) 统一处理 auth / 命令分发 / busy session 路由 / interrupt()
- 理解 gateway/session.py 第 646 行 build_session_key() 为什么同时考虑 chat_type / chat_id / thread_id / user_id，以及什么时候按 user 隔离、什么时候共享 thread
- 理解 handle_message() 与 _handle_message() 的边界 —— adapter 层只做 消息入库，runner 层才做 agent 调度
- 理解 _running_agents 缓存为什么必须有 —— 不缓存就会每条消息重建 AIAgent，prompt caching 立刻失效
需要了解
- MessageType 9 种类型 (TEXT/PHOTO/VIDEO/AUDIO/VOICE/DOCUMENT/STICKER/COMMAND/LOCATION) 怎么被不同 adapter 翻译
- EphemeralReply 与 _schedule_ephemeral_delete() 怎么实现"几秒后自动撤回"
- DeliveryRouter 投递跨平台消息 (cron 输出 / 后台进程完成) 怎么处理目标平台与本地 fallback
- 多 profile multiplexing 下 _session_key_namespace() 怎么把 key 路由到独立 namespace
- start_gateway 的重复实例防护 (acquire_gateway_runtime_lock + takeover_marker) 怎么避免 flap loop
目录
- 一、Gateway 子系统的位置 ── Layer 4 解决什么？
- 二、BasePlatformAdapter ABC ── 窄腰与 3 个 abstractmethod
- 三、MessageEvent 数据契约 ── 9 种类型 + 来源 + 媒体
- 四、GatewayRunner 编排器 ── 鉴权 / 路由 / active session 三段式
- 五、build_session_key 与跨 profile 隔离
- 六、SessionStore 持久化 ── SQLite + JSON 双层
- 七、prompt caching 边界 ── AIAgent 缓存复用
- 八、FAQ 20 问
一、Gateway 子系统的位置 ── Layer 4 解决什么？
Layer 视角 ─ 消息网关这一层解决什么问题？
前 8 篇覆盖了 Agent 怎么调工具（Layer 2）、命令怎么执行（Layer 2.5）、长期记忆怎么挂上（Layer 3）。它们都只关心"agent core 本身怎么转"。但用户实际操作 Hermes 时，70% 的对话来自 聊天平台消息：Telegram 私聊说一句、Discord 群里 @ 一下机器人、Slack 工作区发条消息、微信群里 DTMF 提交一个 cron task。这些 chat 跟 CLI 不同──
- 多对多异步 ── 一个用户可能在 N 个平台同时对话，Hermes 必须知道 "同一个人 / 同一个会话 / 同一条话题" 是谁
- 排队 + 打断 ── 用户可能连续发 3 条消息，agent 还没回，第 4 条又来了；必须能把后面 3 条 合并 / 排队 / 中断+重启
- 媒体 + 命令 + 普通文本 ── 语音、图片、文件、/cmd、纯文本走 4 条不同的传递路径
Gateway 子系统（Layer 4）解决的就是这些问题：把 "聊天平台的 chat 事件" 翻译成 "agent turn 的 message 数组"，反过来再把 agent 输出翻译回平台消息。它跟 Layer 1/2/3 是完全正交的几个层次 ── Layer 4 不在乎 adapter 是 Telegram 还是 Slack，Layer 1 不在乎 turn 是从 CLI 还是 Telegram 来的。
Hermes 给出三个互相绑定的设计约束：
- 窄腰 ABC ── gateway/platforms/base.py 第 1803 行 BasePlatformAdapter(ABC)，定义 3 个 @abstractmethod (connect/disconnect/send) + 大量默认实现
- 编排器单例 ── gateway/run.py 的 GatewayRunner（第 2358 行）封装 auth / 命令分发 / interrupt / shutdown，调用方只对 _handle_message() 不对 adapter 数组
- Session key 唯一来源 ── gateway/session.py 第 646 行 build_session_key()，所有 chat 进来先过这一函数，规则统一，避免不同 adapter 各自实现一套
1.1 全景：22 个 adapter + GatewayRunner 的位置
概览：
Hermes 当前内置 22 个 PlatformAdapter 实现，全部放在 gateway/platforms/，由 gateway/platforms/__init__.py 通过 from .base import BasePlatformAdapter, MessageEvent, SendResult 直接 import（QQ / Yuanbao 走 __getattr__ 懒加载）。它们差异化的是底层协议（Telegram Bot API 长轮询 / Slack Socket Mode / Feishu WebSocket / Email IMAP / webhook HTTP）和消息 schema（inline keyboard / topic / thread / rich card 等），共享的是同一个 BasePlatformAdapter 契约。
定位：
本篇核心源码文件 ──
- gateway/platforms/base.py（4000+ 行）── BasePlatformAdapter ABC（第 1803 行）+ MessageEvent（第 1423 行）+ SessionSource
- gateway/run.py（17210+ 行）── GatewayRunner（第 2358 行）+ start_gateway（第 17042 行）
- gateway/session.py（1500+ 行）── SessionSource（第 71 行）+ build_session_key（第 646 行）+ SessionStore（第 737 行）
- gateway/delivery.py（400+ 行）── DeliveryRouter（第 175 行）
- gateway/platforms/<name>.py × 22 ── 各 adapter 实现
1.2 Gateway 子系统与 Layer 1/2/3 的边界
理由：
为什么 Gateway 是独立 Layer 而不是 Tool 的一部分？
- Tool 是 agent 内部的能力 ── terminal.run()、search_files()、delegate_task() 都是被 model 主动调用的
- Gateway 是 chat 事件到 agent turn 的翻译层 ── 从用户角度看是"把 bot 加进 Telegram"，从代码看是 "每个 chat 进来走 handle_message → _handle_message → AIAgent"
- Agent core 不在乎输入从哪来 ── run_agent.py 拿到的就是 messages 数组，不知道也不需要知道是 CLI / Telegram / Slack / Cron
具体边界 ── BasePlatformAdapter 只暴露 send() / edit_message() / handle_message() 给 runner，不能让 adapter 直接访问 AIAgent 内部。runner 持有 self._running_agents: Dict[str, AIAgent]（gateway/run.py 第 2453 行），adapter 只知道"session_key 跟某条 chat 绑定"。
本节小结
- Gateway 子系统（Layer 4）解决 聊天平台 chat → agent turn 的翻译与持久化
- gateway/platforms/base.py 第 1803 行的 ABC 定义了 3 个 abstract + 大量默认实现
- gateway/run.py 第 2358 行的编排器把 auth / 路由 / interrupt / shutdown 收拢到一个入口
- 与 Tool/Memory 系统正交 ── Agent core 拿到的就是 messages，不知道也不需要知道是哪个 adapter 进来的
二、BasePlatformAdapter ABC ── 窄腰与 3 个 abstractmethod
Layer 视角 ─ ABC 这一层解决什么？
一个 PlatformAdapter 在概念上要做 4 件事：连上平台收消息、断连时清理资源、向 chat 发消息、识别 user/cmd 并把 message 转成结构化事件。再加 改消息 / 删消息 / 发语音 / 发图 等可选能力。Hermes 把这 4 件事的必实现部分定死在 ABC 上，把可选部分做成默认实现 —— 一个最小化的 Telegram 适配器只需覆写 3 个方法 ── connect + disconnect + send。
结果：22 个内置 adapter 全部继承同一个父类，行为差异收敛到 send 的实现细节（rich card 渲染 / inline keyboard / topic 路由）。上层 GatewayRunner._handle_message() 永远拿到一个 BasePlatformAdapter 子类，不需要 import 具体实现。
2.1 3 个核心 abstractmethod
契约：
gateway/platforms/base.py 第 1803 行 class BasePlatformAdapter(ABC) 的 3 个核心方法签名：
- connect（第 2299 行，@abstractmethod async）── async def connect(self) -> bool，连上平台开始收消息，返回 True 表示成功
- disconnect（第 2308 行，@abstractmethod async）── async def disconnect(self) -> None，断连清理
- send（第 2313 行，@abstractmethod async）── async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult，返回 SendResult(success, message_id)
关键约束：
- connect() 必须返回 bool ── start_gateway() 拿 True 计数 (在 gateway/run.py 里)，至少有 1 个 adapter 成功才进入正式启动
- send() 必须返回 SendResult ── 即使发送失败也用 success=False，不抛异常。上层 _send_with_retry()（第 3423 行）靠 success 字段判断重试
- send() 的 reply_to 必须支持原消息 id ── 大多数平台用 message_id 即可，但 Telegram thread 模式要带 thread_id，这是 metadata 的责任
2.2 大量默认实现 ── edit_message / delete_message / send_voice
契约：
可选方法的默认实现（gateway/platforms/base.py）：
- edit_message(chat_id, message_id, content, *, finalize=False)（第 2369 行）── 默认返回 SendResult(success=False)，告诉上层"这个平台不支持 edit，让 stream consumer 用 send() 重发"
- delete_message(chat_id, message_id)（第 2398 行）── 类似，能力级回退
- send_voice / send_image / send_video / send_document（第 2753/2663/2795/2815 行）── 默认实现委托给 send() 或返回失败
- send_typing / stop_typing（第 2589/2598 行）── 默认 no-op（不支持就静默）
- send_slash_confirm / send_clarify / send_private_notice（第 2477/2512/2569 行）── 高层 helper
关键约束：
- 所有可选方法都用 SendResult(success=False) 默认返回 ── 上层通过 success 判定，不抛异常。这是 PR #4926 之后立的规矩，因为原来抛 NotImplementedError 时 stream consumer 没法 fallback 重发
- REQUIRES_EDIT_FINALIZE 类属性（第 2340 行） ── 某些平台 (DingTalk AI Card) 必须显式 finalize=True 才完成 streaming lifecycle；其他平台把它当 no-op
2.3 内部状态机 ── _running / _active_sessions / _pending_messages
契约：
adapter 自身维持 3 个状态字段（gateway/platforms/base.py 第 1836-1902 行）：
- self._running: bool（第 1844 行）── 是否已 connect() 成功；is_connected property（第 2236 行）即返回该值
- self._active_sessions: Dict[str, asyncio.Event]（第 1857 行）── 哪个 session_key 正在被 agent 处理，用来拦截同 session 的并发消息
- self._pending_messages: Dict[str, MessageEvent]（第 1858 行）── 同 session 来新消息时，合并到 pending slot，等当前 turn 结束才喂给下一轮
- self._session_tasks: Dict[str, asyncio.Task]（第 1859 行）── owner task map，让 /stop /new /reset 取消正确的 task
细节：
看 gateway/platforms/base.py 第 1836-1862 行构造与字段（精简）：
class BasePlatformAdapter(ABC):"""Base class for platform adapters."""# Capability flag (markdown-rendering adapters set True)supports_code_blocks: bool = Falsetyped_command_prefix: str = "/"def __init__(self, config: PlatformConfig, platform: Platform):self.config = configself.platform = platformself._message_handler: Optional[MessageHandler] = None# hook e.g. Telegram DM topic recovery rewrites event.source.thread_idself._topic_recovery_fn: Optional[Callable[[Any], Optional[str]]] = Noneself._running = False# Fatal error tracking (token expired / banned / webhook removed)self._fatal_error_code: Optional[str] = Noneself._fatal_error_message: Optional[str] = Noneself._fatal_error_retryable = Trueself._fatal_error_handler = None# Per-session interrupt Event used to gate concurrent messagesself._active_sessions: Dict[str, asyncio.Event] = {}self._pending_messages: Dict[str, MessageEvent] = {}self._session_tasks: Dict[str, asyncio.Task] = {}本节小结
- BasePlatformAdapter(ABC)（第 1803 行）定义 3 个 @abstractmethod ── connect / disconnect / send
- 所有可选方法（edit_message / delete_message / send_voice 等）默认返回
SendResult(success=False) - adapter 自身维持 _active_sessions / _pending_messages / _session_tasks 三组状态
- 单 worker pattern 加上 owner task map，让 /stop / /new / /reset 取消正确的 task
三、MessageEvent 数据契约 ── 9 种类型 + 来源 + 媒体
Layer 视角 ─ 消息契约这一层解决什么？
22 个平台的 chat 事件五花八门：Telegram 给 Update.message.photo[-1].file_id，Slack 给 files[0].url_private，Email 给 raw RFC822 字节。如果让 GatewayRunner 直接 deal with 这些原生对象，下游代码会立刻变成一团 if-else。
Hermes 用一个 统一 dataclass MessageEvent 把所有 chat 事件翻译成同构字段 ── text / message_type / source / media_urls / media_types / auto_skill / channel_prompt / internal。下层 agent 不需要知道来源是 Telegram 还是 Slack。
3.1 MessageEvent 字段
契约：
gateway/platforms/base.py 第 1423 行 @dataclass class MessageEvent 的核心字段（精简）：
- text: str（第 1430 行）── 已 normalize 的纯文本，mention @bot 已 strip，/cmd args 已 split
- message_type: MessageType（第 1431 行）── TEXT/LOCATION/PHOTO/VIDEO/AUDIO/VOICE/DOCUMENT/STICKER/COMMAND 共 9 种
- source: SessionSource（第 1434 行）── 来源标识（看 1.1 节）
- raw_message: Any（第 1437 行）── 平台原始对象，adapter 自己用，下层不读
- message_id / platform_update_id（第 1438/1447 行）── 平台 message id + PTB update id (/restart 用来 ack 边界)
- media_urls / media_types（第 1451/1452 行）── 已经下载到本地的媒体路径列表
- reply_to_message_id / reply_to_text（第 1455/1456 行）── 回复上下文，做 prompt context injection
- auto_skill（第 1460 行）── Telegram DM Topics / Discord channel_skill_bindings 自动加载的 skill
- channel_prompt（第 1464 行）── Discord channel_prompts 注入的 ephemeral system prompt
- channel_context（第 1470 行）── 因 require_mention 错过的 history backfill
- internal: bool（第 1474 行）── synthetic events 标记，跳过 user 授权
3.2 MessageEvent 的 3 个 helper
源码视角 ─ gateway/platforms/base.py 第 1479-1507 行：
def is_command(self) -> bool:"""Check if this is a command message (e.g., /new, /reset)."""return self.text.startswith("/")def get_command(self) -> Optional[str]:"""Extract command name if this is a command message."""if not self.is_command():return Noneparts = self.text.split(maxsplit=1)raw = parts[0][1:].lower() if parts else Noneif raw and "@" in raw:raw = raw.split("@", 1)[0]# Reject file paths: valid command names never contain /if raw and "/" in raw:return Nonereturn rawdef get_command_args(self) -> str:"""Get the arguments after a command."""if not self.is_command():return self.text关键设计：
- 拒绝误识别 ── 第 1493 行
"/" in raw短路，避免将/usr/local这种被作为文本开头误判为 command - 处理
/cmd@botname形式 ── 第 1490 行 split@，Telegram 群组里/help@hermes_bot才能调到正确的 bot
3.3 SessionSource ── 来源描述
结构：
gateway/session.py 第 71 行 @dataclass class SessionSource（精简）：
- platform: Platform（第 80 行）── Platform enum
- chat_id: str（第 81 行）── chat 标识
- chat_type: str = "dm"（第 83 行）──
dm/group/channel/thread4 种 - user_id / user_name / user_id_alt（第 84/85/88 行）── 发送者标识；user_id_alt 给 Signal UUID / Feishu union_id
- thread_id: Optional[str]（第 86 行）── forum topic / Discord thread / Slack thread
- chat_topic / chat_id_alt / guild_id（第 87/89/91 行）── guild = Discord server / Slack workspace
- profile: Optional[str]（第 99 行）── 多 profile multiplexing 路由字段
本节小结
- MessageEvent（第 1423 行）统一了 22 个平台的事件格式
- 9 种 MessageType（第 1401 行）：TEXT/LOCATION/PHOTO/VIDEO/AUDIO/VOICE/DOCUMENT/STICKER/COMMAND
- SessionSource（第 71 行）描述来源，支持 dm/group/channel/thread 四种 chat_type
- get_command() 拒绝误识别（路径形式
/usr/local不会当 cmd）
四、GatewayRunner 编排器 ── 鉴权 / 路由 / active session 三段式
Layer 视角 ─ 编排器这一层解决什么？
ABC 只定义了"每个 adapter 应该能做什么"，但没有回答"调用方怎么用 N 个 adapter + 处理排队 + 鉴权"。如果让 gateway/run.py 每个调用点自己跑鉴权逻辑、agent 缓存、interrupt 检查，会立刻撞到 4 个问题：
- auth 一致性 ── 冷路径要鉴权、热路径 (busy session) 也要鉴权，_handle_message + _handle_active_session_busy_message 必须共享
- agent cache ── 不缓存 AIAgent，prompt caching 失效，Anthropic 上 10x cost
- active session 拦截 ── 同一个 Telegram chat 在 agent 还在跑时再发一条，不能直接调第二次 _handle_message，要走 merge_pending_message_event
- 命令分流 ── /stop / /new / /reset 这几个必须能 取消正在跑的 agent，其他命令 (/approve / /deny / /status / /background / /restart) 只需 inline dispatch
gateway/run.py 的 GatewayRunner（第 2358 行）把这 4 件事一起收拢，由 GatewayAuthorizationMixin、GatewayKanbanWatchersMixin、GatewaySlashCommandsMixin 3 个 mixin 切分职能。
4.1 字段总览
定位：
gateway/run.py GatewayRunner 的关键字段（第 2421-2580 行）：
- self.adapters: Dict[Platform, BasePlatformAdapter]（第 2396 行）── 默认 profile 的 adapter map
- self._profile_adapters: Dict[str, Dict[Platform, BasePlatformAdapter]]（第 2402 行）── 次级 profile 的 adapter 二级 map（多路复用开启才有）
- self.session_store = SessionStore(...)（第 2421 行）── session 持久化 + process_registry reset 保护
- self.delivery_router = DeliveryRouter(self.config)（第 2425 行）── cron 投递
- self._running_agents: Dict[str, Any]（第 2453 行）── session_key → AIAgent，prompt caching 核心
- self._active_session_leases（第 2455 行）── lease lock
- self._pending_messages（第 2456 行）── _queued_events overflow buffer
- self._agent_cache: OrderedDict（第 2504 行）── LRU + idle TTL = 复用 AIAgent
- self._session_model_overrides / reasoning_overrides（第 2509/2512 行）── /model /reasoning per-session override
- self._failed_platforms（第 2523 行）── 重连跟踪
4.2 _handle_message() ── 主入口
源码视角 ─ gateway/run.py 第 7145 行的 docstring 把链路说清楚：
async def _handle_message(self, event: MessageEvent) -> Optional[str]:"""Handle an incoming message from any platform.This is the core message processing pipeline:1. Check user authorization2. Check for commands (/new, /reset, etc.)3. Check for running agent and interrupt if needed4. Get or create session5. Build context for agent6. Run agent conversation7. Return response"""source = event.source# Startup restore gate: queue events while sessions are auto-resumingif getattr(self, "_startup_restore_in_progress", False) \and not getattr(event, "internal", False):self._queue_startup_restore_event(event)return None# Internal events skip user authorization (background completion)is_internal = bool(getattr(event, "internal", False))# Fire pre_gateway_dispatch plugin hookif not is_internal:from hermes_cli.plugins import invoke_hook as _invoke_hook_hook_results = _invoke_hook("pre_gateway_dispatch",event=event, gateway=self, session_store=self.session_store,)关键设计：
- 7 步流水线 docstring（第 7150-7156 行）── auth / cmd 分流 / interrupt / session / context / run / return
- Startup restore gate ── 第 7160-7166 行，重启时 auto-resume 期间真实 inbound 入队，避免和 synthetic resume turn 抢
- pre_gateway_dispatch 钩子（第 7179-7211 行）── 用户来源可由插件改写为 skip / rewrite / allow
4.3 鉴权 ── _is_user_authorized + pairing
源码视角 ─ gateway/run.py 第 7215-7258 行的 authorize / pair 链路：
elif source.user_id is None:# chat-scoped allowlist (e.g. TELEGRAM_GROUP_ALLOWED_CHATS)if not self._is_user_authorized(source):logger.debug("Ignoring message with no user_id from %s", source.platform.value)return None
elif not self._is_user_authorized(source):logger.warning("Unauthorized user: %s (%s) on %s", ...)if source.chat_type == "dm" and self._get_unauthorized_dm_behavior(source.platform) == "pair":platform_name = source.platform.value if source.platform else "unknown"# Rate-limit ALL pairing responsesif self.pairing_store._is_rate_limited(platform_name, source.user_id):return Nonecode = self.pairing_store.generate_code(platform_name, source.user_id, source.user_name or "")if code:adapter = self.adapters.get(source.platform)if adapter:await adapter.send(source.chat_id,f"Hi~ I don't recognize you yet!\n\n"f"Here's your pairing code: `{code}`\n\n"f"Ask the bot owner to run:\n"f"`hermes pairing approve {platform_name} {code}`")关键设计：
- 无 user_id 走 chat-scoped allowlist（第 7215-7224 行）── Telegram 频道 / 匿名管理员 / channel forwards 这种场景
- DM 配对 / Group 静默（第 7228 行）── 没授权的用户在 DM 里给 pairing code；在 group 里直接忽略
- Pairing rate-limit（第 7233 行）── 防止 spam，每秒 N 次降级
4.4 busy 路由 ── _handle_active_session_busy_message
行为：
gateway/run.py 第 4071 行 async def _handle_active_session_busy_message 处理三种行为模式：
- "interrupt"（默认）── 立即调 running_agent.interrupt(event.text)（第 4194 行），agent 下一检查点就退出当前 turn，新消息顶替
- "queue" ── 把消息 merge 到 adapter._pending_messages（第 4179-4184 行），等当前 turn 完才喂下一轮
- "steer" ── 调 running_agent.steer(steer_text)（第 4166 行）注入到运行中的 prompt
关键设计：
- 子代理保护 ── 第 4144-4154 行，detect 到 parent agent 在驱动 subagents 时自动降级 interrupt → queue，避免 interrupt() 级联杀掉 delegate_task，要操作员发 /stop 才彻底取消
- Busy ack 节流 ── 第 4208 行
_BUSY_ACK_COOLDOWN = 30，每 session 30 秒最多发一条 "⏳ running"，避免 spam
4.5 命令分流 ── should_bypass_active_session
源码视角 ─ gateway/platforms/base.py 第 3960-4015 行 handle_message 的命令旁路：
if session_key in self._active_sessions:cmd = event.get_command()from hermes_cli.commands import should_bypass_active_sessionif should_bypass_active_session(cmd):# /stop, /new, /reset cancel the in-flight taskif cmd in {"stop", "new", "reset"}:self._discard_text_debounce(session_key)try:await self._dispatch_active_session_command(event, session_key, cmd)except Exception as e:logger.error("[%s] Command '/%s' dispatch failed: %s", ...)return# /approve, /deny, /status, /background, /restartlogger.debug("[%s] Command '/%s' bypassing active-session guard", ...)try:response = await self._message_handler(event)_text, _eph_ttl = self._unwrap_ephemeral(response)if _text:_r = await self._send_with_retry(chat_id=event.source.chat_id, content=_text,reply_to=_reply_anchor_for_event(event),metadata=_mark_notify_metadata(_thread_meta),)关键设计：
- 3 个 ↘ "cancel + 派发" 命令 (/stop / /new / /reset) ── 必须取消 in-flight adapter task；走 _dispatch_active_session_command 串行化 cancel + runner response + pending drain
- 5 个 ↘ "inline dispatch" 命令 (/approve / /deny / /status / /background / /restart) ── 不取消 in-flight task，直接 _message_handler(event)，返回再 _send_with_retry 回平台
- 不调用 _process_message_background（第 3967-3969 行注释）── 它管理 session lifecycle，与 running task 抢 cleanup (PR #4926 教训)
本节小结
- GatewayRunner（第 2358 行）封装 auth / cmd / busy / interrupt / shutdown
- _handle_message()（第 7145 行）7 步流水线：auth → cmd → interrupt → session → context → run → return
- 3 种 busy 模式：interrupt / queue / steer + 子代理自动降级
- 3 类 bypass 命令：/stop / /new / /reset 取消 task；/approve / /deny / /status / /background / /restart inline dispatch
五、build_session_key 与跨 profile 隔离
Layer 视角 ─ Session key 这一层解决什么？
Hermes 在 GatewayRunner 缓存的是 OrderedDict[str, AIAgent]，key 是 session_key。如果 key 不稳定，缓存立即失效 (每条消息都新建 AIAgent，prompt caching 失效)。如果 key 太宽，两个用户的对话会 bleed 进同一个缓存。
这个 key 的构造规则集中在一个函数 ── build_session_key()（gateway/session.py 第 646 行）。它是全网唯一权威，不允许任何 adapter 自己拼接。
5.1 DM vs Group / Channel / Thread
源码视角 ─ gateway/session.py 第 680-734 行核心规则：
def build_session_key(source: SessionSource,group_sessions_per_user: bool = True,thread_sessions_per_user: bool = False,profile: Optional[str] = None,
) -> str:ns = _session_key_namespace(profile)platform = source.platform.valueif source.chat_type == "dm":dm_chat_id = source.chat_idif source.platform == Platform.WHATSAPP:dm_chat_id = canonical_whatsapp_identifier(source.chat_id)if dm_chat_id:if source.thread_id:return f"{ns}:{platform}:dm:{dm_chat_id}:{source.thread_id}"return f"{ns}:{platform}:dm:{dm_chat_id}"# Fall back to sender id (non-standard adapters / synthetic sources)dm_participant_id = source.user_id_alt or source.user_idif dm_participant_id:...return f"{ns}:{platform}:dm:{dm_participant_id}"...participant_id = source.user_id_alt or source.user_id...key_parts = [ns, platform, source.chat_type]if source.chat_id:key_parts.append(source.chat_id)if source.thread_id:key_parts.append(source.thread_id)# Default: shared thread sessions (UX: everyone in thread sees same convo)isolate_user = group_sessions_per_userif source.thread_id and not thread_sessions_per_user:isolate_user = Falseif isolate_user and participant_id:key_parts.append(str(participant_id))return ":".join(key_parts)关键设计：
- DM 隔离规则（第 682-709 行）── chat_id 在的话用 chat；否则降级到 user_id_alt/user_id，避免"无 chat_id 全部 collapse 到一个
{ns}:{platform}:dm"的 cross-user bleed - Thread 默认共享（第 728 行）── thread_id 存在且 thread_sessions_per_user=False 时，同 thread 内所有用户共享一个 session，这是 Slack thread / Telegram forum topic / Discord thread 的预期 UX
- WhatsApp JID 标准化（第 684-685/698-702 行）── 桥接在 JID/LID 之间反复切换的同一人，避免"两个 isolated per-user session"
5.2 多 Profile multiplexing 与 namespace
源码视角 ─ gateway/session.py 第 626-643 行：
def _session_key_namespace(profile: Optional[str]) -> str:"""Return the key namespace for the active profile.Multiplexing disabled (default): returns "agent:main" — legacy singlegateway namespace. Byte-identical to before, so legacy session JSONsload correctly.Multiplexing enabled: each profile gets its own namespace, so twoprofiles running on the same HERMES_HOME don't share sessions even ifthey're connected to the same Telegram bot."""if profile:return f"agent:{profile}"return "agent:main"关键设计：
- 关 multiplexing 字节相同 ── profile=None →
agent:main，与旧 session JSON 完全兼容 - 开 multiplexing → profile 当 ns ──
agent:coder/agent:casual等；同一个 Telegram bot 可被 2 个 profile 各跑一遍而不串 session - 调用方 ── build_session_key 第 680 行 ns = _session_key_namespace(profile)，profile 默认 None，仅 multiplexing gateway 显式传
5.3 is_shared_multi_user_session
契约：
gateway/session.py 第 605 行 def is_shared_multi_user_session：由 current key 反推，是否是多人共享（thread 默认共享 / DM 永远不共享 / group 按配置）。下层 _is_user_authorized 在 shared session 模式下会放过非 user_id 但 chat-allowlisted 的发送者。
本节小结
- build_session_key()（第 646 行）是 session key 唯一来源，所有 adapter 共用
- DM 用 chat_id 隔离，thread 默认共享（同 thread 内所有人看到同一会话）
- _session_key_namespace()（第 626 行）走
agent:<profile>ns，多 profile multiplexing 不 bleed
六、SessionStore 持久化 ── SQLite + JSON 双层
Layer 视角 ─ 持久化这一层解决什么？
一个会话能跨重启活下来，靠的是 SessionStore 把 key → entry 的映射定期写到磁盘。它的特殊之处在于用了双层：
- SQLite（hermes_state.SessionDB）── 完整 message transcript + metadata，第 4 篇已讲
- JSON index（gateway/session.py _save() 第 790 行）── 仅 key → ID/Title/Source 的轻量 mapping，写盘用 atomic_replace 防中途崩溃
这种 split 是因为 SessionDB 是 Hermes-wide 复用 (CLI / TUI / Gateway 都用)，Gateway 再加一份 JSON 会浪费 (CLI 不需要)；但 Gateway 启动要快速知道"哪些 session 存在、绑定到哪个 chat"，JSON index 比 SQL 查询快几个数量级。
6.1 SessionStore 字段
定位：
gateway/session.py class SessionStore（第 737 行）构造（第 745 行）：
- self._entries: Dict[str, SessionEntry]（第 749 行）── in-memory key → entry 映射
- self._loaded: bool（第 750 行）── 懒加载标志
- self._lock: threading.Lock（第 751 行）── Lazy load 互斥
- self._has_active_processes_fn（第 752 行）── process_registry 回调，用于 /reset 保护
- self._db = SessionDB()（第 758 行）── SQL 后端；失败 → 退到 JSONL
6.2 _resolve_profile_for_key ── key namespace 解析
源码视角 ─ gateway/session.py 第 813-830 行：
def _resolve_profile_for_key(self, source: Optional[SessionSource] = None) -> Optional[str]:"""Return the profile namespace for session keys.Disabled (default): returns None so keys stay in legacy agent:main.Enabled: prefers source.profile (set by /p/<profile>/ URL prefixor per-credential adapter), falling back to the active profile name."""if not getattr(self.config, "multiplex_profiles", False):return Noneif source is not None and source.profile:return source.profiletry:from hermes_cli.profiles import get_active_profile_namereturn get_active_profile_name() or "default"except Exception:return None关键设计：
- /p/<profile>/ URL prefix（第 824 行）── 多路复用开启后，URL 加前缀指定 profile，比全局 active profile 更精细
- per-credential adapter ownership ── 不同 bot token 路由到不同 profile，与 source.profile 双轨
6.3 atomic_replace 写入
契约：
gateway/session.py _save()（第 790 行）：临时文件 + fsync + atomic_replace 三步，确保 SIGTERM / OOM 不会留下半截 JSON。except BaseException 兜底 unlink tmp，避免磁盘被占。
本节小结
- SessionStore（gateway/session.py 第 737 行）双层持久化：SQLite transcript + JSON index
- _save() 用 tmp + fsync + atomic_replace 防半截写入
- _resolve_profile_for_key 把多 profile 路由到 namespace
七、prompt caching 边界 ── AIAgent 缓存复用
Layer 视角 ─ Agent 缓存这一层解决什么？
第 1 篇讲过 Hermes 的"per-conversation prompt caching is sacred" ── 长对话复用 cache 前缀，每次 API call 的 cost 跟 message diff 强相关。Gateway 子系统涉及这个边界的关键问题是 ── 同一个 Telegram chat 来 N 条消息，必须复用同一个 AIAgent，否则 system prompt 重建一次就 invalidate 一次。
7.1 _agent_cache ── LRU + idle TTL
契约：
gateway/run.py GatewayRunner.__init__ 初始化两段关键逻辑（第 2493-2512 行）：
# Cache AIAgent instances per session to preserve prompt caching.
# Without this, a new AIAgent is created per message, rebuilding the
# system prompt (including memory) every turn — breaking prefix cache
# and costing ~10x more on providers with prompt caching (Anthropic).
# Key: session_key, Value: (AIAgent, config_signature_str)
#
# OrderedDict so _enforce_agent_cache_cap() can pop the least-recently-
# used entry (move_to_end() on cache hits, popitem(last=False) for
# eviction). Hard cap via _AGENT_CACHE_MAX_SIZE, idle TTL enforced
# from _session_expiry_watcher().
import threading as _threading
self._agent_cache: "OrderedDict[str, tuple]" = OrderedDict()
self._agent_cache_lock = _threading.Lock()关键设计：
- OrderedDict ── cache hit 时 move_to_end()，eviction 时 popitem(last=False) 实现 LRU
- 2 个上限（gateway/run.py 第 65-66 行）──
_AGENT_CACHE_MAX_SIZE=128+_AGENT_CACHE_IDLE_TTL_SECS=3600.0，长生命周期 gateway 不会无限涨 - config_signature ── 缓存值带 config fingerprint，配置变了自动重 build，不会 cross-config reuse
7.2 _last_resolved_model 兜底
理由：
gateway/run.py 第 2464 行 self._last_resolved_model: Dict[str, str]（注释说明）：—— 上游 mtime-keyed config-cache miss 偶尔返回空 model，会让 AIAgent 用 model="" 初始化，所有 API 调用都 HTTP 400 "No models provided"。last-known-good 兜底保证同一 session 总是能用上一次的模型。
关键设计：
- per-session + 全局
"*"── 单 session 优先，全局兜底 - 无网络调用 ── 不读 config、不查凭证，纯内存
7.3 跨 profile multiplexing 的 cache 隔离
理由：
多 profile 模式下，_agent_cache 不需要拆 ── session_key 已经带 ns 前缀（_session_key_namespace() 第 626 行 agent:coder vs agent:casual），cache key 自动分组到不同 namespace，互不覆盖。
Lesson_running_agents 缓存是 prompt caching 的根基 ── Hermes 每条 chat 进来到 runner，先查 self._running_agents.get(session_key)，缓存命中就直接复用，没命中再 build + insert。LRU + idle TTL 兜底防止 OOM。任何"想给每个 chat 单独的 AIAgent 实例"的设计都会破坏 prompt caching，单条消息 10x 成本上升 ── 这是 PR #4926 修复的 PR 之前 gateway 默认行为的真实问题。
本节小结
- _agent_cache（gateway/run.py 第 2504 行）OrderedDict + LRU + idle TTL，保留 prompt caching
- _last_resolved_model（第 2464 行）兜底 config-cache miss 返回空 model
- 多 profile multiplexing 不需要拆 cache ── session_key 自带 ns 前缀已分组
八、FAQ 20 问
FAQ 分组说明
本节围绕 Gateway 子系统，每条都是"读整个系列前最该知道的事"。从 BasePlatformAdapter 契约、MessageEvent 数据、GatewayRunner 编排、Session key 隔离、持久化、prompt caching 边界六个维度展开。
Q1. BasePlatformAdapter ABC 强制实现哪些方法？
3 个 @abstractmethod async 方法。gateway/platforms/base.py 第 1803 行的 ABC 强制 3 个实现 ── connect（第 2299 行）/ disconnect（第 2308 行）/ send（第 2313 行）。其他如 edit_message / delete_message / send_voice 都是默认 SendResult(success=False)。
Q2. send() 为什么必须返回 SendResult 而不是抛异常？
为了上层能 fallback 重试。gateway/platforms/base.py 第 1552 行 SendResult(success, message_id) 形态：上层 _send_with_retry()（第 3423 行）靠 success=False 判定是否重试，不抛异常确保 stream consumer 能 fallback ── PR #4926 教训：原 NotImplementedError 让 stream 崩到整轮。
Q3. MessageEvent 怎么标识命令消息？
靠 text.startswith("/")，但有 3 个 sanitization。gateway/platforms/base.py 第 1479-1495 行 is_command / get_command ── strip /、转小写、split @botname、拒绝 path-like (/usr/local) 。
Q4. _handle_message() 和 handle_message() 的关系？
两个完全不同的层。BasePlatformAdapter.handle_message()（gateway/platforms/base.py 第 3926 行）是 adapter 层 ── 接收 chat 事件、build session_key、检查 active_session 决定走哪个分支。GatewayRunner._handle_message()（gateway/run.py 第 7145 行）是 runner 层 ── 鉴权、命令分发、agent 调度、返回 response。前者 set 在 adapter.set_message_handler(self._handle_message)。
Q5. busy session 时再来消息怎么走？
3 种模式：interrupt / queue / steer。gateway/run.py 第 4071 行 _handle_active_session_busy_message() ── 默认 interrupt（立即调 running_agent.interrupt()），queue（merge 到 adapter._pending_messages 等下轮），steer（running_agent.steer(text) 注入运行中的 prompt）。
Q6. /stop / /new / /reset 为什么必须取消 agent？
它们是 session 终止命令。gateway/platforms/base.py 第 3978 行 if cmd in {"stop", "new", "reset"}：走 _dispatch_active_session_command，串行化 cancel + runner response + pending drain ── 保证 /new 后下一条消息是新 session，不会被旧 agent 续上。其他命令 (/approve / /deny / /status / /background / /restart) 不 cancel，直接 inline dispatch (第 3989-4014 行)。
Q7. 子代理运行中为什么 busy_input_mode 自动降级？
防止 interrupt 级联杀掉 subagent。gateway/run.py 第 4144-4154 行 demoted_for_subagents ── AIAgent.interrupt() 级联清 _active_children 列表，几分钟的 delegate_task 工作会被一句闲聊 interrupt 干掉。自动降 interrupt → queue。操作员要 cancel 必须显式 /stop（不受 demote 影响）。
Q8. build_session_key() DM 规则？
用 chat_id 优先，降级到 user_id，绝不 collapse 到 agent:main:<platform>:dm。gateway/session.py 第 682-709 行 ── chat_id 在就用 chat；否则用 user_id_alt or user_id；再否则才用 thread_id；最后才 fallback dm bare。注释强调：不 fallback 到 sender 会 cross-user history bleed。
Q9. thread 默认为什么共享 session？
因为这是 thread 语义预期 UX。gateway/session.py 第 728 行 ── if source.thread_id and not thread_sessions_per_user: isolate_user = False：Slack thread / Telegram forum topic / Discord thread 里 所有人共享同一会话，否则 thread 就退化成 N 个独立 DM。要 per-user isolation 显式打开 thread_sessions_per_user=True。
Q10. WhatsApp key 为什么 canonical？
桥接 JID/LID 切换会让 key 漂移。gateway/session.py 第 684-685 + 698-702 行 canonical_whatsapp_identifier() ── 同一 WhatsApp 用户在 JID ↔ LID 之间反复切换，不 canonical 就会被分成两个 isolated per-user session。
Q11. _session_key_namespace 关 multiplexing 时返回什么？
agent:main，字节相同。gateway/session.py 第 626-643 行 ── disable 模式下 profile=None → agent:main，与旧 session JSON 100% 兼容，升级不需要 migrate。
Q12. _agent_cache 上限是多少？
_AGENT_CACHE_MAX_SIZE = 128 + idle TTL 3600.0s。gateway/run.py 第 65-66 行常量。LRU 拍出最旧 + idle 1h 定期清，由 _session_expiry_watcher() 推动。
Q13. start_gateway() 怎么防重复实例？
PID file + scoped lock + takeover marker。gateway/run.py 第 17056-17160 行 ── get_running_pid() 检测已有实例 → 写 takeover marker → terminate_pid(force=False)，等 10s → 还活着就 SIGKILL → force-kill 后仍要 _pid_exists 二次确认避免 2 个 live gateway 抢同一 token (issue #19471)。
Q14. DeliveryRouter 做什么？
Cron 输出 / 后台进程事件投递到原 chat。gateway/delivery.py 第 175 行 class DeliveryRouter ── deliver(target, content)：解析 target=platform:chat_id 字符串，找对应 adapter.send()；本地 fallback 用 _save_full_output() 写到 ~/.hermes/outputs/ 兜底。
Q15. EphemeralReply 怎么实现"几秒后撤回"？
str 子类 + ttl_seconds。gateway/platforms/base.py 第 1575 行 class EphemeralReply(str)，带 _ttl_seconds 标记。adapter 层 _unwrap_ephemeral()（第 3401 行）拆出 (text, ttl)，发送成功后调 _schedule_ephemeral_delete()（第 2443 行）scheduler 调度 adapter.delete_message()。
Q16. _send_with_retry() 怎么定义重试策略？
基于 _is_retryable_error / _is_timeout_error。gateway/platforms/base.py 第 3423 行 ── _is_retryable_error()（第 3382）+ _is_timeout_error()（第 3390）判 timeout → 退避 → 重试到一定次数；非 retryable 立即放弃。
Q17. disconnect() 时的超时是多少？
_ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0。gateway/run.py 第 68 行常量。_adapter_disconnect_timeout_secs()（第 2886 行）返回该值（可被配置覆盖）。这是 graceful shutdown 给每个 adapter 留的清理窗口 ── 超过 5s 直接 force kill，避免 systemd TimeoutStopSec 后面 SIGKILL。
Q18. adapter.connect() 超时是多少？
_PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0。gateway/run.py 第 67 行常量。_connect_adapter_with_timeout()（第 2916 行）用这个值包一层 wait_for，防止某个平台卡死把 start_gateway 拖住。
Q19. MessageType 有几种？
9 种。gateway/platforms/base.py 第 1401 行 ── TEXT / LOCATION / PHOTO / VIDEO / AUDIO / VOICE / DOCUMENT / STICKER / COMMAND。每个 adapter 把平台原生 event 翻译到对应 MessageType，下层不感知来源。
Q20. 22 个内置 adapter 怎么发现？
gateway/platforms/__init__.py 直接 import 3 个；QQ/Yuanbao 懒加载。gateway/platforms/__init__.py 第 11-46 行 ── from .base import BasePlatformAdapter, MessageEvent, SendResult 直接 export；其他 19 个走 __getattr__（PEP 562）首次访问时 import，省 48ms 启动 + 8MB RSS。
FAQ 全篇总纲
- 契约：3 abstract + SendResult fallback（Q1 / Q2 / Q15 / Q16）
- 数据：MessageEvent + SessionSource（Q3 / Q19）
- 编排：_handle_message 7 步 + 3 种 busy mode + subagent 降级（Q4 / Q5 / Q6 / Q7）
- Session：DM/thread 隔离 + WhatsApp canonical + namespace（Q8 / Q9 / Q10 / Q11）
- 启动 / 关闭：start_gateway 防重复 + adapter timeout（Q13 / Q17 / Q18）
- 持久化：DeliveryRouter 投递 + JSON index + SQLite 双层（Q14 / Q20）
下一篇预告
第 10 篇 ─ 20+ 平台 adapter 实现（Telegram / Discord / Slack / 飞书 / 钉钉 / 微信 / Matrix …）（敬请期待）
Gateway 子系统覆盖完了 ── 模型怎么调工具（Layer 2）、怎么执行命令（Layer 2.5）、怎么记长期事实（Layer 3）、怎么处理 chat（Layer 4）。下一篇会展开 22 个 adapter 的差异化：
- Telegram ── Bot API 长轮询 + forum topic + DM topic mode (Bot API 9.4+) + inline keyboard + media group
- Slack ── Socket Mode + mrkdwn + thread anchoring +
"!"命令前缀重写 + slash command integration - Discord ── Gateway WebSocket + 1440-min auto-archive threads + role-based access + guild scope
- Feishu / Wecom / Weixin / Dingtalk / QQBot / Yuanbao ── 国内平台分别用 WebSocket / 加密回调 / wxwork API，密码学签名链路
- Email / SMS ── 拉模式 (IMAP 轮询 + SMTP 发) + Twilio-style webhook 推模式
- WebHook / API server ── 通用 HTTP 入口，给自定义集成
Layer 2 / Layer 2.5 / Layer 3 / Layer 4 / Layer-adapter 五层结构完整后，下一步是 Layer 5 ─ Cron 调度 + 跨进程边界。