From a3af5481bc0aa54938981d4ddf7b8e71bf07c49c Mon Sep 17 00:00:00 2001 From: rp-maintainers Date: Sat, 3 Oct 2026 13:51:50 +0000 Subject: [PATCH 1/3] fix(sdk): preserve provider-neutral actual served tiers --- README.md | 66 ++++- src/routeplane/__init__.py | 3 + src/routeplane/async_client.py | 22 +- src/routeplane/client.py | 24 +- src/routeplane/responses.py | 64 ++++ tests/test_served_tier.py | 516 +++++++++++++++++++++++++++++++++ tests/typing/served_tier.py | 57 ++++ 7 files changed, 739 insertions(+), 13 deletions(-) create mode 100644 src/routeplane/responses.py create mode 100644 tests/test_served_tier.py create mode 100644 tests/typing/served_tier.py diff --git a/README.md b/README.md index e2b02c4..f48a07a 100644 --- a/README.md +++ b/README.md @@ -74,9 +74,10 @@ headers are typed. ## 3. Full — the `Routeplane` client with rich metadata -`Routeplane` subclasses `openai.OpenAI`, so everything works exactly as before — -but it wires up auth for you, lets you set default routing once, and can parse -the gateway's response headers into a typed `RouteplaneMeta`. +`Routeplane` subclasses `openai.OpenAI` and keeps its resource/request API. It +wires up auth, lets you set default routing once, and parses the gateway's +response headers into a typed `RouteplaneMeta`. Chat response tier handling is +provider-neutral at runtime; see the served-tier contract below. ```python from routeplane import Routeplane @@ -213,6 +214,65 @@ parses the gateway's `x-routeplane-*` response headers: `provider`, `trace_id`, `request_id`, `cache`, `guardrails`, `hedged`, `shed`, `budget_remaining`, `budget_warning`, `compliance_warning`, `pii_masked`, `idempotent_replayed`. +### Actual served tier + +`create_with_meta()` returns a `RouteplaneChatCompletion`, and `stream_with_meta()` +yields `RouteplaneChatCompletionChunk` objects, on both sync and async clients. +These exported models retain the installed OpenAI response fields and validation +while widening only returned `service_tier` to an optional strict string. Native +labels such as `standard`, `priority`, `batch`, and future provider labels keep +their exact spelling. Missing or `null` tier evidence remains `None`; serialization +with `exclude_unset=True` preserves omission versus explicit `null`. Non-string, +non-null tiers raise `APIResponseValidationError` even in default tolerant mode. +Unrelated response fields continue to follow the vendor's default or opt-in +`_strict_response_validation=True` behavior. + +```python +from routeplane import Routeplane, RouteplaneChatCompletion + +client = Routeplane(api_key="rp_your_gateway_key") +completion, meta = client.create_with_meta( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) +assert isinstance(completion, RouteplaneChatCompletion) +print(completion.service_tier) # provider-reported evidence, possibly None + +with client.stream_with_meta( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) as stream: + for chunk in stream: + # Continue past a finish_reason: tier-only metadata may arrive afterward. + # choices may be empty, and usage may be omitted on that metadata chunk. + if chunk.service_tier is not None: + print(chunk.service_tier) +``` + +The returned tier is never inferred from a requested tier, renamed, or treated as +proof of a free response, tariff eligibility, complete cost, or billing authority. +Streaming keeps the vendor SSE decoder and `[DONE]` termination; no aggregation or +synthetic usage is added. + +The inherited `client.chat.completions` resource signatures still expose the +**OpenAI vendor model types**. Routeplane's exact-model response-processing hook +also uses the neutral models at runtime for those calls, including raw responses, +but it does not widen the inherited static facade. Use the `*_with_meta` helpers +for a typed neutral result. On OpenAI versions that support custom raw parsing, +an explicit neutral model is also available: + +```python +raw = client.chat.completions.with_raw_response.create( + model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], +) +completion = raw.parse(to=RouteplaneChatCompletion) +``` + +OpenAI 1.0.0 has no `parse(to=...)` argument. The helpers and the runtime hook +support that advertised dependency floor using its existing response processor; +`raw.parse()` receives the neutral runtime model there too. A stock `openai.OpenAI` +client and custom response-model targets retain their own validation behavior. +Stream cleanup follows the installed vendor version; OpenAI 1.0.0 itself has no +`Stream.close()` or `AsyncStream.close()` method. + ## Legacy feedback Use the gateway-generated request ID from response metadata, not the provider's diff --git a/src/routeplane/__init__.py b/src/routeplane/__init__.py index ccd9963..679f14b 100644 --- a/src/routeplane/__init__.py +++ b/src/routeplane/__init__.py @@ -45,12 +45,15 @@ TimeseriesData, UsageData, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk __all__ = [ "Routeplane", "AsyncRouteplane", "RouteplaneStream", "AsyncRouteplaneStream", + "RouteplaneChatCompletion", + "RouteplaneChatCompletionChunk", "headers", "RouteplaneMeta", "RouteplaneRateLimits", diff --git a/src/routeplane/async_client.py b/src/routeplane/async_client.py index bff7a78..644dda7 100644 --- a/src/routeplane/async_client.py +++ b/src/routeplane/async_client.py @@ -16,7 +16,7 @@ import httpx import openai -from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai._types import ResponseT from ._streaming import AsyncRouteplaneStream from .client import DEFAULT_BASE_URL @@ -35,6 +35,7 @@ ResidencyResource, StatusResource, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk, _response_type __all__ = ["AsyncRouteplane"] @@ -139,7 +140,18 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: """ return RouteplaneMeta.from_headers(headers) - async def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMeta]: + def _process_response_data( + self, *, data: object, cast_to: type[ResponseT], response: httpx.Response + ) -> ResponseT: + return super()._process_response_data( + data=data, + cast_to=_response_type(cast_to=cast_to, data=data, response=response), + response=response, + ) + + async def create_with_meta( + self, **kwargs: Any + ) -> Tuple[RouteplaneChatCompletion, RouteplaneMeta]: """Chat completion that also returns the gateway :class:`RouteplaneMeta`. :: @@ -147,10 +159,12 @@ async def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, Routepl completion, meta = await client.create_with_meta(model="gpt-4o", messages=[...]) """ raw = await self.chat.completions.with_raw_response.create(**kwargs) - completion = cast(ChatCompletion, raw.parse()) + completion = cast(RouteplaneChatCompletion, raw.parse()) return completion, RouteplaneMeta.from_headers(raw.headers) - async def stream_with_meta(self, **kwargs: Any) -> AsyncRouteplaneStream[ChatCompletionChunk]: + async def stream_with_meta( + self, **kwargs: Any + ) -> AsyncRouteplaneStream[RouteplaneChatCompletionChunk]: """Streaming chat completion that also exposes ``meta`` on the stream. :: diff --git a/src/routeplane/client.py b/src/routeplane/client.py index cf84681..16c9ea3 100644 --- a/src/routeplane/client.py +++ b/src/routeplane/client.py @@ -1,7 +1,9 @@ """The :class:`Routeplane` client — a thin subclass of ``openai.OpenAI``. -Everything the OpenAI SDK can do works unchanged (``client.chat.completions``, -``client.embeddings``, streaming, retries). The subclass adds three things: +The OpenAI resource API remains available (``client.chat.completions``, +``client.embeddings``, streaming, retries). Chat responses use provider-neutral +returned-tier models at runtime; inherited resource annotations stay vendor-owned. +The subclass adds three things: 1. gateway auth + default ``x-routeplane-*`` routing headers, so callers don't repeat them on every request; @@ -17,7 +19,7 @@ import httpx import openai -from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai._types import ResponseT from ._streaming import RouteplaneStream from .headers import headers as build_headers @@ -35,6 +37,7 @@ ResidencyResource, StatusResource, ) +from .responses import RouteplaneChatCompletion, RouteplaneChatCompletionChunk, _response_type __all__ = ["Routeplane"] @@ -153,7 +156,16 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: """ return RouteplaneMeta.from_headers(headers) - def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMeta]: + def _process_response_data( + self, *, data: object, cast_to: type[ResponseT], response: httpx.Response + ) -> ResponseT: + return super()._process_response_data( + data=data, + cast_to=_response_type(cast_to=cast_to, data=data, response=response), + response=response, + ) + + def create_with_meta(self, **kwargs: Any) -> Tuple[RouteplaneChatCompletion, RouteplaneMeta]: """Chat completion that also returns the gateway :class:`RouteplaneMeta`. :: @@ -162,10 +174,10 @@ def create_with_meta(self, **kwargs: Any) -> Tuple[ChatCompletion, RouteplaneMet print(meta.provider, meta.cache) """ raw = self.chat.completions.with_raw_response.create(**kwargs) - completion = cast(ChatCompletion, raw.parse()) + completion = cast(RouteplaneChatCompletion, raw.parse()) return completion, RouteplaneMeta.from_headers(raw.headers) - def stream_with_meta(self, **kwargs: Any) -> RouteplaneStream[ChatCompletionChunk]: + def stream_with_meta(self, **kwargs: Any) -> RouteplaneStream[RouteplaneChatCompletionChunk]: """Streaming chat completion that also exposes ``meta`` on the stream. ``meta`` is populated from the response headers, which arrive before the diff --git a/src/routeplane/responses.py b/src/routeplane/responses.py new file mode 100644 index 0000000..af2f1ef --- /dev/null +++ b/src/routeplane/responses.py @@ -0,0 +1,64 @@ +"""OpenAI-shaped responses with provider-neutral actual served-tier evidence.""" + +from __future__ import annotations + +from typing import Optional, TypeVar, cast + +import httpx +from openai import APIResponseValidationError +from openai.types.chat import ChatCompletion, ChatCompletionChunk +from pydantic import StrictStr + +__all__ = ["RouteplaneChatCompletion", "RouteplaneChatCompletionChunk"] + + +class RouteplaneChatCompletion(ChatCompletion): + """A completion whose returned tier is an exact provider string or ``None``. + + All other fields and validation come from the installed OpenAI model. The + label is evidence of what served the request, not a tariff or cost claim. + """ + + # Intentionally widen the vendor literal union at this one wire-contract seam. + # The 1.0.0 model predates this field, so the assignment ignore is unused there. + service_tier: Optional[StrictStr] = None # type: ignore[assignment, unused-ignore] + + +class RouteplaneChatCompletionChunk(ChatCompletionChunk): + """A stream chunk with the same provider-neutral returned-tier contract.""" + + # Same deliberate widening, including the dependency floor without this field. + service_tier: Optional[StrictStr] = None # type: ignore[assignment, unused-ignore] + + +_T = TypeVar("_T") + + +def _response_type(*, cast_to: type[_T], data: object, response: httpx.Response) -> type[_T]: + """Select the neutral model and check only tier even in tolerant mode. + + The client's existing response processor still controls validation of every + other field. Exact class matching leaves custom models and other APIs alone. + This hook is present at the supported OpenAI 1.0.0 dependency floor, where + raw-response ``parse(to=...)`` is not available yet. + """ + neutral: type[object] + if cast_to is ChatCompletion: + neutral = RouteplaneChatCompletion + elif cast_to is ChatCompletionChunk: + neutral = RouteplaneChatCompletionChunk + elif cast_to in (RouteplaneChatCompletion, RouteplaneChatCompletionChunk): + neutral = cast_to + else: + return cast_to + + if isinstance(data, dict): + tier = data.get("service_tier") + if tier is not None and not isinstance(tier, str): + raise APIResponseValidationError( + response=response, + body=data, + message="Expected response service_tier to be a string or null.", + ) + + return cast("type[_T]", neutral) diff --git a/tests/test_served_tier.py b/tests/test_served_tier.py new file mode 100644 index 0000000..4aaefde --- /dev/null +++ b/tests/test_served_tier.py @@ -0,0 +1,516 @@ +"""Exercise actual vendor decoding, including strict mode and trailing SSE metadata.""" + +import inspect +import json +import subprocess +import sys +from pathlib import Path + +import httpx +import openai +import pytest +import respx +from openai.types.chat import ChatCompletion, ChatCompletionChunk +from openai.types.create_embedding_response import CreateEmbeddingResponse +from pydantic import ValidationError + +from routeplane import ( + AsyncRouteplane, + Routeplane, + RouteplaneChatCompletion, + RouteplaneChatCompletionChunk, +) + +BASE = "https://api.routeplane.ai/v1" +CHAT = f"{BASE}/chat/completions" +OMITTED = object() +TIERS = [OMITTED, None, "standard", "priority", "batch", "default", "future-tier", " Native "] +USAGE = {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5} +META = { + "x-routeplane-provider": "openrouter", + "x-routeplane-request-id": "rp_request", + "x-routeplane-cache": "bypass", +} +REQUEST = { + "model": "model", + "messages": [{"role": "user", "content": "hi"}], + "temperature": 0.25, + "max_tokens": 16, + # extra_body works even at OpenAI 1.0.0, before its named service_tier input. + "extra_body": {"service_tier": "priority", "provider_option": {"keep": True}}, + "extra_headers": {"x-routeplane-provider": "openrouter", "x-custom": "kept"}, +} + + +def completion(tier=OMITTED): + body = { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1, + "model": "model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "hello"}, + "finish_reason": "stop", + } + ], + "usage": USAGE, + "provider_extension": {"keep": True}, + } + if tier is not OMITTED: + body["service_tier"] = tier + return body + + +def chunk(content=None, finish=None): + return { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 1, + "model": "model", + "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": finish}], + } + + +def frames(tier=OMITTED): + # The provider reports its tier only AFTER the finish chunk, without usage. + finish = chunk(None, "stop") + finish["usage"] = USAGE + metadata = {**chunk(), "choices": [], "provider_extension": {"keep": True}} + if tier is not OMITTED: + metadata["service_tier"] = tier + return [chunk("hello"), finish, metadata] + + +def response(tier, streaming): + if not streaming: + return httpx.Response(200, json=completion(tier), headers=META) + body = "".join(f"data: {json.dumps(frame)}\n\n" for frame in frames(tier)) + # A frame after [DONE] must never become another completion chunk. + body += f"data: [DONE]\n\ndata: {json.dumps(chunk('ignored'))}\n\n" + return httpx.Response( + 200, content=body.encode(), headers={"content-type": "text/event-stream", **META} + ) + + +def assert_request(route, streaming): + sent = route.calls.last.request + assert sent.headers["x-routeplane-api-key"] == "rp_test" + assert sent.headers["authorization"] == "Bearer rp_test" + assert sent.headers["x-routeplane-provider"] == "openrouter" + assert sent.headers["x-custom"] == "kept" + expected = { + "model": "model", + "messages": REQUEST["messages"], + "temperature": 0.25, + "max_tokens": 16, + **REQUEST["extra_body"], + } + if streaming: + expected["stream"] = True + assert json.loads(sent.content) == expected + + +def assert_result(parsed, tier, streaming, strict): + if streaming: + assert len(parsed) == 3 + assert all(isinstance(c, RouteplaneChatCompletionChunk) for c in parsed) + assert parsed[0].choices[0].delta.content == "hello" + assert parsed[0].service_tier is None # never infer the requested priority + assert parsed[1].choices[0].finish_reason == "stop" + usage = parsed[1].usage + assert (usage if isinstance(usage, dict) else usage.dict(exclude_unset=True)) == USAGE + result = parsed[-1] + expected = frames(tier)[-1] + assert result.choices == [] + assert "usage" not in result.dict(exclude_unset=True) + else: + assert isinstance(parsed, RouteplaneChatCompletion) + result = parsed + expected = completion(tier) + assert result.usage.total_tokens == 5 + assert result.usage.prompt_tokens == 3 + assert result.usage.completion_tokens == 2 + assert result.service_tier == (None if tier is OMITTED else tier) + assert result.provider_extension == {"keep": True} + # Preserve the installed vendor's serialization of all unrelated fields. + # OpenAI 1.0.0/Pydantic 1's tolerant construct omits extras under exclude_unset, + # though it retains their attributes. Do not change that vendor behavior. + native = ChatCompletionChunk if streaming else ChatCompletion + base_body = {k: v for k, v in expected.items() if k != "service_tier"} + vendor = native(**base_body) if strict else native.construct(**base_body) + expected_dump = vendor.dict(exclude_unset=True) + if tier is not OMITTED: + expected_dump["service_tier"] = tier + # Omission vs null and the exact native spelling survive in every mode. + assert result.dict(exclude_unset=True) == expected_dump + + +@pytest.mark.parametrize("tier", TIERS) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_sync_actual_served_tier(tier, strict, streaming): + route = respx.post(CHAT).mock(return_value=response(tier, streaming)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + provider="openai", + http_client=httpx.Client(), + _strict_response_validation=strict, + ) as client: + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + meta = stream.meta + parsed = list(stream) + else: + parsed, meta = client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert meta.request_id == "rp_request" + assert meta.cache == "bypass" + assert_result(parsed, tier, streaming, strict) + assert_request(route, streaming) + + +@pytest.mark.parametrize("tier", TIERS) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_actual_served_tier(tier, strict, streaming): + route = respx.post(CHAT).mock(return_value=response(tier, streaming)) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + provider="openai", + http_client=httpx.AsyncClient(), + _strict_response_validation=strict, + ) as client: + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + meta = stream.meta + parsed = [c async for c in stream] + else: + parsed, meta = await client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert meta.request_id == "rp_request" + assert meta.cache == "bypass" + assert_result(parsed, tier, streaming, strict) + assert_request(route, streaming) + + +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}]) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_sync_rejects_invalid_tier(tier, strict, streaming): + respx.post(CHAT).mock(return_value=response(tier, streaming)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=strict, + ) as client: + with pytest.raises(openai.APIResponseValidationError, match="service_tier"): + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + list(stream) + else: + client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}]) +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_rejects_invalid_tier(tier, strict, streaming): + respx.post(CHAT).mock(return_value=response(tier, streaming)) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + _strict_response_validation=strict, + ) as client: + with pytest.raises(openai.APIResponseValidationError, match="service_tier"): + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + [c async for c in stream] + else: + await client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("model", [RouteplaneChatCompletion, RouteplaneChatCompletionChunk]) +@pytest.mark.parametrize("tier", [0, 1.5, True, [], {}, b"standard"]) +def test_exported_models_validate_tier_without_coercion(model, tier): + body = completion() if model is RouteplaneChatCompletion else chunk() + body["service_tier"] = tier + with pytest.raises(ValidationError): + model(**body) + + +@pytest.mark.parametrize( + "neutral,native", + [ + (RouteplaneChatCompletion, ChatCompletion), + (RouteplaneChatCompletionChunk, ChatCompletionChunk), + ], +) +def test_neutral_models_only_change_the_returned_tier_field(neutral, native): + field_attribute = "model_fields" if hasattr(native, "model_fields") else "__fields__" + native_fields = getattr(native, field_attribute) + neutral_fields = getattr(neutral, field_attribute) + assert set(neutral_fields) == set(native_fields) | {"service_tier"} + for name, field in native_fields.items(): + if name != "service_tier": + inherited = neutral_fields[name] + assert inherited.annotation == field.annotation + assert inherited.default == field.default + assert inherited.alias == field.alias + + +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +def test_strict_validation_of_other_fields_is_retained(streaming): + body = chunk() if streaming else completion("standard") + body["created"] = {"invalid": True} + mock_response = ( + httpx.Response( + 200, + content=f"data: {json.dumps(body)}\n\n".encode(), + headers={"content-type": "text/event-stream"}, + ) + if streaming + else httpx.Response(200, json=body) + ) + respx.post(CHAT).mock(return_value=mock_response) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=True, + ) as client: + # OpenAI 1.0.0 exposes Pydantic's validation error directly for SSE; + # recent versions wrap it. Preserve that vendor exception boundary. + with pytest.raises((openai.APIResponseValidationError, ValidationError)): + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + list(stream) + else: + client.create_with_meta(**REQUEST) + + +@pytest.mark.parametrize("streaming", [False, True]) +@respx.mock +async def test_async_strict_validation_of_other_fields_is_retained(streaming): + body = chunk() if streaming else completion("standard") + body["created"] = {"invalid": True} + mock_response = ( + httpx.Response( + 200, + content=f"data: {json.dumps(body)}\n\n".encode(), + headers={"content-type": "text/event-stream"}, + ) + if streaming + else httpx.Response(200, json=body) + ) + respx.post(CHAT).mock(return_value=mock_response) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + _strict_response_validation=True, + ) as client: + with pytest.raises((openai.APIResponseValidationError, ValidationError)): + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + [c async for c in stream] + else: + await client.create_with_meta(**REQUEST) + + +@respx.mock +def test_raw_and_inherited_resource_runtime_boundary(): + respx.post(CHAT).mock(return_value=response("batch", False)) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + _strict_response_validation=True, + ) as client: + direct = client.chat.completions.create(**REQUEST) + assert isinstance(direct, RouteplaneChatCompletion) + assert direct.service_tier == "batch" + raw = client.chat.completions.with_raw_response.create(**REQUEST) + # Works at the advertised floor; the return annotation remains vendor-owned. + assert isinstance(raw.parse(), RouteplaneChatCompletion) + if "to" in inspect.signature(raw.parse).parameters: + assert raw.parse(to=RouteplaneChatCompletion).service_tier == "batch" + + +@respx.mock +def test_retries_stay_on_the_vendor_transport(): + route = respx.post(CHAT).mock(side_effect=[httpx.Response(500), response("standard", False)]) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(), + max_retries=1, + ) as client: + parsed, _ = client.create_with_meta(**REQUEST) + assert parsed.service_tier == "standard" + assert route.call_count == 2 + assert_request(route, False) + + +@respx.mock +async def test_async_retries_stay_on_the_vendor_transport(): + route = respx.post(CHAT).mock(side_effect=[httpx.Response(500), response("standard", False)]) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(), + max_retries=1, + ) as client: + parsed, _ = await client.create_with_meta(**REQUEST) + assert parsed.service_tier == "standard" + assert route.call_count == 2 + assert_request(route, False) + + +class SyncFrames(httpx.SyncByteStream): + def __init__(self): + self.read_count = 0 + self.closed = False + + def __iter__(self): + for frame in frames("batch"): + self.read_count += 1 + yield f"data: {json.dumps(frame)}\n\n".encode() + yield b"data: [DONE]\n\n" + + def close(self): + self.closed = True + + +class AsyncFrames(httpx.AsyncByteStream): + def __init__(self): + self.read_count = 0 + self.closed = False + + async def __aiter__(self): + for frame in frames("batch"): + self.read_count += 1 + yield f"data: {json.dumps(frame)}\n\n".encode() + yield b"data: [DONE]\n\n" + + async def aclose(self): + self.closed = True + + +def test_sync_stream_stays_incremental_and_closes_on_exit(): + wire = SyncFrames() + transport = httpx.MockTransport( + lambda request: httpx.Response( + 200, stream=wire, headers={"content-type": "text/event-stream", **META} + ) + ) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.Client(transport=transport), + ) as client: + with client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + assert wire.read_count == 0 + assert next(stream).choices[0].delta.content == "hello" + assert wire.read_count == 1 + # Earliest OpenAI's Stream had no close method; that vendor limitation + # predates this change. No new eager draining is introduced at the floor. + if hasattr(openai.Stream, "close"): + assert wire.closed + assert wire.read_count == 1 + + +async def test_async_stream_stays_incremental_and_closes_on_exit(): + wire = AsyncFrames() + transport = httpx.MockTransport( + lambda request: httpx.Response( + 200, stream=wire, headers={"content-type": "text/event-stream", **META} + ) + ) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=httpx.AsyncClient(transport=transport), + ) as client: + async with await client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + assert wire.read_count == 0 + assert (await stream.__anext__()).choices[0].delta.content == "hello" + assert wire.read_count == 1 + if hasattr(openai.AsyncStream, "close"): + assert wire.closed + assert wire.read_count == 1 + + +class CustomCompletion(ChatCompletion): + provider_marker: str + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("client_type", [Routeplane, AsyncRouteplane]) +def test_custom_and_other_response_targets_remain_untouched(client_type, strict): + client = client_type(api_key="rp_test", _strict_response_validation=strict) + wire_response = httpx.Response(200, request=httpx.Request("POST", CHAT)) + custom_body = {**completion(), "provider_marker": "kept"} + custom = client._process_response_data( + data=custom_body, + cast_to=CustomCompletion, + response=wire_response, + ) + assert type(custom) is CustomCompletion + assert custom.provider_marker == "kept" + embedding_body = { + "object": "list", + "model": "model", + "data": [{"object": "embedding", "index": 0, "embedding": [0.5, 0.25]}], + "usage": {"prompt_tokens": 3, "total_tokens": 3}, + "service_tier": 123, # An extra on another API must not acquire chat validation. + } + embeddings = client._process_response_data( + data=embedding_body, + cast_to=CreateEmbeddingResponse, + response=wire_response, + ) + assert type(embeddings) is CreateEmbeddingResponse + assert embeddings.data[0].embedding == [0.5, 0.25] + assert embeddings.service_tier == 123 + # No request is sent; close both resource clients deterministically. + if isinstance(client, Routeplane): + client.close() + else: + import asyncio + + asyncio.run(client.close()) + + +def test_served_tier_consumer_type_contract(): + """The existing pytest CI gate must also execute the consumer type fixture.""" + root = Path(__file__).resolve().parents[1] + checked = subprocess.run( + [ + sys.executable, + "-m", + "mypy", + "--strict", + "--cache-dir", + "/dev/null", + "src/", + "tests/typing/served_tier.py", + ], + cwd=root, + capture_output=True, + text=True, + timeout=60, + ) + assert checked.returncode == 0, checked.stdout + checked.stderr diff --git a/tests/typing/served_tier.py b/tests/typing/served_tier.py new file mode 100644 index 0000000..dcf0212 --- /dev/null +++ b/tests/typing/served_tier.py @@ -0,0 +1,57 @@ +"""Static consumer contract: run ``mypy --strict src/ tests/typing/served_tier.py``.""" + +from typing import Optional, Tuple + +from typing_extensions import assert_type + +from routeplane import ( + AsyncRouteplane, + AsyncRouteplaneStream, + Routeplane, + RouteplaneChatCompletion, + RouteplaneChatCompletionChunk, + RouteplaneMeta, + RouteplaneStream, +) + + +def returned_tier(completion: RouteplaneChatCompletion) -> Optional[str]: + assert_type(completion.service_tier, Optional[str]) + completion.service_tier = "unknown-native" + completion.service_tier = None + completion.service_tier = 123 # type: ignore[assignment] + return completion.service_tier + + +def chunk_tier(chunk: RouteplaneChatCompletionChunk) -> Optional[str]: + assert_type(chunk.service_tier, Optional[str]) + chunk.service_tier = "batch" + chunk.service_tier = None + chunk.service_tier = False # type: ignore[assignment] + return chunk.service_tier + + +def sync_helpers(client: Routeplane) -> None: + result = client.create_with_meta(model="model", messages=[]) + assert_type(result, Tuple[RouteplaneChatCompletion, RouteplaneMeta]) + tier: Optional[str] = result[0].service_tier + stream = client.stream_with_meta(model="model", messages=[]) + assert_type(stream, RouteplaneStream[RouteplaneChatCompletionChunk]) + for chunk in stream: + assert_type(chunk, RouteplaneChatCompletionChunk) + assert_type(chunk.service_tier, Optional[str]) + tier = chunk.service_tier + print(tier) + + +async def async_helpers(client: AsyncRouteplane) -> None: + result = await client.create_with_meta(model="model", messages=[]) + assert_type(result, Tuple[RouteplaneChatCompletion, RouteplaneMeta]) + tier: Optional[str] = result[0].service_tier + stream = await client.stream_with_meta(model="model", messages=[]) + assert_type(stream, AsyncRouteplaneStream[RouteplaneChatCompletionChunk]) + async for chunk in stream: + assert_type(chunk, RouteplaneChatCompletionChunk) + assert_type(chunk.service_tier, Optional[str]) + tier = chunk.service_tier + print(tier) From 181b982164d6462b24d64b62374289a3be885fbd Mon Sep 17 00:00:00 2001 From: rp-maintainers Date: Sat, 3 Oct 2026 14:03:11 +0000 Subject: [PATCH 2/3] fix(sdk): preserve opaque vendor response transport typing --- src/routeplane/async_client.py | 3 +- src/routeplane/client.py | 3 +- src/routeplane/responses.py | 9 ++-- tests/test_served_tier.py | 81 ++++++++++++++++++++++++++++++++++ 4 files changed, 91 insertions(+), 5 deletions(-) diff --git a/src/routeplane/async_client.py b/src/routeplane/async_client.py index 644dda7..9be52e1 100644 --- a/src/routeplane/async_client.py +++ b/src/routeplane/async_client.py @@ -141,8 +141,9 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: return RouteplaneMeta.from_headers(headers) def _process_response_data( - self, *, data: object, cast_to: type[ResponseT], response: httpx.Response + self, *, data: object, cast_to: type[ResponseT], response: Any ) -> ResponseT: + # SDK versions use httpx or httpx2; forward the opaque response unchanged. return super()._process_response_data( data=data, cast_to=_response_type(cast_to=cast_to, data=data, response=response), diff --git a/src/routeplane/client.py b/src/routeplane/client.py index 16c9ea3..2d1717f 100644 --- a/src/routeplane/client.py +++ b/src/routeplane/client.py @@ -157,8 +157,9 @@ def meta_from_headers(headers: Mapping[str, str]) -> RouteplaneMeta: return RouteplaneMeta.from_headers(headers) def _process_response_data( - self, *, data: object, cast_to: type[ResponseT], response: httpx.Response + self, *, data: object, cast_to: type[ResponseT], response: Any ) -> ResponseT: + # SDK versions use httpx or httpx2; forward the opaque response unchanged. return super()._process_response_data( data=data, cast_to=_response_type(cast_to=cast_to, data=data, response=response), diff --git a/src/routeplane/responses.py b/src/routeplane/responses.py index af2f1ef..75bc47e 100644 --- a/src/routeplane/responses.py +++ b/src/routeplane/responses.py @@ -2,9 +2,8 @@ from __future__ import annotations -from typing import Optional, TypeVar, cast +from typing import Any, Optional, TypeVar, cast -import httpx from openai import APIResponseValidationError from openai.types.chat import ChatCompletion, ChatCompletionChunk from pydantic import StrictStr @@ -34,13 +33,17 @@ class RouteplaneChatCompletionChunk(ChatCompletionChunk): _T = TypeVar("_T") -def _response_type(*, cast_to: type[_T], data: object, response: httpx.Response) -> type[_T]: +def _response_type(*, cast_to: type[_T], data: object, response: Any) -> type[_T]: """Select the neutral model and check only tier even in tolerant mode. The client's existing response processor still controls validation of every other field. Exact class matching leaves custom models and other APIs alone. This hook is present at the supported OpenAI 1.0.0 dependency floor, where raw-response ``parse(to=...)`` is not available yet. + + The opaque vendor response is forwarded unchanged: supported SDK versions + use unrelated httpx/httpx2 classes and export no shared response type. + Data, model selection, and returned models retain their typed contracts. """ neutral: type[object] if cast_to is ChatCompletion: diff --git a/tests/test_served_tier.py b/tests/test_served_tier.py index 4aaefde..06cb99e 100644 --- a/tests/test_served_tier.py +++ b/tests/test_served_tier.py @@ -1,10 +1,12 @@ """Exercise actual vendor decoding, including strict mode and trailing SSE metadata.""" +import importlib import inspect import json import subprocess import sys from pathlib import Path +from typing import get_type_hints import httpx import openai @@ -514,3 +516,82 @@ def test_served_tier_consumer_type_contract(): timeout=60, ) assert checked.returncode == 0, checked.stdout + checked.stderr + + +def vendor_http_backend(): + """Use the installed SDK's actual transport, including recent httpx2.""" + response_class = get_type_hints(openai.APIResponseValidationError.__init__)["response"] + return importlib.import_module(response_class.__module__.split(".")[0]) + + +def vendor_wire_response(backend, tier, streaming): + if not streaming: + return backend.Response(200, json=completion(tier), headers=META) + body = "".join(f"data: {json.dumps(frame)}\n\n" for frame in frames(tier)) + return backend.Response( + 200, + content=(body + "data: [DONE]\n\n").encode(), + headers={"content-type": "text/event-stream", **META}, + ) + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +def test_sync_with_native_vendor_transport(strict, streaming): + backend = vendor_http_backend() + tier = "batch" if streaming else "standard" + transport = backend.MockTransport( + lambda request: vendor_wire_response(backend, tier, streaming) + ) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.Client(transport=transport), + _strict_response_validation=strict, + ) as client: + if streaming: + with client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + parsed = list(stream) + else: + parsed, meta = client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert_result(parsed, tier, streaming, strict) + + +@pytest.mark.parametrize("strict", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +async def test_async_with_native_vendor_transport(strict, streaming): + backend = vendor_http_backend() + tier = "batch" if streaming else "standard" + transport = backend.MockTransport( + lambda request: vendor_wire_response(backend, tier, streaming) + ) + async with AsyncRouteplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.AsyncClient(transport=transport), + _strict_response_validation=strict, + ) as client: + if streaming: + async with await client.stream_with_meta(**REQUEST) as stream: + assert stream.meta.provider == "openrouter" + parsed = [c async for c in stream] + else: + parsed, meta = await client.create_with_meta(**REQUEST) + assert meta.provider == "openrouter" + assert_result(parsed, tier, streaming, strict) + + +def test_invalid_tier_keeps_the_native_vendor_response_object(): + backend = vendor_http_backend() + wire = vendor_wire_response(backend, 123, False) + transport = backend.MockTransport(lambda request: wire) + with Routeplane( + api_key="rp_test", + base_url=BASE, + http_client=backend.Client(transport=transport), + ) as client: + with pytest.raises(openai.APIResponseValidationError) as failure: + client.create_with_meta(**REQUEST) + assert failure.value.response is wire From 601d7600c146428f84baf18c462c3b78c98e1e94 Mon Sep 17 00:00:00 2001 From: rp-maintainers Date: Sat, 3 Oct 2026 14:11:38 +0000 Subject: [PATCH 3/3] fix(sdk): inspect native transport without annotation evaluation --- tests/test_served_tier.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_served_tier.py b/tests/test_served_tier.py index 06cb99e..9b3c932 100644 --- a/tests/test_served_tier.py +++ b/tests/test_served_tier.py @@ -1,12 +1,10 @@ """Exercise actual vendor decoding, including strict mode and trailing SSE metadata.""" -import importlib import inspect import json import subprocess import sys from pathlib import Path -from typing import get_type_hints import httpx import openai @@ -520,8 +518,10 @@ def test_served_tier_consumer_type_contract(): def vendor_http_backend(): """Use the installed SDK's actual transport, including recent httpx2.""" - response_class = get_type_hints(openai.APIResponseValidationError.__init__)["response"] - return importlib.import_module(response_class.__module__.split(".")[0]) + # Read the trusted SDK binding without evaluating postponed annotations, + # which can contain union syntax Python 3.9 cannot evaluate. + sdk_globals = openai.APIResponseValidationError.__init__.__globals__ + return sdk_globals["httpx2"] if "httpx2" in sdk_globals else sdk_globals["httpx"] def vendor_wire_response(backend, tier, streaming):