一条 Matrix 消息从「路由入队」到「LLM 流式返回」中间到底发生了什么本文以 AgentTeams 场景下的一段真实运行日志为线索逐行定位到 OpenClaw 源码还原 Embedded Agent Runner 的完整执行链路。背景某天晚上我给AgentTeams 多 Agent 框架系统 ClawForge 里的一个 Workercode-analyst运行时是 OpenClaw发了一条消息。Worker 很快回复了 。ClawForge研发全流程代码审查与自愈系统介绍我打开它的 Gateway 日志决定逐行追踪——每一行日志背后是哪段源码在运行。日志来自 AgentTeams 多 Agent 框架ClawForge里的一个 Workercode-analyst运行时是 OpenClaw日志分两段log.log消息从 Matrix 通道进来 → 路由 → 入队 → 触发 hooks《源码解读15》、《源码解读16》的日志内容log2.logEmbedded Agent Runner 真正启动组装 prompt、调 LLM、流式返回《源码解读17》的日志内容下面就是这份日志的源码解读。日志行 1路由解析决定这条消息该给谁[routing] resolveAgentRoute2026-08-11T04:39:16.46500:00 [routing] resolveAgentRoute: channelmatrix accountIddefault peerdirect:admin:... guildIdnone teamIdnone bindings0对应源码src/routing/resolve-route.ts 613调用点在src/channels/inbound-event/envelope.ts:42exportfunctionresolveAgentRoute(input: ResolveAgentRouteInput):ResolvedAgentRoute{// 根据 channel / peer / guildId / teamId 解析出目标 agent 与 session}逻辑消息进入 Gateway 后第一步是路由——把 Matrix 消息解析成目标 agent。bindings0表示没有自定义绑定规则走默认路由。resolveAgentRoute()函数的工作流程标准化输入normalizeTokenchannel/accountId、normalizeIdpeer/guildId/teamId——把所有标识符统一为小写、去空格、去非法字符构建 sessionKey调用buildAgentSessionKey()用agentId channel accountId peer拼出唯一的会话标识Agent 查找缓存AgentLookupCacheWeakMap 实现以 config 对象为 key缓存 Agent ID 的查找结果避免重复遍历Agent 绑定解析核心流程——通过 8 个层级tiers依次匹配8 层绑定的优先级从高到低const tiers [ binding.peer, // 精确匹配发送者 binding.peer.parent, // 父级 peer如 thread 的父消息 binding.peer.wildcard, // 通配符 peer同类型任意用户 binding.guildroles, // Discord 服务器 角色组合 binding.guild, // 仅服务器 binding.team, // Teams 团队 binding.account, // 账号级别 binding.channel, // 渠道级别 ]每一层都调用matchesBindingScope()检查是否满足该层的约束条件guild/team/roles 等。匹配成功后choose()回调生成最终的 route 对象{ agentId, sessionKey, mainSessionKey, matchedBy }。如果所有层都未匹配fallback 到default使用默认 Agent即main。本日志中bindings0表示没有任何自定义绑定消息最终路由到默认 Agentmain。}resolveAgentRoute函数流程normalize → build bindings index → cache check → tiers 匹配 → choose → 返回日志行 2入站分发主流程入口 第一条探针[diagnostic]三段式诊断入队 状态切换idle → processin[diagnostic] message received: channelmatrix chatIdroom:!oXl5... sessionIdunknown ... [diagnostic] message queued: sessionId171b17b9-... queueDepth1 sessionStateidle [diagnostic] session state: previdle newprocessing reasonmessage_start queueDepth1对应源码src/logging/diagnostic.ts:539注意sessionIdunknown—— 这一刻 session 还没解析出来。源码src/auto-reply/reply/dispatch-from-config.ts:278消息进入 session 队列queueDepth1随后 session 状态从idle翻转到processing。两条日志恰好记录入队和开始处理两个相邻时刻。这三行日志来自三个诊断函数记录了消息从到达到进入处理的完整状态流转函数动作日志行logMessageReceived()消息到达发出诊断事件message receivedlogMessageQueued()消息入队queueDepth 1message queuedlogSessionStateChange()会话状态 idle → processingsession state三个函数的结构一致areDiagnosticsEnabledForProcess()—— 全局开关关闭则直接返回diag.isEnabled(debug)—— 日志级别检查debug 才输出emitDiagnosticEvent()—— 发出结构化诊断事件markActivity()—— 标记会话活跃时间状态机的关键逻辑// idle → processing if (params.state processing prevState ! processing) { state.activeQueuedTurn state.queueDepth 0; // 标记有消息待处理 } // processing → idle if (params.state idle) { state.queueDepth Math.max(0, state.queueDepth - 1); // 出队 state.activeQueuedTurn false; }日志行 5、15事件钩子reply_dispatch[plugins] [hooks] running reply_dispatch (1 handlers, first-claim wins)对应源码src/plugins/hooks.ts逻辑主流程走到分发回复这一步时广播一个reply_dispatch钩子注册的 handler这里 1 个被回调执行。first-claim wins表示多个 handler 时第一个认领消息的生效。关键这就是 OpenClaw 的观察者模式——业务代码不直接调用 handler而是发事件handler被通知。它解耦了模块但执行仍是同步串行handler 跑完主流程才继续OpenClaw 的 Hook 系统是一套面向切面编程AOP的实现AOP 概念OpenClaw 实现Join pointrunBeforeXxx/runAfterXxx等入口函数PointcuthookName如reply_dispatchAdvice插件注册的 handler 函数WeavingcreateHookRunner()在 Gateway 启动时完成核心设计createHookRunner(registry, options)export function createHookRunner(registry, options) { // 超时策略 const catchErrors options.catchErrors ?? true; // 默认 fail-open const failurePolicyByHook { before_agent_run: fail-closed, ...options.xxx }; const voidHookTimeoutMsByHook { agent_end: 30_000, before_compaction: 30_000, ... }; const modifyingHookTimeoutMsByHook { before_agent_run: 15_000, ... }; // 三种执行策略 async function runVoidHook() { ... } // 并行 fire-and-forget async function runModifyingHook() { ... } // 顺序执行 累积合并 async function runClaimingHook() { ... } // 顺序执行 first-claim-wins }三种策略对比策略执行方式返回适用场景runVoidHookPromise.all并行无观察类model_call_started、agent_endrunModifyingHookfor循环顺序累积合并修改类before_prompt_buildrunClaimingHookfor循环顺序第一个handled:true抢占类reply_dispatch、inbound_claim日志 → runClaimingHook(reply_dispatch, ...) ← 框架负责找到所有注册的 handler 并执行↓getHooksForName(registry, reply_dispatch) → 找到 1 个注册的 handler↓logger?.debug?.([hooks] running reply_dispatch (1 handlers, first-claim wins)) 输出日志↓runClaimingHooksList(hooks, ...) 遍历执行↓for (const hook of hooks) {result await hook.handler(event, ctx); ← 真正的 handlerMatrix plugin 注册的if (result?.handled) return result; ← 谁先说我处理了谁赢}runClaimingHook 裁判负责调度hook.handler 选手真正干活的那个函数first-claim wins 比赛规则第一个举手说「handled」的选手胜出核心逻辑​⚠️ 关键误区澄清L683 不是「第一个 hook 胜出」而是「第一个返回{handled: true}的 hook 胜出」。如果 hook 返回handled: false或 undefined框架会继续尝试下一个。日志里实际有两个 hook 调用在日志第5行和第15行① [hooks] running reply_dispatch → runReplyDispatch() → hookName: reply_dispatch② [hooks] running before_agent_reply → runBeforeAgentReply() → hookName: before_agent_replyrunBeforeAgentReply 和 runReplyDispatch 是两个不同的 hook各管各的它们各自过滤自己的 handler互不交叉// L1038-1047: reply_dispatchrunReplyDispatch() → runClaimingHook(reply_dispatch, ...)→ getHooksForName(registry, reply_dispatch)→ 只找注册为 reply_dispatch 的 handler// L835-844: before_agent_replyrunBeforeAgentReply() → runClaimingHook(before_agent_reply, ...)→ getHooksForName(registry, before_agent_reply)→ 只找注册为 before_agent_reply 的 handler它们是不同的切入点。日志里两个都触发了按先后顺序先 reply_dispatch消息分发后 before_agent_replyagent 生成回复前。4 个 Claiming 类型的 HookHook 名入口函数时机inbound_claimrunInboundClaim()消息到达插件抢先认领before_dispatchrunBeforeDispatch()分发前拦截reply_dispatchrunReplyDispatch()替代默认分发逻辑before_agent_replyrunBeforeAgentReply()Agent 回复前拦截可返回合成回复日志行 6message dispatch started[diagnostic] message dispatch started: channelmatrix sessionIdunknown sessionKeyagent:main:main sourcereplyResolver对应logMessageDispatchStarted()表示路由和 hook 处理完成消息正式进入分发阶段。日志行 7内存与上下文预检跑之前先体检memoryFlush和preflightCompaction这两行是Agent 执行前的记忆管理检查memoryFlush检查是否需要将上下文持久化tokenCount20033远低于threshold126000不需要 flushpreflightCompaction检查是否需要在发送 prompt 前做上下文压缩token 远低于阈值不需要逻辑真正跑 agent 前先做两道检查——会话记忆要不要刷盘、上下文要不要压缩。两者都是20033 126000阈值未超所以不触发。如果 token 数接近contextWindow的 75%即150000 * 0.85 127500左右就会触发压缩。日志行 9-10SQLite 事务落盘 会话轮次创建[sqlite/transaction] slowSQLitetransaction lock wait [diagnostic] session turncreated: runIdabf06f19-... agentIdmain ...这是 Agent 执行的并发控制层SQLite 锁等待由于 SQLite 在同一时间只允许一个写者session 状态写入前需要等待锁随后生成runId登记本次用户触发的 runtriggeruser。亮点那句slow ... lock wait是串行化的直接证据——如果真有多线程并行无锁就不会有等锁这回事了。// Lane 队列的核心逻辑 enqueue(laneId, task) → 入队等待 dequeue(laneId) → 出队执行FIFO // Session lane 保证同一会话串行Agent lane 限制全局并发waitMs2表示等了 2ms 就拿到执行权——几乎没有竞争。日志行 11-14Session Turn 创建 Lane 并发控制两级 lane 队列Admission 并发控制落地 [diagnostic] laneenqueue: lanesession:agent:main:main queueSize1[diagnostic] lanedequeue: lanesession:agent:main:main waitMs2queueSize0[diagnostic] laneenqueue: lanemain queueSize1[diagnostic] lanedequeue: lanemain waitMs0queueSize0逻辑两级 lane车道队列——第一级session:agent:main:main串行化同一 session 的并发 run第二级main是全局车道。每个 run 先 enqueue 再 dequeuewaitMs是排队等待时长。关键这就是 OpenClaw 的Admission 并发控制——用队列把可能并发的 run 刻意串行化保证同一 session 同时只跑一个 run。再次证明不是并行而是刻意串行。日志行 15-17二次钩子 harness 选择 Agent 执行启动[plugins] [hooks] running before_agent_reply (1 handlers, first-claim wins) [agents/harness] agent harness selected [agent/embedded] embedded run start: runIdabf06f19-... provideragentteams-gateway modelqwen3.7-plus thinkingmedium messageChannelmatrixbefore_agent_replyhookAgent 生成回复前最后一次拦截机会。插件可以在这里返回合成回复绕过 LLM 调用Harness 选中agent harness selected选择 PI Embedded RunnerOpenClaw 内置的 Agent 执行器Embedded Run 启动Agent 执行引擎正式启动使用qwen3.7-plus模型通过agentteams-gateway代理thinking 级别为 medium逻辑回复前再跑before_agent_reply钩子 → 选择执行 harness此处选 Embedded→embedded run start正式启动。到这里log.log结束进入log2.log的 Embedded Runner 执行链路。log2.log日志行 18-19二次路由 工具最小权限 Tool Policy[routing] resolveAgentRoute: channelmatrix accountIddefault peerdirect:admin:matrix-local.agentteams.io:18080 guildIdnone teamIdnone bindings0 [agents/tool-policy] tool policy removed 9 tool(s) via gateway sender owner-only tools.deny: conversations_list, conversations_send, conversations_turn, cron, gateway, nodes, openclaw, sessions, terminal; matched conversations_list, conversations_send, conversations_turn, cron, gateway, nodes, openclaw, sessions, terminal[agents/tool-policy] tool policy removed 9 tool(s) via gateway sender owner-only tools.deny: conversations_list, conversations_send, conversations_turn, cron, gateway, nodes, openclaw, sessions, terminal源码src/routing/resolve-route.ts:613src/agents/tool-policy-audit.ts:197逻辑run 启动后又做一次路由解析并按 tool-policy 过滤工具deny 掉 9 个高危工具。亮点这是多 Agent 框架的最小权限原则——AgentTeams 的 Worker 只拿受限的 Consumer Token像gateway、terminal、cron、sessions这类能操纵宿主的高危工具被直接禁用。安全边界在这里收口。Q A 日志函数散落在resolve-route.ts、diagnostic.ts、hooks.ts、lanes.ts等不同文件里彼此之间没有直接调用关系。这是不是说明它们是多线程并行执行的不是恰恰相反这条链路是严格串行的。 看起来没有调用关系是两个机制叠加的结果日志里的函数大多是诊断探针logMessageReceived、logMessageQueued、logSessionStateChange…它们是被业务主流程在各自阶段直接调用来打日志的——它们是记录点不是流程点当然彼此不互相调用。真正串联流程的是一条串行的 async 流水线dispatchInboundMessage → dispatchReplyFromConfig → …中间用事件钩子hooks插入横切逻辑。打个比方流水线车间里每个工位不调用下一个工位产品消息在传送带上依次经过每个工位日志只是每个工位贴的过站标签。三个铁证① 日志时间戳严格递增、从不交错②lane enqueue/dequeue队列和slow SQLite transaction lock wait锁的存在恰恰是为了串行化并发③唯一会等的是 async IO网络、写库逻辑流程仍是await串行。小结从一条 Matrix 消息到 Agent 开始执行经过了以下阶段Matrix 消息 → 路由解析(resolve-route) → 诊断记录(diagnostic) → Hook 拦截(reply_dispatch) → 消息分发 → 记忆检查(memoryFlush/compaction) → Session Turn 创建 → Lane 并发控制 → Agent Harness 选中 → Embedded Run 启动 → LLM ↔ Tool 循环 LLM ↔ Tool 循环部分代码逻辑见《openclaw源码解读12LLM↔Tool 核心循环》这就是 OpenClaw 从消息抵达 Agent 到开始执行的全链路。注本文中提到的ClawForge是我做的多agent协同的研发全流程代码自愈与审查系统详见《openclaw源码解读12LLM↔Tool 核心循环》正在规划《OpenClaw源码解读》书籍欢迎出版社编辑交流