DeepSeek Harness 事件分发 API 完全指南：context 中的 parallel / emit / serial / bail / waterfall

• 人工智能
• AI Agent
• Agent 框架
• DeepSeek

事件系统是 DeepSeek Harness“一切皆插件”架构的神经中枢。每一个由 Cordis 构建的 context 都天然混入了事件分发 API（event-dispatch API），插件通过ctx.on()注册监听器、通过ctx.parallel()、ctx.emit()、ctx.serial()、ctx.bail()、ctx.waterfall()五种不同策略分发事件，从而实现解耦的扩展与拦截。本文以 docs/cordis-api/events.md 为骨架，结合其底层实现 vendor/cordis/src/events.ts 与 Harness 各子系统（如 docs/subsystems/core.md）中真实的事件声明，系统讲解事件分发的五种模式、监听器生命周期、EventOptions过滤语义，以及 Harness 中agent/*、internal/*等事件的真实用法。读完本文，你将能准确判断“何时用哪种分发模式”，并写出符合 Harness 扩展点约定的监听器代码。

事件分发 API 的定位：每个 context 的内置能力

在 DeepSeek Harness 中，事件分发 API 是混入（mixin）进每个 context的能力——不需要额外引入服务，任何拿到ctx的地方都可以直接调用ctx.on()、ctx.emit()等方法。这一点在 docs/cordis-api/events.md 开头便明确：该 API 混合到每一个上下文中，而 Harness 各子系统声明的事件及分发模式，会由 scripts/gen-cordis-catalog.ts 自动生成到各自所属的子系统页面（生成命令为pnpm run gen-cordis-catalog，对应中文侧文件为 docs/subsystems/core.zh.md）。

从源码结构看，这套 API 的实际载体是一个事件总线服务，安装为ctx.events，同时以类型声明扩展Context接口（见 vendor/cordis/src/events.ts）：

// vendor/cordis/src/events.ts 中 declare module './context.ts' 的简化 parallel(name, ...args): Promise<void> emit(name, ...args): void serial(name, ...args): Promisify<ReturnType<Events[K]>> bail(name, ...args): ReturnType<Events[K]> waterfall(name, ...args): ReturnType<Events[K]> on(name, listener, options?): () => boolean once(name, listener, options?): () => boolean

注意每个方法都有一个带thisArg的重载（如parallel(thisArg, name, ...args)）：显式传入this供监听器使用，同时也作为 context 过滤的依据。所有事件名称与参数类型都由泛型K extends keyof Events约束，保证编译期类型安全。

DispatchMode：五种事件分发策略

EventOptions 与 DispatchMode 一节给出了核心概念——DispatchMode是事件服务使用的事件分发策略，定义于 vendor/cordis/src/events.ts：

type DispatchMode = 'emit' | 'parallel' | 'serial' | 'bail' | 'waterfall'

五种策略的语义（依据原文档与源码注释）：

其中“bail 值（bail value）”有明确判定规则：非null、非false、非undefined的值即为提前终止信号。该判定由 vendor/cordis/src/events.ts 中的isBailed()实现：

export function isBailed(value: any) { return value !== null && value !== false && value !== undefined }

这意味着监听器返回true、0、''、任意对象等都会触发 bail，而返回null/false/undefined表示“继续传递”。

五种分发方法逐个拆解

ctx.parallel(name, ...args)：并发分发

parallel<K extends keyof Events>(name: K, ...args: Parameters<Events[K]>): Promise<void>

并行分发事件，并发运行所有监听器，返回的 Promise 在所有监听器都 settle 后兑现。若任一监听器拒绝，parallel会抛出由所有拒绝原因组成的AggregateError（实现见 vendor/cordis/src/events.ts）：

async parallel(...args: any[]) { const results = await Promise.allSettled(this.dispatch('emit', args).map(async cb => cb(...args))) const errors = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected') if (errors.length) throw new AggregateError(errors.map(error => error.reason)) }

参数说明：

• name：事件名称。
• args：传递给每个监听器的参数。
• 返回值：一个 Promise，在所有监听器均已完成后兑现（rejected 时聚合所有错误）。

典型用途：多个监听器彼此独立、都需要被执行且都需要等待结果，例如日志上报、状态广播等“扇出”场景。

ctx.emit(name, ...args)：同步分发、忽略返回值

emit<K extends keyof Events>(name: K, ...args: Parameters<Events[K]>): void

同步分发事件，忽略监听器的返回值。实现上直接遍历监听器并调用、不处理返回的 Promise（vendor/cordis/src/events.ts）：

emit(...args: any[]) { this.dispatch('emit', args).map(cb => cb(...args)) }

参数说明：

• name：事件名称。
• args：传递给每个监听器的参数。
• 返回值：无（void）。

这是开销最低的“通知式”分发：适合发出信号、触发副作用，且不关心结果、不需要等待异步完成的场景。若监听器返回了 Promise，它也不会被await（注意由此产生的未处理拒绝）。

ctx.serial(name, ...args)：串行分发、等待直到 bail

serial<K extends keyof Events>(name: K, ...args: Parameters<Events[K]>): Promisify<ReturnType<Events[K]>>

依次等待各监听器，直到其中一个提前终止分发。与bail的区别在于serial会await每个监听器的返回值再判定（vendor/cordis/src/events.ts）：

async serial(...args: any[]) { for (const cb of this.dispatch('serial', args)) { const result = await cb(...args) if (isBailed(result)) return result } }

参数说明：

• name：事件名称。
• args：传递给每个监听器的参数。
• 返回值：第一个提前终止值（非null、非false且非undefined）；若没有监听器 bail，则解析为undefined。

典型场景：按注册顺序协商出一个结果，且监听器可能是异步的。Harness 中agent/turn-stopping就采用serial模式（见 docs/subsystems/core.md）。

ctx.bail(name, ...args)：同步分发、遇到首个 bail 值停止

bail<K extends keyof Events>(name: K, ...args: Parameters<Events[K]>): ReturnType<Events[K]>

依次调用各监听器，直到其中一个提前终止分发。与serial的区别是不做异步等待——同步调用每个监听器，一旦返回非null/false/undefined的值就立即返回（vendor/cordis/src/events.ts）：

bail(...args: any[]) { for (const cb of this.dispatch('bail', args)) { const result = cb(...args) if (isBailed(result)) return result } }

参数说明：

• name：事件名称。
• args：传递给每个监听器的参数。
• 返回值：第一个提前终止值（非null、非false且非undefined）。

典型场景：internal/listener就使用 bail 模式——监听器注册时，若某个 bail 监听器返回非空结果，则直接取代默认的注册逻辑（见下文“内部事件”一节）。

ctx.waterfall(name, ...args)：洋葱模型组合、next续接链

waterfall<K extends keyof Events>(name: K, ...args: Parameters<Events[K]>): ReturnType<Events[K]>

分发一个最后一个参数是next续接回调的事件。这是 Harness 中最重要的扩展模式：每个监听器都会包装调用链的其余部分——调用next()会执行下一个监听器，最终执行内置行为；不调用next()则会否决（veto）后续执行（包括内置行为）。

实现如下（vendor/cordis/src/events.ts）：

waterfall(...args: any[]) { const cbs = this.dispatch('waterfall', args) const inner = args.pop() // 最后一个参数是最内层 next const next = () => { const cb = cbs.shift() ?? inner // 依次弹出监听器，耗尽后落到内置行为 return cb(...args) } args.push(next) return next() // 从最外层监听器开始执行 }

参数说明：

• name：事件名称。
• args：监听器参数；最后一个参数是最内层的next（即内置行为）。
• 返回值：最外层监听器的返回值。

执行顺序是“由外向内”的洋葱结构：最先注册/最靠前的监听器最外层，它先拿到next；每个监听器可以选择在调用next()之前/之后做前处理与后处理，也可以直接返回、不调用next()从而否决整条链。waterfall 天然适合拦截-改写语义，Harness 大量扩展点都采用它，例如：

• agent/pre-step（waterfall）：拒绝提议的 step 或替换进入该 step 的消息，调用next()保留当前消息（docs/subsystems/core.md）；
• agent/request（waterfall）：替换冻结的模型调用配置，await next()得到机器将使用的配置，返回替换值来切换（docs/subsystems/core.md）；
• 工具注册表中对tools/ptc-dispatch-log的分发，见 packages/core/tools/src/index.ts：监听器可以改写工具分发的日志内容，失败时回退到原始内容。

监听器注册：on/once与 fiber 所有权

ctx.on(name, listener, options?)

on<K extends keyof Events>(name: K, listener: Events[K], options?: boolean | EventOptions): () => boolean

注册一个归当前 fiber 所有的事件监听器：

• name：要监听的事件名称。
• listener：使用分发参数调用。
• options：监听器选项；布尔值可作为prepend的简写（true等价于{ prepend: true }）。
• 返回值：一个用于移除监听器的资源释放函数（disposer）；调用时若监听器仍处于注册状态则返回true，否则返回undefined。

关键特性是fiber 所有权与自动清理：监听器作为 effect 注册在当前 fiber 上，fiber 卸载（unload）时会自动移除所有监听器，避免插件热重载或停用后留下悬挂回调。从实现看（vendor/cordis/src/events.ts），register()通过ctx.fiber.effect()注册 effect，effect 清理时调用unregister()；若 fiber 已销毁，on()会抛出CordisError('INACTIVE_EFFECT')。此外，监听器在注册时会经过ctx.reflect.bind(listener)绑定，并通过internal/listener事件进行拦截校验（见下文）。

ctx.once(name, listener, options?)

once<K extends keyof Events>(name: K, listener: Events[K], options?: boolean | EventOptions): () => boolean

与on()相同，但监听器在首次调用后自行注销——最多被调用一次。实现上，once()内部先通过on()注册一个包装函数，包装函数第一件事就是调用 disposer 完成自移除，然后再转发给原始监听器（vendor/cordis/src/events.ts）。

适合一次性监听，例如等待某个初始化事件、某个异步任务完成通知。

EventOptions：prepend 与 global

ctx.on()和ctx.once()接受的选项接口定义如下（vendor/cordis/src/events.ts）：

interface EventOptions { /** 把监听器加到同一事件的现有监听器之前。 */ prepend?: boolean /** 无论 context 过滤检查结果如何都接收事件。 */ global?: boolean }

• prepend：控制监听器在队列中的位置。默认push到末尾（后注册的后执行）；为true时unshift到头部（先执行）。由于serial/bail/waterfall的执行顺序敏感，prepend常用于“必须最先/最外层介入”的拦截器。
• global：绕过 context 过滤。事件总线在分发时会根据thisArg携带的 context 过滤器（Context.filter）筛选监听器，只调用与当前 context 匹配的监听器；global: true的监听器无视过滤、总是被调用。这在 vendor/cordis/src/events.ts 的dispatch()中体现：

dispatch(type: string, args: any[]) { const thisArg = typeof args[0] === 'object' || typeof args[0] === 'function' ? args.shift() : null const name: string = args.shift() if (!name.startsWith('internal/')) { this.emit('internal/dispatch', type, name, args, thisArg) } const filter = thisArg?.[Context.filter] return (this._hooks[name] || []) .filter(hook => hook.global || !filter || filter.call(thisArg, hook.ctx)) .map(hook => hook.callback.bind(thisArg)) }

从这段实现还可以读出两个细节：一是internal/*事件不触发internal/dispatch诊断（避免递归）；二是thisArg为对象或函数时会被识别为显式this（同时也用于过滤）。

内置事件：Events接口与internal/*生命周期钩子

除了各子系统声明的事件，事件总线的核心还定义了一批内置框架事件，全部声明在 vendor/cordis/src/events.ts 的Events接口中：

其中两个值得注意：

• internal/listener是bail模式：on()在注册前会先bail('internal/listener', ...)，若返回非空结果，则直接用该结果取代默认注册流程（vendor/cordis/src/events.ts）。事件服务自身正是用它把internal/update监听器存储到 fiber 专用列表里。
• internal/update是waterfall模式，事件服务在其构造函数中注册了一个{ global: true, prepend: true }的监听器来按顺序驱动配置更新回调链（vendor/cordis/src/events.ts）。

在 DeepSeek Harness 中的实战运用

事件系统的完整图景是：API 由 Cordis 核心提供（本页），事件声明按子系统分布在各自的子系统页面。因此，在 Harness 中写监听器通常不需要发明新事件，而是“挂接”既有扩展点。以下是从 docs/subsystems/core.md 摘取的典型示例：

• agent 生命周期（emit）：agent/created、agent/disposed、agent/error、agent/status、agent/session-start——用于观测 Agent 生命周期，实现状态面板、监控、清理等。
• agent 流水线（waterfall）：agent/pre-step（改写进入 step 的消息）、agent/request（替换模型调用配置）、agent/request-error（改写错误处理）——这是对 Agent 循环做“横切”扩展的主要通道。
• agent 决策（serial）：agent/turn-stopping——串行询问各监听器是否应在无工具续接时终止回合。

事件分类的全局视图见 docs/architecture.md 的“Events”部分；各子系统声明的完整事件清单（含参数类型与分发模式）见 docs/subsystems/core.md、docs/subsystems/tools.md 等子系统页面。这些页面由 scripts/gen-cordis-catalog.ts 从源码声明自动生成，确保文档与实现不脱节。

结语：按语义选择分发模式

掌握了五种分发模式与EventOptions的prepend/global语义，再对照 docs/cordis-api/events.md 中的类型签名与各子系统页面的@mode标注，你就能写出与 Harness 事件契约完全一致、可被自动清理、可被 scope 过滤的插件代码。

• 人工智能
• AI Agent
• Agent 框架
• DeepSeek