From d729cade6b7cc24d7e2a1c5b65bc2ae80fe774c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Moura?= Date: Wed, 9 Sep 2026 19:12:57 -0300 Subject: [PATCH] fix(openai): surface gateway errors reported inside an HTTP 200 (#7342) * fix(openai): surface gateway errors reported inside an HTTP 200 OpenAI-compatible gateways commit `200 OK` as soon as the upstream provider accepts a request, so a later provider failure arrives in the body as an `error` object with no `choices`. That reached the SDK's parse helper and surfaced as `TypeError: 'NoneType' object is not iterable`, naming neither the provider, the status, nor the fact that a timeout happened. The four non-streaming paths now inspect the raw body before parsing and raise the exception the upstream code maps to, so a masked 504 is catchable exactly like an honest one. Streaming already had this guard inside the SDK. Co-Authored-By: Claude Opus 5 (1M context) * test(openai): teach the tool-cache fake about with_raw_response The provider now reads the raw body before parsing, so a client double that only implements `create` no longer satisfies it. Same shape as the fixes to the reasoning-effort retry and Snowflake doubles. Co-Authored-By: Claude Opus 5 (1M context) * test(tracing): reset the TraceCollectionListener singleton between tests TraceCollectionListener caches a TraceBatchManager on the class and `_initialized` short-circuits `__init__`, so batch state survives for the whole xdist worker. `test_nested_agent_executor_flow_does_not_finalize_parent_batch` left `trace_batch_id="debug-trace-batch"` behind, which moved every later trace POST from /tracing/ephemeral/batches to /tracing/batches//events. The recorded cassette then stopped matching, the agent retried, and the second call found the cassette consumed -- surfacing as ConnectionError in an unrelated test hundreds of tests later. Reproduced deterministically by running the leaking test followed by tests/tracing/test_trace_enable_disable.py::test_trace_calls_when_enabled_via_env; fails on a68b5e903 too, so this predates the gateway fix it was blocking. An autouse fixture now clears the cached instance after each test. Two canaries pin the invariant and fail without the fixture. Co-Authored-By: Claude Opus 5 (1M context) * test(tracing): drop the unwritable _listeners_setup canary Both review bots flagged that the canary read `_listeners_setup` off the class, where it is always False, so it could never fail. Correct, and the suggested fix does not work either: `BaseEventListener.__init__` calls `setup_listeners` (base_event_listener.py:16), which sets the flag on the instance (trace_listener.py:229), so reading it back through `TraceCollectionListener()` is always True. Neither read observes a leak, so the canary is deleted rather than replaced, with the reasoning recorded so it is not re-added. The same finding showed the fixture was resetting two class attributes that are never assigned at class level. Only dropping `_instance` is load-bearing, so the fixture is now one line. Co-Authored-By: Claude Opus 5 (1M context) * docs(tracing): correct why the _listeners_setup canary is unwritable setup_listeners returns early when tracing is off and no override applies (trace_listener.py:213-220), assigning the flag at :229 only when it actually registers. Construction therefore does not always set it, as the previous note claimed: with tracing disabled the flag never even reaches the instance dict. The instance read reports ambient tracing state rather than isolation, which is a better reason not to assert on it than the one recorded before. Co-Authored-By: Claude Opus 5 (1M context) * fix(tests): clear trace batch state in place instead of dropping the singleton Dropping `TraceCollectionListener._instance` made the next construction re-run `setup_listeners`, re-registering its handlers on the event bus. That broke tests/telemetry/test_task_failure_instrumentation.py, which requires exactly one handler per event: the re-registered `on_task_failed` made two. Verified against a53ecc17f, where the same sequence passes -- the regression was mine. The leak that needed fixing was batch state, not registration, so the fixture now clears the manager's batch fields in place. Handler cleanup already belongs to `cleanup_event_handlers`, and `first_time_handler` keeps its reference to the same manager object. Co-Authored-By: Claude Opus 5 (1M context) * fix(tests): clear tracing context vars so the listener can re-register Bugbot flagged that keeping the singleton leaves `_listeners_setup` set, so after `cleanup_event_handlers` wipes the bus `setup_listeners` returns early (trace_listener.py:208) and tracing silently registers nothing for the rest of the worker. Confirmed: after a tracing-enabled run, re-running setup restores 0 of 119 handler entries. Dropping the singleton fixes that but previously broke test_task_failure_instrumentation. The real cause was a third leak: the `_tracing_enabled` context var stayed set, so the replacement listener still believed tracing was on and re-registered `on_task_failed` next to telemetry's. Clearing the context vars is what makes replacing the listener safe, so the fixture now does both, and a canary pins it (fails with `assert True is False` without the drop). Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- conftest.py | 32 ++ docs/edge/ar/concepts/llms.mdx | 32 ++ docs/edge/en/concepts/llms.mdx | 32 ++ docs/edge/ko/concepts/llms.mdx | 32 ++ docs/edge/pt-BR/concepts/llms.mdx | 32 ++ .../llms/providers/openai/completion.py | 145 ++++- .../openai/test_gateway_error_envelope.py | 540 ++++++++++++++++++ lib/crewai/tests/llms/openai/test_openai.py | 36 +- .../test_tools_reasoning_effort_retry.py | 5 + .../tests/llms/snowflake/test_snowflake.py | 23 +- lib/crewai/tests/test_tool_cache_default.py | 44 +- .../tracing/test_trace_listener_isolation.py | 53 ++ 12 files changed, 969 insertions(+), 37 deletions(-) create mode 100644 lib/crewai/tests/llms/openai/test_gateway_error_envelope.py create mode 100644 lib/crewai/tests/tracing/test_trace_listener_isolation.py diff --git a/conftest.py b/conftest.py index 5987d653f..5936eed5b 100644 --- a/conftest.py +++ b/conftest.py @@ -202,6 +202,38 @@ def cleanup_event_handlers() -> Generator[None, Any, None]: pass +@pytest.fixture(autouse=True, scope="function") +def reset_tracing_state() -> Generator[None, Any, None]: + """Drop the tracing singleton and its context after each test. + + `TraceCollectionListener` is a singleton, so without this three things leak + for the rest of the xdist worker: + + - `TraceBatchManager.trace_batch_id`, which moves later trace POSTs from + `/tracing/ephemeral/batches` to `/tracing/batches//events` until some + unrelated cassette stops matching, naming neither the leak nor its source. + - `_listeners_setup`, which makes `setup_listeners` return early + (`trace_listener.py:208`) after `cleanup_event_handlers` has wiped the bus, + so tracing silently registers nothing and collects no events. + - the `_tracing_enabled` context var, which leaves tracing on for later tests + and re-registers `on_task_failed` alongside telemetry's — breaking + `test_task_failure_instrumentation`, which requires one handler per event. + + All three go together: clearing the context vars is what makes dropping the + singleton safe, because the replacement listener then sees tracing disabled + and registers nothing. Dropping the singleton alone re-registers handlers and + breaks the telemetry test. + """ + yield + + from crewai.events.listeners.tracing import utils as tracing_utils + from crewai.events.listeners.tracing.trace_listener import TraceCollectionListener + + tracing_utils._tracing_enabled.set(None) + tracing_utils._tui_mode.set(False) + TraceCollectionListener._instance = None + + @pytest.fixture(autouse=True, scope="function") def reset_event_state() -> None: """Reset event system state before each test for isolation.""" diff --git a/docs/edge/ar/concepts/llms.mdx b/docs/edge/ar/concepts/llms.mdx index a2b3d9653..863a80ee3 100644 --- a/docs/edge/ar/concepts/llms.mdx +++ b/docs/edge/ar/concepts/llms.mdx @@ -1339,6 +1339,38 @@ llm = LLM( llm = LLM(model="gpt-4") ``` + + + تُعيد البوابات مثل OpenRouter الرمز `200 OK` بمجرد قبول المزود الأساسي للطلب، لذلك يصل انتهاء مهلة المزود داخل جسم الاستجابة بدلًا من رمز الحالة. + + + تُطلق CrewAI الاستثناء نفسه الذي كان رمز الحالة الأصلي سيُنتجه، ومن ثم يلتقط منطق إعادة المحاولة الموجود لديك هذا الفشل المُقنَّع: + + ```python + import openai + from pydantic import BaseModel + + from crewai import LLM + + + class Report(BaseModel): + summary: str + + + llm = LLM(model="openrouter/z-ai/glm-5.3", response_format=Report) + + try: + result = llm.call("Summarize the incident", response_model=Report) + except openai.InternalServerError as e: + # "z-ai/glm-5.3 via openrouter.ai returned HTTP 200 with an upstream error + # and no choices: The operation was aborted (upstream code 504)" + print(f"Upstream provider failed, safe to retry: {e}") + ``` + + + إن استخدام `response_model` كبير أو متداخل بعمق يزيد احتمال انتهاء مهلة المزود. تعامل مع هذه الحالات كأعطال مؤقتة في المزود، وليس كإنتاج النموذج مخرجات منظمة تالفة. + + استخدم نماذج سياق أكبر للمهام الواسعة diff --git a/docs/edge/en/concepts/llms.mdx b/docs/edge/en/concepts/llms.mdx index 02fb97314..957a52e94 100644 --- a/docs/edge/en/concepts/llms.mdx +++ b/docs/edge/en/concepts/llms.mdx @@ -1487,6 +1487,38 @@ llm = LLM( llm = LLM(model="gpt-4") ``` + + + Gateways such as OpenRouter return `200 OK` as soon as the upstream provider accepts the request, so a provider timeout arrives in the response body instead of the status code. + + + CrewAI raises the same exception the upstream code would have produced as a real HTTP status, so a masked failure is caught by the retry handling you already have: + + ```python + import openai + from pydantic import BaseModel + + from crewai import LLM + + + class Report(BaseModel): + summary: str + + + llm = LLM(model="openrouter/z-ai/glm-5.3", response_format=Report) + + try: + result = llm.call("Summarize the incident", response_model=Report) + except openai.InternalServerError as e: + # "z-ai/glm-5.3 via openrouter.ai returned HTTP 200 with an upstream error + # and no choices: The operation was aborted (upstream code 504)" + print(f"Upstream provider failed, safe to retry: {e}") + ``` + + + A large or deeply nested `response_model` makes upstream timeouts more likely. Treat these as transient provider failures, not as the model producing malformed structured output. + + Use larger context models for extensive tasks diff --git a/docs/edge/ko/concepts/llms.mdx b/docs/edge/ko/concepts/llms.mdx index 760377ac1..63b3f861c 100644 --- a/docs/edge/ko/concepts/llms.mdx +++ b/docs/edge/ko/concepts/llms.mdx @@ -995,6 +995,38 @@ LLM 설정을 최대한 활용하는 방법을 알아보세요: llm = LLM(model="gpt-4") ``` + + + OpenRouter와 같은 게이트웨이는 업스트림 공급자가 요청을 수락하는 즉시 `200 OK`를 반환하므로, 공급자 타임아웃은 상태 코드가 아니라 응답 본문에 담겨 도착합니다. + + + CrewAI는 해당 업스트림 코드가 실제 HTTP 상태로 왔을 때 발생시킬 예외와 동일한 예외를 발생시키므로, 이미 구성해 둔 재시도 처리로 가려진 실패를 잡을 수 있습니다: + + ```python + import openai + from pydantic import BaseModel + + from crewai import LLM + + + class Report(BaseModel): + summary: str + + + llm = LLM(model="openrouter/z-ai/glm-5.3", response_format=Report) + + try: + result = llm.call("Summarize the incident", response_model=Report) + except openai.InternalServerError as e: + # "z-ai/glm-5.3 via openrouter.ai returned HTTP 200 with an upstream error + # and no choices: The operation was aborted (upstream code 504)" + print(f"Upstream provider failed, safe to retry: {e}") + ``` + + + 크거나 깊게 중첩된 `response_model`은 업스트림 타임아웃 가능성을 높입니다. 이를 모델이 잘못된 구조화 출력을 생성한 것으로 보지 말고, 일시적인 공급자 장애로 처리하세요. + + 대규모 작업에는 더 큰 컨텍스트 모델을 사용하세요. diff --git a/docs/edge/pt-BR/concepts/llms.mdx b/docs/edge/pt-BR/concepts/llms.mdx index c4cf18ecf..45ef20bbc 100644 --- a/docs/edge/pt-BR/concepts/llms.mdx +++ b/docs/edge/pt-BR/concepts/llms.mdx @@ -919,6 +919,38 @@ Saiba como obter o máximo da configuração do seu LLM: llm = LLM(model="gpt-4") ``` + + + Gateways como o OpenRouter retornam `200 OK` assim que o provedor upstream aceita a requisição, então um timeout do provedor chega no corpo da resposta em vez do código de status. + + + O CrewAI lança a mesma exceção que o código upstream produziria como um status HTTP real, portanto uma falha mascarada é capturada pelo tratamento de retry que você já possui: + + ```python + import openai + from pydantic import BaseModel + + from crewai import LLM + + + class Report(BaseModel): + summary: str + + + llm = LLM(model="openrouter/z-ai/glm-5.3", response_format=Report) + + try: + result = llm.call("Summarize the incident", response_model=Report) + except openai.InternalServerError as e: + # "z-ai/glm-5.3 via openrouter.ai returned HTTP 200 with an upstream error + # and no choices: The operation was aborted (upstream code 504)" + print(f"Upstream provider failed, safe to retry: {e}") + ``` + + + Um `response_model` grande ou profundamente aninhado aumenta a chance de timeouts upstream. Trate esses casos como falhas transitórias do provedor, e não como o modelo produzindo saída estruturada malformada. + + Use modelos de contexto expandido para tarefas extensas diff --git a/lib/crewai/src/crewai/llms/providers/openai/completion.py b/lib/crewai/src/crewai/llms/providers/openai/completion.py index 280c314ea..75e7fb7a7 100644 --- a/lib/crewai/src/crewai/llms/providers/openai/completion.py +++ b/lib/crewai/src/crewai/llms/providers/openai/completion.py @@ -1,20 +1,28 @@ from __future__ import annotations -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Mapping from dataclasses import dataclass, field import json import logging import os -from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict +from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, TypedDict import httpx from openai import ( APIConnectionError, + APIResponseValidationError, + APIStatusError, AsyncOpenAI, + AuthenticationError, BadRequestError, + ConflictError, + InternalServerError, NotFoundError, OpenAI, + PermissionDeniedError, + RateLimitError, Stream, + UnprocessableEntityError, ) from openai.lib.streaming.chat import ChatCompletionStream from openai.types.chat import ( @@ -70,6 +78,95 @@ if TYPE_CHECKING: # per process rather than on every call. _LEARNED_RESPONSES_ONLY_MODELS: set[str] = set() +# Upstream status codes carried inside a 200 body, mapped to the exception the SDK +# raises when the same code arrives as a real HTTP status. Keeping the classes +# identical means a gateway-masked failure is catchable by whatever already handles +# the honest one. Codes outside this table fall back to `InternalServerError` for +# 5xx, otherwise `APIStatusError`. +_UPSTREAM_STATUS_ERRORS: Final[dict[int, type[APIStatusError]]] = { + 400: BadRequestError, + 401: AuthenticationError, + 403: PermissionDeniedError, + 404: NotFoundError, + 409: ConflictError, + 422: UnprocessableEntityError, + 429: RateLimitError, +} + + +def _upstream_status_code(error: Mapping[str, Any]) -> int | None: + """Read an HTTP-like status code out of a gateway error object. + + OpenAI-style bodies put a slug in `code` ("model_not_found"), gateways put the + upstream status there instead; only the latter is a status code. + """ + code = error.get("code") + if isinstance(code, bool): + return None + if isinstance(code, int): + return code if 400 <= code <= 599 else None + if isinstance(code, str) and code.isdigit(): + parsed = int(code) + return parsed if 400 <= parsed <= 599 else None + return None + + +def _raise_for_upstream_error( + body: str, + *, + model: str, + http_response: httpx.Response, +) -> None: + """Raise when a 200 response carries an upstream error instead of choices. + + Gateways commit `200 OK` as soon as a provider accepts the request, so a later + provider failure is reported in the body -- an `error` object and no `choices`. + The OpenAI SDK guards this for streams (`openai/_streaming.py`) but not for + non-streaming responses, where the absent `choices` surfaces from inside the + parse helper as `TypeError: 'NoneType' object is not iterable`, naming neither + the provider nor the status. + """ + try: + payload = json.loads(body) + except ValueError: + # Not JSON, so not an error envelope. A non-JSON 200 already has its own + # (pre-existing) handling in the SDK, which returns the body as `str`. + return + + if not isinstance(payload, Mapping) or payload.get("choices"): + return + + host = http_response.request.url.host + error = payload.get("error") + + if not isinstance(error, Mapping): + raise APIResponseValidationError( + response=http_response, + body=payload, + message=( + f"{model} via {host} returned HTTP 200 with no choices and no error " + f"object; the response does not describe a completion" + ), + ) + + detail = error.get("message") or "no message given" + code = _upstream_status_code(error) + suffix = f" (upstream code {code})" if code is not None else "" + message = ( + f"{model} via {host} returned HTTP 200 with an upstream error and no " + f"choices: {detail}{suffix}" + ) + + if code is None: + raise APIResponseValidationError( + response=http_response, body=error, message=message + ) + + error_cls = _UPSTREAM_STATUS_ERRORS.get(code) or ( + InternalServerError if code >= 500 else APIStatusError + ) + raise error_cls(message, response=http_response, body=error) + class WebSearchResult(TypedDict, total=False): """Result from web search built-in tool.""" @@ -1904,10 +2001,16 @@ class OpenAICompletion(BaseLLM): parse_params = { k: v for k, v in params.items() if k != "response_format" } - parsed_response = self._get_sync_client().beta.chat.completions.parse( + raw_parsed = self._get_sync_client().beta.chat.completions.with_raw_response.parse( **parse_params, response_format=response_model, ) + _raise_for_upstream_error( + raw_parsed.text, + model=self.model, + http_response=raw_parsed.http_response, + ) + parsed_response = raw_parsed.parse() math_reasoning = parsed_response.choices[0].message if math_reasoning.refusal: @@ -1933,9 +2036,17 @@ class OpenAICompletion(BaseLLM): ) return parsed_object - response: ChatCompletion = self._get_sync_client().chat.completions.create( - **params + raw_response = ( + self._get_sync_client().chat.completions.with_raw_response.create( + **params + ) ) + _raise_for_upstream_error( + raw_response.text, + model=self.model, + http_response=raw_response.http_response, + ) + response: ChatCompletion = raw_response.parse() usage = self._extract_openai_token_usage(response) @@ -2330,12 +2441,16 @@ class OpenAICompletion(BaseLLM): parse_params = { k: v for k, v in params.items() if k != "response_format" } - parsed_response = ( - await self._get_async_client().beta.chat.completions.parse( - **parse_params, - response_format=response_model, - ) + raw_parsed = await self._get_async_client().beta.chat.completions.with_raw_response.parse( + **parse_params, + response_format=response_model, ) + _raise_for_upstream_error( + raw_parsed.text, + model=self.model, + http_response=raw_parsed.http_response, + ) + parsed_response = raw_parsed.parse() math_reasoning = parsed_response.choices[0].message if math_reasoning.refusal: @@ -2361,9 +2476,15 @@ class OpenAICompletion(BaseLLM): ) return parsed_object - response: ChatCompletion = ( - await self._get_async_client().chat.completions.create(**params) + raw_response = await self._get_async_client().chat.completions.with_raw_response.create( + **params ) + _raise_for_upstream_error( + raw_response.text, + model=self.model, + http_response=raw_response.http_response, + ) + response: ChatCompletion = raw_response.parse() usage = self._extract_openai_token_usage(response) diff --git a/lib/crewai/tests/llms/openai/test_gateway_error_envelope.py b/lib/crewai/tests/llms/openai/test_gateway_error_envelope.py new file mode 100644 index 000000000..002f98bf4 --- /dev/null +++ b/lib/crewai/tests/llms/openai/test_gateway_error_envelope.py @@ -0,0 +1,540 @@ +"""Gateways that report upstream failures inside an HTTP 200 body. + +OpenRouter (and other OpenAI-compatible gateways) commit ``200 OK`` as soon as the +upstream provider accepts the request, so a later provider failure arrives as an +``error`` object with no ``choices``. Without a guard the absent ``choices`` reaches +the OpenAI SDK's parse helper and surfaces as +``TypeError: 'NoneType' object is not iterable``, naming neither the provider, the +status, nor the fact that a timeout happened. + +These tests drive the real OpenAI SDK over ``httpx.MockTransport``, so the parse +path under test is the one that runs in production. No network is involved. +""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import openai +import pytest +from pydantic import BaseModel + +from crewai.llms.providers.openai.completion import OpenAICompletion +from crewai.llms.providers.openai_compatible.completion import ( + OpenAICompatibleCompletion, +) + + +BASE_URL = "https://openrouter.ai/api/v1" + + +class Answer(BaseModel): + """Minimal structured-output target.""" + + a: str + + +def _error_envelope(message: str, code: object) -> dict[str, Any]: + """A 200 body holding only an upstream error, as gateways send it.""" + return {"error": {"message": message, "code": code}, "id": "gen-abc123"} + + +def _completion_body(content: str) -> dict[str, Any]: + return { + "id": "gen-ok", + "object": "chat.completion", + "created": 1, + "model": "z-ai/glm-5.3", + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "message": {"role": "assistant", "content": content, "refusal": None}, + } + ], + "usage": {"prompt_tokens": 3, "completion_tokens": 4, "total_tokens": 7}, + } + + +def _tool_call_body() -> dict[str, Any]: + body = _completion_body("") + body["choices"][0]["finish_reason"] = "tool_calls" + body["choices"][0]["message"] = { + "role": "assistant", + "content": None, + "refusal": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Lisbon"}'}, + } + ], + } + return body + + +def _sse(chunks: list[dict[str, Any]]) -> bytes: + payload = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks) + return (payload + "data: [DONE]\n\n").encode() + + +def _stream_error_chunk() -> dict[str, Any]: + """OpenRouter's documented mid-stream error shape: error plus an empty delta.""" + return { + "id": "gen-abc123", + "object": "chat.completion.chunk", + "created": 1, + "model": "z-ai/glm-5.3", + "provider": "Z.AI", + "error": {"code": 504, "message": "The operation was aborted"}, + "choices": [ + {"index": 0, "delta": {"content": ""}, "finish_reason": "error"} + ], + } + + +def _make_llm( + responder: Any, + *, + cls: type[OpenAICompletion] = OpenAICompletion, + stream: bool = False, + **kwargs: Any, +) -> OpenAICompletion: + """Build a provider whose SDK clients are pinned to a mock transport. + + Replaces the private clients rather than patching module globals so the test + leaves no state behind for whatever runs next. + """ + llm = cls(model=kwargs.pop("model", "z-ai/glm-5.3"), api_key="sk-test", stream=stream, **kwargs) + transport = httpx.MockTransport(responder) + llm._client = openai.OpenAI( + api_key="sk-test", base_url=BASE_URL, http_client=httpx.Client(transport=transport) + ) + llm._async_client = openai.AsyncOpenAI( + api_key="sk-test", + base_url=BASE_URL, + http_client=httpx.AsyncClient(transport=transport), + ) + return llm + + +def _json_responder(body: dict[str, Any], status: int = 200) -> Any: + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response(status, json=body) + + return respond + + +# --------------------------------------------------------------------------- # +# The reported defect, across the four non-streaming paths +# --------------------------------------------------------------------------- # + + +def test_sync_structured_surfaces_upstream_timeout_not_typeerror() -> None: + """The reported case: response_model= through a gateway that timed out.""" + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises( + openai.InternalServerError, + match=r"z-ai/glm-5\.3 via openrouter\.ai returned HTTP 200 with an upstream " + r"error and no choices: The operation was aborted \(upstream code 504\)", + ): + llm.call("hi", response_model=Answer) + + +def test_sync_plain_surfaces_upstream_timeout_not_typeerror() -> None: + """Not only structured calls: the plain path had the same defect.""" + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises( + openai.InternalServerError, + match=r"returned HTTP 200 with an upstream error and no choices: " + r"The operation was aborted \(upstream code 504\)", + ): + llm.call("hi") + + +@pytest.mark.asyncio +async def test_async_structured_surfaces_upstream_timeout() -> None: + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises( + openai.InternalServerError, + match=r"The operation was aborted \(upstream code 504\)", + ): + await llm.acall("hi", response_model=Answer) + + +@pytest.mark.asyncio +async def test_async_plain_surfaces_upstream_timeout() -> None: + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises( + openai.InternalServerError, + match=r"The operation was aborted \(upstream code 504\)", + ): + await llm.acall("hi") + + +def test_error_message_never_mentions_nonetype() -> None: + """The whole point: the raised error must be actionable.""" + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises(openai.APIStatusError) as caught: + llm.call("hi", response_model=Answer) + + text = str(caught.value) + assert "NoneType" not in text + assert "z-ai/glm-5.3" in text + assert "openrouter.ai" in text + assert "504" in text + + +# --------------------------------------------------------------------------- # +# Upstream status codes map to the class the SDK uses for the honest error +# --------------------------------------------------------------------------- # + + +def test_upstream_429_raises_rate_limit_error() -> None: + """A masked rate limit stays catchable as openai.RateLimitError.""" + llm = _make_llm(_json_responder(_error_envelope("Slow down", 429))) + + with pytest.raises(openai.RateLimitError, match="Slow down"): + llm.call("hi") + + +def test_upstream_502_raises_internal_server_error() -> None: + llm = _make_llm(_json_responder(_error_envelope("Bad gateway", 502))) + + with pytest.raises(openai.InternalServerError, match=r"Bad gateway"): + llm.call("hi") + + +def test_upstream_400_raises_bad_request_error() -> None: + llm = _make_llm(_json_responder(_error_envelope("Malformed schema", 400))) + + with pytest.raises(openai.BadRequestError, match="Malformed schema"): + llm.call("hi") + + +def test_upstream_error_body_is_attached_to_exception() -> None: + """Callers inspecting `.body` get the gateway's own error object.""" + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + + with pytest.raises(openai.InternalServerError) as caught: + llm.call("hi") + + assert caught.value.body == {"message": "The operation was aborted", "code": 504} + assert caught.value.status_code == 200 + + +def test_string_status_code_is_understood() -> None: + """Some gateways stringify the upstream status.""" + llm = _make_llm(_json_responder(_error_envelope("Slow down", "429"))) + + with pytest.raises(openai.RateLimitError, match=r"upstream code 429"): + llm.call("hi") + + +def test_openai_style_slug_code_is_not_treated_as_a_status() -> None: + """`code` is a slug in OpenAI-style bodies, not an HTTP status.""" + llm = _make_llm(_json_responder(_error_envelope("no such model", "model_not_found"))) + + with pytest.raises(openai.APIResponseValidationError, match="no such model"): + llm.call("hi") + + # and the message must not invent a status code + llm2 = _make_llm(_json_responder(_error_envelope("no such model", "model_not_found"))) + with pytest.raises(openai.APIResponseValidationError) as caught: + llm2.call("hi") + assert "upstream code" not in str(caught.value) + + +def test_error_without_message_still_names_the_model() -> None: + llm = _make_llm(_json_responder({"error": {"code": 504}, "id": "gen-x"})) + + with pytest.raises(openai.InternalServerError, match="no message given"): + llm.call("hi") + + +# --------------------------------------------------------------------------- # +# Malformed 200s that carry no error object either +# --------------------------------------------------------------------------- # + + +def test_missing_choices_without_error_object_is_a_validation_error() -> None: + """Distinguishable from a retryable upstream failure, which is the report's ask.""" + llm = _make_llm(_json_responder({"id": "gen-x", "object": "chat.completion"})) + + with pytest.raises( + openai.APIResponseValidationError, + match="returned HTTP 200 with no choices and no error object", + ): + llm.call("hi") + + +def test_empty_choices_list_is_reported_not_indexerror() -> None: + """`choices: []` used to reach `choices[0]` and raise IndexError.""" + body = _completion_body("hi") + body["choices"] = [] + llm = _make_llm(_json_responder(body)) + + with pytest.raises( + openai.APIResponseValidationError, match="no choices and no error object" + ): + llm.call("hi") + + +# --------------------------------------------------------------------------- # +# Existing behaviour that must not change +# --------------------------------------------------------------------------- # + + +def test_happy_path_plain_completion_unchanged() -> None: + llm = _make_llm(_json_responder(_completion_body("hello there"))) + + assert llm.call("hi") == "hello there" + + +def test_happy_path_structured_completion_unchanged() -> None: + llm = _make_llm(_json_responder(_completion_body('{"a": "hi"}'))) + + result = llm.call("hi", response_model=Answer) + + assert isinstance(result, Answer) + assert result.a == "hi" + + +@pytest.mark.asyncio +async def test_happy_path_async_plain_unchanged() -> None: + llm = _make_llm(_json_responder(_completion_body("hello there"))) + + assert await llm.acall("hi") == "hello there" + + +@pytest.mark.asyncio +async def test_happy_path_async_structured_unchanged() -> None: + llm = _make_llm(_json_responder(_completion_body('{"a": "hi"}'))) + + result = await llm.acall("hi", response_model=Answer) + + assert isinstance(result, Answer) + assert result.a == "hi" + + +def test_token_usage_still_tracked_on_happy_path() -> None: + """with_raw_response must not cost us usage accounting.""" + llm = _make_llm(_json_responder(_completion_body("hello there"))) + + llm.call("hi") + + usage = llm.get_token_usage_summary() + assert (usage.prompt_tokens, usage.completion_tokens, usage.total_tokens) == (3, 4, 7) + + +def test_tool_calls_still_returned_to_the_caller() -> None: + """The tool-call branch reads `.choices` too and must be unaffected.""" + llm = _make_llm(_json_responder(_tool_call_body())) + + result = llm.call("what is the weather in Lisbon?") + + assert isinstance(result, list) + assert result[0].function.name == "get_weather" + + +def test_tool_execution_follow_up_turn_unchanged() -> None: + """Tool result -> follow-up completion, the path after a tool executes.""" + calls: list[int] = [] + + def respond(request: httpx.Request) -> httpx.Response: + calls.append(1) + if len(calls) == 1: + return httpx.Response(200, json=_tool_call_body()) + return httpx.Response(200, json=_completion_body("It is sunny in Lisbon")) + + llm = _make_llm(respond) + + result = llm.call( + "what is the weather in Lisbon?", + available_functions={"get_weather": lambda city: f"sunny in {city}"}, + ) + + assert result == "sunny in Lisbon" + + +def test_real_http_error_status_still_raises_its_own_type() -> None: + """A genuine 429 was already handled; it must not route through the new guard.""" + llm = _make_llm( + _json_responder({"error": {"message": "rate limited"}}, status=429) + ) + + with pytest.raises(openai.RateLimitError): + llm.call("hi") + + +def test_length_finish_reason_still_raises_through_raw_response() -> None: + """`with_raw_response` must not drop the SDK's own parse-time errors.""" + body = _completion_body('{"a": "hi"}') + body["choices"][0]["finish_reason"] = "length" + llm = _make_llm(_json_responder(body)) + + with pytest.raises(Exception) as caught: + llm.call("hi", response_model=Answer) + + assert isinstance(caught.value, openai.LengthFinishReasonError) + + +def test_non_json_200_body_is_left_to_existing_handling() -> None: + """Our guard must not turn a non-JSON 200 into a JSONDecodeError.""" + + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, content=b"gateway boom", headers={"content-type": "text/html"} + ) + + llm = _make_llm(respond) + + with pytest.raises(Exception) as caught: + llm.call("hi") + + assert not isinstance(caught.value, json.JSONDecodeError) + + +def test_masked_404_does_not_trigger_the_responses_api_fallback() -> None: + """`_is_responses_only_error` keys off OpenAI's wording, not any 404. + + A gateway 404 inside a 200 must not send the call to /v1/responses. + """ + seen: list[str] = [] + + def respond(request: httpx.Request) -> httpx.Response: + seen.append(request.url.path) + return httpx.Response(200, json=_error_envelope("No endpoints found", 404)) + + llm = _make_llm(respond) + + with pytest.raises(ValueError, match="No endpoints found"): + llm.call("hi") + + assert all("/responses" not in path for path in seen), seen + + +# --------------------------------------------------------------------------- # +# Streaming already had a guard in the SDK; pin that it stays intact +# --------------------------------------------------------------------------- # + + +def test_sync_streaming_error_chunk_already_surfaces_upstream_message() -> None: + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + content=_sse([_stream_error_chunk()]), + headers={"content-type": "text/event-stream"}, + ) + + llm = _make_llm(respond, stream=True) + + with pytest.raises(Exception, match="The operation was aborted"): + llm.call("hi") + + +@pytest.mark.asyncio +async def test_async_streaming_error_chunk_already_surfaces_upstream_message() -> None: + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + content=_sse([_stream_error_chunk()]), + headers={"content-type": "text/event-stream"}, + ) + + llm = _make_llm(respond, stream=True) + + with pytest.raises(Exception, match="The operation was aborted"): + await llm.acall("hi") + + +# --------------------------------------------------------------------------- # +# The subclass users actually reach OpenRouter through +# --------------------------------------------------------------------------- # + + +def test_openai_compatible_subclass_inherits_the_guard() -> None: + """OpenRouter routes through OpenAICompatibleCompletion, not OpenAICompletion.""" + llm = _make_llm( + _json_responder(_error_envelope("The operation was aborted", 504)), + cls=OpenAICompatibleCompletion, + model="z-ai/glm-5.3", + provider="openrouter", + ) + + with pytest.raises( + openai.InternalServerError, + match=r"The operation was aborted \(upstream code 504\)", + ): + llm.call("hi", response_model=Answer) + + +# --------------------------------------------------------------------------- # +# Every entry in the status table, and what the message is allowed to contain +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + ("upstream_code", "expected"), + [ + (400, openai.BadRequestError), + (401, openai.AuthenticationError), + (403, openai.PermissionDeniedError), + (404, openai.NotFoundError), + (409, openai.ConflictError), + (422, openai.UnprocessableEntityError), + (429, openai.RateLimitError), + (500, openai.InternalServerError), + (503, openai.InternalServerError), + (504, openai.InternalServerError), + ], +) +def test_every_upstream_code_maps_to_its_sdk_exception( + upstream_code: int, expected: type[Exception] +) -> None: + """A masked failure must be catchable exactly like the honest one.""" + llm = _make_llm(_json_responder(_error_envelope("upstream said no", upstream_code))) + + # 404 is rewritten to ValueError by the provider's own model-not-found handler, + # which is the same thing it does for a real 404. + if expected is openai.NotFoundError: + with pytest.raises(ValueError, match="upstream said no"): + llm.call("hi") + return + + with pytest.raises(expected, match=f"upstream code {upstream_code}"): + llm.call("hi") + + +def test_message_names_the_host_but_never_url_credentials() -> None: + """The message reaches logs and the model context, so it carries only the host. + + Some gateways accept credentials in the URL; the guard reports + `request.url.host` rather than the URL so those are not echoed. + """ + llm = _make_llm(_json_responder(_error_envelope("The operation was aborted", 504))) + llm._client = openai.OpenAI( + api_key="sk-test", + base_url="https://user:sk-secret123@gw.example/api/v1?api_key=sk-leak456", + http_client=httpx.Client( + transport=httpx.MockTransport( + _json_responder(_error_envelope("The operation was aborted", 504)) + ) + ), + ) + + with pytest.raises(openai.InternalServerError) as caught: + llm.call("hi") + + text = str(caught.value) + assert "gw.example" in text + assert "sk-secret123" not in text + assert "sk-leak456" not in text diff --git a/lib/crewai/tests/llms/openai/test_openai.py b/lib/crewai/tests/llms/openai/test_openai.py index 3dec38f37..994092acd 100644 --- a/lib/crewai/tests/llms/openai/test_openai.py +++ b/lib/crewai/tests/llms/openai/test_openai.py @@ -1,3 +1,4 @@ +import json import os import sys import types @@ -474,6 +475,23 @@ def test_openai_raises_error_when_model_not_supported(): with pytest.raises(ValueError, match="Model.*not found"): llm.call("Hello") + +def _raw_create_double(content: str = "test response") -> MagicMock: + """Double for `chat.completions.with_raw_response.create`. + + `text` has to be real JSON carrying `choices` so the gateway error-envelope + guard sees a well-formed completion and defers to `parse()`. + """ + parsed = MagicMock( + choices=[MagicMock(message=MagicMock(content=content, tool_calls=None))], + usage=MagicMock(prompt_tokens=10, completion_tokens=20, total_tokens=30), + ) + return MagicMock( + text=json.dumps({"choices": [{"index": 0}]}), + **{"parse.return_value": parsed}, + ) + + def test_openai_client_setup_with_extra_arguments(): """ Test that OpenAICompletion is initialized with correct parameters @@ -494,11 +512,10 @@ def test_openai_client_setup_with_extra_arguments(): assert llm._client.max_retries == 3 assert llm._client.timeout == 30 - with patch.object(llm._client.chat.completions, 'create') as mock_create: - mock_create.return_value = MagicMock( - choices=[MagicMock(message=MagicMock(content="test response", tool_calls=None))], - usage=MagicMock(prompt_tokens=10, completion_tokens=20, total_tokens=30) - ) + with patch.object( + llm._client.chat.completions.with_raw_response, 'create' + ) as mock_create: + mock_create.return_value = _raw_create_double() llm.call("Hello") @@ -514,11 +531,10 @@ def test_extra_arguments_are_passed_to_openai_completion(): """ llm = LLM(model="gpt-4o", temperature=0.7, max_tokens=1000, top_p=0.5, max_retries=3) - with patch.object(llm._client.chat.completions, 'create') as mock_create: - mock_create.return_value = MagicMock( - choices=[MagicMock(message=MagicMock(content="test response", tool_calls=None))], - usage=MagicMock(prompt_tokens=10, completion_tokens=20, total_tokens=30) - ) + with patch.object( + llm._client.chat.completions.with_raw_response, 'create' + ) as mock_create: + mock_create.return_value = _raw_create_double() llm.call("Hello, how are you?") diff --git a/lib/crewai/tests/llms/openai/test_tools_reasoning_effort_retry.py b/lib/crewai/tests/llms/openai/test_tools_reasoning_effort_retry.py index 4cd7e62c0..f6a3e9aab 100644 --- a/lib/crewai/tests/llms/openai/test_tools_reasoning_effort_retry.py +++ b/lib/crewai/tests/llms/openai/test_tools_reasoning_effort_retry.py @@ -199,6 +199,11 @@ class TestRetryBehaviour: def create(self, **kwargs): raise tools_effort_error() + @property + def with_raw_response(self): + """The provider reads the raw body to spot 200-wrapped errors.""" + return self + monkeypatch.setattr( llm, "_get_sync_client", diff --git a/lib/crewai/tests/llms/snowflake/test_snowflake.py b/lib/crewai/tests/llms/snowflake/test_snowflake.py index f27e2b5ff..552088044 100644 --- a/lib/crewai/tests/llms/snowflake/test_snowflake.py +++ b/lib/crewai/tests/llms/snowflake/test_snowflake.py @@ -1,8 +1,10 @@ from __future__ import annotations +import json from types import SimpleNamespace from unittest.mock import Mock, patch +import httpx import pytest from crewai.llm import LLM @@ -411,9 +413,26 @@ class TestSnowflakeRequests: ) ], ) - create = Mock(return_value=fake_response) + # The provider reads the raw body first, to spot upstream errors that a + # gateway reported inside an HTTP 200. + create = Mock( + return_value=SimpleNamespace( + text=json.dumps({"choices": [{"index": 0}]}), + parse=lambda: fake_response, + http_response=httpx.Response( + 200, + request=httpx.Request( + "POST", "https://acct.snowflakecomputing.com/api/v2/cortex" + ), + ), + ) + ) fake_client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) + chat=SimpleNamespace( + completions=SimpleNamespace( + with_raw_response=SimpleNamespace(create=create) + ) + ) ) with patch.object(llm, "_get_sync_client", return_value=fake_client): diff --git a/lib/crewai/tests/test_tool_cache_default.py b/lib/crewai/tests/test_tool_cache_default.py index 7a503d269..07ae93733 100644 --- a/lib/crewai/tests/test_tool_cache_default.py +++ b/lib/crewai/tests/test_tool_cache_default.py @@ -14,6 +14,10 @@ identical tool calls followed by a final answer, mirroring the EPD-180 clean-room repro. """ +import json +from types import SimpleNamespace + +import httpx from openai.types.chat import ChatCompletion from pydantic import BaseModel, Field @@ -80,22 +84,36 @@ def make_scripted_llm(): def __init__(self): self.n = 0 + @property + def with_raw_response(self): + """The provider reads the raw body to spot upstream errors that a + gateway reported inside an HTTP 200.""" + return self + def create(self, **params): choice = scripted[min(self.n, len(scripted) - 1)] self.n += 1 - return ChatCompletion.model_validate( - { - "id": f"chatcmpl-fake-{self.n}", - "object": "chat.completion", - "created": 1, - "model": params.get("model", "gpt-4o"), - "choices": [choice], - "usage": { - "prompt_tokens": 10, - "completion_tokens": 5, - "total_tokens": 15, - }, - } + payload = { + "id": f"chatcmpl-fake-{self.n}", + "object": "chat.completion", + "created": 1, + "model": params.get("model", "gpt-4o"), + "choices": [choice], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + }, + } + return SimpleNamespace( + text=json.dumps(payload), + parse=lambda: ChatCompletion.model_validate(payload), + http_response=httpx.Response( + 200, + request=httpx.Request( + "POST", "https://api.openai.com/v1/chat/completions" + ), + ), ) class FakeClient: diff --git a/lib/crewai/tests/tracing/test_trace_listener_isolation.py b/lib/crewai/tests/tracing/test_trace_listener_isolation.py new file mode 100644 index 000000000..8cd87261b --- /dev/null +++ b/lib/crewai/tests/tracing/test_trace_listener_isolation.py @@ -0,0 +1,53 @@ +"""`TraceCollectionListener` is a process-wide singleton, so its state leaks. + +`TraceBatchManager` is cached on the listener class and `_initialized` +short-circuits `__init__`, so anything a test leaves on the batch manager applies +to every later test in the same xdist worker. When `trace_batch_id` leaks, trace +POSTs move from `/tracing/ephemeral/batches` to +`/tracing/batches//events`, the recorded cassette stops matching, and +the failure lands in whichever unrelated test happens to run later — the symptom +names neither the leak nor the test that caused it. + +The autouse `reset_tracing_state` fixture in the root `conftest.py` drops the +singleton and the tracing context vars after every test. These are canaries for +it: run on their own they pass trivially, but under the full suite in random +order they fail if the fixture stops working. + +The context vars matter as much as the singleton. Dropping the listener alone +makes its replacement re-register `on_task_failed` while `_tracing_enabled` is +still set, which breaks `test_task_failure_instrumentation` (one handler per +event). Clearing the context is what makes replacing the listener safe. +""" + +from __future__ import annotations + +from crewai.events.listeners.tracing.trace_listener import TraceCollectionListener + + +def test_batch_manager_starts_clean() -> None: + """No earlier test may leave batch state on the shared singleton. + + Regression: `test_nested_agent_executor_flow_does_not_finalize_parent_batch` + set `trace_batch_id = "debug-trace-batch"` and never restored it, which broke + `test_trace_calls_when_enabled_via_env` several hundred tests later. + """ + manager = TraceCollectionListener().batch_manager + + assert manager.trace_batch_id is None + assert manager.current_batch is None + assert manager.batch_owner_type is None + assert manager.batch_owner_id is None + assert manager.event_buffer == [] + assert manager.defer_session_finalization is False + + +def test_listener_can_register_handlers_again() -> None: + """A leaked listener must not block later handler registration. + + `cleanup_event_handlers` wipes the bus between tests, but `setup_listeners` + returns early when `_listeners_setup` is already set (`trace_listener.py:208`). + A listener surviving a tracing-enabled test would therefore register nothing + for the rest of the worker — tracing silently collects no events instead of + failing visibly. + """ + assert TraceCollectionListener()._listeners_setup is False