自学教程

DeepSeek Harness 事件系统

DeepSeek Harness

DeepSeek Harness 事件系统（Cordis事件总线）

事件系统是 Cordis 微内核实现插件松耦合扩展的核心基础设施。Harness 中智能体生命周期、工具执行钩子、安全拦截校验、提示词流水线、会话轨迹流转全部构建在事件总线之上。

不同于简单的发布‑订阅模型，Cordis 提供五种不同语义的事件分发模式，分别适配广播通知、短路拦截、顺序编排、洋葱中间件流水线、并发扇出等场景。同时支持 TypeScript 声明合并实现类型安全，并且所有通过 ctx.on 注册的监听器都会跟随插件生命周期自动销毁，避免内存泄漏。

一、基础用法：监听与触发事件

插件在 apply(ctx)内部使用 ctx.on() 监听事件，使用对应的分发函数触发事件。通过 ctx.on 注册的监听器属于插件副作用，插件卸载、热重载时会被内核自动移除，不需要手动写注销逻辑。

import type { Context } from '@deepseek‑ai/cordis';
export const name = 'demo‑event';
export function apply(ctx: Context) {
  // 监听会话启动事件
  ctx.on('session/start', (session) => {
    ctx.logger.info('新会话启动', session.id);
  });
  // 触发广播事件
  ctx.emit('worker/ready', { workerId: 'worker‑01' });
}

事件命名遵循官方约定 namespace/action，例如 agent/step、tools/before‑run、security/validate‑command，便于区分子系统。

二、五大事件分发模式

Cordis 提供 5 套分发契约，不同模式返回值、异步行为、短路规则完全不同，需要根据业务场景选择正确模式。

模式API行为说明典型业务场景
广播ctx.emit同步执行所有监听器，忽略返回值，不等待异步 Promise日志打印、状态通知、埋点上报
短路拦截ctx.bail顺序执行，第一个返回非 null/false/undefined 的值直接作为结果并终止后续安全校验、一票否决权限拦截
顺序异步执行ctx.serial按注册顺序 await 每一个监听器；非空返回提前终止系统分阶段初始化、迁移、前置检查
洋葱流水线ctx.waterfall中间件模式；必须调用 next()，不调用直接截断链路提示词转换、请求包装、缓存网关
并发扇出ctx.parallel并发全部监听器，等待全部 Promise 完成，不短路并行执行多个后置统计、无关的后置任务

1. emit 广播

只做通知，监听器返回值全部丢弃；不会等待监听器内部异步操作。适合只通知不处理返回结果的场景。

ctx.emit('worker/ready', { workerId: 'worker‑01' });
ctx.on('worker/ready', (payload) => {
  ctx.logger.info(`worker ${payload.workerId} 就绪`);
});

2. bail 短路拦截（安全校验常用）

按注册顺序执行监听器；只要某个监听器返回非空真值，立刻停止后续监听器，返回该结果。返回null / undefined / false 代表放行，继续执行下一个监听器。

// 安全策略拦截示例
ctx.on('security/validate‑command', (command) => {
  if(command.includes('rm -rf /')) {
    return '检测到高危毁灭性指令，已被安全策略拦截';
  }
});const blockReason = ctx.bail('security/validate‑command', userInputCommand);
if(blockReason) throw new Error(blockReason);

3. serial 顺序异步执行

依次 await 每个异步监听器；得到非空返回值就提前终止链路。适合分阶段启动、多步前置检查流程。

await ctx.serial('system/bootstrap‑phase', runtimeCtx);

4. waterfall 洋葱流水线（中间件）

每一个监听器必须显式调用 await next() 将控制权交给下游；如果不调用 next，流水线直接被截断短路，这是故意设计，用于实现缓存命中、熔断拦截等逻辑。

// 修改提示词：追加时间戳
ctx.on('prompt/transform', async (prompt, next) => {
  const downstream = await next();
  return `${downstream}\n[Timestamp:${Date.now()}]`;
});// 触发流水线处理原始prompt
const finalPrompt = await ctx.waterfall('prompt/transform', rawPrompt, async () => rawPrompt);

⚠️ 重要警告：waterfall 监听器忘记调用 next()，后续全部中间件不会执行，直接截断处理链。

5. parallel 并发扇出

全部监听器并发执行，等待所有异步完成，不会短路。适合多个互不依赖的后置统计、日志上报任务。和 emit 的区别：parallel 会等待 Promise，emit 完全不等待异步逻辑。

三、类型安全事件（TypeScript声明合并）

Cordis 支持通过模块声明合并扩展全局 Events 接口，实现事件的参数、返回值类型推导，编写代码获得自动补全、类型校验。

import type { Context } from '@deepseek‑ai/cordis';// 扩展Cordis全局事件类型
declare module '@deepseek‑ai/cordis' {
  interface Events {
    // 普通广播事件
    'trajectory/step'(step:{index:number;action:string}):void;
    // bail短路事件，返回string代表拦截，void放行
    'security/check'(input:string):string | void;
    // waterfall流水线事件
    'prompt/transform'(input:string, next:()=>Promise<string>):Promise<string>;
  }
}
export function apply(ctx: Context) {
  // 这里 step 对象会获得完整类型提示
  ctx.on('trajectory/step', (step) => {
    ctx.logger.info(步骤#${step.index}:${step.action});
  });
}

四、区分两类事件：Cordis运行时事件 vs 会话持久化事件

这里极易混淆，官方明确区分两套事件体系：

  • Cordis运行时事件：例如 agent/step、tools/result。内存实时分发，直接使用 ctx.on('xxx',handler) 监听；不会自动写入会话日志。
  • 会话持久化事件：turn/*、step/*、tool/call、tool/result、compaction/*。追加写入会话只读日志，用于回放、轨迹展示；不能直接监听该类型字符串，需要监听 session/event，再判断 payload 内部的 event.type 字段。
// 监听持久化会话事件的正确写法
ctx.on('session/event', (evt) => {
  if(evt.type === 'tool/call') {
    ctx.logger.info('工具被调用', evt.data);
  }
});

会话本身就是一份只追加的事件日志，是回放与轨迹的唯一真相源。

五、监听器与副作用生命周期

  • 在 apply 内部通过ctx.on() 注册的监听器，自动作为插件副作用；插件卸载 / 热重载，内核自动解绑全部监听器，不会内存泄漏。
  • 不要在 apply 外部、全局作用域注册事件，脱离 ctx 托管将无法自动清理。
  • 可以接收 ctx.on 返回的 disposer 函数，运行期手动解绑单个监听器。
  • 监听器内部抛出异常会被内核隔离捕获，打印日志，不会打断事件总线其他监听器执行。

六、实战示例：简单工具日志插件

监听工具执行结果，打印调用参数与返回内容，完整可直接运行的插件片段：

import type { Context } from '@deepseek‑ai/cordis';
import '@deepseek‑ai/dsh‑tools';export const name = 'tool‑logger';
export function apply(ctx: Context) {
  ctx.on('tools/result', (exec, result) => {
    ctx.logger.info([tool] ${exec.name}(${JSON.stringify(exec.arguments)}));
    const textOut = result.content
      .filter(b => b.type === 'text')
      .map(b => b.text)
      .join('');
    ctx.logger.info([tool‑result] ${textOut.slice(0,120)});
  });
}

七、常见问题FAQ

Q：emit 和 parallel 的区别？

A：emit 同步触发，完全忽略 Promise，不会等待异步回调；parallel 等待所有监听器的 Promise 全部完成。

Q：waterfall忘记调用 next()会发生什么？

A：流水线直接短路截断，下游所有中间件不会运行；这是用于缓存、熔断的故意设计，写中间件务必不要漏掉 next()调用。

Q：监听器抛异常会不会导致整个事件停止？

A：不会，每个监听器异常被隔离捕获并输出日志，其他监听器继续正常执行。

Q：如何手动解绑事件？

A：保存 ctx.on 返回的 disposer 函数，调用即可手动解绑；插件卸载时无需手动处理，内核自动清理。

Q：我想监听 tool/call，直接写 ctx.on(‘tool/call’) 为什么收不到？

A：tool/call 属于会话持久化事件，不是Cordis运行时事件；需要监听 session/event 再判断 evt.type 字段。

八、开发最佳实践

  1. 所有事件监听写在 apply(ctx) 函数内部，交给插件副作用托管，禁止全局注册监听器。
  2. 根据业务语义选择分发模式：日志通知用 emit；安全拦截用 bail；分阶段启动用 serial；中间件改写用 waterfall；并行统计用 parallel。
  3. 区分Cordis运行时事件与session持久化事件，不要直接监听 turn/step/tool/call 这类持久化事件名字符串。
  4. 开发自定义事件，补充 TypeScript 声明合并，获得类型安全。
  5. waterfall 中间件必须保证调用 next()，除非业务目标就是主动截断链路。

本篇小结

事件系统是 Cordis 微内核的消息中枢。五种分发模式分别适配通知、拦截、顺序编排、中间件流水线、并发扇出；配合类型安全声明合并，以及监听器跟随插件生命周期自动销毁的副作用机制，实现插件之间完全解耦。

开发时必须分清两套事件：内存实时分发的 Cordis 运行时事件，和写入会话日志的持久化会话事件。理解事件系统，就掌握了Harness实现拦截、扩展Agent行为的最核心手段。

标签：

0 条笔记