Provider 与流式协议

内容状态: 已校验
源码仓库: https://github.com/earendil-works/pi
源码版本: 8ce69e9d2b171d173fe4b6b2b6256f1f4411e69d
校验日期: 2026/10/1

一、本章只解决一条主线

阶段 2 已经看到 Agent Loop 消费模型流，阶段 5 已经解释统一 Context 怎样形成。本章补上二者之间的一段。以下按本地源码 8ce69e9d2 核对，代码块是保留关键调用的缩写，省略导入、类型细节和非主线分支，不是可独立运行的完整文件：

Agent Loop
  → ModelRuntime
  → Provider
  → API implementation
  → 厂商请求与流式响应
  → 统一 AssistantMessageEventStream
  → Tool 执行或本轮结束

选择 OpenAI Responses 追踪完整流程，再用 Anthropic Messages 对比协议差异。重点不是记厂商字段，而是理解三次归一化：

运行时归一化
  Provider、认证、模型和 API 分发

请求归一化
  Context、Tool、thinking、cache → 厂商 payload

响应归一化
  厂商事件 → 统一 AssistantMessageEvent

1.1 上游和下游边界

对外的 Models.streamSimple() 仍接受便捷 Context，但 Provider 和 API implementation 接收的已是 TranscriptContext。区别在于：system prompt 和工具声明进入消息历史。

// types.ts / utils/transcript.ts：省略品牌类型与字段细节。
interface Context {
  systemPrompt?: string;
  messages: Message[];
  tools?: Tool[];
}
// normalizeContext() 将上面两个便捷字段折入头部 SystemMessage。
type TranscriptContext = { messages: Message[] };
type Message = SystemMessage | UserMessage | AssistantMessage | ToolResultMessage;

// SystemMessage 除了 content，还可以携带：
// sections、toolsAdded、toolsRemoved。

后续 SystemMessage 可更新指令、命名 section 和工具集合。支持中途系统消息的协议保留更新位置；不支持的协议由 resolveTranscript() 重放系统状态，合并到头部。它不是简单丢弃后来的系统消息。

本章始终追踪同一个例子：用户要求“读取 auth.ts，解释登录逻辑”。请求历史和工具执行按下面的对象交接展开；这里的 read 声明只有模型可见 schema，真正的 execute() 仍由 Agent 持有。

SystemMessage（指令 + read 声明）+ UserMessage
  → Provider 请求
  → AssistantMessage（read 工具调用）
  → Agent 执行 read，追加 ToolResultMessage（auth.ts 内容）
  → 下一次 Provider 请求携带以上历史
  → AssistantMessage（解释登录逻辑）

Provider 输出的是统一事件：

start
text_start / text_delta / text_end
thinking_start / thinking_delta / thinking_end
toolcall_start / toolcall_delta / toolcall_end
done / error

agent-loop.ts 只消费这些事件：

start        → Agent message_start
内容块事件   → Agent message_update
done / error → Agent message_end

若最终消息包含 Tool Call：

AssistantMessage.content 中存在 toolCall
  → 排除 error / aborted；length 生成失败工具结果，不执行
  → 正常调用 executeToolCalls()
  → ToolResultMessage
  → 再次调用 Provider

是否继续还受 finishTurn、steering/follow-up 队列等影响；退出 Core Loop 后，AgentSession 再处理重试、压缩恢复与收尾。这里先聚焦一次模型请求。

二、Model、Provider、API 三层

概念	代码形态	职责
Model	`Model<TApi>` 数据对象	描述模型、能力、价格、窗口和协议
Provider	`Provider<TApi>` 运行时对象	管认证、模型目录和 stream 分发
API	`ProviderStreams`	管具体协议的请求序列化和流解析

2.1 Model：请求说明书

最关键的三个字段：

provider
  向谁认证，由哪个 Provider runtime 管理

api
  使用哪种线上协议

id
  请求哪个具体模型

例如：

provider = openai
api      = openai-responses
id       = gpt-5-mini

多个 Provider 可以复用同一种 OpenAI-compatible API，一个 Provider 也可以管理多种 API。因此完整路由键是：

(provider, api, model id)

用一段读取字段的示意代码串起来：

// 路由示意；实际认证与错误包装见第四节。
const provider = providers.get(model.provider);
const api = apiImplementations[model.api];
const params = { model: model.id /* 其余 payload 由 api 构造 */ };

Model 还携带：

reasoning / input
contextWindow / maxTokens
cost
thinkingLevelMap
compat

这些字段告诉适配器模型支持什么，而不是让上层按模型名猜能力。

2.2 Provider：认证和模型运行时

Provider 拥有：

auth
getModels()
refreshModels()?
stream()
streamSimple()

例如 providers/openai.ts 主要做组装：

OpenAI Provider
  → API key 或 OpenAI Provider 配置的 OAuth 认证
  → OPENAI_MODELS
  → openAIResponsesApi()

它不实现 OpenAI 消息格式。真正的协议逻辑在 API implementation 中。

2.3 API implementation：协议适配器

统一接口：

interface ProviderStreams {
  stream(model, context, options): AssistantMessageEventStream;
  streamSimple(model, context, options): AssistantMessageEventStream;
}

典型实现：

openai-responses.ts
anthropic-messages.ts
google-generative-ai.ts
bedrock-converse-stream.ts

调用路由：

ModelRuntime 按 model.provider 找 Provider
  → Provider 按 model.api 找 API implementation
  → API implementation 按 model.id 构造请求

2.4 新旧入口

当前 coding-agent 主路径是：

ModelRuntime implements Models
  → Provider
  → API implementation

compat.ts 中的全局 getModel()、stream() 和 API registry 是迁移期兼容入口，不是新架构的核心状态。

三、统一流契约

3.1 `streamSimple()` 和 `stream()`

Agent 使用 streamSimple() 表达统一意图：

reasoning = low / medium / high
maxTokens
temperature
signal
cacheRetention

streamSimple() 负责：

估算 Context token
  → 限制 maxTokens
  → 根据 Model 能力修正 reasoning
  → 翻译为协议选项
  → 调用 stream()

例如：

OpenAI
  reasoning = high
  → reasoningEffort = high

Anthropic adaptive thinking
  reasoning = high
  → effort = high

Anthropic 旧模型
  reasoning = high
  → thinkingBudgetTokens

对应 openai-responses.ts:streamSimple() 的缩写是：

getClientApiKey(model.provider, options?.apiKey, options?.headers);
const base = buildBaseOptions(model, context, options, options?.apiKey);
const level = options?.reasoning ? clampThinkingLevel(model, options.reasoning) : undefined;
return stream(model, context, {
  ...base,
  toolChoice: options?.toolChoice,
  reasoningEffort: level === "off" ? undefined : level,
});

统一选项不是原样透传：例如 off 不会作为 OpenAI 的 reasoning effort 发送。

stream() 则接受协议特有选项，例如 OpenAI 的 serviceTier 或 Anthropic 的 thinkingDisplay。

简化理解：

streamSimple()
  统一产品语义

stream()
  协议精细参数

complete() 和 completeSimple() 只是等待对应 stream 的 result()，并不是另一套非流式请求。

3.2 AssistantMessage

Provider 在流期间逐步构造：

interface AssistantMessage {
  role: "assistant";
  content: (TextContent | ThinkingContent | ToolCall)[];
  api: Api;
  provider: string;
  model: string;
  responseId?: string;
  usage: Usage;
  stopReason: "pending" | "stop" | "length" | "toolUse" | "error" | "aborted" | "deferred";
  rawStopReason?: string;
  deferred?: DeferredHandle;
  errorMessage?: string;
}

厂商的 completed、end_turn、max_tokens 等状态必须映射成统一 stopReason。 pending 是累计消息初始化时的未完成状态，不能作为 done.reason。deferred 则表示有延后取回的结果句柄；支持该能力的 Provider 可提供 fetchDeferred() / cancelDeferred()。本章追踪的普通 OpenAI Responses 流式请求不走这条延后路径，不能把所有 done 都解释成“正文已经生成完”。

3.3 事件顺序

成功流：

start
  → block_start
  → block_delta*
  → block_end
  → done

运行中失败：

start
  → 已产生的部分内容
  → error

认证或请求建立阶段失败：

error

这是实现应遵守的契约：每条流以唯一的 done 或 error 结束。失败时，已经开始的内容块不一定收到对应 *_end。直接调用 API implementation 的 streamSimple() 还可能在认证检查时同步抛错； coding-agent 主链由 ModelRuntime 的 lazyStream() 将 setup 异常包成 error stream。

事件中的 contentIndex 指向 AssistantMessage.content，因为回答可能交错包含：

thinking[0]
text[1]
toolCall[2]
text[3]

事件的 partial 是持续变化的同一个累计消息，不是历史快照。

3.4 实时消费与最终结果

event-stream.ts 同时支持：

for await (const event of stream) {
  // 实时更新 UI 和 Agent 状态
}

const finalMessage = await stream.result();

错误流的 result() 通常也会 resolve，只是最终消息的 stopReason 为 error 或 aborted。因此 UI、Session 和 retry 可以处理同一种最终状态。

四、OpenAI Responses 端到端主链

4.1 总时序

sequenceDiagram
    participant Loop as Agent Loop
    participant MR as ModelRuntime
    participant P as Provider
    participant API as OpenAI Responses
    participant HTTP as OpenAI API

    Loop->>MR: streamSimple(model, transcript, options)
    MR-->>Loop: 立即返回 outer stream
    MR->>MR: 认证、headers、baseUrl
    MR->>P: provider.streamSimple()
    P->>API: 按 model.api 分发
    API->>API: reasoning 翻译和消息转换
    API->>HTTP: responses.create(stream=true)
    HTTP-->>API: ResponseStreamEvent
    API-->>Loop: 统一内容块事件
    API-->>Loop: done 或 error

4.2 coding-agent 接入

sdk.ts 创建 Agent 时注入：

Agent.streamFn
  → 读取 timeout、retry、transport
  → ModelRuntime.streamSimple()

同一入口还连接四个 Extension hook：

before_provider_headers
before_provider_request
after_provider_response
provider_stream_event

因此 Extension 可以修改 headers 或 payload；provider_stream_event 观察的是适配器解析后的原始厂商事件，数据归适配器所有，handler 应只读使用。它和归一化后的 text_delta 不是同一层。

Agent.streamFn 的当前主体可缩写为：

streamFn: (model, context, options) => {
  const requestOptions = buildRequestOptions(model, options);
  if (options.sessionId === sessionManager.getSessionId()) {
    cacheWarmer.start({ model, context, options: requestOptions }, cacheContextIsCurrent(model));
  }
  return modelRuntime.streamSimple(model, context, requestOptions);
}

CacheWarmer 是否实际预热还受模式、能力等条件控制；这不是额外的用户 Turn。正文继续追踪 return modelRuntime.streamSimple(...) 产生的回答流。

4.3 ModelRuntime：异步认证，立即返回流

model-runtime.ts 执行：

prepareRequest()
  → 按 model.provider 找 Provider
  → 解析 credential
  → 合并认证、配置、model 和调用方 headers
  → 执行 header hook
  → 合并 env
  → 必要时覆盖 baseUrl
  → provider.streamSimple()

认证刷新和动态 module import 是异步的，但 Models / ProviderStreams 的流接口同步返回可订阅对象，所以使用 lazy.ts：

立即返回 outer AssistantMessageEventStream
  → 后台完成 setup
  → 转发 inner stream

model-runtime.ts 的实际衔接可缩写为：

const transcript = normalizeContext(context);
return lazyStream(model, async () => {
  const prepared = await this.prepareRequest(model, options);
  return prepared.provider.streamSimple(prepared.model, transcript, prepared.options);
});

Agent 允许注入的 StreamFn 也可以返回 Promise，不能把这里的同步接口约束泛化到所有自定义 streamFn。 setup 失败时，lazyStream() 直接产生 error。因此 Provider 不存在、认证缺失、OAuth refresh 失败和动态 import 失败都不会让 Agent Loop 使用另一套异常协议。

4.4 Provider：按 API 分发

createProvider() 可以持有一个 API implementation，也可以持有按 model.api 组织的 map。找不到对应 API 时返回 error stream。

这层只做运行时选择，不转换 transcript。将 models.ts:createProvider() 缩写后，就是：

// single 是单个 API implementation，byApi 是按 api 名组织的 map。
const apiFor = (model) => single ?? byApi?.[model.api];
const dispatch = (model, run) => {
  const streams = apiFor(model);
  if (!streams) return lazyStream(model, async () => { throw new Error("缺少 API 实现"); });
  return run(streams);
};
// Provider 的方法委托给选中的协议实现：
streamSimple: (model, context, options) =>
  dispatch(model, streams => streams.streamSimple(model, context, options));

4.5 API：先清洗历史，再做厂商序列化

openai-responses.ts 先创建空的统一 AssistantMessage，之后所有事件都增量修改它。

请求先经 resolveTranscript() 处理系统消息兼容性，再由 buildParams() 调用 convertResponsesMessages()。后者内部包含下面两步。

第一步：

transformMessages(context.messages, targetModel, normalizeToolCallId)

它处理跨模型 thinking signature、图片、孤立 Tool Call 和外来 ID。

第二步：

convertResponsesMessages()

它把统一内容变成 OpenAI Responses input：

统一内容	OpenAI Responses
system prompt	`developer` 或 `system`
user text / image	`input_text` / `input_image`
assistant text	output `message`
thinking	`reasoning` item
ToolCall	`function_call`
ToolResultMessage	`function_call_output`

前一步保证历史对目标模型合法，后一步保证其符合 OpenAI 请求结构。

4.6 Tool Call ID

OpenAI Responses 同时使用：

call_id
item id

Pi 将二者组合：

ToolCall.id = call_id|item_id

回放 Tool Result 时取 call_id，回放 Tool Call 时恢复二者。跨 Provider 历史会先清洗 ID，不能直接把另一家格式送给 OpenAI。

4.7 Tool 和最终 payload

工具 schema 从 SystemMessage 历史中恢复，已经不再读取 ToolResultMessage.addedToolNames。 utils/transcript.ts:resolveTranscriptTools() 决定声明放在哪里：

// 省略类型；hasNonAdditiveToolChanges 检查移除和同名再次声明。
const anchorsAdditions = supportsToolAdditions && !hasNonAdditiveToolChanges(messages);
return {
  requestTools: anchorsAdditions
    ? (getInitialSystemMessage(messages)?.toolsAdded ?? [])
    : getCurrentTools(messages),
  anchorsAdditions,
};

只有协议支持、且历史只有新增时，初始工具放顶层 tools，后续工具在原 SystemMessage 位置声明。 OpenAI 转换器优先用 additional_tools，否则用 tool_search_call / tool_search_output；若有移除、同名重声明或能力不支持，则顶层发送当前完整工具集合，不再按历史插入新增声明。

例如 read 后扩展开放 search：变化由后续 SystemMessage 的 toolsAdded 表达，和 read 的 ToolResult 是不同消息。新工具不是因为“执行过工具”才有资格声明。

请求发送部分的缩写如下（省略 compat、cache、reasoning、工具语法等分支）：

let params = {
  model: model.id,
  input: convertResponsesMessages(model, context, /* 转换选项省略 */),
  tools: convertResponsesTools(transcriptTools.requestTools, /* 选项省略 */),
  stream: true,
  store: false,
};
const replacement = await options?.onPayload?.(params, model);
if (replacement !== undefined) params = replacement;
const { data: openaiStream, response } = await retryProviderRequest(
  () => client.responses.create(params, requestOptions).withResponse(),
  /* 重试次数、等待上限和 signal 省略 */,
);
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
stream.push({ type: "start", partial: output });
await processResponsesStream(openaiStream, output, stream, model, /* 选项省略 */);

实际代码在工具非空时才设置 tools，并按能力添加 max_output_tokens、reasoning、cache 和 service tier。 onResponse 发生在拿到 HTTP 响应后，start 发生在解析内容流前；两者都不是“回答已完成”。

五、厂商响应怎样变成统一事件

核心实现是 openai-responses-shared.ts 的 processResponsesStream()。

5.1 内容块映射

OpenAI 使用 output_index，Pi 使用统一 contentIndex，所以维护：

Map<output_index, unified content block>

Text：

output_item.added(message)
  → text_start
output_text.delta
  → text_delta
output_item.done(message)
  → text_end

Thinking：

output_item.added(reasoning)
  → thinking_start
reasoning_summary_text.delta / reasoning_text.delta
  → thinking_delta
output_item.done(reasoning)
  → 保存 thinkingSignature
  → thinking_end

thinkingSignature 是下一轮回放所需的 opaque Provider 状态，不只是 UI 文本。

Tool Call：

output_item.added(function_call)
  → toolcall_start
arguments.delta
  → partialJson
  → 容错解析当前 arguments
  → toolcall_delta
output_item.done(function_call)
  → 最终 arguments
  → 删除 partialJson
  → toolcall_end

partialJson 只属于解析过程，不能进入最终 AssistantMessage 或 Session。适配器还支持 custom tool 输入；上面为普通 JSON function call 主线，不是全部 Tool 协议分支。

processResponsesStream() 的职责可缩写为下面的分发骨架（分支体合并为注释）：

const outputSlots = new Map(); // output_index → { block, contentIndex, type }
for await (const event of openaiStream) {
  await options?.onProviderStreamEvent?.(event, model);
  if (event.type === "response.output_item.added") {
    createSlot(event.output_index, event.item); // 创建 block 并发出 *_start
  } else if (event.type === "response.output_text.delta") {
    // 找到 slot，累加 block.text，发出 text_delta，partial 指向同一 output
  } else if (event.type === "response.output_item.done") {
    // 补齐最终内容 / 签名，清理临时字段，发出 *_end
  } else if (event.type === "response.completed" || event.type === "response.incomplete") {
    finalizeResponse(event.response);
  }
  // 省略 reasoning、arguments、失败等分支。
}

因此 toolcall_delta 只表示参数还在拼装。例子中的 read(auth.ts) 要等最终 AssistantMessage 返回到 Agent 后才执行，不会收到一截 JSON 就读取文件。

5.2 终端事件

只有厂商终端事件才证明响应语义完整：

response.completed  → stop
response.incomplete + max_output_tokens → length
response.incomplete + 其他/缺失原因      → error
response.failed     → error

若消息存在 Tool Call，正常 stop 改为 toolUse。

终端事件还结算：

responseId
input / output / cache tokens
reasoning tokens
cost

如果网络流结束却没有终端事件，Pi 生成 error。即使收到终端事件，toolUse 消息里仍有未结束 ToolCall 的解析缓存也会报错（见章末更新）。物理 EOF 和 response.completed 都不能替代工具参数完整性检查。

5.3 回到 Agent Loop

Provider 发出 done 或 error 后：

stream.result()
  → final AssistantMessage
  → 替换 Context 中的 partial message
  → Agent message_end

Core Loop 实际从最终 message.content 提取 Tool Call；error / aborted 提前退出，length 中的调用生成失败工具结果而不执行。正常的 read 调用执行后追加 ToolResultMessage，下一轮请求才把文件内容序列化为 function_call_output。toolUse 是归一化状态，不是唯一的执行条件。

六、Anthropic 与 OpenAI 的差异

Anthropic 的外层结构不变：

streamSimple()
  → stream() / resolveTranscript()
  → buildParams() 内部 transformMessages()
  → HTTP/SSE
  → 统一事件

差异只留在适配器内部。

6.1 消息和 Tool Call

统一内容	OpenAI Responses	Anthropic Messages
初始 system prompt	developer/system input	顶层 `system`；后续更新按 compat 保留或合并
Tool Call	`function_call`	`tool_use`
Tool Result	`function_call_output`	user content 中的 `tool_result`
Tool Call ID	`call_id\|item_id`	`tool_use.id`
thinking replay	reasoning item JSON	thinking + signature

两家最终都生成统一：

{
  type: "toolCall",
  id: "...",
  name: "edit",
  arguments: { path: "README.md" }
}

因此 Agent Loop 不关心原始协议。Tool Result 在下一轮由目标 Provider 重新序列化。

Anthropic 参数流：

content_block_start(tool_use)
  → toolcall_start
input_json_delta
  → toolcall_delta
content_block_stop
  → toolcall_end

Anthropic 的参数拼装对应下面的代码缩写：

// anthropic-messages.ts：input_json_delta 分支，省略 block 查找。
if (block && block.type === "toolCall") {
  block.partialJson += event.delta.partial_json;
  block.arguments = parseStreamingJson(block.partialJson);
  stream.push({
    type: "toolcall_delta", contentIndex: index,
    delta: event.delta.partial_json, partial: output,
  });
}

这里和 OpenAI 的字段名不同，但交给 Agent 的都是 ToolCall。动态工具则有自己的兼容分支：同时支持中途 system/tool changes、初始有活动工具且没有同名定义变化时，可用 tool_addition / tool_removal 及延迟加载声明；否则发送重放后的当前工具集合。不要把 OpenAI 的“只允许新增”规则套到 Anthropic。

6.2 Thinking

Anthropic 支持：

thinking
  可见文本 + signature

redacted_thinking
  占位文本 + opaque signature + redacted=true

有 signature 时可以继续作为 thinking 回放；缺少 signature 时通常降级为普通 text。说明 thinking 文本和 Provider 连续性凭证是两种状态。

6.3 Stop reason、cache 和 usage

Stop reason：

end_turn   → stop
max_tokens → length
tool_use   → toolUse
refusal    → error，并保留 explanation

Cache：

OpenAI
  prompt_cache_key / prompt_cache_retention

Anthropic
  cache_control 可标记顶层 system、工具列表最后一个工具
  对话缓存断点：最后一条协议消息为 user 或 system 时，标记其末尾支持的 block

上述标记还受 cacheRetention 与 compat 支持控制。统一选项表达缓存意图，具体字段由适配器翻译；最后消息为 assistant 时不会套用上述对话末尾标记逻辑。

Usage：

OpenAI
  主要在终端 response

Anthropic
  message_start 提供 input/cache
  message_delta 补 final output

Provider 最终都构造统一 Usage 并计算 cost。

6.4 Compat flags

兼容服务可能不支持 developer role、strict tools、long cache、reasoning effort 或空 thinking signature。这些能力放在 model.compat，由适配器读取。

因此上层不需要散落：

if provider === ...
if model name startsWith ...

七、错误、取消、重试与跨 Provider

7.1 为什么错误编码进 stream

统一错误终点：

error {
  reason: error | aborted
  error: AssistantMessage
}

好处：

保留 partial content
UI 正常收到 message_end
Session 可以持久化失败
AgentSession 可以统一恢复

两类错误：

setup 错误
  Provider、认证、动态 import
  → lazyStream → error

stream 错误
  HTTP、SSE、JSON、early EOF
  → API catch → error

7.2 Abort

AbortSignal 贯穿：

Agent
  → Agent Loop
  → ModelRuntime
  → HTTP request
  → retry wait
  → stream reader

取消后：

stopReason = aborted
已收到内容保留
partialJson 等临时字段删除

Session 可以保留这条失败记录。若下一轮输入仍包含它，transformMessages() 会过滤 error / aborted Assistant；AgentSession 的自动恢复还可用持久化 context_edit 将失败尝试排除出后续模型上下文。保留失败事实和再次发送失败内容是两件事。

7.3 重试与压缩恢复的边界

provider-retry.ts 只处理请求建立阶段的短暂错误，例如 429、5xx 和网络失败。

它不会在已经输出半截 token 后重放整条流，否则会产生重复内容。

机制	所在层	负责什么
Provider retry	HTTP 请求建立	网络、429、5xx
AgentSession retry	产品编排	一次可见运行失败
摘要请求 retry	Compaction 内部	摘要模型请求遇到可重试错误，重试生成摘要
overflow recovery	AgentSession 的 Context 管理	排除失败尝试、压缩、重建上下文，再请求模型；有恢复窗口限制

下面是 openai-responses.ts 的异常收尾缩写，说明“只重试请求建立”和“流中断后交出失败消息”的分界：

try {
  // retryProviderRequest(...) 只包 HTTP 建立。
  await processResponsesStream(openaiStream, output, stream, model, options);
  if (signal?.aborted) throw new Error("Request was aborted");
  // 省略 pending/error 检查。
  stream.push({ type: "done", reason: output.stopReason, message: output });
} catch (error) {
  for (const block of output.content) {
    delete block.index;
    delete block.partialJson;
    delete block.customInput;
  }
  output.stopReason = signal?.aborted ? "aborted" : "error";
  // 省略 errorMessage 格式化；不清空 output.content。
  stream.push({ type: "error", reason: output.stopReason, error: output });
}
stream.end(); // 原码分别在成功与 catch 分支调用。

7.4 跨 Provider 会话

Session 保存统一 Message，而不是厂商 payload：

SessionEntry[]
  → AgentMessage[]
  → Message[]
  → transformMessages(targetModel)
  → 目标 Provider serializer

例如 OpenAI 历史切到 Anthropic：

OpenAI reasoning signature
  → 不能当作 Anthropic signature

call_id|item_id
  → Anthropic ID normalizer

统一 ToolResultMessage（OpenAI 请求时转换为 function_call_output）
  → Anthropic 请求时重新转换为 user/tool_result

跨 Provider 能力来自：

统一持久化格式
  + target-aware transformMessages()
  + target-specific serializer

并不是各家协议天然兼容。

八、验收、新增 Provider 与源码回查

8.1 相关测试

测试	验证重点
`openai-responses-terminal-event.test.ts`	early EOF 和终端状态
`openai-responses-partial-json-cleanup.test.ts`	临时 Tool JSON 不进入 Session
`anthropic-sse-parsing.test.ts`	SSE 容错、refusal、usage
`abort.test.ts`	立即取消、中途取消、取消后继续
`cross-provider-handoff.test.ts`	thinking 和 Tool Call 跨 Provider
`faux-provider.test.ts`	无真实 API 的统一流契约

本次核对了源码与相关测试用例，未运行 Pi 完整测试套件或真实厂商请求；上表列的是可回查的行为证据，不能视为本次 API 联调结果。

8.2 新增 Provider

若能复用现有 API：

定义 Provider factory
定义 auth 和 model catalog
指定已有 API implementation
用 model.compat 描述协议方言
注册 Provider

若是新协议，必须实现：

Context / Tool / reasoning → payload
厂商事件 → 统一内容块事件
usage / stopReason / cost
abort / error / early EOF
Tool Call ID 和 thinking signature 回放

最小消费端检查可以这样写（示意断言，不代表本次执行了真实 API 测试）：

const events = [];
for await (const event of stream) events.push(event);
const terminal = events.filter(e => e.type === "done" || e.type === "error");
assert.equal(terminal.length, 1);
const final = await stream.result();
assert.equal(final.stopReason, terminal[0].reason);
assert.ok(final.content.every(block => !("partialJson" in block)));

实现必须保证终端事件到达。仅调用没有结果的 end() 并不能保证 result() resolve；消费者也需要超时/取消保护来发现有缺陷的实现，不能靠上述 await 自己检测“永远不结束”。

核心不变量：

正常完成的内容块 start → delta* → end；失败可中断，redacted thinking 可无 delta
只有一个 done 或 error
result() 一定得到最终 AssistantMessage
临时解析状态不进入最终消息
物理 EOF 不替代厂商终端事件

8.3 30 秒总结

Pi 用 Model + Provider + API implementation 三层统一模型厂商。Model 描述能力和路由， Provider 管认证与模型目录，API implementation 把统一 Context 转成厂商请求，再把厂商流转成统一 text、thinking、toolcall 和 done/error 事件。Session 只保存统一消息，所以切换 Provider 时由目标转换器重新处理 Tool Call ID、thinking signature 和 Tool Result。

8.4 验收问题

能画出 Agent → ModelRuntime → Provider → API → EventStream；
能区分 Model、Provider 和 API；
能复述 OpenAI Responses 的请求与响应转换；
能解释 Tool Call ID、thinking signature 和终端事件；
能对比 OpenAI 与 Anthropic 的 Tool Result；
能区分 HTTP 重试、运行重试、摘要请求重试和 overflow recovery；
能列出新增 Provider 的实现清单。

8.5 源码回查顺序

按调用链阅读：

types.ts：统一类型和事件；
sdk.ts：Agent 的 streamFn；
model-runtime.ts：认证与 Provider 分发；
models.ts：Provider 和 createProvider()；
openai-responses.ts：请求入口；
openai-responses-shared.ts：消息和流转换；
anthropic-messages.ts：对比实现；
lazy.ts：setup error；
provider-retry.ts：请求重试；
utils/transcript.ts：系统状态重放与工具声明分配；
本章测试表中的协议测试。

8.6 与阶段 9 的连接

阶段 8 最终产出：

AssistantMessageEventStream
  → Agent message_start / message_update / message_end
  → AgentSession event

阶段 9 将继续追踪同一组事件怎样进入 TUI、Print mode、RPC 和 SDK。Provider 层消除了模型厂商差异；下一阶段要理解产品表面如何消除宿主界面差异。

新增：虚拟模型先路由，物理模型再进入 Provider

ModelRuntime 现在另外保存 virtual model 定义，虚拟目录项 api = pi-virtual 本身不能发送给 Provider。会话请求在 AgentSession.prepareRequest 中路由；直接的 streamSimple 调用在 ModelRuntime 中兜底路由。

// ModelRuntime.streamSimple 的虚拟模型分支缩写。
if (isVirtualModel(model)) {
  return lazyStream(model, async () => {
    const route = await this.resolveModel(model, transcript.messages, {
      reason: "direct", thinkingLevel: options?.reasoning ?? "off", signal: options?.signal,
    });
    // 原码按物理模型限制 maxTokens，并在跨 provider 时丢弃 apiKey/headers/env。
    return this.streamSimple(route.model, context, routedOptions);
  });
}
// 然后才是 prepareRequest 认证与 provider.streamSimple。

理由有 user/continuation/retry/direct。路由状态随会话分支保存；direct 请求不保存 router state。 previous 是最近成功实际响应，failed 是已从上下文排除的失败尝试，二者不能互换。跨 Provider 路由不能沿用原厂商凭据。模型能力与预算也必须以本次真实模型为准。

OpenAI Responses 新增了一道完整性检查：即使已经收到 response.completed，若最终 toolUse 消息的 ToolCall 仍留有 partialJson/customInput，说明缺少 output_item.done，适配器会报错而不是交给 Agent 执行。

// processResponsesStream 收流后的检查缩写。
if (output.stopReason === "toolUse") {
  for (const block of output.content) {
    if (block.type === "toolCall" &&
        (block.partialJson !== undefined || block.customInput !== undefined)) {
      throw new Error("unfinished tool call");
    }
  }
}

此外，OpenAI Provider 已不只支持环境 API key，还提供登录认证分支。对于直连 OpenAI 的 ChatGPT 登录 token，buildParams 省略不支持的 max_output_tokens、temperature、prompt_cache_retention/options；不能把前文 payload 示例理解为每种认证方式都发送完全相同字段。Anthropic 还增加 workload identity federation：仅当目标为 anthropic 且没有已解析 key/auth header 时，才使用配置的 federation 环境变量；已有 header 认证不会再被 SDK 默认凭据链覆盖。严格 schema 也有能力检查与兼容转换，参数是否发送由实际身份和能力共同决定。

源码回查：core/virtual-models.ts、core/model-runtime.ts:resolveModel/streamSimple、 core/agent-session.ts:_installAgentRequestProjection、ai/src/api/openai-responses-shared.ts、 ai/src/providers/openai.ts、ai/src/api/openai-responses.ts:buildParams。