refactor: 按策略拆分上游请求准备与多协议非流式格式化

docs/refactor-request-response-boundaries.md

@@ -0,0 +1,81 @@
+# CodeBuddy Proxy: request/response boundary refactor
+
+## Scope and constraints
+
+This is a structural refactor, **not** a protocol or behavior change.
+It follows the already-added domestic `developer -> system` compatibility
+implementation. Existing routes, billing, authentication, retry behavior,
+stream parsing, OpenAI/Anthropic/Responses payload contracts, response status
+handling, and diagnostics remain in place.
+
+The goal is to reduce the responsibility of the large FastAPI entry point
+(`src/codebuddy_proxy/__main__.py`) without moving the high-risk streaming
+state machine.
+
+## Responsibilities
+
+| Module | Responsibility | Pattern |
+| --- | --- | --- |
+| `__main__.py` | HTTP routes, authentication check, stream/non-stream dispatch, lifecycle, error handling | Controller / orchestration |
+| `upstream_request.py` | Prepare the upstream URL, normalized request body, stream flags, authentication and gateway headers | Builder + composable policies |
+| `DomesticDeveloperRolePolicy` | Transform unsupported `developer` messages only for the verified domestic gateway | Message strategy |
+| `InternationalConversationHeaderPolicy` | Inject `X-Conversation-*` headers for the same URLs as before, with nonfatal fallback | Header strategy |
+| `protocol_response.py` | Convert aggregated OpenAI-shaped output into client-compatible objects | Strategy registry |
+| `billing_proxy.py` | Independent authenticated billing endpoints | Existing service module |
+
+## Request preparation order (unchanged)
+
+1. `state.ensure_auth()` and log the original client request summary.
+2. Shallow-copy the body and normalize `tool_choice`.
+3. Resolve the configured upstream chat URL.
+4. Apply domestic developer-role compatibility when applicable.
+5. Run optional `--desensitize` and harness compaction.
+6. Force upstream `stream=True` while retaining the incoming stream preference separately, and default `stream_options`.
+7. Build original authentication / IDE / SSE request headers.
+8. Apply international conversation headers where the existing URL guard matches.
+9. Emit original IDE diagnostic when enabled.
+10. Leave the existing streaming preflight or aggregate code to send the prepared body unchanged.
+
+The `codebuddy.ai` substring check for international headers is deliberately
+preserved as an observable compatibility detail. Tightening hostname matching
+would be a separate behavior-changing fix and is out of scope.
+
+## Nonstream protocol formatting (unchanged)
+
+The aggregation layer continues producing the same OpenAI-style completion.
+`convert_nonstream` is retained as a public/local wrapper, delegating to:
+
+- `OpenAIFormatter`: returns the *same* response object.
+- `AnthropicFormatter`: produces `msg_*`, text/tool-use blocks, stop reason,
+  and mapped usage counters.
+- `ResponsesFormatter`: produces `resp_*`, output-text/function-call blocks,
+  and completion status.
+- Unknown protocol: same identity fallback as before.
+
+The formatter registry accepts custom implementations without editing route
+or upstream transport code. ID generation and the existing clock callback
+remain injectable and produce the same output structures.
+
+## Regression safeguards
+
+- `test_upstream_request_pipeline.py` checks preparation order, original
+  input immutability, stream settings, token headers, tool choice, gateway
+  policy scope, nonblocking international-header failures, and policy injection.
+- `test_protocol_response_strategies.py` checks exact legacy wire payload
+  structures for all supported protocol formatters, malformed tool arguments,
+  empty choices, identity fallback, and custom formatters.
+- `test_intl_conversation_headers.py` replaces a brittle source-inspection
+  assertion with functional verification of emitted headers.
+- Existing protocol, streaming, billing, developer-role, and retry tests are
+  run unchanged.
+
+The known Windows-only `stat.S_IMODE(...) == 0o600` test assertion fails
+on NTFS as it did before this refactor. The rest of the suite is expected
+to pass with that one environment-specific test deselected.
+
+## Next phases intentionally excluded
+
+Streaming SSE parsing / DSML, the model catalog cache, and login/session
+persistence are more stateful and require additional characterization tests.
+They are deliberately left for separate narrowly scoped changes instead of
+being rewritten alongside the request/response boundary.

src/codebuddy_proxy/__main__.py

@@ -48,7 +48,8 @@
 )
 from codebuddy_proxy.codebuddy_client_demo import CodeBuddyClient, CodeBuddyError
 from codebuddy_proxy.billing_proxy import forward_billing
-from codebuddy_proxy.role_compat import adapt_developer_roles, uses_legacy_chat_roles
+from codebuddy_proxy.upstream_request import ChatRequestPreparer
+from codebuddy_proxy.protocol_response import format_nonstream_response
 
 # 尝试导入高级功能模块（可选）
 try:
@@ -1035,61 +1036,23 @@ async def forward_chat(
 
     diagnostic("upstream_request", protocol=protocol, **body_summary(body))
 
-    stream = bool(body.get("stream"))
-    upstream_body = dict(body)
-
-    # 归一化 tool_choice：object 形式 → 函数名字符串（上游只接受 string）
-    if "tool_choice" in upstream_body:
-        upstream_body["tool_choice"] = _normalize_tool_choice(upstream_body["tool_choice"])
-
-    url = state.client.build_url(state.client.profile.chat_path)
-
-    # 国内站当前不接受 developer 角色（400/code 11128）。只对已验证的
-    # copilot.tencent.com 网关做 copy-on-write 转换，不影响客户端请求对象、
-    # 国际站、自定义上游以及已有 assistant/tool 调用历史。
-    messages = upstream_body.get("messages")
-    if (uses_legacy_chat_roles(state.client.profile.key, url)
-            and isinstance(messages, list)):
-        compatible_messages, mapped_count = adapt_developer_roles(messages)
-        if mapped_count:
-            upstream_body["messages"] = compatible_messages
-            diagnostic("upstream_role_compat", protocol=protocol,
-                       backend=state.client.profile.key,
-                       mapping="developer_to_system", count=mapped_count)
-
-    # 在角色兼容转换之后应用脱敏/客户端提示压缩；原来的 developer
-    # 指令现在同 system 一样进入既有处理链路。
-    if state.enable_desensitize:
-        upstream_body = desensitize_body(upstream_body, compact_harness=True)
-
-    # 始终以流式方式请求上游（聚合或转发）
-    upstream_body["stream"] = True
-    upstream_body.setdefault("stream_options", {"include_usage": True})
-
-
-    headers = {
-        "User-Agent": "Mozilla/5.0 (compatible; Genie-IDE/1.0)",
-        **state.client.auth_headers(),
-        "Content-Type": "application/json",
-        "Accept": "text/event-stream",
-    }
-    # INTL (www.codebuddy.ai) 网关的「会话标识」闸门（2026-10-16 起）。
-    # 缺失 X-Conversation-* 头族会被 400 / code 11128 拒绝。
-    # 仅在国际站注入，避免影响国内站 copilot.tencent.com 的现有逻辑。
-    if "codebuddy.ai" in url:
-        try:
-            headers.update(state.client.conversation_headers(upstream_body))
-            diagnostic("intl_conversation_headers",
-                       conversation_id=headers.get("X-Conversation-ID", "")[:12] + "...")
-        except Exception as exc:  # 头注入失败不应阻断主流程
-            diagnostic("intl_conversation_headers_failed", error=str(exc)[:200])
-
-    # 🔍 调试：输出实际发送的IDE识别headers
+    prepared = ChatRequestPreparer(
+        client=state.client,
+        enable_desensitize=state.enable_desensitize,
+        desensitize_body=desensitize_body,
+        normalize_tool_choice=_normalize_tool_choice,
+        diagnostic=diagnostic,
+    ).prepare(body, protocol)
+    stream, upstream_body, url, headers = (
+        prepared.stream, prepared.body, prepared.url, prepared.headers,
+    )
+
+    # Keep the original diagnostics after gateway-specific header injection.
     if state.logger:
-        ide_headers = {k: v for k, v in headers.items() 
-                      if k.startswith("X-IDE-") or k == "X-Product-Version" or k == "X-Machine-Id"}
+        ide_headers = {k: v for k, v in headers.items()
+                       if k.startswith("X-IDE-") or k == "X-Product-Version" or k == "X-Machine-Id"}
         diagnostic("upstream_ide_headers", **ide_headers)
-    
+
     if stream:
         # 上游响应头必须在客户端 StreamingResponse 开始之前读取，否则 400 会被包装成 200。
         upstream_client = httpx.AsyncClient(timeout=httpx.Timeout(30.0, read=300.0))
@@ -1934,78 +1897,8 @@ async def collect_upstream(
 # ============================================================================
 
 def convert_nonstream(data: dict[str, Any], protocol: str, original: dict[str, Any] | None) -> dict[str, Any]:
-    """将聚合的 OpenAI 格式转换为目标协议格式。"""
-    if protocol == "openai":
-        return data
-
-    choice = (data.get("choices") or [{}])[0]
-    message = choice.get("message") or {}
-    content = message.get("content", "")
-
-    if protocol == "anthropic":
-        content_blocks = []
-        if content:
-            content_blocks.append({"type": "text", "text": content})
-        for call in message.get("tool_calls") or []:
-            fn = call.get("function") or {}
-            try:
-                arguments = json.loads(fn.get("arguments", "{}"))
-            except json.JSONDecodeError:
-                arguments = fn.get("arguments", "")
-            content_blocks.append({
-                "type": "tool_use",
-                "id": call.get("id", ""),
-                "name": fn.get("name", ""),
-                "input": arguments
-            })
-        return {
-            "id": "msg_" + uuid.uuid4().hex,
-            "type": "message",
-            "role": "assistant",
-            "model": (original or {}).get("model", data.get("model", "default")),
-            "content": content_blocks,
-            "stop_reason": "tool_use" if message.get("tool_calls") else "end_turn",
-            "stop_sequence": None,
-            "usage": {
-                "input_tokens": (data.get("usage") or {}).get("prompt_tokens", 0),
-                "output_tokens": (data.get("usage") or {}).get("completion_tokens", 0)
-            }
-        }
-
-    elif protocol == "responses":
-        # 构建 content 数组（文本 + 工具调用）
-        content_parts = []
-        if content:
-            content_parts.append({"type": "output_text", "text": content})
-
-        # 处理工具调用
-        for call in message.get("tool_calls") or []:
-            fn = call.get("function") or {}
-            try:
-                arguments = json.loads(fn.get("arguments", "{}"))
-            except json.JSONDecodeError:
-                arguments = fn.get("arguments", "")
-
-            content_parts.append({
-                "type": "function_call",
-                "id": call.get("id", ""),
-                "name": fn.get("name", ""),
-                "arguments": arguments
-            })
-
-        return {
-            "id": "resp_" + uuid.uuid4().hex,
-            "object": "response",
-            "created_at": now_s(),
-            "status": "completed",
-            "output": [{
-                "type": "message",
-                "role": "assistant",
-                "content": content_parts
-            }]
-        }
-
-    return data
+    """Convert a collected response via the protocol's registered formatter."""
+    return format_nonstream_response(data, protocol, original, now_s)
 
 
 # ============================================================================

src/codebuddy_proxy/protocol_response.py

@@ -0,0 +1,132 @@
+"""Non-streaming response format strategies for supported client protocols.
+
+The chat aggregation layer always returns an OpenAI-shaped object; adapters
+here format it for the client's protocol. Keep the legacy payload contracts,
+including their current tool-call placement and argument parsing.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import Any, Callable, Protocol
+
+
+class NonStreamFormatter(Protocol):
+    def format(
+        self, data: dict[str, Any], original: dict[str, Any] | None,
+        now_s: Callable[[], int],
+    ) -> dict[str, Any]: ...
+
+
+class OpenAIFormatter:
+    def format(
+        self, data: dict[str, Any], original: dict[str, Any] | None,
+        now_s: Callable[[], int],
+    ) -> dict[str, Any]:
+        return data
+
+
+def _first_message(data: dict[str, Any]) -> dict[str, Any]:
+    choice = (data.get("choices") or [{}])[0]
+    return choice.get("message") or {}
+
+
+def _arguments(fn: dict[str, Any]) -> Any:
+    try:
+        return json.loads(fn.get("arguments", "{}"))
+    except json.JSONDecodeError:
+        return fn.get("arguments", "")
+
+
+class AnthropicFormatter:
+    def format(
+        self, data: dict[str, Any], original: dict[str, Any] | None,
+        now_s: Callable[[], int],
+    ) -> dict[str, Any]:
+        message = _first_message(data)
+        content = message.get("content", "")
+        content_blocks: list[dict[str, Any]] = []
+        if content:
+            content_blocks.append({"type": "text", "text": content})
+        for call in message.get("tool_calls") or []:
+            fn = call.get("function") or {}
+            content_blocks.append({
+                "type": "tool_use",
+                "id": call.get("id", ""),
+                "name": fn.get("name", ""),
+                "input": _arguments(fn),
+            })
+        return {
+            "id": "msg_" + uuid.uuid4().hex,
+            "type": "message",
+            "role": "assistant",
+            "model": (original or {}).get("model", data.get("model", "default")),
+            "content": content_blocks,
+            "stop_reason": "tool_use" if message.get("tool_calls") else "end_turn",
+            "stop_sequence": None,
+            "usage": {
+                "input_tokens": (data.get("usage") or {}).get("prompt_tokens", 0),
+                "output_tokens": (data.get("usage") or {}).get("completion_tokens", 0),
+            },
+        }
+
+
+class ResponsesFormatter:
+    def format(
+        self, data: dict[str, Any], original: dict[str, Any] | None,
+        now_s: Callable[[], int],
+    ) -> dict[str, Any]:
+        message = _first_message(data)
+        content = message.get("content", "")
+        content_parts: list[dict[str, Any]] = []
+        if content:
+            content_parts.append({"type": "output_text", "text": content})
+        for call in message.get("tool_calls") or []:
+            fn = call.get("function") or {}
+            content_parts.append({
+                "type": "function_call",
+                "id": call.get("id", ""),
+                "name": fn.get("name", ""),
+                "arguments": _arguments(fn),
+            })
+        return {
+            "id": "resp_" + uuid.uuid4().hex,
+            "object": "response",
+            "created_at": now_s(),
+            "status": "completed",
+            "output": [{
+                "type": "message",
+                "role": "assistant",
+                "content": content_parts,
+            }],
+        }
+
+
+class NonStreamResponseRegistry:
+    """Strategy registry, with the historical OpenAI identity fallback."""
+
+    def __init__(self, formatters: dict[str, NonStreamFormatter] | None = None):
+        self._fallback = OpenAIFormatter()
+        self._formatters = formatters if formatters is not None else {
+            "openai": self._fallback,
+            "anthropic": AnthropicFormatter(),
+            "responses": ResponsesFormatter(),
+        }
+
+    def convert(
+        self, data: dict[str, Any], protocol: str,
+        original: dict[str, Any] | None, now_s: Callable[[], int],
+    ) -> dict[str, Any]:
+        formatter = self._formatters.get(protocol, self._fallback)
+        return formatter.format(data, original, now_s)
+
+
+_default_registry = NonStreamResponseRegistry()
+
+
+def format_nonstream_response(
+    data: dict[str, Any], protocol: str,
+    original: dict[str, Any] | None, now_s: Callable[[], int],
+) -> dict[str, Any]:
+    return _default_registry.convert(data, protocol, original, now_s)