利用事件钩子实现请求审计与结果改写

约 35 分钟

使用 emit/waterfall/serial 等事件分发机制，编写能够拦截与过滤大模型输入输出的钩子插件。

核心概念

事件分发模式与生命周期钩子

上一课我们通过 ctx.tools.register 注册了文档解析与分块工具，它更像一个被动的服务台：只有在智能体主动调用时，才会执行路径校验、体积限制与段落切分，属于“被请求才响应”的调用模式。而一个健壮的插件还需要“主动感知”的能力——当智能体启动、消息流转或工具被调用时，都能在恰当的时机观察、审计甚至改写行为。这正是 dsh 事件系统要承接的下一环。可以把这套机制想象成多道安检通道与广播电台的组合：广播电台向所有人广播通知，听众听完各自行动，互不干扰；而安检通道则是串联的流水线，包裹必须逐一通过扫描仪、开箱员和复核员，任何一环发现违禁品都能立即拦截扣留，或者贴上安全标签后再放行给下一位。在 DeepSeek Harness (dsh) 中，事件系统（Event System）正是基于 Cordis 内核构建的松耦合通信与拦截基础设施。它允许插件在不直接依赖具体实现的前提下发布和订阅状态变化。Cordis 提供了五种核心分发契约：用于同步单向广播的 emit、并行异步执行的 parallel、按注册顺序等待并短路的 serial 与同步版 bail，以及支持洋葱模型管道改写与短路拦截的 waterfall。通过在插件中结合 ctx.on 注册监听器，你可以像上一课为工具加上安全边界一样，透明地感知智能体生命周期、审计敏感参数并改写模型输入输出。

深入理解 Cordis 事件契约与类型合并

在开发安全审计与数据改写插件时，选择正确的事件分发模式至关重要。每种模式在 Harness 框架中对应明确的控制语义：

分发模式调用方式异步等待执行顺序与语义
emitctx.emit(name, ...args)否同步广播给所有监听器，忽略返回值。用于通知性事件。
parallelawait ctx.parallel(name, ...args)是并发执行所有异步监听器并等待全部完成。
serialawait ctx.serial(name, ...args)是按顺序串行等待，首个返回非空值（非 null/false/undefined）即终止后续执行。
bailctx.bail(name, ...args)否serial 的同步版本，遇有效返回值立即短路退出。
waterfallawait ctx.waterfall(name, ...args, next)是环绕中间件管道（Around-middleware）。每个监听器接收参数与 next() 回调，可改写返回值或直接短路拦截。

类型化事件（Typed Events）

Harness 推荐使用 TypeScript 声明合并（Declaration Merging）来维持类型安全。通过对 @deepseek-ai/cordis 模块的 Events 接口进行扩展，开发者的 ctx.on 和 ctx.emit 即可获得参数与返回值的自动推导：

typescript
import '@deepseek-ai/cordis'

declare module '@deepseek-ai/cordis' {
  interface Events {
    'audit/tool-call': (payload: { tool: string; args: Record<string, any> }) => void
    'audit/transform': (input: string, next: () => Promise<string>) => Promise<string>
  }
}

所有通过 ctx.on('event-name', handler) 注册的监听器均属于可逆效果（Reversible Effect），当插件被卸载或热重载时，监听器会自动从总线中注销，无需手动清理。

编写审计拦截与参数改写插件

  1. 1

    在项目目录 src/security-audit.ts 中声明类型化事件，定义工具调用前置校验和文本脱敏管道的签名接口。

    typescript
    import type { Context } from '@deepseek-ai/cordis'

declare module '@deepseek-ai/cordis' {
  interface Events {
    'security/validate': (toolName: string, args: Record<string, any>) => boolean | undefined
    'security/sanitize': (input: string, next: () => Promise<string>) => Promise<string>
  }
}
  2. 2

    编写插件主体逻辑，导出 name、inject 与 apply 函数。通过 ctx.on 注册 security/validate（利用 serial/bail 语义进行白名单检查）与 security/sanitize（利用 waterfall 进行敏感词脱敏替换）。

    typescript
    import type { Context } from '@deepseek-ai/cordis'

export const name = 'security-audit'
export const inject = ['tools']

export function apply(ctx: Context) {
  const sensitivePatterns = [/password=[^&\s]+/gi, /sk-[a-zA-Z0-9]{20,}/gi]

  // 1. 阻断违规工具调用（Bail / Serial 校验钩子）
  ctx.on('security/validate', (toolName, args) => {
    if (toolName === 'execute_system_command' && args.command?.includes('rm -rf')) {
      console.warn(`[Security] Blocked dangerous command in ${toolName}`)
      return false // 返回非空/false 触发拦截短路
    }
    return undefined // 放行
  })

  // 2. 敏感数据脱敏改写（Waterfall 管道）
  ctx.on('security/sanitize', async (input, next) => {
    // 先执行下游转换
    const downstream = await next()
    let sanitized = downstream
    for (const pattern of sensitivePatterns) {
      sanitized = sanitized.replace(pattern, '***REDACTED***')
    }
    console.log(`[Audit] Sanitized output from ${downstream.length} to ${sanitized.length} chars`)
    return sanitized
  })

  // 3. 监听全局工具执行结果事件记录日志
  ctx.on('tools/result', (result: any) => {
    console.log(`[Audit Log] Tool completed at ${new Date().toISOString()}`)
  })
}
  3. 3

    在 test-hooks.ts 中构建测试套件，分发 security/validate 与 security/sanitize 事件，验证拦截与文本改写行为。

    typescript
    import { Context } from '@deepseek-ai/cordis'
import * as SecurityAudit from './src/security-audit.js'

async function main() {
  const ctx = new Context()
  ctx.plugin(SecurityAudit)
  await ctx.start()

  // 测试 1: 校验拦截
  const isForbiddenBlocked = ctx.bail('security/validate', 'execute_system_command', { command: 'rm -rf /' })
  console.log('Test Block Result:', isForbiddenBlocked === false ? 'BLOCKED_SUCCESS' : 'FAILED')

  // 测试 2: Waterfall 脱敏改写
  const output = await ctx.waterfall(
    'security/sanitize',
    'Bearer sk-1234567890abcdef1234567890',
    async () => 'Bearer sk-1234567890abcdef1234567890'
  )
  console.log('Sanitized Output:', output)
}

main().catch(console.error)
  4. 4

    执行 Node.js 测试脚本，观察终端输出的阻断标记与脱敏结果。

    bash
    node --loader ts-node/esm test-hooks.ts

你应该看到

控制台打印 Test Block Result: BLOCKED_SUCCESS，且 Sanitized Output 中包含 Bearer ***REDACTED***，证明校验拦截与瀑布流改写均生效。

遇到问题：在 waterfall 事件管道中，下游插件未被触发，且上层请求长时间处于挂起状态或未返回预期结果。

可能原因

  • 某个 upstream waterfall 监听器函数中未执行 await next() 且未直接 return 终态值。
  • 异步监听器中 next() 抛出异常但未被捕获，导致 Promise 链异常中断。
  • 插件加载阶段声明的依赖服务缺失，导致整个 Fiber 处于 PENDING 状态，监听器未实际挂载。

排查步骤

  1. 检查所有 ctx.on('event', async (input, next) => { ... }) 监听器，确保正常放行分支均包含 return await next() 或处理了 next() 的返回值。
  2. 为异步中间件逻辑添加 try...catch 保护块，确保异常时能够优雅降级或向外传递可识别的错误类型。
  3. 检查插件的 export const inject = [...] 声明，确认被依赖的内置服务已成功就绪并处于 ACTIVE 状态。
自检：在 Harness 的 Cordis 事件体系中，为什么请求改写和鉴权网关通常使用 waterfall 或 bail/serial 模式，而不是最基础的 emit 模式？

因为 emit 模式属于单向同步广播，不等待异步 Promise 也完全忽略监听器的返回值，监听器无法将修改后的数据传递给下游，也无法中止后续处理流程。而 waterfall 提供了 Around-middleware 机制，允许通过 await next() 拦截并修改数据流向；bail/serial 则能根据监听器返回的非空判定值立刻短路退出，从而实现阻断非法请求的安全守卫能力。