DSH 从零到一

深色模式

源码拆解 04 · dsh-agent 与 dsh-agent-loop：驱动引擎 ​

本篇拆解两个包：core/agent（Agent 接口、注册表、事件声明）与 core/agent-loop（默认驱动 ReactLoopAgent）。这是全仓库最复杂的部分，但拆开后核心就是一个约 500 行的状态机。

4.1 分工：接口与实现为什么是两个包 ​

包角色类比
dsh-agentAgent 接口 + 注册表 AgentRegistry（ctx.agents）+ agent/* 事件声明 + inbox汽车的"驾驶接口"：方向盘、油门、仪表盘
dsh-agent-loop接口的默认实现 ReactLoopAgent默认的"司机"

为什么拆开？官方注释说得清楚：消费方（比如 ACP 桥接）只依赖 ctx.agents 接口编程，不 import 具体实现包——你可以写一个全新的 driver 注册成 factory 替换掉 agent-loop，而所有消费方代码不变。这是"接缝"思想在 agent 层的兑现（原理篇 3.6）。

4.2 Agent 接口：状态机的外部视图 ​

runtime-types.ts 里的 Agent 接口（完整清单见原理篇 5.5）。源码里最值得注意的两处：

ts
export interface Agent {
  readonly id: SessionId
  readonly session: Session      // agent 与 session 共享同一个 id
  readonly inbox: Inbox          // 待办消息的有序投影
  readonly status: AgentStatus   // 'idle' | 'running'
  readonly ctx: Context          // agent 作用域上下文
  followup(message: UserMessage): void
  steer(message: UserMessage): void
  inject(message: UserMessage): void
  cancel(cause: AgentCancelCause, options?: CancelOptions): void
  whenIdle(): Promise<void>
}

agent 与 session 是同一个东西的两个面：agent.id === session.id。Agent 面回答"现在怎么驱动"，Session 面回答"发生过什么"。接口上的 status 只有 idle / running 两态——没有 disposed 第三态，因为"销毁"不是状态而是从注册表移除（官方注释特意强调了这个设计）。

AgentRegistry（ctx.agents）维护活跃 agent 的注册表，create() 走 factory 委托，register() 是 effect。它还有一对有意思的方法：withInitiator / requireInitiator——在同进程的异步调用链上传递"是谁发起的"这个因果信息（日志、遥测、权限判断都要用）。

4.3 ReactLoopAgent：phase 状态机 ​

打开 agent-loop/src/agent.ts，先看顶部的状态定义（第 38–46 行）：

ts
type Phase =
  | { kind: 'idle'; lastTurn: number }
  | { kind: 'maintenance'; abort: AbortController; lastTurn: number; wakeRequested: boolean }
  | { kind: 'running'; abort: AbortController; turn: number; step: number; wakeRequested: boolean }

整个 driver 是一个三态机：idle（无 driver）→ running（驱动 turn/step）→ 回到 idle，maintenance 是空闲期插入的维护任务态。wakeRequested 处理"运行中又来新输入"的竞态：不打断当前 step，记一笔，step 边界再认领。每个 phase 自带 AbortController——取消 = abort 当前 phase 的 controller，所有下游（模型流、工具执行）通过 signal 感知。

构造函数（第 80–97 行）做了一件关键的事：

ts
this.scope = createScope(loopCtx, this)          // 创建作用域（scope-filtered 事件分发）
this.ctx = this.scope.ctx.extend({ agent: this }) // agent 的作用域上下文
this.dispatch = agentEvents(loopCtx, this)       // 一次性融合的事件派发器
this.inbox = new Inbox(session, { /* 回调里 emit agent/inbox/* 事件 */ })

agent.ctx 就是这里造出来的：一个带作用域过滤的子上下文。之后 setup 回调里注册的一切都发生在它下面——这就是"每个 agent 一个局部世界"的实现位置。

4.4 turn()：第 5 章时序图的代码本体 ​

turn() 方法（第 246–330 行）与原理篇 5.2 的时序图逐行对应：

ts
private async turn(): Promise<boolean> {
  const turn = phase.turn + 1
  this.session.append('turn/start', { turn })          // ① 轮次开始落盘
  // …
  while (true) {
    const decision = await this.preStep(target, { turn, step })   // ② agent/pre-step 瀑布
    if (decision.kind === 'reject') { turnEnds = { kind: 'blocked' }; return false }
    this.session.append('step/start', { turn, step })  // ③ step 开始
    for (const message of decision.messages) {
      this.session.append('user/message', message, { surfaceOp: 'append' })  // ④ 消息落盘
    }
    const stepEnd = await this.step(decision.assembly) // ⑤ 执行 step
    // …
    if (turnEnds && this.inbox.nextStep.length === 0) {
      await this.dispatch.serial('agent/turn-stopping', { turn, signal })  // ⑥ 最后检查点
    }
    if (turnEnds && this.inbox.nextStep.length === 0) break
    target = 'next-step'
  }
  // finally：无论怎么退出，turn/end 一定落盘
  this.session.append('turn/end', { turn, reason: turnEnds! })
}

读这段代码的三个收获：

  1. 落盘先行：turn/start、step/start、user/message 都在动作发生之前写日志——"模型可见即已记录"不是事后补记，而是顺序保证。
  2. finally 保证闭环：turn 无论正常、被取消、还是抛错，turn/end 都会写，reason 区分 completed / aborted / error / max-tokens / blocked。日志结构永远自洽。
  3. 错误结构化：LlmError 保留结构化事实（code/status），其他异常拍平成 { message, code: 'UNKNOWN' } 才落盘——日志里绝不会有不可序列化的 Error 对象。

4.5 step()：一次模型调用 + 工具循环 ​

step()（第 332 行起）内部是一个 while (true) 循环：

ts
private async step(assembly: PromptAssembly): Promise<StepEndReason | null> {
  const system = renderPrompt(assembly)
  while (true) {
    const { request, preparedCall } = await this.buildRequest(/* … */)  // agent/request 瀑布
    const assembler = new BlockAssembler()
    const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)
    for await (const chunk of stream) {
      chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)  // 每个增量落盘
      assembler.push(chunk)
    }
    // …组装 assistant/message、按 executionMode 分类工具调用、executeToolCalls()…
  }
}

buildRequest（第 407 行起）里有一段教科书级的瀑布用法：

ts
const proposedConfig = await this.dispatch.waterfall(
  'agent/request', { /* … */ },
  (): Promise<LlmCallConfig> => Promise.resolve(requestProposal(header)),
)
if (!proposedConfig.provider || !proposedConfig.model) {
  throw new Error(`agent "${this.id}" has no provider/model: set AgentOptions…`)
}

内置行为（瀑布最内层的 next）提供"缺省配置"，任何监听器都能返回一个替换的 LlmCallConfig。工具循环在 tool-calls.ts（executeToolCalls）：按 executionMode 分类 → 有界滚动池调度 → 每个调用走 pre/execute/post 三个瀑布 → 结果落盘。这个文件我们下一篇拆 tools 包时再对读。

4.6 取消与错误恢复：signal 贯穿一切 ​

全文件里 signal.throwIfAborted() 出现在每个 await 之后——取消不是"等当前操作自然结束"，而是在最近的检查点立即抛出。配合 phase.abort 的独立 controller，一个 turn 的取消不会污染下一个 turn（新 turn 换新 controller）。

错误恢复走 agent/request-error 瀑布（声明在 runtime-types.ts）：默认监听器按注册的重试策略返回 { kind: 'retry' }，step() 据此重试或终止。注意恢复的位置语义：失败发生在 step 结束后、turn 结束前——这正是 finally 结构能保证日志闭环的原因。

4.7 一个易读点：dispatch 的融合派发器 ​

agent.ts 第 74 行注释说 "Fused dispatcher, built once in the constructor so hot-path dispatches never allocate"。agentEvents(loopCtx, this) 在构造时把每个 agent/* 事件名和它的 scope 载体预绑定成一个 dispatch 对象——热路径（每个 chunk、每个事件）上的派发不产生任何临时分配。这种"把运行时开销挤到构造期"的工程细节，正是这个"教学级架构"在真实产品里被重度使用后的痕迹。

4.8 本章小结 ​

driver 的本质：一个三态 phase 机 + 一个 while(true) 的 turn/step 嵌套循环 + 贯穿的 AbortSignal + 处处先行落盘的纪律。所有"智能"都在模型侧；dsh 的 agent-loop 不假装聪明，它只保证：可观察（每个边界都有事件）、可拦截（每个决策点都是瀑布）、可恢复（错误结构化、finally 闭环）。

小练习：给 turn() 找 bug

重读 4.4 的 turn() 节选，回答：1）如果 preStep 抛异常，turn/end 会以什么 reason 落盘？2）如果监听器在 agent/turn-stopping 里调用 agent.steer() 塞入新消息，代码第 295–299 行的两次 inbox.nextStep.length === 0 判断各起到什么作用？3）target 变量为什么第一次循环是 next-turn，之后是 next-step？

（提示：三个问题分别对应"finally 闭环"、"数据决定而非监听器顺序决定"、"轮次内插队"三个设计意图。）

基于 DeepSeek Harness（开发者预览版 0.1.0-rc.6）与 Cordis 撰写

教学用途 · 遵循 MIT 协议精神 · 一切皆插件