"""LLM 错误分类：httpx 超时族必须可重试（瞬时网络故障不能直接终结 run）。 线上实测：兼容模式流式响应中途 200 OK 后停发数据 ~17s，抛 ``httpx.ReadTimeout``。 旧名单只认 ``ReadError`` / ``RemoteProtocolError``，ReadTimeout 被归为 generic （不可重试），1 次尝试即兜底报错、整个 run 终止。 """ from __future__ import annotations import asyncio import logging from types import SimpleNamespace import httpx import pytest from langchain_core.messages import AIMessage, AIMessageChunk from deerflow.agents.middlewares.llm_error_handling_middleware import LLM_ERROR_MARKER, LLMErrorHandlingMiddleware from deerflow.models.dsml_tool_calls import DSMLToolCallProtocolError from deerflow.models.patched_openai import OpenAICompatibleUpstreamError, PatchedChatOpenAI def _middleware() -> LLMErrorHandlingMiddleware: config = SimpleNamespace( circuit_breaker=SimpleNamespace(failure_threshold=3, recovery_timeout_sec=30), ) middleware = LLMErrorHandlingMiddleware(app_config=config) middleware.retry_base_delay_ms = 0 middleware.retry_cap_delay_ms = 0 return middleware def _openai_model() -> PatchedChatOpenAI: return PatchedChatOpenAI(model="test-model", api_key="test-key") def test_httpx_read_timeout_is_transient() -> None: assert _middleware()._classify_error(httpx.ReadTimeout("read timed out")) == (True, "transient") def test_httpx_timeout_family_is_transient() -> None: middleware = _middleware() for exc in ( httpx.ConnectTimeout("connect timed out"), httpx.WriteTimeout("write timed out"), httpx.PoolTimeout("pool timed out"), httpx.ConnectError("connection refused"), TimeoutError(), ): assert middleware._classify_error(exc) == (True, "transient"), exc def test_value_error_stays_generic() -> None: assert _middleware()._classify_error(ValueError("boom")) == (False, "generic") def test_openai_compatible_embedded_error_is_retriable_upstream() -> None: error = RuntimeError("OpenAI-compatible upstream application error (code=2056)") assert _middleware()._classify_error(error) == (True, "upstream") def test_openai_compatible_null_choices_error_is_retriable_upstream() -> None: error = TypeError("Received response with null value for 'choices'.") assert _middleware()._classify_error(error) == (True, "upstream") def test_dsml_protocol_error_is_retriable_upstream() -> None: error = DSMLToolCallProtocolError("truncated DSML") assert _middleware()._classify_error(error) == (True, "upstream") def test_insufficient_balance_stays_non_retriable_quota() -> None: """The 200-body compatibility fix must not hide a real billing failure.""" class PaymentError(RuntimeError): status_code = 402 error = PaymentError("Insufficient Balance") assert _middleware()._classify_error(error) == (False, "quota") def test_minimax_token_plan_limit_stays_non_retriable_quota() -> None: error = OpenAICompatibleUpstreamError( code=2056, message="Token Plan limit reached", ) assert _middleware()._classify_error(error) == (False, "quota") def test_non_stream_embedded_application_error_is_raised() -> None: response = { "id": "failed-response", "object": "chat.completion", "created": 0, "model": "test-model", "choices": None, "base_resp": { "status_code": 2056, "status_msg": "Token Plan limit reached", }, } with pytest.raises(OpenAICompatibleUpstreamError, match=r"code=2056") as exc_info: _openai_model()._create_chat_result(response) assert exc_info.value.code == 2056 assert "Token Plan limit reached" not in str(exc_info.value) assert _middleware()._classify_error(exc_info.value) == (False, "quota") def test_openai_compatible_error_classifies_detail_without_exposing_it() -> None: secret = "SECRET_PROVIDER_DETAIL" quota_error = OpenAICompatibleUpstreamError( code=2056, message=f"Token Plan limit reached {secret}", ) busy_error = OpenAICompatibleUpstreamError( code=1001, message=f"Server busy, please retry {secret}", ) assert _middleware()._classify_error(quota_error) == (False, "quota") assert _middleware()._classify_error(busy_error) == (True, "busy") assert secret not in str(quota_error) assert secret not in str(busy_error) secret_code = "API_KEY_LIKE_PROVIDER_CODE" code_error = OpenAICompatibleUpstreamError(code=secret_code, message="Server busy") assert secret_code not in str(code_error) def test_stream_embedded_application_error_is_raised() -> None: chunk = { "id": "failed-response", "object": "chat.completion.chunk", "created": 0, "model": "test-model", "choices": [], "base_resp": { "status_code": 2056, "status_msg": "Token Plan limit reached", }, } with pytest.raises(OpenAICompatibleUpstreamError, match=r"code=2056"): _openai_model()._convert_chunk_to_generation_chunk(chunk, AIMessageChunk, None) @pytest.mark.parametrize("streaming", [False, True]) def test_top_level_error_envelope_is_safe_and_retriable(streaming) -> None: secret = "SECRET_RAW_PROVIDER_ERROR_BODY" response = { "id": "failed-response", "object": "chat.completion.chunk" if streaming else "chat.completion", "created": 0, "model": "test-model", "choices": [], "error": { "code": "server_error", "message": f"Server busy; {secret}", }, } with pytest.raises(OpenAICompatibleUpstreamError) as exc_info: if streaming: _openai_model()._convert_chunk_to_generation_chunk(response, AIMessageChunk, None) else: _openai_model()._create_chat_result(response) assert _middleware()._classify_error(exc_info.value) == (True, "busy") assert secret not in str(exc_info.value) def test_normal_empty_usage_chunk_is_not_treated_as_error() -> None: chunk = { "id": "usage-response", "object": "chat.completion.chunk", "created": 0, "model": "test-model", "choices": [], "usage": { "prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5, }, } result = _openai_model()._convert_chunk_to_generation_chunk(chunk, AIMessageChunk, None) assert result is not None assert result.message.usage_metadata == { "input_tokens": 3, "output_tokens": 2, "total_tokens": 5, "input_token_details": {}, "output_token_details": {}, } def test_read_timeout_is_retried_and_recovers() -> None: """第一次 ReadTimeout、第二次成功：必须重试并返回真实结果。""" middleware = _middleware() calls = {"count": 0} async def handler(_request): calls["count"] += 1 if calls["count"] == 1: raise httpx.ReadTimeout("read timed out") return AIMessage(content="ok") result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler)) assert calls["count"] == 2 assert result.content == "ok" def test_persistent_read_timeout_falls_back_with_transient_reason() -> None: """连续超时到重试上限：兜底消息 reason=transient（而非 generic）。""" middleware = _middleware() async def handler(_request): raise httpx.ReadTimeout("read timed out") result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler)) assert isinstance(result, AIMessage) assert result.additional_kwargs[LLM_ERROR_MARKER] == "transient" @pytest.mark.parametrize("asynchronous", [False, True]) def test_provider_error_logs_stable_fields_without_raw_detail(caplog, asynchronous) -> None: middleware = _middleware() secret = "SECRET_PROVIDER_BODY_OR_API_KEY" class ProviderError(RuntimeError): status_code = 503 code = "gateway_timeout" def sync_handler(_request): raise ProviderError(secret) async def async_handler(_request): raise ProviderError(secret) with caplog.at_level( logging.WARNING, logger="deerflow.agents.middlewares.llm_error_handling_middleware", ): if asynchronous: result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), async_handler)) else: result = middleware.wrap_model_call(SimpleNamespace(), sync_handler) assert isinstance(result, AIMessage) log_text = "\n".join(record.getMessage() for record in caplog.records) assert "LLM_CALL_RETRY" in log_text assert "LLM_CALL_FAILED" in log_text assert "reason=transient" in log_text assert "error_type=ProviderError" in log_text assert "status=503" in log_text assert "provider_code=gateway_timeout" in log_text assert secret not in log_text assert all(record.exc_info is None for record in caplog.records)