andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-09-20 18:13:49 +00:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

fix(openai): surface gateway errors reported inside an HTTP 200 (#7342)

Browse Source 
* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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/<id>/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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
João Moura
2026-09-09 19:12:57 -03:00
committed by GitHub
parent 4ed4aba6cd
commit d729cade6b
12 changed files with 969 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
conftest.py
View File

@@ -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/<id>/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."""

32
docs/edge/ar/concepts/llms.mdx
View File

@@ -1339,6 +1339,38 @@ llm = LLM(
llm = LLM(model="gpt-4")
```
</Tab>
<Tab title="أخطاء البوابة">
<Tip>
تُعيد البوابات مثل OpenRouter الرمز `200 OK` بمجرد قبول المزود الأساسي للطلب، لذلك يصل انتهاء مهلة المزود داخل جسم الاستجابة بدلًا من رمز الحالة.
</Tip>
تُطلق 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}")
```
<Warning>
إن استخدام `response_model` كبير أو متداخل بعمق يزيد احتمال انتهاء مهلة المزود. تعامل مع هذه الحالات كأعطال مؤقتة في المزود، وليس كإنتاج النموذج مخرجات منظمة تالفة.
</Warning>
</Tab>
<Tab title="طول السياق">
<Tip>
استخدم نماذج سياق أكبر للمهام الواسعة

32
docs/edge/en/concepts/llms.mdx
View File

@@ -1487,6 +1487,38 @@ llm = LLM(
llm = LLM(model="gpt-4")
```
</Tab>
<Tab title="Gateway Errors">
<Tip>
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.
</Tip>
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}")
```
<Warning>
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.
</Warning>
</Tab>
<Tab title="Context Length">
<Tip>
Use larger context models for extensive tasks

32
docs/edge/ko/concepts/llms.mdx
View File

@@ -995,6 +995,38 @@ LLM 설정을 최대한 활용하는 방법을 알아보세요:
llm = LLM(model="gpt-4")
```
</Tab>
<Tab title="게이트웨이 오류">
<Tip>
OpenRouter와 같은 게이트웨이는 업스트림 공급자가 요청을 수락하는 즉시 `200 OK`를 반환하므로, 공급자 타임아웃은 상태 코드가 아니라 응답 본문에 담겨 도착합니다.
</Tip>
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}")
```
<Warning>
크거나 깊게 중첩된 `response_model`은 업스트림 타임아웃 가능성을 높입니다. 이를 모델이 잘못된 구조화 출력을 생성한 것으로 보지 말고, 일시적인 공급자 장애로 처리하세요.
</Warning>
</Tab>
<Tab title="컨텍스트 길이">
<Tip>
대규모 작업에는 더 큰 컨텍스트 모델을 사용하세요.

32
docs/edge/pt-BR/concepts/llms.mdx
View File

@@ -919,6 +919,38 @@ Saiba como obter o máximo da configuração do seu LLM:
llm = LLM(model="gpt-4")
```
</Tab>
<Tab title="Erros de Gateway">
<Tip>
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.
</Tip>
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}")
```
<Warning>
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.
</Warning>
</Tab>
<Tab title="Comprimento do Contexto">
<Tip>
Use modelos de contexto expandido para tarefas extensas

145
lib/crewai/src/crewai/llms/providers/openai/completion.py
View File

@@ -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)

540
lib/crewai/tests/llms/openai/test_gateway_error_envelope.py Normal file
View File

@@ -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"<html>gateway boom</html>", 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

36
lib/crewai/tests/llms/openai/test_openai.py
View File

@@ -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?")

5
lib/crewai/tests/llms/openai/test_tools_reasoning_effort_retry.py
View File

@@ -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",

23
lib/crewai/tests/llms/snowflake/test_snowflake.py
View File

@@ -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):

44
lib/crewai/tests/test_tool_cache_default.py
View File

@@ -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:

53
lib/crewai/tests/tracing/test_trace_listener_isolation.py Normal file
View File

@@ -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/<leaked-id>/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