Claude Code 源码阅读 Phase2：Message 与上下文结构

总览Phase 2 程序员视角流程图

Message 在 Agent Loop 中的主线

核心配对关系

assistant message
  content: [ text, tool_use ]
              |
              v
        本地 runtime 执行工具
              |
              v
user message
  content: [ tool_result ]
              |
              v
        下一轮模型继续推理

1src/utils/messages.ts：消息创建入口

createAssistantMessage()

export function createAssistantMessage({
  content,
  usage,
  isVirtual,
}: {
  content: string | BetaContentBlock[]
  usage?: Usage
  isVirtual?: true
}): AssistantMessage {
  return baseCreateAssistantMessage({
    content:
      typeof content === 'string'
        ? [
            {
              type: 'text' as const,
              text: content === '' ? NO_CONTENT_MESSAGE : content,
            } as BetaContentBlock, "comment">// NOTE: citations field is not supported in Bedrock API
          ]
        : content,
    usage,
    isVirtual,
  })
}

关键点：

• assistant 的 content 不一定是字符串，也可以是 content block 数组。
• 如果传入字符串，源码会把它包装成 { type: "text", text: ... }。
• 这说明 Claude Code 内部更倾向统一处理 content block。
• assistant message 可以携带 tool_use block，这是模型请求工具调用的结构。

createUserMessage()

export function createUserMessage({
  content,
  isMeta,
  isVisibleInTranscriptOnly,
  isVirtual,
  isCompactSummary,
  summarizeMetadata,
  toolUseResult,
  mcpMeta,
  uuid,
  timestamp,
  imagePasteIds,
  sourceToolAssistantUUID,
  permissionMode,
  origin,
}: {
  content: string | ContentBlockParam[]
  ...
}): UserMessage

作用：创建 Claude Code 内部的 user message。

关键点：

• user message 不只表示真实用户输入。
• 工具执行结果 tool_result 也会作为 user message 回填给模型。
• 内部结构里同时有 type: "user" 和 message.role: "user"。
• type 更偏 Claude Code 内部分类，message.role 更接近 API 消息角色。

一句话总结：
Message 不是普通聊天字符串，而是 agent loop 中传递状态的结构化对象。

normalizeMessages()

export function normalizeMessages(messages: Message[]): NormalizedMessage[] {
  let isNewChain = false
  return messages.flatMap(message => {
    switch (message.type) {
      ...
    }
  })
}

作用：
把一条可能包含多个 content block 的 message，拆成多条更标准的 normalized message。

解决的问题：

• assistant 一次响应里可能同时包含 text 和 tool_use。
• user message 里也可能包含字符串、图片、tool_result 等不同 block。
• 多 block message 不方便 UI 展示、查找、工具配对。
• 拆开后，每条 normalized message 通常只包含一个 content block。

关键源码：

• 使用 flatMap，表示一条 message 可以拆成多条 message。
• assistant 分支会遍历 message.message.content。
• user 字符串会被转成 { type: "text", text: ... }。
• 拆分后用 deriveUUID(message.uuid, index) 生成稳定的新 uuid。

一句话总结：
normalizeMessages() 是把“API 友好的多 block 消息”转换成“系统内部更容易处理的一 block 消息”。

2工具请求与工具结果判断

位置：src/utils/messages.ts:829-851

isToolUseRequestMessage()

export function isToolUseRequestMessage(
  message: Message,
): message is ToolUseRequestMessage {
  return (
    message.type === 'assistant' &&
    "comment">// Note: stop_reason === 'tool_use' is unreliable -- it's not always set correctly
    message.message.content.some(_ => _.type === 'tool_use')
  )
}

作用：
判断一条 message 是否是 assistant 发出的工具调用请求。

关键逻辑：

• message.type 必须是 "assistant"。
• message.message.content 里必须存在 type === "tool_use" 的 block。
• 源码没有依赖 stop_reason === "tool_use"，因为注释说明它不总是可靠。

结论：
工具请求来自 assistant message，真正可靠的判断依据是 content block 里的 tool_use。

isToolUseResultMessage()

export function isToolUseResultMessage(
  message: Message,
): message is ToolUseResultMessage {
  return (
    message.type === 'user' &&
    ((Array.isArray(message.message.content) &&
      message.message.content[0]?.type === 'tool_result') ||
      Boolean(message.toolUseResult))
  )
}

作用：
判断一条 message 是否是工具执行结果。

关键逻辑：

• message.type 必须是 "user"。
• content 第一个 block 是 tool_result，或者内部字段 toolUseResult 存在。

结论：
工具结果会作为 user message 回填给模型。agent loop 的关键配对关系是：

assistant/tool_use
-> user/tool_result

reorderMessagesInUI()

export function reorderMessagesInUI(
  messages: (
    | NormalizedUserMessage
    | NormalizedAssistantMessage
    | AttachmentMessage
    | SystemMessage
  )[],
  syntheticStreamingToolUseMessages: NormalizedAssistantMessage[],
)

const toolUseGroups = new Map<
  string,
  {
    toolUse: ToolUseRequestMessage | null
    preHooks: AttachmentMessage[]
    toolResult: NormalizedUserMessage | null
    postHooks: AttachmentMessage[]
  }
>()

作用：
为 UI 展示重新排列消息，把同一次工具调用的相关消息聚合在一起。

前置条件：
通常处理的是 normalize 之后的 message，因为函数参数是 NormalizedUserMessage / NormalizedAssistantMessage。

核心结构：
使用 Map<string, ToolUseGroup> 按 tool_use_id 分组。

ToolUseGroup 大致包含：

• toolUse：assistant 发出的工具请求
• preHooks：工具执行前相关附件/钩子消息
• toolResult：user message 形式的工具结果
• postHooks：工具执行后相关附件/钩子消息

关键理解：
这段主要服务 UI 展示，不是 agent loop 的主闭环逻辑。

一句话总结：
reorderMessagesInUI() 把零散的 tool_use、tool_result、hook 附件按 tool_use_id 组织成一组，让用户看到完整的工具调用过程。

3src/services/api/claude.ts：内部消息到 API 消息的转换

userMessageToMessageParam()

export function userMessageToMessageParam(
  message: UserMessage,
  addCache = false,
  enablePromptCaching: boolean,
  querySource?: QuerySource,
): MessageParam {
  if (addCache) {
    if (typeof message.message.content === 'string') {
      return {
        role: 'user',
        content: [
          {
            type: 'text',
            text: message.message.content,
            ...(enablePromptCaching && {
              cache_control: getCacheControl({ querySource }),
            }),
          },
        ],
      }
    } else {
      return {
        role: 'user',
        content: message.message.content.map((_, i) => ({
          ..._,
          ...(i === message.message.content.length - 1
            ? enablePromptCaching
              ? { cache_control: getCacheControl({ querySource }) }
              : {}
            : {}),
        })),
      }
    }
  }
  "comment">// Clone array content to prevent in-place mutations (e.g., insertCacheEditsBlock's
  "comment">// splice) from contaminating the original message. Without cloning, multiple calls
  "comment">// to addCacheBreakpoints share the same array and each splices in duplicate cache_edits.
  return {
    role: 'user',
    content: Array.isArray(message.message.content)
      ? [...message.message.content]
      : message.message.content,
  }
}

作用：
把 Claude Code 内部的 UserMessage 转换成 Anthropic API 的 MessageParam。

核心转换：
内部 UserMessage：

{
type: "user",
message: {
role: "user",
content: ...
},
uuid,
timestamp,
...
}

转换成 API message：

{
role: "user",
content: ...
}

关键点：

• API 不需要 uuid、timestamp、toolUseResult 等内部运行时字段。
• 如果 content 是数组，源码会用 [...content] 浅拷贝，避免后续逻辑原地修改污染原 message。
• addCache 分支是 prompt caching 优化路径，暂时不是 agent loop 主线。

一句话总结：
userMessageToMessageParam() 是“内部 user message”到“模型 API user message”的出口。

assistantMessageToMessageParam()

export function assistantMessageToMessageParam(
  message: AssistantMessage,
  addCache = false,
  enablePromptCaching: boolean,
  querySource?: QuerySource,
): MessageParam {
  if (addCache) {
    if (typeof message.message.content === 'string') {
      return {
        role: 'assistant',
        content: [
          {
            type: 'text',
            text: message.message.content,
            ...(enablePromptCaching && {
              cache_control: getCacheControl({ querySource }),
            }),
          },
        ],
      }
    } else {
      return {
        role: 'assistant',
        content: message.message.content.map((_, i) => ({
          ..._,
          ...(i === message.message.content.length - 1 &&
          _.type !== 'thinking' &&
          _.type !== 'redacted_thinking' &&
          (feature('CONNECTOR_TEXT') ? !isConnectorTextBlock(_) : true)
            ? enablePromptCaching
              ? { cache_control: getCacheControl({ querySource }) }
              : {}
            : {}),
        })),
      }
    }
  }
  return {
    role: 'assistant',
    content: message.message.content,
  }
}

作用：
把 Claude Code 内部的 AssistantMessage 转换成 Anthropic API 的 assistant message。

核心转换：
内部 AssistantMessage：

{
type: "assistant",
message: {
role: "assistant",
content: [...]
},
uuid,
timestamp,
...
}

转换成 API message：

{
role: "assistant",
content: message.message.content
}

关键点：

• assistant message 的 content 可能包含 text block。
• assistant message 的 content 也可能包含 tool_use block。
• 转换到 API message 时，tool_use block 会被保留下来。
• addCache 分支属于 prompt caching 优化路径，不是当前主线。

一句话总结：
assistantMessageToMessageParam() 是“内部 assistant message”到“模型 API assistant message”的出口，并保留 assistant 发出的 tool_use。

4src/query/tokenBudget.ts：token budget 与上下文继续策略

BudgetTracker / createBudgetTracker()

const COMPLETION_THRESHOLD = 0.9
const DIMINISHING_THRESHOLD = 500

export type BudgetTracker = {
  continuationCount: number
  lastDeltaTokens: number
  lastGlobalTurnTokens: number
  startedAt: number
}

export function createBudgetTracker(): BudgetTracker {
  return {
    continuationCount: 0,
    lastDeltaTokens: 0,
    lastGlobalTurnTokens: 0,
    startedAt: Date.now(),
  }
}

作用：
记录 token budget 自动继续过程中的状态。

关键字段：

• continuationCount：已经自动继续了几次。
• lastDeltaTokens：上一次检查时，相比更早一次新增了多少 token。
• lastGlobalTurnTokens：上一次检查时，本 turn 已经消耗的 token 数。
• startedAt：开始追踪的时间，用于后续统计耗时。

关键阈值：

• COMPLETION_THRESHOLD = 0.9：达到预算 90% 附近就倾向停止。
• DIMINISHING_THRESHOLD = 500：连续新增 token 很少时，认为继续收益变小。

一句话总结：
BudgetTracker 是 token budget 自动继续机制的“记账本”，真正的继续/停止判断在后面的 checkTokenBudget()。

TokenBudgetDecision

type ContinueDecision = {
  action: 'continue'
  nudgeMessage: string
  continuationCount: number
  pct: number
  turnTokens: number
  budget: number
}

type StopDecision = {
  action: 'stop'
  completionEvent: {
    continuationCount: number
    pct: number
    turnTokens: number
    budget: number
    diminishingReturns: boolean
    durationMs: number
  } | null
}

export type TokenBudgetDecision = ContinueDecision | StopDecision

作用：
定义 checkTokenBudget() 的返回结果类型。

为什么不用 boolean：
因为继续和停止都需要携带不同的信息。

• continue：需要 nudgeMessage、当前百分比、token 数、继续次数。
• stop：可能需要 completionEvent，记录停止时的统计信息。

核心结构：

• ContinueDecision 用 action: "continue" 表示应该继续。
• StopDecision 用 action: "stop" 表示应该停止。
• TokenBudgetDecision = ContinueDecision | StopDecision 是联合类型。

C++ 类比：
可以理解为 std::variant<ContinueDecision, StopDecision>。
其中 action 字段是判断当前 variant 类型的标签。

一句话总结：
TokenBudgetDecision 把“是否继续”和“继续/停止所需信息”一起表达出来，比 true/false 更适合 agent loop。

checkTokenBudget()

export function checkTokenBudget(
  tracker: BudgetTracker,
  agentId: string | undefined,
  budget: number | null,
  globalTurnTokens: number,
): TokenBudgetDecision {
  if (agentId || budget === null || budget <= 0) {
    return { action: 'stop', completionEvent: null }
  }

  const turnTokens = globalTurnTokens
  const pct = Math.round((turnTokens / budget) * 100)
  const deltaSinceLastCheck = globalTurnTokens - tracker.lastGlobalTurnTokens

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

  if (!isDiminishing && turnTokens < budget * COMPLETION_THRESHOLD) {
    tracker.continuationCount++
    tracker.lastDeltaTokens = deltaSinceLastCheck
    tracker.lastGlobalTurnTokens = globalTurnTokens
    return {
      action: 'continue',
      nudgeMessage: getBudgetContinuationMessage(pct, turnTokens, budget),
      continuationCount: tracker.continuationCount,
      pct,
      turnTokens,
      budget,
    }
  }

  if (isDiminishing || tracker.continuationCount > 0) {
    return {
      action: 'stop',
      completionEvent: {
        continuationCount: tracker.continuationCount,
        pct,
        turnTokens,
        budget,
        diminishingReturns: isDiminishing,
        durationMs: Date.now() - tracker.startedAt,
      },
    }
  }

  return { action: 'stop', completionEvent: null }
}

作用：
当用户指定 token budget 时，判断当前 agent turn 是否应该继续工作，或者停止。

参数：

• tracker：BudgetTracker 记账本。
• agentId：当前是否是 subagent。
• budget：用户指定的 token budget，可能为 null。
• globalTurnTokens：当前 turn 已经消耗的 token 数。

第一步：排除不适用场景
如果满足以下任一条件，直接返回 stop：

• 当前是 subagent：agentId 存在。
• 没有 token budget：budget === null。
• budget 无效：budget <= 0。

第二步：计算当前进度

• turnTokens = globalTurnTokens
• pct = 当前 token / budget * 100
• deltaSinceLastCheck = 当前 token - 上次检查时 token

第三步：判断收益递减
isDiminishing 同时满足：

• 已经自动继续过至少 3 次。
• 本次新增 token 小于 DIMINISHING_THRESHOLD。
• 上次新增 token 也小于 DIMINISHING_THRESHOLD。

其中：

• DIMINISHING_THRESHOLD = 500

第四步：continue 条件
如果：

• 没有收益递减。
• 当前 token 还没达到预算 90%。

则：

• 更新 tracker。
• 返回 action: "continue"。
• 生成 nudgeMessage，提醒模型继续工作，不要总结。

其中：

• COMPLETION_THRESHOLD = 0.9

第五步：stop 条件
如果已经收益递减，或者此前发生过自动继续，则返回：

• action: "stop"
• completionEvent，记录继续次数、token 数、预算、是否收益递减、耗时等。

兜底：
如果没有发生过自动继续，也不满足继续条件，则普通 stop，completionEvent 为 null。

一句话总结：
checkTokenBudget() 是 token budget 自动继续机制的决策函数：没到预算且仍有产出就继续，接近预算或收益递减就停止。

token budget 在 query loop 中的接入

if (decision.action === 'continue') {
          incrementBudgetContinuationCount()
          logForDebugging(
            `Token budget continuation #${decision.continuationCount}: ${decision.pct}% (${decision.turnTokens.toLocaleString()} / ${decision.budget.toLocaleString()})`,
          )
          "comment">// 判断符合继续执行条件，重写state
          state = {
            messages: [
              ...messagesForQuery,
              ...assistantMessages,
              createUserMessage({
                content: decision.nudgeMessage,
                isMeta: true,
              }),
            ],
            toolUseContext,
            autoCompactTracking: tracking,
            maxOutputTokensRecoveryCount: 0,
            hasAttemptedReactiveCompact: false,
            maxOutputTokensOverride: undefined,
            pendingToolUseSummary: undefined,
            stopHookActive: undefined,
            turnCount,
            transition: { reason: 'token_budget_continuation' },
          }
          continue
        }

作用：
把 checkTokenBudget() 的 continue 决策接回 agent loop。

调用参数：

• budgetTracker!：token budget 记账本。
• toolUseContext.agentId：判断当前是否是 subagent。
• getCurrentTurnTokenBudget()：当前 turn 的 token budget。
• getTurnOutputTokens()：当前 turn 已输出 token。

continue 时的核心动作：
重新构造 state.messages：

messages =
messagesForQuery
+ assistantMessages
+ createUserMessage({
content: decision.nudgeMessage,
isMeta: true
})

关键点：

• decision.nudgeMessage 是“继续工作，不要总结”的提示。
• 这条消息用 createUserMessage() 创建。
• isMeta: true 表示它是系统插入的元消息，不是用户真实输入。
• 然后 agent loop 会进入下一轮模型调用。

一句话总结：
token budget 的“继续”不是直接让程序生成，而是往 messages 里追加一条 meta user message，让模型在下一轮继续完成任务。

5src/utils/tokenBudget.ts：解析用户输入里的 token budget

parseTokenBudget()

export function parseTokenBudget(text: string): number | null {
  const startMatch = text.match(SHORTHAND_START_RE)
  if (startMatch) return parseBudgetMatch(startMatch[1]!, startMatch[2]!)
  const endMatch = text.match(SHORTHAND_END_RE)
  if (endMatch) return parseBudgetMatch(endMatch[1]!, endMatch[2]!)
  const verboseMatch = text.match(VERBOSE_RE)
  if (verboseMatch) return parseBudgetMatch(verboseMatch[1]!, verboseMatch[2]!)
  return null
}

作用：
从用户输入文本里识别 token budget，并转换成具体数字。

支持的写法：

• 开头简写：+500k ...
• 结尾简写：... +500k
• 自然语言：use 2m tokens / spend 500k tokens

单位换算：

• k = 1,000
• m = 1,000,000
• b = 1,000,000,000

核心流程：
1. 先匹配开头简写 SHORTHAND_START_RE。
2. 再匹配结尾简写 SHORTHAND_END_RE。
3. 再匹配自然语言形式 VERBOSE_RE。
4. 如果都没有匹配，返回 null。

关键理解：
parseTokenBudget() 只负责把用户文本里的 budget 提取成数字，不负责决定是否继续；继续/停止判断在 src/query/tokenBudget.ts 的 checkTokenBudget()。

一句话总结：
parseTokenBudget() 是 token budget 机制的输入解析器，把 +500k / use 2m tokens 这类文本变成后续 agent loop 能使用的数字预算。

token budget 的 REPL 输入入口

if (feature('TOKEN_BUDGET')) {
    const parsedBudget = input ? parseTokenBudget(input) : null;
    snapshotOutputTokensForTurn(parsedBudget ?? getCurrentTurnTokenBudget());
}

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

作用：
在用户输入进入 agent loop 之前，从输入文本中解析 token budget，并记录到当前 turn 状态里。

关键流程：
1. REPL 拿到用户输入 input。
2. 调用 parseTokenBudget(input)。
3. 如果解析到 budget，就使用解析结果。
4. 如果没有解析到，就沿用当前已有的 getCurrentTurnTokenBudget()。
5. 调用 snapshotOutputTokensForTurn(...) 保存当前 turn 的 token budget。
6. 后续 query loop 通过 getCurrentTurnTokenBudget() 读取，并传给 checkTokenBudget()。

关键源码：

总结Phase 2 真正要记住什么

五个关键结论

1. Message 不是普通字符串，而是带 role/content/uuid/状态字段的结构化对象。
2. assistant 的 tool_use 是工具请求；user 的 tool_result 是工具结果。
3. normalizeMessages() 负责把多 block 消息拆成更容易展示和配对的形式。
4. *MessageToMessageParam() 是内部消息到模型 API 消息的出口。
5. token budget 让 Agent 可以在预算内自动继续，也能在接近上限或收益递减时停止。

最小理解模型

messages = [user_input]

while True:
    api_messages = convert_internal_messages_to_api_params(messages)
    assistant = call_model(api_messages)
    messages.append(assistant)

    if has_tool_use(assistant):
        result = run_tool(assistant.tool_use)
        messages.append(user_tool_result(result))
        continue

    decision = check_token_budget(...)
    if decision.action == "continue":
        messages.append(user_nudge(decision.nudgeMessage))
        continue

    break