refactor(vendor-channels): 收敛跨供应商兼容处理至源→目标通道并引入内容感知源推断

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ## [Unreleased]
 
+- refactor(vendor-channels): 彻底收敛跨供应商兼容性逻辑——删除 `server/request_normalizer.py` 入口通用规范化层，将 `srvtoolu_*` ID 重写、`server_tool_use_delta` 私有块剥离全部迁入源→目标绑定通道（`prepare_zhipu_to_anthropic`、`prepare_zhipu_to_copilot`）；新增 `infer_source_vendor_from_body` 内容感知源推断，在无会话状态的首次请求场景下兜底识别源供应商；`_RouteExecutor._determine_source_vendor` 扩充为三级优先级（failed_tier → session_state → body inference），确保未注册转换对不触发任何清洗；
+- refactor(count-tokens): `/v1/messages/count_tokens` 端点移除无条件 `strip_thinking_blocks` 过度防御，改为基于 `infer_source_vendor_from_body` + `get_transition_channel` 的按需通道清洗，语义与 `/v1/messages` 对齐；
 - fix(request-normalizer): 重设计 zhipu→anthropic 跨供应商 tool_use/tool_result 配对修复——以单遍自包含 `enforce_anthropic_tool_pairing` 替代原有多步串联管线（剥离→重定位→孤儿修复），消除步骤间隐式依赖导致的孤儿 tool_use 漏修问题，彻底根治 `tool_use ids were found without tool_result blocks` 400 异常;
 - refactor(vendor-channels): 将供应商转换通道从「目标 vendor 专属」重构为「源→目标绑定」模型——注册表键从 `target_vendor` 改为 `(source, target)` 二元组，通道函数从 `prepare_for_X` 重命名为 `prepare_X_to_Y`，触发逻辑从 `_needs_vendor_channel` 替换为 `_determine_source_vendor`（基于请求内 `failed_tier_name` 和会话历史推断源 vendor），未注册的转换对（如 anthropic→zhipu）不触发任何通道;
 - feat(vendor-channels): 新增 zhipu→anthropic、zhipu→copilot、copilot→zhipu 三条源→目标绑定转换通道，在跨供应商故障转移时自动清理源 vendor 产物（thinking 块、cache_control 字段、thinking 参数、tool_use/tool_result 配对），消除 `likely format incompatibility (400 + tool_results)` 错误;

docs/arch/testing.md

@@ -65,6 +65,7 @@
 | `test_convert_sse.py`              | Gemini SSE→Anthropic SSE 流适配（单/多 chunk、各 finishReason、边界情况） |
 | `test_copilot_convert_request.py`  | Anthropic→OpenAI 请求格式转换                                             |
 | `test_copilot_convert_response.py` | OpenAI→Anthropic 响应格式转换                                             |
+| `test_vendor_channels.py`          | 源→目标通道：zhipu/copilot 兼容性清洗、tool_use 配对、内容感知源推断      |
 
 ### 2.5 数据模型（model）
 
@@ -95,7 +96,6 @@
 | 测试文件                     | 覆盖范围                                                |
 | ---------------------------- | ------------------------------------------------------- |
 | `test_app_routes.py`         | FastAPI 路由端点测试                                    |
-| `test_request_normalizer.py` | 请求标准化：私有块清洗、tool_use_id 重写、fatal_reasons |
 | `test_cli_usage.py`          | CLI 用量查询命令                                        |
 | `test_banner.py`             | CLI Banner 显示                                         |
 | `test_logging_dual_write.py` | 日志双写机制                                            |

docs/framework.md

@@ -69,7 +69,6 @@ graph TD
         App["<code>app.py</code><br/>应用工厂 + lifespan"]
         Routes["<code>routes.py</code><br/>路由注册"]
         Factory["<code>factory.py</code><br/>Vendor/Tier 构建工厂"]
-        Normalizer["<code>request_normalizer.py</code><br/>请求标准化"]
         Dashboard["<code>dashboard.py</code><br/>状态面板"]
     end
 
@@ -179,15 +178,8 @@ graph TD
 ```mermaid
 flowchart TD
     Client["Client POST /v1/messages"] --> Server["server.routes.messages()"]
-
-    subgraph Normalize ["请求标准化"]
-        Body["body = await request.json()"]
-        Norm["normalize_anthropic_request(body)<br/>清洗私有块 + 重写 tool_use_id"]
-        Body --> Norm
-    end
-
-    Server --> Normalize
-    Norm --> RouteType{"stream?"}
+    Server --> Body["body = await request.json()"]
+    Body --> RouteType{"stream?"}
 
     RouteType -- "true" --> StreamRoute["route_stream()"]
     RouteType -- "false" --> MsgRoute["route_message()"]
@@ -401,7 +393,6 @@ flowchart TD
 | [`app.py`](../src/coding/proxy/server/app.py)                               | FastAPI 应用工厂 `create_app()` + `lifespan` 生命周期管理 |
 | [`factory.py`](../src/coding/proxy/server/factory.py)                       | Vendor/Tier 构建工厂 + 凭证解析                           |
 | [`routes.py`](../src/coding/proxy/server/routes.py)                         | 路由端点按职责分组注册                                    |
-| [`request_normalizer.py`](../src/coding/proxy/server/request_normalizer.py) | 入站请求标准化（清洗供应商私有块）                        |
 | [`responses.py`](../src/coding/proxy/server/responses.py)                   | 响应辅助工具（JSON error / stream error 构建）            |
 | [`dashboard.py`](../src/coding/proxy/server/dashboard.py)                   | 状态面板（Web Dashboard）                                 |
 

src/coding/proxy/convert/vendor_channels.py

@@ -16,13 +16,25 @@
 
 import copy
 import logging
+import re
 from collections.abc import Callable
 from typing import Any
 
 logger = logging.getLogger(__name__)
 
 _THINKING_BLOCK_TYPES = {"thinking", "redacted_thinking"}
 
+# ── Anthropic 工具块 ID 规范 ───────────────────────────────────
+_ANTHROPIC_TOOL_USE_ID_RE = re.compile(r"^toolu_[A-Za-z0-9_]+$")
+_ANTHROPIC_SERVER_TOOL_USE_ID_RE = re.compile(r"^srvtoolu_[A-Za-z0-9_]+$")
+
+# Zhipu 流式响应中出现的非标准供应商私有 content block 类型.
+# Anthropic API 拒绝这些块，需要在跨 vendor 请求体中剥离.
+_ZHIPU_VENDOR_BLOCK_TYPES = {"server_tool_use_delta"}
+
+# Zhipu 内联输出非标准 content block 类型的标识（用于源供应商推断）.
+_ZHIPU_SERVER_TOOL_USE_TYPES = {"server_tool_use", "server_tool_use_delta"}
+
 # ── 转换通道注册表 ─────────────────────────────────────────────
 # (source_vendor, target_vendor) → (body) → (prepared_body, adaptations)
 VENDOR_TRANSITIONS: dict[
@@ -250,6 +262,138 @@ def _strip_cache_control(body: dict[str, Any]) -> int:
     return removed
 
 
+def _remove_vendor_blocks(body: dict[str, Any], block_types: set[str]) -> int:
+    """从 messages[].content[] 中就地移除指定 type 的内容块.
+
+    用于剥离 vendor 私有 content block 类型（如 zhipu 的 ``server_tool_use_delta``），
+    Anthropic API 会拒绝这些非标准块。
+
+    Returns:
+        被移除的块数量。
+    """
+    removed = 0
+    for message in body.get("messages", []):
+        if not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        new_content: list[Any] = []
+        for block in content:
+            if isinstance(block, dict) and block.get("type") in block_types:
+                removed += 1
+                continue
+            new_content.append(block)
+        if removed:
+            message["content"] = new_content
+    return removed
+
+
+def _rewrite_srvtoolu_ids(body: dict[str, Any]) -> tuple[int, dict[str, str]]:
+    """将 zhipu 的 ``server_tool_use`` + ``srvtoolu_*`` ID 改写为标准 Anthropic 形式.
+
+    Anthropic API 要求 tool_use 类型与 ``toolu_*`` 格式的 ID。Zhipu 的
+    ``server_tool_use`` + ``srvtoolu_*`` 在上游 Anthropic 兼容端点可用，但无法
+    透传至其他供应商；同时还需重写紧随其后 user 消息中 ``tool_result.tool_use_id``
+    引用，保持配对关系。
+
+    Returns:
+        (rewritten_count, id_map) — 重写次数与 {原 ID: 新 ID} 映射。
+    """
+    id_map: dict[str, str] = {}
+    counter = 0
+
+    def next_id() -> str:
+        nonlocal counter
+        counter += 1
+        return f"toolu_normalized_{counter}"
+
+    for message in body.get("messages", []):
+        if not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        role = message.get("role")
+        for block in content:
+            if not isinstance(block, dict):
+                continue
+            block_type = block.get("type")
+            block_id = block.get("id")
+
+            # Case A: assistant 消息里的 server_tool_use / srvtoolu_* → 改写
+            if role == "assistant" and block_type in {"tool_use", "server_tool_use"}:
+                if isinstance(block_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
+                    block_id
+                ):
+                    new_id = next_id()
+                    id_map[block_id] = new_id
+                    block["id"] = new_id
+                    block["type"] = "tool_use"
+                elif (
+                    isinstance(block_id, str)
+                    and block_id
+                    and not _ANTHROPIC_TOOL_USE_ID_RE.match(block_id)
+                    and block.get("name")
+                ):
+                    # 非标准 ID（非 toolu_ / srvtoolu_），且具备 name 可改写
+                    new_id = next_id()
+                    id_map[block_id] = new_id
+                    block["id"] = new_id
+                    block["type"] = "tool_use"
+                elif block_type == "server_tool_use" and isinstance(block_id, str):
+                    # 兜底: 类型是 server_tool_use 但 ID 已是标准 toolu_ 形式，仅纠正类型
+                    block["type"] = "tool_use"
+
+            # Case B: user 消息里的 tool_result.tool_use_id 同步重写
+            if block_type == "tool_result":
+                tool_use_id = block.get("tool_use_id")
+                if isinstance(tool_use_id, str) and tool_use_id in id_map:
+                    block["tool_use_id"] = id_map[tool_use_id]
+
+    return len(id_map), id_map
+
+
+def infer_source_vendor_from_body(body: dict[str, Any]) -> str | None:
+    """从请求 body 内容推断源供应商（仅在无会话上下文时作为兜底）.
+
+    启发式（按置信度排序）:
+    - 出现 ``srvtoolu_*`` 格式的 ``tool_use.id`` → zhipu
+    - 出现 ``server_tool_use`` / ``server_tool_use_delta`` 类型的 content block → zhipu
+
+    原则: 只读扫描不修改 body；无匹配返回 None（视作纯净无需跨供应商清洗）。
+
+    Args:
+        body: Anthropic Messages 请求体。
+
+    Returns:
+        推断的源供应商名称（当前仅支持 ``"zhipu"``），无法推断返回 None。
+    """
+    for message in body.get("messages", []):
+        if not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        for block in content:
+            if not isinstance(block, dict):
+                continue
+            block_type = block.get("type")
+            if block_type in _ZHIPU_SERVER_TOOL_USE_TYPES:
+                return "zhipu"
+            block_id = block.get("id")
+            if isinstance(block_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
+                block_id
+            ):
+                return "zhipu"
+            tool_use_id = block.get("tool_use_id")
+            if isinstance(tool_use_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
+                tool_use_id
+            ):
+                return "zhipu"
+    return None
+
+
 # ── copilot → zhipu 转换通道 ─────────────────────────────────────
 
 
@@ -316,17 +460,27 @@ def prepare_zhipu_to_copilot(
     prepared = copy.deepcopy(body)
     adaptations: list[str] = []
 
-    # Step 1: 剥离 thinking/redacted_thinking 块
+    # Step 1: 剥离 zhipu 私有 content block 类型
+    removed_vendor_blocks = _remove_vendor_blocks(prepared, _ZHIPU_VENDOR_BLOCK_TYPES)
+    if removed_vendor_blocks:
+        adaptations.append(f"removed_{removed_vendor_blocks}_zhipu_vendor_blocks")
+
+    # Step 2: 改写 srvtoolu_* ID 与 server_tool_use 类型
+    rewritten, _ = _rewrite_srvtoolu_ids(prepared)
+    if rewritten:
+        adaptations.append(f"rewritten_{rewritten}_srvtoolu_ids")
+
+    # Step 3: 剥离 thinking/redacted_thinking 块
     stripped = strip_thinking_blocks(prepared)
     if stripped:
         adaptations.append(f"stripped_{stripped}_thinking_blocks")
 
-    # Step 2: 移除 cache_control 字段
+    # Step 4: 移除 cache_control 字段
     removed_cc = _strip_cache_control(prepared)
     if removed_cc:
         adaptations.append(f"removed_{removed_cc}_cache_control_fields")
 
-    # Step 3: 强制 tool_use/tool_result 配对
+    # Step 5: 强制 tool_use/tool_result 配对
     pairing_fixes = enforce_anthropic_tool_pairing(prepared.get("messages", []))
     if pairing_fixes:
         adaptations.extend(pairing_fixes)
@@ -343,27 +497,41 @@ def prepare_zhipu_to_anthropic(
     """zhipu → anthropic 转换: 清理 zhipu 产物以适配 Anthropic API.
 
     Anthropic API 要求:
+    - tool_use 类型与 ``toolu_*`` 格式 ID（zhipu 的 ``server_tool_use``/``srvtoolu_*`` 不兼容）
     - 每个 tool_use 必须在紧随的 user 消息中有对应 tool_result
     - thinking blocks 的 signature 必须是 Anthropic 签发（zhipu 签发的无效）
+    - 不接受 ``server_tool_use_delta`` 等 zhipu 私有流式块类型
 
-    此通道执行两项变换:
-    1. enforce_anthropic_tool_pairing: 单遍正向扫描修复配对
-    2. strip_thinking_blocks: 移除非 Anthropic 签发的 thinking 块
+    此通道按顺序执行:
+    1. 剥离 zhipu 私有 block 类型（``server_tool_use_delta``）
+    2. 改写 ``srvtoolu_*`` ID 与 ``server_tool_use`` 类型为标准 Anthropic 形式
+    3. 强制 tool_use/tool_result 配对（单遍正向扫描）
+    4. 剥离 thinking blocks（signature 无效）
 
-    两项变换均为幂等操作，安全地在已清理的请求体上重复执行。
+    所有变换均为幂等操作，安全地在已清理的请求体上重复执行。
 
     Returns:
         (prepared_body, adaptations) — adaptations 为应用的变换描述列表。
     """
     prepared = copy.deepcopy(body)
     adaptations: list[str] = []
 
-    # Step 1: 强制 tool_use/tool_result 配对
+    # Step 1: 剥离 zhipu 私有 content block 类型（如 server_tool_use_delta）
+    removed_vendor_blocks = _remove_vendor_blocks(prepared, _ZHIPU_VENDOR_BLOCK_TYPES)
+    if removed_vendor_blocks:
+        adaptations.append(f"removed_{removed_vendor_blocks}_zhipu_vendor_blocks")
+
+    # Step 2: 改写 srvtoolu_* ID 与 server_tool_use 类型
+    rewritten, _ = _rewrite_srvtoolu_ids(prepared)
+    if rewritten:
+        adaptations.append(f"rewritten_{rewritten}_srvtoolu_ids")
+
+    # Step 3: 强制 tool_use/tool_result 配对
     pairing_fixes = enforce_anthropic_tool_pairing(prepared.get("messages", []))
     if pairing_fixes:
         adaptations.extend(pairing_fixes)
 
-    # Step 2: 剥离 thinking blocks（zhipu signature 无效）
+    # Step 4: 剥离 thinking blocks（zhipu signature 无效）
     stripped = strip_thinking_blocks(prepared)
     if stripped:
         adaptations.append(f"stripped_{stripped}_thinking_blocks")

src/coding/proxy/routing/executor.py

@@ -257,26 +257,41 @@ def _determine_source_vendor(
         target_name: str,
         failed_tier_name: str | None,
         session_record: Any,
+        body: dict[str, Any] | None = None,
     ) -> str | None:
         """确定跨供应商转换的源 vendor.
 
         Priority 1: failed_tier_name（请求内故障转移，最可靠）。
         Priority 2: session_record.provider_state 中有已注册转换的 vendor（跨请求）。
+        Priority 3: 从 body 内容推断（兜底首次请求无会话状态场景）。
         """
+        from ..convert.vendor_channels import (
+            get_transition_channel,
+            infer_source_vendor_from_body,
+        )
+
         # 请求内：刚失败的 tier 就是源
         if failed_tier_name and failed_tier_name != target_name:
             return failed_tier_name
 
         # 跨请求：从会话历史找有注册转换的源
         if session_record is not None and session_record.provider_state:
-            from ..convert.vendor_channels import get_transition_channel
-
             for source in session_record.provider_state:
                 if source != target_name and get_transition_channel(
                     source, target_name
                 ):
                     return source
 
+        # 首次请求兜底：从 body 内容推断（识别 zhipu 产物等）
+        if body is not None:
+            inferred = infer_source_vendor_from_body(body)
+            if (
+                inferred
+                and inferred != target_name
+                and get_transition_channel(inferred, target_name)
+            ):
+                return inferred
+
         return None
 
     async def execute_stream(
@@ -315,7 +330,7 @@ async def execute_stream(
 
             try:
                 source_vendor = _RouteExecutor._determine_source_vendor(
-                    tier.name, failed_tier_name, session_record
+                    tier.name, failed_tier_name, session_record, body
                 )
                 body_for_tier = self._prepare_body_for_tier(body, tier, source_vendor)
                 async for chunk in tier.vendor.send_message_stream(
@@ -481,7 +496,7 @@ async def execute_message(
 
             try:
                 source_vendor = _RouteExecutor._determine_source_vendor(
-                    tier.name, failed_tier_name, session_record
+                    tier.name, failed_tier_name, session_record, body
                 )
                 body_for_tier = self._prepare_body_for_tier(body, tier, source_vendor)
                 resp = await tier.vendor.send_message(body_for_tier, headers)