PR #147：让 Claude 流式片段和最终消息共享同一个内容身份

Buffin · main...fix/claude-native-message-id · 2026-07-16 · 已合并 · 自包含，读完即弃

写法说明：本文按「逐跳走读」展开——每个机制都给真实代码片段（取自分支、经裁剪，青色斜体注释为解读所加，灰色斜体是源码原注释的保留或意译），每段代码标注所在文件。旅程结尾有一张「排查路标」：将来出问题时，症状对应去哪个文件看哪个函数。

1先说结论

Claude 的同一条原生消息会先以 stream_event 逐块输出，再以一个或多个 assistant envelope 给出权威完成内容。以前 Buffin 按“第几个 assistant envelope + envelope 内部下标”生成 itemId；当最终 envelope 拆分、重排或修正文案时，delta 与 completion 会落到不同身份上。

这个 PR 把原生 message.id 提升为消息分组键。流开始时创建组，内容块出现时分配 Buffin 自己的 m1b0 一类 ID，最终 envelope 再按角色和已累积文本认领未完成块。结果是流式文本、最终文本、持久化回放和重复投递都围绕同一个内容身份收敛。

2状态先行：从一个计数器变成消息组

旧状态只有 messageOrdinal，它假定每个最终 assistant envelope 就是一条新消息。新状态显式保存“原生消息 → 内容块”的关系，以及当前生命周期、最近关闭的流、无生命周期回退组和重放标记。

type ClaudeContentBlock = {
  itemId: string
  role: 'assistant' | 'reasoning'
  accumulatedText: string
  completed: boolean
}

type ClaudeMessageGroup = {
  ordinal: number
  blocksByIndex: Map<number | string, ClaudeContentBlock>
  blocksInOrder: ClaudeContentBlock[]
  streamStarted: boolean
  streamClosed: boolean
}

export type ClaudeTranslationState = {
  messageGroups: Map<string, ClaudeMessageGroup>
  assistantEnvelopeGroups: Map<string, ClaudeMessageGroup>
  activeGroup?: ClaudeMessageGroup | undefined
  latestStreamGroup?: ClaudeMessageGroup | undefined
  replayingActiveStream: boolean
  fallbackGroup?: ClaudeMessageGroup | undefined
  allowStandaloneDeltas: boolean
  streamLifecycleSeen: boolean
  nextGroupOrdinal: number
}

blocksByIndex 服务流式阶段：provider index 能快速找到已经声明的块。blocksInOrder 服务完成阶段：最终 envelope 的本地下标不可信，所以要在原始分配顺序里寻找仍未完成的同角色块。两种索引面向两个不同协议阶段。

3旅程 A：一条 Claude 消息如何保持同一个身份

这条旅程从 Claude 的 message_start 开始，经过块声明和 token delta，最后落到一个或多个 authoritative assistant envelope。走完它，就能知道 itemId 在哪里生成、何时复用、哪些输入会被抑制。

A.1旧算法错位在“最终 envelope 的本地下标”

在 main 上，translator 收到一个 assistant envelope 就读取当前消息序号，并用 envelope 的数组下标生成 ID，随后立即把消息序号加一。这个算法对“一个流对应一个完整 envelope，而且内容顺序不变”的输入成立。

const messageOrdinal = state.messageOrdinal
for (const [index, item] of message.content.entries()) {
  const itemId = contentBlockItemId(messageOrdinal, index)
  const event = assistantContentEvent(
    item,
    parsed.is_plan === true,
    itemId,
    occurredAt,
    raw,
  )
  if (event) events.push(event)
}
state.messageOrdinal += 1

Claude 可以把 reasoning 和 text 拆成两个 envelope，而每个 envelope 都从 index 0 开始；也可以在最终内容里改变顺序或补标点。此时“envelope index”不再等于“流中 content block 的身份”，于是同一段内容的 delta 和 completion 会拥有不同 itemId。

A.2生命周期先建立 provider message group

message_start 是分组的入口。translator 读取非空原生 ID，通过 messageGroupForId 取回或创建组，并把它同时记为当前活动组和最近一个有生命周期的组。后者用于最终 envelope 没带 ID 时的关联。

function beginMessageStream(
  state: ClaudeTranslationState,
  event: Record<string, unknown>,
): void {
  state.streamLifecycleSeen = true
  const messageId = readProviderMessageId(getRecord(event.message).id)
  if (!messageId) return

  const group = messageGroupForId(state, messageId)
  state.replayingActiveStream = group.streamStarted
  group.streamStarted = true
  if (!state.replayingActiveStream) group.streamClosed = false
  state.activeGroup = group
  state.latestStreamGroup = group
}

function closeMessageStream(state: ClaudeTranslationState): void {
  state.streamLifecycleSeen = true
  if (state.activeGroup && !state.replayingActiveStream) {
    state.activeGroup.streamClosed = true
  }
  state.activeGroup = undefined
  state.replayingActiveStream = false
}

同一个原生 ID 再次触发 message_start 时，streamStarted 已经为真，因此当前生命周期被标为 replay。它仍然能走完整个协议序列，但不会再次追加文本。

A.3delta 只进入已声明且仍开放的块

content_block_start 先把 provider index、角色和新生成的 itemId 放入活动组。随后到来的 delta 必须命中同一个 index，角色一致，并且块尚未完成。未知 index、已关闭流和重放流都会返回 null。

function streamBlockForDelta(
  state: ClaudeTranslationState,
  role: ClaudeContentBlock['role'],
  index: unknown,
  text: string,
): ClaudeContentBlock | null {
  const streamIndex = readStreamIndex(index)
  if (streamIndex === undefined) return null

  const activeGroup = state.activeGroup
  const group = activeGroup ?? standaloneDeltaGroup(state)
  if (!group || state.replayingActiveStream || group.streamClosed) return null

  const block = group.blocksByIndex.get(streamIndex)
  if (block) {
    if (block.role !== role || block.completed) return null
    block.accumulatedText += text
    return block
  }
  if (activeGroup) return null

  const fallbackBlock = createContentBlock(group, role)
  fallbackBlock.accumulatedText = text
  group.blocksByIndex.set(streamIndex, fallbackBlock)
  return fallbackBlock
}

这里的严格门槛把生命周期当成正常路径的协议证据。只有 runner 明确开启 standalone 模式，而且整次交互尚未见过任何生命周期时，delta 才能自己创建 fallback block。

A.4最终 completion 反向认领未完成块

最终 assistant envelope 先按原生 ID 找组；没有 ID 时，优先找当前或最近的未完成流。真正选择块时不再看 envelope 的本地下标，而是只看同角色、尚未完成的候选块。

如果某个候选块累积出的文本与最终文本完全相同，它优先被认领。这解决同角色块在 final envelope 中换序的问题。如果最终文本被 Claude 修正过，找不到精确相等项，就按原始块分配顺序认领第一个候选；completion 仍沿用流式 itemId，只是文本更新为权威版本。

function completionBlockForText(
  group: ClaudeMessageGroup,
  role: ClaudeContentBlock['role'],
  text: string,
): ClaudeContentBlock {
  const incompleteBlocks = group.blocksInOrder.filter(
    (block) => !block.completed && block.role === role,
  )
  const block =
    incompleteBlocks.find(
      (candidate) => candidate.accumulatedText === text,
    ) ??
    incompleteBlocks[0] ??
    createContentBlock(group, role)
  block.completed = true
  return block
}

如果 assistant envelope 只包含工具块，组仍然会被绑定到该原生消息，但不会凭空消耗一个文本块。后续不同内容的 envelope 可以继续使用同一原生 ID；assistantEnvelopeGroups 用“message ID + 完整 content”区分它们。

A.5缺少生命周期和重复投递走显式分支

真实 runner 同时处理普通 turn 和 slash control，而 Claude 的 stream-json 在这两类交互中都可能省略生命周期。runner 每次开始交互时重置 translation state，并明确允许 standalone delta；纯函数 translator 的默认值仍是关闭，避免测试或其他调用者误把孤立 delta 当成合法流。

private resetTurnDiagnostics(): void {
  this.readerFailure = null
  this.stderrTail = ''
  this.translationState = createClaudeTranslationState({
    // Stream-json can omit lifecycle records for both turns and control commands.
    allowStandaloneDeltas: true,
  })
}

fallback group 一旦被带 ID 的 final envelope 接管，就从 fallbackGroup 槽位移走；下一条 lifecycle-free 消息会分配新组。反过来，只要本次交互已经见过生命周期，standalone 分支就永久关闭，孤立 delta 不能绕过已知协议。

function standaloneDeltaGroup(
  state: ClaudeTranslationState,
): ClaudeMessageGroup | undefined {
  if (!state.allowStandaloneDeltas || state.streamLifecycleSeen) {
    return undefined
  }
  if (!state.fallbackGroup ||
      !fallbackGroupAcceptsDeltas(state.fallbackGroup)) {
    state.fallbackGroup = createMessageGroup(state)
  }
  return state.fallbackGroup
}

4心智模型补丁

5新词表

6测试与风险地图

别被这些现象误导

• 公开事件中的 itemId 仍是 m1b0，不是 Claude UUID；原生 ID 只存在于 translator 的分组状态。
• message_stop 关闭 stream，不等于完成内容块；块要等 authoritative assistant envelope 才把 completed 置为真。
• 工具块不分配 message itemId；它们继续通过原生 tool_use.id 生成 tool.started。
• 每个 turn/control 都在 runner 中重置 translation state，因此 m1b0 会在下一次交互重新出现，这是 run-local 设计。

7覆盖声明

本报告以 GitHub PR #147 的 main...fix/claude-native-message-id 为边界，全量阅读 6 个变更文件的每个 diff hunk：runner.ts、translator.ts、三个 Claude 测试/夹具文件和 provider transcript fixture。核心 translator.ts 同时完整阅读了 base 与 head 版本；报告中的每段代码都从这些一手文件裁剪。

925 行新增中，527 行集中在 translator 单测，另外 58 行位于 runner/session/helper/transcript 测试；没有 rename、锁文件、生成物或纯机械搬运可略读。PR body、7 条 commit subject 和 Claude provider 文档只用于确认动机与现有边界，不作为代码结论的替代证据。

Buffin · PR #147 · 6 files · +925 / −59 · 本报告为一次性理解辅助，不维护、不作为真相源。