refactor: 完善跨格式 normalizer 转换精度，统一 CLI 流式 buffer flush 逻辑

- OpenAI normalizer: 修正 file/image 内容块的标准格式解析与输出，
  assistant 有 tool_calls 时 content 输出 null，非流式 tool_calls
  不再包含 index 字段，保留 system_fingerprint/service_tier roundtrip
- OpenAI CLI normalizer: 支持 reasoning/ThinkingConfig 双向转换，
  补全 parallel_tool_calls/required tool_choice，input_file 解析，
  function_call_output.output 强制字符串化，流式事件补全 item_id/
  output_index/content_index 字段
- Gemini normalizer: 扩展 finishReason 映射，自动检测 tool_use
  stop_reason，保留 safetySettings/cachedContent/generationConfig
  额外字段 roundtrip，error_from_internal 使用精确 HTTP status code
- Claude normalizer: 修正 tool_choice type="tool" 输出，扩展
  extra 提取白名单
- internal.py: 为所有 dataclass 补充跨格式映射文档和修改须知
- stream_bridge: aggregator 新增 open_count/final_count 诊断属性，
  build() 时 flush 未关闭的 open blocks
- CLI handler: 提取 _flush_buffer_with_conversion 统一 prefetch/
  stream 两条路径的 buffer + SSE parser flush 逻辑
- upstream_stream_bridge: 新增事件类型计数和聚合器状态诊断日志
- 新增 fixtures 和测试: roundtrip/to_internal/cross_format/error/
  stream 等多维度转换测试

src/api/handlers/base/cli_event_mixin.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import codecs
 import json
+from collections.abc import Iterator
 from typing import TYPE_CHECKING, Any
 
 from src.api.handlers.base.parsers import get_parser_for_format
@@ -10,6 +12,7 @@ from src.api.handlers.base.stream_context import StreamContext
 from src.api.handlers.base.utils import get_format_converter_registry
 from src.core.logger import logger
 from src.services.provider.behavior import get_provider_behavior
+from src.utils.sse_parser import SSEEventParser
 
 from .cli_sse_helpers import (
     _format_converted_events_to_sse,
@@ -281,6 +284,68 @@ class CliEventMixin:
             if any([new_input, new_output, new_cached, new_cache_creation]):
                 ctx.final_usage = usage
 
+    def _flush_buffer_with_conversion(
+        self: CliHandlerProtocol,
+        ctx: StreamContext,
+        buffer: bytes,
+        decoder: codecs.IncrementalDecoder,
+        sse_parser: SSEEventParser,
+        needs_conversion: bool,
+    ) -> Iterator[bytes]:
+        """flush 字节 buffer 残余数据 + SSE parser 内部缓冲区，并做格式转换。
+
+        正常流结束时调用（区别于异常路径的 _flush_remaining_sse_data）。
+        当 needs_conversion=True 时 yield 转换后的 SSE 行；
+        当 needs_conversion=False 时 yield 原始行透传。
+        """
+        # 1) flush 字节 buffer 中的残余数据（最后一个 chunk 可能不以换行结尾）
+        if buffer:
+            try:
+                remaining = decoder.decode(buffer, True)
+            except Exception:
+                remaining = ""
+            for tail_line in remaining.split("\n"):
+                stripped = tail_line.rstrip("\r")
+                tail_events = sse_parser.feed_line(stripped)
+                if stripped == "":
+                    for event in tail_events:
+                        self._handle_sse_event(
+                            ctx,
+                            event.get("event"),
+                            event.get("data") or "",
+                            record_chunk=not needs_conversion,
+                        )
+                    continue
+                if needs_conversion and stripped:
+                    converted_lines, converted_events = self._convert_sse_line(
+                        ctx, stripped, tail_events
+                    )
+                    self._record_converted_chunks(ctx, converted_events)
+                    for converted_line in converted_lines:
+                        if converted_line:
+                            yield (converted_line + "\n").encode("utf-8")
+                elif stripped:
+                    # 非 conversion 模式：透传原始行
+                    yield (stripped + "\n").encode("utf-8")
+
+        # 2) flush SSE parser 内部缓冲区中的残余事件
+        for event in sse_parser.flush():
+            self._handle_sse_event(
+                ctx,
+                event.get("event"),
+                event.get("data") or "",
+                record_chunk=not needs_conversion,
+            )
+            if needs_conversion:
+                data_str = event.get("data") or ""
+                if data_str and data_str != "[DONE]":
+                    flush_line = f"data: {data_str}"
+                    converted_lines, converted_events = self._convert_sse_line(ctx, flush_line, [])
+                    self._record_converted_chunks(ctx, converted_events)
+                    for converted_line in converted_lines:
+                        if converted_line:
+                            yield (converted_line + "\n").encode("utf-8")
+
     def _finalize_stream_metadata(self, ctx: StreamContext) -> None:
         """
         在记录统计前从 parsed_chunks 中提取额外的元数据 - 子类可覆盖

src/api/handlers/base/cli_prefetch_mixin.py

@@ -515,15 +515,11 @@ class CliPrefetchMixin:
                     if ctx.data_count > 0:
                         last_data_time = time.time()
 
-            # 处理剩余事件
-            flushed_events = sse_parser.flush()
-            for event in flushed_events:
-                self._handle_sse_event(
-                    ctx,
-                    event.get("event"),
-                    event.get("data") or "",
-                    record_chunk=not needs_conversion,
-                )
+            # flush 字节 buffer 残余数据 + SSE parser 内部缓冲区
+            for chunk in self._flush_buffer_with_conversion(
+                ctx, buffer, decoder, sse_parser, needs_conversion
+            ):
+                yield chunk
 
             # 检查是否收到数据
             if ctx.data_count == 0:

src/api/handlers/base/cli_protocol.py

@@ -203,6 +203,15 @@ class CliHandlerProtocol(Protocol):
         converted_events: list[dict[str, Any]],
     ) -> None: ...
 
+    def _flush_buffer_with_conversion(
+        self,
+        ctx: StreamContext,
+        buffer: bytes,
+        decoder: Any,
+        sse_parser: Any,
+        needs_conversion: bool,
+    ) -> Any: ...  # Iterator[bytes]
+
     def _finalize_stream_metadata(
         self,
         ctx: StreamContext,

src/api/handlers/base/cli_stream_mixin.py

@@ -909,14 +909,11 @@ class CliStreamMixin:
                     if ctx.data_count > 0:
                         last_data_time = time.time()
 
-            # 处理剩余事件
-            for event in sse_parser.flush():
-                self._handle_sse_event(
-                    ctx,
-                    event.get("event"),
-                    event.get("data") or "",
-                    record_chunk=not needs_conversion,
-                )
+            # flush 字节 buffer 残余数据 + SSE parser 内部缓冲区
+            for chunk in self._flush_buffer_with_conversion(
+                ctx, buffer, decoder, sse_parser, needs_conversion
+            ):
+                yield chunk
 
             # 检查是否收到数据
             if ctx.data_count == 0:

src/api/handlers/base/upstream_stream_bridge.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import codecs
 import json
+from collections import Counter
 from collections.abc import AsyncIterator
 from typing import Any
 
@@ -148,7 +149,11 @@ async def aggregate_upstream_stream_to_internal_response(
     buffer = b""
     decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
 
+    _event_type_counts: Counter[str] = Counter()
+    _total_events = 0
+
     def _feed_line(normalized_line: str) -> None:
+        nonlocal _total_events
         data_obj, st = parse_provider_stream_line_to_json(normalized_line, provider_api_format)
         if st != "ok" or data_obj is None:
             return
@@ -171,6 +176,9 @@ async def aggregate_upstream_stream_to_internal_response(
                 error_status=parsed.error_type,
             )
 
+        etype = str(data_obj.get("type") or "")
+        _event_type_counts[etype] += 1
+        _total_events += 1
         internal_events = src_norm.stream_chunk_to_internal(data_obj, state)
         aggregator.feed(internal_events)
 
@@ -193,7 +201,35 @@ async def aggregate_upstream_stream_to_internal_response(
         if normalized_tail:
             _feed_line(normalized_tail)
 
-    return aggregator.build()
+    # 诊断日志：在 build() 之前记录聚合器状态
+    open_count = aggregator.open_count
+    final_count = aggregator.final_count
+    if not final_count and not open_count:
+        logger.warning(
+            "[{}] aggregate_upstream_stream: 聚合器无内容, "
+            "open={}, final={}, usage={}, stop_reason={}, "
+            "event_types={}, total_events={}",
+            request_id,
+            open_count,
+            final_count,
+            aggregator.usage,
+            aggregator.stop_reason,
+            dict(_event_type_counts),
+            _total_events,
+        )
+    elif not final_count and open_count:
+        logger.warning(
+            "[{}] aggregate_upstream_stream: final 为空但 open 有内容, "
+            "open={}, final={}, event_types={}, total_events={}",
+            request_id,
+            open_count,
+            final_count,
+            dict(_event_type_counts),
+            _total_events,
+        )
+
+    result = aggregator.build()
+    return result
 
 
 __all__ = [

src/core/api_format/conversion/internal.py

@@ -1,13 +1,19 @@
 """
 格式转换内部表示（Internal / Canonical Format）
 
-该模块定义 Hub-and-Spoke 架构的“中间表示法”，用于把不同 Provider 的请求/响应/流式事件
+该模块定义 Hub-and-Spoke 架构的"中间表示法"，用于把不同 Provider 的请求/响应/流式事件
 统一映射到稳定的内部结构，再转换为目标格式。
 
 设计原则：
 - 类型安全：尽量用 dataclass + Enum 表达语义，便于 IDE/静态检查
 - 可扩展：未知/不可逆字段写入 extra/raw，避免静默丢失
 - 兼容优先：UnknownBlock 在内部保留，但默认在输出阶段丢弃（可观测、可随时调整策略）
+
+字段修改须知：
+- 本文件是所有 normalizer 的共享契约，修改字段语义会同时影响所有格式的输入输出
+- 每个字段的注释标注了各格式的映射关系（OpenAI/Claude/Gemini）
+- 修改前请检查 tests/core/api_format/conversion/ 下的 roundtrip + schema 测试
+- 新增字段应标注 "可选" 并给默认值，避免破坏现有 normalizer
 """
 
 from dataclasses import dataclass, field
@@ -62,7 +68,13 @@ class ErrorType(str, Enum):
 
 @dataclass
 class TextBlock:
-    """文本内容块"""
+    """文本内容块
+
+    Format mapping:
+      OpenAI: message.content (string) / content[].type="text"
+      Claude: content[].type="text"
+      Gemini: parts[].text
+    """
 
     type: ContentType = field(default=ContentType.TEXT, init=False)
     text: str = ""
@@ -71,30 +83,52 @@ class TextBlock:
 
 @dataclass
 class ThinkingBlock:
-    """思考过程内容块（对齐 Gemini thought:true / Claude thinking / OpenAI reasoning_content）"""
+    """思考过程内容块
+
+    Format mapping:
+      OpenAI: message.reasoning_content / delta.reasoning_content
+      Claude: content[].type="thinking" (thinking + signature)
+      Gemini: parts[].thought=true (text + thoughtSignature)
+    """
 
     type: ContentType = field(default=ContentType.THINKING, init=False)
     thinking: str = ""
-    signature: str | None = None  # Gemini thoughtSignature / Claude signature
+    # Claude signature / Gemini thoughtSignature; OpenAI 无对应字段
+    signature: str | None = None
     extra: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
 class ImageBlock:
-    """图片内容块"""
+    """图片内容块
+
+    Format mapping:
+      OpenAI: content[].type="image_url" -> image_url.url (URL or data:mime;base64,...)
+      Claude: content[].type="image" -> source.type="base64" | source.type="url"
+      Gemini: parts[].inlineData (base64) / parts[].fileData (URI)
+    """
 
     type: ContentType = field(default=ContentType.IMAGE, init=False)
-    # base64 编码的图片数据（二选一）
-    data: str | None = None
-    media_type: str | None = None
-    # 或者 URL 引用
-    url: str | None = None
+    data: str | None = None  # base64 encoded image data (mutually exclusive with url)
+    media_type: str | None = None  # MIME type, e.g. "image/png"
+    url: str | None = None  # URL reference (mutually exclusive with data)
     extra: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
 class ToolUseBlock:
-    """工具调用内容块"""
+    """工具调用内容块
+
+    Format mapping:
+      OpenAI: message.tool_calls[].id / .function.name / .function.arguments(JSON str)
+      Claude: content[].type="tool_use" -> id / name / input(dict)
+      Gemini: parts[].functionCall -> name / args(dict); id 由 normalizer 生成
+
+    Contract:
+      tool_id:    roundtrip 保留; OpenAI/Claude 原生提供, Gemini 由 normalizer 合成
+      tool_name:  必须非空
+      tool_input: 已解析的 dict (非 JSON 字符串)
+    """
 
     type: ContentType = field(default=ContentType.TOOL_USE, init=False)
     tool_id: str = ""
@@ -105,12 +139,23 @@ class ToolUseBlock:
 
 @dataclass
 class ToolResultBlock:
-    """工具结果内容块"""
+    """工具结果内容块
+
+    Format mapping:
+      OpenAI: role="tool" message -> tool_call_id + content(string)
+      Claude: content[].type="tool_result" -> tool_use_id + content
+      Gemini: parts[].functionResponse -> name + response(dict)
+
+    Contract:
+      tool_use_id:  关联 ToolUseBlock.tool_id; OpenAI/Claude 必须非空
+      tool_name:    Gemini functionResponse.name 需要; OpenAI/Claude 可为 None
+      output:       结构化输出 (dict/list); 与 content_text 二选一
+      content_text: 纯文本输出; 与 output 二选一
+    """
 
     type: ContentType = field(default=ContentType.TOOL_RESULT, init=False)
-    tool_use_id: str = ""  # 对应的 ToolUseBlock.tool_id（用于 Claude/Antigravity id 字段）
-    tool_name: str | None = None  # 工具名称（用于 Gemini function_response.name 字段）
-    # 工具输出可能是纯文本，也可能是结构化 JSON（Gemini functionResponse 等）
+    tool_use_id: str = ""
+    tool_name: str | None = None
     output: Any = None
     content_text: str | None = None
     is_error: bool = False
@@ -119,11 +164,17 @@ class ToolResultBlock:
 
 @dataclass
 class FileBlock:
-    """文件内容块（PDF、文档等）"""
+    """文件内容块（PDF、文档等）
+
+    Format mapping:
+      OpenAI: content[].type="file" -> file.file_data(data URL) / file.file_id
+      Claude: content[].type="document" -> source.type="base64" / source.type="url"
+      Gemini: parts[].fileData -> fileUri + mimeType
+    """
 
     type: ContentType = field(default=ContentType.FILE, init=False)
-    data: str | None = None  # base64 编码
-    media_type: str | None = None
+    data: str | None = None  # base64 encoded file data
+    media_type: str | None = None  # MIME type
     file_id: str | None = None  # OpenAI file reference
     file_url: str | None = None  # Gemini fileData URI
     filename: str | None = None
@@ -132,12 +183,18 @@ class FileBlock:
 
 @dataclass
 class AudioBlock:
-    """音频内容块"""
+    """音频内容块
+
+    Format mapping:
+      OpenAI: content[].type="input_audio" -> input_audio.data + input_audio.format
+      Claude: content[].type="audio" (planned)
+      Gemini: parts[].inlineData (audio MIME)
+    """
 
     type: ContentType = field(default=ContentType.AUDIO, init=False)
-    data: str | None = None  # base64 编码
-    media_type: str | None = None  # 完整 MIME（如 audio/mp3）
-    format: str | None = None  # 简短格式名（如 mp3, wav）
+    data: str | None = None  # base64 encoded audio data
+    media_type: str | None = None  # full MIME (e.g. audio/mp3)
+    format: str | None = None  # short format name (e.g. mp3, wav)
     extra: dict[str, Any] = field(default_factory=dict)
 
 
@@ -200,19 +257,29 @@ class ToolChoice:
 
 @dataclass
 class InstructionSegment:
-    """系统/开发者指令段（用于保留 OpenAI system/developer 结构与顺序）"""
+    """系统/开发者指令段
 
-    role: Role  # 仅允许 Role.SYSTEM / Role.DEVELOPER
+    OpenAI 区分 system/developer 两种 role, Claude/Gemini 只有 system string.
+    instructions 列表保留 OpenAI 的 role 语义和顺序, system 字段是 join 后的纯文本兜底.
+    """
+
+    role: Role  # Role.SYSTEM / Role.DEVELOPER only
     text: str = ""
     extra: dict[str, Any] = field(default_factory=dict)
 
 
 @dataclass
 class ThinkingConfig:
-    """统一的思考/推理配置（对齐 Claude thinking / Gemini thinkingConfig / OpenAI reasoning_effort）"""
+    """统一的思考/推理配置
+
+    Format mapping:
+      OpenAI: reasoning_effort ("low"/"medium"/"high") -> budget_tokens via lookup table
+      Claude: thinking.type="enabled" + thinking.budget_tokens
+      Gemini: generationConfig.thinkingConfig.thinkingBudget
+    """
 
     enabled: bool = False
-    budget_tokens: int | None = None  # None = provider 默认
+    budget_tokens: int | None = None  # None = provider default
     extra: dict[str, Any] = field(default_factory=dict)
 
 
@@ -227,7 +294,16 @@ class ResponseFormatConfig:
 
 @dataclass
 class InternalRequest:
-    """统一的请求表示"""
+    """统一的请求表示
+
+    Format mapping (key fields):
+      model:        OpenAI/Claude body.model; Gemini URL path param
+      instructions: OpenAI system/developer messages; Claude/Gemini -> join to system string
+      system:       instructions join fallback; Claude system param; Gemini systemInstruction
+      max_tokens:   OpenAI max_tokens/max_completion_tokens; Claude max_tokens; Gemini maxOutputTokens
+      tools:        OpenAI tools[].function; Claude tools[]; Gemini tools[].functionDeclarations
+      tool_choice:  OpenAI tool_choice; Claude tool_choice; Gemini toolConfig.functionCallingConfig
+    """
 
     model: str
     messages: list[InternalMessage]
@@ -297,7 +373,15 @@ class UsageInfo:
 
 @dataclass
 class InternalResponse:
-    """统一的响应表示"""
+    """统一的响应表示
+
+    Format mapping:
+      id:          OpenAI id; Claude id; Gemini (none, synthesized)
+      model:       OpenAI model; Claude model; Gemini model (from metadata)
+      content:     OpenAI choices[0].message; Claude content[]; Gemini candidates[0].content.parts
+      stop_reason: OpenAI finish_reason; Claude stop_reason; Gemini finishReason
+      usage:       OpenAI usage; Claude usage; Gemini usageMetadata
+    """
 
     id: str
     model: str

src/core/api_format/conversion/normalizers/claude.py

@@ -171,7 +171,25 @@ class ClaudeNormalizer(FormatNormalizer):
             tools=tools,
             tool_choice=tool_choice,
             thinking=thinking,
-            extra={"claude": self._extract_extra(request, {"messages"})},
+            extra={
+                "claude": self._extract_extra(
+                    request,
+                    {
+                        "model",
+                        "messages",
+                        "system",
+                        "max_tokens",
+                        "temperature",
+                        "top_p",
+                        "top_k",
+                        "stop_sequences",
+                        "stream",
+                        "tools",
+                        "tool_choice",
+                        "thinking",
+                    },
+                )
+            },
         )
 
         if dropped:
@@ -1151,7 +1169,7 @@ class ClaudeNormalizer(FormatNormalizer):
         if tool_choice.type == ToolChoiceType.REQUIRED:
             return {"type": "any"}
         if tool_choice.type == ToolChoiceType.TOOL:
-            return {"type": "tool_use", "name": tool_choice.tool_name or ""}
+            return {"type": "tool", "name": tool_choice.tool_name or ""}
         return {"type": "auto"}
 
     def _internal_message_to_claude(self, msg: InternalMessage) -> dict[str, Any]:

src/core/api_format/conversion/normalizers/gemini.py

@@ -204,6 +204,11 @@ class GeminiNormalizer(FormatNormalizer):
         "MAX_TOKENS": StopReason.MAX_TOKENS,
         "SAFETY": StopReason.CONTENT_FILTERED,
         "RECITATION": StopReason.CONTENT_FILTERED,
+        "LANGUAGE": StopReason.CONTENT_FILTERED,
+        "BLOCKLIST": StopReason.CONTENT_FILTERED,
+        "PROHIBITED_CONTENT": StopReason.CONTENT_FILTERED,
+        "SPII": StopReason.CONTENT_FILTERED,
+        "IMAGE_SAFETY": StopReason.CONTENT_FILTERED,
         "MALFORMED_FUNCTION_CALL": StopReason.TOOL_USE,
         "OTHER": StopReason.UNKNOWN,
     }
@@ -221,6 +226,29 @@ class GeminiNormalizer(FormatNormalizer):
         ErrorType.UNKNOWN: "INTERNAL",
     }
 
+    # generationConfig 中已被标准化提取的 key，其余写入 extra 以便 roundtrip 保留
+    _GC_KNOWN_KEYS: set[str] = {
+        "max_output_tokens",
+        "maxOutputTokens",
+        "temperature",
+        "top_p",
+        "topP",
+        "top_k",
+        "topK",
+        "stop_sequences",
+        "stopSequences",
+        "response_modalities",
+        "responseModalities",
+        "thinking_config",
+        "thinkingConfig",
+        "candidate_count",
+        "candidateCount",
+        "response_mime_type",
+        "responseMimeType",
+        "response_schema",
+        "responseSchema",
+    }
+
     # =========================
     # Requests
     # =========================
@@ -281,7 +309,48 @@ class GeminiNormalizer(FormatNormalizer):
         )
 
         # 构建 extra，保留原始 gemini 字段
-        extra: dict[str, Any] = {"gemini": self._extract_extra(request, {"contents"})}
+        extra: dict[str, Any] = {
+            "gemini": self._extract_extra(
+                request,
+                {
+                    "model",
+                    "contents",
+                    "system_instruction",
+                    "systemInstruction",
+                    "generation_config",
+                    "generationConfig",
+                    "tools",
+                    "tool_config",
+                    "toolConfig",
+                    "stream",
+                    "safetySettings",
+                    "safety_settings",
+                    "cachedContent",
+                    "cached_content",
+                },
+            )
+        }
+
+        # 保留 generationConfig 中未被标准化的额外字段（seed, presencePenalty 等）
+        raw_gc = (
+            request.get("generation_config")
+            if "generation_config" in request
+            else request.get("generationConfig")
+        )
+        if isinstance(raw_gc, dict):
+            gc_extra = {k: v for k, v in raw_gc.items() if k not in self._GC_KNOWN_KEYS}
+            if gc_extra:
+                extra.setdefault("gemini", {})["generation_config_extra"] = gc_extra
+
+        # 保留 safetySettings（原样透传）
+        raw_safety = request.get("safetySettings") or request.get("safety_settings")
+        if raw_safety:
+            extra.setdefault("gemini", {})["safety_settings"] = raw_safety
+
+        # 保留 cachedContent（原样透传）
+        raw_cached = request.get("cachedContent") or request.get("cached_content")
+        if raw_cached:
+            extra.setdefault("gemini", {})["cached_content"] = raw_cached
 
         # 保留 generationConfig 中的特殊字段（responseModalities, thinkingConfig 等）
         # 这些字段在 _get_generation_config 中已提取，需要单独存储以便转换时使用
@@ -505,6 +574,14 @@ class GeminiNormalizer(FormatNormalizer):
                 if "thinking_config" in orig_gc and "thinkingConfig" not in generation_config:
                     generation_config["thinkingConfig"] = orig_gc["thinking_config"]
 
+            # 恢复 generationConfig 中未被标准化的额外字段
+            # (seed, presencePenalty, frequencyPenalty, logprobs, speechConfig, mediaResolution 等)
+            gc_extra = gemini_extra.get("generation_config_extra")
+            if isinstance(gc_extra, dict):
+                for k, v in gc_extra.items():
+                    if k not in generation_config:
+                        generation_config[k] = v
+
         raw_contents: list[dict[str, Any]] = []
         last_idx = len(internal.messages) - 1
         for idx, msg in enumerate(internal.messages):
@@ -570,6 +647,15 @@ class GeminiNormalizer(FormatNormalizer):
         if tool_config:
             result["tool_config"] = tool_config
 
+        # 恢复 safetySettings 和 cachedContent（Gemini -> Gemini 透传）
+        if isinstance(gemini_extra, dict):
+            safety = gemini_extra.get("safety_settings")
+            if safety:
+                result["safetySettings"] = safety
+            cached = gemini_extra.get("cached_content")
+            if cached:
+                result["cachedContent"] = cached
+
         return result
 
     # =========================
@@ -577,7 +663,7 @@ class GeminiNormalizer(FormatNormalizer):
     # =========================
 
     def response_to_internal(self, response: dict[str, Any]) -> InternalResponse:
-        rid = str(response.get("id") or "")
+        rid = str(response.get("responseId") or response.get("id") or "")
         model = str(response.get("modelVersion") or response.get("model") or "")
 
         candidates = response.get("candidates") or []
@@ -594,6 +680,11 @@ class GeminiNormalizer(FormatNormalizer):
         if finish_reason is not None:
             stop_reason = self._FINISH_REASON_TO_STOP.get(str(finish_reason), StopReason.UNKNOWN)
 
+        # Gemini returns finishReason=STOP for function calls; detect tool use from content
+        has_tool_use = any(isinstance(b, ToolUseBlock) for b in blocks)
+        if has_tool_use and stop_reason != StopReason.TOOL_USE:
+            stop_reason = StopReason.TOOL_USE
+
         usage_info = self._usage_metadata_to_internal(response.get("usageMetadata"))
 
         extra: dict[str, Any] = {}
@@ -1147,8 +1238,20 @@ class GeminiNormalizer(FormatNormalizer):
 
     def error_from_internal(self, internal: InternalError) -> dict[str, Any]:
         status = self._ERROR_TYPE_TO_GEMINI_STATUS.get(internal.type, "INTERNAL")
+        code_map: dict[ErrorType, int] = {
+            ErrorType.INVALID_REQUEST: 400,
+            ErrorType.AUTHENTICATION: 401,
+            ErrorType.PERMISSION_DENIED: 403,
+            ErrorType.NOT_FOUND: 404,
+            ErrorType.RATE_LIMIT: 429,
+            ErrorType.OVERLOADED: 503,
+            ErrorType.CONTENT_FILTERED: 400,
+            ErrorType.CONTEXT_LENGTH_EXCEEDED: 400,
+            ErrorType.SERVER_ERROR: 500,
+            ErrorType.UNKNOWN: 500,
+        }
         payload: dict[str, Any] = {
-            "code": 400 if internal.type == ErrorType.INVALID_REQUEST else 500,
+            "code": code_map.get(internal.type, 500),
             "message": internal.message,
             "status": status,
         }

src/core/api_format/conversion/normalizers/openai.py

@@ -376,6 +376,14 @@ class OpenAINormalizer(FormatNormalizer):
 
         extra: dict[str, Any] = {}
 
+        # 保留 system_fingerprint / service_tier 以便 roundtrip 还原
+        sys_fp = response.get("system_fingerprint")
+        if sys_fp is not None:
+            extra.setdefault("openai", {})["system_fingerprint"] = sys_fp
+        svc_tier = response.get("service_tier")
+        if svc_tier is not None:
+            extra.setdefault("openai", {})["service_tier"] = svc_tier
+
         choices = response.get("choices") or []
         if isinstance(choices, list) and len(choices) > 1:
             extra.setdefault("openai", {})["choices"] = choices
@@ -384,8 +392,22 @@ class OpenAINormalizer(FormatNormalizer):
         message = choice0.get("message") if isinstance(choice0, dict) else None
         message = message if isinstance(message, dict) else {}
 
+        # reasoning_content -> ThinkingBlock
+        reasoning_content = message.get("reasoning_content")
+        reasoning_blocks: list[ContentBlock] = []
+        if (
+            isinstance(reasoning_content, str)
+            and reasoning_content
+            and reasoning_content != "[undefined]"
+        ):
+            reasoning_blocks.append(ThinkingBlock(thinking=reasoning_content))
+
         blocks, dropped = self._openai_content_to_blocks(message.get("content"))
 
+        # thinking blocks first (align with Claude thinking-first convention)
+        if reasoning_blocks:
+            blocks = reasoning_blocks + blocks
+
         # tool_calls -> ToolUseBlock
         tool_calls = message.get("tool_calls") or []
         if isinstance(tool_calls, list):
@@ -432,6 +454,7 @@ class OpenAINormalizer(FormatNormalizer):
             "object": "chat.completion",
             "created": int(time.time()),
             "model": model_name,
+            "system_fingerprint": None,
             "choices": [],
         }
 
@@ -448,10 +471,13 @@ class OpenAINormalizer(FormatNormalizer):
                 message["reasoning_content"] = reasoning_text
 
         content_value = self._blocks_to_openai_content(content_blocks)
+        # assistant 有 tool_calls 时 content 允许为 null；否则回退为空字符串
         if content_value is not None:
             message["content"] = content_value
-        else:
+        elif tool_blocks:
             message["content"] = None
+        else:
+            message["content"] = ""
 
         if tool_blocks:
             message["tool_calls"] = [
@@ -492,11 +518,17 @@ class OpenAINormalizer(FormatNormalizer):
                     usage_out["completion_tokens_details"] = ctd
             out["usage"] = usage_out
 
-        return out
+        # 还原 system_fingerprint / service_tier（roundtrip 保留）
+        openai_resp_extra = internal.extra.get("openai", {})
+        if isinstance(openai_resp_extra, dict):
+            sfp = openai_resp_extra.get("system_fingerprint")
+            if sfp is not None:
+                out["system_fingerprint"] = sfp
+            st = openai_resp_extra.get("service_tier")
+            if st is not None:
+                out["service_tier"] = st
 
-    # =========================
-    # Streaming
-    # =========================
+        return out
 
     def stream_chunk_to_internal(
         self, chunk: dict[str, Any], state: StreamState
@@ -732,7 +764,7 @@ class OpenAINormalizer(FormatNormalizer):
                 block_type = ss.get(f"block_type_{event.block_index}")
                 if block_type == ContentType.THINKING.value:
                     # 对齐 AM：thinking 内容输出为 reasoning_content
-                    out.append(base_chunk({"reasoning_content": event.text_delta, "content": None}))
+                    out.append(base_chunk({"reasoning_content": event.text_delta}))
                 else:
                     out.append(base_chunk({"content": event.text_delta}))
             return out
@@ -808,20 +840,12 @@ class OpenAINormalizer(FormatNormalizer):
 
         if isinstance(event, ToolCallDeltaEvent):
             tool_index = self._ensure_tool_call_index(ss, event.tool_id)
-            out.append(
-                base_chunk(
-                    {
-                        "tool_calls": [
-                            {
+            # 后续 delta 只需 index + function.arguments；id/type 仅在 ContentBlockStartEvent 首次发送
+            tc_delta: dict[str, Any] = {
                 "index": tool_index,
-                                "id": event.tool_id,
-                                "type": "function",
                 "function": {"arguments": event.input_delta},
             }
-                        ]
-                    }
-                )
-            )
+            out.append(base_chunk({"tool_calls": [tc_delta]}))
             return out
 
         if isinstance(event, MessageStopEvent):
@@ -1107,7 +1131,7 @@ class OpenAINormalizer(FormatNormalizer):
                 InternalMessage(
                     role=Role.USER,
                     content=[tr_block],
-                    extra=self._extract_extra(msg, {"role", "content"}),
+                    extra=self._extract_extra(msg, {"role", "content", "tool_call_id"}),
                 ),
                 dropped,
             )
@@ -1202,26 +1226,40 @@ class OpenAINormalizer(FormatNormalizer):
                 continue
 
             if ptype == "file":
-                file_data = part.get("file_data")
-                file_id = part.get("file_id")
-                if isinstance(file_data, dict):
+                # 标准 OpenAI 格式: {"type": "file", "file": {"file_id": ...}}
+                # 或 {"type": "file", "file": {"file_data": "data:mime;base64,...", "filename": ...}}
+                file_obj = part.get("file")
+                if isinstance(file_obj, dict):
+                    fid = file_obj.get("file_id")
+                    fdata = file_obj.get("file_data")
+                    fname = file_obj.get("filename")
+                    if isinstance(fdata, str) and fdata:
+                        # file_data 是 data URL 格式: "data:mime;base64,..."
+                        mime_type = None
+                        raw_data = fdata
+                        if fdata.startswith("data:") and ";base64," in fdata:
+                            header, raw_data = fdata.split(";base64,", 1)
+                            mime_type = header[len("data:") :]
                         blocks.append(
                             FileBlock(
-                            data=file_data.get("data"),
-                            media_type=file_data.get("mime_type"),
-                            filename=file_data.get("filename"),
-                            extra=self._extract_extra(part, {"type", "file_data"}),
+                                data=raw_data,
+                                media_type=mime_type,
+                                filename=fname,
+                                extra=self._extract_extra(part, {"type", "file"}),
                             )
                         )
-                elif isinstance(file_id, str) and file_id:
+                    elif isinstance(fid, str) and fid:
                         blocks.append(
                             FileBlock(
-                            file_id=file_id,
-                            extra=self._extract_extra(part, {"type", "file_id"}),
+                                file_id=fid,
+                                filename=fname,
+                                extra=self._extract_extra(part, {"type", "file"}),
                             )
                         )
                     else:
                         blocks.append(UnknownBlock(raw_type="file", payload=part))
+                else:
+                    blocks.append(UnknownBlock(raw_type="file", payload=part))
                 continue
 
             if ptype == "input_audio":
@@ -1511,36 +1549,27 @@ class OpenAINormalizer(FormatNormalizer):
                     text_parts.append(b.text)
                 continue
             if isinstance(b, ImageBlock):
-                # 区分两种图片来源：
-                # 1. URL 引用（OpenAI 原生格式）-> multipart content
-                # 2. base64 内嵌数据（格式转换来的）-> markdown 格式
-                if b.url and not b.data:
-                    # OpenAI 原生格式：URL 引用的图片，使用 multipart content
-                    parts.append({"type": "image_url", "image_url": {"url": b.url}})
-                elif b.data and b.media_type:
-                    # 格式转换来的图片（base64 内嵌），使用 markdown 格式
+                if b.data and b.media_type:
+                    # base64 data -> data URL in image_url format
                     data_url = f"data:{b.media_type};base64,{b.data}"
-                    text_parts.append(f"![image]({data_url})")
+                    parts.append({"type": "image_url", "image_url": {"url": data_url}})
                 elif b.url:
-                    # 有 URL 也有 data，优先使用 URL
                     parts.append({"type": "image_url", "image_url": {"url": b.url}})
                 continue
 
             if isinstance(b, FileBlock):
                 if b.data and b.media_type:
-                    # OpenAI file content part
-                    file_part: dict[str, Any] = {
-                        "type": "file",
-                        "file_data": {
-                            "mime_type": b.media_type,
-                            "data": b.data,
-                        },
-                    }
+                    # 标准 OpenAI 格式: {"type": "file", "file": {"file_data": "data:mime;base64,...", "filename": ...}}
+                    data_url = f"data:{b.media_type};base64,{b.data}"
+                    file_inner: dict[str, Any] = {"file_data": data_url}
                     if b.filename:
-                        file_part["file_data"]["filename"] = b.filename
-                    parts.append(file_part)
+                        file_inner["filename"] = b.filename
+                    parts.append({"type": "file", "file": file_inner})
                 elif b.file_id:
-                    parts.append({"type": "file", "file_id": b.file_id})
+                    file_inner_id: dict[str, Any] = {"file_id": b.file_id}
+                    if b.filename:
+                        file_inner_id["filename"] = b.filename
+                    parts.append({"type": "file", "file": file_inner_id})
                 elif b.file_url:
                     # 回退为文本描述
                     text_parts.append(f"[File: {b.file_url}]")
@@ -1576,8 +1605,9 @@ class OpenAINormalizer(FormatNormalizer):
         if text_parts:
             return "\n".join(text_parts)
 
-        # OpenAI content 可以是空字符串；但作为响应 message.content 通常允许为 ""/None。
-        return ""
+        # 无任何内容：返回 None，让调用方决定是输出 null 还是空字符串
+        # （assistant 有 tool_calls 时 content 应为 null；user 消息 content 可为空字符串）
+        return None
 
     def _split_blocks(
         self, blocks: list[ContentBlock]
@@ -1677,7 +1707,13 @@ class OpenAINormalizer(FormatNormalizer):
             out["reasoning_content"] = "".join(thinking_texts)
 
         content_value = self._blocks_to_openai_content(content_blocks)
-        out["content"] = content_value if content_value is not None else ""
+        # assistant 有 tool_calls 时 content 允许为 null；否则回退为空字符串
+        if content_value is not None:
+            out["content"] = content_value
+        elif tool_blocks:
+            out["content"] = None
+        else:
+            out["content"] = ""
 
         if tool_blocks:
             out["tool_calls"] = [
@@ -1704,8 +1740,8 @@ class OpenAINormalizer(FormatNormalizer):
         }
 
     def _tool_use_block_to_openai_call(self, block: ToolUseBlock, index: int) -> dict[str, Any]:
+        # index 参数仅用于 fallback id 生成；非流式响应的 tool_calls 数组不应包含 index 字段
         return {
-            "index": index,
             "id": block.tool_id or f"call_{index}",
             "type": "function",
             "function": {

src/core/api_format/conversion/normalizers/openai_cli.py

@@ -17,12 +17,15 @@ from typing import Any
 
 from src.core.api_format.conversion.field_mappings import (
     ERROR_TYPE_MAPPINGS,
+    REASONING_EFFORT_TO_THINKING_BUDGET,
     RETRYABLE_ERROR_TYPES,
+    THINKING_BUDGET_TO_REASONING_EFFORT,
 )
 from src.core.api_format.conversion.internal import (
     ContentBlock,
     ContentType,
     ErrorType,
+    FileBlock,
     FormatCapabilities,
     ImageBlock,
     InstructionSegment,
@@ -34,6 +37,7 @@ from src.core.api_format.conversion.internal import (
     StopReason,
     TextBlock,
     ThinkingBlock,
+    ThinkingConfig,
     ToolChoice,
     ToolChoiceType,
     ToolDefinition,
@@ -125,6 +129,24 @@ class OpenAICliNormalizer(FormatNormalizer):
 
         max_tokens = self._optional_int(request.get("max_output_tokens", request.get("max_tokens")))
 
+        # parallel_tool_calls
+        parallel_tool_calls: bool | None = None
+        ptc = request.get("parallel_tool_calls")
+        if ptc is not None:
+            parallel_tool_calls = bool(ptc)
+
+        # reasoning -> ThinkingConfig (Responses API uses reasoning.effort)
+        thinking: ThinkingConfig | None = None
+        reasoning = request.get("reasoning")
+        if isinstance(reasoning, dict):
+            effort = reasoning.get("effort")
+            if isinstance(effort, str) and effort in REASONING_EFFORT_TO_THINKING_BUDGET:
+                thinking = ThinkingConfig(
+                    enabled=True,
+                    budget_tokens=REASONING_EFFORT_TO_THINKING_BUDGET[effort],
+                    extra={"reasoning_effort": effort, "reasoning": reasoning},
+                )
+
         internal = InternalRequest(
             model=model,
             messages=messages,
@@ -137,7 +159,28 @@ class OpenAICliNormalizer(FormatNormalizer):
             stream=bool(request.get("stream") or False),
             tools=tools,
             tool_choice=tool_choice,
-            extra={"openai_cli": self._extract_extra(request, {"input"})},
+            thinking=thinking,
+            parallel_tool_calls=parallel_tool_calls,
+            extra={
+                "openai_cli": self._extract_extra(
+                    request,
+                    {
+                        "model",
+                        "input",
+                        "instructions",
+                        "max_output_tokens",
+                        "max_tokens",
+                        "temperature",
+                        "top_p",
+                        "stop",
+                        "stream",
+                        "tools",
+                        "tool_choice",
+                        "parallel_tool_calls",
+                        "reasoning",
+                    },
+                )
+            },
         )
 
         return internal
@@ -204,6 +247,26 @@ class OpenAICliNormalizer(FormatNormalizer):
         if internal.tool_choice:
             result["tool_choice"] = self._tool_choice_to_openai(internal.tool_choice)
 
+        # thinking -> reasoning (Responses API)
+        if internal.thinking and internal.thinking.enabled:
+            # 优先还原原始 reasoning 对象
+            original_reasoning = internal.thinking.extra.get("reasoning")
+            if isinstance(original_reasoning, dict):
+                result["reasoning"] = original_reasoning
+            else:
+                effort = internal.thinking.extra.get("reasoning_effort")
+                if not effort and internal.thinking.budget_tokens is not None:
+                    for threshold, level in THINKING_BUDGET_TO_REASONING_EFFORT:
+                        if internal.thinking.budget_tokens <= threshold:
+                            effort = level
+                            break
+                if effort:
+                    result["reasoning"] = {"effort": effort}
+
+        # parallel_tool_calls
+        if internal.parallel_tool_calls is not None:
+            result["parallel_tool_calls"] = internal.parallel_tool_calls
+
         # 还原 OpenAI Responses API 的其他字段（黑名单：已单独处理的字段不还原）
         handled_keys = {
             "model",
@@ -217,6 +280,8 @@ class OpenAICliNormalizer(FormatNormalizer):
             "stream",
             "tools",
             "tool_choice",
+            "parallel_tool_calls",
+            "reasoning",
         }
         for key, value in openai_cli_extra.items():
             if key not in handled_keys and key not in result:
@@ -291,7 +356,25 @@ class OpenAICliNormalizer(FormatNormalizer):
     ) -> dict[str, Any]:
         output_items: list[dict[str, Any]] = []
 
-        # 构建 output items：message（文本）和 function_call（工具调用）
+        # 构建 output items：reasoning（思考）、message（文本）、function_call（工具调用）
+        # 按 Responses API 顺序：reasoning -> message -> function_call
+        rs_idx = 0
+        for block in internal.content:
+            if isinstance(block, ThinkingBlock) and block.thinking:
+                rs_id = (
+                    f"rs_{internal.id or 'resp'}"
+                    if rs_idx == 0
+                    else f"rs_{internal.id or 'resp'}_{rs_idx}"
+                )
+                output_items.append(
+                    {
+                        "type": "reasoning",
+                        "id": rs_id,
+                        "summary": [{"type": "summary_text", "text": block.thinking}],
+                    }
+                )
+                rs_idx += 1
+
         text = self._collapse_internal_text(internal.content)
         if text:
             output_items.append(
@@ -897,6 +980,7 @@ class OpenAICliNormalizer(FormatNormalizer):
                     {
                         "type": "response.content_part.added",
                         "sequence_number": self._next_seq(ss),
+                        "item_id": message_id,
                         "output_index": output_index,
                         "content_index": 0,
                         "part": {"type": "output_text", "text": "", "annotations": []},
@@ -904,10 +988,15 @@ class OpenAICliNormalizer(FormatNormalizer):
                 )
             ss["text_started"] = True
             ss["collected_text"] = str(ss.get("collected_text") or "") + event.text_delta
+            message_id = f"msg_{state.message_id or 'stream'}"
+            output_index = ss.get("message_output_index") or 0
             out.append(
                 {
                     "type": "response.output_text.delta",
                     "sequence_number": self._next_seq(ss),
+                    "item_id": message_id,
+                    "output_index": output_index,
+                    "content_index": 0,
                     "delta": event.text_delta,
                 }
             )
@@ -970,10 +1059,14 @@ class OpenAICliNormalizer(FormatNormalizer):
             ),
         }
         if ss.get("text_started"):
+            msg_output_index = ss.get("message_output_index") or 0
             out.append(
                 {
                     "type": "response.output_text.done",
                     "sequence_number": self._next_seq(ss),
+                    "item_id": message_id,
+                    "output_index": msg_output_index,
+                    "content_index": 0,
                     "text": final_text,
                 }
             )
@@ -982,7 +1075,8 @@ class OpenAICliNormalizer(FormatNormalizer):
                 {
                     "type": "response.content_part.done",
                     "sequence_number": self._next_seq(ss),
-                    "output_index": ss.get("message_output_index") or 0,
+                    "item_id": message_id,
+                    "output_index": msg_output_index,
                     "content_index": 0,
                     "part": {"type": "output_text", "text": final_text, "annotations": []},
                 }
@@ -1151,6 +1245,7 @@ class OpenAICliNormalizer(FormatNormalizer):
             (blocks, extra, has_tool_use): 内容块列表、extra 信息、是否包含工具调用
         """
         text_parts: list[str] = []
+        thinking_blocks: list[ContentBlock] = []
         blocks: list[ContentBlock] = []
         has_tool_use = False
 
@@ -1199,13 +1294,30 @@ class OpenAICliNormalizer(FormatNormalizer):
 
                 if item_type in ("output_text", "text") and isinstance(item.get("text"), str):
                     text_parts.append(item.get("text") or "")
+                    continue
+
+                if item_type == "reasoning":
+                    # reasoning output item -> ThinkingBlock
+                    summary = item.get("summary")
+                    summary_parts: list[str] = []
+                    if isinstance(summary, list):
+                        for s in summary:
+                            if isinstance(s, dict) and s.get("type") == "summary_text":
+                                t = s.get("text")
+                                if isinstance(t, str) and t:
+                                    summary_parts.append(t)
+                    thinking_text = "\n".join(summary_parts)
+                    if thinking_text:
+                        thinking_blocks.append(ThinkingBlock(thinking=thinking_text))
+                    continue
 
         # 兼容：部分实现可能直接给 output_text
         if not text_parts and isinstance(payload.get("output_text"), str):
             text_parts.append(payload.get("output_text") or "")
 
-        # 文本块放在前面，工具调用块在后面（与 Claude 的 content 顺序一致）
+        # thinking 在前，文本在中，工具调用在后（与 Claude 的 content 顺序一致）
         result_blocks: list[ContentBlock] = []
+        result_blocks.extend(thinking_blocks)
         text = "".join(text_parts)
         if text:
             result_blocks.append(TextBlock(text=text))
@@ -1399,6 +1511,24 @@ class OpenAICliNormalizer(FormatNormalizer):
                 else:
                     blocks.append(UnknownBlock(raw_type=ptype, payload=part))
                 continue
+            if ptype == "input_file":
+                file_data = part.get("file_data")
+                file_id = part.get("file_id")
+                filename = part.get("filename")
+                fb = FileBlock(filename=filename)
+                if isinstance(file_data, str) and file_data:
+                    # file_data 是 data URL: "data:mime;base64,..."
+                    if file_data.startswith("data:") and ";base64," in file_data:
+                        header, _, data = file_data.partition(",")
+                        fb.media_type = header.split(";")[0].split(":", 1)[-1]
+                        fb.data = data
+                    else:
+                        fb.data = file_data
+                elif isinstance(file_id, str) and file_id:
+                    fb.file_id = file_id
+                fb.extra = self._extract_extra(part, {"type", "file_data", "file_id", "filename"})
+                blocks.append(fb)
+                continue
             blocks.append(UnknownBlock(raw_type=ptype or "unknown", payload=part))
         return blocks
 
@@ -1428,15 +1558,20 @@ class OpenAICliNormalizer(FormatNormalizer):
                     continue
 
                 if isinstance(block, ToolResultBlock):
+                    # Responses API function_call_output.output 必须是字符串
+                    if block.content_text is not None:
+                        output_str = block.content_text
+                    elif isinstance(block.output, str):
+                        output_str = block.output
+                    elif block.output is not None:
+                        output_str = json.dumps(block.output, ensure_ascii=False)
+                    else:
+                        output_str = ""
                     out.append(
                         {
                             "type": "function_call_output",
                             "call_id": block.tool_use_id,
-                            "output": (
-                                block.content_text
-                                if block.content_text is not None
-                                else block.output
-                            ),
+                            "output": output_str,
                         }
                     )
                     continue
@@ -1567,6 +1702,11 @@ class OpenAICliNormalizer(FormatNormalizer):
                 return ToolChoice(
                     type=ToolChoiceType.AUTO, extra={"openai_cli": {"tool_choice": tool_choice}}
                 )
+            if tool_choice == "required":
+                return ToolChoice(
+                    type=ToolChoiceType.REQUIRED,
+                    extra={"openai_cli": {"tool_choice": tool_choice}},
+                )
             return ToolChoice(type=ToolChoiceType.AUTO, extra={"raw": tool_choice})
 
         if isinstance(tool_choice, dict):

src/core/api_format/conversion/stream_bridge.py

@@ -166,7 +166,30 @@ class InternalStreamAggregator:
                 self._open.clear()
                 continue
 
+    @property
+    def open_count(self) -> int:
+        """当前未关闭的 block 数量。"""
+        return len(self._open)
+
+    @property
+    def final_count(self) -> int:
+        """已完成的 block 数量。"""
+        return len(self._final)
+
+    @property
+    def usage(self) -> UsageInfo | None:
+        return self._usage
+
+    @property
+    def stop_reason(self) -> StopReason | None:
+        return self._stop_reason
+
     def build(self) -> InternalResponse:
+        # Flush remaining open blocks (best-effort) in case MessageStopEvent was never received.
+        for idx, b in list(self._open.items()):
+            self._final.setdefault(idx, b.finalize())
+        self._open.clear()
+
         rid = self._id or self._fallback_id
         model = self._model or self._fallback_model
         content = [self._final[k] for k in sorted(self._final.keys())]

tests/core/api_format/conversion/fixtures/__init__.py

@@ -0,0 +1,6 @@
+"""
+Format conversion test fixtures.
+
+Provides golden internal representations, format-specific fixtures,
+stream fixtures, error fixtures, and assertion helpers.
+"""

tests/core/api_format/conversion/fixtures/assertions.py

@@ -0,0 +1,361 @@
+"""
+Assertion helpers for format conversion tests.
+
+Provides semantic comparison functions that check meaningful equivalence
+while tolerating format-specific differences (extra fields, id regeneration, etc.).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from src.core.api_format.conversion.internal import (
+    ContentBlock,
+    ImageBlock,
+    InternalMessage,
+    InternalRequest,
+    InternalResponse,
+    StopReason,
+    TextBlock,
+    ThinkingBlock,
+    ToolDefinition,
+    ToolResultBlock,
+    ToolUseBlock,
+    UnknownBlock,
+)
+from src.core.api_format.conversion.stream_events import (
+    ContentBlockStartEvent,
+    ContentDeltaEvent,
+    InternalStreamEvent,
+    MessageStopEvent,
+    ToolCallDeltaEvent,
+)
+
+
+def assert_internal_request_matches(
+    actual: InternalRequest,
+    expected: InternalRequest,
+    required_fields: set[str],
+) -> None:
+    """Verify that actual InternalRequest matches expected on required fields."""
+    if "model" in required_fields:
+        assert (
+            actual.model == expected.model
+        ), f"model mismatch: {actual.model!r} != {expected.model!r}"
+
+    if "messages" in required_fields:
+        # Merge consecutive same-role messages before comparison,
+        # since normalizers may merge/split them during conversion.
+        actual_msgs = _merge_consecutive_same_role(actual.messages)
+        expected_msgs = _merge_consecutive_same_role(expected.messages)
+        assert len(actual_msgs) == len(
+            expected_msgs
+        ), f"message count mismatch: {len(actual_msgs)} != {len(expected_msgs)}"
+        for i, (a, e) in enumerate(zip(actual_msgs, expected_msgs)):
+            assert a.role == e.role, f"message[{i}] role mismatch: {a.role} != {e.role}"
+            assert_content_blocks_match_unordered(a.content, e.content, context=f"message[{i}]")
+
+    if "system" in required_fields:
+        # Allow either system or instructions to carry the system prompt
+        actual_sys = actual.system or _join_instructions(actual.instructions)
+        expected_sys = expected.system or _join_instructions(expected.instructions)
+        assert actual_sys == expected_sys, f"system mismatch: {actual_sys!r} != {expected_sys!r}"
+
+    if "max_tokens" in required_fields:
+        assert (
+            actual.max_tokens == expected.max_tokens
+        ), f"max_tokens mismatch: {actual.max_tokens} != {expected.max_tokens}"
+
+    if "stream" in required_fields:
+        assert actual.stream == expected.stream
+
+    if "tools" in required_fields:
+        assert_tools_match(actual.tools, expected.tools)
+
+    if "tool_choice" in required_fields:
+        if expected.tool_choice is not None:
+            assert actual.tool_choice is not None, "tool_choice is None but expected non-None"
+            assert (
+                actual.tool_choice.type == expected.tool_choice.type
+            ), f"tool_choice.type mismatch: {actual.tool_choice.type} != {expected.tool_choice.type}"
+
+
+def assert_internal_response_matches(
+    actual: InternalResponse,
+    expected: InternalResponse,
+    required_fields: set[str],
+) -> None:
+    """Verify that actual InternalResponse matches expected on required fields."""
+    if "content" in required_fields:
+        assert_content_blocks_match(actual.content, expected.content, context="response")
+
+    if "stop_reason" in required_fields:
+        assert (
+            actual.stop_reason == expected.stop_reason
+        ), f"stop_reason mismatch: {actual.stop_reason} != {expected.stop_reason}"
+
+    if "usage" in required_fields and expected.usage is not None:
+        assert actual.usage is not None, "usage is None but expected non-None"
+        assert actual.usage.input_tokens == expected.usage.input_tokens
+        assert actual.usage.output_tokens == expected.usage.output_tokens
+
+
+def assert_content_blocks_match(
+    actual_blocks: list[ContentBlock],
+    expected_blocks: list[ContentBlock],
+    *,
+    context: str = "",
+) -> None:
+    """Verify content block lists are semantically equivalent (ignoring extra)."""
+    # Filter out UnknownBlock (allowed to be lost)
+    actual_meaningful = [b for b in actual_blocks if not isinstance(b, UnknownBlock)]
+    expected_meaningful = [b for b in expected_blocks if not isinstance(b, UnknownBlock)]
+
+    assert len(actual_meaningful) == len(expected_meaningful), (
+        f"{context} block count mismatch: {len(actual_meaningful)} != {len(expected_meaningful)}"
+        f"\n  actual types: {[type(b).__name__ for b in actual_meaningful]}"
+        f"\n  expected types: {[type(b).__name__ for b in expected_meaningful]}"
+    )
+
+    for i, (a, e) in enumerate(zip(actual_meaningful, expected_meaningful)):
+        prefix = f"{context}.block[{i}]" if context else f"block[{i}]"
+        assert type(a) is type(
+            e
+        ), f"{prefix} type mismatch: {type(a).__name__} != {type(e).__name__}"
+
+        if isinstance(a, TextBlock) and isinstance(e, TextBlock):
+            assert a.text == e.text, f"{prefix} text mismatch: {a.text!r} != {e.text!r}"
+
+        elif isinstance(a, ToolUseBlock) and isinstance(e, ToolUseBlock):
+            assert (
+                a.tool_name == e.tool_name
+            ), f"{prefix} tool_name mismatch: {a.tool_name!r} != {e.tool_name!r}"
+            assert (
+                a.tool_input == e.tool_input
+            ), f"{prefix} tool_input mismatch: {a.tool_input} != {e.tool_input}"
+            # tool_id may be regenerated, just verify non-empty
+            assert bool(a.tool_id), f"{prefix} tool_id is empty"
+
+        elif isinstance(a, ToolResultBlock) and isinstance(e, ToolResultBlock):
+            assert bool(a.tool_use_id), f"{prefix} tool_use_id is empty"
+            # content_text or output should be semantically equivalent
+            # Normalizers may parse JSON strings into dicts, so compare semantically
+            a_val = _normalize_tool_output(a)
+            e_val = _normalize_tool_output(e)
+            assert a_val == e_val, f"{prefix} tool result mismatch: {a_val!r} != {e_val!r}"
+
+        elif isinstance(a, ThinkingBlock) and isinstance(e, ThinkingBlock):
+            assert (
+                a.thinking == e.thinking
+            ), f"{prefix} thinking mismatch: {a.thinking!r} != {e.thinking!r}"
+
+        elif isinstance(a, ImageBlock) and isinstance(e, ImageBlock):
+            if e.url:
+                assert a.url == e.url, f"{prefix} image url mismatch"
+            if e.data:
+                assert a.data == e.data, f"{prefix} image data mismatch"
+            if e.media_type:
+                assert a.media_type == e.media_type, f"{prefix} media_type mismatch"
+
+
+def assert_tools_match(
+    actual: list[ToolDefinition] | None,
+    expected: list[ToolDefinition] | None,
+) -> None:
+    """Verify tool definitions match."""
+    if expected is None:
+        return
+    assert actual is not None, "tools is None but expected non-None"
+    assert len(actual) == len(expected), f"tools count mismatch: {len(actual)} != {len(expected)}"
+    for i, (a, e) in enumerate(zip(actual, expected)):
+        assert a.name == e.name, f"tool[{i}].name mismatch: {a.name!r} != {e.name!r}"
+        if e.description is not None:
+            assert a.description == e.description, f"tool[{i}].description mismatch"
+        if e.parameters is not None:
+            assert a.parameters == e.parameters, f"tool[{i}].parameters mismatch"
+
+
+def assert_content_blocks_match_unordered(
+    actual_blocks: list[ContentBlock],
+    expected_blocks: list[ContentBlock],
+    *,
+    context: str = "",
+) -> None:
+    """Verify content blocks are semantically equivalent regardless of order.
+
+    Groups blocks by type and compares within each group. This tolerates
+    reordering that normalizers may introduce during roundtrip (e.g. placing
+    tool_result before or after text within the same message).
+    """
+    actual_meaningful = [b for b in actual_blocks if not isinstance(b, UnknownBlock)]
+    expected_meaningful = [b for b in expected_blocks if not isinstance(b, UnknownBlock)]
+
+    assert len(actual_meaningful) == len(expected_meaningful), (
+        f"{context} block count mismatch: {len(actual_meaningful)} != {len(expected_meaningful)}"
+        f"\n  actual types: {[type(b).__name__ for b in actual_meaningful]}"
+        f"\n  expected types: {[type(b).__name__ for b in expected_meaningful]}"
+    )
+
+    def _group_by_type(blocks: Sequence[ContentBlock]) -> dict[type, list[ContentBlock]]:
+        groups: dict[type, list[ContentBlock]] = {}
+        for b in blocks:
+            groups.setdefault(type(b), []).append(b)
+        return groups
+
+    actual_groups = _group_by_type(actual_meaningful)
+    expected_groups = _group_by_type(expected_meaningful)
+
+    assert set(actual_groups.keys()) == set(expected_groups.keys()), (
+        f"{context} block type sets differ: "
+        f"{[t.__name__ for t in actual_groups]} != {[t.__name__ for t in expected_groups]}"
+    )
+
+    for btype in expected_groups:
+        a_list = actual_groups[btype]
+        e_list = expected_groups[btype]
+        assert len(a_list) == len(
+            e_list
+        ), f"{context} {btype.__name__} count mismatch: {len(a_list)} != {len(e_list)}"
+        for i, (a, e) in enumerate(zip(a_list, e_list)):
+            prefix = f"{context}.{btype.__name__}[{i}]" if context else f"{btype.__name__}[{i}]"
+            if isinstance(a, TextBlock) and isinstance(e, TextBlock):
+                assert a.text == e.text, f"{prefix} text mismatch: {a.text!r} != {e.text!r}"
+            elif isinstance(a, ToolUseBlock) and isinstance(e, ToolUseBlock):
+                assert a.tool_name == e.tool_name, f"{prefix} tool_name mismatch"
+                assert a.tool_input == e.tool_input, f"{prefix} tool_input mismatch"
+            elif isinstance(a, ToolResultBlock) and isinstance(e, ToolResultBlock):
+                a_val = _normalize_tool_output(a)
+                e_val = _normalize_tool_output(e)
+                assert a_val == e_val, f"{prefix} tool result mismatch: {a_val!r} != {e_val!r}"
+            elif isinstance(a, ThinkingBlock) and isinstance(e, ThinkingBlock):
+                assert a.thinking == e.thinking, f"{prefix} thinking mismatch"
+            elif isinstance(a, ImageBlock) and isinstance(e, ImageBlock):
+                if e.url:
+                    assert a.url == e.url, f"{prefix} image url mismatch"
+                if e.data:
+                    assert a.data == e.data, f"{prefix} image data mismatch"
+
+
+def _merge_consecutive_same_role(
+    messages: list[InternalMessage],
+) -> list[InternalMessage]:
+    """Merge consecutive messages with the same role into one (for semantic comparison)."""
+    if not messages:
+        return []
+    merged: list[InternalMessage] = []
+    for msg in messages:
+        if merged and merged[-1].role == msg.role:
+            merged[-1] = InternalMessage(
+                role=msg.role,
+                content=list(merged[-1].content) + list(msg.content),
+            )
+        else:
+            merged.append(InternalMessage(role=msg.role, content=list(msg.content)))
+    return merged
+
+
+def assert_internal_requests_equivalent(
+    a: InternalRequest,
+    b: InternalRequest,
+    lossy_fields: set[str] | None = None,
+) -> None:
+    """Verify two InternalRequests are semantically equivalent after roundtrip."""
+    lossy = lossy_fields or set()
+
+    assert a.model == b.model
+    if "messages" not in lossy:
+        # Merge consecutive same-role messages before comparison,
+        # since normalizers may merge/split them during roundtrip.
+        a_msgs = _merge_consecutive_same_role(a.messages)
+        b_msgs = _merge_consecutive_same_role(b.messages)
+        assert len(a_msgs) == len(
+            b_msgs
+        ), f"message count mismatch after merge: {len(a_msgs)} != {len(b_msgs)}"
+        for i, (am, bm) in enumerate(zip(a_msgs, b_msgs)):
+            assert am.role == bm.role, f"message[{i}] role mismatch after roundtrip"
+            # Use order-insensitive comparison: normalizers may reorder blocks
+            # within a message during roundtrip (e.g. tool_result before/after text).
+            assert_content_blocks_match_unordered(
+                am.content, bm.content, context=f"roundtrip.message[{i}]"
+            )
+    if "system" not in lossy:
+        a_sys = a.system or _join_instructions(a.instructions)
+        b_sys = b.system or _join_instructions(b.instructions)
+        assert a_sys == b_sys
+    if "max_tokens" not in lossy:
+        assert a.max_tokens == b.max_tokens
+    if "tools" not in lossy:
+        assert_tools_match(a.tools, b.tools)
+
+
+def assert_stream_text_matches(
+    events: list[InternalStreamEvent],
+    expected_text: str,
+) -> None:
+    """Verify that stream events produce the expected text when concatenated."""
+    parts: list[str] = []
+    for evt in events:
+        if isinstance(evt, ContentDeltaEvent) and evt.text_delta:
+            parts.append(evt.text_delta)
+    actual = "".join(parts)
+    assert actual == expected_text, f"stream text mismatch: {actual!r} != {expected_text!r}"
+
+
+def assert_stream_stop_reason_matches(
+    events: list[InternalStreamEvent],
+    expected: StopReason,
+) -> None:
+    """Verify that the stream ends with the expected stop reason."""
+    stop_events = [e for e in events if isinstance(e, MessageStopEvent)]
+    assert stop_events, "no MessageStopEvent found in stream events"
+    last = stop_events[-1]
+    assert (
+        last.stop_reason == expected
+    ), f"stream stop_reason mismatch: {last.stop_reason} != {expected}"
+
+
+def _join_instructions(instructions: list) -> str | None:
+    if not instructions:
+        return None
+    parts = [seg.text for seg in instructions if seg.text]
+    return "\n\n".join(parts) or None
+
+
+def _normalize_tool_output(block: ToolResultBlock) -> object:
+    """Normalize tool output for comparison (parse JSON strings to dicts)."""
+    import json
+
+    val = block.content_text if block.content_text is not None else block.output
+    if val is None:
+        return ""
+    if isinstance(val, str):
+        try:
+            return json.loads(val)
+        except (json.JSONDecodeError, TypeError):
+            return val
+    return val
+
+
+def assert_stream_has_tool_call(
+    events: list[InternalStreamEvent],
+    expected_tool_name: str,
+) -> None:
+    """Verify that stream events contain a tool call with the expected name."""
+    from src.core.api_format.conversion.internal import ContentType
+
+    tool_starts = [
+        e
+        for e in events
+        if isinstance(e, ContentBlockStartEvent) and e.block_type == ContentType.TOOL_USE
+    ]
+    assert tool_starts, "no tool call ContentBlockStartEvent found in stream events"
+    names = [e.tool_name for e in tool_starts]
+    assert (
+        expected_tool_name in names
+    ), f"tool name {expected_tool_name!r} not found in stream tool starts: {names}"
+
+    # Verify there are ToolCallDeltaEvents with non-empty input
+    tool_deltas = [e for e in events if isinstance(e, ToolCallDeltaEvent)]
+    assert tool_deltas, "no ToolCallDeltaEvent found in stream events"
+    combined = "".join(d.input_delta for d in tool_deltas)
+    assert combined, "tool call input_delta is empty after concatenation"

tests/core/api_format/conversion/fixtures/error_fixtures.py

@@ -0,0 +1,258 @@
+"""
+Error fixtures for each format.
+
+Each fixture defines a format-specific error response and the expected
+InternalError it should produce.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from src.core.api_format.conversion.internal import ErrorType
+
+
+@dataclass
+class ErrorFixture:
+    """A format-specific error fixture."""
+
+    error_response: dict[str, Any]
+    expected_type: ErrorType
+    expected_message: str
+
+
+# ===================================================================
+# Claude error responses
+# ===================================================================
+
+_CLAUDE_ERRORS: dict[str, ErrorFixture] = {
+    "invalid_request": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "invalid_request_error",
+                "message": "max_tokens must be a positive integer",
+            },
+        },
+        expected_type=ErrorType.INVALID_REQUEST,
+        expected_message="max_tokens must be a positive integer",
+    ),
+    "rate_limit": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "rate_limit_error",
+                "message": "Rate limit exceeded",
+            },
+        },
+        expected_type=ErrorType.RATE_LIMIT,
+        expected_message="Rate limit exceeded",
+    ),
+    "auth_error": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "authentication_error",
+                "message": "Invalid API key",
+            },
+        },
+        expected_type=ErrorType.AUTHENTICATION,
+        expected_message="Invalid API key",
+    ),
+    "overloaded": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "overloaded_error",
+                "message": "Overloaded",
+            },
+        },
+        expected_type=ErrorType.OVERLOADED,
+        expected_message="Overloaded",
+    ),
+    "server_error": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "api_error",
+                "message": "Internal server error",
+            },
+        },
+        expected_type=ErrorType.SERVER_ERROR,
+        expected_message="Internal server error",
+    ),
+    "not_found": ErrorFixture(
+        error_response={
+            "type": "error",
+            "error": {
+                "type": "not_found_error",
+                "message": "Model not found",
+            },
+        },
+        expected_type=ErrorType.NOT_FOUND,
+        expected_message="Model not found",
+    ),
+}
+
+# ===================================================================
+# OpenAI Chat error responses
+# ===================================================================
+
+_OPENAI_CHAT_ERRORS: dict[str, ErrorFixture] = {
+    "invalid_request": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Invalid value for max_tokens",
+                "type": "invalid_request_error",
+                "param": "max_tokens",
+                "code": None,
+            }
+        },
+        expected_type=ErrorType.INVALID_REQUEST,
+        expected_message="Invalid value for max_tokens",
+    ),
+    "rate_limit": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Rate limit reached",
+                "type": "rate_limit_exceeded",
+                "param": None,
+                "code": "rate_limit_exceeded",
+            }
+        },
+        expected_type=ErrorType.RATE_LIMIT,
+        expected_message="Rate limit reached",
+    ),
+    "auth_error": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Incorrect API key provided",
+                "type": "invalid_api_key",
+                "param": None,
+                "code": "invalid_api_key",
+            }
+        },
+        expected_type=ErrorType.AUTHENTICATION,
+        expected_message="Incorrect API key provided",
+    ),
+    "server_error": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "The server had an error",
+                "type": "server_error",
+                "param": None,
+                "code": "server_error",
+            }
+        },
+        expected_type=ErrorType.SERVER_ERROR,
+        expected_message="The server had an error",
+    ),
+}
+
+# ===================================================================
+# OpenAI CLI (Responses API) error responses
+# ===================================================================
+
+_OPENAI_CLI_ERRORS: dict[str, ErrorFixture] = {
+    "invalid_request": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Invalid input",
+                "type": "invalid_request_error",
+                "code": None,
+            }
+        },
+        expected_type=ErrorType.INVALID_REQUEST,
+        expected_message="Invalid input",
+    ),
+    "rate_limit": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Rate limit reached",
+                "type": "rate_limit_exceeded",
+                "code": "rate_limit_exceeded",
+            }
+        },
+        expected_type=ErrorType.RATE_LIMIT,
+        expected_message="Rate limit reached",
+    ),
+    "server_error": ErrorFixture(
+        error_response={
+            "error": {
+                "message": "Internal server error",
+                "type": "server_error",
+                "code": "server_error",
+            }
+        },
+        expected_type=ErrorType.SERVER_ERROR,
+        expected_message="Internal server error",
+    ),
+}
+
+# ===================================================================
+# Gemini error responses
+# ===================================================================
+
+_GEMINI_ERRORS: dict[str, ErrorFixture] = {
+    "invalid_request": ErrorFixture(
+        error_response={
+            "error": {
+                "code": 400,
+                "message": "Invalid value for field",
+                "status": "INVALID_ARGUMENT",
+            }
+        },
+        expected_type=ErrorType.INVALID_REQUEST,
+        expected_message="Invalid value for field",
+    ),
+    "rate_limit": ErrorFixture(
+        error_response={
+            "error": {
+                "code": 429,
+                "message": "Resource exhausted",
+                "status": "RESOURCE_EXHAUSTED",
+            }
+        },
+        expected_type=ErrorType.RATE_LIMIT,
+        expected_message="Resource exhausted",
+    ),
+    "auth_error": ErrorFixture(
+        error_response={
+            "error": {
+                "code": 401,
+                "message": "API key not valid",
+                "status": "UNAUTHENTICATED",
+            }
+        },
+        expected_type=ErrorType.AUTHENTICATION,
+        expected_message="API key not valid",
+    ),
+    "server_error": ErrorFixture(
+        error_response={
+            "error": {
+                "code": 500,
+                "message": "Internal error encountered",
+                "status": "INTERNAL",
+            }
+        },
+        expected_type=ErrorType.SERVER_ERROR,
+        expected_message="Internal error encountered",
+    ),
+}
+
+
+# ===================================================================
+# Registry
+# ===================================================================
+
+ERROR_FIXTURES: dict[str, dict[str, ErrorFixture]] = {
+    "claude:chat": _CLAUDE_ERRORS,
+    "claude:cli": _CLAUDE_ERRORS,
+    "openai:chat": _OPENAI_CHAT_ERRORS,
+    "openai:cli": _OPENAI_CLI_ERRORS,
+    "gemini:chat": _GEMINI_ERRORS,
+    "gemini:cli": _GEMINI_ERRORS,
+}
+
+ERROR_ALL_FORMATS = list(ERROR_FIXTURES.keys())

tests/core/api_format/conversion/fixtures/golden_internal.py

@@ -0,0 +1,425 @@
+"""
+Internal golden fixtures.
+
+Each fixture defines the canonical InternalRequest / InternalResponse
+that all normalizers must produce (or consume) for a given scenario.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from src.core.api_format.conversion.internal import (
+    ImageBlock,
+    InstructionSegment,
+    InternalMessage,
+    InternalRequest,
+    InternalResponse,
+    Role,
+    StopReason,
+    TextBlock,
+    ThinkingBlock,
+    ToolChoice,
+    ToolChoiceType,
+    ToolDefinition,
+    ToolResultBlock,
+    ToolUseBlock,
+    UsageInfo,
+)
+
+
+@dataclass
+class GoldenFixture:
+    """A golden internal fixture for a specific scenario."""
+
+    fixture_id: str
+    description: str
+    internal_request: InternalRequest
+    internal_response: InternalResponse
+    # Fields that MUST be correctly converted by every normalizer
+    required_fields: set[str] = field(default_factory=set)
+    # Fields that may be lost during conversion (format-specific extras)
+    lossy_fields: set[str] = field(default_factory=set)
+
+
+# ---------------------------------------------------------------------------
+# Shared constants
+# ---------------------------------------------------------------------------
+_MODEL = "test-model"
+_SYSTEM = "You are a helpful assistant."
+_TOOL_DEF = ToolDefinition(
+    name="get_weather",
+    description="Get the current weather for a location.",
+    parameters={
+        "type": "object",
+        "properties": {
+            "location": {"type": "string", "description": "City name"},
+        },
+        "required": ["location"],
+    },
+)
+
+_TOOL_ID = "tool_call_001"
+
+_REQUIRED_REQUEST = {"model", "messages", "system"}
+_REQUIRED_RESPONSE = {"content", "stop_reason"}
+
+
+# ---------------------------------------------------------------------------
+# simple_text: single-turn text conversation
+# ---------------------------------------------------------------------------
+SIMPLE_TEXT = GoldenFixture(
+    fixture_id="simple_text",
+    description="Single-turn text conversation with system prompt",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(role=Role.USER, content=[TextBlock(text="Hello, how are you?")]),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_001",
+        model=_MODEL,
+        content=[TextBlock(text="I'm doing well, thank you!")],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=10, output_tokens=8, total_tokens=18),
+    ),
+    required_fields={"model", "messages", "system", "max_tokens", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# multi_turn: multi-turn conversation
+# ---------------------------------------------------------------------------
+MULTI_TURN = GoldenFixture(
+    fixture_id="multi_turn",
+    description="Multi-turn conversation with user/assistant alternation",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(role=Role.USER, content=[TextBlock(text="What is 2+2?")]),
+            InternalMessage(role=Role.ASSISTANT, content=[TextBlock(text="4")]),
+            InternalMessage(role=Role.USER, content=[TextBlock(text="And 3+3?")]),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_002",
+        model=_MODEL,
+        content=[TextBlock(text="6")],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=20, output_tokens=1, total_tokens=21),
+    ),
+    required_fields={"model", "messages", "system", "content", "stop_reason"},
+)
+
+# ---------------------------------------------------------------------------
+# tool_use: tool call + tool result
+# ---------------------------------------------------------------------------
+TOOL_USE = GoldenFixture(
+    fixture_id="tool_use",
+    description="Single tool call with result in conversation history",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(
+                role=Role.USER, content=[TextBlock(text="What is the weather in Tokyo?")]
+            ),
+            InternalMessage(
+                role=Role.ASSISTANT,
+                content=[
+                    TextBlock(text="Let me check the weather for you."),
+                    ToolUseBlock(
+                        tool_id=_TOOL_ID,
+                        tool_name="get_weather",
+                        tool_input={"location": "Tokyo"},
+                    ),
+                ],
+            ),
+            InternalMessage(
+                role=Role.USER,
+                content=[
+                    ToolResultBlock(
+                        tool_use_id=_TOOL_ID,
+                        content_text='{"temperature": 22, "condition": "sunny"}',
+                    ),
+                ],
+            ),
+            InternalMessage(role=Role.USER, content=[TextBlock(text="Thanks!")]),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+        tools=[_TOOL_DEF],
+    ),
+    internal_response=InternalResponse(
+        id="resp_003",
+        model=_MODEL,
+        content=[TextBlock(text="The weather in Tokyo is 22C and sunny.")],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=50, output_tokens=12, total_tokens=62),
+    ),
+    required_fields={"model", "messages", "system", "tools", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# tool_use_response: response that contains a tool call (not end_turn)
+# ---------------------------------------------------------------------------
+TOOL_USE_RESPONSE = GoldenFixture(
+    fixture_id="tool_use_response",
+    description="Response that is a tool call (stop_reason=tool_use)",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(
+                role=Role.USER, content=[TextBlock(text="What is the weather in Tokyo?")]
+            ),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+        tools=[_TOOL_DEF],
+    ),
+    internal_response=InternalResponse(
+        id="resp_004",
+        model=_MODEL,
+        content=[
+            ToolUseBlock(
+                tool_id=_TOOL_ID,
+                tool_name="get_weather",
+                tool_input={"location": "Tokyo"},
+            ),
+        ],
+        stop_reason=StopReason.TOOL_USE,
+        usage=UsageInfo(input_tokens=30, output_tokens=15, total_tokens=45),
+    ),
+    required_fields={"model", "messages", "tools", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# thinking: response with thinking block
+# ---------------------------------------------------------------------------
+THINKING = GoldenFixture(
+    fixture_id="thinking",
+    description="Response with thinking/reasoning content",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(role=Role.USER, content=[TextBlock(text="Solve: 15 * 23")]),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=2048,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_005",
+        model=_MODEL,
+        content=[
+            ThinkingBlock(thinking="15 * 23 = 15 * 20 + 15 * 3 = 300 + 45 = 345"),
+            TextBlock(text="345"),
+        ],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=15, output_tokens=20, total_tokens=35),
+    ),
+    required_fields={"model", "messages", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# image_url: image input via URL
+# ---------------------------------------------------------------------------
+IMAGE_URL = GoldenFixture(
+    fixture_id="image_url",
+    description="Image input via URL",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(
+                role=Role.USER,
+                content=[
+                    ImageBlock(url="https://example.com/image.png", media_type="image/png"),
+                    TextBlock(text="What is in this image?"),
+                ],
+            ),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_006",
+        model=_MODEL,
+        content=[TextBlock(text="I see a cat.")],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=100, output_tokens=5, total_tokens=105),
+    ),
+    required_fields={"model", "messages", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# image_base64: image input via base64
+# ---------------------------------------------------------------------------
+IMAGE_BASE64 = GoldenFixture(
+    fixture_id="image_base64",
+    description="Image input via base64 data",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(
+                role=Role.USER,
+                content=[
+                    ImageBlock(data="iVBORw0KGgo=", media_type="image/png"),
+                    TextBlock(text="Describe this image."),
+                ],
+            ),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_007",
+        model=_MODEL,
+        content=[TextBlock(text="A small icon.")],
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=80, output_tokens=3, total_tokens=83),
+    ),
+    required_fields={"model", "messages", "content", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# empty_response: response with no content
+# ---------------------------------------------------------------------------
+EMPTY_RESPONSE = GoldenFixture(
+    fixture_id="empty_response",
+    description="Empty response (no content blocks)",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(role=Role.USER, content=[TextBlock(text="Say nothing.")]),
+        ],
+        instructions=[InstructionSegment(role=Role.SYSTEM, text=_SYSTEM)],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+    ),
+    internal_response=InternalResponse(
+        id="resp_008",
+        model=_MODEL,
+        content=[],  # Normalizers typically drop empty text blocks
+        stop_reason=StopReason.END_TURN,
+        usage=UsageInfo(input_tokens=10, output_tokens=0, total_tokens=10),
+    ),
+    required_fields={"model", "stop_reason"},
+)
+
+
+# ---------------------------------------------------------------------------
+# tool_choice_auto: tool_choice=auto
+# ---------------------------------------------------------------------------
+TOOL_CHOICE_AUTO = GoldenFixture(
+    fixture_id="tool_choice_auto",
+    description="Request with tool_choice=auto",
+    internal_request=InternalRequest(
+        model=_MODEL,
+        messages=[
+            InternalMessage(role=Role.USER, content=[TextBlock(text="Help me.")]),
+        ],
+        system=_SYSTEM,
+        max_tokens=1024,
+        stream=False,
+        tools=[_TOOL_DEF],
+        tool_choice=ToolChoice(type=ToolChoiceType.AUTO),
+    ),
+    internal_response=InternalResponse(
+        id="resp_009",
+        model=_MODEL,
+        content=[TextBlock(text="Sure!")],
+        stop_reason=StopReason.END_TURN,
+    ),
+    required_fields={"model", "messages", "tools", "tool_choice"},
+)
+
+
+# ---------------------------------------------------------------------------
+# Registry of all golden fixtures
+# ---------------------------------------------------------------------------
+ALL_GOLDEN_FIXTURES: dict[str, GoldenFixture] = {
+    f.fixture_id: f
+    for f in [
+        SIMPLE_TEXT,
+        MULTI_TURN,
+        TOOL_USE,
+        TOOL_USE_RESPONSE,
+        THINKING,
+        IMAGE_URL,
+        IMAGE_BASE64,
+        EMPTY_RESPONSE,
+        TOOL_CHOICE_AUTO,
+    ]
+}
+
+# Fixture IDs that all formats must support (core scenarios)
+CORE_FIXTURE_IDS = ["simple_text", "multi_turn", "tool_use", "empty_response"]
+
+# Fixture IDs for extended scenarios (some formats may not support)
+EXTENDED_FIXTURE_IDS = [
+    "tool_use_response",
+    "thinking",
+    "image_url",
+    "image_base64",
+    "tool_choice_auto",
+]
+
+ALL_FIXTURE_IDS = CORE_FIXTURE_IDS + EXTENDED_FIXTURE_IDS
+
+# ---------------------------------------------------------------------------
+# Known normalizer limitations for extended fixtures.
+#
+# Maps (format_id, fixture_id, test_layer) -> reason string.
+# test_layer: "to_internal", "from_internal", "roundtrip", "cross_request", "cross_response"
+#
+# These are documented limitations of the current normalizer implementations,
+# NOT bugs to fix. Tests will skip these combinations.
+# ---------------------------------------------------------------------------
+KNOWN_LIMITATIONS: dict[tuple[str, str, str], str] = {}
+
+# Formats where response_to_internal loses ThinkingBlock (source limitation)
+_THINKING_RESPONSE_LOSSY_SOURCES = {"openai:cli"}
+
+# Fixtures where the response's thinking block is lost when target format
+# doesn't support ThinkingBlock in non-streaming responses.
+_THINKING_RESPONSE_LOSSY_TARGETS = {"openai:cli"}
+
+
+def is_cross_format_limited(
+    source: str,
+    target: str,
+    fixture_id: str,
+    layer: str,
+) -> str | None:
+    """Return a reason string if this cross-format combo is a known limitation, else None."""
+    # thinking response: openai:cli doesn't support ThinkingBlock in non-streaming
+    if fixture_id == "thinking" and layer == "cross_response":
+        if source in _THINKING_RESPONSE_LOSSY_SOURCES:
+            return f"{source} does not parse thinking content into ThinkingBlock"
+        if target in _THINKING_RESPONSE_LOSSY_TARGETS:
+            return f"{target} does not preserve ThinkingBlock in non-streaming responses"
+
+    return None

tests/core/api_format/conversion/fixtures/stream_fixtures.py

@@ -0,0 +1,504 @@
+"""
+Stream fixtures for each format.
+
+Each fixture defines a sequence of format-specific SSE chunks and the
+expected internal stream events / final text they should produce.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from src.core.api_format.conversion.internal import StopReason
+
+from .golden_internal import _MODEL
+
+
+@dataclass
+class StreamFixture:
+    """A stream fixture for a specific format and scenario."""
+
+    chunks: list[dict[str, Any]]
+    expected_text: str
+    expected_stop_reason: StopReason
+    # Fields that may differ across formats
+    lossy_fields: set[str] = field(default_factory=set)
+
+
+# ===================================================================
+# Claude Chat / CLI stream chunks
+# ===================================================================
+
+_CLAUDE_STREAM_TEXT_CHUNKS: list[dict[str, Any]] = [
+    {
+        "type": "message_start",
+        "message": {
+            "id": "msg_stream_001",
+            "type": "message",
+            "role": "assistant",
+            "model": _MODEL,
+            "content": [],
+            "stop_reason": None,
+            "usage": {"input_tokens": 10, "output_tokens": 0},
+        },
+    },
+    {
+        "type": "content_block_start",
+        "index": 0,
+        "content_block": {"type": "text", "text": ""},
+    },
+    # PLACEHOLDER_DELTAS
+]
+
+# Add text deltas
+_CLAUDE_STREAM_TEXT_CHUNKS.extend(
+    [
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": "Hello, "},
+        },
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "text_delta", "text": "world!"},
+        },
+        {"type": "content_block_stop", "index": 0},
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn"},
+            "usage": {"output_tokens": 5},
+        },
+        {"type": "message_stop"},
+    ]
+)
+
+
+_CLAUDE_STREAM_TEXT = StreamFixture(
+    chunks=_CLAUDE_STREAM_TEXT_CHUNKS,
+    expected_text="Hello, world!",
+    expected_stop_reason=StopReason.END_TURN,
+)
+
+
+# Claude stream tool call
+_CLAUDE_STREAM_TOOL_CALL = StreamFixture(
+    chunks=[
+        {
+            "type": "message_start",
+            "message": {
+                "id": "msg_stream_tc_001",
+                "type": "message",
+                "role": "assistant",
+                "model": _MODEL,
+                "content": [],
+                "stop_reason": None,
+                "usage": {"input_tokens": 20, "output_tokens": 0},
+            },
+        },
+        {
+            "type": "content_block_start",
+            "index": 0,
+            "content_block": {"type": "tool_use", "id": "tool_call_s01", "name": "get_weather"},
+        },
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "input_json_delta", "partial_json": '{"location":'},
+        },
+        {
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {"type": "input_json_delta", "partial_json": ' "Tokyo"}'},
+        },
+        {"type": "content_block_stop", "index": 0},
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "tool_use"},
+            "usage": {"output_tokens": 10},
+        },
+        {"type": "message_stop"},
+    ],
+    expected_text="",
+    expected_stop_reason=StopReason.TOOL_USE,
+)
+
+
+# ===================================================================
+# OpenAI Chat stream chunks
+# ===================================================================
+
+_OPENAI_CHAT_STREAM_TEXT = StreamFixture(
+    chunks=[
+        {
+            "id": "chatcmpl-stream-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"role": "assistant", "content": ""},
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-stream-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [{"index": 0, "delta": {"content": "Hello, "}, "finish_reason": None}],
+        },
+        {
+            "id": "chatcmpl-stream-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [{"index": 0, "delta": {"content": "world!"}, "finish_reason": None}],
+        },
+        {
+            "id": "chatcmpl-stream-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+        },
+    ],
+    expected_text="Hello, world!",
+    expected_stop_reason=StopReason.END_TURN,
+)
+
+
+# OpenAI Chat stream tool call
+_OPENAI_CHAT_STREAM_TOOL_CALL = StreamFixture(
+    chunks=[
+        {
+            "id": "chatcmpl-stream-tc-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {
+                        "role": "assistant",
+                        "content": None,
+                        "tool_calls": [
+                            {
+                                "index": 0,
+                                "id": "call_tc_001",
+                                "type": "function",
+                                "function": {"name": "get_weather", "arguments": ""},
+                            }
+                        ],
+                    },
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-stream-tc-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {
+                        "tool_calls": [{"index": 0, "function": {"arguments": '{"location":'}}]
+                    },
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-stream-tc-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [
+                {
+                    "index": 0,
+                    "delta": {"tool_calls": [{"index": 0, "function": {"arguments": ' "Tokyo"}'}}]},
+                    "finish_reason": None,
+                }
+            ],
+        },
+        {
+            "id": "chatcmpl-stream-tc-001",
+            "object": "chat.completion.chunk",
+            "model": _MODEL,
+            "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
+        },
+    ],
+    expected_text="",
+    expected_stop_reason=StopReason.TOOL_USE,
+)
+
+
+# ===================================================================
+# OpenAI CLI (Responses API) stream chunks
+# ===================================================================
+
+_OPENAI_CLI_STREAM_TEXT = StreamFixture(
+    chunks=[
+        {
+            "type": "response.created",
+            "response": {
+                "id": "resp_stream_001",
+                "object": "response",
+                "model": _MODEL,
+                "status": "in_progress",
+                "output": [],
+            },
+        },
+        {
+            "type": "response.output_item.added",
+            "output_index": 0,
+            "item": {
+                "type": "message",
+                "id": "msg_stream_001",
+                "role": "assistant",
+                "status": "in_progress",
+                "content": [],
+            },
+        },
+        {
+            "type": "response.content_part.added",
+            "output_index": 0,
+            "content_index": 0,
+            "part": {"type": "output_text", "text": ""},
+        },
+        {
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": "Hello, ",
+        },
+        {
+            "type": "response.output_text.delta",
+            "output_index": 0,
+            "content_index": 0,
+            "delta": "world!",
+        },
+        {
+            "type": "response.output_text.done",
+            "output_index": 0,
+            "content_index": 0,
+            "text": "Hello, world!",
+        },
+        {
+            "type": "response.output_item.done",
+            "output_index": 0,
+            "item": {
+                "type": "message",
+                "id": "msg_stream_001",
+                "role": "assistant",
+                "status": "completed",
+                "content": [{"type": "output_text", "text": "Hello, world!"}],
+            },
+        },
+        {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_stream_001",
+                "object": "response",
+                "model": _MODEL,
+                "status": "completed",
+                "output": [
+                    {
+                        "type": "message",
+                        "id": "msg_stream_001",
+                        "role": "assistant",
+                        "status": "completed",
+                        "content": [{"type": "output_text", "text": "Hello, world!"}],
+                    }
+                ],
+                "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
+            },
+        },
+    ],
+    expected_text="Hello, world!",
+    expected_stop_reason=StopReason.END_TURN,
+)
+
+
+# OpenAI CLI stream tool call
+_OPENAI_CLI_STREAM_TOOL_CALL = StreamFixture(
+    chunks=[
+        {
+            "type": "response.created",
+            "response": {
+                "id": "resp_stream_tc_001",
+                "object": "response",
+                "model": _MODEL,
+                "status": "in_progress",
+                "output": [],
+            },
+        },
+        {
+            "type": "response.output_item.added",
+            "output_index": 0,
+            "item": {
+                "type": "function_call",
+                "call_id": "fc_001",
+                "id": "fc_001",
+                "name": "get_weather",
+                "status": "in_progress",
+                "arguments": "",
+            },
+        },
+        {
+            "type": "response.function_call_arguments.delta",
+            "output_index": 0,
+            "item_id": "fc_001",
+            "delta": '{"location":',
+        },
+        {
+            "type": "response.function_call_arguments.delta",
+            "output_index": 0,
+            "item_id": "fc_001",
+            "delta": ' "Tokyo"}',
+        },
+        {
+            "type": "response.function_call_arguments.done",
+            "output_index": 0,
+            "item_id": "fc_001",
+            "arguments": '{"location": "Tokyo"}',
+        },
+        {
+            "type": "response.output_item.done",
+            "output_index": 0,
+            "item": {
+                "type": "function_call",
+                "call_id": "fc_001",
+                "id": "fc_001",
+                "name": "get_weather",
+                "status": "completed",
+                "arguments": '{"location": "Tokyo"}',
+            },
+        },
+        {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_stream_tc_001",
+                "object": "response",
+                "model": _MODEL,
+                "status": "completed",
+                "output": [
+                    {
+                        "type": "function_call",
+                        "call_id": "fc_001",
+                        "id": "fc_001",
+                        "name": "get_weather",
+                        "status": "completed",
+                        "arguments": '{"location": "Tokyo"}',
+                    }
+                ],
+                "usage": {"input_tokens": 20, "output_tokens": 10, "total_tokens": 30},
+            },
+        },
+    ],
+    expected_text="",
+    expected_stop_reason=StopReason.TOOL_USE,
+)
+
+
+# ===================================================================
+# Gemini Chat / CLI stream chunks
+# ===================================================================
+
+_GEMINI_STREAM_TEXT = StreamFixture(
+    chunks=[
+        {
+            "candidates": [
+                {
+                    "content": {"role": "model", "parts": [{"text": "Hello, "}]},
+                    "index": 0,
+                }
+            ],
+            "modelVersion": _MODEL,
+        },
+        {
+            "candidates": [
+                {
+                    "content": {"role": "model", "parts": [{"text": "world!"}]},
+                    "index": 0,
+                    "finishReason": "STOP",
+                }
+            ],
+            "usageMetadata": {
+                "promptTokenCount": 10,
+                "candidatesTokenCount": 5,
+                "totalTokenCount": 15,
+            },
+            "modelVersion": _MODEL,
+        },
+    ],
+    expected_text="Hello, world!",
+    expected_stop_reason=StopReason.END_TURN,
+)
+
+
+# Gemini stream tool call (Gemini emits complete tool calls atomically)
+_GEMINI_STREAM_TOOL_CALL = StreamFixture(
+    chunks=[
+        {
+            "candidates": [
+                {
+                    "content": {
+                        "role": "model",
+                        "parts": [
+                            {
+                                "functionCall": {
+                                    "name": "get_weather",
+                                    "args": {"location": "Tokyo"},
+                                }
+                            },
+                        ],
+                    },
+                    "index": 0,
+                    "finishReason": "STOP",
+                }
+            ],
+            "usageMetadata": {
+                "promptTokenCount": 20,
+                "candidatesTokenCount": 10,
+                "totalTokenCount": 30,
+            },
+            "modelVersion": _MODEL,
+        },
+    ],
+    expected_text="",
+    expected_stop_reason=StopReason.END_TURN,
+)
+
+
+# ===================================================================
+# Registry
+# ===================================================================
+
+STREAM_FIXTURES: dict[str, dict[str, StreamFixture]] = {
+    "claude:chat": {
+        "stream_text": _CLAUDE_STREAM_TEXT,
+        "stream_tool_call": _CLAUDE_STREAM_TOOL_CALL,
+    },
+    "claude:cli": {
+        "stream_text": _CLAUDE_STREAM_TEXT,
+        "stream_tool_call": _CLAUDE_STREAM_TOOL_CALL,
+    },
+    "openai:chat": {
+        "stream_text": _OPENAI_CHAT_STREAM_TEXT,
+        "stream_tool_call": _OPENAI_CHAT_STREAM_TOOL_CALL,
+    },
+    "openai:cli": {
+        "stream_text": _OPENAI_CLI_STREAM_TEXT,
+        "stream_tool_call": _OPENAI_CLI_STREAM_TOOL_CALL,
+    },
+    "gemini:chat": {
+        "stream_text": _GEMINI_STREAM_TEXT,
+        "stream_tool_call": _GEMINI_STREAM_TOOL_CALL,
+    },
+    "gemini:cli": {
+        "stream_text": _GEMINI_STREAM_TEXT,
+        "stream_tool_call": _GEMINI_STREAM_TOOL_CALL,
+    },
+}
+
+STREAM_FIXTURE_IDS = ["stream_text", "stream_tool_call"]
+STREAM_ALL_FORMATS = list(STREAM_FIXTURES.keys())

tests/core/api_format/conversion/test_claude_normalizer.py

@@ -218,7 +218,7 @@ def test_claude_stream_chunk_and_event_roundtrip_basic() -> None:
     n = ClaudeNormalizer()
     state = StreamState()
 
-    chunks = [
+    chunks: list[dict[str, Any]] = [
         {
             "type": "message_start",
             "message": {
@@ -350,6 +350,7 @@ def test_claude_system_array_format() -> None:
     internal = n.request_to_internal(req)
 
     # system 数组中的多个 text 应该用 \n\n 连接
+    assert internal.system is not None
     assert "x-anthropic-billing-header" in internal.system
     assert "You are Claude Code" in internal.system
     assert "Extract file paths" in internal.system

tests/core/api_format/conversion/test_cross_format_roundtrip.py

@@ -0,0 +1,101 @@
+"""
+Layer 3: Cross-format roundtrip tests.
+
+Verifies that converting A -> internal -> B -> internal preserves
+semantic equivalence across all format pairs.
+"""
+
+from __future__ import annotations
+
+import itertools
+
+import pytest
+
+from src.core.api_format.conversion.registry import (
+    format_conversion_registry,
+    register_default_normalizers,
+)
+
+from .fixtures.assertions import assert_internal_request_matches, assert_internal_response_matches
+from .fixtures.format_fixtures import ALL_FORMATS, FORMAT_FIXTURES, get_format_fixture
+from .fixtures.golden_internal import ALL_FIXTURE_IDS, ALL_GOLDEN_FIXTURES, is_cross_format_limited
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _ensure_normalizers_registered() -> None:
+    register_default_normalizers()
+
+
+def _cross_format_combos() -> list[tuple[str, str, str]]:
+    """Generate (source, target, fixture_id) where fixture exists for source."""
+    combos = []
+    for source, target in itertools.permutations(ALL_FORMATS, 2):
+        for fid in ALL_FIXTURE_IDS:
+            if fid in FORMAT_FIXTURES.get(source, {}):
+                combos.append((source, target, fid))
+    return combos
+
+
+_COMBOS = _cross_format_combos()
+
+
+def _combo_id(combo: tuple[str, str, str]) -> str:
+    return f"{combo[0]}->{combo[1]}:{combo[2]}"
+
+
+class TestCrossFormatRequest:
+    """source.request -> internal -> target.request -> internal: matches golden."""
+
+    @pytest.mark.parametrize("source,target,fixture_id", _COMBOS, ids=_combo_id)
+    def test_cross_format_request(self, source: str, target: str, fixture_id: str) -> None:
+        limitation = is_cross_format_limited(source, target, fixture_id, "cross_request")
+        if limitation:
+            pytest.skip(limitation)
+
+        src_norm = format_conversion_registry.get_normalizer(source)
+        tgt_norm = format_conversion_registry.get_normalizer(target)
+        assert src_norm is not None and tgt_norm is not None
+
+        fixture = get_format_fixture(source, fixture_id)
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        # source -> internal
+        internal = src_norm.request_to_internal(fixture.request)
+
+        # internal -> target native
+        target_native = tgt_norm.request_from_internal(internal)
+
+        # target native -> internal (should still match golden)
+        internal2 = tgt_norm.request_to_internal(target_native)
+
+        assert_internal_request_matches(internal2, golden.internal_request, golden.required_fields)
+
+
+class TestCrossFormatResponse:
+    """source.response -> internal -> target.response -> internal: matches golden."""
+
+    @pytest.mark.parametrize("source,target,fixture_id", _COMBOS, ids=_combo_id)
+    def test_cross_format_response(self, source: str, target: str, fixture_id: str) -> None:
+        limitation = is_cross_format_limited(source, target, fixture_id, "cross_response")
+        if limitation:
+            pytest.skip(limitation)
+
+        src_norm = format_conversion_registry.get_normalizer(source)
+        tgt_norm = format_conversion_registry.get_normalizer(target)
+        assert src_norm is not None and tgt_norm is not None
+
+        fixture = get_format_fixture(source, fixture_id)
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        # source -> internal
+        internal = src_norm.response_to_internal(fixture.response)
+
+        # internal -> target native
+        target_native = tgt_norm.response_from_internal(internal)
+
+        # target native -> internal
+        internal2 = tgt_norm.response_to_internal(target_native)
+
+        assert_internal_response_matches(
+            internal2, golden.internal_response, golden.required_fields
+        )

tests/core/api_format/conversion/test_error_conversion_full.py

@@ -0,0 +1,104 @@
+"""
+Layer 5: Error conversion tests (fixture-driven).
+
+Verifies that each normalizer correctly converts format-specific error
+responses to/from InternalError, and that error type mapping is correct.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.core.api_format.conversion.registry import (
+    format_conversion_registry,
+    register_default_normalizers,
+)
+
+from .fixtures.error_fixtures import ERROR_ALL_FORMATS, ERROR_FIXTURES
+from .fixtures.schema_validators import get_error_validator
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _ensure_normalizers_registered() -> None:
+    register_default_normalizers()
+
+
+def _error_combos() -> list[tuple[str, str]]:
+    combos = []
+    for fmt in ERROR_ALL_FORMATS:
+        for eid in ERROR_FIXTURES.get(fmt, {}):
+            combos.append((fmt, eid))
+    return combos
+
+
+_COMBOS = _error_combos()
+
+
+class TestErrorToInternal:
+    """Verify format-specific error -> InternalError."""
+
+    @pytest.mark.parametrize("format_id,error_id", _COMBOS)
+    def test_error_to_internal(self, format_id: str, error_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = ERROR_FIXTURES[format_id][error_id]
+        internal = normalizer.error_to_internal(fixture.error_response)
+
+        assert (
+            internal.type == fixture.expected_type
+        ), f"error type mismatch: {internal.type} != {fixture.expected_type}"
+        assert (
+            internal.message == fixture.expected_message
+        ), f"error message mismatch: {internal.message!r} != {fixture.expected_message!r}"
+
+
+class TestErrorRoundtrip:
+    """Verify error -> internal -> error -> internal preserves type and message."""
+
+    @pytest.mark.parametrize("format_id,error_id", _COMBOS)
+    def test_error_roundtrip(self, format_id: str, error_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = ERROR_FIXTURES[format_id][error_id]
+
+        # First pass
+        internal1 = normalizer.error_to_internal(fixture.error_response)
+
+        # Reconstruct
+        reconstructed = normalizer.error_from_internal(internal1)
+
+        # Second pass
+        internal2 = normalizer.error_to_internal(reconstructed)
+
+        assert (
+            internal1.type == internal2.type
+        ), f"error type changed after roundtrip: {internal1.type} -> {internal2.type}"
+        assert (
+            internal1.message == internal2.message
+        ), f"error message changed after roundtrip: {internal1.message!r} -> {internal2.message!r}"
+
+
+class TestErrorFromInternalSchema:
+    """Verify InternalError -> format-specific error conforms to API schema."""
+
+    @pytest.mark.parametrize("format_id,error_id", _COMBOS)
+    def test_error_from_internal_schema(self, format_id: str, error_id: str) -> None:
+        validator = get_error_validator(format_id)
+        if validator is None:
+            pytest.skip(f"No error schema validator for {format_id}")
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = ERROR_FIXTURES[format_id][error_id]
+        internal = normalizer.error_to_internal(fixture.error_response)
+        reconstructed = normalizer.error_from_internal(internal)
+
+        errors = validator(reconstructed)
+        assert (
+            not errors
+        ), f"Error schema validation failed for {format_id} ({error_id}):\n" + "\n".join(
+            f"  - {e}" for e in errors
+        )

tests/core/api_format/conversion/test_gemini_normalizer.py

@@ -195,7 +195,7 @@ def test_gemini_stream_chunk_and_event_roundtrip_basic() -> None:
     n = GeminiNormalizer()
     state = StreamState()
 
-    chunks = [
+    chunks: list[dict[str, Any]] = [
         {
             "candidates": [{"content": {"parts": [{"text": "Hel"}], "role": "model"}, "index": 0}],
             "modelVersion": "gemini-1.5",

tests/core/api_format/conversion/test_normalizer_roundtrip.py

@@ -0,0 +1,96 @@
+"""
+Layer 2: Normalizer roundtrip tests.
+
+Verifies that converting A -> internal -> A preserves semantic equivalence.
+"""
+
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from src.core.api_format.conversion.registry import (
+    format_conversion_registry,
+    register_default_normalizers,
+)
+
+from .fixtures.assertions import assert_internal_requests_equivalent
+from .fixtures.format_fixtures import ALL_FORMATS, FORMAT_FIXTURES, get_format_fixture
+from .fixtures.golden_internal import ALL_FIXTURE_IDS, KNOWN_LIMITATIONS
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _ensure_normalizers_registered() -> None:
+    register_default_normalizers()
+
+
+def _available_combos() -> list[tuple[str, str]]:
+    combos = []
+    for fmt in ALL_FORMATS:
+        for fid in ALL_FIXTURE_IDS:
+            if fid in FORMAT_FIXTURES.get(fmt, {}):
+                combos.append((fmt, fid))
+    return combos
+
+
+_COMBOS = _available_combos()
+
+
+class TestRequestRoundtrip:
+    """A.request -> internal -> A.request -> internal: two internals should match."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_request_roundtrip(self, format_id: str, fixture_id: str) -> None:
+        limitation = KNOWN_LIMITATIONS.get((format_id, fixture_id, "roundtrip"))
+        if limitation:
+            pytest.skip(limitation)
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = get_format_fixture(format_id, fixture_id)
+
+        # First pass: native -> internal
+        internal1 = normalizer.request_to_internal(fixture.request)
+
+        # Reconstruct: internal -> native
+        # Deep copy because some normalizers mutate the input (e.g. _coerce_claude_message_sequence)
+        reconstructed = normalizer.request_from_internal(copy.deepcopy(internal1))
+
+        # Second pass: native -> internal
+        internal2 = normalizer.request_to_internal(reconstructed)
+
+        # The two internal representations should be semantically equivalent
+        assert_internal_requests_equivalent(internal1, internal2, lossy_fields=fixture.lossy_fields)
+
+
+class TestResponseRoundtrip:
+    """A.response -> internal -> A.response -> internal: two internals should match."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_response_roundtrip(self, format_id: str, fixture_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = get_format_fixture(format_id, fixture_id)
+
+        # First pass
+        internal1 = normalizer.response_to_internal(fixture.response)
+
+        # Reconstruct
+        reconstructed = normalizer.response_from_internal(internal1)
+
+        # Second pass
+        internal2 = normalizer.response_to_internal(reconstructed)
+
+        # Compare content blocks (the core semantic payload)
+        from .fixtures.assertions import assert_content_blocks_match
+
+        assert_content_blocks_match(
+            internal1.content, internal2.content, context="response roundtrip"
+        )
+        # Stop reason should be preserved
+        assert (
+            internal1.stop_reason == internal2.stop_reason
+        ), f"stop_reason changed after roundtrip: {internal1.stop_reason} -> {internal2.stop_reason}"

tests/core/api_format/conversion/test_normalizer_to_internal.py

@@ -0,0 +1,159 @@
+"""
+Layer 1: Normalizer to_internal / from_internal tests.
+
+Verifies that each normalizer correctly converts format-specific
+requests/responses to/from the canonical internal representation.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.core.api_format.conversion.registry import (
+    format_conversion_registry,
+    register_default_normalizers,
+)
+
+from .fixtures.assertions import (
+    assert_internal_request_matches,
+    assert_internal_response_matches,
+)
+from .fixtures.format_fixtures import ALL_FORMATS, FORMAT_FIXTURES, get_format_fixture
+from .fixtures.golden_internal import ALL_FIXTURE_IDS, ALL_GOLDEN_FIXTURES, KNOWN_LIMITATIONS
+from .fixtures.schema_validators import (
+    get_request_validator,
+    get_response_validator,
+)
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _ensure_normalizers_registered() -> None:
+    """Ensure all normalizers are registered before tests run."""
+    register_default_normalizers()
+
+
+def _available_combos() -> list[tuple[str, str]]:
+    """Generate (format_id, fixture_id) pairs where fixture exists for format."""
+    combos = []
+    for fmt in ALL_FORMATS:
+        for fid in ALL_FIXTURE_IDS:
+            if fid in FORMAT_FIXTURES.get(fmt, {}):
+                combos.append((fmt, fid))
+    return combos
+
+
+_COMBOS = _available_combos()
+
+
+class TestRequestToInternal:
+    """Verify format-specific request -> InternalRequest."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_request_to_internal(self, format_id: str, fixture_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None, f"No normalizer for {format_id}"
+
+        fixture = get_format_fixture(format_id, fixture_id)
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        internal = normalizer.request_to_internal(fixture.request)
+
+        # Exclude lossy fields from comparison
+        effective_required = golden.required_fields - fixture.lossy_fields
+        assert_internal_request_matches(internal, golden.internal_request, effective_required)
+
+
+class TestResponseToInternal:
+    """Verify format-specific response -> InternalResponse."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_response_to_internal(self, format_id: str, fixture_id: str) -> None:
+        limitation = KNOWN_LIMITATIONS.get((format_id, fixture_id, "to_internal_response"))
+        if limitation:
+            pytest.skip(limitation)
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None, f"No normalizer for {format_id}"
+
+        fixture = get_format_fixture(format_id, fixture_id)
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        internal = normalizer.response_to_internal(fixture.response)
+
+        assert_internal_response_matches(internal, golden.internal_response, golden.required_fields)
+
+
+class TestRequestFromInternal:
+    """Verify InternalRequest -> format-specific request produces valid output."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_request_from_internal(self, format_id: str, fixture_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None, f"No normalizer for {format_id}"
+
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        result = normalizer.request_from_internal(golden.internal_request)
+
+        # The result should be a valid dict that can be parsed back
+        assert isinstance(result, dict), f"Expected dict, got {type(result)}"
+        # Model should be preserved
+        model_key = "model"
+        if format_id.startswith("gemini"):
+            # Gemini doesn't put model in request body
+            pass
+        else:
+            assert result.get(model_key) == golden.internal_request.model
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_request_from_internal_schema(self, format_id: str, fixture_id: str) -> None:
+        """Validate output structure conforms to the target API schema."""
+        validator = get_request_validator(format_id)
+        if validator is None:
+            pytest.skip(f"No request schema validator for {format_id}")
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+        result = normalizer.request_from_internal(golden.internal_request)
+        errors = validator(result)
+        assert (
+            not errors
+        ), f"Schema validation failed for {format_id} request ({fixture_id}):\n" + "\n".join(
+            f"  - {e}" for e in errors
+        )
+
+
+class TestResponseFromInternal:
+    """Verify InternalResponse -> format-specific response produces valid output."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_response_from_internal(self, format_id: str, fixture_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None, f"No normalizer for {format_id}"
+
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+
+        result = normalizer.response_from_internal(golden.internal_response)
+
+        assert isinstance(result, dict), f"Expected dict, got {type(result)}"
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_response_from_internal_schema(self, format_id: str, fixture_id: str) -> None:
+        """Validate output structure conforms to the target API schema."""
+        validator = get_response_validator(format_id)
+        if validator is None:
+            pytest.skip(f"No response schema validator for {format_id}")
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        golden = ALL_GOLDEN_FIXTURES[fixture_id]
+        result = normalizer.response_from_internal(golden.internal_response)
+        errors = validator(result)
+        assert (
+            not errors
+        ), f"Schema validation failed for {format_id} response ({fixture_id}):\n" + "\n".join(
+            f"  - {e}" for e in errors
+        )

tests/core/api_format/conversion/test_openai_normalizer.py

@@ -260,7 +260,7 @@ def test_openai_stream_chunk_and_event_roundtrip_basic() -> None:
     n = OpenAINormalizer()
     state = StreamState()
 
-    chunks = [
+    chunks: list[dict[str, Any]] = [
         {
             "id": "chatcmpl_stream_1",
             "object": "chat.completion.chunk",

tests/core/api_format/conversion/test_stream_conversion.py

@@ -0,0 +1,110 @@
+"""
+Layer 4: Stream conversion tests.
+
+Verifies that each normalizer correctly converts format-specific stream
+chunks to/from internal stream events.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.core.api_format.conversion.registry import (
+    format_conversion_registry,
+    register_default_normalizers,
+)
+from src.core.api_format.conversion.stream_state import StreamState
+
+from .fixtures.assertions import (
+    assert_stream_has_tool_call,
+    assert_stream_stop_reason_matches,
+    assert_stream_text_matches,
+)
+from .fixtures.schema_validators import get_stream_chunk_validator
+from .fixtures.stream_fixtures import (
+    STREAM_ALL_FORMATS,
+    STREAM_FIXTURE_IDS,
+    STREAM_FIXTURES,
+)
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _ensure_normalizers_registered() -> None:
+    register_default_normalizers()
+
+
+def _stream_combos() -> list[tuple[str, str]]:
+    combos = []
+    for fmt in STREAM_ALL_FORMATS:
+        for fid in STREAM_FIXTURE_IDS:
+            if fid in STREAM_FIXTURES.get(fmt, {}):
+                combos.append((fmt, fid))
+    return combos
+
+
+_COMBOS = _stream_combos()
+
+
+class TestStreamToInternal:
+    """Verify format-specific stream chunks -> InternalStreamEvent sequence."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_stream_to_internal(self, format_id: str, fixture_id: str) -> None:
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = STREAM_FIXTURES[format_id][fixture_id]
+        state = StreamState(model=fixture.chunks[0].get("model", ""))
+
+        all_events = []
+        for chunk in fixture.chunks:
+            events = normalizer.stream_chunk_to_internal(chunk, state)
+            all_events.extend(events)
+
+        assert_stream_text_matches(all_events, fixture.expected_text)
+        assert_stream_stop_reason_matches(all_events, fixture.expected_stop_reason)
+
+        if fixture_id == "stream_tool_call":
+            assert_stream_has_tool_call(all_events, "get_weather")
+
+
+class TestStreamFromInternalSchema:
+    """Verify internal events -> format-specific chunks conform to API schema."""
+
+    @pytest.mark.parametrize("format_id,fixture_id", _COMBOS)
+    def test_stream_roundtrip_schema(self, format_id: str, fixture_id: str) -> None:
+        """Parse chunks -> internal events -> reconstruct chunks, validate schema."""
+        validator = get_stream_chunk_validator(format_id)
+        if validator is None:
+            pytest.skip(f"No stream chunk schema validator for {format_id}")
+
+        normalizer = format_conversion_registry.get_normalizer(format_id)
+        assert normalizer is not None
+
+        fixture = STREAM_FIXTURES[format_id][fixture_id]
+
+        # Phase 1: chunks -> internal events
+        in_state = StreamState(model=fixture.chunks[0].get("model", ""))
+        all_events = []
+        for chunk in fixture.chunks:
+            events = normalizer.stream_chunk_to_internal(chunk, in_state)
+            all_events.extend(events)
+
+        # Phase 2: internal events -> output chunks, validate each
+        out_state = StreamState(
+            message_id=in_state.message_id or "chatcmpl-test",
+            model=in_state.model or "test-model",
+        )
+        all_errors: list[str] = []
+        for event in all_events:
+            output_chunks = normalizer.stream_event_from_internal(event, out_state)
+            for out_chunk in output_chunks:
+                errors = validator(out_chunk)
+                if errors:
+                    all_errors.extend(f"[{type(event).__name__}] {e}" for e in errors)
+
+        assert (
+            not all_errors
+        ), f"Stream schema validation failed for {format_id} ({fixture_id}):\n" + "\n".join(
+            f"  - {e}" for e in all_errors
+        )