diff --git a/.github/renovate.json b/.github/renovate.json index a5f14e5a..c8e423c1 100644 --- a/.github/renovate.json +++ b/.github/renovate.json @@ -55,7 +55,7 @@ { "description": "LLM SDKs — always review manually", "groupName": "LLM SDKs", - "matchPackageNames": ["openai", "anthropic", "mcp"], + "matchPackageNames": ["openai", "httpx2", "anthropic", "mcp"], "schedule": ["before 9am on Monday"], "automerge": false }, diff --git a/CHANGELOG.md b/CHANGELOG.md index cd947895..4eb7756b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -152,6 +152,16 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. ### Changed +- **OpenAI SDK v3 and its HTTPX2 default transport are now supported (#1009).** + Chat Completions and Responses streams normalize native HTTPX2 connection + deaths through the same retry boundary as legacy HTTPX-backed providers, + including failures observed after safe cross-thread client closure during a + model-registry reload. The OpenAI v3 runtime escape hatch for explicitly + injected legacy HTTPX clients remains supported. OpenAI connections now + follow HTTPX2's operating-system trust store by default; deployments that + relied on a modified `certifi` bundle must install that CA in the system + store or set `SSL_CERT_FILE` / `SSL_CERT_DIR`. + - **Log event rename: `drain_stream.post_finish_blip` is now `stream.post_finish_blip`; its `usage_captured` field is retained.** The single-shot drain normalizes mid-body transport deaths through the same diff --git a/docs/eval.md b/docs/eval.md index 94a7f28f..1e183300 100644 --- a/docs/eval.md +++ b/docs/eval.md @@ -177,7 +177,8 @@ Runs a complete multi-turn conversation: 1. Appends the user message. 2. Checks `_cancelled` event — stops if set (timeout cleanup). -3. Calls the model API (non-streaming). +3. Calls the model through the production streaming provider path and drains + the result. 4. If tool calls are returned, executes them (with stdout suppressed) and logs each call to `self.tool_call_log`. 5. Repeats up to `max_turns` or until the model responds without tool calls. @@ -190,14 +191,15 @@ Parallel tool calls are capped at 10 per turn to prevent degenerate repetition. Each test runs in a `ThreadPoolExecutor(max_workers=1)` with a per-test timeout (`--test-timeout`). Each attempt gets its own `OpenAI` client with -a matching httpx read timeout. On timeout, three layers of defense prevent -zombie connections: +a matching per-read HTTP transport timeout. Because a trickling stream can +continually reset that read timeout, three layers bound the harness and stop +follow-on work: -1. **httpx timeout**: Per-request read timeout aborts the HTTP call and - releases the server slot. +1. **Executor wall clock**: The harness stops waiting after `--test-timeout`. 2. **`_cancelled` event**: Prevents the orphan thread from starting new turns. -3. **`run_client.close()`**: Closes the connection pool to abort any - in-flight request. +3. **`run_client.close()`**: Retires the connection pool and prevents reuse. + HTTPX2 does not promise that cross-thread client closure immediately aborts + an active body read; that read unwinds on its next wire event or read timeout. ### Retry Logic @@ -214,7 +216,8 @@ Each test case runs in isolation: 1. A fresh temp directory is created. 2. Setup files are written to the temp directory. 3. The working directory is changed to the temp directory. -4. A per-attempt `OpenAI` client is created with httpx timeout matching `--test-timeout`. +4. A per-attempt `OpenAI` client is created with an HTTP transport timeout + matching `--test-timeout`. 5. A new `HeadlessSession` is created with the current developer prompt. 6. `send_headless()` runs the user prompt through the conversation loop. 7. The tool log is scored against expected actions. diff --git a/pyproject.toml b/pyproject.toml index 54d0f420..df11df66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,11 +23,14 @@ classifiers = [ "Topic :: Scientific/Engineering :: Artificial Intelligence", ] dependencies = [ - # 2.45 adds the GPT-5.6 typed fields used below. Version 3 replaces the - # httpx transport family, so migrate the streaming retry boundary first. - "openai>=2.45,<3", + # Version 3 moves the default transport to HTTPX2. Keep major upgrades + # deliberate because the stream retry boundary depends on that contract. + "openai>=3,<4", "anthropic>=0.117", # tracks the release current at claude-opus-5 onboarding; hard runtime floor is still 0.105 (mid-conversation system blocks) — Opus 5 itself needs no new SDK surface (model ids are opaque strings; "refusal" has been in the StopReason literal since ~0.95). Raise this when adopting fast mode / server-side fallbacks / advisor / mid-conversation tool changes, which DO need newer typed params. "httpx>=0.28", + # Direct because the provider boundary catches this exception family; + # OpenAI v3's transitive dependency alone is not an import contract. + "httpx2>=2.7,<3", "mcp>=1.27,<2", # v2 is a breaking rewrite (2.0.0a1 live 2026-06-11; stable ~2026-07-27) — streamablehttp_client removed, 2-tuple transport, snake_case types; migrate deliberately "starlette>=1.3.1", # CVE-2026-54282 (path->authority host spoof) + CVE-2026-54283 (url-encoded form DoS); supersedes the PYSEC-2026-161 host-header path-injection floor "uvicorn>=0.34", diff --git a/tests/test_drain_stream.py b/tests/test_drain_stream.py index 95542bcd..8a4246bd 100644 --- a/tests/test_drain_stream.py +++ b/tests/test_drain_stream.py @@ -9,6 +9,8 @@ provider_blocks, trailing-citation fold). from __future__ import annotations +import httpx +import httpx2 import pytest from turnstone.core.providers import ( @@ -418,11 +420,18 @@ class TestTransportGuarded: """``transport_guarded`` — drain's conversion rule for consumers that keep streaming semantics (the interactive loop).""" - @pytest.mark.parametrize("exc_name", ["ReadError", "RemoteProtocolError"]) - def test_pre_finish_transport_error_becomes_retryable_incomplete(self, exc_name): - import httpx - - exc_cls = getattr(httpx, exc_name) + @pytest.mark.parametrize( + "exc_cls", + [ + httpx.ReadError, + httpx.RemoteProtocolError, + httpx2.ReadError, + httpx2.RemoteProtocolError, + ], + ids=["httpx-read", "httpx-protocol", "httpx2-read", "httpx2-protocol"], + ) + def test_pre_finish_transport_error_becomes_retryable_incomplete(self, exc_cls): + exc_name = exc_cls.__name__ def chunks(): yield StreamChunk(content_delta="partial") @@ -434,14 +443,17 @@ class TestTransportGuarded: next(it) assert isinstance(excinfo.value.__cause__, exc_cls) - def test_message_byte_matches_drains_shape(self): + @pytest.mark.parametrize( + "exc_cls", + [httpx.ReadError, httpx2.ReadError], + ids=["httpx", "httpx2"], + ) + def test_message_byte_matches_drains_shape(self, exc_cls): # The wrapper and the drain are the SAME conversion rule; any test # or log filter pinned to drain's message must match the wrapper's. - import httpx - def chunks(): yield StreamChunk(content_delta="partial") - raise httpx.ReadError("[SSL] record layer failure (_ssl.c:2590)") + raise exc_cls("[SSL] record layer failure (_ssl.c:2590)") with pytest.raises(IncompleteStreamError) as guarded: list(transport_guarded(chunks())) @@ -449,17 +461,20 @@ class TestTransportGuarded: drain_stream(chunks()) assert str(guarded.value) == str(drained.value) - def test_post_finish_blip_ends_stream_cleanly(self, caplog): + @pytest.mark.parametrize( + "exc_cls", + [httpx.ReadError, httpx2.ReadError], + ids=["httpx", "httpx2"], + ) + def test_post_finish_blip_ends_stream_cleanly(self, caplog, exc_cls): # The generation already completed — the blip only cost trailing # metadata, so the stream ends instead of raising. import logging - import httpx - def chunks(): yield StreamChunk(content_delta="done") yield StreamChunk(finish_reason="stop") - raise httpx.ReadError("late blip") + raise exc_cls("late blip") with caplog.at_level(logging.WARNING, logger="turnstone.core.providers._protocol"): out = list(transport_guarded(chunks())) @@ -502,32 +517,33 @@ class TestErrorPropagation: # The generation completed (finish reason in hand) — a trailing # transport blip forfeits only trailing metadata (here: the usage # chunk), never the completed result. - import httpx - def chunks(): yield StreamChunk(content_delta="whole answer") yield StreamChunk(finish_reason="stop") - raise httpx.ReadError("late blip") + raise httpx2.ReadError("late blip") result = drain_stream(chunks()) assert result.content == "whole answer" assert result.finish_reason == "stop" assert result.usage is None - def test_httpx_transport_error_becomes_retryable_incomplete(self): + @pytest.mark.parametrize( + "exc_cls", + [httpx.RemoteProtocolError, httpx2.RemoteProtocolError], + ids=["httpx", "httpx2"], + ) + def test_transport_error_becomes_retryable_incomplete(self, exc_cls): # Streaming moves the body read out of the SDK's wrapped request: - # a mid-body wire failure surfaces as a raw httpx.TransportError - # no retry predicate recognizes. The drain re-raises it (chained, + # a mid-body wire failure surfaces as a raw HTTPX-family TransportError + # no retry predicate recognizes. The drain re-raises it (chained, # message preserved) as the retryable IncompleteStreamError. - import httpx - def chunks(): yield StreamChunk(content_delta="partial") - raise httpx.RemoteProtocolError("peer closed connection") + raise exc_cls("peer closed connection") with pytest.raises(IncompleteStreamError, match="RemoteProtocolError") as excinfo: drain_stream(chunks()) - assert isinstance(excinfo.value.__cause__, httpx.RemoteProtocolError) + assert isinstance(excinfo.value.__cause__, exc_cls) def test_mid_stream_exception_propagates_verbatim(self): # Retry/deadline/fallback policy is the caller's — the drain adds diff --git a/tests/test_eval_nudges.py b/tests/test_eval_nudges.py index 327cf4b3..a991ec9a 100644 --- a/tests/test_eval_nudges.py +++ b/tests/test_eval_nudges.py @@ -2753,7 +2753,7 @@ class TestRunResourceLifecycle: assert not is_storage_initialized() def test_a_hung_generation_is_bounded_by_the_wall_clock(self, monkeypatch): - """The per-request httpx timeout cannot bound a STREAM — a + """The per-request HTTP transport timeout cannot bound a STREAM — a trickling response resets the read timeout indefinitely — so without the executor wall clock a hung generation occupies a run slot forever and is scored as a body regression when the sweep diff --git a/tests/test_sdk_stream_boundary.py b/tests/test_sdk_stream_boundary.py index 2176dcb6..9dcf6eea 100644 --- a/tests/test_sdk_stream_boundary.py +++ b/tests/test_sdk_stream_boundary.py @@ -1,17 +1,19 @@ """Offline pins of the SDK boundary behaviors the #937 retry design rests on. -Three facts, each probed against the REAL SDKs over mock/loopback +Four facts, each probed against the REAL SDKs over mock/loopback transports (no network, no live backend): -1. The OpenAI SDK's ``max_retries`` covers request time only — a - mid-BODY death produces no re-request, and the raw ``httpx.ReadError`` - escapes the chunk iterator unwrapped. -2. The Anthropic ``messages.stream()`` helper propagates the same shape. -3. Closing an httpx-backed SDK client from another thread while a read is - blocked (the ``ModelRegistry.reload()`` shape) surfaces as an - ``httpx.TransportError`` on the blocked ``next()`` — which is why - ``_stream_response``'s mid-stream re-issue ladder re-resolves the - registry binding before re-creating. +1. OpenAI v3's ``max_retries`` covers request time only — a mid-BODY death + produces no re-request, and the raw ``httpx2.ReadError`` escapes both Chat + Completions and Responses chunk iterators unwrapped. +2. OpenAI v3's runtime-only legacy-client path preserves the old ``httpx`` + exception family when an application explicitly injects that client. +3. The Anthropic ``messages.stream()`` helper propagates the ``httpx`` shape. +4. Closing an OpenAI v3 default client from another thread while a read is + blocked (the ``ModelRegistry.reload()`` shape) completes safely; a later + wire release surfaces as an ``httpx2.TransportError`` on the blocked + ``next()``. The production ``transport_guarded`` seam must normalize it + before the retry gate. If an SDK/httpx upgrade changes any of these, the ``transport_guarded`` conversion (and the retry gate consuming it) must be re-verified — these @@ -27,6 +29,7 @@ import time import anthropic import httpx +import httpx2 import openai import pytest @@ -39,6 +42,12 @@ CHAT_CHUNK = ( '"finish_reason":null}]}' + LF + LF ) +RESPONSES_EVENT = ( + 'data: {"type":"response.output_text.delta","sequence_number":0,' + '"item_id":"item_1","output_index":0,"content_index":0,' + '"delta":"hello","logprobs":[]}' + LF + LF +) + ANTHROPIC_EVENTS = ( "event: message_start" + LF @@ -71,6 +80,17 @@ class _DyingStream(httpx.SyncByteStream): raise httpx.ReadError("[SSL] record layer failure (_ssl.c:2590)") +class _Httpx2DyingStream(httpx2.SyncByteStream): + """HTTPX2 response body: one SSE payload, then a wire death.""" + + def __init__(self, payload: bytes) -> None: + self._payload = payload + + def __iter__(self): + yield self._payload + raise httpx2.ReadError("[SSL] record layer failure (_ssl.c:2590)") + + def _dying_transport(payload: str, requests: list) -> httpx.MockTransport: def handler(request: httpx.Request) -> httpx.Response: requests.append(request) @@ -84,12 +104,70 @@ def _dying_transport(payload: str, requests: list) -> httpx.MockTransport: return httpx.MockTransport(handler) -def test_openai_midbody_death_is_unwrapped_readerror_and_no_rerequest(): +def _httpx2_dying_transport(payload: str, requests: list) -> httpx2.MockTransport: + def handler(request: httpx2.Request) -> httpx2.Response: + requests.append(request) + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=_Httpx2DyingStream(payload.encode()), + request=request, + ) + + return httpx2.MockTransport(handler) + + +def test_openai_v3_chat_midbody_death_is_unwrapped_httpx2_error_and_no_rerequest(): requests: list = [] client = openai.OpenAI( api_key="probe", base_url="http://probe.invalid/v1", - http_client=httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)), + http_client=httpx2.Client(transport=_httpx2_dying_transport(CHAT_CHUNK, requests)), + max_retries=2, + ) + stream = client.chat.completions.create( + model="m", messages=[{"role": "user", "content": "hi"}], stream=True + ) + texts = [] + with pytest.raises(httpx2.ReadError) as excinfo: + for chunk in stream: + if chunk.choices and chunk.choices[0].delta.content: + texts.append(chunk.choices[0].delta.content) + # The retry gate matches on the class NAME; pin the exact identity the + # SDK lets escape, and that it is the HTTPX2 transport family. + assert type(excinfo.value).__name__ == "ReadError" + assert isinstance(excinfo.value, httpx2.TransportError) + assert texts == ["hello"] # the request succeeded; the BODY died + assert len(requests) == 1 # max_retries never re-requested mid-body + + +def test_openai_v3_responses_midbody_death_is_unwrapped_httpx2_error_and_no_rerequest(): + requests: list = [] + client = openai.OpenAI( + api_key="probe", + base_url="http://probe.invalid/v1", + http_client=httpx2.Client(transport=_httpx2_dying_transport(RESPONSES_EVENT, requests)), + max_retries=2, + ) + stream = client.responses.create(model="m", input="hi", stream=True) + texts = [] + with pytest.raises(httpx2.ReadError) as excinfo: + for event in stream: + if event.type == "response.output_text.delta": + texts.append(event.delta) + assert type(excinfo.value).__name__ == "ReadError" + assert isinstance(excinfo.value, httpx2.TransportError) + assert texts == ["hello"] + assert len(requests) == 1 + + +def test_openai_v3_legacy_httpx_midbody_death_keeps_legacy_error_family(): + requests: list = [] + legacy_http_client = httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)) + client = openai.OpenAI( + api_key="probe", + base_url="http://probe.invalid/v1", + http_client=legacy_http_client, # type: ignore[arg-type] max_retries=2, ) stream = client.chat.completions.create( @@ -100,12 +178,10 @@ def test_openai_midbody_death_is_unwrapped_readerror_and_no_rerequest(): for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: texts.append(chunk.choices[0].delta.content) - # The retry gate matches on the class NAME; pin the exact identity the - # SDK lets escape, and that it is the httpx transport family. assert type(excinfo.value).__name__ == "ReadError" assert isinstance(excinfo.value, httpx.TransportError) - assert texts == ["hello"] # the request succeeded; the BODY died - assert len(requests) == 1 # max_retries never re-requested mid-body + assert texts == ["hello"] + assert len(requests) == 1 def test_anthropic_midbody_death_is_unwrapped_readerror_and_no_rerequest(): @@ -136,13 +212,59 @@ def test_anthropic_midbody_death_is_unwrapped_readerror_and_no_rerequest(): assert len(requests) == 1 -def test_cross_thread_client_close_surfaces_transport_error_to_blocked_read(): - """The ``ModelRegistry.reload()`` shape: an admin-thread ``client.close()`` - under a worker blocked in ``next()`` must surface as an - ``httpx.TransportError`` (so ``transport_guarded`` converts it and the - retry gate passes) — not as a plain ``RuntimeError`` the gate would - treat as fatal.""" - body = f"{len(CHAT_CHUNK):x}" + CRLF + CHAT_CHUNK + CRLF +@pytest.mark.parametrize("surface", ["chat", "responses"]) +def test_closed_v3_default_client_creation_stays_sdk_wrapped_and_unarmed(surface: str): + """A re-create on the client closed by reload is still a creation error. + + OpenAI v3 must wrap the default HTTPX2 client's ``RuntimeError`` as its + retryable ``APIConnectionError`` before either adapter can arm the stream. + This preserves the creation-vs-mid-stream classifier while the original + normalized stream death remains available to the outer re-issue ladder. + """ + from turnstone.core.providers import create_provider + + client = openai.OpenAI( + api_key="probe", + base_url="http://probe.invalid/v1", + max_retries=0, + ) + client.close() + cancel_ref: list = [] + provider = create_provider("openai-compatible", api_surface=surface) + + with pytest.raises(openai.APIConnectionError) as excinfo: + provider.create_streaming( + client=client, + model="m", + messages=[{"role": "user", "content": "hi"}], + cancel_ref=cancel_ref, + ) + + assert cancel_ref == [] + assert type(excinfo.value).__name__ in provider.retryable_error_names + assert type(excinfo.value.__cause__) is RuntimeError + + +@pytest.mark.parametrize( + ("surface", "payload"), + [("chat", CHAT_CHUNK), ("responses", RESPONSES_EVENT)], +) +def test_cross_thread_v3_default_client_close_then_wire_release_is_normalized( + surface: str, payload: str +): + """The ``ModelRegistry.reload()`` shape stays safe at the retry seam. + + OpenAI v3's synchronous HTTPX2 client does not promise that cross-thread + ``close()`` itself interrupts a blocked body read. Pin the behavior + Turnstone needs instead: closing from the admin thread completes safely + while a worker is in ``next()``, and the subsequent wire release reaches + ``transport_guarded`` as a provider-retryable ``IncompleteStreamError`` + for both OpenAI streaming adapters. + """ + from turnstone.core.providers import create_provider, transport_guarded + from turnstone.core.providers._protocol import IncompleteStreamError + + body = f"{len(payload):x}" + CRLF + payload + CRLF response_head = ( "HTTP/1.1 200 OK" + CRLF @@ -153,59 +275,100 @@ def test_cross_thread_client_close_surfaces_transport_error_to_blocked_read(): + CRLF ) listener = socket.create_server(("127.0.0.1", 0)) + listener.settimeout(10.0) port = listener.getsockname()[1] - client_closed = threading.Event() + release_peer = threading.Event() reader_blocked = threading.Event() + close_done = threading.Event() + first_content: list[str] = [] + reader_errors: list[BaseException] = [] + closer_errors: list[BaseException] = [] + server_errors: list[BaseException] = [] def serve() -> None: - conn, _ = listener.accept() - conn.recv(65536) - conn.sendall((response_head + body).encode()) - # Hold the connection (no second chunk) so the reader blocks, and - # release only once the client has been closed under it. - client_closed.wait(timeout=10.0) - conn.close() + try: + conn, _ = listener.accept() + with conn: + conn.settimeout(10.0) + conn.recv(65536) + conn.sendall((response_head + body).encode()) + # Keep the peer open through close_done: any reader error + # before release_peer is therefore caused by close(), not EOF. + if not release_peer.wait(timeout=15.0): + raise AssertionError("peer release was never signalled") + except BaseException as exc: + server_errors.append(exc) - client = openai.OpenAI(api_key="probe", base_url=f"http://127.0.0.1:{port}/v1", max_retries=0) + client = openai.OpenAI( + api_key="probe", + base_url=f"http://127.0.0.1:{port}/v1", + max_retries=0, + timeout=5.0, + ) + + def read_stream() -> None: + try: + provider = create_provider("openai-compatible", api_surface=surface) + chunks = provider.create_streaming( + client=client, + model="m", + messages=[{"role": "user", "content": "hi"}], + ) + it = transport_guarded(chunks) + first_content.append(next(it).content_delta) + reader_blocked.set() + next(it) + except BaseException as exc: + reader_errors.append(exc) def closer() -> None: - reader_blocked.wait(timeout=10.0) - time.sleep(0.5) # let the reader enter the blocking socket read - client.close() - client_closed.set() + try: + if not reader_blocked.wait(timeout=10.0): + raise AssertionError("reader never reached the blocked body read") + time.sleep(0.5) # let the reader enter the blocking socket read + client.close() + except BaseException as exc: + closer_errors.append(exc) + finally: + close_done.set() server_thread = threading.Thread(target=serve) + reader_thread = threading.Thread(target=read_stream) closer_thread = threading.Thread(target=closer) server_thread.start() + reader_thread.start() closer_thread.start() try: - stream = client.chat.completions.create( - model="m", messages=[{"role": "user", "content": "hi"}], stream=True - ) - it = iter(stream) - first = next(it) # the one sent chunk arrives; the wire then idles - assert first.choices[0].delta.content == "hello" - reader_blocked.set() - with pytest.raises(httpx.TransportError) as excinfo: - next(it) # blocked read, killed by the cross-thread close() - # Platform/timing-dependent: the killed read surfaces as ReadError - # (EBADF from the blocked recv) or, where the reader observes EOF - # first, RemoteProtocolError (chunked body never terminated). Both - # are TransportError members of _BACKEND_STREAM_EXC_NAMES, so - # transport_guarded converts either and the retry gate passes — the - # property this pin exists for. - assert type(excinfo.value).__name__ in {"ReadError", "RemoteProtocolError"} + assert reader_blocked.wait(timeout=10.0) + assert close_done.wait(timeout=10.0) + assert closer_errors == [] + # The server has not closed its peer yet, proving close() completed + # safely rather than merely returning after an EOF unblocked it. + assert server_errors == [] + release_peer.set() + reader_thread.join(timeout=10.0) finally: - # Unblock and join both threads on every exit path so nothing + # Unblock and join every thread on every exit path so nothing # outlives the test (leaked-thread guard). reader_blocked.set() - client_closed.set() + release_peer.set() closer_thread.join(timeout=10.0) + reader_thread.join(timeout=10.0) server_thread.join(timeout=10.0) listener.close() with contextlib.suppress(Exception): client.close() + assert first_content == ["hello"] + assert len(reader_errors) == 1 + exc = reader_errors[0] + assert isinstance(exc, IncompleteStreamError) + assert any(name in str(exc) for name in ("ReadError", "RemoteProtocolError")) + cause = exc.__cause__ + assert isinstance(cause, httpx2.TransportError) + assert type(cause).__name__ in {"ReadError", "RemoteProtocolError"} + assert server_errors == [] assert not closer_thread.is_alive() + assert not reader_thread.is_alive() assert not server_thread.is_alive() @@ -240,7 +403,7 @@ class TestEagerAppendContract: requests: list = [] client = openai.OpenAI( api_key="probe", - http_client=httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)), + http_client=httpx2.Client(transport=_httpx2_dying_transport(CHAT_CHUNK, requests)), ) self._armed_at_return(OpenAIChatCompletionsProvider(), client) assert len(requests) == 1 # the HTTP call happened inside create @@ -251,7 +414,7 @@ class TestEagerAppendContract: requests: list = [] client = openai.OpenAI( api_key="probe", - http_client=httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)), + http_client=httpx2.Client(transport=_httpx2_dying_transport(RESPONSES_EVENT, requests)), ) self._armed_at_return(OpenAIResponsesProvider(), client) assert len(requests) == 1 diff --git a/tests/test_session_backend_error_format.py b/tests/test_session_backend_error_format.py index 3b9b282b..841df685 100644 --- a/tests/test_session_backend_error_format.py +++ b/tests/test_session_backend_error_format.py @@ -1,6 +1,6 @@ """Tests for :meth:`ChatSession._format_backend_error`. -The helper turns bare backend-boundary exceptions (httpx ``ReadTimeout``, +The helper turns bare backend-boundary exceptions (HTTPX/HTTPX2 ``ReadTimeout``, OpenAI SDK ``APITimeoutError`` / ``APIConnectionError`` / ``NotFoundError`` / ``RateLimitError`` / ``AuthenticationError``) into operator-actionable messages that include the provider, base URL, and @@ -225,7 +225,7 @@ def test_rate_limit_with_overflow_phrasing_is_not_mislabeled_overflow(): def _stream_death_exemplars() -> list[BaseException]: """One realistic instance per name in ``_BACKEND_STREAM_EXC_NAMES``: - the normalized shape the guarded iterators raise, plus the raw httpx + the normalized shape the guarded iterators raise, plus the raw HTTPX-family names for any future unguarded path.""" from turnstone.core.providers import IncompleteStreamError diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py index f38ae250..a6e7090b 100644 --- a/turnstone/core/providers/_openai_chat.py +++ b/turnstone/core/providers/_openai_chat.py @@ -452,11 +452,11 @@ class OpenAIChatCompletionsProvider: # operator-declared ``finish_reason_optional`` capability: on a # server that never sends finish reasons, a stream that ended # CLEANLY — the SDK ends iteration on [DONE]; an abrupt connection - # death raises httpx.TransportError out of this generator — after - # delivering output is a completed generation. Everywhere else a - # clean finish-less end is indistinguishable from a generation - # that died behind a clean-closing proxy/ASGI layer, so the shim - # stays DISARMED and the drain's complete-or-error gate raises + # death raises the active HTTP client's TransportError out of this + # generator — after delivering output is a completed generation. + # Everywhere else a clean finish-less end is indistinguishable + # from a generation that died behind a clean-closing proxy/ASGI layer, + # so the shim stays DISARMED and the drain's complete-or-error gate raises # (retryable) instead of blessing possibly-truncated text. # Reasoning counts as delivered output — a thinking model that # spent its budget before emitting content is still a completed diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index f7b6ba26..884e4c6e 100644 --- a/turnstone/core/providers/_protocol.py +++ b/turnstone/core/providers/_protocol.py @@ -228,10 +228,12 @@ def transport_guarded(chunks: Iterator[StreamChunk]) -> Iterator[StreamChunk]: Streaming moves the body read out of the SDK's ``APIConnectionError``-wrapped request into raw iteration, so a mid-body wire death (connection drop, TLS record failure, read - timeout) surfaces as a bare ``httpx.TransportError`` no retry - predicate recognizes. This wrapper is that conversion rule made - reusable for consumers that keep streaming semantics (the - interactive loop); :func:`drain_stream` applies the same rule for + timeout) surfaces as a bare transport error no retry predicate + recognizes. OpenAI v3's default client raises ``httpx2`` errors; + Anthropic, Turnstone's own HTTP clients, and the OpenAI v3 legacy-client + escape hatch raise ``httpx`` errors. This wrapper is the one conversion + rule for both families, reusable by consumers that keep streaming + semantics (the interactive loop); :func:`drain_stream` applies it for the single-shot lanes. - A ``TransportError`` BEFORE any finish reason re-raises (chained) @@ -243,7 +245,11 @@ def transport_guarded(chunks: Iterator[StreamChunk]) -> Iterator[StreamChunk]: - Everything else — chunks, exhaustion, non-transport exceptions — passes through untouched. """ - import httpx # noqa: PLC0415 — heavyweight; deferred off the type-module import path + # Both libraries are heavyweight; keep them off this module's dataclass- + # only import path. ``httpx`` remains Turnstone's application transport, + # while ``httpx2`` is the OpenAI v3 default transport. + import httpx # noqa: PLC0415 + import httpx2 # noqa: PLC0415 finish_seen = False usage_seen = False @@ -253,7 +259,7 @@ def transport_guarded(chunks: Iterator[StreamChunk]) -> Iterator[StreamChunk]: sc = next(iterator) except StopIteration: return - except httpx.TransportError as exc: + except (httpx.TransportError, httpx2.TransportError) as exc: if finish_seen: # usage_captured distinguishes "completed result kept but # its spend went missing from usage accounting" (the chat @@ -348,7 +354,7 @@ def drain_stream( Raises whatever the underlying stream raises — retry/deadline/fallback policy stays with the caller, exactly as with the old non-streaming - transport — EXCEPT httpx transport failures, normalized by + transport — EXCEPT HTTPX/HTTPX2 transport failures, normalized by :func:`transport_guarded` (the one conversion rule, shared with the interactive loop): a mid-body death before the finish reason is re-raised (chained) as :class:`IncompleteStreamError`, restoring the diff --git a/turnstone/core/session.py b/turnstone/core/session.py index c700a87a..9fc751bb 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -1935,11 +1935,11 @@ _BACKEND_AUTH_EXC_NAMES: frozenset[str] = frozenset( _BACKEND_RATE_LIMIT_EXC_NAMES: frozenset[str] = frozenset({"RateLimitError"}) # Mid-response stream deaths: the normalized shape every guarded iterator # raises (``IncompleteStreamError`` from ``drain_stream`` / -# ``transport_guarded``) plus the raw httpx names for any future unguarded -# path (defense in depth). Unioning them into ``_BACKEND_KNOWN_EXC_NAMES`` -# is required — ``_format_backend_error`` gates on that set before the -# branch lookups — and makes these three names ineligible for -# ``_is_ctx_overflow``'s text-based overflow detection (its class +# ``transport_guarded``) plus the raw HTTPX/HTTPX2 names for any future +# unguarded path (defense in depth). Unioning them into +# ``_BACKEND_KNOWN_EXC_NAMES`` is required — ``_format_backend_error`` gates +# on that set before the branch lookups — and makes these three names +# ineligible for ``_is_ctx_overflow``'s text-based overflow detection (its class # self-gate): harmless, since their texts are fixed transport/SSL strings # that never carry overflow phrases. _BACKEND_STREAM_EXC_NAMES: frozenset[str] = frozenset( @@ -7456,7 +7456,7 @@ class ChatSession: can arrive as several exception classes); everything else is matched by class name (see the ``_BACKEND_*_EXC_NAMES`` sets above) so the same helper covers - httpx ``ReadTimeout`` / ``ConnectError``, OpenAI SDK + HTTPX/HTTPX2 ``ReadTimeout`` / ``ConnectError``, OpenAI SDK ``APITimeoutError`` / ``APIConnectionError`` / ``NotFoundError`` / ``RateLimitError`` / ``AuthenticationError``, and the Anthropic SDK equivalents (which share names). diff --git a/turnstone/eval/core.py b/turnstone/eval/core.py index 798f88d3..d1c916ff 100644 --- a/turnstone/eval/core.py +++ b/turnstone/eval/core.py @@ -232,7 +232,7 @@ class HeadlessSession(ChatSession): - auto_approve is always True - Tool calls are recorded into a structured log - All stdout output is suppressed - - send_headless() uses non-streaming API + - send_headless() drains the production streaming provider path """ def __init__( @@ -307,8 +307,8 @@ class HeadlessSession(ChatSession): ) -> list[dict[str, Any]]: """Run a complete conversation turn headlessly. - Uses non-streaming API calls. Captures all tool calls into - self.tool_call_log. + Drains production streaming provider calls into single-shot results. + Captures all tool calls into ``self.tool_call_log``. Returns the tool call log: list of dicts with keys: tool: str, args: dict, result: str (truncated), ok: bool, @@ -533,9 +533,9 @@ def run_with_lifecycle( past the teardown (which closes only the last one built). * *drive* (returned by ``build_session``) — submitted to a one-worker executor and bounded by ``future.result(test_timeout)``. - The per-request httpx timeout cannot bound a STREAM: a trickling - response resets the read timeout indefinitely, so without the - wall clock a hung generation occupies a run slot forever. On + The per-request HTTP transport timeout cannot bound a STREAM: a + trickling response resets the read timeout indefinitely, so without + the wall clock a hung generation occupies a run slot forever. On timeout the session is DROPPED, not closed: the shutdown did not wait, so the worker is still inside the drive, and ``close()``'s bounded shell-join would trade a bounded leak for a blocked @@ -736,8 +736,8 @@ def _run_single_test( ) def _build_client() -> Any: - # Per-attempt client with request-level timeout so httpx aborts - # the HTTP request itself — no zombie connections on the server. + # Per-attempt client with a per-read timeout. The executor wall clock + # above separately bounds a trickling stream that keeps resetting it. return OpenAI( base_url=client.base_url, api_key=client.api_key, diff --git a/uv.lock b/uv.lock index 6e1ed29b..45ae1efd 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,8 @@ resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'win32'", "python_full_version < '3.13' and sys_platform == 'win32'", "python_full_version == '3.13.*' and sys_platform != 'win32'", - "python_full_version < '3.13' and sys_platform != 'win32'", + "python_full_version == '3.12.*' and sys_platform == 'emscripten'", + "(python_full_version < '3.12' and sys_platform == 'emscripten') or (python_full_version < '3.13' and sys_platform != 'emscripten' and sys_platform != 'win32')", ] [[package]] @@ -972,6 +973,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, ] +[[package]] +name = "httpcore2" +version = "2.10.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "h11", marker = "python_full_version != '3.12.*' or sys_platform != 'emscripten'" }, + { name = "truststore", marker = "python_full_version != '3.12.*' or sys_platform != 'emscripten'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a9/83/a896fc59940fc5a6e2aff3a4be1d92fa890112936803b331cae75a993c34/httpcore2-2.10.0.tar.gz", hash = "sha256:13c0cc3d1919d4f28457f60cd2c2abe04113a8af184ccf1142811beba936f9dc", size = 67427, upload-time = "2026-08-09T09:11:32.123Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e5/4f/d149104195a35e2853a2fc203a8e3477747e58c80e17dda686dace174383/httpcore2-2.10.0-py3-none-any.whl", hash = "sha256:7df06cfb34070cae4f7c89be69dc1095eca138e9704ceffb98d25c1912ab6f01", size = 83000, upload-time = "2026-08-09T09:11:29.555Z" }, +] + [[package]] name = "httpx" version = "0.28.1" @@ -996,6 +1010,32 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d2/fd/6668e5aec43ab844de6fc74927e155a3b37bf40d7c3790e49fc0406b6578/httpx_sse-0.4.3-py3-none-any.whl", hash = "sha256:0ac1c9fe3c0afad2e0ebb25a934a59f4c7823b60792691f779fad2c5568830fc", size = 8960, upload-time = "2025-10-10T21:48:21.158Z" }, ] +[[package]] +name = "httpx2" +version = "2.10.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio", marker = "sys_platform != 'emscripten'" }, + { name = "httpcore2", marker = "sys_platform != 'emscripten'" }, + { name = "httpx2-jsfetch", marker = "python_full_version >= '3.12' and sys_platform == 'emscripten'" }, + { name = "idna" }, + { name = "truststore", marker = "sys_platform != 'emscripten'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/bd/3d/f9a8c07a3884f3e5b26205e8436a18b3af61c5d53192c3bea235574dbbec/httpx2-2.10.0.tar.gz", hash = "sha256:8741d7329fe2c7885fc9ceb61c8217acfb87a85f75723714b89ebf7ad7196338", size = 98749, upload-time = "2026-08-09T09:11:33.24Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b9/6d/a637d52449d98a6892d9a4dc0262587afdb6a66f201871842dce5a97b1c1/httpx2-2.10.0-py3-none-any.whl", hash = "sha256:5e3194a432701e1cc6f69a8b1b2fa199ef907013fede8d9a09a2c5b7b8141a18", size = 94355, upload-time = "2026-08-09T09:11:30.882Z" }, +] + +[[package]] +name = "httpx2-jsfetch" +version = "1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/cd/c4/0e5636363151a2a1795e0a77617168b9ca438e1748ec05fc9b5687f93d64/httpx2_jsfetch-1.0.tar.gz", hash = "sha256:70a0e3eabfef7cce5ad9c629f7d01ca05e418f586646f4ddf14782e4c1454c60", size = 6872, upload-time = "2026-08-07T00:13:07.492Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9b/43/832f631d32e4f1211caa2ba368317739fe71f0b8530e4c9d15dc454bac2a/httpx2_jsfetch-1.0-py3-none-any.whl", hash = "sha256:cb916b707601e69a07721aabc8f3f6659be3a6893bc1ff5c6f9e02241df2da32", size = 6382, upload-time = "2026-08-07T00:13:06.567Z" }, +] + [[package]] name = "idna" version = "3.18" @@ -1568,21 +1608,21 @@ wheels = [ [[package]] name = "openai" -version = "2.53.0" +version = "3.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, { name = "distro" }, - { name = "httpx" }, + { name = "httpx2" }, { name = "jiter" }, { name = "pydantic" }, { name = "sniffio" }, { name = "tqdm" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ef/cf/36e3e7235fdf6d125c052acc0970924611b17a20a4fe580596faf4566a65/openai-2.53.0.tar.gz", hash = "sha256:baf5802ad08980e1d9d561e1b996e800c8bcd14af5847c6d0e7a5cc59e4d4116", size = 1099435, upload-time = "2026-08-03T21:42:01.664Z" } +sdist = { url = "https://files.pythonhosted.org/packages/54/8c/2f500e8be09d1ae98c530467962535198b02cd4550cd418bbbaedc8b2910/openai-3.0.0.tar.gz", hash = "sha256:ffd00ef1678d70957e1f1ed98d5bfcf1d661f41ea4482f22e7d0144a66435a49", size = 1123740, upload-time = "2026-08-12T01:55:50.849Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/78/0f/cc6afea3542a5142c5d8fc8211c5e059a8375105d004a41dfa2c7948dbb0/openai-2.53.0-py3-none-any.whl", hash = "sha256:c694ffc747a3c4d1663ef2b07b811315a476164ee5efa3a993967349ebca7618", size = 1659829, upload-time = "2026-08-03T21:41:59.581Z" }, + { url = "https://files.pythonhosted.org/packages/7b/0d/9850e7eddb5e66da4439ed503e78e09ad1fd0195e6df51e4236c75763581/openai-3.0.0-py3-none-any.whl", hash = "sha256:8d32ac3a6647a66910d6cb8a64f0fa5a6c823604b6e82db83d9d055c6709bd51", size = 1665775, upload-time = "2026-08-12T01:55:48.678Z" }, ] [[package]] @@ -2568,6 +2608,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f9/1c/01bfd571a64e7f270e6bab5e33777debe0edc56759233ce84f27dec92d14/tqdm-4.70.0-py3-none-any.whl", hash = "sha256:7f585706bfddbdebf89daac705b2dfcc16890130727d3197ca62c732b4310953", size = 80184, upload-time = "2026-07-27T11:33:13.167Z" }, ] +[[package]] +name = "truststore" +version = "0.10.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" }, +] + [[package]] name = "turnstone" version = "1.8.0a7" @@ -2581,6 +2630,7 @@ dependencies = [ { name = "cryptography" }, { name = "httpx" }, { name = "httpx-sse" }, + { name = "httpx2" }, { name = "lacme" }, { name = "mcp" }, { name = "openai" }, @@ -2636,10 +2686,11 @@ requires-dist = [ { name = "discord-py", marker = "extra == 'discord'", specifier = ">=2.4" }, { name = "httpx", specifier = ">=0.28" }, { name = "httpx-sse", specifier = ">=0.4" }, + { name = "httpx2", specifier = ">=2.7,<3" }, { name = "lacme", specifier = ">=1.0.5" }, { name = "mcp", specifier = ">=1.27,<2" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.14" }, - { name = "openai", specifier = ">=2.45,<3" }, + { name = "openai", specifier = ">=3,<4" }, { name = "pillow", specifier = ">=10" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.2" }, { name = "pydantic", specifier = ">=2.0" },