这是《DeepSeek Harness 权威指南 》系列第 13 篇，也是二开线（B 线）的第 6 篇。源码基线为 deepseek-harness @ 47f9438（v0.1.0-rc.5）。

前五篇都在讲“能力”：工具、adapter、策略、结果投影。这篇把镜头拉回驱动者：一个 agent 收到输入后，会话日志里到底发生了什么。作者 demo 使用官方测试同款依赖和一个 scripted adapter——它不调用任何真实 provider。

B4 里你看到了工具结果如何进入日志。但“谁发起调用、turn 怎么开、怎么收束”还没有连起来。本篇补上这一环。

一条 turn 的会话事件流

图：followup 把消息放进 inbox；driver 开 turn、claim 消息、跑 step、请求模型、执行工具，最后以 turn/end 收束。

一、先跑起来：五段事件流

作者本机项目（rex-hugo/.tmp-research/dsh-b5-agent，完整代码见 GitHub：rex-dhs-core/dsh-b5-agent ）

dsh-b5-agent/
└── agent-event-stream-demo.ts

运行：

node --import tsx E:/coding/rex-hugo/.tmp-research/dsh-b5-agent/agent-event-stream-demo.ts

1. 依赖组装：与官方测试同款

demo 挂载的插件集合与官方 packages/core/agent-loop/tests/agent.spec.ts:12-22 完全一致：

async function harness(): Promise<{ ctx: Context; adapter: ScriptedAdapter; agent: import('@deepseek-ai/dsh-agent').Agent }> {
  const ctx = new Context()
  await ctx.plugin(LlmRuntime)
  await ctx.plugin(SessionStore)
  await ctx.plugin(SystemPrompt)
  await ctx.plugin(ToolRuntime)
  await ctx.plugin(AgentRegistry)
  await ctx.plugin(AgentLoop, { agents: [] })
  await ctx.plugin(Commands)
  ctx.tools.register(greetTool)
  ctx.commands.register({
    name: 'ping',
    description: 'Local ping command.',
    handler: () => ({ kind: 'success', text: 'pong' }),
  })
  const adapter = new ScriptedAdapter([])
  ctx.llm.registerAdapter(['mock'], adapter)
  const agent = ctx.agentLoop.create(SessionId('b5-agent'), { provider: 'mock', model: 'mock' })
  return { ctx, adapter, agent }
}

逐行解释：

  • LlmRuntime 是 LLM 服务（B2 的主角）；SessionStore 提供会话；SystemPrompt 组装提示词；ToolRuntime 是工具注册表（B1）；AgentRegistry 提供 Agent 抽象；AgentLoop 才是真正驱动 turn 的循环。
  • ctx.agentLoop.create(SessionId(...), { provider: 'mock', model: 'mock' }) 创建一个 agent，路由到名为 mock 的 adapter。
  • ctx.llm.registerAdapter(['mock'], adapter) 把 author-local 的 scripted adapter 注册到 mock provider（B2 的路由机制）。
  • ctx.commands.register(...) 注册一个本地 /ping 命令（第六节）。

scripted adapter 是官方测试 MockAdapter 的同形副本（packages/core/agent-loop/tests/mock-adapter.ts）：每个模型请求消费脚本中的下一个条目，可能是 textResponse(...)、toolCallResponse(...) 或 'hang'。作者副本同样标注为 author-local：

class ScriptedAdapter extends LlmAdapter {
  requests: GenerateOptions[] = []
  constructor(private script: (StreamChunk[] | 'hang')[]) { super() }

  override resolveModel(provider: string, model: string): Promise<LlmResolvedModelInfo> {
    return Promise.resolve({ provider, id: model, name: model })
  }

  async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
    this.requests.push(options)
    const entry = this.script.shift()
    if (!entry) throw new Error('ScriptedAdapter: script exhausted')
    if (entry === 'hang') {
      yield { type: 'block-start', index: 0, blockType: 'text' }
      yield { type: 'text-delta', index: 0, text: 'partial' }
      await new Promise<void>((_resolve, reject) => {
        if (options.signal?.aborted) { reject(new Error('aborted')); return }
        options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
      })
      return
    }
    for (const chunk of entry) {
      if (options.signal?.aborted) throw new Error('aborted')
      yield chunk
    }
  }
}

'hang' 条目是关键：它流出一个 partial 分片后等待 abort signal——这正是第五节 cancel 场景的“悬停模型”替身。它不证明真实 LLM 会如何响应取消；它只证明 loop 会把 abort signal 传给 adapter 并收敛。

2. 订阅会话事件

demo 用 ctx.on('session/event', ...) 订阅该 agent 会话的全部事件，并压缩打印：

ctx.on('session/event', (session, event) => {
  if (session === agent.session) lines.push({ type: event.type, data: event.data })
})
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }))
await agent.whenIdle()

whenIdle() 等待 driver 完成当前所有活动；这是官方测试驱动 agent 的标准方式（agent.spec.ts 每个用例都用它）。

二、Agent 的三种输入与一种中止

ReactLoopAgent 的输入 API（packages/core/agent-loop/src/agent.ts:113-140）：

  send(message: UserMessage, target: InboxTarget, wakeup: boolean): void {
    // Waking input cannot join an aborted activity, so it starts the next turn.
    // Captured before the insertion so a reentrant cancel from a splice observer cannot reclassify it.
    const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted
    const resolvedTarget = wakingAfterAbort ? 'next-turn' : target
    this.inbox.splice(resolvedTarget, Infinity, 0, [message])
    if (wakeup) this.wakeDriver(wakingAfterAbort)
  }

  followup(input: UserMessage): void {
    this.send(input, 'next-turn', true)
  }

  steer(input: UserMessage): void {
    this.send(input, 'next-step', true)
  }

  inject(input: UserMessage): void {
    this.send(input, 'next-step', false)
  }

  cancel(cause: AgentCancelCause, options: CancelOptions = {}): void {
    if (!options.keepInbox) {
      this.inbox.clear()
      if (this.phase.kind !== 'idle') this.phase.wakeRequested = false
    }
    if (this.phase.kind !== 'idle') this.phase.abort.abort(cause)
  }

三句话总结：

方法inbox 目标wake语义
followupnext-turntrue开启新 turn：用户提问
steernext-steptrue当前 turn 的 step 边界进入；idle 时退化为 woken prompt turn
injectnext-stepfalse只入栈不唤醒：插件上下文、文件变更通知
cancel——清空 inbox（可 keepInbox）+ abort 当前 phase

steer 的“idle 退化”不是文档话术：demo 的 C 段就是 idle 时 steer，结果与 followup 一样跑完一个完整 turn（见下）。

输入目标与收束方式对照

图：输入进 inbox 的目标决定“何时进入”；turn/end 的 reason 决定“怎么收束”；command 事件绕过 turn 直接落日志。

三、turn 生命周期：事件序列逐行解读

demo A 段（一次 followup('hi')）的完整事件流如下（节选：... 与 {...} 为作者缩写，其余每一行与真实 stdout 逐字一致；完整 31 行见本机文件 agent-event-stream-demo-output.txt）：

=== A: followup text turn ===
  agent/inbox/spliced  {"target":"next-turn","start":0,"inserted":[{...text:"hi",source:{kind:"user"}...}]}
  turn/start  {"turn":1}
  agent/inbox/spliced  {"target":"next-turn","start":0,"removedCount":1,"inserted":[]}
  step/start  {"turn":1,"step":1}
  user/message  "hi"
  request/header  {config:{provider:"mock",model:"mock"}, system:"You are an AI agent powered by DeepSeek Harness.", tools:[greet schema], reason:"initial"}
  request/context  {"provider":"mock","model":"mock"}
  assistant/chunk  block-start
  assistant/chunk  "H"
  assistant/chunk  "e"
  ...（每个字符一条 text-delta）...
  assistant/chunk  block-end
  assistant/chunk  usage
  assistant/chunk  finish
  assistant/message  text:"Hello from mock"
  step/end  {"turn":1,"step":1}
  turn/end  completed
  status=idle modelRequests=1

这条序列值得逐行读一遍：

  1. agent/inbox/spliced（insert）：消息进入 inbox，目标 next-turn。注意消息带着自己的 id 与 source: {kind:'user'}——来源在入口就被记录。
  2. turn/start {turn:1}：driver 开启 turn 1。
  3. agent/inbox/spliced（claim）：removedCount:1 表示消息被 claim 出队。
  4. step/start {turn:1, step:1}：step 是“一次模型调用 + 它请求的工具执行”。注意 turn 与 step 都从 1 开始编号。
  5. user/message：消息进入模型可见 surface（B4 的投影规则）。
  6. request/header + request/context：本次请求的完整配置快照（provider/model/system/tools）与路由元数据。reason:"initial" 表示这是该会话第一份 header。
  7. assistant/chunk × N：adapter 流出的每个 StreamChunk 逐条落日志——token 级回放保真就在这里（一个字符一条 delta）。
  8. assistant/message：组装后的完整 assistant 消息（surface 事件）。
  9. step/end → turn/end completed：step 收束、turn 收束，status=idle。

要点：模型请求本身没有单独的事件，它体现为 request/header 快照 + 随后的 chunk 流；assistant/message 才是进入 transcript 的组装结果。

四、工具调用如何进入事件流

demo B 段（followup('greet Rex')，脚本为 toolCallResponse → textResponse('Done')）的关键部分：

=== B: tool-call turn (loop continues after tool/result) ===
  assistant/chunk  tool-call-delta
  assistant/chunk  tool-call-delta
  assistant/chunk  block-end
  assistant/chunk  finish
  assistant/message  tool-call
  tool/call  greet({"name":"Rex"})
  tool/result  Hello, Rex!
  step/end  {"turn":1,"step":1}
  step/start  {"turn":1,"step":2}
  assistant/message  text:"Done"
  step/end  {"turn":1,"step":2}
  turn/end  completed
  status=idle modelRequests=2

与 A 段的差异只有一处，但含义重大：

  • 模型流以 finish(tool-calls) 结束，assistant/message 是 tool-call 块；
  • loop 把工具调用送进 registry（B1-B4 的全部流水线），落 tool/call（arguments 原文）与 tool/result（Hello, Rex!）；
  • 然后 loop 没有收束 turn，而是在同一个 turn 里开 step 2 再请求一次模型——模型这次输出文本 Done，turn 才以 completed 收束；
  • modelRequests=2：一个 turn 内两次模型调用，这就是“工具结果回填后继续对话”的机制。

五、cancel：aborted 收束

demo D 段让 adapter 停在 'hang'（已流出 partial），然后 agent.cancel({kind:'user'})：

=== D: cancel aborts the turn ===
  assistant/chunk  block-start
  assistant/chunk  "partial"
  step/end  {"turn":1,"step":1}
  turn/end  aborted:user
  status=idle modelRequests=1
  • turn/end 的 reason 是 aborted，cause 是 user；
  • 悬停的 adapter 流收到 abort signal 后以 aborted 错误收敛（ScriptedAdapter 的 abort listener）；
  • agent 回到 idle，whenIdle() 正常返回。

这条路径说明取消是协作式的：loop 通过 exec.signal / options.signal 通知下游，但下游必须自己收敛。真实 provider 的取消行为、资源释放时序都要在真实 endpoint 上另行验证。

六、slash command：绕过 turn 的日志事件

命令注册与执行（packages/interaction/commands/src/index.ts:296-338）——commands.execute(agent, line, signal)：

  @Remote
  async execute(
    agent: Agent,
    line: string,
    signal: AbortSignal,
  ): Promise<CommandExecution | undefined> {
    const parsed = parseCommand(line)
    if (parsed === undefined) return undefined
    const command = this.view(agent).get(parsed.name)
    if (command === undefined) return undefined
    if (signal.aborted) throw abortError(signal)
    const commandId = this.mintCommandId()
    this.appendLifecycle(agent.session, 'command/run', {
      commandId,
      name: parsed.name,
      ...command.definition.recordInput === false ? {} : { args: parsed.rawInput },
      source: { kind: 'user' },
    })
    const invocation = Object.freeze({ commandId, agent, rawInput: parsed.rawInput, signal })
    let result: CommandResult
    try {
      const output = command.definition.handler(invocation)
      result = normalizeResult(parsed.name, await withAbort(Promise.resolve(output), signal))
    } catch (error: unknown) {
      ...
      throw error
    }
    this.appendLifecycle(agent.session, 'command/done', {
      commandId, kind: result.kind,
      ...result.text === undefined ? {} : { text: result.text },
      ...
    })
    return Object.freeze({ commandId, result })
  }

事件 payload（packages/interaction/commands/src/types.ts:88-100）：

    'command/run': { commandId: CommandId; name: string; args?: string; source: CommandSource }
    'command/done': {
      commandId: CommandId
      kind: 'success' | 'error'
      text?: string
      sourceEventSeq?: number
    }

demo E 段的真实输出：

=== E: slash command ===
  execution={"commandId":"cmd-4437f190-1","result":{"kind":"success","text":"pong"}}
  command/run  ping  hello
  command/done  success "pong"

四个要点：

  1. command/run 与 command/done 是log-only事件，配 commandId 成对——官方注释明确说它们“never model surface”（绝不进入模型 transcript）；
  2. 它们不经 turn 包装：命令不是模型输入，handler 直接执行；
  3. parseCommand 在入口完成解析，args 是未再解析的原文（含分隔空白）；
  4. 语法错误或未知命令名时什么都不落日志（“admission misses … log nothing”）——未进入 handler 的调用没有审计记录。

七、常见错误写法

过强或错误的说法更准确的表述
“inject 会唤醒 agent。”inject wake: false，idle 时只入栈不开 turn。
“steer 只能在运行中有效。”idle 时 steer 退化为 woken prompt turn（demo C 实证）。
“cancel 会立刻终止模型。”cancel 发送 abort signal；adapter 必须协作收敛，真实 provider 行为需单独验证。
“command 会发给模型。”command/run 与 command/done 是 log-only，不进入 transcript。
“每次用户输入都是一个 step。”一次输入开一个 turn；一个 turn 内可以有多个 step（工具链续跑）。
“assistant/chunk 可以直接当消息用。”chunk 是 token 级回放记录；进入 transcript 的是组装后的 assistant/message。
“本机 scripted adapter 证明真实 provider 行为。”它只证明 loop 与 adapter 的契约（信号、chunk、finish），不证明任何真实模型。

八、下一步

B5 的核心结论：agent 循环是一个事件生产者。输入经 inbox 按目标（next-turn / next-step）进入，turn 与 step 由 driver 开合，模型与工具的结果都变成会话日志里可回放的事件，命令则绕过 turn 直接落成成对日志。

设计自己的集成时，先问：

  1. 输入应该开新 turn（followup）还是插进当前 step（steer / inject）？
  2. 我的 UI 如何订阅事件流？（session/event + agent/status）
  3. 取消路径是否协作？（adapter / 工具是否转发 signal 并收敛）
  4. 命令该进 transcript 还是只进日志？（注册为 slash command 就是后者）

下一篇：B6：把二开能力打包成可交付插件

一条 turn 从输入到收束，全部事实都在会话日志里；followup 开 turn、工具续跑同 turn、cancel 以 aborted 收束、command 只留成对日志——记住这四个边界，就不会再把“驱动者”和“事件流”混为一谈。