From 7870d8091b48712cb4a802cb4994f49ef374ae49 Mon Sep 17 00:00:00 2001 From: ThreeFish Date: Mon, 6 Jul 2026 15:55:29 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(vendor-logging):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E9=94=99=E8=AF=AF=E6=97=A5=E5=BF=97=E4=B8=AD?= =?UTF-8?q?=E6=96=87=E4=B9=B1=E7=A0=81,=E6=94=B6=E6=95=9B=20bytes=20?= =?UTF-8?q?=E8=A7=A3=E7=A0=81=E4=B8=BA=E5=8D=95=E4=B8=80=20helper;?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上游供应商返回 4xx/5xx 时,流式错误分支将 response.aread() 得到的 bytes 经 %s 直接格式化,Python 走 repr() 导致非 ASCII 的 UTF-8 字节被转义为 \xe6\x82\xa8 之类的不可读序列(中文乱码)。 - model/vendor.py 新增公开工具函数 decode_error_body(raw, limit=500): 先整体 UTF-8 解码(errors="replace" 容忍非法字节)再按字符截断,避免 在多字节边界切断;作为「错误响应体日志解码」的单一事实源; - base.py / copilot.py / antigravity.py 三处同型流式错误日志统一复用该 helper;其中 antigravity 消除了原先重复的 decode 调用(scope 检查与日志 共用一次解码); - model 包(vendor.__all__ 与 __init__)同步 re-export; - 新增 tests/test_vendor_helpers.py:单元测试覆盖中文解码/字符截断/非法字节/ 空值/回归对照,集成测试经 BaseVendor 流式路径(MockTransport 返回 400+中文) 以 caplog 断言日志含可读中文且无字节转义。 🤖 Generated with [Claude Code](https://github.com/claude), [CodeX](https://openai.com), [Gemini](https://github.com/apps/gemini-code-assist) Co-Authored-By: Aurelius Huang --- src/coding/proxy/model/__init__.py | 2 + src/coding/proxy/model/vendor.py | 15 ++++ src/coding/proxy/vendors/antigravity.py | 8 +- src/coding/proxy/vendors/base.py | 3 +- src/coding/proxy/vendors/copilot.py | 3 +- tests/test_vendor_helpers.py | 107 ++++++++++++++++++++++++ uv.lock | 2 +- 7 files changed, 133 insertions(+), 7 deletions(-) create mode 100644 tests/test_vendor_helpers.py 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..8ecc08f 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,13 @@ async def send_message_stream( if response.status_code >= 400: self._on_error_status(response.status_code) error_body = await response.aread() - self._mark_scope_error_if_needed( - error_body.decode("utf-8", errors="ignore"), - ) + decoded_body = decode_error_body(error_body) + self._mark_scope_error_if_needed(decoded_body) logger.warning( "%s stream error: status=%d body=%s", self.get_name(), response.status_code, - error_body[:500], + decoded_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_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" }, From 4d0f6228a514f46f1f2b70ea28f35c2f50f21465 Mon Sep 17 00:00:00 2001 From: ThreeFish Date: Mon, 6 Jul 2026 16:14:10 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(vendor-antigravity):=20=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=20scope=20=E6=A3=80=E6=B5=8B=E6=94=B9=E7=94=A8?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E8=A7=A3=E7=A0=81=20body,=E4=B8=8E=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E5=B1=95=E7=A4=BA=E6=88=AA=E6=96=AD=E8=A7=A3=E8=80=A6?= =?UTF-8?q?,=E8=A1=A5=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95;?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 流式错误路径此前把 decode_error_body(error_body)(截断至前 500 字符) 既用于日志展示、又传入 _mark_scope_error_if_needed 做子串检测。检测这类 判定性逻辑不应受日志展示上限约束:当上游 scope 错误体在 ACCESS_TOKEN_SCOPE_INSUFFICIENT 标记出现前已超过 500 字符时,标记会被 截断丢弃,token 不再标记 needs_reauth 而继续以 scope 不足的凭证重试。 - 检测改用完整解码文本 error_body.decode("utf-8", errors="replace"), 日志展示仍用 decode_error_body 的 500 字符截断版本,二者关注点解耦; - 与非流式路径(send_message)传入完整 response.text 的行为恢复一致; - 补充回归测试:构造标记位于第 500 字符之后的 403 错误体,断言检测仍触发 (旧截断实现下该用例失败)。 🤖 Generated with [Claude Code](https://github.com/claude), [CodeX](https://openai.com), [Gemini](https://github.com/apps/gemini-code-assist) Co-Authored-By: Aurelius Huang --- src/coding/proxy/vendors/antigravity.py | 9 +++-- tests/test_antigravity.py | 45 +++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/src/coding/proxy/vendors/antigravity.py b/src/coding/proxy/vendors/antigravity.py index 8ecc08f..c2b261b 100644 --- a/src/coding/proxy/vendors/antigravity.py +++ b/src/coding/proxy/vendors/antigravity.py @@ -533,13 +533,16 @@ async def send_message_stream( if response.status_code >= 400: self._on_error_status(response.status_code) error_body = await response.aread() - decoded_body = decode_error_body(error_body) - self._mark_scope_error_if_needed(decoded_body) + # 检测用完整解码文本(判定性逻辑不应受日志展示上限影响); + # 日志展示则用 decode_error_body 的 500 字符截断版本。 + self._mark_scope_error_if_needed( + error_body.decode("utf-8", errors="replace"), + ) logger.warning( "%s stream error: status=%d body=%s", self.get_name(), response.status_code, - decoded_body, + 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(