Agent 与 Agent Loop：turn、step、inbox

整个 harness 里唯一一段具体循环逻辑，逐行走读

本章基准：dsh 0.1.1-rc.2，commit b150a551b8d465e31e418e1b2eaf5e79bbb7d28e

主要源码：packages/core/agent/src/（4 个文件、约 1338 行）、packages/core/agent-loop/src/（7 个文件、约 1960 行）

前三章讲的是"骨架"：Cordis 的四个原语、patch 分层怎么把一堆插件装成一个产品。这一章开始讲"肉"——真正让模型转起来的那段代码。

dsh 官方对这个包的定位写得很直白（packages/core/agent-loop/README.md）：

This is the only package in the harness that contains concrete loop logic. Everything else is an abstract service or a plugin against extension points — new behavior goes into plugins, not here.

整个 harness 里只有这一个包含具体循环逻辑的包。 227 个 workspace package 里，只有 dsh-agent-loop 一个知道"什么叫一步"。这句话是第 1 章"Everything is a Plugin"的另一面：插件化不是把逻辑打散，而是把逻辑收缩到一个点，然后在这个点周围开出足够多的扩展孔。

所以本章要回答的是三个问题：

1. 这个"唯一的循环"到底长什么样？（ReactLoopAgent.turn() / step() 的逐行走读）
2. 外部输入是怎么在不打断循环的前提下插进来的？（Inbox 的两条队列、send(target × wakeup)）
3. 为什么它要长这么复杂？（每一处 signal.throwIfAborted()、每一个 finally 背后都有一个真实的竞态）

dsh web ──▶ AgentLoop 服务 ──create/resume──▶ ReactLoopAgent（唯一的具体驱动器）
                                                │
                              ┌─────────────────┴──────────────────┐
                              │  while (await this.turn()) {}      │
                              │    turn: 开 turn → 循环 step        │
                              │      step: 组装 → 请求 → 工具        │
                              └────────────────────────────────────┘
                                        ▲               │
                                  Inbox │               │ SessionEvent
                     next-turn / next-step 两条队列       ▼ append-only 日志

全章图示：assets/ch04-turn-step-inbox.svg

1. 三个时间单位：session、turn、step

先把词对齐。这三个词在很多 Agent 框架里被混着用，dsh 里它们是三个有精确边界、且都写进日志的东西。

关键在第三行。一个 step = 一次模型请求。 不是"一次用户交互"，也不是"一条消息"。模型返回文本、没有 tool call → step 结束、turn 也结束；模型返回 3 个 tool call → 执行完 3 个工具，回到 while 顶部，开新的 step（第二次模型请求），把工具结果喂回去。所以"一个 turn 里有几个 step"等于"这轮任务里模型被调用了几次"。

为什么要给这两层都发日志事件、都编号？因为几乎所有横切功能都需要在其中一个尺度上挂钩：

• 压缩（第 10 章）关心的是 step 边界——请求快撑爆上下文窗口了；
• 权限审批（第 9 章）关心的是工具调用，落在 step 内部；
• 会话标题生成、遥测关心的是 turn 结束；
• 用户"打断并改口"关心的是 step 边界（我不想等你把整个任务做完）；
• 用户"追加一个新任务"关心的是 turn 边界（等你把手上这个做完）。

最后两条直接决定了 Inbox 为什么是两条队列。

还有一个词值得澄清：dsh 里没有"round"这个一等概念。step 内部的 LLM 重试（agent/request-error 返回 { kind: 'retry' }）会在同一个 step 里重新发一次请求（agent-loop/src/agent.ts:339 的 while (true)），不递增 step 号——因为那不是"模型又说了一次话"，那是同一次说话的重试。这个区分很讲究：step 号是模型可见历史的坐标，重试不产生新的模型可见历史。

2. Inbox：两条队列，两个时间尺度

2.1 一个矩阵，三个别名

Agent 接口上只有一个真正的输入原语（packages/core/agent/src/runtime-types.ts:117）：

send(message: UserMessage, target: InboxTarget, wakeup: boolean): void

InboxTarget 只有两个值（packages/core/agent/src/types.ts）：

export type InboxTarget = 'next-turn' | 'next-step'

两个维度做笛卡尔积，就是 dsh 全部的输入语义。三个公开别名是这个矩阵的固定预设（agent-loop/src/agent.ts:122-132）：

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) }

第四格没有别名但可用——这是"原语 + 预设"设计的直接好处：send() 是完备的，别名只是可读性糖。任何插件想要一个新的输入语义，不需要改 agent-loop，选一格 send() 就行。

inject 那一格（next-step + 不唤醒）尤其重要，它是"模型可见 ⟺ 已落日志"这条铁律的落地机制之一：插件想给模型加一段上下文，唯一的合法通道是往 inbox 里塞一条 UserMessage，而 inbox 的每次变动都会写日志。想绕过日志直接改请求消息？agent/request waterfall 只给你改配置，不给你改 messages（runtime-types.ts:236 的 JSDoc 明确写了 this waterfall cannot mutate messages）。

2.2 Inbox 是投影，不是状态

这是最容易看漏的一点：Inbox 不是一个内存队列，它是日志的投影（packages/core/agent/src/inbox.ts:25，类注释就一句：A replay-once projection that incrementally consumes later inbox splices）。

构造函数（inbox.ts:29）从会话日志重放：

for (const event of session.events.slice(session.header.seedLength ?? 0)) {
  if (event.type !== 'agent/inbox/spliced') continue
  try {
    this.apply(event.data)
  } catch (error: unknown) {
    throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error })
  }
}

所以进程重启、resume 一个持久化会话，用户上次没跑完就排在队列里的消息还在。这不是额外做的功能，这是"inbox 是投影"的免费推论。

而每一次修改，先写日志、后改内存（inbox.ts:158 的 mutate()）：

const event = this.session.append('agent/inbox/spliced', splice)
const removed = inbox.splice(actualStart, actualDeleteCount, ...event.data.inserted)

顺序是刻意的，JSDoc 说明了原因（inbox.ts:132）：

The durable event commits before the live projection mutates, so synchronous session/event observers see the pre-splice lists and can reconstruct the removed messages from the normalized coordinates.

同步观察者拿到事件时，内存投影还是变动前的状态。事件里带的是 { target, start, removedCount, inserted } 这样的规范化坐标，观察者拿坐标一减，就能算出被删掉的是哪几条消息——而事件本身不需要冗余地存一份"被删的消息内容"。这是一个很省的设计：用"事件 + 变动前状态"表达删除，而不是把删除的内容也写进日志。

mutate() 里还有一整段在做 Array.prototype.splice 的语义归一化（inbox.ts:169-181）：负 start 折回、NaN 归零、deleteCount 截断到边界。为什么这么讲究？因为归一化后的坐标要写进持久化日志并在下次重放，start: -1 这种相对坐标在重放时含义会漂移。所以日志里只允许绝对坐标，归一化必须发生在写事件之前。

validate()（inbox.ts:203）在写事件之前做两件事：坐标是否越界，以及——MessageId 在两条队列合起来范围内是否唯一：

for (const message of splice.target === 'next-turn'
  ? [...candidate, ...this.nextStep]
  : [...this.nextTurn, ...candidate]) {
  if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
  ids.add(message.id)
}

跨队列唯一，所以 remove(messageId) / replace(messageId, …) 这类按身份操作的 API 不需要调用者告诉它在哪条队列里（locate() 自己扫两条）。UI 上"取消我刚才排的那条消息"就是这么实现的。

2.3 claim：turn 边界的原子领取

claim()（inbox.ts:71）是驱动器每个 step 边界的读操作，它的签名很说明问题：

claim(target: InboxTarget, turn: number): UserMessage[] {
  const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)
  if (target === 'next-turn') {
    claimed.push(...this.mutate('next-turn', 0, 1, [], false))
  }
  for (const message of claimed) this.notifications.claimed(message, turn)
  return claimed
}

三个细节：

① 无论在 turn 边界还是 step 边界，next-step 总是被全部取走。 只有 turn 边界会额外取 next-turn 的一条（0, 1）。这就是"一条 follow-up 消息独占一个 turn"，而 steering/injection 是"搭便车"。

② 领取用的是 discardRemoved: false。 同一个 mutate()，claim 传 false，用户取消传 true。差别在日志里：取消会带 outcome: 'canceled'，领取不带（纯删除）。这一个字段撑起了第 10 节要讲的 foldConsumedWork —— "这批输入被跑掉了"和"这批输入被丢掉了"必须在日志上可区分。

③ 领取顺序是 next-step 在前、next-turn 在后。 于是模型看到的消息顺序是"先 steering，后新任务"。而 clear()（inbox.ts:58）反过来先清 next-step 再清 next-turn——它有一句注释说这是刻意的（clearing next-step before next-turn）：取消时先掐掉最紧急的那条通道。

3. 驱动器的相位机：idle / maintenance / running

ReactLoopAgent 只有一个可变状态（agent-loop/src/agent.ts:38-46）：

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

而对外只有两个状态（runtime-types.ts:50）：

export type AgentStatus = 'idle' | 'running'

maintenance 对外报告为 idle（agent.ts:99）。这个刻意的信息隐藏是给"两次 turn 之间的活儿"留的位置：压缩（第 10 章）、生成会话标题、写持久化快照——这些工作要独占 agent（不能一边压缩历史一边发请求），但从用户视角 agent 确实是闲着的，UI 不该显示"思考中"。

runMaintenance()（agent.ts:142）就是这个位置的门：

if (this.phase.kind !== 'idle') throw new Error(`agent "${this.id}" already has active work`)

同步抛错，不排队。抢不到就是抢不到——因为"排队等 maintenance"的语义没有拥有者，与其猜，不如让调用方自己决定重试还是放弃。

3.1 wake latch：一个真实 bug 留下的疤

wakeDriver()（agent.ts:172）是本章最值得细看的 40 行。它的注释里带了一个 Agent Note 链接（2026-08-07-cancel-convergence-wake-latch.md），说明这段逻辑是从一个真实缺陷里长出来的。

问题是这样的：用户点了"停止"，驱动器的 AbortController 已经 abort，但驱动器还在收敛（正在跑 finally、正在等工具 drain）。此时用户又敲了一条新消息。这条消息不能加入正在死掉的那个活动，但也不能丢掉。

send()（agent.ts:113）先给这条消息定性：

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)

注意注释：Captured before the insertion so a reentrant cancel from a splice observer cannot reclassify it。分类必须发生在插入之前——因为插入会同步写日志、同步派发 session/event，而某个观察者完全可能在回调里再喊一次 cancel()。如果分类在插入之后读 phase，同一条消息就会被两次不同的读法定性成两个结果。

然后 wakeDriver() 决定这次唤醒是"立刻起飞"还是"闩住等收敛"：

if (this.phase.kind !== 'idle') {
  const reason = this.phase.abort.signal.reason as AgentCancelCause | undefined
  if (reason?.kind !== 'disposed' && (this.phase.kind === 'maintenance' || wakeAfterAbort)) {
    this.phase.wakeRequested = true
  }
  return
}

三种情况分得很干净：

• maintenance 中 → 闩住，等 maintenance 结束时重放（agent.ts:158）；
• 已 abort 的 running 中 → 闩住，等 kick() 的 finally 重放（agent.ts:220）；
• disposed 原因的取消 → 不闩。 这条最关键：销毁中的 agent 如果还闩着唤醒，dispose() 里的 await machine.whenIdle() 就会等一个刚被自己唤醒的新 turn，teardown 永远等不完。注释直说了：disposal never latches, so teardown waits on no model turn。

而重放时还要再确认一次（agent.ts:220）：

if (wakeRequested && this.inbox.hasPending) this.wakeDriver()

闩了 + 队列里确实还有东西，才真起飞。因为闩住期间那条消息完全可能被用户撤销掉。

反过来，如果 agent 本来就 idle，wakeDriver 会无条件开一个 turn，即使那条消息随后被清掉了。这是刻意的（agent.ts:164-169 的 JSDoc）：A wake sent while idle always opens its turn boundary, even when its message was cleared。为什么？因为"用户发了一条消息"这件事本身应该在日志里留下一个 turn 边界；后面 turn() 里会有专门的分支处理"claim 出来是空的"（第 4 节）。只有闩住的重放才需要二次确认——因为那个闩本身可能已经过期了。

3.2 whenIdle 的 do-while

async whenIdle(): Promise<void> {
  let activity: Promise<void>
  do {
    await (activity = this.activityDone)
  } while (activity !== this.activityDone)
}

为什么不是简单 await this.activityDone？因为 latch 会在旧活动收敛的同一时刻接上新活动，activityDone 被换成了新的 promise。循环到"await 完之后字段没被换过"才算真静止。

这个模式（await 一个可变的 promise 字段，直到它不再变）在任何"收敛可能触发后继工作"的系统里都值得记住——它比"计数器 + 条件变量"更难写错。

4. turn()：一次 turn 的逐行走读

turn()（agent.ts:246）是 85 行的一个方法，返回 boolean：还有没有下一个 turn。外层就一句（agent.ts:212）：

while (await this.turn()) {}

按执行顺序拆开看。

① 开 turn。

const turn = phase.turn + 1
try {
  this.session.append('turn/start', { turn })
} catch (error: unknown) {
  this.throwError(error)
}
phase.turn = turn

注意 append 被 try 包住了。日志写入本身可以失败（持久化后端拒绝、事件类型未注册），而这类失败必须走 throwError() —— 它先发 agent/error 事件再抛（agent.ts:203），保证 UI 能看到"turn 都没开起来"这件事。

② 内层 while：一个 turn 里的多个 step。

let turnEnds: TurnEndReason | null = null
let target: InboxTarget = 'next-turn'
while (true) {
  signal.throwIfAborted()
  const step = phase.step + 1
  const decision = await this.preStep(target, { turn, step })

target 只有第一次是 next-turn，循环末尾会改成 next-step（agent.ts:300）。这行代码就是"一个 turn 只消费一条 follow-up"的全部实现。

③ pre-step 拒绝。

if (decision.kind === 'reject') {
  turnEnds = { kind: 'blocked' }
  return false
}

插件（比如 plan mode、权限守卫、hooks）可以在模型请求发出之前否掉这一步。此时 turn 以 blocked 结束，而且已经领取的消息不会退回 inbox。这个语义很反直觉，但 dsh 在 runtime-types.ts:188 明确写了：

If the proposed step is rejected, the claimed message ends here: it is neither discarded nor re-emitted as a user/message, and the turn closes without a step.

为什么不退回？因为退回意味着下一次唤醒会再次领取、再次被同一个插件拒绝——一个无限循环。让它就地终止，并且在 turn/end 上留下 blocked，是可诊断的。第 10 节会看到 foldConsumedWork 专门把 blocked 算作"对这批输入负了责"。

④ 两个空批次分支。 这两个 if 是本方法最烧脑的地方：

if (turnEnds && decision.messages.length === 0) break
// A removed waking message or an enter decision rewritten to empty
// still owns the initial turn boundary, but it spends no model call.
if (phase.step === 0 && decision.messages.length === 0) {
  turnEnds = { kind: 'completed' }
  return false
}

• 第一个：已经跑过至少一个 step（turnEnds 非空）且这次没新东西 → 正常收尾。
• 第二个：一个 step 都没跑（phase.step === 0）且领出来是空的 → 这就是 3.1 节留下的尾巴。唤醒时那条消息已经被撤销了，或者 agent/pre-step 把消息改写成了空数组。turn 边界照样开、照样关（日志里留下一对 turn/start/turn/end），但一次模型调用都不花。

第二个分支是"idle 唤醒无条件开 turn"这个决定的必然代价，dsh 选择付这个代价，换来"每条用户输入在日志里都有对应的 turn 边界"这个更强的性质。

⑤ 跑一个 step。

this.session.append('step/start', { turn, step })
phase.step = step
try {
  for (const message of decision.messages) {
    this.session.append('user/message', message, { surfaceOp: 'append' })
  }
  const stepEnd = await this.step(decision.assembly)
  if (turnEnds === null || turnEnds.kind !== 'max-tokens') turnEnds = stepEnd
} finally {
  this.session.append('step/end', { turn, step })
}

三个点：

• user/message 是在 step/start 之后才落日志的。 消息离开 inbox（claim）和进入模型可见历史（user/message）是两个不同的事件、两个不同的时刻。中间那段就是 agent/pre-step 的窗口——它能改写、能拒绝，而这两件事都发生在消息"正式成为历史"之前。
• step/end 在 finally 里。 无论 step 是正常完成、抛错还是被取消，step/end 一定配对。日志的括号必须永远闭合，否则 deriveMessages()（第 5 章）和 fork 校验（要求"没有未闭合的 turn/step"）就全乱了。
• max-tokens 是粘性的。 一旦某个 step 撞了输出上限，后面的 step 就算正常完成也不能把 turn 的结局降级回 completed。因为对调用方来说"这轮输出被截断过"是必须知道的事实。

⑥ turn-stopping：数据说话。

if (turnEnds && this.inbox.nextStep.length === 0) {
  await this.dispatch.serial('agent/turn-stopping', { turn, signal })
  signal.throwIfAborted()
}
if (turnEnds && this.inbox.nextStep.length === 0) break
target = 'next-step'

同一个条件判断了两次，中间夹着一次 serial 派发。这是 dsh 里最漂亮的一个设计，runtime-types.ts:261 的 JSDoc 讲得比代码清楚：

a listener that objects steers (agent.steer(...)) and the machine re-reads its inbox: fresh steering runs another step, none closes the turn. Data decides, so listener order cannot change the outcome.

一个想阻止 turn 结束的监听器（典型例子：loop-hygiene 守卫发现模型说"我做完了"但 todo 列表还没清空）不返回 false、不抛异常、不调用什么 preventDefault()——它调 agent.steer()，往 inbox 里塞一条真实的 steering 消息。然后循环重新读 inbox：有东西就再跑一个 step，没东西就收工。

好处是决定性的：监听器顺序不影响结果。 三个监听器都想续命，结果是塞了三条 steering，跑一个 step 就都消费掉了；一个塞了一个撤销了，最终看 inbox 里剩什么。如果换成"返回 false 表示拒绝结束"，那么就必须回答"A 说结束 B 说不结束怎么办"、"顺序颠倒结果一样吗"这些没有正确答案的问题。

反方向的控制也是数据：工具结果可以带 concludesTurn（见第 8 节），提前结束一个工具循环——同样不是短路，而是往数据里写一个事实。

⑦ 收尾。

} catch (error: unknown) {
  if (signal.aborted) {
    turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }
    throw error
  }
  turnEnds = {
    kind: 'error',
    error: error instanceof LlmError ? error.failure : { message: errorChain(error), code: 'UNKNOWN' },
  }
  this.throwError(error)
} finally {
  this.session.append('turn/end', { turn, reason: turnEnds! })
}

每一条退出路径都赋了 turnEnds，finally 无条件写 turn/end。错误被结构化：LlmError 保留它自己的 failure（provider 返回的 code、message、可重试性），其他一切压成 errorChain(error) 文本 + UNKNOWN code。这样"这个 turn 为什么失败"永远是日志里的一个可读字段，而不是一句只在控制台里出现过的堆栈。

最后：

if (!this.inbox.hasPending) return false
phase.abort = new AbortController()
phase.wakeRequested = false
phase.step = 0
return true

换一个全新的 AbortController。 上一个 turn 可能是被 abort 掉的，那个 controller 已经污染；下一个 turn 必须从干净的取消状态开始。同时把 wakeRequested 清零——因为闩是设在旧 controller 的语境里的，而现在活的驱动器自己就会去领队列，注释说得很准：A fresh controller makes a latch set on the old one stale。

数一下这个方法里的 signal.throwIfAborted()：7 次（agent.ts:252, 264, 278, 294, 297 以及 preStep 里的两次）。每一处都对应一个"这里有个 await，取消可能在 await 期间到达"的位置。这种密度不是防御性偏执，而是 await 语义的必然要求：JS 里每一个 await 都是一个可以被外部世界改变的时间点。

5. preStep()：组装、投影、瀑布

private async preStep(target: InboxTarget, position: { turn: number; step: number }): Promise<PreparedStep> {
  const signal = this.phase.abort.signal
  const claimed = this.inbox.claim(target, position.turn)
  const assembly = await this.loopCtx.systemPrompt.assemble(assembleContextFor(this, signal))
  signal.throwIfAborted()
  const sections = renderContextSections(assembly)
  const context = this.runtimeContext.project(joinContextSections(sections), sections)
  const decision = await this.dispatch.waterfall(
    'agent/pre-step', { messages: claimed, ...position, signal },
    (): Promise<PreStepDecision> => Promise.resolve<PreStepDecision>({
      kind: 'enter',
      messages: context === undefined ? claimed : [...claimed, context],
    }),
  )
  signal.throwIfAborted()
  return decision.kind === 'reject' ? decision : { ...decision, assembly }
}

四步：领取 → 组装系统提示词与工具（第 6 章）→ 投影运行时上下文 → 过瀑布。

assembleContextFor(this, signal)（agent/src/dispatch.ts:174）是个只有三行的小函数，但它的存在有一条硬理由（JSDoc）：

Build the prompt assembly context with agent and scope set together, so agent-scoped prompt and tool contributions cannot be silently omitted.

agent 和 scope 必须同时设，否则 agent 作用域的贡献会被静默漏掉。把"两个字段必须一起设"变成"一个函数调用"，是消灭这类 bug 的标准手法。

5.1 RuntimeContextProjection：让"当前状态"不变成 KV-cache 杀手

runtime-context.ts 这 76 行解决一个非常现实的问题。

系统提示词里的动态部分（当前时间、cwd、git 分支、plan 状态、todo 列表……）每个 step 都可能变。如果把它放进 system prompt，那 system prompt 每步都变，整个 KV cache 每步都失效——这是性能和成本上的灾难。

dsh 的做法：把动态上下文作为一条 user/message 追加在历史末尾，前缀完全不动。而 RuntimeContextProjection 负责避免这条消息重复堆积：

project(current: string, sections: readonly ContextSnapshotSection[]): UserMessage | undefined {
  if (this.retained === undefined && current.length === 0) return
  const snapshot = current.length === 0 ? CLEARED : current
  if (this.retained?.text === snapshot) return
  return createUserMessage({ /* … */ })
}

• 内容和上次保留的快照一样 → 返回 undefined，不加消息；
• 变了 → 生成新快照；
• 从有变成空 → 不是什么都不做，而是发一条明确的"作废"消息：

const CLEARED = 'Current runtime context: none. Earlier runtime-context snapshots no longer apply.'

这一条很讲究。历史里还留着上一条快照（append-only 日志删不掉），模型会看见它。如果什么都不说，模型就会以为那个旧快照还有效。所以必须显式地告诉模型"忘了它"。这是 append-only 架构里最典型的一类推理：你不能删除，你只能补充一条让旧事实失效的新事实。

而"上次保留的快照是哪条"这个状态自己也是从日志重建的（构造函数里从后往前扫 user/message，并检查 session.surface），并且通过监听 session/event 保持同步。为什么要检查 surface？因为压缩（第 10 章）可能已经把那条旧快照从模型可见表面上换掉了——那它就不算"保留中"，下一步应该重新发一条。

6. step()：一次模型请求的全过程

step()（agent.ts:332）的骨架是 while (true) —— 为重试而存在。

① 构造请求并调用。

const { request, preparedCall } = await this.buildRequest(
  turn, step, assembly.tools, system, this.session.deriveMessages(), signal,
)

this.session.deriveMessages()。这一行是整个 dsh 架构的心脏：消息数组从来不是被维护的状态，它是每次现算出来的日志投影。 没有 this.messages.push(...)，没有"历史数组"。第 5 章会详细讲 deriveMessages() 怎么从 append-only 事件流算出 provider 要的 messages，以及压缩、fork、tool-result pruning 为什么可以只改日志表面就生效。

② 流式响应逐块落日志。

const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)
for await (const chunk of stream) {
  signal.throwIfAborted()
  chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)
  assembler.push(chunk)
}

每个 chunk 都是一个持久事件，并且记下它的 seq。后面完整的 assistant/message 会带上 sourceEventSeqs: chunkSeqs——这是"这条完整消息由这些 chunk 合成"的溯源关系。UI 靠它做增量渲染和重放对齐。

③ 被打断时保住已经生成的部分。

if (signal.aborted) {
  const content = assembler.interruptedBlocks()
  if (content.length > 0) {
    this.session.append('assistant/message', { /* … */ interrupted: true, /* … */ })
  }
}
throw error

用户中途点了停止，模型已经吐出的半句话要留在历史里并标记 interrupted: true。否则下一个 turn 的模型会看到自己上次凭空消失了一段，行为会很奇怪。interruptedBlocks() 负责把半成品块收成合法的 content（截断的 tool-call 参数不能当有效 tool-call 交出去）。

④ 失败：一个可插拔的重试决策点。

if (finish.kind === 'error' || finish.kind === 'aborted') {
  const action = await this.dispatch.waterfall('agent/request-error', {
    turn, step, provider: request.provider, failure: finish.failure,
    retryPolicy: preparedCall?.retryPolicy, signal,
  }, () => Promise.resolve<RequestErrorAction>(undefined))
  if (action?.kind !== 'retry') {
    throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
  }
  continue
}

默认值是 undefined —— 循环自己不重试。重试策略是插件的事（retryPolicy 作为素材递给它）。continue 回到 while 顶部，重新 buildRequest()（所以重试会重新走 agent/request 瀑布、重新算 messages——如果期间发生了压缩，重试就用压缩后的历史，这正是想要的），但 step 号不变。

⑤ 成功：落 assistant/message，然后判断这个 step 怎么收。

if (finish.kind === 'max-tokens') return { kind: 'max-tokens' }
const toolCalls = message.content.filter(block => block.type === 'tool-call')
if (toolCalls.length === 0) return { kind: 'completed' }
const { concluded } = await executeToolCalls(/* … */)
return concluded ? { kind: 'completed' } : null

三个返回值，语义精确：

null 是"这个 step 结束了但 turn 没结束"的编码。回到 turn() 里那个 if (turnEnds && …)，turnEnds 为 null 意味着两个 break 条件都不成立，循环继续——工具结果需要喂回模型。整个 ReAct 循环就靠这一个 null 驱动。

7. buildRequest()：请求头的折叠与"变了才记"

buildRequest()（agent.ts:426）是很容易被当成样板代码略过的一段，但它承载了一个重要的日志设计：请求配置不是每次都记，而是"变了才记"。

const seedConfig = deepFreeze(structuredClone(
  this.requestHeaderLogged ? requestProposal(persistedHeader!) : { ...route, /* … */ },
))
const proposedConfig = await this.dispatch.waterfall('agent/request', { turn, step, signal },
  () => Promise.resolve(seedConfig))

第一次请求的种子来自 AgentOptions（配置里声明的 provider/model/maxTokens）；之后的种子来自日志里已记录的 header。所以插件在第 3 步把模型换成了 deepseek-reasoner，第 4 步的种子就是 reasoner——切换是有状态的，而这个状态存在日志里，不在内存变量里。resume 一个会话，模型选择自动延续。

requestProposal()（agent.ts:55）在把 header 交给插件之前，会剥掉 adapter 填的默认值：

if (header.adapterDefaults.reasoningEffort === true) delete proposal.reasoningEffort
if (header.adapterDefaults.maxTokens === true) delete proposal.maxTokens

为什么？因为"某个 adapter 替你填的 maxTokens: 8192"和"用户显式要求 8192"是两件不同的事。如果不剥，第一次请求后 adapter 的默认值就会固化成显式配置，换模型时不会重新解析。adapterDefaults 这个布尔标记的全部意义就是区分这两者。

然后是折叠：

const baseline = this.session.requestHeader()
if (!this.requestHeaderLogged) {
  this.session.append('request/header', { header, reason: baseline === undefined ? 'initial' : 'resume' })
  this.requestHeaderLogged = true
} else if (baseline === undefined || !headerEquals(baseline, header)) {
  this.session.append('request/header', { header, reason: 'change' })
}

三个 reason：initial（全新会话的第一次）、resume（这个驱动器实例接手了一个已有日志）、change（真的变了）。一个跑了 40 个 step 的会话，如果模型和工具集一直没变，日志里只有一条 request/header。要复原第 37 步用的是什么配置，往前找最近的 header 就行。

这是 append-only 日志里"状态字段"的标准处理：记录变化，不记录快照。 同样的手法在紧接的 request/context（provider/model/contextWindow）上再用了一次。

顺带一提：markAgentLoopRequest(deepFreeze({ … }))。请求对象被冻结，并且打上标记进一个进程内的身份集合。dsh-agent-loop/invariant companion 用它做运行时不变量校验——独立地从日志重建一遍消息边界和请求头，跟实际发出的请求比对。"模型可见 ⟺ 已落日志"这条规则有一个自动化的看门狗。

8. 工具调度：模型顺序、滚动池、屏障

tool-calls.ts 的 289 行只做一件事：把一个 step 里的 N 个 tool call 跑完。难点全在"并发"和"顺序"的冲突上。

模块头注释把契约写全了：

Exclusive calls form barriers; parallel calls use a bounded rolling pool and are reclassified before start. Dispatch may overlap, while policy, results, and result context remain model-ordered.

分组是动态的。 executeToolCalls（tool-calls.ts:62）不预先切好组：

const mode = ctx.tools.executionMode(first.exec).kind
const group = mode === 'parallel' ? planned.slice(next) : [first]

每次只看当前第一个调用的模式；如果是 parallel，就乐观地把剩下全部当候选，进 runGroup 后再一个个复核（fillPool 里 if (nextToStart > 0 && mode === 'parallel' && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break）。为什么要复核？ 因为工具注册表是活的：前一个工具（比如 tool-cordis，第 12 章）可能在执行过程中挂载了新插件、改变了某个工具的并发模式。注释写得直接：Commit before classifying again so registry changes affect unstarted calls。

滚动池，不是批。

while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {

inFlight.size < maxParallelToolCalls（默认 10），一个完成就补一个，不是"10 个一批、等齐再下一批"。

提交严格按模型顺序。 commitReady()（tool-calls.ts:146）：

while (committed < group.length) {
  const slot = slots[committed]
  if (slot === undefined) break
  // …finalize / finish → appendToolResult → acceptContext → concluded ||= …
  committed++
}

committed 只沿连续的槽位前进。第 3 个先跑完、第 1 个还没完？第 3 个的结果就在槽里等着，不落日志。因为模型看到的 tool result 顺序必须和它自己发出 tool call 的顺序一致——否则同一次交互重放两遍会得到不同的历史。派发可以乱序，提交必须有序。

取消要留下完整的历史。

if (aborted) {
  for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
  return { consumed: group.length, aborted: true, concluded }
}

appendSkippedToolCall（tool-calls.ts:249）给每个没来得及启动的调用补一对 tool/call + tool/result，结果是合成的错误：

content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
isError: true,
error: { message: 'tool call aborted before dispatch', info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },

这不是礼貌，这是协议要求：绝大多数 provider 都规定每个 tool call 必须有配对的 tool result，缺一个则整段历史不合法，下次请求直接 400。所以取消路径必须把账做平。

对比一下调度器自身失败的路径（prepare 抛了、内部断言炸了）：注释明确 A terminal scheduler failure preserves already-recorded tool/call events without fabricating results。不编造结果——因为那种情况下 dsh 不知道工具到底跑没跑，编一个"失败"可能是撒谎。宁可留一个不完整的日志让人看见，也不写一个假的事实。这个区分（可确定的取消 vs 不可确定的崩溃）非常值得学。

concludesTurn。

concluded ||= result.concludesTurn === true

一个工具可以在结果里宣布"这轮该结束了"（典型：exit_plan_mode、把控制权交还用户的工具）。传回 step() 后 return concluded ? { kind: 'completed' } : null。但它不短路：runtime-types.ts:268 写明 The conclusion never short-circuits already-submitted next-step work —— 同 step 的 additionalContexts 和竞态的 steering 照样会跑，turn 只在 inbox 真的干了才关。又是"数据决定"。

9. foldConsumedWork()：为什么 turn/end 不够

consumed-work.ts 只有 108 行，但它的模块注释是整个 agent 包里最值得读的一段设计说明：

The turn and step vocabulary alone cannot answer this. A turn that stops before its first step leaves a turn/end shaped exactly like the balanced no-op turns a rejection or an empty claim produces, so reading turns in isolation either credits cut-short work as finished or convicts every no-op.

问题很具体："用户交给这个 agent 的活儿，到底干了没有？"

光看 turn 答不出来。第 4 节里有三种都会产生"一对 turn/start + turn/end、中间零个 step"的情况：

1. 唤醒消息被撤销了 → 什么活儿都没接，不算欠账；
2. agent/pre-step 拒绝了 → 接了活儿，但永远不会跑；
3. 取消发生在第一个 step 之前 → 接了活儿，被打断。

只看 turn，这三个长得一模一样。缺的信息在 inbox 的日志里：removedCount 和 outcome: 'canceled'。所以 foldConsumedWork()（consumed-work.ts:68）一遍扫过日志，同时维护三个集合：

case 'agent/inbox/spliced': {
  const { removedCount, outcome, inserted } = event.data
  if (removedCount === undefined) break
  if (outcome === 'canceled') droppedUnrun ||= inserted.length === 0
  else if (open !== undefined) claimed.add(open)
  break
}

• 带 canceled 且没有替换插入 → 有活儿被丢了没跑（replace 会同时插入新消息，那是换身份不是丢活儿，所以要判 inserted.length === 0）；
• 不带 outcome 的删除 → 那是 claim，给当前打开的 turn 记一笔"接过活儿"。

然后 turn/end 时判定：

if (stepped.delete(turn) || (claimed.delete(turn) && accountsForClaim(reason))) {
  end = event
  droppedUnrun = false
}

跑过 step 的 turn 无条件算数；只 claim 过没跑 step 的 turn，要看它的结束原因是否"对这批输入负了责"（accountsForClaim，consumed-work.ts:42）：

case 'completed': return false      // 领了个空，本来就没活儿
case 'blocked':
case 'aborted':
case 'interrupted':
case 'error': return true           // 接了活儿，明确交代了
default: return true                // 未知结局也算负责，绝不当成功

那个 default: return true 上面有 4 行注释解释为什么：TurnEndReasonMap 是可合并扩展的，后端可以加新的结束原因，而一个说不清的结局绝不能读成成功。这是"未知情况偏向保守"的正确方向选择。

这个函数的调用者是 subagent 委派（第 11 章）和 SDK：父 agent 要知道子 agent 到底把任务干完了还是被掐了。而它的签名只有 (events: readonly SessionEvent[]) —— 纯函数，输入只有日志。模块注释点出了这为什么重要：

every input is the log itself: no caller has to sample live state before cancelling, so a cancellation issued by anyone — the owner's teardown, an ancestor's interrupt, an unloading plugin — reads the same.

取消的发起者是谁都不影响读数。 如果这个判断依赖内存里的运行时状态（"我取消前先记一下当时跑到哪了"），那么每一个可能发起取消的地方都得记一次，还得记得一样。让它成为日志的纯函数，这类 bug 就不存在了。

10. 生命周期：一个 agent 的诞生与销毁

AgentLoop（agent-loop/src/index.ts:296）是服务壳，ReactLoopAgent 是驱动器，两者的关系是工厂与产品：

export class AgentLoop extends Service implements AgentFactory {
  static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']

五个依赖，全是接口服务，一个具体实现都没有。这就是第 1 章说的"抽象脊椎"。

构造函数里做了三件 Cordis 式的事（index.ts:349-353）：

ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
ctx.effect(() => ctx.agents.setFactory(this), 'agentLoop.setFactory()')
ctx.systemPrompt.variable('provider', context => context.agent?.options.provider)

注册自己为工厂是effect，所以 HMR 卸载 agent-loop 时工厂槽自动腾空，下一个版本能接上（第 2 章）。提示词变量也一样。

10.1 prepare()：先装好拆的路，再开始装

prepare()（index.ts:459）是 120 行的一个方法，它的注释直接给了纲领：

The teardown is registered with the factory and the owner fiber BEFORE publication, so a mid-setup unload rolls everything back.

顺序是反直觉但正确的：先注册 teardown，再创建资源。

let machine: ReactLoopAgent | undefined
let detachSession: (() => void) | undefined
let detachAgent: (() => void) | undefined
const dispose = (ownerTriggered = false): Promise<void> => (disposing ??= (async () => { … })())
const untrack = this.ownership.track(dispose)
unfollowOwner = ownerCtx.effect(() => () => { … return dispose(true) }, `agentLoop.lifecycle(${id})`)

dispose 闭包捕获的是可变槽位（machine、detachSession、detachAgent 都还是 undefined），注释解释得很清楚：It is registered BEFORE any resource exists, over mutable slots, so an unload arriving while the scope is still minting finds a working disposer instead of a leak. 卸载信号在"scope 还在铸造中"的瞬间到达时，能找到一个已经能工作的 disposer，而不是一个漏掉的资源。

disposing ??= 是记忆化：多个拥有者（调用方的 signal、owner fiber 的 unload、工厂 teardown）可能同时喊拆，但只能拆一次，所有人 await 同一个静止点。

拆的顺序是装的逆序：

if (machine === undefined) await machineReady.promise
machine.cancel({ kind: 'disposed' })
await machine.whenIdle()
await machine.scope.dispose()
// finally:
detachAgent?.()
detachSession?.()

停驱动器 → 等静止 → 拆 scope → 退出 registry → 退出 session registry。注意 machine.cancel({ kind: 'disposed' }) —— 回到 3.1 节，正是这个 disposed 原因让 wake latch 不生效，whenIdle() 才能真的等到静止。两个相隔几百行的代码点，靠一个 discriminant 值咬合在一起。

10.2 publish()：发布是一个事务

publish: (source) => {
  assertLive()
  detachSession = agent.ctx.sessions.enter(session)
  detachAgent = loopCtx.agents.enter(agent, ownerCtx.agent)
  agent.ctx.sessions.announce(session)
  assertLive()
  loopCtx.agents.announce(agent)
  assertLive()
  emitAgentEvent(loopCtx, agent, 'agent/session-start', { source })
  assertLive()
  return { agent, dispose }
}

enter 和 announce 是分开的两个动作（agent/src/index.ts:474 和 :549）。enter 插入 registry 但不广播，announce 才广播 agent/created。为什么要拆？因为异步 setup 需要在"已经占了位、但外界还看不见"的状态下跑完——观察者永远不该看到一个半配置好的 agent。

而 enter() 里那个碰撞检查（index.ts:482）是整个系统的权威仲裁点：

if (this.store.has(id)) throw new Error(`agent "${id}" is already registered`)

注释：Concurrent create/resume operations may both prepare, but only one exact entry can publish. 两个并发的 resume 同一个 id，都可以走完准备和 setup，但只有一个能 enter 成功，另一个抛错并回滚自己的私有资源。这比"先抢一个锁再准备"简单得多，代价是失败者白干一次——对低频的 resume 操作来说是对的取舍。

enter() 返回的 detach 闭包还处理一个精妙的时序（index.ts:502）：

if (entry.announcing) {
  entry.detachRequested = true
  return
}

如果一个 agent/created 的同步监听器立刻决定销毁这个 agent（比如它检查出配置不合法），detach 不能立刻执行——否则同一次派发里后面的监听器会看到一个已经消失的 agent，而且 agent/disposed 会在 agent/created 派发完成之前发出。所以推迟到派发展开完（announce 的 finally 里 if (entry.detachRequested) this.detachEntered(entry)）。created/disposed 这对边永远配对，且永远按顺序。

配套的还有一条（index.ts:523）：

if (!entry.announced) return

enter 了但没 announce 就回滚的 agent，不发 agent/disposed —— 因为它对外界从来没存在过，发一个 disposed 就是在日志里编造一条不可能的生命周期边。

10.3 withInitiator：AsyncLocalStorage 把 agent 沿异步链传下去

AgentRegistry 里有个东西第一眼很意外（agent/src/index.ts:259）：

private readonly initiators = new AsyncLocalStorage<Agent | undefined>()

驱动器启动时（agent.ts:192）：

this.loopCtx.agents.withInitiator(this, () => this.kick())

于是从 kick() 往下的整条异步调用链里，任何代码都能 ctx.agents.requireInitiator() 拿到"是哪个 agent 引起的我"。executeToolCalls 第一行就是这么拿 session 的（tool-calls.ts:69）。

为什么不直接传参？packages/AGENTS.md 里有专门一条规则回答了这个：

Initiator-owned private chains derive, then capture. Under ctx.agents.withInitiator(), recover the Agent at each orchestration entry, derive agent.session, and let operation-local helpers close over it. Keep Agent and Session explicit at lifecycle, session-log, service, authority, worker/process, persistence, and wire interfaces.

界限很清楚：编排入口用 ALS 恢复，然后局部闭包捕获；跨越生命周期、日志、服务、授权、进程、持久化、网线这些界面时必须显式传参。 ALS 是给"深层私有调用链"省去参数管道的，不是给公共接口省事的——一个从网线上收到消息的接收器绝不能靠 ALS 猜自己在为谁服务（它必须先验证身份再建立边界，withInitiator 的 JSDoc 明说了 this method does neither）。

还有一个反向操作 withoutInitiator()（index.ts:356）：清除继承的 initiator。用在懒初始化的共享资源上——共享定时器、连接池、导出器不该"归第一个碰巧初始化它的 agent 所有"。这是个真实且容易踩的坑：共享资源继承了某个 agent 的身份，那个 agent 一销毁，共享资源的归因就全错了。

销毁时的 disposeInitiators()（index.ts:625）先 closing（拒绝新边界）、再排空活跃边界、最后 initiators.disable()。三段式收敛，避免"服务已经拆了但还有继承的异步链在跑"。

11. 这套设计的代价

① 读懂 turn() 需要同时装着五六个不变量。 85 行里有 7 处 throwIfAborted、两个语义不同的空批次分支、一个粘性 max-tokens、一个双重检查的 turn-stopping。每一处都有理由，但组合起来的认知负担很高。dsh 用大量注释和 Agent Note 链接来缓解，但这段代码确实不适合"顺手改一下"。

② "唯一的具体循环"意味着扩展点的设计压力全在一个地方。 想加一个"在模型响应之后、工具执行之前"的钩子？现在没有，tools/pre-execute 是每个工具一次而不是每组一次。加一个就要改 agent-loop、改 docs/architecture.md、改两个 SDK 的期望输出。扩展点的收益是真的，成本也是真的。

③ Inbox 是投影 → 每次输入都写日志。 用户敲一条消息、插件注入一段上下文、取消清空队列，全都是持久化写。对本地 SQLite 无所谓，但换成网络持久化后端，输入延迟就直接暴露给用户了。

④ ALS 的边界是靠规则和评审守的，不是靠类型。 requireInitiator() 在任何地方都能编译通过。哪里该用、哪里必须显式传参，全靠 packages/AGENTS.md 那一条规则和人的判断。这是 dsh 少数几个"约定大于机制"的地方。

⑤ 一次模型调用一个 step，粒度是固定的。 想做"一次请求内多轮 speculative 工具执行"这类实验性调度，现有的 turn/step 编号体系装不下——step 号是模型可见历史的坐标，不能随便加。

12. 可迁移的经验

① 把"决定"变成数据，而不是回调的返回值。 agent/turn-stopping 是本章最值得抄的一招：想续命的监听器不返回 false，它往队列里塞一条真实消息，然后循环重读队列。监听器顺序立刻变得无关，"多个监听器意见冲突"这个问题从存在变成不存在。任何"多个观察者都想影响一个决定"的场合都适用。

② 队列不是状态，是日志的投影。 Inbox 的所有性质——重启后还在、UI 能重放、取消可审计、观察者能重建被删的内容——都是"它是投影"这一个决定的推论，没有一条是额外实现的功能。

③ 记录变化，不记录快照。 request/header 的 initial/resume/change 三态：跑 40 步、配置没变就只有一条记录，而任意时刻的配置总能靠"往前找最近一条"复原。这比每步存一份完整配置省得多，也比"只在内存里维护"可靠得多。

④ 用一个纯函数回答"到底干了没有"。 foldConsumedWork(events) 的输入只有日志，所以无论取消是谁发起的，读数都一致。凡是"判断某件事的状态"要先采样运行时状态的设计，都会在多个发起者出现时不一致。

⑤ 先注册 teardown，再创建资源。 prepare() 让 disposer 闭包捕获可变槽位，于是"资源正在创建中就被要求销毁"这个最难的时序有了正确答案。同时用 disposing ??= 记忆化，让所有并发的拆除者 await 同一个静止点。

⑥ 区分"可确定的失败"和"不可确定的失败"。 取消时给未启动的工具补一个合成的错误结果（协议要求账要平），调度器自己崩溃时不编造结果（因为不知道工具跑没跑）。前者是补账，后者是撒谎，中间那条线要划清楚。

⑦ append-only 的世界里，"清除"是一条新的事实。 RuntimeContextProjection 的 CLEARED 字符串：你不能删掉模型已经看过的旧快照，只能补一条"之前那个不算了"。任何 append-only / 事件溯源系统都会遇到这个形状的问题。

附：本章源码引用表