# 流处理器与工具调用生命周期

> 处理器如何把 LLMEvent 流翻译成 Part：工具调用的生命周期、步末结算与溢出判定、重试、死循环检测与中断收尾。

- 作者：David（道雾轩）
- 专栏：OpenCode 源码解读（https://daiw.net/manual/opencode.md）
- 最后更新：2026-09-15
- 原文：https://daiw.net/manual/opencode/processor
- 转载与引用：请注明出处并附原文链接（https://daiw.net/about/copyright）

[主循环](https://daiw.net/manual/opencode/prompt-loop)每一步都新建一条 assistant 消息，并为它创建一个处理器句柄（`packages/opencode/src/session/processor.ts`）。句柄的核心方法是 `process(streamInput)`：把 [LLM 层](https://daiw.net/manual/opencode/llm-bridge)吐出的事件流逐个翻译成 Part 写进这条消息，最后告诉循环三选一的结果——`compact`、`stop` 或 `continue`。

## 先拍快照

`create` 做的第一件事是拍文件快照。注释解释了为什么不能等 `step-start` 事件再拍（`processor.ts:99`）：

> Pre-capture snapshot before the LLM stream starts. The AI SDK may execute tools internally before emitting start-step events, so capturing inside the event handler can be too late.

AI SDK 可能在发出 start-step 之前就在内部执行了工具，等事件到了再拍，文件也许已经被改过。快照的用途见[快照与撤销](https://daiw.net/manual/opencode/snapshot-revert)。

## process 的骨架

`processor.ts:650`，删减后：

```ts
          yield* Effect.gen(function* () {
            // ...
            yield* status.set(ctx.sessionID, { type: "busy" })
            const stream = llm.stream(streamInput)

            yield* stream.pipe(
              Stream.tap((event) => handleEvent(event)),
              Stream.takeUntil(() => ctx.needsCompaction),
              Stream.runDrain,
            )
          }).pipe(
            Effect.onInterrupt(() =>
              // ...
            ),
            // ...
            Effect.retry(
              SessionRetry.policy({
                // ...
              }),
            ),
            Effect.catch(halt),
            Effect.ensuring(cleanup()),
          )

          if (ctx.needsCompaction) return "compact"
          if (ctx.blocked || ctx.assistantMessage.error) return "stop"
          return "continue"
```

几层包装各司其职：`takeUntil` 让某个 step-finish 一旦判定上下文超限就立刻停止消费；用户中断时标记 `aborted` 并按 AbortError 走 `halt`；其他错误交给重试策略，**重试会重跑整个流**；重试耗尽或不可重试则由 `halt` 记录错误；`cleanup` 无论成败都会执行。

## 事件到 Part 的映射

`handleEvent`（`processor.ts:278`）是一个大 switch：

| LLMEvent | 处理 |
| --- | --- |
| `text-start`、`text-delta`、`text-end` | 建 text Part；增量走不落库的 `updatePartDelta`；结束时经插件钩子改写后整段写入 |
| `reasoning-start`、`reasoning-delta`、`reasoning-end` | 同上，生成 reasoning Part，按 ID 可多段并存 |
| `tool-input-start`、`tool-input-delta`、`tool-input-end` | `ensureToolCall`：首次出现时建 pending 的 tool Part |
| `tool-call` | 参数到齐，改为 running，并做死循环检测 |
| `tool-result`、`tool-error` | 改为 completed 或 error |
| `step-start` | 写 step-start Part，带快照哈希 |
| `step-finish` | 结算用量，写 step-finish 与 patch Part，判断溢出 |
| `provider-error` | 直接抛错，交给重试或 `halt` |

另有两道防线：正在生成压缩摘要时，任何工具调用都直接抛错；没有对应 start 的 reasoning 增量则被静默丢弃。

## 工具调用的一生

处理器并不执行工具。执行发生在 AI SDK（或原生链路）调用 `session/tools.ts` 包装的 `execute` 时，处理器只记账：每个调用在 `ctx.toolcalls` 里登记一个 `Deferred`。

```mermaid
sequenceDiagram
  participant M as 模型事件流
  participant P as 处理器
  participant T as 工具 execute
  M->>P: tool-input-start
  P->>P: 建 pending Part，登记 Deferred
  M->>P: tool-call 完整参数
  P->>P: 改为 running，死循环检测
  T->>P: metadata 回调刷新进度
  M->>P: tool-result 或 tool-error
  P->>P: 改为 completed 或 error，Deferred 完成
```

失败原因若是权限或提问被用户拒绝，处理器置 `ctx.blocked = ctx.shouldBreak`，本步结束后返回 `stop`；配置了 `experimental.continue_loop_on_deny` 时则不停，模型会带着“被拒绝”的结果继续。用户中断时流已断开、等不来 `tool-result`，所以 `tools.ts` 在工具执行完后检查 `abortSignal`，自己调用 `completeToolCall` 补记结果。

### 死循环检测

`processor.ts:353`，删减后（阈值 `DOOM_LOOP_THRESHOLD = 3`）：

```ts
            const recentParts = parts.slice(-DOOM_LOOP_THRESHOLD)

            if (
              recentParts.length !== DOOM_LOOP_THRESHOLD ||
              !recentParts.every(
                (part) =>
                  part.type === "tool" &&
                  part.tool === value.name &&
                  part.state.status !== "pending" &&
                  JSON.stringify(part.state.input) === JSON.stringify(input),
              )
            ) {
              return
            }

            const agent = yield* agents.get(ctx.assistantMessage.agent)
            yield* permission.ask({
              permission: "doom_loop",
              // ...
            })
```

当前消息最后 3 个 Part 都是同名工具、参数 JSON 完全一致且都已离开 pending，就发起一次 `doom_loop` 权限询问。内置 agent 的默认规则里 `doom_loop` 为 `ask`（`packages/opencode/src/agent/agent.ts:121`），即把“要不要继续”交给用户。注意检测范围只是当前这条 assistant 消息，而主循环每一步都会新建消息。

## 步末结算

`step-finish` 分支（`processor.ts:435`）依次：再拍一次快照；收尾所有 reasoning；若 Anthropic 报告丢弃了思考块就记警告（注释说这意味着历史在签名块背后被改动过）；用 `Session.getUsage` 算用量与费用；写 step-finish Part；与步首快照比较，有改动就写 patch Part；fork 出差异统计；最后判断溢出。

`getUsage`（`session.ts:338`）的细节：AI SDK v6 起 `inputTokens` 已含缓存命中，代码显式减去缓存读写；`output` 扣掉推理 token，推理按输出单价计费（标了 TODO）；单价支持按上下文长度分档；全程用 `decimal.js` 计算。

`cost` 是累加，`tokens` 却是覆盖——消息上保留的是最后一步的用量，溢出判断用的正是它。

## 溢出判定

`packages/opencode/src/session/overflow.ts:10`：

```ts
export function usable(input: { cfg: ConfigV1.Info; model: Provider.Model; outputTokenMax?: number }) {
  const context = input.model.limit.context
  if (context === 0) return 0

  const reserved =
    input.cfg.compaction?.reserved ??
    Math.min(COMPACTION_BUFFER, ProviderTransform.maxOutputTokens(input.model, input.outputTokenMax))
  return input.model.limit.input
    ? Math.max(0, input.model.limit.input - reserved)
    : Math.max(0, context - ProviderTransform.maxOutputTokens(input.model, input.outputTokenMax))
}
```

模型声明了独立的输入上限时，可用量 = 输入上限 − 预留（默认取 2 万与最大输出的较小者）；否则 = 上下文窗口 − 最大输出（模型输出上限与 32000 取小）。`isOverflow` 在关闭自动压缩或窗口未知时恒为假，否则比较本步 token 总量是否达到可用量。判定为真就置 `needsCompaction`，`process` 返回 `compact`，后续见[上下文压缩](https://daiw.net/manual/opencode/compaction)。

## 重试

`retry.ts` 最多重试 5 次；间隔从 2 秒起每次翻倍，加最多 25% 的随机抖动，没有响应头时封顶 30 秒；响应头里有 `retry-after-ms` 或 `retry-after`（秒数或 HTTP 日期）时以它为准。

`ContextOverflowError` 永不重试，因为它该走压缩。`APIError` 在 SDK 标记可重试、状态码为 5xx、或消息与响应体命中一组正则（429、rate limit、overloaded、各类网络错误与超时）时重试。每次重试前，会话状态被设为 `retry`，带上尝试次数、原因和下次重试时间。

## halt 与 cleanup

`halt` 用 `MessageV2.fromError` 把异常归类成消息错误。`ContextOverflowError` 是特例：关闭了自动压缩就记为错误并置 `idle`，否则置 `needsCompaction`，让本步返回 `compact`。

`cleanup` 补写剩余的 patch Part，给未结束的 text 与 reasoning 补上结束时间，给未完成的工具调用最多 250 毫秒收尾；仍未完成的一律改成 `error`，错误文本为“Tool execution aborted”，并打上 `metadata.interrupted = true`。这个标记在两处被读取：主循环判断结束时忽略这类“孤儿”调用；`toModelMessages` 把仍处于 pending 或 running 的调用补成错误结果，注释写明 Anthropic 要求每个 tool_use 都有对应的 tool_result。

下一篇：[系统提示词与指令文件](https://daiw.net/manual/opencode/system-prompt)——每一步发给模型的 system 由哪些部分拼成。