Merge pull request #84 from ThreeFish-AI/vk/b2ca-fix-500-200

fix(error-handling): 修复流式路径裸 500 异常并增强错误日志

Author: ThreeFish-AI

src/coding/proxy/routing/executor.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 from typing import Any, AsyncIterator
@@ -40,6 +41,34 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _log_http_error_detail(tier_name: str, exc: Exception, *, is_stream: bool = False) -> None:
+    """记录 HTTP 错误的详细信息（状态码 / 响应体摘要 / 异常类型）.
+
+    替代原先单行 ``logger.warning("Tier %s stream failed: %s", ...)``，
+    在非 200 响应时输出更丰富的诊断上下文，便于跟踪上游故障根因。
+    """
+    detail_parts = [f"Tier {tier_name} {'stream' if is_stream else 'message'} failed:"]
+    detail_parts.append(f"  exc_type={type(exc).__name__}")
+    if isinstance(exc, httpx.HTTPStatusError) and exc.response is not None:
+        resp = exc.response
+        detail_parts.append(f"  status={resp.status_code}")
+        body_preview = (resp.text[:300] if resp.text else "(empty)") if resp.content else "(no content)"
+        detail_parts.append(f"  response_body={body_preview}")
+        # 尝试提取 error type / message
+        try:
+            payload = resp.json() if resp.content else None
+        except Exception:
+            payload = None
+        if isinstance(payload, dict):
+            err = payload.get("error", {})
+            if isinstance(err, dict):
+                detail_parts.append(f"  error_type={err.get('type', 'N/A')}")
+                detail_parts.append(f"  error_msg={err.get('message', 'N/A')[:200]}")
+    else:
+        detail_parts.append(f"  message={str(exc)[:300]}")
+    logger.warning("\n".join(detail_parts))
+
 # tier.name → 上游 Vendor 协议标签映射（用于 token 用量日志标注）
 _VENDOR_PROTOCOL_LABEL_MAP: dict[str, str] = {
     "anthropic": "Anthropic",
@@ -143,12 +172,23 @@ async def execute_stream(
                     raise
 
             except (httpx.HTTPStatusError, httpx.TimeoutException, httpx.ConnectError, httpx.ReadError) as exc:
-                logger.warning("Tier %s stream failed: %s", tier.name, exc)
+                _log_http_error_detail(tier.name, exc, is_stream=True)
                 should_continue, failed_tier_name, last_exc = await self._handle_http_error(tier, exc, is_last, failed_tier_name, last_exc, is_stream=True)
                 if should_continue:
                     continue
                 if is_last:
                     raise
+            except Exception as exc:
+                logger.error(
+                    "Tier %s stream unexpected error: %s: %s",
+                    tier.name, type(exc).__name__, exc,
+                    exc_info=True,
+                )
+                tier.record_failure()
+                failed_tier_name = tier.name
+                if not is_last:
+                    continue
+                raise
 
         if last_exc:
             raise last_exc
@@ -229,12 +269,23 @@ async def execute_message(
                 continue
 
             except (httpx.TimeoutException, httpx.ConnectError, httpx.ReadError) as exc:
-                logger.warning("Tier %s connection error: %s", tier.name, exc)
+                _log_http_error_detail(tier.name, exc, is_stream=False)
                 tier.record_failure()
                 failed_tier_name = tier.name
                 if is_last:
                     raise
                 continue
+            except Exception as exc:
+                logger.error(
+                    "Tier %s message unexpected error: %s: %s",
+                    tier.name, type(exc).__name__, exc,
+                    exc_info=True,
+                )
+                tier.record_failure()
+                failed_tier_name = tier.name
+                if not is_last:
+                    continue
+                raise
 
         if incompatible_reasons:
             raise NoCompatibleVendorError("当前请求包含仅客户端/MCP 可安全承接的能力，未找到兼容供应商", reasons=incompatible_reasons)

src/coding/proxy/server/routes.py

@@ -43,6 +43,16 @@ async def _stream_proxy(router: Any, body: dict, headers: dict) -> Any:
     except httpx.HTTPStatusError as exc:
         error_type, message = extract_stream_http_error(exc)
         yield stream_error_event(error_type, message)
+    except Exception as exc:
+        logger.error(
+            "_stream_proxy 未预期异常: %s: %s",
+            type(exc).__name__, exc,
+            exc_info=True,
+        )
+        yield stream_error_event(
+            "api_error",
+            f"内部错误: {type(exc).__name__}: {exc}",
+        )
 
 
 def register_core_routes(app: Any, router: Any) -> None:
@@ -76,6 +86,13 @@ async def messages(request: Request) -> Response:
             return json_error_response(503, error_type="authentication_error", message=str(exc))
         except (httpx.TimeoutException, httpx.ConnectError, httpx.ReadError) as exc:
             return json_error_response(502, error_type="api_error", message=f"上游不可达: {exc}")
+        except Exception as exc:
+            logger.error(
+                "messages() 非流式路径未预期异常: %s: %s",
+                type(exc).__name__, exc,
+                exc_info=True,
+            )
+            return json_error_response(500, error_type="api_error", message=f"内部错误: {type(exc).__name__}")
         return Response(content=resp.raw_body or b"{}", status_code=resp.status_code, media_type="application/json")
 
     @app.post("/v1/messages/count_tokens")

tests/test_app_routes.py

@@ -317,6 +317,76 @@ async def failing_route_message(body, headers):
     assert "socket closed" in data["error"]["message"]
 
 
+def test_stream_unexpected_exception_returns_sse_error_not_500():
+    """流式路径的未预期异常应返回 SSE error event（HTTP 200 + event: error），而非框架级 500.
+
+    验证 catch-all Exception 处理器将未知异常转换为结构化 SSE 错误事件，
+    客户端可正常解析错误信息而非收到裸 HTTP 500。
+    """
+    app = create_app(ProxyConfig(
+        primary={"enabled": False},
+        fallback={"enabled": True},
+        database={"path": "/tmp/test-coding-proxy-unexpected.db"},
+    ))
+
+    async def failing_route_stream(body, headers):
+        raise ValueError("unexpected parsing error")
+        yield  # pragma: no cover
+
+    app.state.router.route_stream = failing_route_stream
+
+    with TestClient(app) as client:
+        with client.stream(
+            "POST",
+            "/v1/messages",
+            json={
+                "model": "claude-sonnet-4-20250514",
+                "messages": [{"role": "user", "content": "Hi"}],
+                "stream": True,
+            },
+        ) as resp:
+            body = b"".join(resp.iter_bytes()).decode()
+
+    # 关键断言：必须是 200（SSE 流正常关闭），不能是 500
+    assert resp.status_code == 200
+    assert "event: error" in body
+    assert '"type": "api_error"' in body
+    # 确认包含异常类型信息，便于调试定位
+    assert "ValueError" in body or "unexpected" in body.lower()
+
+
+def test_non_stream_unexpected_exception_returns_500_json_not_raw_500():
+    """非流式路径的未预期异常应返回结构化 JSON 错误（含异常类型），而非框架级原始 500.
+
+    验证 catch-all Exception 处理器将未知异常转换为 JSON 格式的 api_error 响应，
+    服务端日志记录完整堆栈（exc_info=True），客户端可从响应体获取错误类型。
+    """
+    app = create_app(ProxyConfig(
+        primary={"enabled": False},
+        fallback={"enabled": True},
+        database={"path": "/tmp/test-coding-proxy-nonstream-unexpected.db"},
+    ))
+
+    async def failing_route_message(body, headers):
+        raise RuntimeError("internal state corruption")
+
+    app.state.router.route_message = failing_route_message
+
+    with TestClient(app) as client:
+        resp = client.post(
+            "/v1/messages",
+            json={
+                "model": "claude-sonnet-4-20250514",
+                "messages": [{"role": "user", "content": "Hi"}],
+            },
+        )
+
+    assert resp.status_code == 500
+    data = resp.json()
+    assert data["error"]["type"] == "api_error"
+    assert "RuntimeError" in data["error"]["message"]
+
+
 def test_messages_normalizes_vendor_tool_blocks_before_routing():
     """入口应先规范化 server_tool_use，再交给高优先级 tier."""
     app = create_app(ProxyConfig(

tests/test_router_executor.py

@@ -346,6 +346,33 @@ async def test_non_last_tier_continues_on_connect_error(self):
         resp = await exec_inst.execute_message({"model": "test"}, {})
         assert resp.status_code == 200
 
+    @pytest.mark.asyncio
+    async def test_unexpected_exception_on_last_tier_propagates(self):
+        """最后一层非流式未预期异常应向上传播."""
+        vendor = _mock_vendor()
+        vendor.send_message.side_effect = KeyError("missing config key")
+        exec_inst = _executor([_make_tier(vendor)])
+
+        with pytest.raises(KeyError, match="missing config key"):
+            await exec_inst.execute_message({"model": "test"}, {})
+
+    @pytest.mark.asyncio
+    async def test_unexpected_exception_on_non_last_tier_continues(self):
+        """非最后一层非流式未预期异常应触发故障转移."""
+        bad = _mock_vendor("bad")
+        bad.send_message.side_effect = RuntimeError("internal state corruption")
+
+        good = _mock_vendor("good")
+        good_resp = VendorResponse(
+            status_code=200, raw_body=b'{}',
+            usage=UsageInfo(input_tokens=1, output_tokens=1),
+        )
+        good.send_message = AsyncMock(return_value=good_resp)
+
+        exec_inst = _executor([_make_tier(bad), _make_tier(good)])
+        resp = await exec_inst.execute_message({"model": "test"}, {})
+        assert resp.status_code == 200
+
 
 # ── execute_stream 测试 ──────────────────────────────────
 
@@ -410,6 +437,51 @@ async def _raise_http(*a, **kw):
             async for _ in exec_inst.execute_stream({"model": "test"}, {}):
                 pass
 
+    @pytest.mark.asyncio
+    async def test_stream_unexpected_exception_continues_to_next_tier(self):
+        """非最后一层流式未预期异常应触发故障转移到下一层."""
+        bad_vendor = _mock_vendor("bad")
+
+        async def _raise_unexpected(*a, **kw):
+            raise ValueError("upstream returned garbage")
+            yield  # noqa: PYS101
+
+        bad_vendor.send_message_stream = _raise_unexpected
+
+        good_vendor = _mock_vendor("good")
+
+        async def _good_stream(*a, **kw):
+            yield b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n'
+
+        good_vendor.send_message_stream = _good_stream
+
+        exec_inst = _executor([
+            _make_tier(bad_vendor),
+            _make_tier(good_vendor),
+        ])
+
+        collected = []
+        async for chunk, name in exec_inst.execute_stream({"model": "test"}, {}):
+            collected.append((chunk, name))
+        assert len(collected) > 0
+        assert name == "good"
+
+    @pytest.mark.asyncio
+    async def test_stream_unexpected_exception_on_last_tier_propagates(self):
+        """最后一层流式未预期异常应向上传播（由 _stream_proxy 接管）."""
+        vendor = _mock_vendor()
+
+        async def _raise_unexpected(*a, **kw):
+            raise RuntimeError("stream corrupted")
+            yield  # noqa: PYS101
+
+        vendor.send_message_stream = _raise_unexpected
+        exec_inst = _executor([_make_tier(vendor)])
+
+        with pytest.raises(RuntimeError, match="stream corrupted"):
+            async for _ in exec_inst.execute_stream({"model": "test"}, {}):
+                pass  # noqa: PLC0107
+
 
 # ── 错误处理测试 ─────────────────────────────────────────