第 7 篇里,轮次与步骤的每个边界都伴着事件:claim 时逐条发 agent/inbox/claimed,轮次收尾时发 agent/turn-stopping。要改 dsh 的行为,第一件事不是找函数,而是选事件。架构文档(docs/architecture.zh.md:76)只给了一句话:事件就是扩展点,选对事件域是大多数改动的第一个决定。本篇把这句话展开:四个事件域各是什么、waterfall 如何委托决策、为什么监听顺序改变不了结果,以及面对一个改动该挂到哪里。
两个消费口径先分开:docs/agent-lifecycle.zh.md 末尾的指引说得很直白——需要可回放 transcript 数据的 SDK 用户消费 session/event;agent/* 是用于队列与状态、提示词拦截、请求构造、steering 和错误处理的实时协调接口。同一件事在两个域里各有一个"版本",服务不同的读者,这篇的目的就是让你每次都知道该写进哪个版本、该听哪个版本。
四个事件域
dsh 的事件不是一张平铺的清单,而是四个域,分发语义完全不同。docs/event-producer-consumer.zh.md 用一张生产方/消费方矩阵记录了每个事件谁发谁听;本文按语义把它们归入四张表,并回答两个判据问题:这件事重载之后还要不要存在?这个监听者要不要改变结果?
域一:持久会话事件。追加进会话日志、经 session/event 广播,是回放与冷读的事实来源。核心变体共十三种(含 session/end-seed),插件可以往 SessionEventMap 里声明合并新变体,不必改动拥有该类型的包。判据只有一个:这个事实在进程重启、会话 resume 之后还必须存在,就得落在这里。
| 事件 | 记录的事实 |
|---|---|
turn/start·turn/end | 轮次边界与结局 reason |
step/start·step/end | 步骤边界与步骤结局 |
system/message | 系统提示词 surface 节点(第 9 篇) |
user/message | 进入步骤的用户输入 |
assistant/message·assistant/attempt | 成功结算与仅日志的失败尝试 |
tool/call·tool/result | 工具调用与结果配对 |
request/header·request/context | 路由信封与请求级上下文 |
agent/inbox/spliced | inbox 的结构化变更 |
域二:实时 agent/* emit。进程内瞬态通知,发了就走,监听者的返回值无人读。以 agent/status 为例:它每次翻转都带目标状态,running 的开始时刻是"waking 投递同步保留取消之后",idle 的含义是"没有任何驱动器或维护任务遗留"——它证明的是整个排空区间,不证明某个轮次仍然打开(第 7 篇的 Phase 三态才是内部真相)。agent/assistant-stream 值得单独看一眼:它的 end 帧带着 settlement 结果——要么 committed(附持久事件的类型与 seq),要么 abandoned(放弃,没有持久结算);chunk 帧是瞬态的,UI 应该按"可丢"对待。想要逐 token 的实时渲染听它,想要逐 token 的历史记录去读 assistant/message 里嵌入的紧凑流。
| 事件 | 通知的内容 |
|---|---|
agent/status | idle ⇄ running 翻转 |
agent/inbox/inserted·claimed·discarded | 单条消息的 inbox 生命周期 |
agent/assistant-stream | start/chunk/end 帧,chunk 是瞬态的 |
agent/error | 步骤或轮次失败,verbatim 错误 |
agent/disposed | agent 离开注册表 |
域三:实时 waterfall。监听者拿到 next(),不调用它即短路,返回值是决策。真实的挂载可以很直观地核对:agent/request-error 上有 compaction-basic(上下文溢出恢复)和 llm-retry(退避重试);tools/execute 上有 timeout-policy(超时包装);tools/pre-execute 和 tools/post-execute 上有 hooks 桥接与各类策略。这些包都不改动循环代码,只挂监听。
| 事件 | 决策类型 |
|---|---|
agent/pre-step | reject 或 enter(messages) |
agent/request | 替换冻结的调用配置 |
agent/request-error | retry 或保持终态 |
llm/stream | 替换流式响应源 |
tools/pre-execute | allow / deny / cancel / ask |
tools/execute | 环绕分发的结果(中间件) |
tools/post-execute | accept 或 block |
system-prompt/assemble | 改写组装结果 |
tools/ptc-dispatch-log | 改写 PTC 子调用的日志副本 |
域四:实时 serial。按注册顺序逐个 await,没有 next()。全仓库只有两个:agent/created(创建期初始化,监听失败回滚创建,AgentLoop 在所有监听完成前扣住已排队输入)和 agent/turn-stopping(轮次收尾检查点)。真实的监听者包括 hooks 桥接与 workspace-changes,用途是"在边界提交前做最后一次布置"。
两组对比最能说明语义差别。持久 vs 实时问的是:进程重启、会话重载之后还在不在。tool/result 进日志,模型靠它继续对话;tools/result 只是进程内通知观察者,观察者抛错会被隔离,事实已经落盘不受影响。agent/inbox/spliced 是投影 fold 的输入,没有活跃 Agent 也能冷读;agent/inbox/claimed 是给 UI 的一次性提示,错过即无。waterfall vs serial 问的是:有没有 next() 可委托。agent/created 的监听者不能"把创建交给下游决定",它只能在自己的回调里完成初始化;agent/turn-stopping 同理,下面细讲。
持久事件还有一个结构细节:surface 变体可以在 sourceEventSeqs 里列出被引用的较早事件,并携带 surfaceOp(追加或替换区间)。派生历史不是第二份存储,而是这些节点按序折叠的结果——deriveMessages()(packages/core/session/src/index.ts:856)每次从日志现算,事件改、历史变,没有同步失真的机会。
flowchart TD E["harness 全部事件"] --> P["持久会话事件 session/event"] E --> A["实时 agent/* emit"] E --> W["实时 waterfall 有 next()"] E --> S["实时 serial 无 next()"] P --> P1["turn/step 边界、system/user/assistant 消息"] P --> P2["tool/call·result、request/header·context、inbox/spliced"] A --> A1["agent/status、agent/error、agent/disposed"] A --> A2["inbox 单条通知、assistant-stream 帧"] W --> W1["agent/pre-step、agent/request、agent/request-error"] W --> W2["tools/pre·execute·post-execute、llm/stream、system-prompt/assemble"] S --> S1["agent/created"] S --> S2["agent/turn-stopping"] P -.->|"重载后仍在,冷读可重建"| PD["回放与投影的事实源"]
waterfall:next() 即委托
四个域里最常用的是 waterfall,机制是 Cordis 的 ctx.waterfall(thisArg, name, ...args, next):监听者要么返回自己的决策(短路,下游不再执行),要么 await next() 取得默认决策或下游监听者的决策,在其基础上修改后返回。分发时还有作用域过滤——agent 事件以 agent 为 subject,agent.ctx 里的监听只收到那个 agent 的事件(dispatch.ts:107 的融合分发器负责注入 payload)。
关键事实是:默认决策就是最内层的 next()。没有任何监听者时,机器自己的行为被包装成最后一个 next。每个默认值都值得读一遍,因为它们就是"什么都不装插件时系统的行为":
| 事件 | 默认决策 | 代码位置 |
|---|---|---|
agent/pre-step | enter(已领取批次 + context) | agent.ts:276 |
agent/request | seedConfig(agent options 或已记录 header) | agent.ts:576 |
tools/pre-execute | allow | tools/src/index.ts:1505 |
tools/post-execute | accept(原样接受) | tools/src/index.ts:1782 |
agent/pre-step 的默认值是 enter,且携带已领取批次——第 7 篇说过,pre-step 拒绝或空首批会产生零步骤轮次,所以这个默认点的另一侧不是"放行一切"那么简单,而是"以领取的内容进入步骤"。agent/request 的默认值分两个阶段:首个请求用 agent options 播种,之后的请求折叠日志里最新的 request/header。tools/pre-execute 默认 allow,deny 与 ask 都要监听者显式选择;tools/post-execute 默认 accept,替换 content 或 value 只能二选一(第 10 篇展开)。
以 agent/request 的签名为例(runtime-types.ts:984):监听者拿到 { agent, turn, step, signal } 与 next: () => Promise<LlmCallConfig>,返回替换后的配置。包装 next() 的价值在于可以保留下游结论:agent/pre-step 的包装者先 await next() 拿到下游的 enter 决策,替换消息的同时保留 startsRequestSeries 声明——除非有意替换,下游的劳动不被清零。这正是中间件模式在事件系统里的形状:包一层,改一点,传下去。
waterfall 之外还要记住一条纪律:模型可见的内容必须走已记录的通道。agent/request 的注释明确说,这个 waterfall 不能改消息——想改模型看到的内容,写 system/message 或 user/message(第 9 篇),而不是在请求对象上动刀。
agent/turn-stopping:串行的 data decides
回看第 7 篇的状态机:轮次自然停止、next-step 为空时,循环发出 agent/turn-stopping,先于最后一次 steering 排空运行。它是 serial——逐个 await 监听者,没有 next() 可委托。一个想阻止轮次关闭的监听者,手上只有一个动作:agent.steer(...) 往 inbox 写一条数据。随后机器重读 inbox:有 steering 就再跑一个步骤,没有就关闭轮次。
这就是 "data decides":监听者不能返回"我反对"来否决,只能写数据,让机器依据持久状态做决定。两个监听者一先一后各写一条 steering,与交换顺序执行,最终 inbox 里的内容相同,结果就相同——监听顺序无法影响结果。反过来,若允许直接否决,后注册的监听就能推翻先注册的,决策语义就被注册顺序劫持了。
把 serial 的两个成员放在一起看,形状是统一的:agent/created 也不委托——监听者不能"让下游决定是否创建",失败就是失败,创建回滚,两个 id 都不发布。serial 的语义是责任,不是协商;协商的形状属于 waterfall。
这个设计的另一半是对称的:提前结束轮次的控制也是数据。工具结果携带 concludesTurn 标记(ToolExecutionSuccess 的字段),轮次在该步骤收尾;但已提交的工作不会被短路,同步骤的 additionalContexts 与竞速 steering 照常执行(第 7 篇的收尾纪律)。写数据,等机器读——两个方向的控制都收敛到同一条原则。
该用哪个事件:一张决策树
举两个真实改动收尾。改日志结构?那是事实,要么扩展 SessionEventMap 加持久变体,要么写一条已有的持久事件;不要往 agent/* 里塞——重载之后没人记得。给某个工具加审批策略?拦截点在 tools/pre-execute(执行前)而不是 tools/result(结算后),后者只通知,来不及;若策略要求"每次询问",返回 ask,剩下的交给审批服务。想清楚这两个例子,决策树只是把它们图形化:
flowchart TD
Q{"你要改动什么?"} -->|"一个必须在重载后存活的事实"| A["挂持久会话事件或写入会话日志"]
Q -->|"观察或拦截进行中的 agent 工作"| B["挂 agent/* 实时事件"]
Q -->|"给某个能力 seam 附加策略"| C["挂能力事件 fs/* 或 tools/*,无需依赖循环包"]
Q -->|"改写一次决策或包一层中间件"| D{"调用方提供 next() 吗?"}
D -->|"有 next,默认决策可委托"| E["waterfall 监听"]
D -->|"无 next,只能表达立场"| F["serial 事件,立场写成数据"]文字版同样短:要持久,用会话事件,或者干脆把你的事实写进日志;要观察或拦截进行中的工作,用 agent/*——状态翻转听 agent/status,流式渲染听 agent/assistant-stream,steering 效果听 agent/inbox/claimed;要给 seam 附加策略(文件系统的写意图、工具的执行前检查),用能力事件,它们不 import 循环包;要包中间件或替换决策,找 waterfall;没有 next() 的地方(创建、收尾),记得你的唯一语言是数据。还有一个常见误判值得提醒:需要"每次"都被通知的事实用 emit,需要"决定结果"的事实用 waterfall,需要"事后翻不了案"的否决用 serial 加数据。判断不了时,回到那个判据:重载之后还要不要存在。
小结
事件就是扩展点,而扩展点按四个域分发:持久会话事件是事实,emit 是通知,waterfall 是委托决策,serial 是按序执行的检查点。选错域的代价很实在——把决策塞进 emit,没人读你的返回值;把事实塞进 emit,重启之后就丢了。waterfall 的默认决策永远是最内层的 next();serial 的 agent/turn-stopping 则用 data decides 把否决权从监听顺序手里收回给持久状态。
事件的总数有限,而挂载其上的策略无限。下一篇我们看其中最有分量的一条持久通道:系统提示词如何组装,又如何经准入机制进入派生历史。