diff --git a/src/coding/proxy/model/__init__.py b/src/coding/proxy/model/__init__.py index 4921e60..e2b70bf 100644 --- a/src/coding/proxy/model/__init__.py +++ b/src/coding/proxy/model/__init__.py @@ -25,6 +25,7 @@ UsageInfo, VendorCapabilities, VendorResponse, + decode_error_body, decode_json_body, extract_error_message, sanitize_headers_for_synthetic_response, @@ -83,6 +84,7 @@ "CopilotModelCatalog", "RequestCapabilities", "UsageInfo", + "decode_error_body", "decode_json_body", "extract_error_message", "sanitize_headers_for_synthetic_response", diff --git a/src/coding/proxy/model/vendor.py b/src/coding/proxy/model/vendor.py index dd5ef4f..867069a 100644 --- a/src/coding/proxy/model/vendor.py +++ b/src/coding/proxy/model/vendor.py @@ -75,6 +75,20 @@ def extract_error_message( return text[:500] if text else None +def decode_error_body(raw: bytes, limit: int = 500) -> str: + """将上游错误响应体（bytes）安全解码为可读文本，供日志展示. + + 直接以 ``%s`` 格式化 ``bytes`` 会走 ``repr()``，导致非 ASCII 的 UTF-8 + 字节被转义为 ``\\xe6\\x82\\xa8`` 之类的不可读序列（中文乱码）。本函数： + + - 使用 ``errors="replace"`` 容忍非法字节，非法字节降级为 ``�`` 而非抛异常， + 确保日志路径绝对健壮； + - 先整体解码再按字符截断，避免在多字节 UTF-8 边界切断产生乱码 + （因此 ``limit`` 语义为「字符数」而非「字节数」）。 + """ + return raw.decode("utf-8", errors="replace")[:limit] + + # ═══════════════════════════════════════════════════════════════ # 供应商核心数据类型 # ═══════════════════════════════════════════════════════════════ @@ -229,6 +243,7 @@ def age_seconds(self) -> int | None: "CopilotMisdirectedRequest", "CopilotModelCatalog", # 工具函数 + "decode_error_body", "decode_json_body", "extract_error_message", "sanitize_headers_for_synthetic_response", diff --git a/src/coding/proxy/vendors/antigravity.py b/src/coding/proxy/vendors/antigravity.py index b4d7199..c2b261b 100644 --- a/src/coding/proxy/vendors/antigravity.py +++ b/src/coding/proxy/vendors/antigravity.py @@ -24,6 +24,7 @@ VendorCapabilities, VendorResponse, _sanitize_headers_for_synthetic_response, + decode_error_body, ) # GoogleOAuthTokenManager 已从 antigravity_token_manager.py 合并至本文件末尾 @@ -532,14 +533,16 @@ async def send_message_stream( if response.status_code >= 400: self._on_error_status(response.status_code) error_body = await response.aread() + # 检测用完整解码文本(判定性逻辑不应受日志展示上限影响); + # 日志展示则用 decode_error_body 的 500 字符截断版本。 self._mark_scope_error_if_needed( - error_body.decode("utf-8", errors="ignore"), + error_body.decode("utf-8", errors="replace"), ) logger.warning( "%s stream error: status=%d body=%s", self.get_name(), response.status_code, - error_body[:500], + decode_error_body(error_body), ) raise httpx.HTTPStatusError( f"{self.get_name()} API error: {response.status_code}", diff --git a/src/coding/proxy/vendors/base.py b/src/coding/proxy/vendors/base.py index d1434bc..1c4c9c3 100644 --- a/src/coding/proxy/vendors/base.py +++ b/src/coding/proxy/vendors/base.py @@ -25,6 +25,7 @@ UsageInfo, VendorCapabilities, VendorResponse, + decode_error_body, decode_json_body, extract_error_message, sanitize_headers_for_synthetic_response, @@ -328,7 +329,7 @@ async def send_message_stream( "%s stream error: status=%d body=%s", self.get_name(), response.status_code, - error_body[:500], + decode_error_body(error_body), ) raise httpx.HTTPStatusError( f"{self.get_name()} API error: {response.status_code}", diff --git a/src/coding/proxy/vendors/copilot.py b/src/coding/proxy/vendors/copilot.py index b159976..a8e7f0f 100644 --- a/src/coding/proxy/vendors/copilot.py +++ b/src/coding/proxy/vendors/copilot.py @@ -25,6 +25,7 @@ VendorResponse, _decode_json_body, _extract_error_message, + decode_error_body, ) from .copilot_models import ( # noqa: F401 CopilotMisdirectedRequest, @@ -423,7 +424,7 @@ async def _stream_from_client( "%s stream error: status=%d body=%s", self.get_name(), response.status_code, - error_body[:500], + decode_error_body(error_body), ) raise httpx.HTTPStatusError( f"{self.get_name()} API error: {response.status_code}", diff --git a/tests/test_antigravity.py b/tests/test_antigravity.py index cc93127..675d995 100644 --- a/tests/test_antigravity.py +++ b/tests/test_antigravity.py @@ -275,6 +275,51 @@ def test_mark_scope_error_if_needed(): assert diagnostics["token_manager"]["error_kind"] == "insufficient_scope" +@pytest.mark.asyncio +async def test_stream_scope_detection_scans_full_body_not_log_truncation(): + """流式 scope 检测必须扫描完整 body,而非日志展示用的 500 字符截断. + + 回归守护:``decode_error_body`` 的 ``[:500]`` 截断仅服务日志展示;若把它 + 同时喂给 ``_mark_scope_error_if_needed`` 的子串检测,则标记出现在 500 字符 + 之后的错误体会漏检,token 不会被标记 ``needs_reauth``。本用例构造标记位于 + 500 字符之后的 403 错误体,断言检测仍然触发(旧的截断实现下会失败)。 + """ + marker = "ACCESS_TOKEN_SCOPE_INSUFFICIENT" + # message 填充 600 字符,把 details 中的 marker 挤到第 500 字符之后 + body = ( + f'{{"error":{{"message":"{"x" * 600}","status":"PERMISSION_DENIED",' + f'"details":[{{"reason":"{marker}"}}]}}}}' + ).encode() + assert body.decode().index(marker) > 500 # 前置条件:marker 确在 500 之后 + + vendor = AntigravityVendor(AntigravityConfig(), FailoverConfig(), ModelMapper([])) + vendor._token_manager.get_token = AsyncMock(return_value="tok") + vendor._discover_project_id = AsyncMock(return_value="") # 避免真实网络发现 + vendor._client = httpx.AsyncClient( + base_url=vendor._base_url, + transport=httpx.MockTransport( + lambda _req: httpx.Response( + 403, content=body, headers={"content-type": "application/json"} + ) + ), + ) + + with pytest.raises(httpx.HTTPStatusError): + async for _ in vendor.send_message_stream( + { + "model": "claude-sonnet-4-20250514", + "messages": [{"role": "user", "content": "Hi"}], + }, + {}, + ): + pass + + await vendor.close() + + diagnostics = vendor.get_diagnostics() + assert diagnostics["token_manager"]["error_kind"] == "insufficient_scope" + + def test_antigravity_supports_request_with_tools_thinking_and_metadata(): vendor = AntigravityVendor(AntigravityConfig(), FailoverConfig(), ModelMapper([])) supported, reasons = vendor.supports_request( diff --git a/tests/test_vendor_helpers.py b/tests/test_vendor_helpers.py new file mode 100644 index 0000000..322da7a --- /dev/null +++ b/tests/test_vendor_helpers.py @@ -0,0 +1,107 @@ +"""供应商工具函数测试 — 聚焦 decode_error_body 的编码正确性与流式错误日志回归. + +背景：上游 4xx/5xx 错误的流式日志曾将 ``bytes`` 响应体经 ``%s`` 直接格式化， +Python 对 ``bytes`` 走 ``repr()``，导致非 ASCII 的 UTF-8 字节被转义为 +``\\xe6\\x82\\xa8`` 之类的不可读序列（中文乱码）。本测试锚定该根因并守护修复。 +""" + +import logging + +import httpx +import pytest + +from coding.proxy.config.schema import AnthropicConfig, FailoverConfig +from coding.proxy.model.vendor import decode_error_body +from coding.proxy.vendors.anthropic import AnthropicVendor + +# ── 单元测试：decode_error_body ────────────────────────────── + +_ZH_TEXT = "您的账户已达到速率限制,请您控制请求频率" + + +class TestDecodeErrorBody: + def test_chinese_bytes_decoded_readable(self): + """中文 UTF-8 bytes 应解码为可读字符,不残留转义序列或 bytes 前缀.""" + raw = f'{{"message":"[1302][{_ZH_TEXT}]"}}'.encode() + out = decode_error_body(raw) + assert _ZH_TEXT in out + assert "\\x" not in out # 无字节转义 + assert not out.startswith("b'") # 非 bytes repr + assert isinstance(out, str) + + def test_truncates_by_characters(self): + """limit 语义为字符数;超长输入按字符截断.""" + raw = ("你" * 1000).encode() + out = decode_error_body(raw, limit=100) + assert len(out) == 100 + assert out == "你" * 100 # 未在多字节边界切断 + + def test_invalid_bytes_do_not_raise(self): + """非法字节以 errors='replace' 降级为占位符,绝不抛异常.""" + raw = b"\xff\xfe invalid" + out = decode_error_body(raw) + assert isinstance(out, str) + assert "�" in out # U+FFFD replacement character + + def test_empty_bytes(self): + assert decode_error_body(b"") == "" + + def test_ascii_passthrough(self): + raw = b'{"error":{"type":"rate_limit_error"}}' + assert decode_error_body(raw) == '{"error":{"type":"rate_limit_error"}}' + + def test_regression_repr_vs_decode(self): + """回归对照:复现旧 bug(%s 对 bytes 走 repr)并证明新实现修复之.""" + raw = _ZH_TEXT.encode() + # 旧行为:bytes 经 %s → repr → 转义字节序列(刻意保留 % 格式化以精确 + # 复现旧 logger.warning("...%s...", error_body) 的缺陷路径,故 noqa UP031) + assert "\\x" in "%s" % raw # noqa: UP031 + # 新行为:先解码 → 可读中文,无转义 + assert "\\x" not in decode_error_body(raw) + + +# ── 集成测试:BaseVendor 流式错误日志路径 ──────────────────── + + +@pytest.mark.asyncio +async def test_stream_error_log_decodes_chinese(caplog): + """经 BaseVendor.send_message_stream 的流式错误日志应输出可读中文. + + 以 AnthropicVendor(纯继承 BaseVendor,无重试/包装)具象化, + 注入挂载 MockTransport 的 AsyncClient 返回 400 + 含中文的 body, + 断言 WARNING 日志文本含中文且不含字节转义序列。 + """ + error_payload = f'{{"error":{{"message":"{_ZH_TEXT}"}}}}'.encode() + + def _handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 400, + content=error_payload, + headers={"content-type": "application/json; charset=utf-8"}, + ) + + vendor = AnthropicVendor(AnthropicConfig(), FailoverConfig()) + # 注入 MockTransport 客户端(base_url 与真实一致,仅替换 transport) + vendor._client = httpx.AsyncClient( + base_url=vendor._base_url, + transport=httpx.MockTransport(_handler), + ) + + with caplog.at_level(logging.WARNING): + with pytest.raises(httpx.HTTPStatusError): + async for _ in vendor.send_message_stream( + {"model": "claude-opus-4-6", "messages": []}, + {"authorization": "Bearer sk-test"}, + ): + pass + + await vendor.close() + + stream_errors = [ + r.getMessage() for r in caplog.records if "stream error" in r.getMessage() + ] + assert stream_errors, "未捕获到 stream error 日志" + msg = stream_errors[0] + assert _ZH_TEXT in msg # 中文可读 + assert "\\x" not in msg # 无字节转义 + assert "body=b'" not in msg # 非 bytes repr diff --git a/uv.lock b/uv.lock index b9197bf..7682827 100644 --- a/uv.lock +++ b/uv.lock @@ -74,7 +74,7 @@ wheels = [ [[package]] name = "coding-proxy" -version = "0.5.2a5" +version = "0.5.2a7" source = { editable = "." } dependencies = [ { name = "aiosqlite" },