From fb0adfd65d55aa2ca7e9b75e9280fd955f96d580 Mon Sep 17 00:00:00 2001 From: marcus Date: Thu, 22 Oct 2026 23:47:29 +0800 Subject: [PATCH] refactor: isolate upstream request policies and protocol formatters --- docs/refactor-request-response-boundaries.md | 81 ++++++++ src/codebuddy_proxy/__main__.py | 145 ++----------- src/codebuddy_proxy/protocol_response.py | 132 ++++++++++++ src/codebuddy_proxy/upstream_request.py | 147 +++++++++++++ test_intl_conversation_headers.py | 62 +++--- test_protocol_response_strategies.py | 135 ++++++++++++ test_upstream_request_pipeline.py | 207 +++++++++++++++++++ 7 files changed, 759 insertions(+), 150 deletions(-) create mode 100644 docs/refactor-request-response-boundaries.md create mode 100644 src/codebuddy_proxy/protocol_response.py create mode 100644 src/codebuddy_proxy/upstream_request.py create mode 100644 test_protocol_response_strategies.py create mode 100644 test_upstream_request_pipeline.py diff --git a/docs/refactor-request-response-boundaries.md b/docs/refactor-request-response-boundaries.md new file mode 100644 index 0000000..8c253d8 --- /dev/null +++ b/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. diff --git a/src/codebuddy_proxy/__main__.py b/src/codebuddy_proxy/__main__.py index cb8dbf3..f4e61a3 100755 --- a/src/codebuddy_proxy/__main__.py +++ b/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) # ============================================================================ diff --git a/src/codebuddy_proxy/protocol_response.py b/src/codebuddy_proxy/protocol_response.py new file mode 100644 index 0000000..25210f6 --- /dev/null +++ b/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) diff --git a/src/codebuddy_proxy/upstream_request.py b/src/codebuddy_proxy/upstream_request.py new file mode 100644 index 0000000..73a0048 --- /dev/null +++ b/src/codebuddy_proxy/upstream_request.py @@ -0,0 +1,147 @@ +"""Build CodeBuddy chat requests without coupling gateway rules to HTTP streaming. + +The preparation pipeline is intentionally ordered: +normalize tool choice -> gateway message rules -> optional desensitization -> +stream flags -> authenticated headers -> gateway header rules. + +Gateway rules are small strategies rather than branches inside forward_chat(). +The builder performs no I/O beyond the existing client auth/header methods. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Protocol + +from codebuddy_proxy.role_compat import adapt_developer_roles, uses_legacy_chat_roles + +Diagnostic = Callable[..., None] + + +@dataclass(frozen=True) +class PreparedChatRequest: + """Prepared immutable envelope; body and header dictionaries remain mutable.""" + + url: str + headers: dict[str, str] + body: dict[str, Any] + stream: bool + + +class MessagePolicy(Protocol): + def apply( + self, body: dict[str, Any], *, client: Any, url: str, + protocol: str, diagnostic: Diagnostic, + ) -> None: ... + + +class HeaderPolicy(Protocol): + def apply( + self, headers: dict[str, str], *, body: dict[str, Any], + client: Any, url: str, diagnostic: Diagnostic, + ) -> None: ... + + +class DomesticDeveloperRolePolicy: + """Scope the legacy-role workaround to the verified domestic gateway.""" + + def apply( + self, body: dict[str, Any], *, client: Any, url: str, + protocol: str, diagnostic: Diagnostic, + ) -> None: + messages = body.get("messages") + if (uses_legacy_chat_roles(client.profile.key, url) + and isinstance(messages, list)): + compatible, count = adapt_developer_roles(messages) + if count: + body["messages"] = compatible + diagnostic( + "upstream_role_compat", protocol=protocol, + backend=client.profile.key, + mapping="developer_to_system", count=count, + ) + + +class InternationalConversationHeaderPolicy: + """Preserve the existing URL guard and best-effort header injection. + + Deliberately retain the original 'codebuddy.ai' substring check to avoid + changing which configured gateways receive X-Conversation-* headers. + """ + + def apply( + self, headers: dict[str, str], *, body: dict[str, Any], + client: Any, url: str, diagnostic: Diagnostic, + ) -> None: + if "codebuddy.ai" in url: + try: + headers.update(client.conversation_headers(body)) + diagnostic( + "intl_conversation_headers", + conversation_id=headers.get("X-Conversation-ID", "")[:12] + "...", + ) + except Exception as exc: # A header failure never blocks forwarding. + diagnostic("intl_conversation_headers_failed", error=str(exc)[:200]) + + +class ChatRequestPreparer: + """Compose gateway strategies with the existing request transformations. + + Dependencies are injected to avoid importing the FastAPI entry point and + to keep normalization/desensitization behavior under its existing owners. + """ + + def __init__( + self, + *, + client: Any, + enable_desensitize: bool, + desensitize_body: Callable[..., dict[str, Any]], + normalize_tool_choice: Callable[[Any], Any], + diagnostic: Diagnostic, + message_policies: tuple[MessagePolicy, ...] = (DomesticDeveloperRolePolicy(),), + header_policies: tuple[HeaderPolicy, ...] = (InternationalConversationHeaderPolicy(),), + ) -> None: + self.client = client + self.enable_desensitize = enable_desensitize + self.desensitize_body = desensitize_body + self.normalize_tool_choice = normalize_tool_choice + self.diagnostic = diagnostic + self.message_policies = message_policies + self.header_policies = header_policies + + def prepare(self, body: dict[str, Any], protocol: str) -> PreparedChatRequest: + stream = bool(body.get("stream")) + upstream_body = dict(body) + + if "tool_choice" in upstream_body: + upstream_body["tool_choice"] = self.normalize_tool_choice( + upstream_body["tool_choice"] + ) + + url = self.client.build_url(self.client.profile.chat_path) + for policy in self.message_policies: + policy.apply( + upstream_body, client=self.client, url=url, protocol=protocol, + diagnostic=self.diagnostic, + ) + + if self.enable_desensitize: + upstream_body = self.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)", + **self.client.auth_headers(), + "Content-Type": "application/json", + "Accept": "text/event-stream", + } + for policy in self.header_policies: + policy.apply( + headers, body=upstream_body, client=self.client, url=url, + diagnostic=self.diagnostic, + ) + + return PreparedChatRequest(url=url, headers=headers, body=upstream_body, stream=stream) diff --git a/test_intl_conversation_headers.py b/test_intl_conversation_headers.py index 3083961..a658b1e 100644 --- a/test_intl_conversation_headers.py +++ b/test_intl_conversation_headers.py @@ -9,9 +9,6 @@ 3. 注入条件严格限定 INTL —— 国内站 URL 不得带这些头 """ -import json - -import codebuddy_proxy.__main__ as proxy from codebuddy_proxy.codebuddy_client_demo import CodeBuddyClient REQUIRED_HEADERS = ( @@ -111,26 +108,43 @@ def test_headers_tolerate_missing_or_malformed_body(): def test_intl_gate_is_scoped_to_codebuddy_ai(): - """注入条件必须严格限定 INTL —— 国内站 URL 不得命中。 + """Verify emitted headers rather than inspecting a particular source file. - 这是作者在 issue #11 里明确要求的「避免影响国内站现有逻辑」。 - 以源码里的判定表达式为准，防止后续重构把范围放宽。 + The policy is allowed to move between modules during refactoring; the + observable contract is that only the international URL receives headers. """ - import inspect - - source = inspect.getsource(proxy) - # 找到 forward_chat 里注入 conversation headers 的那处守卫 - marker = "conversation_headers" - assert marker in source, "forward_chat 未注入 conversation headers" - - idx = source.index(marker) - window = source[max(0, idx - 600):idx] - # 只看可执行代码，剔除注释（注释里会提到国内站做说明） - code = "\n".join( - line for line in window.splitlines() - if line.strip() and not line.strip().startswith("#") - ) - assert "codebuddy.ai" in code, \ - "注入守卫必须包含 codebuddy.ai 判定（仅 INTL）" - assert "copilot.tencent.com" not in code, \ - "注入守卫不得覆盖国内站" + from types import SimpleNamespace + + from codebuddy_proxy.upstream_request import ChatRequestPreparer + + class FakeClient: + def __init__(self, endpoint: str): + self.endpoint = endpoint + self.profile = SimpleNamespace(key="global", chat_path="/v2/chat/completions") + self.conversation_calls = 0 + + def build_url(self, path): + return self.endpoint + path + + def auth_headers(self): + return {"Authorization": "Bearer test"} + + def conversation_headers(self, body): + self.conversation_calls += 1 + return {name: "test-conversation-id" for name in REQUIRED_HEADERS} + + for endpoint, should_inject in [ + ("https://www.codebuddy.ai", True), + ("https://copilot.tencent.com", False), + ("https://example.test", False), + ]: + client = FakeClient(endpoint) + prepared = ChatRequestPreparer( + client=client, + enable_desensitize=False, + desensitize_body=lambda body, **kwargs: body, + normalize_tool_choice=lambda choice: choice, + diagnostic=lambda *args, **kwargs: None, + ).prepare(_body("keep system context"), "openai") + assert client.conversation_calls == int(should_inject) + assert all((key in prepared.headers) == should_inject for key in REQUIRED_HEADERS) diff --git a/test_protocol_response_strategies.py b/test_protocol_response_strategies.py new file mode 100644 index 0000000..0408e59 --- /dev/null +++ b/test_protocol_response_strategies.py @@ -0,0 +1,135 @@ +"""Golden contracts for the non-streaming protocol formatter strategies.""" + +from copy import deepcopy +from types import SimpleNamespace + +import pytest + +import codebuddy_proxy.__main__ as proxy +import codebuddy_proxy.protocol_response as formats +from codebuddy_proxy.protocol_response import NonStreamResponseRegistry + + +@pytest.fixture +def fixed_clock_and_id(monkeypatch): + monkeypatch.setattr(formats.uuid, "uuid4", lambda: SimpleNamespace(hex="deadbeef")) + monkeypatch.setattr(proxy, "now_s", lambda: 1710000000) + + +def _data(*, content="hello", calls=None): + return { + "id": "chatcmpl-1", + "object": "chat.completion", + "model": "upstream-model", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": content, "tool_calls": calls}, + "finish_reason": "stop", + }], + "usage": {"prompt_tokens": 11, "completion_tokens": 7, "total_tokens": 18}, + } + + +def _calls(): + return [ + {"id": "a", "function": {"name": "read", "arguments": '{"path":"a.txt"}'}}, + {"id": "b", "function": {"name": "shell", "arguments": "not json"}}, + ] + + +@pytest.mark.parametrize("protocol", ["openai", "unknown", ""]) +def test_openai_and_unknown_protocols_preserve_identity(protocol): + data = _data(calls=_calls()) + assert proxy.convert_nonstream(data, protocol, None) is data + + +def test_anthropic_text_only_contract(fixed_clock_and_id): + data = _data() + original = deepcopy(data) + output = proxy.convert_nonstream(data, "anthropic", {"model": "client-model"}) + assert output == { + "id": "msg_deadbeef", + "type": "message", + "role": "assistant", + "model": "client-model", + "content": [{"type": "text", "text": "hello"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 7}, + } + assert data == original + + +def test_anthropic_tool_call_args_fallback_preserves_existing_structure(fixed_clock_and_id): + output = proxy.convert_nonstream( + _data(content="", calls=_calls()), "anthropic", None, + ) + assert output == { + "id": "msg_deadbeef", + "type": "message", + "role": "assistant", + "model": "upstream-model", + "content": [ + {"type": "tool_use", "id": "a", "name": "read", "input": {"path": "a.txt"}}, + {"type": "tool_use", "id": "b", "name": "shell", "input": "not json"}, + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 7}, + } + + +def test_responses_text_and_tools_contract(fixed_clock_and_id): + data = _data(content="ready", calls=_calls()) + original = deepcopy(data) + assert proxy.convert_nonstream(data, "responses", None) == { + "id": "resp_deadbeef", + "object": "response", + "created_at": 1710000000, + "status": "completed", + "output": [{ + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "ready"}, + {"type": "function_call", "id": "a", "name": "read", "arguments": {"path": "a.txt"}}, + {"type": "function_call", "id": "b", "name": "shell", "arguments": "not json"}, + ], + }], + } + assert data == original + + +@pytest.mark.parametrize("protocol", ["anthropic", "responses"]) +def test_empty_choice_and_missing_usage_keep_defaults(fixed_clock_and_id, protocol): + output = proxy.convert_nonstream( + {"model": "upstream", "choices": [], "usage": None}, protocol, None, + ) + if protocol == "anthropic": + assert output["content"] == [] + assert output["stop_reason"] == "end_turn" + assert output["usage"] == {"input_tokens": 0, "output_tokens": 0} + assert output["model"] == "upstream" + else: + assert output["output"][0]["content"] == [] + + +def test_missing_function_arguments_becomes_empty_dict(fixed_clock_and_id): + tool = {"id": "t", "function": {"name": "read"}} + anthropic = proxy.convert_nonstream(_data(calls=[tool]), "anthropic", None) + responses = proxy.convert_nonstream(_data(calls=[tool]), "responses", None) + assert anthropic["content"][1]["input"] == {} + assert responses["output"][0]["content"][1]["arguments"] == {} + + +def test_custom_protocol_strategy_can_be_registered_without_changing_routes(): + class Custom: + def format(self, data, original, now_s): + return {"custom": True, "original": original, "now": now_s()} + + registry = NonStreamResponseRegistry({"new_protocol": Custom()}) + data = _data() + assert registry.convert(data, "new_protocol", {"model": "m"}, lambda: 24) == { + "custom": True, "original": {"model": "m"}, "now": 24, + } + assert registry.convert(data, "unknown", None, lambda: 24) is data diff --git a/test_upstream_request_pipeline.py b/test_upstream_request_pipeline.py new file mode 100644 index 0000000..b09fa99 --- /dev/null +++ b/test_upstream_request_pipeline.py @@ -0,0 +1,207 @@ +"""Contract tests for the refactored upstream request preparation pipeline. + +No real credentials or network calls are used. +""" + +from copy import deepcopy +from types import SimpleNamespace + +import pytest + +from codebuddy_proxy.upstream_request import ChatRequestPreparer + + +class FakeClient: + def __init__(self, *, endpoint="https://copilot.tencent.com", backend="domestic"): + self.endpoint = endpoint + self.profile = SimpleNamespace(key=backend, chat_path="/v2/chat/completions") + self.calls = [] + self.fail_conversation_headers = False + + def build_url(self, path): + self.calls.append(("build_url", path)) + return self.endpoint + path + + def auth_headers(self): + self.calls.append(("auth_headers", None)) + return { + "Authorization": "Bearer test", + "X-IDE-Type": "vscode", + "X-Machine-Id": "machine-1", + } + + def conversation_headers(self, body): + self.calls.append(("conversation_headers", deepcopy(body))) + if self.fail_conversation_headers: + raise ValueError("invalid conversation metadata") + return { + "X-Conversation-ID": "0123456789abcdef", + "X-Conversation-Message-ID": "new-message", + "X-Conversation-Request-ID": "new-request", + } + + +def _preparer(client, *, desensitize=False, effects=None, desensitize_fn=None, + message_policies=None, header_policies=None): + effects = effects if effects is not None else [] + kwargs = {} + if message_policies is not None: + kwargs["message_policies"] = message_policies + if header_policies is not None: + kwargs["header_policies"] = header_policies + return ChatRequestPreparer( + client=client, + enable_desensitize=desensitize, + desensitize_body=desensitize_fn or (lambda body, **kw: body), + normalize_tool_choice=lambda value: ( + value["function"]["name"] if isinstance(value, dict) else value + ), + diagnostic=lambda event, **fields: effects.append((event, fields)), + **kwargs, + ) + + +@pytest.mark.parametrize("stream", [True, False, None]) +def test_preparer_normalizes_choice_and_forces_stream_without_mutating_client(stream): + client = FakeClient() + body = { + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + "tool_choice": {"type": "function", "function": {"name": "read"}}, + "stream": stream, + } + original = deepcopy(body) + req = _preparer(client).prepare(body, "openai") + + assert req.url == "https://copilot.tencent.com/v2/chat/completions" + assert req.stream is bool(stream) + assert req.body["stream"] is True + assert req.body["stream_options"] == {"include_usage": True} + assert req.body["tool_choice"] == "read" + assert req.headers == { + "User-Agent": "Mozilla/5.0 (compatible; Genie-IDE/1.0)", + "Authorization": "Bearer test", + "X-IDE-Type": "vscode", + "X-Machine-Id": "machine-1", + "Content-Type": "application/json", + "Accept": "text/event-stream", + } + assert body == original + assert req.body is not body + assert client.calls == [ + ("build_url", "/v2/chat/completions"), + ("auth_headers", None), + ] + + +def test_explicit_stream_options_are_preserved_by_setdefault(): + original_options = {"include_usage": False, "custom": "yes"} + body = {"stream_options": original_options, "stream": True} + prepared = _preparer(FakeClient()).prepare(body, "openai") + assert prepared.body["stream_options"] is original_options + assert original_options == {"include_usage": False, "custom": "yes"} + + +@pytest.mark.parametrize("endpoint,backend,roles", [ + ("https://copilot.tencent.com", "domestic", ["system", "user"]), + ("https://www.codebuddy.ai", "global", ["developer", "user"]), + ("https://somewhere.example", "domestic", ["developer", "user"]), + ("https://copilot.tencent.com", "global", ["developer", "user"]), +]) +def test_gateway_message_strategy_scope_and_diagnostics(endpoint, backend, roles): + client = FakeClient(endpoint=endpoint, backend=backend) + effects = [] + messages = [{"role": "developer", "content": "guidance"}, {"role": "user", "content": "question"}] + body = {"messages": messages} + original = deepcopy(body) + req = _preparer(client, effects=effects).prepare(body, "openai") + + assert [m["role"] for m in req.body["messages"]] == roles + assert body == original + compat = [fields for name, fields in effects if name == "upstream_role_compat"] + assert len(compat) == (1 if roles[0] == "system" else 0) + if compat: + assert compat[0] == { + "protocol": "openai", "backend": "domestic", + "mapping": "developer_to_system", "count": 1, + } + + +def test_transform_steps_run_in_existing_order(): + effects = [] + seen = [] + client = FakeClient() + + def desensitize(body, **kwargs): + seen.append((deepcopy(body), kwargs)) + return {**body, "messages": [{"role": "system", "content": "masked"}]} + + body = { + "stream": False, + "messages": [{"role": "developer", "content": "sensitive"}], + } + req = _preparer( + client, desensitize=True, desensitize_fn=desensitize, effects=effects, + ).prepare(body, "openai") + + assert seen[0][0]["messages"][0]["role"] == "system" + assert seen[0][0]["stream"] is False + assert seen[0][1] == {"compact_harness": True} + assert req.body["messages"][0]["content"] == "masked" + assert req.body["stream"] is True + assert body["messages"][0]["role"] == "developer" + assert [name for name, _ in effects] == ["upstream_role_compat"] + + +def test_intl_conversation_headers_get_fully_transformed_body(): + client = FakeClient(endpoint="https://www.codebuddy.ai", backend="global") + effects = [] + req = _preparer(client, effects=effects).prepare( + {"stream": False, "messages": [{"role": "developer", "content": "guide"}]}, + "openai", + ) + conv_call = [details for name, details in client.calls if name == "conversation_headers"] + assert len(conv_call) == 1 + assert conv_call[0] == req.body + assert conv_call[0]["stream"] is True + assert conv_call[0]["messages"][0]["role"] == "developer" + assert req.headers["X-Conversation-ID"] == "0123456789abcdef" + assert effects == [("intl_conversation_headers", {"conversation_id": "0123456789ab..."})] + + +def test_intl_headers_failure_is_best_effort_and_diagnostic_is_truncated(): + client = FakeClient(endpoint="https://www.codebuddy.ai", backend="global") + client.fail_conversation_headers = True + effects = [] + req = _preparer(client, effects=effects).prepare( + {"messages": [{"role": "user", "content": "hello"}]}, "openai", + ) + assert "X-Conversation-ID" not in req.headers + assert req.headers["Authorization"] == "Bearer test" + assert effects == [ + ("intl_conversation_headers_failed", {"error": "invalid conversation metadata"}), + ] + + +def test_custom_gateway_strategies_can_be_composed_without_modifying_builder(): + steps = [] + + class MessageRule: + def apply(self, body, *, client, url, protocol, diagnostic): + steps.append("message") + body["extra"] = protocol + + class HeaderRule: + def apply(self, headers, *, body, client, url, diagnostic): + steps.append("headers") + assert body["extra"] == "anthropic" + assert body["stream"] is True + headers["X-Extra"] = "enabled" + + req = _preparer( + FakeClient(), message_policies=(MessageRule(),), header_policies=(HeaderRule(),), + ).prepare({"model": "test"}, "anthropic") + + assert steps == ["message", "headers"] + assert req.headers["X-Extra"] == "enabled" + assert req.body["extra"] == "anthropic"