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