From 37df62ef7f2356c7ed8b0e78157d57c761a3ae8a Mon Sep 17 00:00:00 2001 From: basil-k-aji-dev <70605804+basil-k-aji-dev@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:23:22 +0530 Subject: [PATCH 1/3] test(responses): pin the chat bridge's multi-turn replay invariants A stateless Responses client has no previous_response_id to lean on, so each turn it appends the bridge's own output items back onto input and resends the whole transcript. Every existing test for this bridge translates one turn in isolation, which is the case that cannot see a transcript degrading: #42005 reports single-turn checks passing 12/12 while the loop lost its tool call on the fifth turn and returned the model's reasoning as visible text. These pin what has to hold at every depth: a replayed function_call reaches the provider as a native tool_calls entry, a reasoning item never becomes visible content and never arrives as a user turn, no tool result precedes the assistant message announcing it, the tools survive, and message count grows by a constant per turn. Seven output shapes across four transcript depths, with no upstream: these assert the request the bridge builds, so they are provider-independent and need no network or cassette. Refs #42005 --- ...test_responses_bridge_multi_turn_replay.py | 290 ++++++++++++++++++ 1 file changed, 290 insertions(+) create mode 100644 tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py 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 new file mode 100644 index 00000000000..11e559cf66a --- /dev/null +++ b/tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py @@ -0,0 +1,290 @@ +""" +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" From 3a09d212612802ce638044ef7af75dbce360d5a7 Mon Sep 17 00:00:00 2001 From: basil-k-aji-dev <70605804+basil-k-aji-dev@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:53:06 +0530 Subject: [PATCH 2/3] test(responses): pin the chat bridge's multi-turn replay invariants A stateless Responses client appends the bridge's own output items back onto input every turn and resends the whole transcript. Every existing test for this bridge translates one turn in isolation, which cannot see a transcript degrading: #42005 reports single-turn checks passing while the loop lost its tool call on the fifth turn and returned the model's reasoning as visible text. Nine output shapes across four transcript depths assert that a replayed function call still reaches the provider as a native tool call, reasoning never becomes visible content, the provider sees exactly the user turns the client sent, no tool result is orphaned, and the tools survive. Refs #42005 --- .../test_litellm_completion_responses.py | 169 ++++++++++ ...test_responses_bridge_multi_turn_replay.py | 290 ------------------ 2 files changed, 169 insertions(+), 290 deletions(-) delete mode 100644 tests/unit/responses/litellm_completion_transformation/test_responses_bridge_multi_turn_replay.py 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" From 1203de1b760d36b2855c874e510768e2286efbbe Mon Sep 17 00:00:00 2001 From: basil-k-aji-dev <70605804+basil-k-aji-dev@users.noreply.github.com> Date: Sun, 20 Sep 2026 18:00:29 +0530 Subject: [PATCH 3/3] style(responses): build the replay fixtures without mutation AGENTS.md asks new code to build values in one shot rather than seeding an empty collection and mutating it --- .../test_litellm_completion_responses.py | 57 +++++++++---------- 1 file changed, 28 insertions(+), 29 deletions(-) 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 cb93e4db829..8c331c86eac 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 @@ -5189,8 +5189,8 @@ def _replay_field(message, key, default=None): 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_tool_call_ids(message) -> tuple: + return tuple(_replay_field(call, "id") for call in _replay_field(message, "tool_calls") or ()) def _replay_visible_text(message): @@ -5200,15 +5200,15 @@ def _replay_visible_text(message): 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}"} +def _replay_reasoning_item(turn, summary=True, encrypted=False, content_blocks=False) -> dict: 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 + return { + "type": "reasoning", + "id": f"rs_{turn}", + "summary": [{"type": "summary_text", "text": text}] if summary else [], + **({"content": [{"type": "reasoning_text", "text": text}]} if content_blocks else {}), + **({"encrypted_content": "gAAAAABopaque=="} if encrypted else {}), + } def _replay_call_item(turn, suffix=""): @@ -5285,27 +5285,23 @@ _REPLAY_SHAPES: Final = { } -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 _replay_transcript(turns, shape) -> tuple: + opening = ({"role": "user", "content": "Collect vault tokens one at a time, starting at slot 1."},) + return opening + tuple(item for turn in range(1, turns + 1) for item in shape(turn)) -def _bridge_replay(transcript): +def _bridge_replay(transcript) -> dict: return LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( model="openai/kimi-k3", - input=transcript, + input=list(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 +def _expected_replay_call_ids(turns, shape_name) -> frozenset: + suffixes = ("", "b") if shape_name == "parallel_calls" else ("",) + return frozenset(f"call_{turn}{suffix}" for turn in range(1, turns + 1) for suffix in suffixes) @pytest.mark.parametrize("shape_name", sorted(_REPLAY_SHAPES)) @@ -5315,7 +5311,7 @@ class TestBridgeMultiTurnReplay: 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)} + sent = frozenset(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" @@ -5331,12 +5327,15 @@ class TestBridgeMultiTurnReplay: 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)) + messages = _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name]))["messages"] + for index, message in enumerate(messages): + if _replay_field(message, "role") != "tool": + continue + call_id = _replay_field(message, "tool_call_id") + announced = frozenset( + announced_id for earlier in messages[:index] for announced_id in _replay_tool_call_ids(earlier) + ) + assert call_id in announced, f"tool result {call_id} arrives before any message announces it" def test_the_tools_survive_the_whole_transcript(self, shape_name, turns): request = _bridge_replay(_replay_transcript(turns, _REPLAY_SHAPES[shape_name]))