第 3 章：Agent 接口与 agent-loop——turn/step 状态机

从「Agent 接口」到「默认驱动」

第 1 章说过：dsh 的 Agent 循环本身是插件。这句话在包结构上立刻可见——packages/core/agent/ 定义 接口与注册表（ctx.agents），packages/core/agent-loop/ 提供默认实现（ReactLoopAgent）。agent 包不依赖 agent-loop；所有扩展插件只依赖 agent。想换循环实现？实现同一个 Agent 接口，通过 AgentFactory 注入注册表即可——其他一切子系统对此无感。

接口本身（packages/core/agent/src/runtime-types.ts）小而锋利：

export interface Agent {
  /** The single identity shared with {@link session}. */
  readonly id: SessionId
  /** The live session this agent drives; its log is the durable source of truth. */
  readonly session: Session
  /** The agent-owned projection of durable pending work. */
  readonly inbox: Inbox
  /** The current lifecycle state, mirrored on every `agent/status` transition. */
  readonly status: AgentStatus
  /** Agent-scoped context; its contributions are agent-local, unwind on disposal, and reject registration afterward. */
  readonly ctx: Context

  cancel(cause: AgentCancelCause, options?: CancelOptions): void
  whenIdle(): Promise<void>
  runMaintenance<T>(task: (signal: AbortSignal) => Promise<T>): Promise<T>
  send(message: UserMessage, target: InboxTarget, wakeup: boolean): void
  followup(message: UserMessage): void
  steer(message: UserMessage): void
  inject(message: UserMessage): void
}

几个值得注意的形态决定：

• id 是 SessionId——agent 与 session 共享同一个身份。没有「会话的 id」与「agent 的 id」两套编号。
• status 是只读属性而非方法，取值只有 'idle' | 'running'；disposal 不是第三种状态，而是一个取消原因（{kind: 'disposed'}）。
• cancel 携带一个有类型的取消原因（AgentCancelCause：user / parent / hook / disposed），原因随 AbortSignal.reason 传播，首个原因胜出。
• followup/steer/inject 是 send 的三个固定预设别名——下一节看它们的区别。

Inbox：把收件箱做成日志投影

Agent 不直接持有「待办消息队列」——队列是会话日志的投影。每条入队/出队都先落一条 agent/inbox/spliced 事件，再改内存投影；进程重启后从日志重放就能恢复队列。这就是「模型可见 ⟺ 已记录」不变式在队列上的体现。

Inbox 里有两条待办列表，由 InboxTarget 区分：

claim(target, turn) 的实现（inbox.ts）：

  /**
   * Remove and return the complete batch proposed for one step, publishing
   * each claimed message. The durable splices are pure deletions.
   */
  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
  }

一次 claim = 清空 next-step 全表 +（仅当目标是 next-turn 时）取 1 条 next-turn。所以「模型跑完一步、工具结果塞回 next-step、循环立刻认领进入下一步」是一条连续带；而「新用户输入进入 next-turn、只有 turn 边界才消费」是另一条节奏。底层的 mutate() 体现了持久化优先：

    this.validate(splice)
    const event = this.session.append('agent/inbox/spliced', splice)
    const removed = inbox.splice(actualStart, actualDeleteCount, ...event.data.inserted)
    if (discardRemoved) {
      for (const message of removed) this.notifications.discarded(message)
    }
    for (const message of event.data.inserted) this.notifications.inserted(message)
    return removed

先 session.append 落日志，再改内存投影，最后发通知——顺序不可颠倒。至于 send 的 wakeup 参数：target 决定消息进哪条列表，wakeup 决定是否唤醒 driver。有一个精妙的边角：如果唤醒消息到达时当前活动已经 abort，它会被改投到 next-turn（agent.ts 的 send：wakingAfterAbort 判定），等取消收敛后再开新一轮跑——「正在被取消的工作不能收新活」。

agentEvents：三语义分发器

Agent 事件由 agentEvents()（dispatch.ts）生成一个「融合分发器」：把 agent 对象注入每个事件的 payload（spread 在前，所以监听器永远无法覆盖注入的 subject），并把 scopeTarget(agent, agent) 作为事件路由载体。三种语义：

    emit(name, payload) {
      // Cordis emit invokes callbacks through Array.map: one synchronous throw
      // starves later listeners, and returned promises are discarded. Agent
      // notifications are non-vetoing, so resolve the same filtered callback
      // set ourselves and contain both failure modes independently.
      const args: unknown[] = [carrier, name, fused(payload)]
      const callbacks = ctx.events.dispatch('emit', args)
      for (const callback of callbacks) {
        try {
          const returned: unknown = callback(...args)
          void Promise.resolve(returned).catch((error: unknown) => {
            ctx.logger.warn(`agent event "${name}" listener rejected: ${String(error)}`)
          })
        } catch (error: unknown) {
          ctx.logger.warn(`agent event "${name}" listener threw: ${String(error)}`)
        }
      }
    },

注意这段注释——Cordis 原生的 emit 用 Array.map 调用回调，一个同步 throw 会饿死后面的监听者；agent 通知是不可否决的，所以 dsh 自己遍历回调、把两种失败模式独立包住。这是「框架默认行为不够好时，在插件里修」的微观例子。

分发器之外，ctx.agents 还提供 withInitiator：用双层 AsyncLocalStorage（initiators 存 Agent、initiatorRuns 存运行链）建立进程内因果链。requireInitiator() 在无链时抛错——工具执行时 exec.agent 就来自这里（第 5 章）。

turn/step 状态机：ReactLoopAgent 的心脏

ReactLoopAgent（packages/core/agent-loop/src/agent.ts）用三个 phase 表达生命周期：idle（无事可做）、maintenance（后台维护任务占用）、running（driver 在跑，携带 turn/step 计数与自己的 AbortController）。驱动入口是 kick()——while (await this.turn()) {} 循环，只要 turn 返回 true（还有活）就继续。

turn 循环骨架（agent.ts，注释保留原文）：

    const turn = phase.turn + 1
    try {
      this.session.append('turn/start', { turn })
    } catch (error: unknown) {
      this.throwError(error)
    }
    phase.turn = turn
    let turnEnds: TurnEndReason | null = null
    let target: InboxTarget = 'next-turn'
    try {
      while (true) {
        signal.throwIfAborted()
        const step = phase.step + 1
        const decision = await this.preStep(target, { turn, step })
        if (decision.kind === 'reject') {
          turnEnds = { kind: 'blocked' }
          return false
        }
        // 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
        }
        signal.throwIfAborted()
        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' })
          }
          // max-tokens is sticky: once any step hits the ceiling, later steps
          // that complete normally must not downgrade the turn outcome.
          const stepEnd = await this.step(decision.assembly)
          // max-tokens stays sticky: a later completed step must not
          // downgrade the turn outcome.
          if (turnEnds === null || turnEnds.kind !== 'max-tokens') turnEnds = stepEnd
        } finally {
          this.session.append('step/end', { turn, step })
        }
        // ...
      }
    } finally {
      try {
        // oxlint-disable-next-line typescript/no-non-null-assertion -- every exit assigns a turn ending
        this.session.append('turn/end', { turn, reason: turnEnds! })
      } catch (error: unknown) {
        this.throwError(error)
      }
    }

值得展开的细节：

• pre-step 是唯一的输入闸门：preStep() 先 inbox.claim，再 systemPrompt.assemble 装配提示词，再经 agent/pre-step 瀑布。瀑布的默认 next() 返回 {kind: 'enter', messages: claimed}——监听器可以 reject（turn 以 blocked 关闭）、可以改写消息、可以追加（运行时上下文投影就在这注入，第 6 章）。
• 空消息的 turn 仍然成立：被移除的唤醒消息、或被重写成空的首个 enter，都「拥有」turn 边界但零模型调用——日志里能看出「这个 turn 尝试过但没有内容」。
• max-tokens sticky：一旦某一步撞上输出上限，后续正常完成的 step 不能把 turn 结局降级为 completed。三行代码（turnEnds === null || turnEnds.kind !== 'max-tokens'）守住一个微妙的语义。
• turn-stopping 是续命钩子：turn 要关但 next-step 队列已空时，先发 agent/turn-stopping（serial）。监听器可以在这里 agent.steer(...) 塞新输入——机器随后重读 inbox，队列非空就继续下一 step。这是「监听器能救活一个即将结束的 turn」的机制，第 13 章的目标续跑、第 16 章的 hooks 都会用到。
• 所有失败都有结构化出口：abort → {kind: 'aborted', reason}；其他错误 → LlmError 保留事实，否则扁平化成 {message, code: 'UNKNOWN'}。turn/end 一定带 TurnEndReason（completed/aborted/blocked/error/max-tokens/interrupted）。

step() 内部（下一章会看到对应的日志事件）：buildRequest 组装冻结的请求（经 agent/request 瀑布）、llm.stream 拉 chunk（每条 chunk 先落 assistant/chunk 日志再喂给 BlockAssembler）、失败走 agent/request-error 瀑布（监听器返回 {kind:'retry'} 就重试）、消息落 assistant/message、工具调用交给 executeToolCalls（第 5 章），工具结果作为上下文经 acceptor 塞回 next-step inbox。

scope：按 agent 隔离的注册边界

「每个 agent 有自己的注册视图」由 packages/core/scope/ 实现。核心是 createScope(ctx, key)：用 ctx.plugin(无操作插件) 铸造一条新 fiber，fiber.ctx.extend({[kScope]: key}) 得到 scoped context——通过它注册的一切都是该 fiber 的 effect，dispose 时整体回滚。

export function createScope(ctx: Context, key: ScopeKey, options?: CreateScopeOptions): Scope {
  if (options?.parent !== undefined) bindScopeParent(key, options.parent)
  const fiber = ctx.plugin(scope)
  const scoped: Context = fiber.ctx.extend({ [kScope]: key })
  let disposing: Promise<void> | undefined
  return {
    ctx: scoped,
    rawDispose: fiber.dispose,
    dispose: () => (disposing ??= quiesceFiber(fiber)),
  }
}

路由语义在 scopeTarget：无标签的 ctx 全局放行，带标签的只放行 key 自身或祖先链上的标签——事件只向上流。于是「一个常驻组合（如 preset 的 standing mount，第 16 章）可以观察它下面的所有子 agent，子 agent 看不到父之上的兄弟」。注册视图则相反地向下继承：ScopedLayers 用「全局层 + 按 key 覆盖层」存储，chainLayers(scope) 祖先在前、最近者在后，同名注册最近者胜——工具注册表（第 5 章）与 system-prompt（第 6 章）都建在它上面。

AgentLoop：工厂与发布事务

packages/core/agent-loop/src/index.ts 的 AgentLoop 服务实现 AgentFactory，负责把「一个 session」变成「一个活着的 agent」：setup → enter → announce（发 agent/created）→ session-start → 启动机器的发布事务。加上 runMaintenance（idle 期才能认领的后台任务槽，占用期间 status 仍是 idle，新 wake 在 inbox 排队）与 whenIdle（跟随 activityDone 直到收敛），一个 agent 的完整生命周期契约就闭合了。

实践应用

1. 收件箱即日志投影：队列不放在内存，每条变更先落事件再改投影——重启可恢复、可审计、可与「模型可见 ⟺ 已记录」共享同一条不变式。
2. 瀑布的默认值是最内层 next()：让「框架默认行为」与「扩展行为」同构，拦截点就能否决、改写、短路任意输入——pre-step 的 reject/enter 是完整的 around-middleware。
3. 取消是带类型的协议：AgentCancelCause + AbortSignal.reason + wake 闩锁 + keepInbox，把「取消」从布尔标志升格为可区分意图、可收敛、可恢复的领域概念。
4. 隔离靠作用域链而非克隆：per-agent 注册边界用 scope key + fiber 实现，事件向上流、注册向下继承——「最近者胜」是标准的覆盖语义，换组合是 O(1) 的父链重绑。

总结

这一章拆完了 Agent 接口、Inbox、分发器与 turn/step 状态机：循环的每个动作都落在日志事件与瀑布扩展点上，队列是日志投影，取消是类型化协议，隔离是作用域链。但还有一半没讲——循环写进日志的那些事件（turn/*、user/message、assistant/*、tool/*）本身是什么？下一章进入会话日志：为什么 dsh 敢说「模型看到的一切都能从日志重建」，以及 append-only 日志如何在类型层面保持可扩展。