DeepSeek Harness: 事件系统
最后更新：2026-08-31
事件是 Cordis 插件间松耦合通信的核心机制——插件不直接调用彼此，而是通过事件广播和订阅协作。四种事件模式覆盖了从简单通知到复杂管道的全部场景。
💡 提示：选择事件模式的关键问题是"需不需要拦截"——纯通知用 emit，可中断用 bail，串行处理用 serial，链式传递用 waterfall。
📋 前置知识：已完成 14-inject.md，理解依赖注入
1. 你将学到
- emit：普通事件广播
- bail：可中断事件
- serial：串行事件
- waterfall：瀑布式事件
- 事件域：session/agent/capability
- 自定义事件与类型安全
2. emit：普通事件广播
(1) 基本用法
emit 是最简单的事件模式——广播通知，所有监听器都会收到，不关心返回值：
TYPESCRIPT
// 发射事件
ctx.emit('session/created', { id: 'abc-123', user: 'Alice' })
// 监听事件
ctx.on('session/created', (data) => {
ctx.logger.info(`new session: ${data.id}`)
})
(2) 特点
| 特点 | 说明 |
|---|---|
| 广播 | 所有监听器都会被调用 |
| 无返回值 | 不关心监听器的返回值 |
| 无中断 | 监听器不能阻止后续监听器执行 |
| 并行 | 监听器按注册顺序执行 |
(3) 典型场景
- 通知类：
session/created、plugin/loaded - 日志类：
tool/executed、llm/request - 统计类：
request/completed、error/occurred
(4) ▶ 示例 4
TYPESCRIPT
ctx.on('session/created', (data) => {
ctx.logger.info(`logger: ${data.id}`)
})
ctx.on('session/created', (data) => {
ctx.metrics.inc('session_count')
})
ctx.on('session/created', (data) => {
ctx.cache.set(`session:${data.id}`, data)
})
ctx.emit('session/created', { id: 'abc', user: 'Alice' })
// 三个监听器都会执行
3. bail：可中断事件
(1) 基本用法
bail 是可中断事件——监听器返回非 undefined 值时，后续监听器不再执行：
TYPESCRIPT
// 监听器可以"拦截"事件
ctx.on('tool/beforeExecute', (data) => {
if (data.tool === 'shell' && data.params.command.includes('rm')) {
return { denied: true, reason: 'dangerous command' }
}
})
ctx.bail('tool/beforeExecute', { tool: 'shell', params: { command: 'rm -rf /' } })
// → 返回 { denied: true, reason: 'dangerous command' }
// 后续监听器不会执行
(2) 特点
| 特点 | 说明 |
|---|---|
| 可中断 | 返回非 undefined 值即中断 |
| 短路 | 第一个返回值的监听器终止传播 |
| 有返回值 | bail 调用返回拦截值 |
| 顺序敏感 | 先注册的监听器优先拦截 |
(3) 典型场景
- 权限检查：
tool/beforeExecute（拒绝危险操作） - 内容过滤：
message/beforeSend（过滤敏感内容） - 条件跳过：
task/beforeRun（跳过不适用的任务）
(4) ▶ 示例 4
TYPESCRIPT
// 审批策略插件
ctx.on('tool/beforeExecute', (data) => {
const policy = getApprovalPolicy(data.tool)
if (policy === 'deny') {
return { denied: true, reason: `${data.tool} is denied by policy` }
}
if (policy === 'ask') {
return { pending: true, requiresApproval: true }
}
// 返回 undefined → 不拦截，继续传播
})
// 沙箱插件（在审批之后）
ctx.on('tool/beforeExecute', (data) => {
if (!isInSandbox(data.params.cwd)) {
return { denied: true, reason: 'execution outside sandbox' }
}
})
(5) ▶ 示例 5
TYPESCRIPT
const result = ctx.bail('tool/beforeExecute', data)
if (result) {
// 被拦截
ctx.logger.warn('tool execution denied:', result.reason)
} else {
// 未被拦截，可以执行
await executeTool(data)
}
4. serial：串行事件
(1) 基本用法
serial 按顺序执行异步监听器，每个监听器等前一个完成后再执行：
TYPESCRIPT
ctx.on('session/initialized', async (data) => {
await loadUserPreferences(data.userId)
})
ctx.on('session/initialized', async (data) => {
await setupWorkspace(data.workspaceId)
})
ctx.on('session/initialized', async (data) => {
await warmCache(data.projectPath)
})
// 三个监听器串行执行
await ctx.serial('session/initialized', { userId: 'alice', workspaceId: 'ws-1' })
(2) 特点
| 特点 | 说明 |
|---|---|
| 串行 | 监听器按顺序逐个执行 |
| 异步 | 支持 async 监听器 |
| 等待完成 | serial 调用等待所有监听器完成 |
| 不中断 | 监听器不能阻止后续执行 |
(3) 典型场景
- 初始化流程：
session/initialized（按顺序加载配置、建立连接、预热缓存） - 清理流程：
session/closing（按顺序保存数据、断开连接、清理临时文件） - 数据管道：
data/transform（按步骤转换数据）
(4) 与 emit 的区别
TYPESCRIPT
// emit: 并行（不等待）
ctx.emit('session/created', data) // 不等监听器完成
// serial: 串行（等待）
await ctx.serial('session/created', data) // 等所有监听器完成
5. waterfall：瀑布式事件
(1) 基本用法
waterfall 将前一个监听器的返回值传给下一个监听器，形成链式传递：
TYPESCRIPT
ctx.on('message/format', async (data, next) => {
data.text = data.text.trim()
return next(data)
})
ctx.on('message/format', async (data, next) => {
data.text = data.text.replace(/\s+/g, ' ')
return next(data)
})
ctx.on('message/format', async (data, next) => {
data.text = data.text.substring(0, 4096)
return next(data)
})
const result = await ctx.waterfall('message/format', { text: ' hello world ' })
// result.text === 'hello world' (trim → collapse → truncate)
(2) 特点
| 特点 | 说明 |
|---|---|
| 链式传递 | 前一个的输出是后一个的输入 |
| next() 调用 | 监听器必须调用 next() 传递给下一个 |
| 可修改 | 每个监听器可以修改数据 |
| 可中断 | 不调用 next() 即中断链 |
(3) next() 函数
每个 waterfall 监听器接收 next 参数：
TYPESCRIPT
ctx.on('event/name', async (data, next) => {
// 修改 data
data.field = newValue
// 调用 next 传递给下一个监听器
return next(data)
// 不调用 next → 中断链，data 不再传递
})
(4) 典型场景
- 消息处理：
message/format（格式化 → 过滤 → 截断） - 请求管道：
request/process（认证 → 授权 → 处理 → 日志） - 数据转换：
data/transform（解析 → 校验 → 标准化 → 输出）
(5) 中断链
TYPESCRIPT
ctx.on('request/process', async (data, next) => {
if (!data.authenticated) {
return { error: 'unauthenticated' } // 不调用 next，链中断
}
return next(data)
})
6. 事件域
(1) 域的划分
Cordis 事件按域划分，用 / 分隔：
TEXT
📖 仅展示
session/created → session 域
session/destroyed → session 域
tool/beforeExecute → tool 域（capability 子域）
tool/afterExecute → tool 域
agent/initialized → agent 域
llm/request → llm 域
(2) 核心事件域
| 域 | 前缀 | 典型事件 |
|---|---|---|
| session | session/ |
created, destroyed, forked |
| agent | agent/ |
initialized, stopped, error |
| tool | tool/ |
beforeExecute, afterExecute, error |
| llm | llm/ |
request, response, stream, error |
| fiber | fiber/ |
created, active, disposing, disposed, errored |
| config | config/ |
updated, validated |
(3) 域的作用
事件域不是语法糖——框架根据域做优化：
- 事件过滤：只订阅特定域的事件
- 作用域隔离：session 域的事件在会话上下文中传播
- 审计分组：按域收集事件日志
(4) 订阅特定域
TYPESCRIPT
// 订阅 tool 域的所有事件
ctx.on('tool/*', (eventName, data) => {
ctx.logger.info(`tool event: ${eventName}`)
})
7. 自定义事件与类型安全
(1) 声明自定义事件
TYPESCRIPT
// events.ts
interface MyPluginEvents {
'my-plugin/data-loaded': { source: string; count: number }
'my-plugin/data-error': { source: string; error: Error }
}
declare module '@deepseek-ai/cordis' {
interface Events extends MyPluginEvents {}
}
(2) 类型安全的事件发射
TYPESCRIPT
ctx.emit('my-plugin/data-loaded', { source: 'api', count: 42 }) // ✅ 类型正确
ctx.emit('my-plugin/data-loaded', { wrong: true }) // ❌ 类型错误
(3) 类型安全的监听
TYPESCRIPT
ctx.on('my-plugin/data-loaded', (data) => {
// data 自动推断为 { source: string; count: number }
ctx.logger.info(`loaded ${data.count} items from ${data.source}`)
})
(4) 事件类型定义模式
TYPESCRIPT
// 插件内部事件
interface InternalEvents {
'cache/hit': { key: string; age: number }
'cache/miss': { key: string }
'cache/evicted': { key: string; reason: string }
}
// 扩展全局事件接口
declare module '@deepseek-ai/cordis' {
interface Events extends InternalEvents {}
}
// 导出供其他插件使用
export type CacheEvents = InternalEvents
(5) 事件命名规范
TEXT
📖 仅展示
{domain}/{verb-past-tense} ✅ session/created
{domain}/{verb-present} ✅ tool/execute (进行中)
{domain}/before{Action} ✅ tool/beforeExecute (前置)
{domain}/after{Action} ✅ tool/afterExecute (后置)
{domain}/{noun}-{state} ✅ fiber/errored (状态)
❓ 常见问题
Q emit 和 bail 能用同一个事件名吗？
A 不建议。虽然技术上可以，但 emit 的监听器和 bail 的监听器混在一起会导致混乱。建议用不同事件名区分。
Q waterfall 中的 next() 忘记调用会怎样？
A 链会中断，后续监听器不执行。waterfall 调用返回当前数据。这可能是故意的（条件中断），也可能是 bug（忘记调用）。
Q 事件监听器可以注册多次吗？
A 可以。同一个监听器函数注册多次会被调用多次。使用
ctx.off() 移除时需要引用同一个函数。Q 事件监听器的执行顺序可以控制吗？
A 默认按注册顺序。某些框架支持优先级参数，但 Cordis 当前按 FIFO 顺序执行。
Q 异步监听器在 emit 中会被等待吗？
A 不会。emit 不等待异步监听器完成。如果需要等待，使用 serial。
Q 如何查看所有已注册的事件监听器？
A
typescript ctx.logger.info('listeners:', ctx.listenerCount('tool/beforeExecute')) ---📖 小节
- 四种事件模式：emit（广播）、bail（可中断）、serial（串行异步）、waterfall（链式传递）
- emit 适合通知，bail 适合拦截，serial 适合有序初始化，waterfall 适合数据管道
- 事件域按
/分隔：session/agent/tool/llm/fiber/config - 自定义事件通过
declare module扩展 Events 接口实现类型安全 - waterfall 的 next() 调用是传递关键，忘记调用会中断链
📝 作业
1. ⭐ 基础题：编写一个插件，用 emit 广播 my-plugin/loaded 事件，在另一个插件中监听并输出日志。
2. ⭐⭐ 进阶题：实现一个审批拦截器，用 bail 在 tool/beforeExecute 事件中拦截包含 rm 的 shell 命令。测试：执行 ls 正常，执行 rm -rf / 被拦截。
3. ⭐⭐⭐ 挑战题：用 waterfall 实现一个消息处理管道：trim → 去除敏感词 → 截断超长文本。每个步骤是一个独立的监听器，中间步骤可以修改数据，最后一步返回最终结果。编写测试验证管道行为。
