fix(responses): drop bridge-minted reasoning items from OpenAI replays (#44132)

* fix(responses): drop bridge-minted reasoning items from OpenAI replays

* fix(responses): send id-less stored reasoning items without a made-up id

The chat-to-Responses bridge gave a stored reasoning item with no id an rs_<n> id that OpenAI rejects (404 without encrypted content, 400 with it); an id-less item is accepted and verified by OpenAI itself. Decoding encrypted_content now keeps the verifiable thinking blocks of a mixed array instead of rejecting the whole array, and the verifiable-block rule lives in the shared module.

* test(integration): audit minted reasoning item replay across responses, chat bridge and messages

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-02 19:39:48 -07:00 • committed by GitHub
parent dac31e3d5d
commit fe910889f7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1841 additions and 40 deletions

View file

@ -271,16 +271,20 @@ def _flat_responses_tool_choice(choice_type: str, name: str) -> ToolChoiceFuncti
def _reasoning_item_to_response_input(
r_item: ChatCompletionReasoningItem,
) -> dict[str, object]:
"""Convert a stored ChatCompletionReasoningItem back to a Responses API input item."""
r_input: Final[dict[str, object]] = {
"""Convert a stored ChatCompletionReasoningItem back to a Responses API input item.
An item without an id is sent without one: the Responses API accepts that and
verifies the encrypted content on its own, while it rejects any id it did not mint.
"""
item_id: Final = r_item.get("id")
encrypted_content: Final = r_item.get("encrypted_content")
return {
"type": "reasoning",
"id": r_item.get("id") or f"rs_{id(r_item)}",
**({"id": item_id} if item_id else {}),
# summary is always required by the Responses API, even when empty
"summary": r_item.get("summary") or [],
**({"encrypted_content": encrypted_content} if encrypted_content else {}),
}
if r_item.get("encrypted_content"):
r_input["encrypted_content"] = r_item["encrypted_content"]
return r_input
class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):

View file

@ -24,6 +24,7 @@ from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.openai.chat.gpt_5_transformation import is_gpt_reasoning_series_name
from litellm.responses.litellm_completion_transformation.custom_tools import TOOL_CALL_ITEM_ID_PREFIX_BY_TYPE
from litellm.responses.litellm_completion_transformation.reasoning_items import is_litellm_minted_reasoning_item
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import *
from litellm.types.responses.main import *
@ -46,6 +47,7 @@ _NO_TOOL_UPDATE: Final[Mapping[str, object]] = MappingProxyType({})
_MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS: Final = ("gpt-4", "gpt-3.5", "chatgpt-4o", "o1", "o3", "o4")
_PROVIDERS_WITH_OPENAI_SCHEMA_VALIDATOR: Final = frozenset({LlmProviders.AZURE, LlmProviders.OPENAI})
_PROVIDERS_VALIDATING_TOOL_CALL_ITEM_IDS: Final = frozenset({LlmProviders.AZURE, LlmProviders.OPENAI})
_PROVIDERS_REPLAYING_ONLY_THEIR_OWN_REASONING: Final = _PROVIDERS_VALIDATING_TOOL_CALL_ITEM_IDS
class _ReasoningSupportEntry(BaseModel):
@ -317,7 +319,7 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
tools: Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None,
litellm_params: GenericLiteLLMParams,
) -> tuple[str | ResponseInputParam, Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None]:
validated_input: Final = self._validate_input_param(input)
validated_input: Final = self._validate_input_param(self._drop_bridge_minted_reasoning_items(input))
stripped_input, stripped_tools = self.remove_cache_control_flag_from_input_and_tools(
model=model, input=validated_input, tools=tools
)
@ -390,6 +392,12 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
return input, tools
def _drop_bridge_minted_reasoning_items(self, input: str | ResponseInputParam) -> str | ResponseInputParam:
if self.custom_llm_provider not in _PROVIDERS_REPLAYING_ONLY_THEIR_OWN_REASONING or not isinstance(input, list):
return input
replayable_items: Final = [item for item in input if not is_litellm_minted_reasoning_item(item)]
return cast("ResponseInputParam", replayable_items) # cast-ok: the surviving items keep their shape
def _drop_foreign_tool_call_item_ids(self, input: str | ResponseInputParam) -> str | ResponseInputParam:
if self.custom_llm_provider not in _PROVIDERS_VALIDATING_TOOL_CALL_ITEM_IDS or not isinstance(input, list):
return input

View file

@ -0,0 +1,73 @@
import json
import uuid
from collections.abc import Iterator, Mapping, Sequence
from typing import Final
from pydantic import BaseModel, TypeAdapter, ValidationError
REASONING_ITEM_ID_PREFIX: Final = "rs_"
_JSON_LIST: Final = TypeAdapter(list[object])
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
def mint_reasoning_item_id() -> str:
return f"{REASONING_ITEM_ID_PREFIX}{uuid.uuid4()}"
def is_verifiable_thinking_block(block: Mapping[str, object]) -> bool:
block_type: Final = block.get("type")
if block_type == "thinking":
return bool(block.get("signature"))
if block_type == "redacted_thinking":
return bool(block.get("data"))
return False
def encode_thinking_blocks(thinking_blocks: Sequence[Mapping[str, object]]) -> str | None:
preserved: Final = [block for block in thinking_blocks if is_verifiable_thinking_block(block)]
return json.dumps(preserved, separators=(",", ":")) if preserved else None
def _json_objects(members: Sequence[object]) -> Iterator[Mapping[str, object]]:
for member in members:
try:
yield _JSON_OBJECT.validate_python(member)
except ValidationError:
continue
def decode_thinking_blocks(encrypted_content: object) -> tuple[Mapping[str, object], ...] | None:
if not isinstance(encrypted_content, str) or not encrypted_content.strip():
return None
try:
decoded: Final = _JSON_LIST.validate_json(encrypted_content)
except ValidationError:
return None
blocks: Final = tuple(block for block in _json_objects(decoded) if is_verifiable_thinking_block(block))
return blocks or None
def is_minted_reasoning_item_id(item_id: object) -> bool:
if not isinstance(item_id, str) or not item_id.startswith(REASONING_ITEM_ID_PREFIX):
return False
suffix: Final = item_id.removeprefix(REASONING_ITEM_ID_PREFIX)
try:
parsed: Final = uuid.UUID(suffix)
except ValueError:
return False
return parsed.version == 4 and str(parsed) == suffix
def is_litellm_minted_reasoning_item(item: object) -> bool:
try:
fields: Final = _JSON_OBJECT.validate_python(
item.model_dump(exclude_none=True) if isinstance(item, BaseModel) else item
)
except ValidationError:
return False
if fields.get("type") != "reasoning":
return False
return (
is_minted_reasoning_item_id(fields.get("id"))
or decode_thinking_blocks(fields.get("encrypted_content")) is not None
)

View file

@ -11,6 +11,7 @@ from litellm.responses.litellm_completion_transformation.custom_tools import (
is_custom_tool_call,
serialize_tool_call_arguments,
)
from litellm.responses.litellm_completion_transformation.reasoning_items import mint_reasoning_item_id
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
@ -944,7 +945,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if (hasattr(delta, "reasoning_content") and delta.reasoning_content) or _delta_has_signed_thinking_block(delta):
self._reasoning_active = True
if self._cached_reasoning_item_id is None:
self._cached_reasoning_item_id = f"rs_{uuid.uuid4()}"
self._cached_reasoning_item_id = mint_reasoning_item_id()
self._reasoning_item_id = self._cached_reasoning_item_id
event = OutputItemAddedEvent(
@ -1027,7 +1028,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
# Ensure we have a valid reasoning_item_id
self._cached_reasoning_item_id = (
self._reasoning_item_id or self._cached_reasoning_item_id or f"rs_{uuid.uuid4()}"
self._reasoning_item_id
or self._cached_reasoning_item_id
or mint_reasoning_item_id()
)
reasoning_item_id = self._cached_reasoning_item_id
@ -1186,7 +1189,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
reasoning_content: Final = chunk.choices[0].delta.reasoning_content
if self._cached_reasoning_item_id is None:
self._cached_reasoning_item_id = f"rs_{uuid.uuid4()}"
self._cached_reasoning_item_id = mint_reasoning_item_id()
return ReasoningSummaryTextDeltaEvent(
type=ResponsesAPIStreamEvents.REASONING_SUMMARY_TEXT_DELTA,

View file

@ -105,6 +105,7 @@ from .custom_tools import (
unwrap_custom_tool_arguments,
validated_allowed_callers,
)
from .reasoning_items import decode_thinking_blocks, encode_thinking_blocks, mint_reasoning_item_id
NamespaceNameMap: TypeAlias = Mapping[str, tuple[str, str]]
NamespaceTool: TypeAlias = Mapping[str, object]
@ -1494,39 +1495,16 @@ class LiteLLMCompletionResponsesConfig:
Returns None for anything this deployment did not write, so a genuinely
opaque blob is still skipped rather than forwarded as garbage.
"""
encrypted_content: Final[object] = input_item.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content.strip():
decoded: Final = decode_thinking_blocks(input_item.get("encrypted_content"))
if decoded is None:
return None
try:
decoded: Final[object] = cast(object, json.loads(encrypted_content)) # cast-ok: json.loads returns Any
except ValueError:
return None
if not isinstance(decoded, list):
return None
blocks: Final = tuple(
cast( # cast-ok: shape validated by _is_replayable_thinking_block
return tuple(
cast( # cast-ok: decode_thinking_blocks keeps verifiable thinking blocks only
ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
block,
)
for block in decoded
if isinstance(block, Mapping) and LiteLLMCompletionResponsesConfig._is_replayable_thinking_block(block)
)
return blocks or None
@staticmethod
def _is_replayable_thinking_block(block: Mapping[str, object]) -> bool:
"""
A thinking block is only worth replaying when the provider can verify
it: a ``thinking`` block needs its signature, a ``redacted_thinking``
block needs its opaque data.
"""
block_type: Final[object] = block.get("type")
if block_type == "thinking":
return bool(block.get("signature"))
if block_type == "redacted_thinking":
return bool(block.get("data"))
return False
@staticmethod
def _is_input_item_tool_call_output(input_item: Mapping[str, object]) -> bool:
@ -2559,8 +2537,7 @@ class LiteLLMCompletionResponsesConfig:
@staticmethod
def _encode_thinking_blocks(message: Message) -> str | None:
thinking_blocks: Final[Sequence[Mapping[str, object]]] = getattr(message, "thinking_blocks", None) or ()
preserved: Final = tuple(block for block in thinking_blocks if block.get("signature") or block.get("data"))
return json.dumps(preserved, separators=(",", ":")) if preserved else None
return encode_thinking_blocks(thinking_blocks)
@staticmethod
def _extract_reasoning_output_items(
@ -2577,7 +2554,7 @@ class LiteLLMCompletionResponsesConfig:
return [
GenericResponseOutputItem(
type="reasoning",
id=f"rs_{uuid.uuid4()}",
id=mint_reasoning_item_id(),
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
),

View file

@ -0,0 +1,271 @@
from __future__ import annotations
import base64
import json
import os
import re
import uuid
from collections import deque
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import Final
from urllib.parse import urlsplit
from integration._support import claude_code as cc
from integration._support.wire import Reply, Request
from pydantic import JsonValue, TypeAdapter
MARKER: Final = re.compile(r"marker-([0-9a-f]{32})")
THOUGHT: Final = "plan the answer"
USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 30, "output_tokens": 5, "total_tokens": 35}
CHAT_USAGE: Final[dict[str, JsonValue]] = {"prompt_tokens": 30, "completion_tokens": 5, "total_tokens": 35}
CLAUDE_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 20, "output_tokens": 7}
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
ITEMS: Final = TypeAdapter(list[dict[str, JsonValue]])
MINTED_ID: Final = re.compile(r"^rs_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
_INNER_ID: Final = re.compile(r"response_id:([^;]+)")
_WRAPPER_PREFIX: Final = "litellm:custom_llm_provider:"
_PROXY_WRAPPED_PREFIX: Final = "litellm_proxy:responses_api:response_id:"
def signature(marker: str) -> str:
return f"sig-{marker}"
def answer(marker: str | None) -> str:
return "ok" if marker is None else f"answer marker-{marker}"
def newest_marker(text: str) -> str | None:
found: Final = MARKER.findall(text)
return str(found[-1]) if found else None
def error(status: int, message: str, code: str) -> Reply:
body: Final = {"error": {"message": message, "type": "invalid_request_error", "param": None, "code": code}}
return Reply(status=status, body=json.dumps(body).encode())
def sse(event: Mapping[str, JsonValue]) -> bytes:
return f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode()
def chat_sse(frame: Mapping[str, JsonValue]) -> bytes:
return b"data: " + json.dumps(frame).encode() + b"\n\n"
def thinking_json(marker: str) -> str:
return json.dumps([{"type": "thinking", "thinking": THOUGHT, "signature": signature(marker)}])
def minted_item(marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
return {"type": "reasoning", "id": f"rs_{uuid.uuid4()}", "encrypted_content": thinking_json(marker), **extra}
def agents_sdk_history(marker: str, *reasoning: dict[str, JsonValue]) -> list[dict[str, JsonValue]]:
return [
{"role": "user", "content": "Pick a city and look up its weather."},
*reasoning,
{
"type": "message",
"id": f"msg_{uuid.uuid4()}",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": "Prague", "annotations": []}],
},
{"type": "function_call", "call_id": "call_weather", "name": "weather", "arguments": '{"city": "Prague"}'},
{"type": "function_call_output", "call_id": "call_weather", "output": '{"celsius": 18}'},
{"role": "user", "content": f"Now answer marker-{marker}"},
]
def without(history: Sequence[dict[str, JsonValue]], dropped: Sequence[dict[str, JsonValue]]) -> list[JsonValue]:
return [item for item in history if all(item is not gone for gone in dropped)]
def reasoning_items(body: Mapping[str, JsonValue]) -> list[dict[str, JsonValue]]:
return [item for item in ITEMS.validate_python(body["input"]) if item.get("type") == "reasoning"]
def _decoded_wrapper(value: str) -> str | None:
try:
decoded: Final = base64.b64decode(value.removeprefix("resp_"), validate=True).decode()
except (ValueError, UnicodeDecodeError):
return None
return decoded if decoded.startswith(_WRAPPER_PREFIX) else None
def response_identities(value: str) -> frozenset[str]:
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_if_encrypted_with
salt: Final = os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt")
opened: Final = decrypt_if_encrypted_with(value.removeprefix("resp_"), salt)
sealed: Final = opened is not None and opened.startswith(_PROXY_WRAPPED_PREFIX)
wrapped: Final = opened.removeprefix(_PROXY_WRAPPED_PREFIX).split(";", 1)[0] if sealed and opened else value
decoded: Final = _decoded_wrapper(wrapped)
if decoded is None:
return frozenset({wrapped})
inner: Final = _INNER_ID.search(decoded)
assert inner is not None, decoded
return frozenset({wrapped, inner.group(1)})
def same_response(left: str, right: str) -> bool:
return bool(response_identities(left) & response_identities(right))
@dataclass(frozen=True, slots=True)
class ResponsesVendor:
claude_model: str = cc.OPUS
pause_between_chunks: float = 0
minted: deque[str] = field(default_factory=deque)
def respond(self, request: Request) -> Reply:
path: Final = urlsplit(request.target).path
if request.method == "GET":
return Reply(body=json.dumps({"object": "list", "data": [{"id": "gpt-5.6", "object": "model"}]}).encode())
body: Final = JSON_OBJECT.validate_json(request.body)
if path.endswith("/messages"):
return self._claude(body)
if path.endswith("/chat/completions"):
return self._chat(body)
assert path.endswith("/responses"), request.target
verdict: Final = self._verdict(body)
return verdict if verdict is not None else self._responses(body)
def _verdict(self, body: Mapping[str, JsonValue]) -> Reply | None:
received: Final = body.get("input")
if isinstance(received, str):
return None
items: Final = ITEMS.validate_python(received)
if not items and "previous_response_id" not in body:
return error(
400, 'One of "input" or "previous_response_id" must be provided.', "missing_required_parameter"
)
for index, item in enumerate(items):
if item.get("type") != "reasoning":
continue
item_id: Final = item.get("id")
if item_id is not None and not isinstance(item_id, str):
return error(400, f"Invalid type for 'input[{index}].id': expected a string.", "invalid_type")
if "summary" not in item:
return error(
400, f"Missing required parameter: 'input[{index}].summary'.", "missing_required_parameter"
)
if item_id == "":
return error(400, f"Invalid 'input[{index}].id': empty string.", "invalid_value")
if isinstance(item_id, str) and item_id not in self.minted:
return error(404, f"Item with id '{item_id}' not found.", "invalid_request_error")
return None
def _responses(self, body: Mapping[str, JsonValue]) -> Reply:
marker: Final = newest_marker(json.dumps(body))
tag: Final = uuid.uuid4().hex
self.minted.append(f"rs_{tag}")
reasoning: Final[dict[str, JsonValue]] = {
"id": f"rs_{tag}",
"type": "reasoning",
"summary": [],
"encrypted_content": f"gAAAAA-vendor-{tag}",
}
message: Final[dict[str, JsonValue]] = {
"id": f"msg_{tag}",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": answer(marker), "annotations": []}],
}
response: Final[dict[str, JsonValue]] = {
"id": f"resp_{tag}",
"object": "response",
"created_at": 1,
"status": "completed",
"model": body["model"],
"output": [reasoning, message],
"usage": USAGE,
}
if body.get("stream") is not True:
return Reply(body=json.dumps(response).encode())
events: Final[tuple[dict[str, JsonValue], ...]] = (
{
"type": "response.created",
"sequence_number": 0,
"response": {**response, "status": "in_progress", "output": []},
},
{"type": "response.output_item.added", "sequence_number": 1, "output_index": 0, "item": reasoning},
{"type": "response.output_item.done", "sequence_number": 2, "output_index": 0, "item": reasoning},
{
"type": "response.output_item.added",
"sequence_number": 3,
"output_index": 1,
"item": {**message, "content": []},
},
{
"type": "response.output_text.delta",
"sequence_number": 4,
"item_id": f"msg_{tag}",
"output_index": 1,
"content_index": 0,
"delta": answer(marker),
},
{"type": "response.output_item.done", "sequence_number": 5, "output_index": 1, "item": message},
{"type": "response.completed", "sequence_number": 6, "response": response},
)
return Reply(
content_type="text/event-stream",
chunks=tuple(sse(event) for event in events),
pause_between_chunks=self.pause_between_chunks,
)
def _chat(self, body: Mapping[str, JsonValue]) -> Reply:
marker: Final = newest_marker(json.dumps(body))
tag: Final = uuid.uuid4().hex
if body.get("stream") is not True:
return Reply(
body=json.dumps(
{
"id": f"chatcmpl-{tag}",
"object": "chat.completion",
"created": 1,
"model": body["model"],
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": answer(marker)},
"finish_reason": "stop",
}
],
"usage": CHAT_USAGE,
}
).encode()
)
chunk: Final[dict[str, JsonValue]] = {
"id": f"chatcmpl-{tag}",
"object": "chat.completion.chunk",
"created": 1,
"model": body["model"],
}
frames: Final[tuple[dict[str, JsonValue], ...]] = (
{**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": answer(marker)}}]},
{**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], "usage": CHAT_USAGE},
)
return Reply(
content_type="text/event-stream",
chunks=(*(chat_sse(frame) for frame in frames), b"data: [DONE]\n\n"),
pause_between_chunks=self.pause_between_chunks,
)
def _claude(self, body: Mapping[str, JsonValue]) -> Reply:
marker: Final = newest_marker(json.dumps(body))
content: Final = (
{"type": "thinking", "thinking": THOUGHT, "signature": signature(marker or "")},
{"type": "text", "text": answer(marker)},
)
identity: Final = f"msg_{uuid.uuid4().hex}"
if body.get("stream") is True:
return Reply(
content_type="text/event-stream",
chunks=cc.message_stream(identity, self.claude_model, content, CLAUDE_USAGE),
pause_between_chunks=self.pause_between_chunks,
)
return Reply(body=cc.message_reply(identity, self.claude_model, content, CLAUDE_USAGE))

View file

@ -0,0 +1,523 @@
import asyncio
import json
import re
import signal
import socket
import threading
import time
import uuid
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from pathlib import Path
from queue import SimpleQueue
from types import MappingProxyType
from typing import Final, Literal
from urllib.parse import urlsplit
import httpx
import psutil
import pytest
import websockets
import yaml
from integration._support import claude_code as cc
from integration._support import responses_vendor as rv
from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment
from integration._support.database import read_rows
from integration._support.process import OwnedProxy, owned_proxy_process
from integration._support.tls import server_context, write_self_signed_cert
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue
_GPT: Final = "gpt-5.6"
_CODEX: Final = "gpt-5.3-codex"
_OPENAI_KEY: Final = "synthetic-openai-key"
_CONFIG_MODEL: Final = "responses-minted-reasoning-chaos"
_FOUNDRY_BASE: Final = "http://minted-reasoning-audit.services.ai.azure.com"
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
_CACHE_BUST: Final[Mapping[str, JsonValue]] = MappingProxyType({"cache": {"no-cache": True}})
Endpoint = Literal["responses", "chat", "messages"]
@dataclass(frozen=True, slots=True)
class _Call:
endpoint: Endpoint
stream: bool
marker: str
@dataclass(frozen=True, slots=True)
class _Served:
call: _Call
status: int
text: str
call_id: str
@dataclass(frozen=True, slots=True)
class _Models:
responses: str
chat: str
messages: str
def of(self, endpoint: Endpoint) -> str:
match endpoint:
case "responses":
return self.responses
case "chat":
return self.chat
case "messages":
return self.messages
def _register(scenario: Scenario, api_base: str) -> _Models:
return _Models(
responses=scenario.model(model=f"openai/{_GPT}", api_base=api_base, api_key=_OPENAI_KEY),
chat=scenario.model(model=f"openai/{_CODEX}", api_base=api_base, api_key=_OPENAI_KEY),
messages=scenario.model(model=f"anthropic/{cc.OPUS}", api_base=api_base, api_key=cc.ANTHROPIC_API_KEY),
)
def _path(endpoint: Endpoint) -> str:
match endpoint:
case "responses":
return "/v1/responses"
case "chat":
return "/v1/chat/completions"
case "messages":
return "/v1/messages"
def _body(models: _Models, call: _Call) -> dict[str, JsonValue]:
common: Final[dict[str, JsonValue]] = {
"model": models.of(call.endpoint),
"stream": call.stream,
"num_retries": 0,
**_CACHE_BUST,
}
match call.endpoint:
case "responses":
return {**common, "input": rv.agents_sdk_history(call.marker, rv.minted_item(call.marker))}
case "chat":
return {
**common,
"messages": [
{"role": "user", "content": "Pick a city."},
{
"role": "assistant",
"content": "Prague",
"reasoning_items": [
{"type": "reasoning", "encrypted_content": f"gAAAAA-stored-{call.marker}", "summary": []}
],
},
{"role": "user", "content": f"Name a landmark marker-{call.marker}"},
],
}
case "messages":
return {
**common,
"max_tokens": 64,
"messages": [
{"role": "user", "content": "Pick a city."},
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": rv.THOUGHT, "signature": rv.signature(call.marker)},
{"type": "text", "text": "Prague"},
],
},
{"role": "user", "content": f"Name a landmark marker-{call.marker}"},
],
}
def _calls(count: int, endpoints: tuple[Endpoint, ...]) -> tuple[_Call, ...]:
return tuple(
_Call(endpoint=endpoints[index % len(endpoints)], stream=index % 2 == 1, marker=uuid.uuid4().hex)
for index in range(count)
)
async def _send(client: httpx.AsyncClient, key: str, models: _Models, call: _Call) -> _Served:
async with client.stream(
"POST",
_path(call.endpoint),
json=_body(models, call),
headers={"Authorization": f"Bearer {key}", "anthropic-version": "2023-06-01"},
) as response:
raw: Final = await response.aread()
return _Served(call, response.status_code, raw.decode(), response.headers.get("x-litellm-call-id", ""))
async def _burst(
base_url: str, key: str, models: _Models, calls: tuple[_Call, ...], *, tolerate_transport_errors: bool = False
) -> tuple[_Served, ...]:
async with httpx.AsyncClient(base_url=base_url, timeout=60, trust_env=False) as client:
results: Final = await asyncio.gather(
*(_send(client, key, models, call) for call in calls), return_exceptions=tolerate_transport_errors
)
for result in results:
assert not isinstance(result, BaseException) or isinstance(result, httpx.TransportError), repr(result)
return tuple(result for result in results if isinstance(result, _Served))
def _frames(text: str) -> list[dict[str, JsonValue]]:
return [rv.JSON_OBJECT.validate_json(line[6:]) for line in text.splitlines() if line.startswith("data: {")]
def _response_id(served: _Served) -> str:
if not served.call.stream:
return str(rv.JSON_OBJECT.validate_json(served.text)["id"])
frames: Final = _frames(served.text)
match served.call.endpoint:
case "responses":
(completed,) = [frame for frame in frames if frame.get("type") == "response.completed"]
return str(rv.JSON_OBJECT.validate_python(completed["response"])["id"])
case "chat":
return str(frames[0]["id"])
case "messages":
(start,) = [frame for frame in frames if frame.get("type") == "message_start"]
return str(rv.JSON_OBJECT.validate_python(start["message"])["id"])
def _assert_answered_with_its_own_marker(served: _Served) -> None:
assert served.status == 200, served.text
assert set(rv.MARKER.findall(served.text)) == {served.call.marker}, served.text
def _assert_forwarded_without_a_minted_item(request: Request, marker: str) -> None:
body: Final = rv.JSON_OBJECT.validate_json(request.body)
path: Final = urlsplit(request.target).path
assert "no-cache" not in request.body.decode(), request.body
if path.endswith("/messages"):
(assistant,) = [turn for turn in rv.ITEMS.validate_python(body["messages"]) if turn["role"] == "assistant"]
assert assistant["content"] == [
{"type": "thinking", "thinking": rv.THOUGHT, "signature": rv.signature(marker)},
{"type": "text", "text": "Prague"},
], assistant
return
assert path.endswith("/responses"), request.target
items: Final = rv.reasoning_items(body)
if body["model"] == _CODEX:
assert items == [{"type": "reasoning", "encrypted_content": f"gAAAAA-stored-{marker}", "summary": []}], items
return
assert items == [], body["input"]
def _spend_rows(models: _Models, expected: int) -> list[dict[str, JsonValue]]:
return eventually(
lambda: read_rows(
'SELECT request_id, status FROM "LiteLLM_SpendLogs" WHERE model_group IN (%s, %s, %s)',
(models.responses, models.chat, models.messages),
),
lambda found: len(found) >= expected,
seconds=70,
)
def _assert_each_lands_once(
rows: list[dict[str, JsonValue]], failed: tuple[_Served, ...], served: tuple[_Served, ...]
) -> None:
by_status: Final = {str(row["request_id"]): str(row["status"]) for row in rows}
assert len(by_status) == len(rows) == len(failed) + len(served), rows
for item in failed:
assert by_status.get(item.call_id) == "failure", (item.call_id, rows)
for item in served:
(match,) = [request_id for request_id in by_status if rv.same_response(request_id, _response_id(item))]
assert by_status[match] == "success", rows
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
probe.bind(("127.0.0.1", 0))
return int(probe.getsockname()[1])
def _health_counts(gateway: Gateway, model: str) -> tuple[int, int]:
response: Final = gateway.request("GET", f"/health?model={model}", None)
assert response.status_code in (200, 503), response.text
health: Final = rv.JSON_OBJECT.validate_json(response.text)
return int(str(health["healthy_count"])), int(str(health["unhealthy_count"]))
def _marked(received: tuple[Request, ...]) -> dict[str, Request]:
marked: Final = {marker: request for request in received if (marker := rv.newest_marker(request.body.decode()))}
assert len(marked) == sum(1 for request in received if rv.newest_marker(request.body.decode())), received
return marked
@pytest.mark.timeout(180)
async def test_vendor_outage_fails_each_replay_cleanly_and_the_recovered_vendor_gets_them_without_minted_items(
gateway: Gateway,
) -> None:
port: Final = _free_port()
while_down: Final = _calls(15, ("responses", "chat", "messages"))
after: Final = _calls(15, ("responses", "chat", "messages"))
with gateway.scenario() as scenario:
models: Final = _register(scenario, f"http://127.0.0.1:{port}")
failed: Final = await _burst(str(gateway.client.base_url), gateway.key, models, while_down)
assert len(failed) == 15
for item in failed:
assert item.status == 500 and "Cannot connect to host" in item.text, (item.status, item.text)
assert "answer marker" not in item.text, item.text
assert item.call_id, item
assert _health_counts(gateway, models.responses) == (0, 1)
with wire_server(rv.ResponsesVendor().respond, port=port) as wire:
assert _health_counts(gateway, models.responses) == (1, 0)
wire.drain()
served: Final = await _burst(str(gateway.client.base_url), gateway.key, models, after)
assert len(served) == 15
for item in served:
_assert_answered_with_its_own_marker(item)
forwarded: Final = _marked(wire.drain())
assert set(forwarded) == {call.marker for call in after}, sorted(forwarded)
for marker, request in forwarded.items():
_assert_forwarded_without_a_minted_item(request, marker)
_assert_each_lands_once(_spend_rows(models, 30), failed, served)
async def test_slow_vendor_streams_are_each_forwarded_once_without_the_minted_item(gateway: Gateway) -> None:
calls: Final = tuple(_Call("responses", True, uuid.uuid4().hex) for _ in range(10))
with wire_server(rv.ResponsesVendor(pause_between_chunks=0.3).respond) as wire, gateway.scenario() as scenario:
models: Final = _register(scenario, wire.url)
served: Final = await _burst(str(gateway.client.base_url), gateway.key, models, calls)
assert len(served) == 10
for item in served:
_assert_answered_with_its_own_marker(item)
assert "response.completed" in item.text, item.text
received: Final = wire.drain()
assert len(received) == 10, [request.target for request in received]
forwarded: Final = _marked(received)
assert set(forwarded) == {call.marker for call in calls}
for marker, request in forwarded.items():
_assert_forwarded_without_a_minted_item(request, marker)
_assert_each_lands_once(_spend_rows(models, 10), (), served)
def _chaos_config(wire: Wire, tmp_path: Path) -> Path:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["model_list"] = [
{
"model_name": _CONFIG_MODEL,
"litellm_params": {"model": f"openai/{_GPT}", "api_base": wire.url, "api_key": _OPENAI_KEY},
}
]
path: Final = tmp_path / "responses-minted-reasoning-chaos.yaml"
path.write_text(yaml.safe_dump(config))
return path
def _open_upstream_connections(pid: int, upstream: str) -> int:
port: Final = urlsplit(upstream).port
return sum(
1
for connection in psutil.Process(pid).net_connections(kind="tcp")
if connection.status == psutil.CONN_ESTABLISHED and connection.raddr and connection.raddr.port == port
)
@pytest.mark.timeout(240)
async def test_worker_sigkill_mid_burst_leaves_the_sibling_dropping_the_minted_item(
gateway: Gateway, tmp_path: Path
) -> None:
calls: Final = tuple(_Call("responses", False, uuid.uuid4().hex) for _ in range(20))
release: Final = threading.Event()
held_markers: Final[SimpleQueue[str]] = SimpleQueue()
vendor: Final = rv.ResponsesVendor()
def held(request: Request) -> Reply:
if request.method == "GET":
return vendor.respond(request)
marker: Final = rv.newest_marker(request.body.decode())
assert marker is not None, request.body
held_markers.put(marker)
assert release.wait(timeout=60), "The burst was never released"
return vendor.respond(request)
with wire_server(held) as wire:
path: Final = _chaos_config(wire, tmp_path)
with owned_proxy_process(gateway, tmp_path, {}, config=path, workers=2) as owned:
candidate: Final = owned.gateway
models: Final = _Models(_CONFIG_MODEL, _CONFIG_MODEL, _CONFIG_MODEL)
workers: Final = eventually(
lambda: tuple(int(pid) for pid in _STARTED_WORKER.findall(owned.log.read_text())),
lambda pids: len(pids) == 2,
seconds=30,
)
burst: Final = asyncio.create_task(
_burst(str(candidate.client.base_url), candidate.key, models, calls, tolerate_transport_errors=True)
)
await asyncio.to_thread(eventually, held_markers.qsize, lambda size: size == 20, 60)
held_by: Final = MappingProxyType({pid: _open_upstream_connections(pid, wire.url) for pid in workers})
assert sum(held_by.values()) == 20, held_by
victim_pid, survivor_pid = sorted(workers, key=held_by.__getitem__)
victim: Final = psutil.Process(victim_pid)
victim.suspend()
victim.send_signal(signal.SIGKILL)
release.set()
served: Final = await burst
assert held_by[survivor_pid] >= 10, held_by
assert len(served) == held_by[survivor_pid], (held_by, len(served))
for item in served:
_assert_answered_with_its_own_marker(item)
follow_up: Final = _Call("responses", False, uuid.uuid4().hex)
(answered,) = await _burst(str(candidate.client.base_url), candidate.key, models, (follow_up,))
_assert_answered_with_its_own_marker(answered)
forwarded: Final = _marked(tuple(request for request in wire.drain() if request.method == "POST"))
assert set(forwarded) == {call.marker for call in (*calls, follow_up)}, sorted(forwarded)
for marker, request in forwarded.items():
_assert_forwarded_without_a_minted_item(request, marker)
@dataclass(frozen=True, slots=True)
class _Rig:
wire: Wire
proxy: OwnedProxy
cert: Path
key: Path
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
directory: Final = tmp_path_factory.mktemp("minted-reasoning-rig")
cert, key = write_self_signed_cert(directory)
copilot: Final = directory / "copilot"
chatgpt: Final = directory / "chatgpt"
copilot.mkdir()
chatgpt.mkdir()
with gateway_from_environment() as gateway, wire_server(rv.ResponsesVendor().respond) as wire:
(copilot / "api-key.json").write_text(
json.dumps(
{"token": "synthetic-copilot-token", "expires_at": time.time() + 3600, "endpoints": {"api": wire.url}}
)
)
(chatgpt / "auth.json").write_text(
json.dumps(
{
"access_token": "synthetic-chatgpt-token",
"account_id": "acct-synthetic",
"expires_at": time.time() + 3600,
}
)
)
overrides: Final = {
"GITHUB_COPILOT_TOKEN_DIR": str(copilot),
"CHATGPT_TOKEN_DIR": str(chatgpt),
"CHATGPT_API_BASE": wire.url,
"SSL_CERT_FILE": str(cert),
"HTTP_PROXY": wire.url,
"NO_PROXY": "127.0.0.1,localhost",
}
with owned_proxy_process(gateway, directory, overrides, workers=2) as owned:
yield _Rig(wire, owned, cert, key)
def _replay(gateway: Gateway, model: str, history: list[dict[str, JsonValue]], stream: bool) -> httpx.Response:
return gateway.request("POST", "/v1/responses", {"model": model, "input": history, "stream": stream, **_CACHE_BUST})
@dataclass(frozen=True, slots=True)
class _LoginDeployment:
label: str
model: str
api_key: str | None
_LOGIN_DEPLOYMENTS: Final = (
_LoginDeployment("github_copilot", f"github_copilot/{_CODEX}", None),
_LoginDeployment("chatgpt", f"chatgpt/{_CODEX}", None),
_LoginDeployment("azure_ai-foundry-host", "azure_ai/deepseek-v3", "synthetic-azure-key"),
)
@pytest.mark.timeout(240)
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
@pytest.mark.parametrize("deployment", _LOGIN_DEPLOYMENTS, ids=[deployment.label for deployment in _LOGIN_DEPLOYMENTS])
def test_login_backed_and_foundry_deployments_forward_the_minted_item_unchanged(
rig: _Rig, deployment: _LoginDeployment, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker, summary=[])
history: Final = rv.agents_sdk_history(marker, minted)
api_base: Final = _FOUNDRY_BASE if deployment.label.startswith("azure_ai") else rig.wire.url
rig.wire.drain()
with rig.proxy.gateway.scenario() as scenario:
parameters: Final[dict[str, JsonValue]] = {"model": deployment.model, "api_base": api_base}
model: Final = scenario.model(
**parameters, **({} if deployment.api_key is None else {"api_key": deployment.api_key})
)
response: Final = _replay(rig.proxy.gateway, model, history, stream)
received: Final = rig.wire.drain()
assert len(received) == 1, [(request.method, request.target) for request in received]
target: Final = urlsplit(received[0].target)
assert target.path.endswith("/responses"), received[0].target
if deployment.label.startswith("azure_ai"):
assert target.scheme == "http" and target.netloc == urlsplit(_FOUNDRY_BASE).netloc, received[0].target
items: Final = rv.reasoning_items(rv.JSON_OBJECT.validate_json(received[0].body))
assert items == [minted], items
assert response.status_code == 404, response.text
assert f"Item with id '{minted['id']}' not found" in response.text, response.text
@pytest.mark.timeout(240)
async def test_websocket_session_forwards_the_minted_item_as_before(rig: _Rig) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, minted)
frames: Final[SimpleQueue[tuple[str, str]]] = SimpleQueue()
async def vendor(connection: websockets.ServerConnection) -> None:
first: Final = await connection.recv()
frames.put((str(connection.request.path), str(first)))
tag: Final = uuid.uuid4().hex
response: Final[dict[str, JsonValue]] = {
"id": f"resp_{tag}",
"object": "response",
"created_at": 1,
"status": "completed",
"model": _GPT,
"output": [
{
"id": f"msg_{tag}",
"type": "message",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": rv.answer(marker), "annotations": []}],
}
],
"usage": rv.USAGE,
}
created: Final = {
"type": "response.created",
"sequence_number": 0,
"response": {**response, "status": "in_progress", "output": []},
}
await connection.send(json.dumps(created))
await connection.send(json.dumps({"type": "response.completed", "sequence_number": 1, "response": response}))
await connection.wait_closed()
gateway: Final = rig.proxy.gateway
async with websockets.serve(vendor, "127.0.0.1", 0, ssl=server_context(rig.cert, rig.key)) as server:
port: Final = server.sockets[0].getsockname()[1]
with gateway.scenario() as scenario:
model: Final = scenario.model(
model=f"openai/{_GPT}", api_base=f"https://127.0.0.1:{port}", api_key=_OPENAI_KEY
)
session_url: Final = (
f"{str(gateway.client.base_url).rstrip('/').replace('http://', 'ws://')}/v1/responses?model={model}"
)
async with websockets.connect(
session_url, additional_headers={"Authorization": f"Bearer {gateway.key}"}
) as session:
await session.send(json.dumps({"type": "response.create", "model": model, "input": history}))
received: Final[list[dict[str, JsonValue]]] = []
while not received or received[-1].get("type") != "response.completed":
received.append(rv.JSON_OBJECT.validate_json(str(await session.recv())))
assert [event["type"] for event in received] == ["response.created", "response.completed"], received
completed: Final = rv.JSON_OBJECT.validate_python(received[-1]["response"])
(message,) = rv.ITEMS.validate_python(completed["output"])
assert rv.ITEMS.validate_python(message["content"])[0]["text"] == rv.answer(marker), message
assert frames.qsize() == 1
path, first = frames.get_nowait()
assert path.startswith("/responses?") and f"model={_GPT}" in path, path
assert rv.JSON_OBJECT.validate_json(first)["input"] == history, first

View file

@ -0,0 +1,783 @@
import json
import threading
import time
import uuid
from collections import deque
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from types import EllipsisType, MappingProxyType
from typing import Final
from urllib.parse import urlsplit
import anthropic
import httpx
import openai
import pytest
from integration._support import claude_code as cc
from integration._support import responses_vendor as rv
from integration._support.client import Gateway, Scenario, eventually
from integration._support.database import read_rows
from integration._support.wire import Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
_GPT: Final = "gpt-5.6"
_CODEX: Final = "gpt-5.3-codex"
_CLAUDE: Final = cc.OPUS
_OPENAI_KEY: Final = "synthetic-openai-key"
_AZURE_KEY: Final = "synthetic-azure-key"
_CACHE_BUST: Final[Mapping[str, JsonValue]] = MappingProxyType({"cache": {"no-cache": True}})
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_ITEMS: Final = TypeAdapter(list[dict[str, JsonValue]])
@dataclass(frozen=True, slots=True)
class _Deployment:
label: str
model: str
api_key: str
target: str
extra: Mapping[str, JsonValue] = MappingProxyType({})
strips_message_status: bool = False
types_untyped_items_as_messages: bool = False
model_info: Mapping[str, JsonValue] | None = None
def register(self, scenario: Scenario, wire: Wire) -> str:
return scenario.model(
model=self.model, api_base=wire.url, api_key=self.api_key, model_info=self.model_info, **dict(self.extra)
)
def on_wire(self, items: Sequence[JsonValue]) -> list[JsonValue]:
return [self._as_sent(item) for item in items]
def _as_sent(self, item: JsonValue) -> JsonValue:
if not isinstance(item, dict):
return item
if self.strips_message_status and item.get("type") == "message":
return {key: value for key, value in item.items() if key != "status"}
if self.types_untyped_items_as_messages and "type" not in item:
return {**item, "type": "message"}
return item
_OPENAI: Final = _Deployment("openai", f"openai/{_GPT}", _OPENAI_KEY, "/responses")
_AZURE: Final = _Deployment(
"azure",
f"azure/{_GPT}",
_AZURE_KEY,
"/openai/v1/responses?api-version=preview",
MappingProxyType({"api_version": "preview"}),
strips_message_status=True,
)
_AZURE_AI_OPENAI_HOST: Final = _Deployment(
"azure_ai-rewritten-to-azure",
f"azure_ai/{_GPT}",
_AZURE_KEY,
"/openai/v1/responses?api-version=preview",
strips_message_status=True,
)
_DROPPING: Final = (_OPENAI, _AZURE, _AZURE_AI_OPENAI_HOST)
_KEEPING: Final = (
_Deployment("litellm_proxy", f"litellm_proxy/{_GPT}", "synthetic-proxy-key", "/responses"),
_Deployment("databricks", "databricks/gpt-5.6", "synthetic-databricks-key", "/responses"),
_Deployment("openrouter", f"openrouter/openai/{_GPT}", "synthetic-openrouter-key", "/responses"),
_Deployment("xai", "xai/grok-4.7", "synthetic-xai-key", "/responses"),
_Deployment("hosted_vllm", "hosted_vllm/qwen3", "synthetic-vllm-key", "/responses"),
_Deployment("fireworks_ai", "fireworks_ai/accounts/fireworks/models/kimi", "synthetic-fireworks-key", "/responses"),
_Deployment("volcengine", "volcengine/doubao", "synthetic-volcengine-key", "/responses"),
_Deployment("manus", "manus/manus-1", "synthetic-manus-key", "/responses"),
_Deployment("edenai", "edenai/openai/gpt-5.6", "synthetic-edenai-key", "/responses"),
_Deployment(
"perplexity",
"perplexity/sonar-pro",
"synthetic-perplexity-key",
"/v1/responses",
types_untyped_items_as_messages=True,
),
_Deployment("bedrock_mantle", "bedrock_mantle/openai.gpt-oss-120b", "synthetic-mantle-key", "/v1/responses"),
_Deployment(
"bedrock",
"bedrock/openai.gpt-oss-120b-1:0",
"synthetic-bedrock-key",
"/openai/v1/responses",
MappingProxyType({"aws_region_name": "us-east-1"}),
model_info=MappingProxyType({"supported_endpoints": ["/v1/responses"]}),
),
*(
_Deployment(slug, f"{slug}/{model}", f"synthetic-{slug}-key", "/responses")
for slug, model in (
("sail", "sail-1"),
("neosantara", "nusantara-base"),
("tensormesh", "qwen3"),
("parasail", "parasail-gpt-oss-120b"),
("empiriolabs", "empirio-1"),
("meta", "llama-4-maverick"),
("cortecs", "gpt-oss-120b"),
("pinstripes", "gpt-5.6"),
("prism", "gpt-oss-120b"),
)
),
)
def _base_url(gateway: Gateway) -> str:
return str(gateway.client.base_url).rstrip("/")
def _sdk(gateway: Gateway) -> openai.OpenAI:
return openai.OpenAI(
base_url=f"{_base_url(gateway)}/v1",
api_key=gateway.key,
max_retries=0,
http_client=httpx.Client(trust_env=False, timeout=60),
)
def _async_sdk(gateway: Gateway) -> openai.AsyncOpenAI:
return openai.AsyncOpenAI(
base_url=f"{_base_url(gateway)}/v1",
api_key=gateway.key,
max_retries=0,
http_client=httpx.AsyncClient(trust_env=False, timeout=60),
)
def _claude_sdk(gateway: Gateway) -> anthropic.Anthropic:
return anthropic.Anthropic(
base_url=_base_url(gateway),
api_key=gateway.key,
max_retries=0,
http_client=httpx.Client(trust_env=False, timeout=60),
)
def _create(
client: openai.OpenAI, model: str, history: Sequence[Mapping[str, JsonValue]], stream: bool
) -> dict[str, JsonValue]:
if not stream:
return client.responses.create(model=model, input=list(history), extra_body=dict(_CACHE_BUST)).model_dump()
events: Final = list(
client.responses.create(model=model, input=list(history), stream=True, extra_body=dict(_CACHE_BUST))
)
completed: Final = [event for event in events if event.type == "response.completed"]
assert len(completed) == 1, [event.type for event in events]
return completed[0].response.model_dump()
async def _create_async(
client: openai.AsyncOpenAI, model: str, history: Sequence[Mapping[str, JsonValue]], stream: bool
) -> dict[str, JsonValue]:
if not stream:
return (
await client.responses.create(model=model, input=list(history), extra_body=dict(_CACHE_BUST))
).model_dump()
events: Final = [
event
async for event in await client.responses.create(
model=model, input=list(history), stream=True, extra_body=dict(_CACHE_BUST)
)
]
completed: Final = [event for event in events if event.type == "response.completed"]
assert len(completed) == 1, [event.type for event in events]
return completed[0].response.model_dump()
def _raw(
gateway: Gateway, path: str, body: Mapping[str, JsonValue], *, key: str | None | EllipsisType = ...
) -> httpx.Response:
with httpx.Client(base_url=_base_url(gateway), trust_env=False, timeout=60) as client:
bearer: Final = gateway.key if key is ... else key
headers: Final = {} if bearer is None else {"Authorization": f"Bearer {bearer}"}
with client.stream("POST", path, json={**body, **_CACHE_BUST}, headers=headers) as response:
response.read()
return response
def _completed_payload(response: httpx.Response) -> dict[str, JsonValue]:
if not response.headers.get("content-type", "").startswith("text/event-stream"):
return _JSON_OBJECT.validate_json(response.content)
frames: Final = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: {")]
completed: Final = [frame for frame in frames if frame.get("type") == "response.completed"]
assert len(completed) == 1, [frame.get("type") for frame in frames]
return _JSON_OBJECT.validate_python(completed[0]["response"])
def _answer_text(payload: Mapping[str, JsonValue]) -> str:
messages: Final = [item for item in _ITEMS.validate_python(payload["output"]) if item.get("type") == "message"]
assert len(messages) == 1, payload
return str(_ITEMS.validate_python(messages[0]["content"])[0]["text"])
def _only_request(wire: Wire) -> tuple[Request, dict[str, JsonValue]]:
received: Final = wire.drain()
assert len(received) == 1, [(request.method, request.target) for request in received]
return received[0], _JSON_OBJECT.validate_json(received[0].body)
def _assert_spend_rows(model: str, response_ids: Sequence[str]) -> None:
rows: Final = eventually(
lambda: read_rows('SELECT request_id, status FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)),
lambda found: len(found) >= len(response_ids),
seconds=70,
)
logged: Final = {str(row["request_id"]): str(row["status"]) for row in rows}
assert len(logged) == len(rows) == len(response_ids), rows
for response_id in response_ids:
(match,) = [logged_id for logged_id in logged if rv.same_response(logged_id, response_id)]
assert logged[match] == "success", rows
def _assert_vendor_body(
body: Mapping[str, JsonValue], backend: str, forwarded: Sequence[JsonValue], stream: bool
) -> None:
assert body["model"] == backend, body
assert body["input"] == list(forwarded), body["input"]
assert body.get("stream", False) is stream, body
assert "cache" not in body and "no-cache" not in json.dumps(body), body
def _backend_of(deployment: _Deployment) -> str:
return deployment.model.split("/", 1)[1]
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
@pytest.mark.parametrize("deployment", _DROPPING, ids=[deployment.label for deployment in _DROPPING])
def test_agents_sdk_history_replays_to_openai_shaped_vendors_without_the_minted_item(
gateway: Gateway, deployment: _Deployment, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, minted)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = deployment.register(scenario, wire)
payload: Final = _create(_sdk(gateway), model, history, stream)
assert _answer_text(payload) == f"answer marker-{marker}", payload
request, body = _only_request(wire)
assert request.target == deployment.target, request.target
_assert_vendor_body(body, _backend_of(deployment), deployment.on_wire(rv.without(history, (minted,))), stream)
_assert_spend_rows(model, (str(payload["id"]),))
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
async def test_async_openai_sdk_replays_without_the_minted_item(gateway: Gateway, stream: bool) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, minted)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
payload: Final = await _create_async(_async_sdk(gateway), model, history, stream)
assert _answer_text(payload) == f"answer marker-{marker}", payload
request, body = _only_request(wire)
assert request.target == "/responses", request.target
_assert_vendor_body(body, _GPT, rv.without(history, (minted,)), stream)
@pytest.mark.parametrize("path", ["/v1/responses", "/responses", "/openai/v1/responses"])
def test_every_responses_route_alias_drops_the_minted_item(gateway: Gateway, path: str) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, minted)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
response: Final = _raw(gateway, path, {"model": model, "input": history})
assert response.status_code == 200, response.text
assert _answer_text(_completed_payload(response)) == f"answer marker-{marker}"
_, body = _only_request(wire)
_assert_vendor_body(body, _GPT, rv.without(history, (minted,)), False)
def test_identical_replays_each_land_one_spend_row(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
history: Final = rv.agents_sdk_history(marker, rv.minted_item(marker))
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
first: Final = _completed_payload(_raw(gateway, "/v1/responses", {"model": model, "input": history}))
second: Final = _completed_payload(_raw(gateway, "/v1/responses", {"model": model, "input": history}))
assert first["id"] != second["id"]
assert len(wire.drain()) == 2
_assert_spend_rows(model, (str(first["id"]), str(second["id"])))
def _decoded_thinking(item: Mapping[str, JsonValue]) -> list[dict[str, JsonValue]]:
encrypted: Final = item["encrypted_content"]
assert isinstance(encrypted, str), item
return _ITEMS.validate_json(encrypted)
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
def test_claude_turn_replays_to_openai_without_its_item_and_to_claude_with_its_thinking(
gateway: Gateway, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
claude: Final = scenario.model(model=f"anthropic/{_CLAUDE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY)
gpt: Final = _OPENAI.register(scenario, wire)
question: Final[dict[str, JsonValue]] = {"role": "user", "content": f"Pick a city marker-{marker}"}
produced: Final = _completed_payload(
_raw(gateway, "/v1/responses", {"model": claude, "input": [question], "stream": stream})
)
reasoning, message = _ITEMS.validate_python(produced["output"])
assert reasoning["type"] == "reasoning" and rv.MINTED_ID.match(str(reasoning["id"])), reasoning
assert "summary" not in reasoning, reasoning
(block,) = _decoded_thinking(reasoning)
assert (block["type"], block["signature"]) == ("thinking", rv.signature(marker)), block
assert message["type"] == "message", message
producing_request, producing_body = _only_request(wire)
assert producing_request.target == "/v1/messages"
follow_up: Final = uuid.uuid4().hex
history: Final[list[dict[str, JsonValue]]] = [
question,
reasoning,
message,
{"role": "user", "content": f"Name a landmark marker-{follow_up}"},
]
to_openai: Final = _raw(gateway, "/v1/responses", {"model": gpt, "input": history, "stream": stream})
assert to_openai.status_code == 200, to_openai.text
assert _answer_text(_completed_payload(to_openai)) == f"answer marker-{follow_up}"
openai_request, openai_body = _only_request(wire)
assert openai_request.target == "/responses"
_assert_vendor_body(openai_body, _GPT, [question, message, history[3]], stream)
to_claude: Final = _raw(gateway, "/v1/responses", {"model": claude, "input": history, "stream": stream})
assert to_claude.status_code == 200, to_claude.text
claude_request, claude_body = _only_request(wire)
assert claude_request.target == "/v1/messages"
messages: Final = _ITEMS.validate_python(claude_body["messages"])
assistant: Final = [turn for turn in messages if turn["role"] == "assistant"]
assert len(assistant) == 1, messages
assert assistant[0]["content"] == [
{"type": "thinking", "thinking": block["thinking"], "signature": rv.signature(marker)},
{"type": "text", "text": _answer_text(produced)},
], assistant[0]
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
@pytest.mark.parametrize("deployment", _KEEPING, ids=[deployment.label for deployment in _KEEPING])
def test_other_responses_providers_forward_the_minted_item_unchanged(
gateway: Gateway, deployment: _Deployment, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker, summary=[])
history: Final = rv.agents_sdk_history(marker, minted)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = deployment.register(scenario, wire)
response: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history, "stream": stream})
request, body = _only_request(wire)
assert urlsplit(request.target).path.endswith("/responses"), request.target
assert body["input"] == deployment.on_wire(history), body["input"]
assert response.status_code == 404, response.text
assert f"Item with id '{minted['id']}' not found" in response.text, response.text
@pytest.mark.parametrize(
("prefix", "forwarded_blocks"),
[
("litellm_proxy", ("thinking", "text", "tool_use")),
("openai", ("text", "tool_use")),
],
)
def test_chained_hop_through_this_proxy_to_claude(
gateway: Gateway, prefix: str, forwarded_blocks: tuple[str, ...]
) -> None:
marker: Final = uuid.uuid4().hex
minted: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, minted)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
claude: Final = scenario.model(model=f"anthropic/{_CLAUDE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY)
outer: Final = scenario.model(model=f"{prefix}/{claude}", api_base=_base_url(gateway), api_key=gateway.key)
response: Final = _raw(gateway, "/v1/responses", {"model": outer, "input": history})
assert response.status_code == 200, response.text
assert _answer_text(_completed_payload(response)) == f"answer marker-{marker}"
request, body = _only_request(wire)
assert request.target == "/v1/messages"
assistant: Final = [turn for turn in _ITEMS.validate_python(body["messages"]) if turn["role"] == "assistant"]
assert len(assistant) == 1, body["messages"]
blocks: Final = _ITEMS.validate_python(assistant[0]["content"])
assert tuple(str(block["type"]) for block in blocks) == forwarded_blocks, blocks
if "thinking" in forwarded_blocks:
assert blocks[0] == {"type": "thinking", "thinking": rv.THOUGHT, "signature": rv.signature(marker)}, blocks[
0
]
@dataclass(frozen=True, slots=True)
class _Hostile:
label: str
item: dict[str, JsonValue]
status: int
forwarded: bool
detail: str = ""
on_wire: Mapping[str, JsonValue] | None = None
def _hostile_cases() -> tuple[_Hostile, ...]:
marker: Final = "0" * 32
signed: Final = {"type": "thinking", "thinking": rv.THOUGHT, "signature": rv.signature(marker)}
unsigned: Final = {"type": "thinking", "thinking": rv.THOUGHT}
summary: Final[list[JsonValue]] = [{"type": "summary_text", "text": "thought about it"}]
big_blob: Final = "x" * 5000
big_blocks: Final = json.dumps([signed] * 60)
assert len(big_blocks) > 5000
return (
_Hostile(
"uppercase-uuid4-id",
{"type": "reasoning", "id": f"rs_{str(uuid.uuid4()).upper()}", "summary": []},
404,
True,
"Item with id",
),
_Hostile(
"minted-id-with-summary", {"type": "reasoning", "id": f"rs_{uuid.uuid4()}", "summary": summary}, 200, False
),
_Hostile(
"idless-opaque-blob", {"type": "reasoning", "encrypted_content": "gAAAAA-opaque", "summary": []}, 200, True
),
_Hostile(
"idless-unverifiable-blocks",
{"type": "reasoning", "encrypted_content": json.dumps([unsigned]), "summary": []},
200,
True,
),
_Hostile(
"idless-mixed-blocks",
{
"type": "reasoning",
"encrypted_content": json.dumps([unsigned, {"type": "text", "text": "x"}, signed]),
"summary": [],
},
200,
False,
),
_Hostile("int-id", {"type": "reasoning", "id": 7, "summary": []}, 400, True, "input"),
_Hostile("list-id", {"type": "reasoning", "id": ["rs_x"], "summary": []}, 400, True, "input"),
_Hostile("empty-id", {"type": "reasoning", "id": "", "summary": summary}, 400, True, "empty string"),
_Hostile("int-encrypted-content", {"type": "reasoning", "encrypted_content": 7, "summary": []}, 200, True),
_Hostile(
"list-encrypted-content", {"type": "reasoning", "encrypted_content": [signed], "summary": []}, 200, True
),
_Hostile("empty-encrypted-content", {"type": "reasoning", "encrypted_content": "", "summary": []}, 200, True),
_Hostile("five-kb-blob", {"type": "reasoning", "encrypted_content": big_blob, "summary": []}, 200, True),
_Hostile(
"five-kb-signed-blocks", {"type": "reasoning", "encrypted_content": big_blocks, "summary": []}, 200, False
),
_Hostile(
"null-id-null-encrypted",
{"type": "reasoning", "id": None, "encrypted_content": None, "summary": []},
200,
True,
on_wire={"type": "reasoning", "id": None, "summary": []},
),
_Hostile(
"message-with-minted-looking-id",
{
"type": "message",
"id": f"rs_{uuid.uuid4()}",
"role": "assistant",
"content": [{"type": "output_text", "text": "x", "annotations": []}],
},
200,
True,
),
)
_HOSTILE: Final = _hostile_cases()
@pytest.mark.parametrize("case", _HOSTILE, ids=[case.label for case in _HOSTILE])
def test_hostile_reasoning_items_reach_the_vendor_or_are_dropped_as_classified(
gateway: Gateway, case: _Hostile
) -> None:
marker: Final = uuid.uuid4().hex
history: Final = rv.agents_sdk_history(marker, case.item)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
response: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history})
assert response.status_code == case.status, response.text
assert case.detail in response.text, response.text
received: Final = wire.drain()
if response.status_code >= 400 and not received:
return
assert len(received) == 1, [(request.method, request.target) for request in received]
body: Final = _JSON_OBJECT.validate_json(received[0].body)
expected: Final = (
[case.on_wire if item is case.item and case.on_wire is not None else item for item in history]
if case.forwarded
else rv.without(history, (case.item,))
)
assert body["input"] == expected, body["input"]
assert response.status_code == case.status
if case.status == 200:
assert _answer_text(_completed_payload(response)) == f"answer marker-{marker}"
unrelated: Final = _raw(gateway, "/v1/responses", {"model": model, "input": f"ping marker-{marker}"})
assert unrelated.status_code == 200, unrelated.text
def test_vendor_owned_reasoning_item_from_a_producing_turn_is_kept(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
question: Final[dict[str, JsonValue]] = {"role": "user", "content": f"Pick a city marker-{marker}"}
produced: Final = _completed_payload(_raw(gateway, "/v1/responses", {"model": model, "input": [question]}))
reasoning, message = _ITEMS.validate_python(produced["output"])
assert str(reasoning["id"]).startswith("rs_") and not rv.MINTED_ID.match(str(reasoning["id"])), reasoning
wire.drain()
follow_up: Final = uuid.uuid4().hex
history: Final[list[dict[str, JsonValue]]] = [
question,
reasoning,
message,
{"role": "user", "content": f"Name a landmark marker-{follow_up}"},
]
response: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history})
assert response.status_code == 200, response.text
_, body = _only_request(wire)
assert body["input"] == history, body["input"]
def test_two_minted_items_are_both_dropped_and_a_minted_only_history_goes_out_empty(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
first: Final = rv.minted_item(marker)
second: Final = rv.minted_item(marker)
history: Final = rv.agents_sdk_history(marker, first, second)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
response: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history})
assert response.status_code == 200, response.text
_, body = _only_request(wire)
assert body["input"] == rv.without(history, (first, second)), body["input"]
lonely: Final = _raw(gateway, "/v1/responses", {"model": model, "input": [rv.minted_item(marker)]})
assert lonely.status_code == 400, lonely.text
assert "previous_response_id" in lonely.text and "must be provided" in lonely.text, lonely.text
_, lonely_body = _only_request(wire)
assert lonely_body["input"] == [], lonely_body
def test_a_megabyte_of_minted_thinking_is_dropped_while_the_proxy_stays_responsive(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
block: Final = {"type": "thinking", "thinking": "t" * 4000, "signature": rv.signature(marker)}
encrypted: Final = json.dumps([block] * 256)
assert len(encrypted) > 1_000_000
minted: Final[dict[str, JsonValue]] = {
"type": "reasoning",
"id": f"rs_{uuid.uuid4()}",
"encrypted_content": encrypted,
}
history: Final = rv.agents_sdk_history(marker, minted)
latencies: Final[deque[float]] = deque()
done: Final = threading.Event()
def probe() -> None:
with httpx.Client(base_url=_base_url(gateway), trust_env=False, timeout=30) as client:
while not done.is_set():
started: Final = time.monotonic()
assert client.get("/health/liveliness").status_code == 200
latencies.append(time.monotonic() - started)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
prober: Final = threading.Thread(target=probe)
prober.start()
started: Final = time.monotonic()
response: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history})
elapsed: Final = time.monotonic() - started
done.set()
prober.join(timeout=35)
assert response.status_code == 200, response.text[:500]
assert elapsed < 20, elapsed
assert latencies and max(latencies) < 5, (max(latencies), len(latencies))
_, body = _only_request(wire)
assert body["input"] == rv.without(history, (minted,))
def test_unauthenticated_replay_never_reaches_the_vendor_and_other_keys_keep_working(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
history: Final = rv.agents_sdk_history(marker, rv.minted_item(marker))
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = _OPENAI.register(scenario, wire)
other: Final = scenario.key(models=[model])
anonymous: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history}, key=None)
assert anonymous.status_code == 401, anonymous.text
forged: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history}, key="sk-not-a-key")
assert forged.status_code == 401, forged.text
assert wire.drain() == ()
failing: Final = _raw(
gateway,
"/v1/responses",
{
"model": model,
"input": rv.agents_sdk_history(marker, {"type": "reasoning", "id": "rs_" + "f" * 32, "summary": []}),
},
)
assert failing.status_code == 404, failing.text
assert "rs_" + "f" * 32 in failing.text, failing.text
healthy: Final = _raw(gateway, "/v1/responses", {"model": model, "input": history}, key=other)
assert healthy.status_code == 200, healthy.text
assert [request.target for request in wire.drain()] == ["/responses", "/responses"]
def _chat_history(marker: str, reasoning_items: Sequence[Mapping[str, JsonValue]]) -> list[dict[str, JsonValue]]:
return [
{"role": "user", "content": "Pick a city."},
{"role": "assistant", "content": "Prague", "reasoning_items": [dict(item) for item in reasoning_items]},
{"role": "user", "content": f"Name a landmark marker-{marker}"},
]
def _chat_create(client: openai.OpenAI, model: str, messages: Sequence[Mapping[str, JsonValue]], stream: bool) -> str:
if not stream:
completion: Final = client.chat.completions.create(
model=model, messages=list(messages), extra_body=dict(_CACHE_BUST)
)
return str(completion.choices[0].message.content)
chunks: Final = list(
client.chat.completions.create(model=model, messages=list(messages), stream=True, extra_body=dict(_CACHE_BUST))
)
return "".join(str(chunk.choices[0].delta.content or "") for chunk in chunks if chunk.choices)
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
def test_chat_bridge_replays_a_stored_reasoning_item_without_inventing_an_id(gateway: Gateway, stream: bool) -> None:
marker: Final = uuid.uuid4().hex
stored: Final[dict[str, JsonValue]] = {
"type": "reasoning",
"encrypted_content": f"gAAAAA-stored-{marker}",
"summary": [],
}
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"openai/{_CODEX}", api_base=wire.url, api_key=_OPENAI_KEY)
answer: Final = _chat_create(_sdk(gateway), model, _chat_history(marker, (stored,)), stream)
assert answer == f"answer marker-{marker}"
request, body = _only_request(wire)
assert request.target == "/responses"
assert body["model"] == _CODEX
assert rv.reasoning_items(body) == [stored], body["input"]
async def test_chat_bridge_async_client_replays_a_stored_reasoning_item_without_inventing_an_id(
gateway: Gateway,
) -> None:
marker: Final = uuid.uuid4().hex
stored: Final[dict[str, JsonValue]] = {
"type": "reasoning",
"encrypted_content": f"gAAAAA-stored-{marker}",
"summary": [],
}
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"openai/{_CODEX}", api_base=wire.url, api_key=_OPENAI_KEY)
completion: Final = await _async_sdk(gateway).chat.completions.create(
model=model, messages=_chat_history(marker, (stored,)), extra_body=dict(_CACHE_BUST)
)
assert completion.choices[0].message.content == f"answer marker-{marker}"
_, body = _only_request(wire)
assert rv.reasoning_items(body) == [stored], body["input"]
def test_chat_bridge_keeps_a_vendor_minted_id_and_sends_an_empty_item_bare(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"openai/{_CODEX}", api_base=wire.url, api_key=_OPENAI_KEY)
produced: Final = _sdk(gateway).chat.completions.create(
model=model,
messages=[{"role": "user", "content": f"Pick a city marker-{marker}"}],
extra_body=dict(_CACHE_BUST),
)
message: Final = produced.choices[0].message.model_dump()
(stored,) = _ITEMS.validate_python(message["reasoning_items"])
assert str(stored["id"]).startswith("rs_") and str(stored["encrypted_content"]).startswith("gAAAAA-vendor-"), (
stored
)
wire.drain()
follow_up: Final = uuid.uuid4().hex
answer: Final = _chat_create(_sdk(gateway), model, _chat_history(follow_up, (stored,)), False)
assert answer == f"answer marker-{follow_up}"
_, body = _only_request(wire)
assert rv.reasoning_items(body) == [
{"type": "reasoning", "id": stored["id"], "summary": [], "encrypted_content": stored["encrypted_content"]}
], body["input"]
bare: Final = uuid.uuid4().hex
assert (
_chat_create(_sdk(gateway), model, _chat_history(bare, ({"type": "reasoning", "summary": []},)), False)
== f"answer marker-{bare}"
)
_, bare_body = _only_request(wire)
assert rv.reasoning_items(bare_body) == [{"type": "reasoning", "summary": []}], bare_body["input"]
def test_chat_mode_model_takes_the_same_assistant_message_on_the_chat_wire(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
stored: Final[dict[str, JsonValue]] = {
"type": "reasoning",
"encrypted_content": f"gAAAAA-stored-{marker}",
"summary": [],
}
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"openai/{_GPT}", api_base=wire.url, api_key=_OPENAI_KEY)
assert _chat_create(_sdk(gateway), model, _chat_history(marker, (stored,)), False) == f"answer marker-{marker}"
request, body = _only_request(wire)
assert request.target == "/chat/completions"
messages: Final = _ITEMS.validate_python(body["messages"])
assert [turn["role"] for turn in messages] == ["user", "assistant", "user"], messages
assert messages[1]["content"] == "Prague", messages[1]
def _thinking_turns(marker: str) -> list[dict[str, JsonValue]]:
return [
{"role": "user", "content": "Pick a city."},
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": rv.THOUGHT, "signature": rv.signature(marker)},
{"type": "text", "text": "Prague"},
],
},
{"role": "user", "content": f"Name a landmark marker-{marker}"},
]
def _messages_create(
client: anthropic.Anthropic, model: str, messages: Sequence[Mapping[str, JsonValue]], stream: bool
) -> str:
if not stream:
reply: Final = client.messages.create(
model=model, max_tokens=64, messages=list(messages), extra_body=dict(_CACHE_BUST)
)
return "".join(block.text for block in reply.content if block.type == "text")
with client.messages.stream(
model=model, max_tokens=64, messages=list(messages), extra_body=dict(_CACHE_BUST)
) as stream_reply:
final: Final = stream_reply.get_final_message()
return "".join(block.text for block in final.content if block.type == "text")
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
def test_messages_endpoint_replays_claude_thinking_to_claude_unchanged(gateway: Gateway, stream: bool) -> None:
marker: Final = uuid.uuid4().hex
turns: Final = _thinking_turns(marker)
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{_CLAUDE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY)
assert _messages_create(_claude_sdk(gateway), model, turns, stream) == f"answer marker-{marker}"
request, body = _only_request(wire)
assert request.target == "/v1/messages"
assert body["messages"] == turns, body["messages"]
assert body.get("stream", False) is stream, body
@pytest.mark.parametrize("backend", [_CODEX, _GPT])
@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"])
def test_messages_endpoint_on_an_openai_model_sends_an_idless_reasoning_item(
gateway: Gateway, backend: str, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(rv.ResponsesVendor().respond) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"openai/{backend}", api_base=wire.url, api_key=_OPENAI_KEY)
assert (
_messages_create(_claude_sdk(gateway), model, _thinking_turns(marker), stream) == f"answer marker-{marker}"
)
request, body = _only_request(wire)
assert request.target == "/responses"
assert body.get("stream", False) is stream, body
(item,) = rv.reasoning_items(body)
assert "id" not in item and "summary" in item, item

View file

@ -3950,6 +3950,27 @@ def test_stored_reasoning_items_win_over_thinking_blocks():
assert reasoning_items[0]["id"] == "rs_real"
@pytest.mark.parametrize("missing_id", [None, ""])
def test_a_stored_reasoning_item_without_an_id_is_replayed_without_inventing_one(missing_id):
"""The Responses API rejects every id it did not mint, so no id beats a made-up one."""
handler = LiteLLMResponsesTransformationHandler()
stored_item = {"type": "reasoning", "summary": [], "encrypted_content": "enc_abc"}
messages = [
{
"role": "assistant",
"content": "Denver is sunny.",
"reasoning_items": [stored_item if missing_id is None else {**stored_item, "id": missing_id}],
},
]
input_items, _ = handler.convert_chat_completion_messages_to_responses_api(messages)
(reasoning_item,) = [item for item in input_items if item.get("type") == "reasoning"]
assert "id" not in reasoning_item
assert reasoning_item["encrypted_content"] == "enc_abc"
assert reasoning_item["summary"] == []
def test_convert_chat_completion_messages_to_responses_api_tool_result_with_tool_reference():
"""Tool-search tool_reference blocks have no Responses API equivalent: skip them, never stringify them."""
from litellm.completion_extras.litellm_responses_transformation.transformation import (

View file

@ -10,6 +10,7 @@ import litellm
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.azure.responses.transformation import AzureOpenAIResponsesAPIConfig
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.responses.litellm_completion_transformation.transformation import LiteLLMCompletionResponsesConfig
from litellm.types.llms.openai import (
ImageGenerationPartialImageEvent,
OutputTextDeltaEvent,
@ -18,6 +19,7 @@ from litellm.types.llms.openai import (
ResponsesAPIStreamEvents,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import Choices, Message, ModelResponse
_ARTIFACT_FIELD_PATTERN: Final = r'^(?!__.*__$)[^\p{Cc}\p{Cf}\p{Zl}\p{Zp}"\\./[\]]{1,200}$'
@ -941,6 +943,80 @@ class TestOpenAIResponsesAPIConfig:
assert norm["input"][1]["type"] == "custom_tool_call"
assert "namespace" not in norm["input"][1]
@staticmethod
def _claude_turn_bridged_to_responses_output() -> list:
claude_turn = ModelResponse(
id="chatcmpl-claude",
model="claude-sonnet-4-5",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(
role="assistant",
content="Paris is 22C and sunny.",
reasoning_content="Check Paris first.",
thinking_blocks=[
{"type": "thinking", "thinking": "Check Paris first.", "signature": "sig-paris"}
],
),
)
],
)
bridged = LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response(
request_input="Weather in Paris?", responses_api_request={}, chat_completion_response=claude_turn
)
return list(bridged.output)
@pytest.mark.parametrize("config", [OpenAIResponsesAPIConfig(), AzureOpenAIResponsesAPIConfig()])
def test_claude_reasoning_minted_by_the_bridge_is_dropped_before_the_history_reaches_openai(self, config):
saved_claude_turn = json.loads(
json.dumps([item.model_dump() for item in self._claude_turn_bridged_to_responses_output()])
)
bridge_reasoning = [item for item in saved_claude_turn if item["type"] == "reasoning"]
assert len(bridge_reasoning) == 1
openai_reasoning = {
"id": "rs_08d3a89dbb92277a006abf04f4266087d0b4eedacd7848f306",
"type": "reasoning",
"summary": [],
"encrypted_content": "gAAAAABo-opaque-openai-blob",
}
history = [
{"role": "user", "content": "Weather in Paris?"},
*saved_claude_turn,
openai_reasoning,
{"role": "user", "content": "And Berlin?"},
]
request = config.transform_responses_api_request(
model="gpt-5.6",
input=history,
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
outbound = request["input"]
assert len(outbound) == len(history) - 1
assert [item["id"] for item in outbound if item.get("type") == "reasoning"] == [openai_reasoning["id"]]
assert LiteLLMCompletionResponsesConfig._decode_thinking_blocks_from_input_item(bridge_reasoning[0]) == (
{"type": "thinking", "thinking": "Check Paris first.", "signature": "sig-paris"},
)
def test_bridge_minted_reasoning_is_dropped_when_handed_back_as_pydantic_output_items(self):
history = [*self._claude_turn_bridged_to_responses_output(), {"role": "user", "content": "And Berlin?"}]
request = self.config.transform_responses_api_request(
model="gpt-5.6",
input=history,
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert len(request["input"]) == len(history) - 1
assert all(item.get("type") != "reasoning" for item in request["input"])
class TestAzureResponsesAPIConfig:
def setup_method(self):

View file

@ -0,0 +1,62 @@
import json
from litellm.responses.litellm_completion_transformation.reasoning_items import (
decode_thinking_blocks,
encode_thinking_blocks,
is_litellm_minted_reasoning_item,
is_minted_reasoning_item_id,
mint_reasoning_item_id,
)
A_PROVIDER_OWNED_REASONING_ITEM_ID = "rs_08d3a89dbb92277a006abf04f4266087d0b4eedacd7848f306"
A_PROVIDER_OWNED_ENCRYPTED_BLOB = "gAAAAABo-opaque-provider-blob"
SIGNED_BLOCK = {"type": "thinking", "thinking": "Paris first.", "signature": "sig-paris"}
UNSIGNED_BLOCK = {"type": "thinking", "thinking": "never signed"}
REDACTED_BLOCK = {"type": "redacted_thinking", "data": "opaque"}
def test_minted_ids_are_recognized_and_provider_owned_ids_are_not():
minted = mint_reasoning_item_id()
assert is_minted_reasoning_item_id(minted)
assert not is_minted_reasoning_item_id(A_PROVIDER_OWNED_REASONING_ITEM_ID)
assert not is_minted_reasoning_item_id(minted.replace("-", ""))
assert not is_minted_reasoning_item_id(minted.removeprefix("rs_"))
assert not is_minted_reasoning_item_id(None)
def test_encoded_thinking_blocks_decode_back_to_the_verifiable_blocks_only():
encoded = encode_thinking_blocks([SIGNED_BLOCK, UNSIGNED_BLOCK, REDACTED_BLOCK])
assert encoded is not None
assert decode_thinking_blocks(encoded) == (SIGNED_BLOCK, REDACTED_BLOCK)
assert encode_thinking_blocks([UNSIGNED_BLOCK]) is None
assert decode_thinking_blocks(A_PROVIDER_OWNED_ENCRYPTED_BLOB) is None
assert decode_thinking_blocks(json.dumps(SIGNED_BLOCK)) is None
assert decode_thinking_blocks(json.dumps([{"type": "text", "text": "not thinking"}])) is None
def test_decoding_keeps_the_verifiable_blocks_of_a_mixed_array_and_skips_the_rest():
mixed = json.dumps([SIGNED_BLOCK, "a stray string", 7, None, UNSIGNED_BLOCK, {"type": "thinking"}, REDACTED_BLOCK])
assert decode_thinking_blocks(mixed) == (SIGNED_BLOCK, REDACTED_BLOCK)
assert decode_thinking_blocks(json.dumps(["only", "strings", 3])) is None
assert decode_thinking_blocks(json.dumps([UNSIGNED_BLOCK])) is None
def test_a_reasoning_item_is_litellm_minted_by_its_id_or_by_its_encoded_thinking_blocks():
assert is_litellm_minted_reasoning_item({"type": "reasoning", "id": mint_reasoning_item_id(), "summary": []})
assert is_litellm_minted_reasoning_item(
{
"type": "reasoning",
"id": A_PROVIDER_OWNED_REASONING_ITEM_ID,
"encrypted_content": encode_thinking_blocks([SIGNED_BLOCK]),
}
)
assert not is_litellm_minted_reasoning_item(
{
"type": "reasoning",
"id": A_PROVIDER_OWNED_REASONING_ITEM_ID,
"summary": [],
"encrypted_content": A_PROVIDER_OWNED_ENCRYPTED_BLOB,
}
)
assert not is_litellm_minted_reasoning_item({"type": "message", "id": mint_reasoning_item_id(), "role": "assistant"})
assert not is_litellm_minted_reasoning_item("a bare string input")