Skip to main content

查询引擎：54,000 行里最不像代码的那一块

如果你只盯工具系统的 grep / Glob / FileEdit，看到的是「拼装好的乐高」。但乐高谁来驱动、怎么一轮轮地拼、拼到一半发现问题怎么回退——这些全靠查询引擎。源码里这一块接近 54,000 行，几乎全是状态机和恢复逻辑，是整个项目里最不像「应用代码」、最像「分布式系统核心」的一部分。

QueryEngine：从便利函数到生命周期管理​

本节源码：src/QueryEngine.ts

源码早期，查询逻辑塞在一个 ask() 便利函数里——填参数、调 API、yield 消息，调用方用着方便，但测试和复用都很难。源码里专门有一段注释解释这次重构的动机：

QueryEngine owns the query lifecycle and session state for a conversation. It extracts the core logic from ask() into a standalone class that can be used by both the headless/SDK path and (in a future phase) the REPL. One QueryEngine per conversation. Each submitMessage() call starts a new turn within the same conversation. State (messages, file cache, usage, etc.) persists across turns.

核心抽象：一次对话 = 一个 QueryEngine 实例。submitMessage() 不重新跑 setup，而是接着上次的 mutableMessages / readFileState / usage stats 继续跑。这意味着：

  • 对话历史是引擎的私有状态，不是每次重新塞进来的参数
  • 跨轮的 token 累加、permission denial 累积、discovered skill 集合都是引擎保管的
  • 同样的对话可以用 REPL、SDK、cowork 三种入口驱动，只要走同一个 QueryEngine 实例，行为完全一致
export class QueryEngine {
  private config: QueryEngineConfig
  private mutableMessages: Message[]
  private abortController: AbortController
  private permissionDenials: SDKPermissionDenial[]
  private totalUsage: NonNullableUsage
  private readFileState: FileStateCache
  private discoveredSkillNames = new Set<string>()
  private loadedNestedMemoryPaths = new Set<string>()

  async *submitMessage(
    prompt: string | ContentBlockParam[],
  ): AsyncGenerator<SDKMessage, void, unknown>
}

注意几个关键字段：

  • mutableMessages——对话历史，是引擎最关心的状态
  • totalUsage——所有 turn 的 token 累加，用于 budget 控制
  • permissionDenials——每一轮失败一次就 push 进去，最终生成 SDK 的 permission_denials 字段
  • discoveredSkillNames / loadedNestedMemoryPaths——turn-scoped 状态，但跨两轮之间用 clear() 重置（防止 SDK 模式下无限增长）

query() 函数：Async Generator 而非 Promise​

本节源码：src/query.ts

export async function* query(
  params: QueryParams,
): AsyncGenerator<
  | StreamEvent
  | RequestStartEvent
  | Message
  | TombstoneMessage
  | ToolUseSummaryMessage,
  Terminal
>

返回类型是 AsyncGenerator 不是 Promise，这不是品味问题，而是工程必须：

  • token 是流式返回的——LLM 的响应是一个字一个字吐出来的，前端 UI 需要边收边渲染，不能等全部到齐
  • 工具调用是迭代的——一次 query 可能产生 N 轮工具调用（agent loop），每一轮都要 yield 给调用方
  • 错误是可恢复的——max_output_tokens 触发的 64k 重试、prompt-too-long 触发的 collapse drain + reactive compact，都需要流式 yield 出中间状态，然后重新进入循环

调用方拿到这个 generator 后，可以用 for await 一边消费消息一边干别的。状态被 close 在 queryLoop 内部。

内部 queryLoop：一个带类型化 state transition 的状态机​

本节源码：src/query.ts

query() 包装了 queryLoop()，后者才是真正干活的地方。它最值得学习的设计是显式的 State 对象 + transition 字段：

type State = {
  messages: Message[]
  toolUseContext: ToolUseContext
  autoCompactTracking: AutoCompactTrackingState | undefined
  maxOutputTokensRecoveryCount: number
  hasAttemptedReactiveCompact: boolean
  maxOutputTokensOverride: number | undefined
  pendingToolUseSummary: Promise<ToolUseSummaryMessage | null> | undefined
  stopHookActive: boolean | undefined
  turnCount: number
  // Why the previous iteration continued. Undefined on first iteration.
  // Lets tests assert recovery paths fired without inspecting message contents.
  transition: Continue | undefined
}

while (true) {
  let { toolUseContext } = state
  const {
    messages,
    autoCompactTracking,
    maxOutputTokensRecoveryCount,
    ...
  } = state
  // ... loop body ...
  
  state = { ... }  // continue/recover/exit
  continue
}

源码注释解释这套设计：

Mutable cross-iteration state. The loop body destructures this at the top of each iteration so reads stay bare-name (messages, toolUseContext). Continue sites write state = { ... } instead of 9 separate assignments.

每个 continue 站点都构造一个完整的新 State 对象，而不是修改 9 个字段。这样：

  • 字段跨站点流动一目了然（看 state = { ... } 一行就知道变了什么）
  • 测试不用 mock 单个字段的修改，直接断言最终 State
  • 9 个字段的同步边界不再隐藏在局部 mutation 里

transition 字段记录**「上一轮为什么 continue」**——这让单测可以验证「max_output_tokens 重试是否真的发生过」，不用 grep yield 出来的消息内容。

前调用管线：snip → microcompact → contextCollapse → autocompact​

本节源码：src/services/compact/autoCompact.ts

每次进入 API 调用之前，会按固定顺序跑 4 道压缩：

snip         → 折叠早期历史
↓
microcompact  → 替换旧的工具结果
↓
contextCollapse → 分层折叠（如果开启了 CONTEXT_COLLAPSE feature）
↓
autocompact   → 真正逼近窗口上限时做大手术式压缩

源码注释解释了顺序背后的设计：

// Apply snip before microcompact (both may run — they are not mutually exclusive).
// snipTokensFreed is plumbed to autocompact so its threshold check reflects
// what snip removed; tokenCountWithEstimation alone can't see it (reads usage
// from the protected-tail assistant, which survives snip unchanged).
let snipTokensFreed = 0
if (feature('HISTORY_SNIP')) {
  const snipResult = snipModule!.snipCompactIfNeeded(messagesForQuery)
  messagesForQuery = snipResult.messages
  snipTokensFreed = snipResult.tokensFreed
  // ...
}

// Apply microcompact before autocompact
queryCheckpoint('query_microcompact_start')
const microcompactResult = await deps.microcompact(...)

顺序的逻辑：

  • snip 在 microcompact 之前——snip 折叠的是历史老消息，microcompact 处理的是工具结果老条目，两者不重叠但都释放 token。先 snip 把老消息直接砍掉，microcompact 就不用处理那些已经消失的消息
  • microcompact 在 autocompact 之前——microcompact 是「增量小手术」，autocompact 是「大手术」。先做小手术可能就让总 token 回到阈值以下，autocompact 就不用触发
  • contextCollapse 在 autocompact 之前——因为 contextCollapse 也算小手术，且 AUTOCOMPACT_THRESHOLD 的检查读的是 token 估算，先 collapse 可能让 autocompact 当轮直接跳过
// Project the collapsed context view and maybe commit more collapses.
// Runs BEFORE autocompact so that if collapse gets us under the
// autocompact threshold, autocompact is a no-op and we keep granular
// context instead of a single summary.
if (feature('CONTEXT_COLLAPSE') && contextCollapse) {
  const collapseResult = await contextCollapse.applyCollapsesIfNeeded(...)
  messagesForQuery = collapseResult.messages
}

这套管线的精髓是：能不触发大手术就别触发。每次 autocompact 都会让对话历史从「细粒度的原始消息」变成「一段摘要」，损失大量细节。先试成本低的小手术，token 降不下来了再做大手术。

3 套恢复机制：模型的天花板被 3 层兜底撑住​

本节源码：src/query.ts

LLM 调用可能失败的原因很多，每种都需要不同的恢复策略。QueryEngine 处理了 3 大类故障：

1. 输出太长：max_output_tokens​

模型被截断时，源码的处理是先升额度，再降容错：

const MAX_OUTPUT_TOKENS_RECOVERY_LIMIT = 3

if (isWithheldMaxOutputTokens(lastMessage)) {
  if (capEnabled && maxOutputTokensOverride === undefined) {
    // 第一轮恢复：从默认 8k 升到 ESCALATED_MAX_TOKENS（64k）
    logEvent('tengu_max_tokens_escalate', { escalatedTo: ESCALATED_MAX_TOKENS })
    state = {
      ...
      maxOutputTokensOverride: ESCALATED_MAX_TOKENS,
      transition: { reason: 'max_output_tokens_escalate' },
    }
    continue
  }

  if (maxOutputTokensRecoveryCount < MAX_OUTPUT_TOKENS_RECOVERY_LIMIT) {
    // 第二轮恢复：插入元消息让模型自己手动分块
    const recoveryMessage = createUserMessage({
      content:
        `Output token limit hit. Resume directly — no apology, no recap of what you were doing. ` +
        `Pick up mid-thought if that is where the cut happened. Break remaining work into smaller pieces.`,
      isMeta: true,
    })
    state = {
      ...
      maxOutputTokensRecoveryCount: maxOutputTokensRecoveryCount + 1,
      transition: {
        reason: 'max_output_tokens_recovery',
        attempt: maxOutputTokensRecoveryCount + 1,
      },
    }
    continue
  }
  // 恢复次数用尽，yield lastMessage 抛出错误
}

精妙之处：

  • 第一轮升级 token 上限——同一请求用更大的 max_tokens 重试，无感恢复，调用方不知道发生过
  • 第二轮才插入元消息——升级还失败才告诉模型「你输出太长，分块吧」，避免无谓地干扰模型
  • 元消息的内容不是「对不起，输出超限，请分块」——而是「Resume directly — no apology, no recap of what you were doing.」直接告诉模型别浪费时间在客套和回顾上，立刻继续。精简指令节省的就是 token

2. prompt too long：双路兜底​

API 返回「上下文太长」错误时，QueryEngine 同时保留两条恢复路径，按优先级 fallback：

const isWithheld413 =
  lastMessage?.type === 'assistant' &&
  lastMessage.isApiErrorMessage &&
  isPromptTooLongMessage(lastMessage)

// 路径 1：contextCollapse 先把所有已折叠的 staged collapses 一次性释放
if (
  feature('CONTEXT_COLLAPSE') &&
  contextCollapse &&
  state.transition?.reason !== 'collapse_drain_retry'
) {
  const drained = contextCollapse.recoverFromOverflow(...)
  if (drained.committed > 0) {
    state = {
      ...,
      transition: { reason: 'collapse_drain_retry', committed: drained.committed },
    }
    continue
  }
}

// 路径 2：reactive compact 做完整摘要
if ((isWithheld413 || isWithheldMedia) && reactiveCompact) {
  const compacted = await reactiveCompact.tryReactiveCompact(...)
  if (compacted) {
    state = {
      ...,
      transition: { reason: 'reactive_compact_retry' },
    }
    continue
  }
}

为什么要两条路径：

  • 路径 1（collapse drain）保持细粒度——把之前 collapsed 但还没发布的视图释放回去，仍然是原始消息粒度，模型「看到」的还是细节
  • 路径 2（reactive compact）退到摘要——这是兜底，drain 没东西可放时直接摘要整个历史

源码注释解释了为什么不直接走摘要：

Single-shot on each — if a retry still 413's, the next stage handles it or the error surfaces.

两段都不能无限重试——state.transition?.reason !== 'collapse_drain_retry' 这段检查保证下轮不再 drain，下轮要么 reactive compact 成功要么错误浮上来。避免抢救机制本身变成性能黑洞。

3. token budget 续杯​

最后一层是用户设定的成本上限：

if (feature('TOKEN_BUDGET')) {
  const decision = checkTokenBudget(
    budgetTracker!,
    toolUseContext.agentId,
    getCurrentTurnTokenBudget(),
    getTurnOutputTokens(),
  )

  if (decision.action === 'continue') {
    incrementBudgetContinuationCount()
    state = {
      ...,
      transition: { reason: 'token_budget_continuation' },
    }
    continue
  }
}

checkTokenBudget 的逻辑：

const COMPLETION_THRESHOLD = 0.9
const DIMINISHING_THRESHOLD = 500

const isDiminishing =
  tracker.continuationCount >= 3 &&
  deltaSinceLastCheck < DIMINISHING_THRESHOLD &&
  tracker.lastDeltaTokens < DIMINISHING_THRESHOLD

if (!isDiminishing && turnTokens < budget * COMPLETION_THRESHOLD) {
  // 续杯
}

if (isDiminishing || tracker.continuationCount > 0) {
  // 停
}

两个判断条件：

  • 未达 90%——单次续杯
  • 第 3 次续杯后增量 < 500 token——说明在「无效输出」（重复无效内容），不再续杯

diminishing returns 是这个机制最聪明的设计。如果模型一直在产出大量无效 token，续杯只会浪费钱——在涨 token 但产出没涨的情况下，停比续好。

流式回落：两端必须一致​

本节源码：src/services/tools/StreamingToolExecutor.ts

LLM 调用支持流式输出。但流式过程中可能会触发 fallback——主模型限流了，自动回退到备用模型。流式回落后会产生一个关键问题：之前流到的部分内容，tool_use_id 已经发了，但现在接的是另一个模型的响应。

源码在回落时做了一个关键操作——为之前流到的 assistant 消息发 tombstone（墓碑）：

if (streamingFallbackOccured) {
  // Yield tombstones for orphaned messages so they're removed from UI and transcript.
  // These partial messages (especially thinking blocks) have invalid signatures
  // that would cause "thinking blocks cannot be modified" API errors.
  for (const msg of assistantMessages) {
    yield { type: 'tombstone' as const, message: msg }
  }
  
  // Discard pending results from the failed streaming attempt and create
  // a fresh executor. This prevents orphan tool_results (with old tool_use_ids)
  // from being yielded after the fallback response arrives.
  if (streamingToolExecutor) {
    streamingToolExecutor.discard()
    streamingToolExecutor = new StreamingToolExecutor(...)
  }
}

为什么 tombstone 不能省？因为 partial 流到的内容里可能含有 thinking blocks，这些 blocks 在 API 端是有签名的。回退到不同模型后，这些 thinking 签名失效，重发就 400 错误。Tombstone 是一种特殊的「删除信号」——告诉调用方「这些消息作废，不要在 UI 或 transcript 里显示」。

源码注释里 "tombstone" 这个词本身就揭示了意图——墓碑标记「这里曾经有内容，但现在不存在」，是一种优雅的逻辑删除。

错误抑制：withhold 要两端对齐​

本节源码：src/query.ts

LLM 返回可恢复错误（prompt-too-long、max-output-tokens 等）时，源代码不会立刻 yield 给调用方，而是临时 withhold，等确定恢复不了再 yield：

let withheld = false
if (reactiveCompact?.isWithheldPromptTooLong(message)) {
  withheld = true
}
if (reactiveCompact?.isWithheldMediaSizeError(message)) {
  withheld = true
}
if (isWithheldMaxOutputTokens(message)) {
  withheld = true
}
if (!withheld) {
  yield yieldMessage
}

源码注释点出了一个微妙的配对要求：

Withhold (inside the stream loop) and recovery (after) must agree; CACHED_MAY_BE_STALE can flip during the 5-30s stream, and withhold-without-recover would eat the message.

Withhold 时和恢复时的判断必须对齐——否则消息就丢了。但有个工程难题：流式可能持续 5-30 秒，期间 features 状态可能变化（feature flags 用 CACHED_MAY_BE_STALE 表示可能过期）。源码的解决方案是 mediaRecoveryEnabled 在进流式循环之前就 hoist 一次：

// Hoist media-recovery gate once per turn. Withholding (inside the stream loop) and
// recovery (after) must agree; CACHED_MAY_BE_STALE can flip during the 5-30s stream,
// and withhold-without-recover would eat the message.
const mediaRecoveryEnabled =
  reactiveCompact?.isReactiveCompactEnabled() ?? false
if (
  !compactionResult &&
  ...
  !(
    reactiveCompact?.isReactiveCompactEnabled() && isAutoCompactEnabled()
  )
  ...
) { ...blocking check... }

hoist（提升）一次，整个流式和恢复期都用同一个值。这是分布式系统里典型的「避免中途判断翻转」模式——只要确定一次，就不变了。

给 AI 产品开发者的教训​

本节源码：src/QueryEngine.ts

QueryEngine 不是 API client。它拥有对话状态——历史、token 累加、permission denials、discovered skills。把它当成「远程 LLM 调用的代理」会设计错，因为代价在于「跨轮的连续性」。

几层兜底要用不同的恢复策略​

失败模式恢复策略关键参数
max_output_tokens（模型被截断）第一轮升 64k（同请求无感升级），第二轮才插元消息分块MAX_OUTPUT_TOKENS_RECOVERY_LIMIT = 3
prompt too long（上下文超限）优先 drain 已折叠的 collapse（细粒度），再退到 reactive compact（摘要兜底）hasAttemptedReactiveCompact flag 防止重试循环
用户预算耗尽90% 阈值 + 增量 < 500 token 才停DIMINISHING_THRESHOLD 防无效刷 token

好的恢复机制不是「一种兜底打天下」，而是按代价排序的多层组合：代价低的小手术先用，大手术兜底，且每层都有限次避免重试风暴。

状态机的 transition 字段值得借鉴​

普通的状态机字段只关心「当前是什么状态」；Claude Code 的 queryLoop 多了一个 transition 字段记录「从上一状态怎么到这里的」。这一个字段让单元测试从「解析 yield 出来的消息反推路径」变成「直接断言 transition 字段」——测试复杂度下降一个数量级。

Tombstone 比「删除消息」更优雅​

当一段流消息因为 fallback 失效需要被回退时，源代码不直接删除，而是 yield 一个 tombstone 标记「这里本来有内容，但现在作废」。订阅方（REPL UI、transcript 持久化、转写服务）各自决定如何处理 tombstone——UI 把它当作消息被撤回，transcript 直接忽略。这是事件溯源（event sourcing）思维的轻量化应用：不维护「当前状态」，而是记录「发生了什么事件」，状态由事件推导。

按约定没跑 pnpm build，请 pnpm start 验证 sidebar 显示是否符合预期。

Last updated on by Kimi Gao