diff --git a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py index 7b9de4644b4..cb93e4db829 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -5172,3 +5172,172 @@ async def test_bridge_rejects_untranslatable_tool_choice_with_a_400(stream: bool ) assert exc_info.value.status_code == 400 assert "tool_choice={'type': 'file_search'}" in str(exc_info.value) + + +_REPLAY_TOOL: Final = { + "type": "function", + "name": "vault_token", + "parameters": {"type": "object", "properties": {"slot": {"type": "integer"}}, "required": ["slot"]}, +} +_REASONING_MARKER: Final = "SECRET-CHAIN-OF-THOUGHT" +_REPLAY_TURNS: Final = [1, 2, 5, 8] + + +def _replay_field(message, key, default=None): + if isinstance(message, dict): + return message.get(key, default) + return getattr(message, key, default) + + +def _replay_tool_call_ids(message): + return [_replay_field(call, "id") for call in _replay_field(message, "tool_calls") or []] + + +def _replay_visible_text(message): + content = _replay_field(message, "content") + if content is None: + return "" + return content if isinstance(content, str) else json.dumps(content) + + +def _replay_reasoning_item(turn, summary=True, encrypted=False, content_blocks=False): + item: dict = {"type": "reasoning", "id": f"rs_{turn}"} + text = f"{_REASONING_MARKER}-{turn}" + item["summary"] = [{"type": "summary_text", "text": text}] if summary else [] + if content_blocks: + item["content"] = [{"type": "reasoning_text", "text": text}] + if encrypted: + item["encrypted_content"] = "gAAAAABopaque==" + return item + + +def _replay_call_item(turn, suffix=""): + return { + "type": "function_call", + "id": f"fc_{turn}{suffix}", + "call_id": f"call_{turn}{suffix}", + "name": "vault_token", + "arguments": json.dumps({"slot": turn}), + "status": "completed", + } + + +def _replay_call_output_item(turn, suffix=""): + return {"type": "function_call_output", "call_id": f"call_{turn}{suffix}", "output": f"tok-{turn}{suffix}"} + + +def _replay_assistant_text_item(turn): + return { + "type": "message", + "id": f"msg_{turn}", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": f"Fetching slot {turn}.", "annotations": []}], + } + + +_REPLAY_SHAPES: Final = { + "reasoning_summary": lambda t: [ + _replay_reasoning_item(t), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "reasoning_empty_summary": lambda t: [ + _replay_reasoning_item(t, summary=False), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "reasoning_encrypted_only": lambda t: [ + _replay_reasoning_item(t, summary=False, encrypted=True), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "reasoning_content_blocks": lambda t: [ + _replay_reasoning_item(t, summary=False, content_blocks=True), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "assistant_text_beside_the_call": lambda t: [ + _replay_assistant_text_item(t), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "reasoning_and_assistant_text": lambda t: [ + _replay_reasoning_item(t), + _replay_assistant_text_item(t), + _replay_call_item(t), + _replay_call_output_item(t), + ], + "no_reasoning_item": lambda t: [_replay_call_item(t), _replay_call_output_item(t)], + "user_speaks_again_each_turn": lambda t: [ + _replay_reasoning_item(t), + _replay_call_item(t), + _replay_call_output_item(t), + {"role": "user", "content": f"Now do slot {t + 1}."}, + ], + "parallel_calls": lambda t: [ + _replay_reasoning_item(t), + _replay_call_item(t), + _replay_call_item(t, "b"), + _replay_call_output_item(t), + _replay_call_output_item(t, "b"), + ], +} + + +def _replay_transcript(turns, shape): + transcript: list = [{"role": "user", "content": "Collect vault tokens one at a time, starting at slot 1."}] + for turn in range(1, turns + 1): + transcript.extend(shape(turn)) + return transcript + + +def _bridge_replay(transcript): + return LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( + model="openai/kimi-k3", + input=transcript, + responses_api_request={"tools": [_REPLAY_TOOL], "max_output_tokens": 500}, + custom_llm_provider="openai", + ) + + +def _expected_replay_call_ids(turns, shape_name): + ids = {f"call_{turn}" for turn in range(1, turns + 1)} + if shape_name == "parallel_calls": + ids |= {f"call_{turn}b" for turn in range(1, turns + 1)} + return ids + + +@pytest.mark.parametrize("shape_name", sorted(_REPLAY_SHAPES)) +@pytest.mark.parametrize("turns", _REPLAY_TURNS) +class TestBridgeMultiTurnReplay: + """A stateless client appends the bridge's own output back onto input every turn""" + + def test_every_replayed_call_reaches_the_provider_as_a_tool_call(self, shape_name, turns): + messages = _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name]))["messages"] + sent = {call_id for message in messages for call_id in _replay_tool_call_ids(message)} + missing = _expected_replay_call_ids(turns, shape_name) - sent + assert not missing, f"{sorted(missing)} were replayed but reach the provider as no tool call" + + def test_reasoning_never_becomes_visible_content(self, shape_name, turns): + for message in _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name]))["messages"]: + role = _replay_field(message, "role") + assert _REASONING_MARKER not in _replay_visible_text(message), f"reasoning is visible {role} content" + + def test_the_provider_sees_exactly_the_user_turns_the_client_sent(self, shape_name, turns): + transcript = _replay_transcript(turns, _REPLAY_SHAPES[shape_name]) + expected = sum(1 for item in transcript if item.get("role") == "user") + got = sum(1 for m in _bridge_replay(transcript)["messages"] if _replay_field(m, "role") == "user") + assert got == expected, f"{got} user messages reach the provider, but the client sent {expected}" + + def test_no_tool_result_is_orphaned(self, shape_name, turns): + announced: set = set() + for message in _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name]))["messages"]: + if _replay_field(message, "role") == "tool": + call_id = _replay_field(message, "tool_call_id") + assert call_id in announced, f"tool result {call_id} arrives before any message announces it" + announced.update(_replay_tool_call_ids(message)) + + def test_the_tools_survive_the_whole_transcript(self, shape_name, turns): + request = _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name])) + assert len(request.get("tools") or []) == 1, "the tool definition is gone from the request" diff --git a/tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py b/tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py deleted file mode 100644 index 11e559cf66a..00000000000 --- a/tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py +++ /dev/null @@ -1,290 +0,0 @@ -""" -Replay-invariant contract tests for the Responses -> chat completions bridge. - -A stateless Responses client has no ``previous_response_id`` to lean on, so -every turn it appends the bridge's own ``output`` items back onto ``input`` -verbatim and resends the whole transcript. That is the only option against a -provider whose Messages API is stateless, and it is the case single-turn -translation tests do not cover: each turn in isolation can translate perfectly -while the accumulated transcript still degrades. - -The invariants pinned here, from the discussion on #42005: - -* every ``function_call`` the client replays reaches the provider as a native - ``tool_calls`` entry on an assistant message, not as prose; -* a ``reasoning`` item never becomes visible ``content`` on any message, and - never arrives with ``role="user"`` -- replaying the model's own hidden - chain-of-thought as something the user said is worse than dropping it; -* every ``tool`` result is preceded by the assistant message carrying its - ``tool_call_id``, so no result is orphaned; -* none of the above decays as the transcript grows. - -The upstream is deliberately absent. These assert the request the bridge builds, -which makes them provider-independent and deterministic -- no Kimi, no network, -no recorded cassette to rot. -""" - -import json - -import pytest - -from litellm.responses.litellm_completion_transformation.transformation import ( - LiteLLMCompletionResponsesConfig, -) - -VAULT_TOOL = { - "type": "function", - "name": "vault_token", - "description": "Return an opaque random token for a slot. Never guess it.", - "parameters": { - "type": "object", - "properties": {"slot": {"type": "integer"}}, - "required": ["slot"], - }, -} - - -def _field(message, key, default=None): - if isinstance(message, dict): - return message.get(key, default) - return getattr(message, key, default) - - -def _tool_call_ids(message): - ids = [] - for call in _field(message, "tool_calls") or []: - ids.append(_field(call, "id")) - return ids - - -def _visible_text(message): - """Text a provider would read as prompt content, flattened to one string.""" - content = _field(message, "content") - if content is None: - return "" - if isinstance(content, str): - return content - return json.dumps(content) - - -def _reasoning_item(turn, *, summary=True, encrypted=False, content_blocks=False): - item = {"type": "reasoning", "id": f"rs_{turn}"} - secret = f"SECRET-CHAIN-OF-THOUGHT-{turn}" - if summary: - item["summary"] = [{"type": "summary_text", "text": secret}] - else: - item["summary"] = [] - if content_blocks: - item["content"] = [{"type": "reasoning_text", "text": secret}] - if encrypted: - item["encrypted_content"] = "gAAAAABopaque==" - return item - - -def _function_call_item(turn): - return { - "type": "function_call", - "id": f"fc_{turn}", - "call_id": f"call_{turn}", - "name": "vault_token", - "arguments": json.dumps({"slot": turn}), - "status": "completed", - } - - -def _function_call_output_item(turn): - return { - "type": "function_call_output", - "call_id": f"call_{turn}", - "output": f"tok-{turn}-opaque", - } - - -def _assistant_text_item(turn): - return { - "type": "message", - "id": f"msg_{turn}", - "role": "assistant", - "status": "completed", - "content": [{"type": "output_text", "text": f"Fetching slot {turn}.", "annotations": []}], - } - - -def _build_transcript(turns, shape): - """The `input` a stateless client holds after `turns` completed turns.""" - transcript = [ - { - "role": "user", - "content": "Collect vault tokens one at a time using vault_token, starting at slot 1.", - } - ] - for turn in range(1, turns + 1): - transcript.extend(shape(turn)) - return transcript - - -def _bridge(transcript): - return LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( - model="openai/kimi-k3", - input=transcript, - responses_api_request={"tools": [VAULT_TOOL], "max_output_tokens": 500}, - custom_llm_provider="openai", - ) - - -# Every shape a real `output` array takes across providers. Each one is replayed -# verbatim, because that is what the client is told to do. -SHAPES = { - "reasoning_summary": lambda t: [ - _reasoning_item(t), - _function_call_item(t), - _function_call_output_item(t), - ], - "reasoning_empty_summary": lambda t: [ - _reasoning_item(t, summary=False), - _function_call_item(t), - _function_call_output_item(t), - ], - "reasoning_encrypted_only": lambda t: [ - _reasoning_item(t, summary=False, encrypted=True), - _function_call_item(t), - _function_call_output_item(t), - ], - "reasoning_content_blocks": lambda t: [ - _reasoning_item(t, summary=False, content_blocks=True), - _function_call_item(t), - _function_call_output_item(t), - ], - "assistant_text_beside_the_call": lambda t: [ - _assistant_text_item(t), - _function_call_item(t), - _function_call_output_item(t), - ], - "reasoning_and_assistant_text": lambda t: [ - _reasoning_item(t), - _assistant_text_item(t), - _function_call_item(t), - _function_call_output_item(t), - ], - "no_reasoning_item": lambda t: [ - _function_call_item(t), - _function_call_output_item(t), - ], -} - -# Eight turns because the reported degradation appeared on the fifth; one turn -# is the control that says a failure at depth is about accumulation. -TURN_COUNTS = [1, 2, 5, 8] - - -@pytest.mark.parametrize("shape_name", sorted(SHAPES)) -@pytest.mark.parametrize("turns", TURN_COUNTS) -class TestResponsesBridgeMultiTurnReplay: - def test_every_replayed_call_reaches_the_provider_as_a_native_tool_call(self, shape_name, turns): - """A replayed function_call must not degrade into prose.""" - request = _bridge(_build_transcript(turns, SHAPES[shape_name])) - - sent = set() - for message in request["messages"]: - sent.update(_tool_call_ids(message)) - - expected = {f"call_{turn}" for turn in range(1, turns + 1)} - assert expected <= sent, ( - f"{sorted(expected - sent)} were replayed by the client but reach the provider as " - f"no tool call at all after {turns} turns" - ) - - def test_reasoning_never_becomes_visible_content(self, shape_name, turns): - """The model's hidden chain-of-thought must not re-enter as prompt text.""" - request = _bridge(_build_transcript(turns, SHAPES[shape_name])) - - for message in request["messages"]: - text = _visible_text(message) - assert "SECRET-CHAIN-OF-THOUGHT" not in text, ( - f"reasoning surfaced as visible {_field(message, 'role')!r} content " - f"after {turns} turns: {text[:120]!r}" - ) - - def test_reasoning_is_never_replayed_as_something_the_user_said(self, shape_name, turns): - """A reasoning item has no `role`, so a naive default makes it a user turn.""" - request = _bridge(_build_transcript(turns, SHAPES[shape_name])) - - user_messages = [m for m in request["messages"] if _field(m, "role") == "user"] - assert len(user_messages) == 1, ( - f"the transcript holds one real user turn, but {len(user_messages)} user messages " - f"reach the provider after {turns} turns" - ) - - def test_no_tool_result_is_orphaned(self, shape_name, turns): - """A tool result whose call never precedes it is rejected by most providers.""" - messages = _bridge(_build_transcript(turns, SHAPES[shape_name]))["messages"] - - announced = set() - for message in messages: - if _field(message, "role") == "tool": - call_id = _field(message, "tool_call_id") - assert call_id in announced, ( - f"tool result {call_id!r} arrives before any assistant message announces it, " - f"after {turns} turns" - ) - announced.update(_tool_call_ids(message)) - - def test_the_tools_survive_the_whole_transcript(self, shape_name, turns): - """A transcript that loses its tools explains a model answering in prose.""" - request = _bridge(_build_transcript(turns, SHAPES[shape_name])) - - tools = request.get("tools") or [] - assert len(tools) == 1, f"the tool definition is gone from the request after {turns} turns" - - -@pytest.mark.parametrize("shape_name", sorted(SHAPES)) -def test_replay_grows_by_exactly_one_exchange_per_turn(shape_name): - """ - Message count must grow linearly. - - Both real failure modes show up here before they show up anywhere else: a - turn that stops producing its assistant message, and a merge that folds two - turns into one and loses a call on the way. - """ - shape = SHAPES[shape_name] - counts = [len(_bridge(_build_transcript(turns, shape))["messages"]) for turns in (1, 2, 3, 4)] - deltas = {later - earlier for earlier, later in zip(counts, counts[1:])} - - assert len(deltas) == 1, ( - f"messages per replayed turn is not constant across the transcript: counts {counts}" - ) - - -@pytest.mark.parametrize("turns", TURN_COUNTS) -def test_two_calls_in_one_turn_are_both_replayed(turns): - """Parallel tool calls share a turn; neither may be dropped by the merge.""" - - def shape(turn): - return [ - _reasoning_item(turn), - _function_call_item(turn), - { - "type": "function_call", - "id": f"fc_{turn}b", - "call_id": f"call_{turn}b", - "name": "vault_token", - "arguments": json.dumps({"slot": turn + 100}), - "status": "completed", - }, - _function_call_output_item(turn), - { - "type": "function_call_output", - "call_id": f"call_{turn}b", - "output": f"tok-{turn}b-opaque", - }, - ] - - messages = _bridge(_build_transcript(turns, shape))["messages"] - - sent = set() - for message in messages: - sent.update(_tool_call_ids(message)) - - expected = {f"call_{turn}" for turn in range(1, turns + 1)} - expected |= {f"call_{turn}b" for turn in range(1, turns + 1)} - assert expected <= sent, f"{sorted(expected - sent)} were dropped after {turns} turns"