diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py
index 2a48a82..d6e8bf2 100644
--- a/src/polygateway/transports/openai_compat.py
+++ b/src/polygateway/transports/openai_compat.py
@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any
 import httpx
 
 from polygateway.errors import (
+    PolyGatewayError,
     RequestRejectedError,
     ResultInvalidError,
     SourceDeadError,
@@ -30,6 +31,7 @@ from polygateway.providers import (
     resolve_thinking,
 )
 from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
+from polygateway.transports._http_errors import compose_message, summarize_body
 from polygateway.types import EmbeddingTransportResult, SourceConfig, TransportResult
 
 if TYPE_CHECKING:
@@ -107,40 +109,59 @@ def _parse_retry_after(raw: str | None) -> float | None:
     return seconds if seconds > 0 else None
 
 
-def _translate_429(source: SourceConfig, body_text: str, headers: Mapping[str, str]) -> Exception:
+def _translate_429(
+    source: SourceConfig, body_text: str, headers: Mapping[str, str], ctx: dict[str, Any]
+) -> Exception:
+    """429 细分。`body_text` 必须是**未截断的原文**——`ctx["body_text"]` 是摘要,
+    头尾保留会破坏 JSON 结构,拿它解析会让超长 body 的配额耗尽退化成普通限速
+    (该源不再 force_open),把一个诊断改进变成治理 bug(issue #10 实现红线)。
+    """
     try:
         err_type = json.loads(body_text).get("error", {}).get("type", "")
     except (json.JSONDecodeError, AttributeError):
         err_type = ""
+    summary = ctx["body_text"]
     if err_type == "insufficient_quota":
         return SourceDeadError(
-            f"{source.name} 配额耗尽(insufficient_quota)",
-            source_name=source.name,
-            status_code=429,
-            operation="chat",
+            compose_message(f"{source.name} 配额耗尽(insufficient_quota)", summary), **ctx
         )
     return TransientError(
-        f"{source.name} 限速: 429",
+        compose_message(f"{source.name} 限速: 429", summary),
         retry_after_s=_parse_retry_after(headers.get("retry-after")),
-        source_name=source.name,
-        status_code=429,
-        operation="chat",
+        **ctx,
     )
 
 
+def _classify(status: int) -> tuple[type[PolyGatewayError], str]:
+    """状态码 → (错误类, message 标签);映射与 ARCH §6.2 逐条相同,本次零变更。"""
+    if status in (401, 403):
+        return SourceDeadError, "凭据失效/欠费"
+    if status == 400:
+        return RequestRejectedError, "请求被拒"
+    if status >= 500:
+        return TransientError, "瞬时错误"
+    return RequestRejectedError, "客户端错误"
+
+
 def _status_to_error(
     source: SourceConfig, status: int, body_text: str, headers: Mapping[str, str]
 ) -> Exception:
-    ctx: dict[str, Any] = {"source_name": source.name, "status_code": status, "operation": "chat"}
-    if status in (401, 403):
-        return SourceDeadError(f"{source.name} 凭据失效/欠费: {status}", **ctx)
-    if status == 400:
-        return RequestRejectedError(f"{source.name} 请求被拒: 400", **ctx)
+    """非 2xx → 领域错误,**全部分支**携带响应体摘要(issue #10)。
+
+    摘要只算一次,message 与 `body_text` 共用同一份串: 两份不同长度会让"遥测里
+    看到的"与"下游 catch 到的"对不上,排查时反而多一层困惑。
+    """
+    summary = summarize_body(body_text)
+    ctx: dict[str, Any] = {
+        "source_name": source.name,
+        "status_code": status,
+        "operation": "chat",
+        "body_text": summary,
+    }
     if status == 429:
-        return _translate_429(source, body_text, headers)
-    if status >= 500:
-        return TransientError(f"{source.name} 瞬时错误: {status}", **ctx)
-    return RequestRejectedError(f"{source.name} 客户端错误: {status}", **ctx)
+        return _translate_429(source, body_text, headers, ctx)
+    cls, label = _classify(status)
+    return cls(compose_message(f"{source.name} {label}: {status}", summary), **ctx)
 
 
 def _strip_think(content: str) -> tuple[str, str]:
diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py
index 5ac6bbd..62c7389 100644
--- a/tests/unit/test_openai_compat.py
+++ b/tests/unit/test_openai_compat.py
@@ -691,6 +691,116 @@ class TestErrorTranslation:
             await _complete(_transport_for(handler), _source())
 
 
+# issue #10 原文给出的真实响应体(一字不改): 关键在于 code 收尾
+_REJECT_BODY = (
+    '{"error":{"message":"<400> ***.***.InvalidParameter: The image format is illegal '
+    'and cannot be opened","type":"invalid_request_error","param":"",'
+    '"code":"invalid_parameter_error"}}'
+)
+
+
+class TestErrorBodyRetention:
+    """issue #10: 网关说了什么必须活着离开翻译层——message 与字段各留一份。"""
+
+    @pytest.mark.parametrize(
+        ("status", "exc"),
+        [
+            (400, RequestRejectedError),
+            (401, SourceDeadError),
+            (403, SourceDeadError),
+            (404, RequestRejectedError),
+            (500, TransientError),
+            (503, TransientError),
+        ],
+    )
+    async def test_every_non_2xx_branch_keeps_the_body(self, status, exc):
+        def handler(request):
+            return httpx.Response(status, content=_REJECT_BODY.encode())
+
+        with pytest.raises(exc) as ei:
+            await _complete(_transport_for(handler), _source())
+        assert ei.value.body_text == _REJECT_BODY
+        # message 与字段共用同一份串: 遥测里看到的与下游 catch 到的不得打架
+        assert str(ei.value).endswith(f" | {_REJECT_BODY}")
+        assert "invalid_parameter_error" in str(ei.value)
+
+    async def test_rate_limited_429_keeps_body_and_retry_after(self):
+        def handler(request):
+            return httpx.Response(
+                429,
+                content=b'{"error":{"message":"per-minute cap 3"}}',
+                headers={"retry-after": "2.5"},
+            )
+
+        with pytest.raises(TransientError) as ei:
+            await _complete(_transport_for(handler), _source())
+        assert "per-minute cap 3" in str(ei.value)
+        assert ei.value.retry_after_s == 2.5  # 摘要不得干扰既有解析
+
+    async def test_insufficient_quota_429_keeps_body(self):
+        body = json.dumps(
+            {"error": {"type": "insufficient_quota", "message": "daily budget spent"}}
+        )
+
+        def handler(request):
+            return httpx.Response(429, content=body.encode())
+
+        with pytest.raises(SourceDeadError) as ei:
+            await _complete(_transport_for(handler), _source())
+        assert "daily budget spent" in str(ei.value)
+
+    async def test_oversized_insufficient_quota_still_classified_dead(self):
+        """实现红线: 类型判定必须读**原文**。
+
+        摘要会破坏 JSON 结构,若改用摘要解析,超长 body 的配额耗尽将退化成普通
+        限速——配额已耗尽的源不再 force_open,一个诊断改进就变成了治理 bug。
+        """
+        body = json.dumps(
+            {"padding": "P" * 4000, "error": {"type": "insufficient_quota", "message": "spent"}}
+        )
+        assert len(body) > 2048
+
+        def handler(request):
+            return httpx.Response(429, content=body.encode())
+
+        with pytest.raises(SourceDeadError):
+            await _complete(_transport_for(handler), _source())
+
+    async def test_empty_body_leaves_no_dangling_separator(self):
+        def handler(request):
+            return httpx.Response(400, content=b"")
+
+        with pytest.raises(RequestRejectedError) as ei:
+            await _complete(_transport_for(handler), _source())
+        assert str(ei.value) == "qwen_1 请求被拒: 400"
+        assert ei.value.body_text == ""
+
+    @pytest.mark.parametrize("body", [b"<html>gateway down</html>", b"\xff\xfe not utf-8"])
+    async def test_non_json_and_non_utf8_bodies_do_not_explode(self, body):
+        def handler(request):
+            return httpx.Response(400, content=body)
+
+        with pytest.raises(RequestRejectedError) as ei:
+            await _complete(_transport_for(handler), _source())
+        assert ei.value.status_code == 400  # 分类不受 body 形态影响
+
+    async def test_non_stream_path_keeps_the_body(self):
+        def handler(request):
+            return httpx.Response(400, content=_REJECT_BODY.encode())
+
+        with pytest.raises(RequestRejectedError) as ei:
+            await _complete(_transport_for(handler), _source(), stream=False)
+        assert ei.value.body_text == _REJECT_BODY
+
+    async def test_embedding_path_keeps_the_body(self):
+        def handler(request):
+            return httpx.Response(400, content=_REJECT_BODY.encode())
+
+        with pytest.raises(RequestRejectedError) as ei:
+            await _transport_for(handler).embed(texts=["hi"], source=_source(), call_id="cid-embed")
+        assert ei.value.body_text == _REJECT_BODY
+
+
 class TestLifecycle:
     async def test_aclose_idempotent(self):
         def handler(request):