Claude Code QueryEngine 深度分析

// src/QueryEngine.ts:184-207
export class QueryEngine {
  private config: QueryEngineConfig
  private mutableMessages: Message[]       // 消息历史（跨轮次持久化）
  private abortController: AbortController  // 中断控制
  private permissionDenials: SDKPermissionDenial[]  // 权限拒绝记录
  private totalUsage: NonNullableUsage      // 累积 token 用量
  private readFileState: FileStateCache     // 文件读取缓存
  private discoveredSkillNames = new Set<string>()  // 技能发现追踪
  private loadedNestedMemoryPaths = new Set<string>()  // 嵌套 memory 路径
}

export type QueryEngineConfig = {
  cwd: string                        // 工作目录
  tools: Tools                       // 可用工具列表
  commands: Command[]                 // 斜杠命令
  mcpClients: MCPServerConnection[]  // MCP 服务器连接
  agents: AgentDefinition[]          // 子 agent 定义
  canUseTool: CanUseToolFn           // 工具权限判断函数
  getAppState: () => AppState        // 获取应用状态
  setAppState: (f: (prev: AppState) => AppState) => void
  initialMessages?: Message[]        // 初始消息（会话恢复）
  readFileCache: FileStateCache      // 文件状态缓存
  customSystemPrompt?: string        // 自定义系统提示词
  appendSystemPrompt?: string        // 追加系统提示词
  userSpecifiedModel?: string        // 用户指定模型
  fallbackModel?: string             // 降级模型
  thinkingConfig?: ThinkingConfig    // Thinking 配置
  maxTurns?: number                  // 最大轮次数
  maxBudgetUsd?: number              // 最大 USD 预算
  taskBudget?: { total: number }     // API 侧 token 预算
  jsonSchema?: Record<string, unknown>  // 结构化输出 schema
  verbose?: boolean                  // 详细输出
  replayUserMessages?: boolean       // SDK 模式下重放用户消息
  includePartialMessages?: bool     // 是否包含流式中间事件
  abortController?: AbortController  // 外部中断控制
  snipReplay?: (...) => ...          // snip 压缩回放处理器
}

// src/QueryEngine.ts:1249-1285
const engine = new QueryEngine({ /* config */ })
yield* engine.submitMessage(prompt, { uuid: promptUuid, isMeta })

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
  transition: Continue | undefined  // 上一次迭代的继续原因
}

export async function* queryModelWithStreaming({
  messages, systemPrompt, thinkingConfig, tools, signal, options
}): AsyncGenerator<StreamEvent | AssistantMessage | SystemAPIErrorMessage> {
  return yield* withStreamingVCR(messages, async function* () {
    yield* queryModel(messages, systemPrompt, thinkingConfig, tools, signal, options)
  })
}

return {
  model: normalizeModelStringForAPI(options.model),
  messages: addCacheBreakpoints(messagesForAPI, ...),
  system,                           // 系统提示词 (带 cache_control)
  tools: allTools,                  // 工具定义数组
  tool_choice: options.toolChoice,
  betas: betasParams,               // beta headers
  metadata: getAPIMetadata(),       // user_id, device_id, session_id
  max_tokens: maxOutputTokens,
  thinking,                         // { type: 'adaptive' } 或 { budget_tokens, type: 'enabled' }
  temperature,                      // 仅 thinking 禁用时发送
  context_management,               // API 侧上下文管理策略
  output_config: { effort, task_budget, format },
  speed,                            // fast mode
  ...extraBodyParams,               // Bedrock/Vertex 额外参数
}

switch (part.type) {
  case 'message_start':
    // 初始化 partialMessage, 记录 ttftMs, 更新 usage
  case 'content_block_start':
    // 按 block 类型初始化: text/tool_use/thinking/server_tool_use
    // tool_use: contentBlocks[i] = { ...block, input: '' }
  case 'content_block_delta':
    // text_delta: 累加 text
    // input_json_delta: 累加 input JSON 字符串
    // thinking_signature_delta: 累加 signature
  case 'content_block_stop':
    // 构建 AssistantMessage, yield 给调用方
    // 解析 tool_use input JSON
  case 'message_delta':
    // 更新 usage, stop_reason
    // 写回已 yield 的 message 对象（直接属性修改）
    // 计算 USD 成本
  case 'message_stop':
    break
}

let streamingToolExecutor = useStreamingToolExecution
  ? new StreamingToolExecutor(toolUseContext.options.tools, canUseTool, toolUseContext)
  : null

// 流中：每收到 tool_use block 立即加入执行队列
for (const toolBlock of msgToolUseBlocks) {
  streamingToolExecutor.addTool(toolBlock, message)
}
// 流中：获取已完成的结果
for (const result of streamingToolExecutor.getCompletedResults()) { ... }
// 流结束后：获取剩余结果
for (const update of streamingToolExecutor.getRemainingResults()) { ... }

const toolUpdates = streamingToolExecutor
  ? streamingToolExecutor.getRemainingResults()
  : runTools(toolUseBlocks, assistantMessages, canUseTool, toolUseContext)

for await (const update of toolUpdates) {
  if (update.message) {
    yield update.message
    toolResults.push(...normalizeMessagesForAPI([update.message], tools))
  }
  if (update.newContext) {
    updatedToolUseContext = { ...update.newContext, queryTracking }
  }
}

const next: State = {
  messages: [...messagesForQuery, ...assistantMessages, ...toolResults],
  toolUseContext: toolUseContextWithQueryTracking,
  autoCompactTracking: tracking,
  turnCount: nextTurnCount,
  maxOutputTokensRecoveryCount: 0,
  hasAttemptedReactiveCompact: false,
  pendingToolUseSummary: nextPendingToolUseSummary,
  maxOutputTokensOverride: undefined,
  stopHookActive,
  transition: { reason: 'next_turn' },
}
state = next  // while(true) 继续

export async function* withRetry<T>(
  getClient: () => Promise<Anthropic>,
  operation: (client, attempt, context) => Promise<T>,
  options: RetryOptions,
): AsyncGenerator<SystemAPIErrorMessage, T>

if (consecutive529Errors >= MAX_529_RETRIES) {
  if (options.fallbackModel) {
    throw new FallbackTriggeredError(options.model, options.fallbackModel)
  }
}

if (innerError instanceof FallbackTriggeredError && fallbackModel) {
  currentModel = fallbackModel
  attemptWithFallback = true
  // 清空 assistant messages, 重新发起请求
  assistantMessages.length = 0
  toolResults.length = 0
  toolUseBlocks.length = 0
}

type StoredCostState = {
  totalCostUSD: number               // 总 USD 费用
  totalAPIDuration: number           // 总 API 耗时
  totalAPIDurationWithoutRetries: number  // 不含重试的 API 耗时
  totalToolDuration: number          // 总工具执行耗时
  totalLinesAdded: number            // 总添加行数
  totalLinesRemoved: number          // 总删除行数
  lastDuration: number | undefined   // 最近一次耗时
  modelUsage: { [modelName: string]: ModelUsage }  // 按模型分类的用量
}

// src/services/api/claude.ts:2993-3005
export function accumulateUsage(totalUsage, messageUsage): NonNullableUsage {
  return {
    input_tokens: totalUsage.input_tokens + messageUsage.input_tokens,
    cache_creation_input_tokens: totalUsage.cache_creation_input_tokens + messageUsage.cache_creation_input_tokens,
    cache_read_input_tokens: totalUsage.cache_read_input_tokens + messageUsage.cache_read_input_tokens,
    output_tokens: totalUsage.output_tokens + messageUsage.output_tokens,
    // ...
  }
}

const costUSDForPart = calculateUSDCost(resolvedModel, usage)
costUSD += addToTotalSessionCost(costUSDForPart, usage, options.model)

// src/query/tokenBudget.ts:45-93
export function checkTokenBudget(tracker, agentId, budget, globalTurnTokens) {
  // 90% 阈值：未达到则注入 nudge 消息鼓励继续
  // diminishing returns：连续 3 次且增量 <500 token 则提前停止
  // 返回 continue 或 stop 决策
}

export const getSystemContext = memoize(async () => {
  return {
    gitStatus,           // git 分支、状态、最近提交
    cacheBreaker,        // 缓存破坏标记（ant-only）
  }
})

export const getUserContext = memoize(async () => {
  return {
    claudeMd,            // CLAUDE.md 文件内容
    currentDate,         // 当前日期
  }
})

// 1. 获取系统提示词各部分
const { defaultSystemPrompt, userContext, systemContext } = await fetchSystemPromptParts({
  tools, mainLoopModel, additionalWorkingDirectories, mcpClients, customSystemPrompt
})

// 2. 组装最终系统提示词
const systemPrompt = asSystemPrompt([
  ...(customPrompt !== undefined ? [customPrompt] : defaultSystemPrompt),
  ...(memoryMechanicsPrompt ? [memoryMechanicsPrompt] : []),
  ...(appendSystemPrompt ? [appendSystemPrompt] : []),
])

type Message =
  | UserMessage      // { type: 'user', message: { role: 'user', content }, uuid, timestamp }
  | AssistantMessage // { type: 'assistant', message: { role: 'assistant', content, usage, stop_reason }, uuid }
  | SystemMessage    // { type: 'system', subtype: '...', content }
  | StreamEvent      // { type: 'stream_event', event: RawStreamEvent }
  | AttachmentMessage  // { type: 'attachment', attachment: {...} }
  | ProgressMessage  // { type: 'progress', toolUseID, data }
  | TombstoneMessage // { type: 'tombstone', message: Message } — UI 删除标记
  | ToolUseSummaryMessage  // { type: 'tool_use_summary', summary, precedingToolUseIds }

export type QueryDeps = {
  callModel: typeof queryModelWithStreaming  // API 调用
  microcompact: typeof microcompactMessages  // 微压缩
  autocompact: typeof autoCompactIfNeeded    // 自动压缩
  uuid: () => string                         // UUID 生成
}

export type QueryConfig = {
  sessionId: SessionId
  gates: {
    streamingToolExecution: boolean  // 流式工具执行
    emitToolUseSummaries: boolean    // 工具使用摘要
    isAnt: boolean                   // ant 用户
    fastModeEnabled: boolean         // 快速模式
  }
}