fix(vendor-logging): 修复流式错误日志中文乱码并解耦 antigravity scope 检测;

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",

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",

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}",

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}",

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}",

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(

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