DeepSeek Harness 能“在运行中改写自己“，全靠这个叫 Cordis 的插件框架（附完整源码解析）

qq_56972807

898人浏览 · 2026-08-16 22:47:42

qq_56972807 · 2026-08-16 22:47:42 发布

Cordis 深入解析：DeepSeek Harness 背后的插件框架（核心概念 → 源码原理 → 生命周期 → 实战，一篇全看懂）

本文基于 @deepseek-ai/cordis v4.0.1 源码逐行研读，配合 DeepSeek Harness 官方文档与 dsh-tools 真实源码，是全网少有的 Cordis 中文深度解析。

你将依次看到：五个核心概念 → Context Proxy 与依赖注入源码拆解 → Fiber 生命周期与 effect 副作用系统 → 事件分发 5 模式与真实 DSH 插件实战。

引言：为什么 DeepSeek Harness 值得研究它的底座

2026 年，DeepSeek Harness 成为 AI Agent 领域最受关注的开源运行时之一。论文《让 Agent 在运行中改写自己》让它火出圈——但多数人只看到了 Agent 能力本身，忽略了支撑这一切的插件框架。

Harness 的架构核心是一个名为 Cordis 的元框架：它不是 LLM、不是工具链、不是 UI，而是把"llm 服务、tools 服务、sessions 服务、agent 协调"这些子系统像乐高一样拼装起来的那层胶水。

Cordis 的作者是 Shigma（Koishi 机器人框架的作者），DSH 以 vendor 方式引入了它（@deepseek-ai/cordis，MIT 协议）。

📌 相关阅读：

官方入门：deepseek-harness/docs/cordis-primer.zh.md
论文解读：DeepSeek Harness背后的论文：让Agent在运行中改写自己

第一部分：核心概念与源码架构

一、Cordis 到底解决什么问题？

先看传统应用架构的三个痛点：

痛点	传统做法	Cordis 的做法
模块间依赖	手动 import / 手动初始化顺序	`ctx.<服务名>` 按 key 查找，`inject` 声明依赖，框架自动排序
功能开关	if/else 写死在代码里	插件即服务，`ctx.plugin()` 一行挂载/卸载
热更新	重启进程	Fiber 状态机 + effect 副作用自动撤销，reload 即换即生效

一句话总结：Cordis 让"应用 = 一组可插拔、可热重载、彼此通过依赖关系自动协调的服务"。

二、五个核心概念（官方原版提炼）

官方 primer 把 Cordis 浓缩成五句话，我逐条拆解：

1️⃣ 插件是实现 Service 的对象

插件有三种合法形态，Registry 会统一归一化为一个 callback（注册表身份键）：

// ① 函数插件：ctx 是上下文，config 是配置
function myPlugin(ctx: Context, config: MyConfig) {
  // 插件体：在这里注册服务、监听事件
}
myPlugin.inject = ['llm']        // 声明依赖：等 llm 服务就绪才启动
myPlugin.Config = MyConfigSchema // 配置校验（standard-schema 格式）

// ② 类插件（Service 子类）：生命周期由框架挂载/卸载
class MyService extends Service { ... }

// ③ 对象插件
const plugin = {
  name: 'my-plugin',
  Config: schema,
  inject: ['tools'],
  apply(ctx, config) { ... }
}

2️⃣ 上下文是服务的容器

每个服务占据一个稳定的 ctx.<key>：

ctx.llm       // LLM 服务
ctx.tools     // 工具流水线
ctx.sessions  // 会话管理
ctx.agents    // Agent 协调

关键设计：其他插件通过 key 查找服务，而不是 import 具体实现。 这就是依赖倒置——换一个 LLM 实现，所有依赖 ctx.llm 的插件无需改动。

3️⃣ 通过 `inject` 声明服务依赖

// 声明需要哪些服务
myPlugin.inject = ['llm', 'tools']

// 或者用对象形式，附带拦截配置
myPlugin.inject = {
  llm: { model: 'deepseek-v4' },  // 这个插件上下文里，llm 配置被合并覆写
}

框架会等所有依赖就绪后才启动插件；依赖消失自动卸载，依赖恢复自动重载。加载顺序完全由依赖图决定，不需要手动编排启动序列。

4️⃣ 类型化事件用于通信

服务通过 TypeScript 声明合并注册事件名，然后按 5 种模式分发（第四部分细讲）：

// 监听（自动随插件卸载）
ctx.on('message/send', (msg) => { ... })

// 分发
ctx.emit('message/send', msg)       // 通知，不等待
ctx.waterfall('internal/config', ...) // 中间件链，可拦截/替换
ctx.parallel('task/run', ...)       // 并行扇出
ctx.serial('task/query', ...)       // 按序直到有结果

5️⃣ 注册是可逆的副作用

提示词片段、工具 schema、监听器、服务……全部通过 ctx.effect() 或 ctx.on() 安装，reload / teardown 时自动按逆序撤销：

ctx.effect(() => {
  // 立即执行的注册逻辑
  registerSomething()

  // 返回 disposer：卸载时逆序执行
  return () => unregisterSomething()
})

这是整个框架最优雅的设计之一——没有内存泄漏，没有手动清理。

三、源码架构总览（v4.0.1，仅 9 个模块）

node_modules/@deepseek-ai/cordis/src/
├── context.ts    Context 类（Proxy 代理 + extend/isolate/intercept 作用域）
├── registry.ts   RegistryService（ctx.plugin/ctx.inject）、Plugin 类型、@Inject 装饰器
├── fiber.ts      Fiber 类（插件生命周期状态机 + effect 系统）★ 最核心
├── events.ts     EventsService（事件总线、5 种分发模式）、内置事件表
├── service.ts    Service 抽象基类（服务注册、可调用服务、配置合并）
├── reflect.ts    ReflectService（ctx.get/set/provide/mixin，服务解析）
├── logger.ts     结构化日志门面（Exporter 可插拔）
├── utils.ts      基础设施（DisposableList、traceable proxy、composeError）
└── index.ts      统一出口

依赖关系（谁 import 谁）：

context ──→ registry / events / reflect / logger / fiber
registry ──→ fiber
fiber ──→ context / reflect
reflect ──→ context / fiber / utils

四、内置服务的职责划分

new Context() 时安装了 4 个核心服务，它们的职责边界非常清晰：

服务	ctx 上暴露	职责
`RegistryService`	`ctx.plugin` / `ctx.inject`	插件注册、形态归一化、Fiber 创建
`EventsService`	`ctx.on` / `ctx.emit` / `ctx.waterfall` …	事件总线、5 种分发、监听器生命周期
`ReflectService`	`ctx.get` / `ctx.set` / `ctx.provide` / `ctx.mixin`	Context Proxy 的引擎：服务查找与注册
`LoggerService`	`ctx.logger(name)`	命名日志门面，Exporter 可插拔

注意 ReflectService 的构造器里有一段关键代码——核心服务的方法被 mixin 到了 ctx 上：

this.mixin('reflect', ['get', 'set', 'provide', 'accessor', 'mixin'])
this.mixin('fiber', ['runtime', 'effect'])
this.mixin('registry', ['inject', 'plugin'])
this.mixin('events', ['on', 'once', 'parallel', 'emit', 'serial', 'bail', 'waterfall'])

所以 ctx.on() 本质是 ctx.events.on()，ctx.plugin() 本质是 ctx.registry.plugin()——ctx 是一个聚合了所有核心能力的门面对象。

五、一个最小可运行的 Cordis 应用

光看概念不够，直接上代码。先装包：

npm i @deepseek-ai/cordis

然后写一个最小应用：

// app.ts
import { Context } from '@deepseek-ai/cordis'

// ① 创建根上下文（会自动装好 4 个核心服务）
const app = new Context()

// ② 定义一个服务插件：提供一个 greet 服务
const greetingService = {
  name: 'greeting',
  apply(ctx) {
    // 提供服务：ctx.greet 可用了
    ctx.provide('greet', (name: string) => `Hello, ${name}!`)
  },
}

// ③ 定义一个依赖该服务的插件
const consumer = {
  name: 'consumer',
  inject: ['greet'],              // 等 greet 就绪才启动
  apply(ctx) {
    // 这里可以安全使用 ctx.greet（依赖已满足）
    ctx.on('ready', () => {
      console.log((ctx.greet as any)('Cordis'))
    })
  },
}

// ④ 挂载插件
app.plugin(greetingService)
app.plugin(consumer)

// ⑤ 触发事件验证
app.emit('ready')

运行 npx tsx app.ts，输出：

Hello, Cordis!

这段代码演示了 4 个核心概念：插件形态、服务提供/消费、inject 依赖声明、事件通信。而"卸载时自动清理"体现在：ctx.provide 和 ctx.on 返回的都是 disposer，框架挂载/卸载自动调用。

六、Cordis 在 DSH 里实际怎么用

看真实的 DSH profile 配置（~/.dsh/profiles/web/）：

# cordis.yml（根，勿改）—— 空列表，整棵树靠 patch 组合
[]

# cordis.patch.yml（用户层，最后应用）—— 用户插件在这里注册
- insert:
    - id: skill-hub
      name: 'dsh-skills'

// package.json 里的 dsh.profile.bundles —— 组合顺序
{
  "dsh": {
    "profile": {
      "bundles": [
        "@deepseek-ai/dsh-base",   // 底座（llm/sessions/agents 等核心服务）
        "@deepseek-ai/dsh-web-app", // Web 界面
        "dsh-attachments",          // 附件插件
        "dsh-inspector",            // 检查器插件
        "dsh-skills"                // 技能插件
      ]
    }
  }
}

加载机制：每个 bundle 包内自带 cordis.patch.yml（insert: 声明插件 id + 包名），DSH 按 bundles 顺序组合成一棵插件树，用户层 patch 最后应用可覆盖/禁用/新增。!!js 表达式由 cordis-plugin-include 解析。

这就是为什么你 dsh plugin add xxx 装插件后，插件能自动出现在设置页——它本质是往这棵 Cordis 插件树里插了一个节点。

七、Cordis 生态全家桶（DSH 实际安装的包）

DSH 的 node_modules 里实际安装了这些 Cordis 相关包，看清它们的分工，你对整个生态就有完整地图了：

包名	职责
`@deepseek-ai/cordis`	核心框架（Context / Fiber / 事件 / 服务）
`cordis-plugin-loader`	插件加载器：维护 EntryTree，按模块名 import 插件、应用配置、同步插件图
`cordis-plugin-include`	文件型插件树：把 YAML 配置解析成 loader 条目（支持 `!!js` 表达式）
`cordis-plugin-hmr`	热模块替换：改插件源码自动卸载旧→加载新
`cordis-plugin-group`	插件分组管理
`cordis-plugin-timer`	定时任务集成

📌 Loader 的 Entry 模型（写配置前必懂）：

await root.plugin(Loader, { baseUrl: import.meta.url })
const id = await root.loader.create({
  name: './plugins/example',   // 模块说明符
  config: { enabled: true },   // 插件配置
})
root.loader.update(id, { config: { enabled: false } })  // 热更新
root.loader.remove(id)         // 卸载

DSH 官方文档导航（深入学习路径）

文档	内容
`docs/cordis-primer.zh.md`	五个核心概念速览（本文的官方依据）
`docs/user/develop/framework/index.zh.md`	插件模型 + Fiber 状态机 + 自动清理 + HMR
`docs/user/develop/framework/service.zh.md`	服务与依赖（Service 基类、类型声明）
`docs/user/develop/framework/events.zh.md`	事件系统 5 模式 + 类型安全事件
`docs/user/develop/subsystems/core.md`	生成的核心服务/事件参考（`cordis-surface` 区块）
`docs/cordis-tutorial/`	手把手教程（从零搭同一套生命周期/服务/事件）

第二部分：Context Proxy 与依赖注入源码拆解

八、一个灵魂问题

ctx.llm   // 就这么一行，服务就拿到了

如果你写过 Java Spring，会想到依赖注入容器；如果你写过 Node，会觉得这不过是个对象属性。但 Cordis 的做法两者都不是——它是一个 Proxy 代理的上下文对象，每次属性读取都会走一套完整的服务解析管线。

这一部分回答三个问题：

ctx 为什么是 Proxy？Proxy 的 get 陷阱里发生了什么？
ctx.provide('llm', impl) 注册的服务存在哪？怎么被找到的？
inject: ['llm'] 声明的依赖，为什么能"等就绪再启动"？

九、Context：一个自指 Proxy

看 src/context.ts 的构造函数（我加了注释）：

export class Context {
  constructor() {
    // 隔离映射：服务名 → 作用域标签
    this[symbols.isolate] = Object.create(null)
    // 拦截映射：服务名 → 该服务的插件配置覆写
    this[symbols.intercept] = Object.create(null)

    // ★ 关键：this 被包了一层 Proxy，返回的是代理对象
    const self = new Proxy<this>(this, ReflectService.handler)
    this.root = self           // root 指向代理后的自己
    this.baseUrl = undefined

    // 根 Fiber（uid=0，代表应用本身）
    this.fiber = new Fiber(self, {}, Object.create(null), null, () => [])

    // 安装 4 个核心服务（都装在代理对象 self 上）
    this.reflect = new ReflectService(self)
    this.registry = new RegistryService(self)
    this.events = new EventsService(self)
    this.logger = new LoggerService(self)

    this.fiber._disposables.clear()
    return self                // ★ 返回的是 Proxy！
  }
}

注意 return self —— new Context() 拿到的不是类实例本身，而是它的 Proxy。所有对 ctx 的属性读写都会先经过 ReflectService.handler 的 get/set 陷阱。

十、Proxy get 陷阱：服务解析管线

ReflectService.handler 的 get 是整个框架的心脏（src/reflect.ts 135-171 行）：

static handler: ProxyHandler<Context> = {
  get: (target, prop, ctx: Context) => {
    // ① 特殊属性（symbol、prototype、then、_ 开头、数字字符串）直接放行
    if (isSpecialProperty(prop)) {
      return Reflect.get(target, prop, ctx)
    }

    // ② 实例自有属性直接返回（如 ctx.fiber、ctx.events）
    if (Reflect.has(target, prop)) {
      return getTraceable(ctx, Reflect.get(target, prop, ctx))
    }

    // ③ 到这里：prop 不是内置属性 → 当成"服务名"解析
    const error = new Error(`cannot get property "${prop}" without inject`)

    try {
      const def = target.reflect.props[prop]
      // accessor（计算属性，如 mixin 出来的 ctx.on）走 accessor 分支
      if (def?.type === 'accessor') {
        return def.get.call(ctx, ctx[symbols.receiver], error)
      }

      // 根上下文（无 runtime）：直接查服务存储
      if (!ctx.fiber.runtime) return ctx.reflect.get(prop, false)

      // ★ 插件上下文：走 internal/get waterfall（可被拦截！）
      return ctx.events.waterfall('internal/get', ctx, prop, error, () => {
        // 真正的查找逻辑：沿着 fiber 链向上找
        const key = target[symbols.isolate][prop]
        let fiber = (ctx[symbols.shadow] ?? ctx).fiber
        while (true) {
          const impl = fiber.store?.[prop]      // 当前 fiber 有没有提供这个服务？
          if (impl) return getTraceable(ctx, impl.value)
          if (prop in fiber.inject) {
            // 声明了依赖但还没就绪 → 抛错
            error.message = `cannot get required service "${prop}" in inactive context`
            throw error
          }
          if (!fiber.runtime) throw error        // 到根了还没有 → 抛错
          if (fiber.parent[symbols.isolate][prop] !== key) throw error  // 跨了隔离作用域 → 抛错
          fiber = fiber.parent.fiber             // 沿父链继续找
        }
      })
    } catch (e: any) {
      throw e === error ? enhanceError(e) : e    // 美化错误栈（long stack trace）
    }
  },
  // set / has 陷阱类似，这里省略
}

这段代码的三个关键设计

设计 1：读服务 = 沿 Fiber 链向上查找

let fiber = currentFiber
while (true) {
  const impl = fiber.store?.[prop]   // 每个 Fiber 的 store 存它提供的服务
  if (impl) return impl.value
  fiber = fiber.parent.fiber         // 找不到就找父级
}

子上下文可以覆盖父级提供的同名服务（自己的 store 优先），这正是 isolate() 作用域隔离的基础。

设计 2：查找被包在 internal/get waterfall 里

这意味着——任何插件都能拦截"读取 ctx.llm"这个动作！

// 某个插件注册一个 internal/get 监听器
ctx.on('internal/get', (ctx, name, error, next) => {
  if (name === 'llm') {
    return myLlmProxy   // 偷梁换柱：所有 ctx.llm 读取都被替换
  }
  return next()         // 否则放行
})

这是"Agent 在运行中改写自己"的底层能力之一。

设计 3：错误即诊断

cannot get property "llm" without inject —— 你没声明依赖就乱读服务，报错信息直接告诉你该加 inject。配合 composeError 的 long stack trace，错误能一路追到插件调用点。

十一、服务存哪了？ReflectService 的 store

ctx.provide('llm', impl) 的实现（src/reflect.ts 277-305 行）：

provide(name: string, value?: any, check?: () => boolean) {
  // ★ provide 本身也是一个 effect！卸载自动注销
  return this.ctx.fiber.effect(() => {
    if (!this.props[name]) {
      this.props[name] ??= { type: 'service' }
    } else if (this.props[name].type !== 'service') {
      throw new Error(`property "${name}" is already declared as ${this.props[name].type}`)
    }

    // 给服务名分配一个隔离标签（root 上首次分配）
    this.ctx.root[symbols.isolate][name] ??= Symbol(name)
    const key = this.ctx[symbols.isolate][name]

    // ★ 存储：按隔离标签 key 存进 store
    const impl: Impl = { name, value, fiber: this.ctx.fiber, check }
    if (this.store[key]) {
      throw new Error(`service "${name}" has been registered at <${this.store[key].fiber.name}>`)
    }
    this.store[key] = impl
    this.ctx.fiber.store![name] = impl   // 同时挂到 fiber.store（供向上查找）

    // ★ 服务可用 → 唤醒所有依赖它的插件
    if (this.ctx.fiber.state === FiberState.ACTIVE) {
      this.notify([name])
    }

    // disposer：注销 + 再次唤醒依赖方
    return async () => {
      delete this.store[key]
      const fibers = this.notify([name])
      await Promise.allSettled(fibers.map(fiber => fiber.await()))
      delete this.ctx.fiber.store![name]
    }
  }, `ctx.provide(${JSON.stringify(name)})`)
}

数据结构全景

ReflectService.store         隔离标签(symbol) → Impl
ReflectService.props         服务名 → Property（service | accessor）
Fiber.store                  服务名 → Impl（每个 fiber 自己提供的）
Fiber.inject                 服务名 → 拦截配置（每个 fiber 依赖的）
Context[isolate]             服务名 → 隔离标签
Context[intercept]           服务名 → 配置覆写

Impl 是核心记录：

interface Impl {
  name: string        // 服务名
  value?: any         // 服务实现
  fiber: Fiber        // 提供方（拥有生命周期）
  check?: () => boolean  // 可用性谓词
}

十二、依赖注入的完整闭环：`inject` → 等待 → 启动

12.1 `ctx.plugin()` 创建 Fiber（`src/registry.ts` 316-336 行）

plugin(plugin: Plugin, config?: any, getOuterStack = buildOuterStack()) {
  // ① 归一化形态：函数 → 函数本身；{apply} 对象 → apply 方法
  const callback = this.resolve(plugin)
  if (!callback) throw new Error('invalid plugin, expect function or object with an "apply" method')

  // ② 找/建 Runtime 记录（同一插件的所有 Fiber 共享）
  let runtime = this._internal.get(callback)
  if (!runtime) {
    runtime = { name: plugin.name, callback, fibers: new DisposableList(), Config: plugin.Config }
    this._internal.set(callback, runtime)
  }

  // ③ 创建 Fiber（关键！传入解析好的 inject 映射）
  const fiber = new Fiber(this.ctx, config, Inject.resolve(plugin.inject), runtime, getOuterStack)

  // ④ 包装成 PromiseLike：await fiber 会等加载完成
  const wrapped = Object.create(fiber) as Fiber & PromiseLike<Fiber>
  wrapped.then = (onFulfilled, onRejected) => fiber.await().then(onFulfilled, onRejected)
  return wrapped
}

12.2 Fiber 构造：检查依赖 → 刷新 epoch（`src/fiber.ts` 314-319 行）

// 发布 internal/plugin 事件后：
if (this.uid !== null && parent.fiber.state !== FiberState.UNLOADING) {
  for (const name of Object.keys(this.inject)) {
    this._checkImpl(name)   // 逐个检查依赖服务是否已提供
  }
  this._refresh()           // 计算 epoch，决定加载还是等待
}

12.3 epoch：依赖状态的指纹

_checkImpl + _refresh 的组合是整个依赖注入的精髓：

_checkImpl(name: string) {
  const impl = this.ctx.reflect._getImpl(name, true)  // 严格模式：只认 ACTIVE 的提供方
  if (!impl) return delete this._store[name]          // 服务还没就绪 → 记录缺失
  try {
    if (impl.check && !impl.check.call(getTraceable(this.ctx, impl.value))) {
      return delete this._store[name]                 // check 谓词不通过 → 视为缺失
    }
  } catch (error) {
    impl.fiber.ctx.logger.error(error)
    return delete this._store[name]
  }
  this._store[name] = impl                            // 就绪 → 记入 store
}

_refresh() {
  let epoch: string = ''
  for (const name of Object.keys(this.inject)) {
    const impl = this._store[name]
    if (!impl) { epoch = INACTIVE; break }            // 任一缺失 → epoch = '__INACTIVE__'
    epoch += ':' + impl.fiber.uid                     // 否则拼上提供方 uid
  }
  this._setEpoch(epoch)                               // epoch 变了 → 触发加载/卸载
}

epoch 的设计非常精巧：

'__INACTIVE__' = 有依赖没就绪 → UNLOADING / 等待
':1:2:3' = 依赖齐全（提供方 fiber uid 1、2、3）→ LOADING / ACTIVE
epoch 包含提供方 uid → 提供方换了（比如 llm 实现被替换），epoch 变化，依赖方自动重启

12.4 通知机制：服务变化 → 依赖方自动响应

ReflectService.notify（314-336 行）——服务提供/注销时被调用：

notify(names: string[], filter?) {
  const fibers: Fiber[] = []
  for (const runtime of this.ctx.registry.values()) {     // 遍历所有插件
    for (const fiber of runtime.fibers) {
      let hasUpdate = false
      for (const name of names) {
        if (!(name in fiber.inject)) continue              // 只关心声明了依赖的
        if (!filter?.(fiber.ctx, name)) continue           // 隔离作用域过滤
        hasUpdate = true
        fiber._checkImpl(name)                             // 重查依赖
      }
      if (!hasUpdate) continue
      fiber._refresh()                                     // 重算 epoch → 可能触发加载/卸载
      fibers.push(fiber)
    }
  }
  // 发 internal/service 事件通知
  for (const name of names) { ... this.ctx.events.emit(self, 'internal/service', name, impl?.value) }
  return fibers
}

完整闭环：

插件A: ctx.provide('llm', impl)
  → 写入 store + fiber.store
  → notify(['llm'])
    → 遍历所有插件
      → 插件B inject 里有 'llm' → _checkImpl 重查
        → 找到 ACTIVE 提供方 → _refresh
          → epoch 从 INACTIVE 变为 ':1'
            → _setEpoch → _reload()
              → 插件B 的 apply(ctx) 执行！ctx.llm 可用

这就是"声明依赖，自动就绪"的完整真相——不是魔法，是一个事件驱动的状态机。

十三、`ctx.get()` vs 直接读属性

官方 API 里 ctx.get(name) 是"不要求 inject 的读取"：

// 严格模式（默认）：只返回 ACTIVE 提供方的服务
ctx.get('llm')
// 非严格：PENDING 也返回（可能拿到尚未就绪的实现）
ctx.get('llm', false)

而 ctx.llm 直接读属性时，如果当前 fiber 声明了 inject: ['llm'] 但还没就绪，会抛错（cannot get required service "llm" in inactive context）——这是 Cordis 主动帮你抓"用了没就绪的依赖"这种 bug。

十四、DSH 真实源码剖析：ToolRuntime 服务类

理论讲完，直接看 DSH 真实的服务类长什么样。@deepseek-ai/dsh-tools 的 ToolRuntime（lib/types/index.d.ts）：

import { Context, Service } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'

// ★ 服务类：继承 Service，在 ctx 上注册为 ctx.tools
export declare class ToolRuntime extends Service {
  static inject: string[]          // 服务也可以依赖其他服务！
  static Config: z<Config>         // 服务自己的配置 schema

  constructor(ctx: Context, config?: Config) { ... }

  // 各种私有实现：并行调度器、取消状态、内容 finalizer ...
  private readonly layers
  private readonly defaultMode
  private readonly maxParallelSubCalls
}

然后通过 TS 声明合并把 ctx.tools 和工具事件挂到 Context/Events 上（这是 Cordis 类型安全的来源）：

declare module '@deepseek-ai/cordis' {
  interface Context {
    tools: ToolRuntime;   // ★ ctx.tools 类型
  }
  interface Events {
    // ★ 工具执行流水线事件（全是 waterfall 模式）
    'tools/pre-execute'(this: Scoped<ToolRuntime>, exec: ToolExecution, next: () => Promise<PreToolDecision>): Promise<PreToolDecision>;
    'tools/execute'(this: Scoped<ToolRuntime>, exec: ToolDispatchExecution, next: () => Promise<ToolExecutionResult>): Promise<ToolExecutionResult>;
    'tools/post-execute'(this: Scoped<ToolRuntime>, exec: ToolExecution, result: Readonly<ToolExecutionResult>, next: () => Promise<PostToolDecision>): Promise<PostToolDecision>;
    'tools/result'(this: Scoped<ToolRuntime>, exec: Readonly<ToolExecution>, result: Readonly<ToolExecutionResult>): undefined;  // emit 模式
    'tools/change'(): void;       // emit 模式
  }
}

这段真实代码教了我们什么

服务 = Service 子类：super(ctx, name) 自动 ctx.reflect.provide，static inject 声明服务依赖，static Config 声明配置 schema
事件名 = 命名空间/动作：tools/pre-execute、tools/execute、tools/post-execute——DSH 的所有事件遵循 namespace/action 命名
工具执行是 waterfall 流水线：pre-execute（审批/拦截）→ execute（超时/重试/指标包装）→ post-execute（接受/替换/丰富结果）→ result（emit 观察）。这就是"工具调用"在 DSH 里的完整生命周期！
Scoped 类型：this: Scoped<ToolRuntime> 表示事件按 agent 作用域过滤分发——不同 agent 的工具调用事件互不串扰

💡 这个设计的意义：拦截工具调用、记录工具结果、给工具加超时重试，都不需要改 dsh-tools 源码——注册一个对应事件的监听器即可。DSH 的扩展性就是这么来的。

官方文档的 MetricsService 例子（service.md）

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

export default class MetricsService extends Service {
  static inject = ['llm']               // 服务依赖其他服务

  constructor(ctx: Context) {
    super(ctx, 'metrics')               // 注册为 ctx.metrics
  }

  record(event: string, value: number) {
    // ...
  }
}

消费方：

export const inject = ['metrics']
export function apply(ctx: Context) {
  ctx.metrics.record('tool_call', 1)    // 类型安全！
}

小结：一张图看懂 Context Proxy

你的代码:  ctx.llm
              │
              ▼
   ┌──────────────────────────┐
   │  Proxy get 陷阱           │
   │  ① 特殊属性？→ 直接返回     │
   │  ② 自有属性？→ 返回         │
   │  ③ 否则当服务名解析：        │
   │     internal/get waterfall │
   │       └─ 沿 fiber 链向上找   │
   │          ├─ store 命中 → 返回 │
   │          ├─ inject 未就绪→抛错│
   │          └─ 到根没有 → 抛错   │
   └──────────────────────────┘

第三部分：Fiber 生命周期与 effect 副作用系统

十五、Fiber 是什么？

Fiber = 一个插件的一次运行实例。

每次 ctx.plugin(plugin) 都会创建一个 Fiber。它跟踪：

这个插件的依赖状态（inject 的服务是否就绪）
经过校验的配置
生命周期阶段（PENDING / LOADING / ACTIVE / …）
所有已注册的 effects（副作用）及其清理函数

类比：如果 Cordis 是操作系统，插件是程序，Fiber 是进程。同一个插件可以在不同上下文里跑多个 Fiber（多个"进程"），互不干扰。

十六、生命周期状态机

源码 src/fiber.ts 147-154 行：

export const enum FiberState {
  PENDING,   // 等待依赖服务
  LOADING,   // 插件回调执行中
  ACTIVE,    // 已加载、正在提供服务
  FAILED,    // 回调/配置抛错
  DISPOSED,  // 已卸载，不可重启
  UNLOADING, // disposers 正在执行
}

状态转移图

        ┌──────────┐
        │ PENDING  │◄──────────────┐
        └────┬─────┘               │
             │ 依赖全部就绪        │
             ▼                     │
        ┌──────────┐   依赖缺失    │
        │ LOADING  │───────────────┘
        └────┬─────┘
             │ 回调执行完
             ▼
        ┌──────────┐
        │  ACTIVE  │
        └────┬─────┘
             │ dispose() / 依赖消失
             ▼
        ┌───────────┐
        │ UNLOADING │  ← disposers 逆序执行中
        └────┬──────┘
             │ 清理完成
             ▼
        ┌───────────┐
        │ DISPOSED  │  终态，不可重启
        └───────────┘

FAILED 是旁路：LOADING 或 config 校验抛错 → FAILED。错误被 ctx.logger.error 记录，await fiber 时可以重抛给调用方。

状态转移的实现：`_setEpoch` → `_reload` / `_unload`

private _setEpoch(epoch: string) {
  const oldEpoch = this._runner.epoch
  if (epoch === oldEpoch) return        // epoch 没变 → 无操作
  this._runner.epoch = epoch
  if (this.inertia) return              // 已有转移在进行 → 等它完成

  this._updateState(() => {
    if (epoch !== INACTIVE && oldEpoch === INACTIVE) {
      // 从"不可用"变"可用" → 加载
      this.inertia = this._reload()
      return FiberState.LOADING
    } else {
      // 从"可用"变"不可用" → 卸载
      this.inertia = this._unload()
      return FiberState.UNLOADING
    }
  })
}

注意 inertia 字段——它代表"正在进行的加载/卸载转移"。Fiber 用 await 方法等待所有 inertia 完成：

async await() {
  while (this.inertia) {
    await this.inertia
  }
  if (this._error) throw this._error    // 启动失败 → 重抛
  return this
}

这就是 await ctx.plugin(x) 能等到插件真正加载完成的原理。

`_reload`：插件启动（关键细节）

private async _reload() {
  this.store = { ...this._store }        // 快照依赖实现
  const oldEpoch = this._runner.epoch
  try {
    await Promise.resolve()              // 让出事件循环（防重入）
    // 检查 epoch 是否还是旧的（期间可能被 dispose）
    if (this._runner.epoch === oldEpoch) {
      this.config = this._resolveConfig(this._config)   // 校验配置
      await this._execute(this._runner)   // ★ 执行插件回调！
      this._error = undefined
    }
  } catch (reason) {
    this.ctx.logger.error(reason)
    this._error = reason
    this._runner.epoch = INACTIVE
  }
  // 收尾：如果没有新的转移请求，inertia 置空
  this._updateState(() => {
    if (this._runner.epoch === oldEpoch) {
      this.inertia = undefined
    } else {
      this.inertia = this._unload()       // 期间又变了 → 继续卸载
      return FiberState.UNLOADING
    }
  })
}

这里有个非常严谨的防竞态设计：await Promise.resolve() 让出后检查 epoch，防止"加载途中被 dispose"导致插件代码在已卸载的上下文里运行。

`_execute`：执行插件回调的三种形态

execute: function () {
  if (isConstructor(runtime.callback)) {
    // ① 类插件：new 出来，跑 init hooks 和 [init] 方法
    const instance = new runtime.callback(this.ctx, this.config)
    for (const hook of instance?.[symbols.initHooks] ?? []) {
      hook()
    }
    return instance?.[symbols.init]?.()
  } else {
    // ② 函数/对象插件：直接调用 apply(ctx, config)
    return runtime.callback(this.ctx, this.config)
  }
}

十七、effect 副作用系统：注册即清理

17.1 基础用法

ctx.effect(() => {
  // 立即执行的注册逻辑
  const timer = setInterval(() => {}, 1000)
  server.listen(8080)

  // 返回 disposer：卸载时逆序执行
  return () => {
    clearInterval(timer)
    server.close()
  }
})

接受的返回值形态（Effect 类型）：

返回	含义
`() => void`	单个 disposer
`Promise<() => void>`	异步 disposer
`Iterable<() => void>`	生成器，逐个 yield disposer
`AsyncIterable<() => void>`	异步生成器
`undefined` / `null`	无清理

17.2 一个 effect 能产生多个 disposer

ctx.effect(function* () {
  yield registerA()          // 第一个 disposer
  yield registerB()          // 第二个 disposer
  yield registerC()          // 第三个 disposer
  // 卸载时按 C → B → A 逆序执行
})

17.3 逆序清理 + 幂等

Fiber.effect 的实现核心（424-441 行）：

const disposables: Disposable[] = []
let disposing = false
const dispose = () => {
  if (disposing) return disposalTask   // ★ 幂等：二次调用直接返回进行中的任务
  disposing = true
  let task!: void | Promise<void>
  // ★ 逆序执行：splice(0).reverse()
  for (const disposable of disposables.splice(0).reverse()) {
    if (task) {
      task = task.then(() => runDisposable(disposable))   // 串行 await
    } else {
      const result = runDisposable(disposable)
      if (isObject(result) && 'then' in result) {
        task = result
      }
    }
  }
  return disposalTask = task
}

逆序执行的理由：后注册的依赖先注册的（比如先注册了 server 再注册了路由，卸载时先关路由再关 server），保证资源释放顺序正确。

17.4 所有注册 API 都是 effect

这是 Cordis 最优雅的一点——你接触到的每个注册 API，底层都是 effect：

// ctx.on → EventsService.register → ctx.fiber.effect(...)
on(name, listener, options) {
  ...
  return this.register(label, hooks, listener, options)
}
register(label, hooks, callback, options) {
  return this.ctx.fiber.effect(() => {
    hooks[method]({ ctx: this.ctx, callback, ...options })
    return () => this.unregister(hooks, callback)   // disposer = 反注册
  }, label)
}

// ctx.provide → ReflectService.provide → ctx.fiber.effect(...)
provide(name, value, check) {
  return this.ctx.fiber.effect(() => {
    ...注册逻辑...
    return async () => { ...注销逻辑... }
  }, `ctx.provide(${JSON.stringify(name)})`)
}

// ctx.mixin → ctx.fiber.effect(function* () { ... yield self.accessor(...) })

所以"卸载插件无残留"不是靠约定，而是机制保证——每个注册都绑定到创建它的 Fiber，Fiber 卸载必然逆序触发所有 disposer。

17.5 诊断：EffectMeta 树

每个 effect 带 label，嵌套 effect 形成树：

interface EffectMeta {
  label: string        // 如 `ctx.on("message/send")`、`ctx.provide("llm")`
  children: EffectMeta[]  // 这个 effect 运行期间注册的嵌套 effect
}

fiber.getEffects() 返回诊断树，这是调试"谁注册了什么"的利器。

十八、热重载的完整流程

现在把前面的知识串起来，看一次完整的热重载：

1. 插件B 正在 ACTIVE，依赖服务 llm（提供方 fiber uid=1）
2. 用户调用 ctx.update(newConfig) 或插件A重载 llm
3. 依赖变化 → notify(['llm'])
4. 插件B.fiber._checkImpl('llm') → 找到新实现
5. 插件B.fiber._refresh() → epoch 从 ':1' 变 ':2'（提供方换了）
6. _setEpoch → 旧 epoch ≠ 新 epoch → _unload()
7. UNLOADING：逆序执行所有 disposer（监听器移除、服务注销、定时器清理）
8. _unload 结束 → 检查 epoch：还是新的 → _reload()
9. LOADING：重新执行 apply(ctx, config)
10. ACTIVE：插件B 以新配置/新依赖重新上线

整个过程不用重启进程，插件代码无感知。 这就是"Agent 在运行中改写自己"的运行时基础。

十九、update：配置热更新

Fiber.update(config)（736-753 行）：

update(config: any, noSave = false) {
  this.assertActive()
  this._config = config
  if (this.state !== FiberState.ACTIVE) {
    // 还没激活：延迟到能激活时再解析
    this._error = undefined
    this._setEpoch(INACTIVE)
    this._refresh()
    return
  }
  config = this._resolveConfig(config)          // 先校验新配置
  return this.context.waterfall(this, 'internal/update', config, noSave, () => {
    this.config = config
    this._error = undefined
    return this.restart()                        // 默认行为：重启插件
  })
}

关键设计：internal/update 是 waterfall——其他插件可以拦截、否决或替换这次更新。比如某个插件可以监听 internal/update 说"这个配置不能改，我 veto"，或者 HMR 工具可以接管热更新逻辑。

二十、错误处理与 long stack trace

异步代码的错误栈会丢失调用点，Cordis 用 composeError + buildOuterStack 解决：

export function composeError<T>(callback: (info: StackInfo) => T, getOuterStack = buildOuterStack()): T {
  const info: StackInfo = { offset: 1, error: new Error() }
  try {
    const result: any = callback(info)
    if (isObject(result) && 'then' in result) {
      // 异步：把外层调用栈拼接到错误栈末尾
      return result.then(undefined, (reason) => handleError(info, reason, getOuterStack))
    }
    return result
  } catch (reason) {
    handleError(info, reason, getOuterStack)
  }
}

handleError 会找到错误栈与内部栈的分界点，把插件调用点的栈帧拼接进去，让你一眼看到"这个异步错误是从哪个插件的哪行代码引发的"。

二十一、dispose 的三重保证（官方文档原文）

官方 framework 文档明确规定了 fiber.dispose() 的语义：

const fiber = ctx.plugin(myPlugin)

// 手动提前终止
await fiber.dispose()

dispose 保证：

该插件拥有的所有注册均被移除（监听器、服务、工具、effect 全部逆序清理）
它的子插件也被递归卸载（ctx.plugin(childPlugin) 创建的嵌套 Fiber 一起销毁）
返回的 Promise 会在所有异步清理完成后兑现（等所有 disposer 跑完）

官方生命周期示例

export function apply(ctx: Context) {
  console.log('plugin loading')

  ctx.effect(() => {
    console.log('effect registered')
    return () => console.log('effect cleaned up')
  })
}

加载时输出：

plugin loading
effect registered

卸载时输出：

effect cleaned up

⚠️ 官方提醒：异步 disposer 并发执行

插件卸载时，处置器按注册顺序的逆序开始调用，但多个异步处置器会并发执行，不保证逐个完成。存在顺序依赖的清理步骤必须放进同一个 ctx.effect() 返回的处置器中，由该处置器负责串行等待。

// ❌ 错误：两个 effect 的异步清理会并发跑，顺序无保证
ctx.effect(async () => { await stopServerA() })
ctx.effect(async () => { await stopServerB() })

// ✅ 正确：顺序敏感的清理放进同一个 effect
ctx.effect(async () => {
  await stopServerA()
  await stopServerB()
})

二十二、HMR 热模块替换

DSH 的开发体验里有个杀手锏：改插件源码 → 自动热替换。官方文档说明（通过 cordis.yml 加载 @deepseek-ai/cordis-plugin-hmr 后）：

卸载旧插件（清理所有注册）
重新加载新代码
执行新的 apply

因为插件注册会被自动清理，热替换不会保留旧实例的任何残留——这是 effect 系统的直接红利。对比 Node 原生 require.cache 清缓存的粗暴做法，Cordis 的热替换是"状态机驱动的优雅重载"。

💡 这也解释了为什么 DSH 插件开发体验这么好：改一行代码，效果即时可见，不用重启整个 Harness。

二十三、Fiber 能抄什么：给框架作者的启示

状态机驱动插件生命周期：PENDING→LOADING→ACTIVE 的显式状态 + epoch 指纹，比"回调+标志位"可靠得多
effect 即注册：所有副作用统一走 effect，逆序清理、幂等、可诊断——框架层面消灭内存泄漏
inertia 防竞态：加载/卸载转移期间的状态变化排队处理，杜绝"加载中被卸载"的并发 bug
waterfall 做扩展点：config 校验、服务读取、配置更新都是 waterfall，第三方可拦截
await 语义：ctx.plugin() 返回 PromiseLike，调用方可以等待加载完成或捕获启动错误

第四部分：事件分发 5 模式与真实 DSH 插件实战

二十四、为什么事件分发需要 5 种模式？

Node 的 EventEmitter 只有一种：emit（同步调用所有监听器）。但真实应用里"通知"的语义千差万别：

想通知一下，不关心结果 → emit
想并行跑所有监听器再等结果 → parallel
想按序询问，第一个给出答案就停 → serial / bail
想拦截/包装某个操作 → waterfall（中间件）

Cordis 把 DispatchMode 定义为 5 种，每种语义明确、互不替代：

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

二十五、五种模式详解

1️⃣ `emit` — 同步通知

// 源码：同步调用所有监听器，不等待返回值
emit(...args: any[]) {
  this.dispatch('emit', args).map(cb => cb(...args))
}

不 await，监听器返回值被忽略
按注册顺序调用
适合：广播通知（“消息发出了”、“插件加载了”）

2️⃣ `parallel` — 并行扇出

// 源码：Promise.allSettled 并行执行
async parallel(...args: any[]) {
  const results = await Promise.allSettled(this.dispatch('emit', args).map(async cb => cb(...args)))
  const errors = results.filter(r => r.status === 'rejected')
  if (errors.length) throw new AggregateError(errors.map(e => e.reason))
}

await 所有监听器完成
并行执行，一个失败不影响其他（allSettled）
全部失败才抛 AggregateError
适合：需要等所有异步监听器做完的事（如"采集完成，让所有扩展都落盘"）

3️⃣ `serial` — 按序等待，首个结果短路

// 源码：逐个 await，遇到 bail 值（非 null/false/undefined）就停
async serial(...args: any[]) {
  for (const cb of this.dispatch('serial', args)) {
    const result = await cb(...args)
    if (isBailed(result)) return result    // 非 null/false/undefined 即短路
  }
}

await 每个监听器（支持异步）
按注册顺序执行
第一个返回非空值就停止，返回该值
适合：路由查询（“谁能处理这个请求？”）、权限判断

4️⃣ `bail` — 同步短路

// 源码：同步调用，遇到 bail 值即停
bail(...args: any[]) {
  for (const cb of this.dispatch('bail', args)) {
    const result = cb(...args)
    if (isBailed(result)) return result
  }
}

和 serial 一样短路，但同步、不 await。
适合：同步的决策链（“谁来兜底？”）、校验链。

5️⃣ `waterfall` — 环绕中间件（精髓！）

// 源码：监听器组合成洋葱模型，最后一个参数是 next
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()
}

waterfall 是 Cordis 最强大的分发模式，本质是洋葱模型中间件：

监听器1 ──► 监听器2 ──► 监听器3 ──► 内置实现(inner)
   ▲                                    │
   └──────────── 返回值逐层回传 ─────────┘

每个监听器接收 (...args, next)：

调用 next() → 执行下游（可以包装返回值）
不调用 next() → 短路（veto），下游全部跳过

waterfall 实战：拦截配置更新

// 内置的 internal/update 就是 waterfall
ctx.on('internal/update', (config, noSave, next) => {
  if (config.model === 'forbidden-model') {
    console.log('已否决配置更新')
    return            // 不调 next() → 否决！
  }
  return next()       // 放行
})

五种模式对比表

模式	await?	顺序	短路?	返回值	典型用途
`emit`	❌	注册序	❌	忽略	广播通知
`parallel`	✅	并行	❌	聚合错误	并行任务
`serial`	✅	注册序	✅(非空)	首个结果	异步路由
`bail`	❌	注册序	✅(非空)	首个结果	同步决策链
`waterfall`	❌	洋葱	✅(不调next)	链式结果	中间件/拦截

二十六、上下文过滤：`ctx.on` 的隔离魔法

EventsService.dispatch 里有个细节（165-175 行）：

dispatch(type: string, args: any[]) {
  const thisArg = typeof args[0] === 'object' || typeof args[0] === 'function' ? args.shift() : null
  const name: string = args.shift()
  ...
  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))
}

监听器注册时带着自己的 ctx。分发时可以传一个 thisArg（带 filter），只有同一作用域的监听器才会被调用。这是 isolate() 作用域隔离在事件层面的延伸——不同用户的会话，事件互不串扰。

二十七、内置事件表（Events 接口）

src/events.ts 定义了一组 internal/* 内置事件，DSH 的扩展点全靠它们：

事件	模式	触发时机
`internal/plugin`	emit	插件 Fiber 创建/销毁
`internal/status`	emit	Fiber 状态变化
`internal/config`	waterfall	插件配置解析（可拦截）
`internal/update`	waterfall	配置更新（可否决）
`internal/get`	waterfall	读服务时（可偷梁换柱）
`internal/set`	waterfall	写服务时
`internal/listener`	bail	注册监听器时（可替换）
`internal/service`	emit	服务绑定变化
`internal/dispatch`	emit	事件分发前（诊断）

DSH 的 ctx.tools、ctx.llm 扩展点本质上就是这些 internal 事件——工具注册、模型流式输出拦截，都是插件监听/触发这些事件实现的。

二十八、实战：手写一个 DSH 风格插件

现在把前面所有知识串起来，写一个完整的插件。场景：一个"模型输出过滤器"插件——监听 LLM 输出，把敏感词替换成 ***，并提供配置。

28.1 插件结构（对象插件形态）

// model-filter.ts
import type { Context } from '@deepseek-ai/cordis'

// 配置 schema（standard-schema 格式，Cordis 用它做校验）
const Config = {
  '~standard': {
    version: 1,
    vendor: 'cordis-demo',
    validate(value: unknown) {
      const result: any = {
        value: { sensitiveWords: [] as string[], enabled: true, ...(value as object) },
      }
      if (!Array.isArray(result.value.sensitiveWords)) {
        return { issues: [{ message: 'sensitiveWords 必须是数组' }] }
      }
      return result
    },
  },
}

export const modelFilterPlugin = {
  name: 'model-filter',
  Config,                          // 配置校验
  inject: ['llm'],                 // 依赖 llm 服务
  apply(ctx: Context, config: any) {
    const { sensitiveWords, enabled } = config
    const disposers: Array<() => void> = []

    // ① effect：注册一个内部服务
    disposers.push(ctx.effect(() => {
      console.log(`[model-filter] 已启动，敏感词: ${sensitiveWords.join(', ')}`)
      return () => console.log('[model-filter] 已卸载')
    }))

    // ② 监听 LLM 输出事件（假设 harness 有 llm/stream 事件，这里演示）
    disposers.push(ctx.on('llm/stream' as any, (chunk: any, next: any) => {
      if (!enabled) return next()
      let text = String(chunk?.text ?? '')
      for (const word of sensitiveWords) {
        text = text.split(word).join('***')
      }
      chunk.text = text
      return next()          // waterfall：改完继续传递
    }))

    // ③ 提供自己的服务：让别的插件能调用 ctx.modelFilter
    disposers.push(ctx.provide('modelFilter', {
      addWord(word: string) {
        sensitiveWords.push(word)
        console.log(`[model-filter] 新增敏感词: ${word}`)
      },
      list() {
        return [...sensitiveWords]
      },
    }))

    // ④ 监听配置更新（演示 internal/update 拦截）
    disposers.push(ctx.on('internal/update' as any, (newConfig: any, noSave: any, next: any) => {
      if (newConfig?.sensitiveWords?.length > 100) {
        console.warn('[model-filter] 敏感词过多，已否决更新')
        return    // 不调 next() = 否决
      }
      return next()
    }))

    // ⑤ 卸载时统一清理（其实 ctx.on/provide/effect 已自动清理，这里演示手动）
    return () => {
      for (const d of disposers) d()
    }
  },
}

28.2 使用插件

import { Context } from '@deepseek-ai/cordis'
import { modelFilterPlugin } from './model-filter'

const app = new Context()

// 先提供 llm 服务（模拟）
app.plugin({
  name: 'fake-llm',
  apply(ctx) {
    ctx.provide('llm', {
      async chat(messages: any[]) {
        return { text: 'hello world, this is a test' }
      },
    })
  },
})

// 挂载过滤器（inject: ['llm'] 会自动等 llm 就绪）
const fiber = app.plugin(modelFilterPlugin, {
  sensitiveWords: ['hello'],
  enabled: true,
})

// 等插件真正加载完成
await fiber

// 通过服务调用
const filter = (app as any).modelFilter
console.log(filter.list())          // ['hello']
filter.addWord('test')

// 触发 LLM 输出事件
app.emit('llm/stream' as any, { text: 'hello world, this is a test' })
// 输出: [model-filter] 已启动，敏感词: hello
//       [model-filter] 新增敏感词: test
//       （监听器把 hello 和 test 替换成 ***）

// 更新配置（会被 internal/update 校验）
await (fiber as any).update({ sensitiveWords: [], enabled: true })

// 卸载插件 → 自动清理所有副作用
await fiber.dispose()
// 输出: [model-filter] 已卸载

28.3 这个例子演示了什么

机制	代码位置
插件形态（对象 + name/Config/inject/apply）	28.1 整体
配置校验（standard-schema）	`Config`
依赖注入（inject: [‘llm’]）	`inject: ['llm']`
effect 副作用（启动/卸载日志）	①
事件监听（waterfall 拦截输出）	② `ctx.on('llm/stream', ...)`
服务提供（ctx.provide）	③
事件拦截（internal/update 否决）	④
Fiber 生命周期（await/update/dispose）	28.2
自动清理（dispose 后无残留）	28.3

二十九、DSH 真实实战：工具调用流水线事件

前面第十四节我们看了 DSH 的 tools 事件声明。这一节用真实代码演示怎么用这些事件写插件——这是官方 events.md 文档里的完整例子，我加了详细注释：

29.1 工具日志插件（官方示例）

import type { Context } from '@deepseek-ai/cordis'
import '@deepseek-ai/dsh-tools'    // 引入类型声明，让事件类型安全

export const name = 'tool-logger'

export function apply(ctx: Context) {
  // 监听工具结果（emit 模式，纯观察）
  ctx.on('tools/result', (exec, result) => {
    console.log(`[tool] ${exec.name}(${JSON.stringify(exec.arguments)})`)
    const text = result.content
      .map(block => block.type === 'text' ? block.text : '')
      .join('')
    console.log(`[tool result] ${text.slice(0, 100)}`)
  })
}

29.2 工具审批插件（waterfall 拦截，DSH 的灵魂用法）

DSH 的工具流水线是 pre-execute → execute → post-execute → result 四段 waterfall，每一段都能拦截。写一个"敏感工具审批"插件：

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

export const name = 'tool-guard'

export function apply(ctx: Context) {
  // pre-execute：工具派发前审批
  ctx.on('tools/pre-execute', async (exec, next) => {
    if (exec.name === 'run_code' && exec.arguments.includes('rm -rf')) {
      // 不调 next() = 否决！这个工具调用会被拦截
      console.warn(`[tool-guard] 已拦截危险操作: ${exec.name}`)
      return { decision: 'deny', reason: '危险 shell 命令被拦截' }
    }
    return next()   // 放行
  })

  // post-execute：结果出来后可以替换/丰富
  ctx.on('tools/post-execute', (exec, result, next) => {
    if (result.status === 'error' && exec.name === 'fetch') {
      console.error(`[tool-guard] fetch 失败，重试一次`)
      // 这里可以返回重试后的新结果
    }
    return next()
  })
}

29.3 作用域过滤：Scoped

注意事件签名里的 this: Scoped<ToolRuntime>——这表示事件按 agent 作用域过滤分发：

// agent A 的工具调用，只有 A 作用域内的监听器能收到
ctx.on('tools/result', (exec, result) => {
  // exec.agent 是调用者 agent
  console.log(`[agent:${exec.agent}] 调用了 ${exec.name}`)
})

这就是多 agent 场景下事件不串扰的机制——isolate() 作用域隔离在事件层面的体现。

29.4 类型安全事件声明合并（官方 events.md）

给 Cordis 加自定义类型化事件：

import '@deepseek-ai/cordis'

declare module '@deepseek-ai/cordis' {
  interface Events {
    'my-plugin/ready': (payload: { id: string }) => void
    'my-plugin/check': (input: string) => boolean | undefined
    'my-plugin/transform': (input: string, next: () => Promise<string>) => Promise<string>
  }
}

// 从此 ctx.on / ctx.emit 自动推断参数类型
ctx.on('my-plugin/ready', ({ id }) => { ... })
ctx.emit('my-plugin/ready', { id: 'worker-1' })

29.5 会话事件 vs Cordis 事件（避坑指南）

官方文档特意提醒了一个容易踩的坑：

turn/*、step/*、tool/call、tool/result 和 compaction/* 是持久化的会话事件类型，不是同名 Cordis 事件。需要观察它们时，监听 session/event 并检查 event.type。

// ❌ 错误：这些是会话记录，不是 Cordis 事件
ctx.on('tool/call', handler)

// ✅ 正确：监听 session/event 再筛选
ctx.on('session/event', (event) => {
  if (event.type === 'tool/call') { ... }
})

三十、DSH 真实插件对比：dsh-skills 的注册方式

看看真实 DSH 插件怎么注册（~/.dsh/profiles/web/node_modules/dsh-skills/cordis.patch.yml）：

- insert:
    - id: skill-hub
      name: 'dsh-skills'

配合 package.json：

{
  "dsh": {
    "profile": {
      "bundles": ["@deepseek-ai/dsh-base", "@deepseek-ai/dsh-web-app", "dsh-skills"]
    }
  }
}

DSH 的 loader 把 YAML patch 解析成 ctx.plugin() 调用——id 是 Fiber 标识，name 是包名。你写的插件只要导出 name/Config/inject/apply（或 Service 子类），就能被 DSH 以同样的方式加载。这就是为什么理解 Cordis 是理解 DSH 插件开发的钥匙。

总结

从五个核心概念 → Context Proxy → Fiber 状态机 → 事件分发，我们完整走了一遍 Cordis 的设计：

插件 = 服务，注册即 effect，卸载零残留
上下文 = 服务容器，Proxy 代理 + 依赖注入自动编排
Fiber = 生命周期状态机，epoch 驱动热重载，inertia 防竞态
事件 = 5 种分发模式，waterfall 提供中间件能力，internal/* 事件是 DSH 的扩展点

Cordis 最值得学的不是 API，而是设计哲学：显式状态机、副作用可逆、依赖声明式、扩展点事件化。这套设计对任何"可插拔、可热更新的应用框架"都有直接借鉴意义。

如果这篇文章对你有帮助，欢迎点赞收藏转发；有任何问题评论区见，我尽量都回。

参考

源码：@deepseek-ai/cordis v4.0.1（src/context.ts / src/registry.ts / src/fiber.ts / src/events.ts / src/service.ts / src/reflect.ts / src/utils.ts）
DSH 真实服务源码：@deepseek-ai/dsh-tools（lib/types/index.d.ts）
官方文档：cordis-primer.zh.md
DeepSeek Harness：github.com/deepseek-ai/deepseek-harness

DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

我用 Qwen3.8-Max 做了一个采购合同风险审查工具：从能分析，到能定位、复核和交付

DeepSeek技术社区

我用 Qwen3.8-MAX 搭了一个教研纪要 Agent，一次集体备课几秒出七段、分歧还不被和稀泥

DeepSeek技术社区

我用 Qwen3.8-Max 做了缠论结构观察员，跑完了从开发到修复的全过程

Qwen3.8-Max 没有替我判断方向，也没有把缠论包装成预测工具。它在开发过程中支持多 Agent 并行，能够理解较长的产品约束和算法边界，拆分文档与实现任务；当子任务停滞后回到主上下文继续推进；再根据“笔断裂”“中枢过宽”这种图上的具体问题，定位到算法文件并补回归测试。复杂项目不能靠一次生成直接交付。Agent Team 会卡住，初版算法也会错。模型放进开发闭环里才有意义，最后还是要靠可检查

DeepSeek技术社区

所有评论(0)

查看更多评论

qq_56972807

@qq_56972807

已为社区贡献1条内容

DeepSeek Harness 能“在运行中改写自己“，全靠这个叫 Cordis 的插件框架（附完整源码解析）

qq_56972807

Cordis 深入解析：DeepSeek Harness 背后的插件框架（核心概念 → 源码原理 → 生命周期 → 实战，一篇全看懂）

引言：为什么 DeepSeek Harness 值得研究它的底座

第一部分：核心概念与源码架构

一、Cordis 到底解决什么问题？

二、五个核心概念（官方原版提炼）

1️⃣ 插件是实现 Service 的对象

2️⃣ 上下文是服务的容器

3️⃣ 通过 inject 声明服务依赖

4️⃣ 类型化事件用于通信

5️⃣ 注册是可逆的副作用

三、源码架构总览（v4.0.1，仅 9 个模块）

四、内置服务的职责划分

五、一个最小可运行的 Cordis 应用

六、Cordis 在 DSH 里实际怎么用

七、Cordis 生态全家桶（DSH 实际安装的包）

DSH 官方文档导航（深入学习路径）

第二部分：Context Proxy 与依赖注入源码拆解

八、一个灵魂问题

九、Context：一个自指 Proxy

十、Proxy get 陷阱：服务解析管线

这段代码的三个关键设计

十一、服务存哪了？ReflectService 的 store

数据结构全景

十二、依赖注入的完整闭环：inject → 等待 → 启动

12.1 ctx.plugin() 创建 Fiber（src/registry.ts 316-336 行）

12.2 Fiber 构造：检查依赖 → 刷新 epoch（src/fiber.ts 314-319 行）

12.3 epoch：依赖状态的指纹

12.4 通知机制：服务变化 → 依赖方自动响应

十三、ctx.get() vs 直接读属性

十四、DSH 真实源码剖析：ToolRuntime 服务类

这段真实代码教了我们什么

官方文档的 MetricsService 例子（service.md）

小结：一张图看懂 Context Proxy

第三部分：Fiber 生命周期与 effect 副作用系统

十五、Fiber 是什么？

十六、生命周期状态机

状态转移图

状态转移的实现：_setEpoch → _reload / _unload

_reload：插件启动（关键细节）

_execute：执行插件回调的三种形态

十七、effect 副作用系统：注册即清理

17.1 基础用法

17.2 一个 effect 能产生多个 disposer

17.3 逆序清理 + 幂等

17.4 所有注册 API 都是 effect

17.5 诊断：EffectMeta 树

十八、热重载的完整流程

十九、update：配置热更新

二十、错误处理与 long stack trace

二十一、dispose 的三重保证（官方文档原文）

官方生命周期示例

⚠️ 官方提醒：异步 disposer 并发执行

二十二、HMR 热模块替换

二十三、Fiber 能抄什么：给框架作者的启示

第四部分：事件分发 5 模式与真实 DSH 插件实战

二十四、为什么事件分发需要 5 种模式？

二十五、五种模式详解

1️⃣ emit — 同步通知

2️⃣ parallel — 并行扇出

3️⃣ serial — 按序等待，首个结果短路

4️⃣ bail — 同步短路

5️⃣ waterfall — 环绕中间件（精髓！）

waterfall 实战：拦截配置更新

五种模式对比表

二十六、上下文过滤：ctx.on 的隔离魔法

二十七、内置事件表（Events 接口）

二十八、实战：手写一个 DSH 风格插件

28.1 插件结构（对象插件形态）

28.2 使用插件

28.3 这个例子演示了什么

二十九、DSH 真实实战：工具调用流水线事件

29.1 工具日志插件（官方示例）

29.2 工具审批插件（waterfall 拦截，DSH 的灵魂用法）

29.3 作用域过滤：Scoped

29.4 类型安全事件声明合并（官方 events.md）

29.5 会话事件 vs Cordis 事件（避坑指南）

三十、DSH 真实插件对比：dsh-skills 的注册方式

总结

3️⃣ 通过 `inject` 声明服务依赖

十二、依赖注入的完整闭环：`inject` → 等待 → 启动

12.1 `ctx.plugin()` 创建 Fiber（`src/registry.ts` 316-336 行）

12.2 Fiber 构造：检查依赖 → 刷新 epoch（`src/fiber.ts` 314-319 行）

十三、`ctx.get()` vs 直接读属性

状态转移的实现：`_setEpoch` → `_reload` / `_unload`

`_reload`：插件启动（关键细节）

`_execute`：执行插件回调的三种形态

1️⃣ `emit` — 同步通知

2️⃣ `parallel` — 并行扇出

3️⃣ `serial` — 按序等待，首个结果短路

4️⃣ `bail` — 同步短路

5️⃣ `waterfall` — 环绕中间件（精髓！）

二十六、上下文过滤：`ctx.on` 的隔离魔法