diff --git a/litellm/litellm_core_utils/prompt_templates/common_utils.py b/litellm/litellm_core_utils/prompt_templates/common_utils.py index 591cae6dfa4..3c1b3cf964b 100644 --- a/litellm/litellm_core_utils/prompt_templates/common_utils.py +++ b/litellm/litellm_core_utils/prompt_templates/common_utils.py @@ -1788,7 +1788,8 @@ _MarkedT: Final = TypeVar("_MarkedT", bound=Mapping[str, object]) def with_prompt_cache_breakpoint(target: _MarkedT, marker: object) -> _MarkedT: - if marker is None: + """A breakpoint the client already wrote on the block wins, TTL included.""" + if marker is None or target.get("prompt_cache_breakpoint") is not None: return target marked: Final = {**target, "prompt_cache_breakpoint": marker} return cast(_MarkedT, marked) # cast-ok: same block shape as the input plus the marker key diff --git a/litellm/llms/openai/responses/transformation.py b/litellm/llms/openai/responses/transformation.py index abd3d03b139..7c159c6a6a2 100644 --- a/litellm/llms/openai/responses/transformation.py +++ b/litellm/llms/openai/responses/transformation.py @@ -10,14 +10,17 @@ from typing_extensions import ReadOnly, TypedDict import litellm from litellm._logging import verbose_logger +from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook from litellm.litellm_core_utils.core_helpers import process_response_headers from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import ( safe_convert_created_field, ) from litellm.litellm_core_utils.prompt_templates.common_utils import ( + cache_control_as_prompt_cache_breakpoint, drop_non_python_regex_patterns, flatten_combinators_and_drop_non_python_regex_patterns, + with_prompt_cache_breakpoint, ) from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.litellm_core_utils.url_utils import encode_url_path_segment @@ -51,6 +54,75 @@ _MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS: Final = ("gpt-4", "gpt-3 _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 +_ROLES_READING_PROMPT_CACHE_BREAKPOINTS: Final = frozenset({"system", "developer", "user"}) +_PART_TYPES_READING_PROMPT_CACHE_BREAKPOINTS: Final = frozenset({"input_text", "input_image", "input_file"}) +_INPUT_ITEM: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) +_CONTENT_PARTS: Final[TypeAdapter[Sequence[Mapping[str, object]]]] = TypeAdapter(Sequence[Mapping[str, object]]) + + +class _DeploymentHost(BaseModel): + """`responses()` builds the params from raw kwargs, where callers may spell the host `base_url`.""" + + model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True) + + api_base: object = None + base_url: object = None + + +def _provider_name(provider: LlmProviders | str) -> str: + """Configs generated from a provider JSON file return the slug as a plain str.""" + return provider.value if isinstance(provider, LlmProviders) else provider + + +def _item_with_prompt_cache_breakpoints(item: object) -> object: + """Marks on assistant output and on tools are left for the strip that follows: OpenAI writes no cache there.""" + try: + fields: Final = _INPUT_ITEM.validate_python(item) + except ValidationError: + return item + if fields.get("type") == "function_call_output": + return _with_prompt_cache_breakpoints_under(fields, "output") + if fields.get("type", "message") == "message" and fields.get("role") in _ROLES_READING_PROMPT_CACHE_BREAKPOINTS: + return _with_prompt_cache_breakpoints_under(fields, "content") + return item + + +def _with_prompt_cache_breakpoints_under(item: Mapping[str, object], key: str) -> Mapping[str, object]: + value: Final = item.get(key) + item_mark: Final = item.get("cache_control") + unmarked_item: Final = {name: field for name, field in item.items() if name != "cache_control"} + if isinstance(value, str): + if item_mark is None: + return item + text_part: Final[Mapping[str, object]] = {"type": "input_text", "text": value} + return {**unmarked_item, key: [with_prompt_cache_breakpoint(text_part, PromptCacheBreakpoint(mode="explicit"))]} + try: + parts: Final = _CONTENT_PARTS.validate_python(value) + except ValidationError: + return item + swapped_parts: Final = tuple(_part_with_prompt_cache_breakpoint(part) for part in parts) + return {**unmarked_item, key: list(_with_breakpoint_on_last_reading_part(swapped_parts, item_mark))} + + +def _part_with_prompt_cache_breakpoint(part: Mapping[str, object]) -> Mapping[str, object]: + if part.get("type") not in _PART_TYPES_READING_PROMPT_CACHE_BREAKPOINTS: + return part + return cache_control_as_prompt_cache_breakpoint(part) + + +def _with_breakpoint_on_last_reading_part( + parts: Sequence[Mapping[str, object]], item_mark: object +) -> Sequence[Mapping[str, object]]: + if item_mark is None: + return parts + reading_indexes: Final = [ + index for index, part in enumerate(parts) if part.get("type") in _PART_TYPES_READING_PROMPT_CACHE_BREAKPOINTS + ] + if not reading_indexes: + return parts + last: Final = reading_indexes[-1] + marked_last: Final = with_prompt_cache_breakpoint(parts[last], PromptCacheBreakpoint(mode="explicit")) + return (*parts[:last], marked_last, *parts[last + 1 :]) class _ReasoningSupportEntry(LiteLLMBaseModel): @@ -293,19 +365,22 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): litellm_params: GenericLiteLLMParams, headers: dict, ) -> dict: - """Strip Anthropic-only `cache_control` markers before sending to OpenAI. - - OpenAI's Responses API rejects unknown fields on input content blocks - with HTTP 400 ("Unknown parameter: 'input[0].content[0].cache_control'"). - Chat Completions strips these in - `remove_cache_control_flag_from_messages_and_tools`; mirror that here. + """Anthropic-only `cache_control` marks become `prompt_cache_breakpoint` on the same block + where the deployment reads them (gpt-5.6 and newer on OpenAI's API, or a host opted in + through `prompt_cache_options`) and are stripped everywhere else, because OpenAI's + Responses API rejects unknown fields on input content blocks with HTTP 400 + ("Unknown parameter: 'input[0].content[0].cache_control'"). """ + request_params: Final = cast( # cast-ok: the base signature leaves the params bag untyped + "Mapping[str, object]", response_api_optional_request_params + ) replay_safe_input, sanitized_tools = self._prepared_input_and_tools( model=model, input=input, tools=response_api_optional_request_params.get("tools"), litellm_params=litellm_params, + prompt_cache_options=request_params.get("prompt_cache_options"), ) if sanitized_tools is not None: response_api_optional_request_params["tools"] = sanitized_tools @@ -321,10 +396,18 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): input: str | ResponseInputParam, tools: Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None, litellm_params: GenericLiteLLMParams, + prompt_cache_options: object = None, ) -> tuple[str | ResponseInputParam, Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None]: validated_input: Final = self._validate_input_param(self._drop_bridge_minted_reasoning_items(input)) + marked_input: Final = ( + self._input_with_prompt_cache_breakpoints(validated_input) + if self._reads_prompt_cache_breakpoints( + model=model, litellm_params=litellm_params, prompt_cache_options=prompt_cache_options + ) + else validated_input + ) stripped_input, stripped_tools = self.remove_cache_control_flag_from_input_and_tools( - model=model, input=validated_input, tools=tools + model=model, input=marked_input, tools=tools ) object_schema_tools: Final = self._tools_with_object_parameters(model=model, tools=stripped_tools) sanitized_tools: Final = self._sanitized_tool_schemas_for_openai( @@ -332,6 +415,24 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): ) return self._drop_foreign_tool_call_item_ids(stripped_input), sanitized_tools + def _reads_prompt_cache_breakpoints( + self, model: str, litellm_params: GenericLiteLLMParams, prompt_cache_options: object + ) -> bool: + host: Final = _DeploymentHost.model_validate(litellm_params) + return AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint( + model=model, + custom_llm_provider=_provider_name(self.custom_llm_provider), + api_base=host.api_base or host.base_url, + prompt_cache_options=prompt_cache_options, + ) + + @staticmethod + def _input_with_prompt_cache_breakpoints(input: str | ResponseInputParam) -> str | ResponseInputParam: + if not isinstance(input, list): + return input + marked_items: Final = [_item_with_prompt_cache_breakpoints(item) for item in input] + return cast("ResponseInputParam", marked_items) # cast-ok: items keep their shape, marks renamed in place + def _tools_with_object_parameters( self, model: str, tools: Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None ) -> Sequence[ALL_RESPONSES_API_TOOL_PARAMS] | None: @@ -951,11 +1052,15 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): compact_path: Final = parsed_url.path.rstrip("/") + "/compact" url: Final = str(parsed_url.copy_with(path=compact_path)) + request_params: Final = cast( # cast-ok: the base signature leaves the params bag untyped + "Mapping[str, object]", response_api_optional_request_params + ) replay_safe_input, sanitized_tools = self._prepared_input_and_tools( model=model, input=input, tools=response_api_optional_request_params.get("tools"), litellm_params=litellm_params, + prompt_cache_options=request_params.get("prompt_cache_options"), ) if sanitized_tools is not None: response_api_optional_request_params["tools"] = sanitized_tools diff --git a/tests/integration/providers/test_openai_responses_prompt_cache_breakpoint_wire.py b/tests/integration/providers/test_openai_responses_prompt_cache_breakpoint_wire.py new file mode 100644 index 00000000000..f64023d1af7 --- /dev/null +++ b/tests/integration/providers/test_openai_responses_prompt_cache_breakpoint_wire.py @@ -0,0 +1,229 @@ +import uuid +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from typing import Final + +import httpx +import openai +import pytest +from integration._support import prompt_cache_breakpoint as pcb +from integration._support import responses_vendor as rv +from integration._support.client import Gateway, gateway_from_environment +from integration._support.wire import Request, Wire, wire_server +from pydantic import JsonValue + +pytestmark: Final = pytest.mark.timeout(120) + +_EPHEMERAL: Final[Mapping[str, JsonValue]] = {"type": "ephemeral"} +_EPHEMERAL_1H: Final[Mapping[str, JsonValue]] = {"type": "ephemeral", "ttl": "1h"} +_OPTED_IN: Final[Mapping[str, JsonValue]] = {"prompt_cache_options": {"mode": "explicit"}} +_LOOKUP_TOOL: Final[Mapping[str, JsonValue]] = { + "type": "function", + "name": "lookup", + "parameters": {"type": "object", "properties": {}}, +} + + +@dataclass(frozen=True, slots=True) +class _Native: + gateway: Gateway + wire: Wire + reading: str + stripping: str + spend: pcb.SpendLogs + + +@pytest.fixture(scope="module") +def native() -> Iterator[_Native]: + with ( + wire_server(pcb.respond) as wire, + gateway_from_environment() as gateway, + gateway.scenario() as scenario, + pcb.spend_logs() as spend, + ): + api_base: Final = f"{wire.url}/v1" + yield _Native( + gateway, + wire, + scenario.model(model=pcb.MODEL, api_base=api_base, **_OPTED_IN), + scenario.model(model=pcb.MODEL, api_base=api_base), + spend, + ) + + +def _v1(gateway: Gateway) -> str: + return str(gateway.client.base_url).rstrip("/") + "/v1" + + +def _input_text(text: str) -> dict[str, JsonValue]: + return {"type": "input_text", "text": text} + + +def _marked( + block: Mapping[str, JsonValue], cache_control: Mapping[str, JsonValue] = _EPHEMERAL +) -> dict[str, JsonValue]: + return {**block, "cache_control": dict(cache_control)} + + +def _marked_history(marker: str) -> list[JsonValue]: + return [ + { + "type": "message", + "role": "developer", + "content": [_input_text("Answer briefly"), _marked(_input_text("Stable rules"), _EPHEMERAL_1H)], + }, + _marked({"role": "user", "content": pcb.prompt(marker)}), + ] + + +def _responses(native: _Native, model: str, input: Sequence[JsonValue], *, stream: bool = False) -> httpx.Response: + return native.gateway.request( + "POST", "/v1/responses", {"model": model, "input": list(input), "stream": stream, **pcb.NO_CACHE} + ) + + +def _answered(response: httpx.Response, marker: str) -> str: + assert response.status_code == 200, response.text + assert rv.answer(marker) in response.text, response.text + return response.headers["x-litellm-call-id"] + + +def _wire_body(request: Request, *, stream: bool = False) -> dict[str, JsonValue]: + body: Final = pcb.body_of(request) + assert body["model"] == "gpt-6.1-sol", body + assert (body.get("stream") is True) is stream, body + assert b"cache_control" not in request.body, request.body + return body + + +def _assert_marked_history_on_wire(body: Mapping[str, JsonValue], marker: str) -> None: + assert body["prompt_cache_options"] == {"mode": "explicit"}, body + assert body["input"] == [ + { + "type": "message", + "role": "developer", + "content": [ + _input_text("Answer briefly"), + pcb.marked(_input_text("Stable rules"), pcb.EXPLICIT), + ], + }, + {"role": "user", "content": [pcb.marked(_input_text(pcb.prompt(marker)), pcb.EXPLICIT)]}, + ], body + + +def test_openai_sdk_cache_control_reaches_the_wire_as_a_prompt_cache_breakpoint_on_the_same_block( + native: _Native, +) -> None: + marker: Final = uuid.uuid4().hex + with openai.OpenAI(api_key=native.gateway.key, base_url=_v1(native.gateway), max_retries=0) as client: + raw: Final = client.responses.with_raw_response.create( + model=native.reading, input=_marked_history(marker), extra_body=dict(pcb.NO_CACHE) + ) + response: Final = raw.parse() + assert pcb.answers(response.id, marker), response + assert response.output_text == rv.answer(marker), response + _assert_marked_history_on_wire(_wire_body(pcb.posted(native.wire, marker)), marker) + native.spend.landed(native.reading, raw.headers["x-litellm-call-id"], marker) + + +def test_stream_carries_the_marks_the_same_way(native: _Native) -> None: + marker: Final = uuid.uuid4().hex + call_id: Final = _answered(_responses(native, native.reading, _marked_history(marker), stream=True), marker) + _assert_marked_history_on_wire(_wire_body(pcb.posted(native.wire, marker), stream=True), marker) + native.spend.landed(native.reading, call_id, marker) + + +@pytest.mark.parametrize( + ("output", "expected"), + ( + ("found it", [pcb.marked(_input_text("found it"), pcb.EXPLICIT)]), + ( + [_input_text("first"), _input_text("last")], + [_input_text("first"), pcb.marked(_input_text("last"), pcb.EXPLICIT)], + ), + ), + ids=("string-output", "list-output"), +) +def test_cache_control_on_a_function_call_output_marks_its_last_part( + native: _Native, output: JsonValue, expected: JsonValue +) -> None: + marker: Final = uuid.uuid4().hex + history: Final[list[JsonValue]] = [ + {"role": "user", "content": pcb.prompt(marker)}, + {"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"}, + _marked({"type": "function_call_output", "call_id": "call_1", "output": output}), + ] + call_id: Final = _answered(_responses(native, native.reading, history), marker) + request: Final = pcb.posted(native.wire, marker) + _wire_body(request) + assert pcb.function_output(pcb.input_items(request), "call_1") == expected, ( + "OpenAI caches up to a breakpoint on a function_call_output part (prompt caching guide, read 2026-10-12)" + ) + native.spend.landed(native.reading, call_id, marker) + + +def test_cache_control_on_an_assistant_turn_and_on_a_tool_is_dropped(native: _Native) -> None: + marker: Final = uuid.uuid4().hex + history: Final[list[JsonValue]] = [ + {"role": "user", "content": "first question"}, + { + "type": "message", + "role": "assistant", + "content": [_marked({"type": "output_text", "text": "earlier answer"})], + }, + {"role": "user", "content": pcb.prompt(marker)}, + ] + response: Final = native.gateway.request( + "POST", + "/v1/responses", + {"model": native.reading, "input": history, "tools": [_marked(_LOOKUP_TOOL)], **pcb.NO_CACHE}, + ) + call_id: Final = _answered(response, marker) + request: Final = pcb.posted(native.wire, marker) + body: Final = _wire_body(request) + assert b"prompt_cache_breakpoint" not in request.body, request.body + assert pcb.content_of(pcb.input_items(request), "assistant") == [ + {"type": "output_text", "text": "earlier answer"} + ], "OpenAI accepts a breakpoint on output_text but writes no cache for it (live gpt-5.6, 2026-10-12)" + assert body["tools"] == [dict(_LOOKUP_TOOL)], body + native.spend.landed(native.reading, call_id, marker) + + +def test_cache_control_is_stripped_where_the_deployment_does_not_read_breakpoints(native: _Native) -> None: + marker: Final = uuid.uuid4().hex + call_id: Final = _answered(_responses(native, native.stripping, _marked_history(marker)), marker) + request: Final = pcb.posted(native.wire, marker) + body: Final = _wire_body(request) + assert "prompt_cache_options" not in body, body + assert b"prompt_cache_breakpoint" not in request.body, request.body + assert body["input"] == [ + { + "type": "message", + "role": "developer", + "content": [_input_text("Answer briefly"), _input_text("Stable rules")], + }, + {"role": "user", "content": pcb.prompt(marker)}, + ], body + native.spend.landed(native.stripping, call_id, marker) + + +def test_a_breakpoint_the_client_already_wrote_is_left_alone_beside_a_converted_mark(native: _Native) -> None: + marker: Final = uuid.uuid4().hex + history: Final[list[JsonValue]] = [ + { + "role": "user", + "content": [ + pcb.marked(_input_text(pcb.prompt(marker)), pcb.EXPLICIT_30M), + _marked(_input_text("and nothing else")), + ], + } + ] + call_id: Final = _answered(_responses(native, native.reading, history), marker) + request: Final = pcb.posted(native.wire, marker) + _wire_body(request) + (item,) = pcb.input_items(request) + assert item["content"] == [ + pcb.marked(_input_text(pcb.prompt(marker)), pcb.EXPLICIT_30M), + pcb.marked(_input_text("and nothing else"), pcb.EXPLICIT), + ], item + native.spend.landed(native.reading, call_id, marker) diff --git a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py index c040b161ba4..25d929ae025 100644 --- a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py +++ b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py @@ -30,6 +30,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import ( strip_encrypted_reasoning_from_messages, system_messages_first, update_messages_with_model_file_ids, + with_prompt_cache_breakpoint, ) _ARTIFACT_FIELD_PATTERN: Final = r'^(?!__.*__$)[^\p{Cc}\p{Cf}\p{Zl}\p{Zp}"\\./[\]]{1,200}$' @@ -2280,3 +2281,28 @@ def test_a_block_without_cache_control_is_returned_untouched() -> None: unmarked: Final = {"type": "text", "text": "rules"} assert cache_control_as_prompt_cache_breakpoint(unmarked) is unmarked + + +def test_a_breakpoint_the_client_wrote_survives_a_message_level_mark_ttl_included() -> None: + block: Final = { + "type": "input_text", + "text": "rules", + "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}, + } + + assert with_prompt_cache_breakpoint(block, {"mode": "explicit"}) is block + + +def test_a_breakpoint_the_client_wrote_survives_a_per_part_cache_control() -> None: + both: Final = { + "type": "input_text", + "text": "rules", + "cache_control": {"type": "ephemeral"}, + "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}, + } + + assert cache_control_as_prompt_cache_breakpoint(both) == { + "type": "input_text", + "text": "rules", + "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}, + } diff --git a/tests/unit/llms/openai/responses/test_openai_responses_transformation.py b/tests/unit/llms/openai/responses/test_openai_responses_transformation.py index 2121e04dcac..2a35031202a 100644 --- a/tests/unit/llms/openai/responses/test_openai_responses_transformation.py +++ b/tests/unit/llms/openai/responses/test_openai_responses_transformation.py @@ -1,5 +1,5 @@ import json -from collections.abc import Mapping +from collections.abc import Callable, Mapping from types import SimpleNamespace from typing import Final, cast from unittest.mock import AsyncMock, MagicMock, Mock, patch @@ -2303,6 +2303,314 @@ class TestToolSchemaRegexPatternWiring: assert result["tools"][0] is tool +_OPENAI_API: Final = GenericLiteLLMParams(api_base="https://api.openai.com/v1") +_OTHER_HOST: Final = GenericLiteLLMParams(api_base="https://gateway.example/v1") +_EXPLICIT_OPT_IN: Final = {"prompt_cache_options": {"mode": "explicit"}} +_EPHEMERAL: Final = {"type": "ephemeral"} +_EXPLICIT_BREAKPOINT: Final = {"mode": "explicit"} + + +def _marked_input(cache_control: dict[str, str] | None) -> list[dict[str, object]]: + mark: Final = {} if cache_control is None else {"cache_control": cache_control} + return [ + { + "role": "developer", + "content": [ + {"type": "input_text", "text": "Be concise."}, + {"type": "input_text", "text": "Be helpful.", **mark}, + ], + }, + {"type": "message", "role": "user", "content": "look", **mark}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "looking", **mark}]}, + {"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_1", "output": "found", **mark}, + {"type": "function_call", "call_id": "call_2", "name": "lookup", "arguments": "{}"}, + { + "type": "function_call_output", + "call_id": "call_2", + "output": [{"type": "input_text", "text": "first"}, {"type": "input_text", "text": "last"}], + **mark, + }, + ] + + +def _marked_tool(cache_control: dict[str, str] | None) -> dict[str, object]: + mark: Final = {} if cache_control is None else {"cache_control": cache_control} + return {"type": "function", "name": "lookup", "parameters": {"type": "object", "properties": {}}, **mark} + + +def _openai_like_responses_config() -> OpenAIResponsesAPIConfig: + from litellm.llms.openai_like.responses.transformation import OpenAILikeResponsesConfig + + return OpenAILikeResponsesConfig() + + +def _generated_sail_responses_config() -> OpenAIResponsesAPIConfig: + from litellm.llms.openai_like.dynamic_config import create_responses_config_class + from litellm.llms.openai_like.json_loader import JSONProviderRegistry + + sail = JSONProviderRegistry.get("sail") + assert sail is not None + return create_responses_config_class(sail)() + + +class TestCacheControlBecomesPromptCacheBreakpoint: + """Clients that speak Anthropic's dialect mark blocks with `cache_control`. On a deployment that + reads OpenAI's `prompt_cache_breakpoint` the mark moves onto the same block instead of being + stripped, so explicit-mode prompt caching keeps working through the proxy.""" + + @pytest.fixture(autouse=True) + def _no_openai_api_base_override(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + monkeypatch.delenv("OPENAI_API_BASE", raising=False) + monkeypatch.setattr(litellm, "api_base", None) + + def _transform( + self, + model: str, + litellm_params: GenericLiteLLMParams, + cache_control: dict[str, str] | None, + extra_params: dict[str, object] | None = None, + config: OpenAIResponsesAPIConfig | None = None, + ) -> dict[str, object]: + return (config or OpenAIResponsesAPIConfig()).transform_responses_api_request( + model=model, + input=_marked_input(cache_control), + response_api_optional_request_params={"tools": [_marked_tool(cache_control)], **(extra_params or {})}, + litellm_params=litellm_params, + headers={}, + ) + + @pytest.mark.parametrize( + ("litellm_params", "extra_params"), + [ + pytest.param(_OPENAI_API, {}, id="api.openai.com"), + pytest.param( + GenericLiteLLMParams(base_url="https://api.openai.com/v1"), {}, id="api.openai.com-as-base_url" + ), + pytest.param(GenericLiteLLMParams(), {}, id="default-host"), + pytest.param(_OTHER_HOST, _EXPLICIT_OPT_IN, id="other-host-opted-in"), + ], + ) + def test_marks_reach_gpt_5_6_as_explicit_breakpoints_on_the_same_blocks( + self, litellm_params: GenericLiteLLMParams, extra_params: dict[str, object] + ) -> None: + result = self._transform("gpt-5.6", litellm_params, {**_EPHEMERAL, "ttl": "1h"}, extra_params) + + assert result["input"] == [ + { + "role": "developer", + "content": [ + {"type": "input_text", "text": "Be concise."}, + {"type": "input_text", "text": "Be helpful.", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}, + ], + }, + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "look", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}], + }, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "looking"}]}, + {"type": "function_call", "call_id": "call_1", "name": "lookup", "arguments": "{}"}, + { + "type": "function_call_output", + "call_id": "call_1", + "output": [{"type": "input_text", "text": "found", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}], + }, + {"type": "function_call", "call_id": "call_2", "name": "lookup", "arguments": "{}"}, + { + "type": "function_call_output", + "call_id": "call_2", + "output": [ + {"type": "input_text", "text": "first"}, + {"type": "input_text", "text": "last", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}, + ], + }, + ] + assert result["tools"] == [_marked_tool(None)] + assert result.get("prompt_cache_options") == extra_params.get("prompt_cache_options") + + def test_per_part_marks_win_over_the_item_mark_and_keep_their_own_block(self) -> None: + result = OpenAIResponsesAPIConfig().transform_responses_api_request( + model="gpt-5.6", + input=[ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "a", "cache_control": _EPHEMERAL}, + {"type": "input_image", "image_url": "https://img.example/x.png", "detail": "auto"}, + {"type": "input_text", "text": "b"}, + ], + "cache_control": _EPHEMERAL, + } + ], + response_api_optional_request_params={}, + litellm_params=_OPENAI_API, + headers={}, + ) + + assert result["input"] == [ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "a", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}, + {"type": "input_image", "image_url": "https://img.example/x.png", "detail": "auto"}, + {"type": "input_text", "text": "b", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}, + ], + } + ] + + @pytest.mark.parametrize( + ("model", "litellm_params"), + [ + pytest.param("gpt-4o", _OPENAI_API, id="model-older-than-gpt-5.6"), + pytest.param("gpt-5.6", _OTHER_HOST, id="other-host-not-opted-in"), + pytest.param( + "gpt-5.6", GenericLiteLLMParams(base_url="https://gateway.example/v1"), id="other-host-as-base_url" + ), + ], + ) + def test_marks_are_still_stripped_where_the_deployment_does_not_read_breakpoints( + self, model: str, litellm_params: GenericLiteLLMParams + ) -> None: + marked = self._transform(model, litellm_params, _EPHEMERAL) + unmarked = self._transform(model, litellm_params, None) + + assert marked == unmarked + assert "prompt_cache_breakpoint" not in json.dumps(marked) + + def test_non_openai_subclass_still_strips_on_a_gpt_5_6_deployment(self) -> None: + from litellm.llms.hosted_vllm.responses.transformation import HostedVLLMResponsesAPIConfig + + config = HostedVLLMResponsesAPIConfig() + marked = self._transform("gpt-5.6", _OPENAI_API, _EPHEMERAL, _EXPLICIT_OPT_IN, config=config) + unmarked = self._transform("gpt-5.6", _OPENAI_API, None, _EXPLICIT_OPT_IN, config=config) + + assert marked == unmarked + assert "prompt_cache_breakpoint" not in json.dumps(marked["input"]) + + @pytest.mark.parametrize( + "make_config", [_openai_like_responses_config, _generated_sail_responses_config], ids=["openai_like", "sail"] + ) + def test_a_config_naming_its_provider_as_a_plain_str_still_strips_on_a_gpt_5_6_deployment( + self, make_config: Callable[[], OpenAIResponsesAPIConfig] + ) -> None: + config = make_config() + assert type(config.custom_llm_provider) is str + + marked = self._transform("gpt-5.6", _OPENAI_API, _EPHEMERAL, _EXPLICIT_OPT_IN, config=config) + unmarked = self._transform("gpt-5.6", _OPENAI_API, None, _EXPLICIT_OPT_IN, config=config) + + assert marked == unmarked + assert "prompt_cache_breakpoint" not in json.dumps(marked["input"]) + + def test_a_breakpoint_the_client_wrote_on_the_same_block_wins_over_the_mark_ttl_included(self) -> None: + kept = {"mode": "explicit", "ttl": "30m"} + result = OpenAIResponsesAPIConfig().transform_responses_api_request( + model="gpt-5.6", + input=[ + { + "role": "user", + "content": [ + { + "type": "input_text", + "text": "a", + "prompt_cache_breakpoint": kept, + "cache_control": _EPHEMERAL, + } + ], + }, + { + "role": "user", + "content": [{"type": "input_text", "text": "b", "prompt_cache_breakpoint": kept}], + "cache_control": _EPHEMERAL, + }, + ], + response_api_optional_request_params={}, + litellm_params=_OPENAI_API, + headers={}, + ) + + assert result["input"] == [ + {"role": "user", "content": [{"type": "input_text", "text": "a", "prompt_cache_breakpoint": kept}]}, + {"role": "user", "content": [{"type": "input_text", "text": "b", "prompt_cache_breakpoint": kept}]}, + ] + + def test_marks_on_parts_openai_does_not_cache_are_dropped_and_the_item_mark_skips_them(self) -> None: + audio = {"type": "input_audio", "input_audio": {"data": "UklGRg==", "format": "wav"}} + result = OpenAIResponsesAPIConfig().transform_responses_api_request( + model="gpt-5.6", + input=[ + {"role": "user", "content": [{**audio, "cache_control": _EPHEMERAL}], "cache_control": _EPHEMERAL}, + { + "role": "user", + "content": [{"type": "input_text", "text": "a"}, {**audio, "cache_control": _EPHEMERAL}], + "cache_control": _EPHEMERAL, + }, + ], + response_api_optional_request_params={}, + litellm_params=_OPENAI_API, + headers={}, + ) + + assert result["input"] == [ + {"role": "user", "content": [audio]}, + { + "role": "user", + "content": [ + {"type": "input_text", "text": "a", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}, + audio, + ], + }, + ] + + def test_items_and_content_of_a_shape_openai_will_reject_pass_through_to_the_strip(self) -> None: + """Validating the body is upstream's job. The swap leaves what it cannot read alone, so the + caller gets OpenAI's 400 for it instead of a failure inside the proxy.""" + result = OpenAIResponsesAPIConfig().transform_responses_api_request( + model="gpt-5.6", + input=[ + "a bare string where an item belongs", + { + "type": "function_call_output", + "call_id": "call_1", + "output": {"result": 1}, + "cache_control": _EPHEMERAL, + }, + {"role": "user", "content": "look", "cache_control": _EPHEMERAL}, + ], + response_api_optional_request_params={}, + litellm_params=_OPENAI_API, + headers={}, + ) + + assert result["input"] == [ + "a bare string where an item belongs", + {"type": "function_call_output", "call_id": "call_1", "output": {"result": 1}}, + { + "role": "user", + "content": [{"type": "input_text", "text": "look", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}], + }, + ] + + def test_compact_request_swaps_marks_the_same_way(self) -> None: + _, data = OpenAIResponsesAPIConfig().transform_compact_response_api_request( + model="gpt-5.6", + input=[{"role": "user", "content": "look", "cache_control": _EPHEMERAL}], + response_api_optional_request_params=dict(_EXPLICIT_OPT_IN), + litellm_params=_OTHER_HOST, + headers={}, + api_base="https://gateway.example/v1/responses", + ) + + assert data["input"] == [ + { + "role": "user", + "content": [{"type": "input_text", "text": "look", "prompt_cache_breakpoint": _EXPLICIT_BREAKPOINT}], + } + ] + + class TestReasoningFollowsModelSupport: """Responses API clients like Codex send `reasoning` on every request, and OpenAI 400s it on non-reasoning models like gpt-4o. drop_params must strip it there, the same way the