diff --git a/litellm/integrations/anthropic_cache_control_hook.py b/litellm/integrations/anthropic_cache_control_hook.py index f4f3b00dda0..ef2edbf1007 100644 --- a/litellm/integrations/anthropic_cache_control_hook.py +++ b/litellm/integrations/anthropic_cache_control_hook.py @@ -104,6 +104,13 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool: return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES +# Set by a caller whose message list is not the one that goes upstream -- today the +# Responses API layer, whose `instructions` only becomes a system message further down. +# Tells this hook to hand role-targeted points to the pass holding the final messages +# rather than spending them on a list that is still missing some of their targets. +CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_points" + + class AnthropicCacheControlHook(CustomPromptManagement): def get_chat_completion_prompt( self, @@ -128,6 +135,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): - non_default_params: dict - params with any global cache controls """ # Extract cache control injection points + carry_unmatched: Final = bool(non_default_params.pop(CARRY_UNMATCHED_MESSAGE_POINTS, False)) injection_points: Final[list[CacheControlInjectionPoint]] = non_default_params.pop( "cache_control_injection_points", [] ) @@ -161,12 +169,25 @@ class AnthropicCacheControlHook(CustomPromptManagement): non_default_params.get("prompt_cache_options"), ) ) + # A provisional message list defers every role-targeted point to the pass holding + # the final one: a role with no message here may have one there, and settling all + # of them in one pass is what lets config order decide the shared breakpoint + # budget. An ordinal names a different message once a later layer builds its own + # list, so it is placed here or not at all. + carried_message_points: Final[Sequence[CacheControlMessageInjectionPoint]] = ( + tuple(point for point in message_points if point.get("index") is None) if carry_unmatched else () + ) + applied_message_points: Final[Sequence[CacheControlMessageInjectionPoint]] = ( + tuple(point for point in message_points if point.get("index") is not None) + if carry_unmatched + else tuple(message_points) + ) reserved_blocks: Final = ( 1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0 ) breakpoints_before: Final = AnthropicCacheControlHook._count_request_cache_breakpoints(processed_messages) processed_messages = self._apply_message_injections( - points=message_points, + points=applied_message_points, messages=processed_messages, max_blocks=MAX_CACHE_CONTROL_BLOCKS - reserved_blocks, openai_dialect=openai_dialect, @@ -177,10 +198,15 @@ class AnthropicCacheControlHook(CustomPromptManagement): ): non_default_params.setdefault("prompt_cache_options", PromptCacheOptions(mode="explicit")) - # Pass through non-message injection points for provider-specific handling - if remaining_points: + # Points this pass did not place: non-message ones for the provider transform, and + # the deferred role-targeted ones. Deferring is what reaches the Responses API's + # `instructions`, which is only a system message once the bridge builds one. The + # judged stamp is what makes it safe: the next pass must not re-judge points + # against messages this pass already marked (see `_should_stand_down`). + carried_points: Final[Sequence[CacheControlInjectionPoint]] = (*remaining_points, *carried_message_points) + if carried_points: non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged( - remaining_points + carried_points ) return model, processed_messages, non_default_params @@ -218,7 +244,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): @staticmethod def _apply_message_injections( - points: list[CacheControlMessageInjectionPoint], + points: Sequence[CacheControlMessageInjectionPoint], messages: list[AllMessageValues], max_blocks: int, openai_dialect: bool = False, diff --git a/litellm/responses/main.py b/litellm/responses/main.py index d6ebc44ac52..8a682c701e0 100644 --- a/litellm/responses/main.py +++ b/litellm/responses/main.py @@ -1,6 +1,7 @@ import asyncio import contextvars -from collections.abc import Coroutine, Iterable, Mapping +from collections.abc import Coroutine, Generator, Iterable, Mapping +from contextlib import contextmanager from functools import partial from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast @@ -13,6 +14,7 @@ from litellm.completion_extras.litellm_responses_transformation.transformation i LiteLLMResponsesTransformationHandler, ) from litellm.constants import request_timeout +from litellm.integrations.anthropic_cache_control_hook import CARRY_UNMATCHED_MESSAGE_POINTS from litellm.litellm_core_utils.asyncify import run_async_function from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.prompt_templates.common_utils import ( @@ -390,6 +392,60 @@ async def aresponses_api_with_mcp( return response +def _bridges_to_chat_completions( + responses_api_provider_config: BaseResponsesAPIConfig | None, use_chat_completions_api: bool +) -> bool: + """Whether the request reaches its provider as a chat completion, not a Responses call.""" + return responses_api_provider_config is None or use_chat_completions_api is True + + +def _will_bridge_to_chat_completions( + model: str, custom_llm_provider: str | None, use_chat_completions_api: bool +) -> bool: + """``_bridges_to_chat_completions`` for callers running before the provider config is resolved. + + Resolving the config is a pure lookup, so this asks the same question the dispatch + asks rather than restating its condition. Both callers resolve the provider before + this runs, so the only way to be wrong is a prompt manager that moves the model + across the bridge boundary, which would leave the deferred points to a pass that + never comes. + """ + normalized_model: Final = _normalize_openai_chat_completions_responses_model(model) + if custom_llm_provider is None: + return True + return _bridges_to_chat_completions( + ProviderConfigManager.get_provider_responses_api_config( + model=normalized_model[0], provider=custom_llm_provider + ), + use_chat_completions_api or normalized_model[1], + ) + + +@contextmanager +def _prompt_management_sees_a_provisional_message_list( + kwargs: dict[str, Any], # mutable-ok: the signal is read and popped out of the caller's own kwargs + bridged: bool, +) -> Generator[None, None]: + """Tell the cache-control hook that this layer's messages are not the ones sent upstream. + + A Responses request keeps its system prompt in ``instructions``, which only becomes a + system message when the chat-completion bridge builds one, so a role-targeted point + is placed by the bridge's pass rather than this one. + + Only raised for a request that will be bridged. A provider serving Responses natively + gets no second pass, so this layer is the last one that can place anything and handing + a point forward there drops it. + """ + if not bridged: + yield + return + kwargs[CARRY_UNMATCHED_MESSAGE_POINTS] = True + try: + yield + finally: + kwargs.pop(CARRY_UNMATCHED_MESSAGE_POINTS, None) + + @client async def aresponses( input: str | ResponseInputParam, @@ -467,19 +523,25 @@ async def aresponses( client_input: list[AllMessageValues] = [{"role": "user", "content": input}] else: client_input = [item for item in input if isinstance(item, dict) and "role" in item] - ( - model, - merged_input, - merged_optional_params, - ) = await litellm_logging_obj.async_get_chat_completion_prompt( - model=model, - messages=client_input, - non_default_params=kwargs, - prompt_id=prompt_id, - prompt_variables=prompt_variables, - prompt_label=kwargs.get("prompt_label", None), - prompt_version=kwargs.get("prompt_version", None), - ) + with _prompt_management_sees_a_provisional_message_list( + kwargs, + bridged=_will_bridge_to_chat_completions( + model, custom_llm_provider, bool(kwargs.get("use_chat_completions_api")) + ), + ): + ( + model, + merged_input, + merged_optional_params, + ) = await litellm_logging_obj.async_get_chat_completion_prompt( + model=model, + messages=client_input, + non_default_params=kwargs, + prompt_id=prompt_id, + prompt_variables=prompt_variables, + prompt_label=kwargs.get("prompt_label", None), + prompt_version=kwargs.get("prompt_version", None), + ) input = cast( str | ResponseInputParam, ResponsesAPIRequestUtils.merge_prompt_management_input( @@ -566,6 +628,7 @@ def _apply_prompt_management_to_responses_call( litellm_logging_obj: LiteLLMLoggingObj | None, kwargs: dict[str, Any], local_vars: dict[str, object], + use_chat_completions_api: bool, ) -> tuple[str | ResponseInputParam, str, str | None]: async_merged: Final[Mapping[str, object] | None] = kwargs.pop("_async_prompt_merged_params", None) if async_merged is not None: @@ -585,19 +648,23 @@ def _apply_prompt_management_to_responses_call( if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and litellm_logging_obj.should_run_prompt_management_hooks( prompt_id=prompt_id, non_default_params=kwargs ): - ( - model, - merged_input, - merged_optional_params, - ) = litellm_logging_obj.get_chat_completion_prompt( - model=model, - messages=client_input, - non_default_params=kwargs, - prompt_id=prompt_id, - prompt_variables=prompt_variables, - prompt_label=kwargs.get("prompt_label", None), - prompt_version=kwargs.get("prompt_version", None), - ) + with _prompt_management_sees_a_provisional_message_list( + kwargs, + bridged=_will_bridge_to_chat_completions(model, custom_llm_provider, use_chat_completions_api), + ): + ( + model, + merged_input, + merged_optional_params, + ) = litellm_logging_obj.get_chat_completion_prompt( + model=model, + messages=client_input, + non_default_params=kwargs, + prompt_id=prompt_id, + prompt_variables=prompt_variables, + prompt_label=kwargs.get("prompt_label", None), + prompt_version=kwargs.get("prompt_version", None), + ) input = cast( str | ResponseInputParam, ResponsesAPIRequestUtils.merge_prompt_management_input( @@ -961,6 +1028,7 @@ def responses( litellm_logging_obj=litellm_logging_obj, kwargs=kwargs, local_vars=local_vars, + use_chat_completions_api=use_chat_completions_api, ) ######################################################### @@ -1063,7 +1131,7 @@ def responses( if _file_search_dispatch is not None: return _file_search_dispatch - if responses_api_provider_config is None or use_chat_completions_api is True: + if _bridges_to_chat_completions(responses_api_provider_config, use_chat_completions_api): return litellm_completion_transformation_handler.response_api_handler( model=model, input=input, diff --git a/tests/test_litellm/responses/test_responses_api_request_body.py b/tests/test_litellm/responses/test_responses_api_request_body.py index 14eb9ab6e12..5fd53fda01b 100644 --- a/tests/test_litellm/responses/test_responses_api_request_body.py +++ b/tests/test_litellm/responses/test_responses_api_request_body.py @@ -41,9 +41,7 @@ def _minimal_responses_api_payload(response_id: str, model: str) -> dict: "id": "msg_1", "status": "completed", "role": "assistant", - "content": [ - {"type": "output_text", "text": "Done.", "annotations": []} - ], + "content": [{"type": "output_text", "text": "Done.", "annotations": []}], } ], "parallel_tool_calls": True, @@ -83,9 +81,9 @@ class MockResponse: def _assert_request_body_matches(request_body: dict, expected_body: dict) -> None: for key, expected_value in expected_body.items(): assert key in request_body, f"Missing key in request body: {key}" - assert ( - request_body[key] == expected_value - ), f"Mismatch for key {key}: got {request_body[key]!r}, expected {expected_value!r}" + assert request_body[key] == expected_value, ( + f"Mismatch for key {key}: got {request_body[key]!r}, expected {expected_value!r}" + ) @pytest.mark.asyncio @@ -100,9 +98,7 @@ async def test_aresponses_context_management_and_shell_request_body_matches_expe "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post", new_callable=AsyncMock, ) as mock_post: - mock_post.return_value = MockResponse( - _minimal_responses_api_payload("resp_ctx_shell_test", "gpt-4o"), 200 - ) + mock_post.return_value = MockResponse(_minimal_responses_api_payload("resp_ctx_shell_test", "gpt-4o"), 200) await litellm.aresponses( model="openai/gpt-4o", @@ -426,7 +422,18 @@ async def test_aresponses_websocket_strips_responses_routing_prefix_from_openai_ _INJECTION_POINT_INPUT = [{"role": "system", "content": "You are terse."}, {"role": "user", "content": "hi"}] -_SYSTEM_INJECTION_POINT = [{"location": "message", "role": "system"}] +_SYSTEM_POINT = {"location": "message", "role": "system"} +_USER_POINT = {"location": "message", "role": "user"} +_SYSTEM_INJECTION_POINT = [_SYSTEM_POINT] +_ANTHROPIC_MESSAGES_PAYLOAD = { + "id": "msg_1", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5", + "content": [{"type": "text", "text": "Done."}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 5}, +} def _sent_body(mock_post) -> dict: @@ -598,3 +605,178 @@ def test_responses_custom_api_base_sends_no_openai_markers(): body = _sent_body(mock_post) assert body["input"] == _INJECTION_POINT_INPUT assert "prompt_cache_options" not in body + + +@pytest.mark.asyncio +async def test_injection_points_still_reach_a_native_responses_provider(): + """Providers that serve Responses natively never reach the chat-completions bridge, + so this layer is their only chance to inject and must keep doing so.""" + injected_client = AsyncHTTPHandler() + mock_post = AsyncMock(return_value=MockResponse(_minimal_responses_api_payload("resp_native", "gpt-5.6"), 200)) + injected_client.post = mock_post + + await litellm.aresponses( + model="openai/gpt-5.6", + api_key="fake-api-key", + input=copy.deepcopy(_INJECTION_POINT_INPUT), + cache_control_injection_points=copy.deepcopy(_SYSTEM_INJECTION_POINT), + client=injected_client, + ) + + body = _sent_body(mock_post) + assert body["input"][0]["content"][0]["prompt_cache_breakpoint"] == {"mode": "explicit"} + assert "cache_control_injection_points" not in body + + +async def _bridged_body(mock_post, *, points, input, instructions="You are a documentation assistant."): + injected_client = AsyncHTTPHandler() + injected_client.post = mock_post + + await litellm.aresponses( + model="anthropic/claude-sonnet-4-5", + api_key="fake-api-key", + instructions=instructions, + input=copy.deepcopy(input), + cache_control_injection_points=copy.deepcopy(points), + client=injected_client, + ) + return _sent_body(mock_post) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "content", + [ + pytest.param("hi", id="string-content"), + pytest.param([{"type": "input_text", "text": "hi there friend"}], id="list-content"), + ], +) +@pytest.mark.parametrize( + "points", + [ + pytest.param([_SYSTEM_POINT], id="system-only"), + pytest.param([_USER_POINT, _SYSTEM_POINT], id="mixed-user-and-system"), + ], +) +async def test_instructions_are_marked_when_the_bridge_builds_the_system_message(points, content): + """The system prompt lives in `instructions`, which is not a message until the bridge + builds one, so the point targeting it matches nothing at the Responses layer. + + Carrying it forward is what marks it at all. Carrying it *stamped* is what keeps a + second point that did match from stranding it: without the stamp the next pass reads + litellm's own marks as client breakpoints and stands the whole configuration down. + """ + mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200)) + body = await _bridged_body(mock_post, points=points, input=[{"role": "user", "content": content}]) + + assert body["system"][0]["cache_control"] == {"type": "ephemeral"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("instructions", [None, "You are a documentation assistant."]) +async def test_positional_points_address_the_input_item_the_caller_indexed(instructions): + """`index` counts the caller's `input` items, and the Responses layer is where that + list still is, so a matched positional point must be spent there and never re-resolved + against the bridge's list, where the system message shifts every ordinal by one.""" + mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200)) + body = await _bridged_body( + mock_post, + points=[{"location": "message", "index": 0}], + input=[{"role": "user", "content": [{"type": "input_text", "text": "hi there friend"}]}], + instructions=instructions, + ) + + assert body["messages"][0]["content"][0]["cache_control"] == {"type": "ephemeral"} + if instructions: + assert "cache_control" not in json.dumps(body["system"]) + + +@pytest.mark.asyncio +async def test_out_of_bounds_positional_points_are_not_revived_by_a_longer_list(): + """An ordinal addresses the list in front of the pass that reads it. + + Carrying one forward would re-resolve it against the bridge's longer list, where an + index that named nothing in the caller's `input` can land on a real message -- the + system prompt included. Positional points are resolved where they were written or not + at all. + """ + mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200)) + body = await _bridged_body( + mock_post, + points=[{"location": "message", "index": 1}], + input=[{"role": "user", "content": [{"type": "input_text", "text": "only item"}]}], + ) + + assert "cache_control" not in json.dumps(body["system"]) + assert "cache_control" not in json.dumps(body["messages"]) + + +def _four_user_turns() -> list: + return [ + item + for i in range(4) + for item in ( + {"role": "user", "content": [{"type": "input_text", "text": f"msg{i}"}]}, + {"role": "assistant", "content": [{"type": "output_text", "text": f"reply{i}", "annotations": []}]}, + ) + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "points,instructions,system_marked,marked_messages", + [ + pytest.param([_SYSTEM_POINT, _USER_POINT], "You are terse.", True, [0, 2, 4], id="earlier-point-wins"), + pytest.param([_USER_POINT, _SYSTEM_POINT], "You are terse.", False, [0, 2, 4, 6], id="reversed-order-reverses"), + pytest.param([_USER_POINT, _SYSTEM_POINT], None, False, [0, 2, 4, 6], id="target-never-built-costs-nothing"), + ], +) +async def test_config_order_decides_who_wins_the_shared_breakpoint_budget( + points, instructions, system_marked, marked_messages +): + """Injection points are honoured in config order, earlier ones winning scarce slots. + + A role-targeted point is placed a pass later than a positional one, so the four + breakpoints it competes for are shared across both passes. Every role point being + settled in the pass that holds the final list -- rather than the earlier pass holding + a slot for one it cannot place -- is what keeps that competition ordered in both + directions, and what stops a point whose target is never built from costing anything. + """ + mock_post = AsyncMock(return_value=MockResponse(_ANTHROPIC_MESSAGES_PAYLOAD, 200)) + body = await _bridged_body(mock_post, points=points, input=_four_user_turns(), instructions=instructions) + + assert ("cache_control" in json.dumps(body.get("system", []))) is system_marked + assert [i for i, msg in enumerate(body["messages"]) if "cache_control" in json.dumps(msg)] == marked_messages + + +@pytest.mark.asyncio +async def test_a_native_responses_provider_places_every_point_itself(): + """A provider serving Responses natively gets no second pass. + + This layer is the last one that can place anything, so handing a point forward here + drops it -- and an unmatchable point must not cost a matching one its slot either. + The request has to be known to be bridged before anything is deferred. + """ + input_items = _four_user_turns() + + async def _marked_indices(points): + injected_client = AsyncHTTPHandler() + mock_post = AsyncMock(return_value=MockResponse(_minimal_responses_api_payload("resp_native", "gpt-5.6"), 200)) + injected_client.post = mock_post + await litellm.aresponses( + model="openai/gpt-5.6", + api_key="fake-api-key", + input=copy.deepcopy(input_items), + cache_control_injection_points=copy.deepcopy(points), + client=injected_client, + ) + body = _sent_body(mock_post) + return [i for i, item in enumerate(body["input"]) if "prompt_cache_breakpoint" in json.dumps(item)] + + user_only = await _marked_indices([_USER_POINT]) + # The system point can never match here: nothing turns `instructions` into a message + # on the native path, so it must not cost the user point a slot. + with_unmatchable_system = await _marked_indices([_SYSTEM_POINT, _USER_POINT]) + + assert user_only == [0, 2, 4, 6] + assert with_unmatchable_system == user_only