From 48c266b3cf1a3da0c691644dcb1478d7cbe6ce79 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:33:38 -0700 Subject: [PATCH] test(responses): add black box tests for chat turn history replay --- .../integration/_support/responses_replay.py | 214 ++++++++++++++++ ...st_responses_bridge_history_replay_wire.py | 233 ++++++++++++++++++ ..._tool_loop_responses_history_replay_sdk.py | 37 +++ 3 files changed, 484 insertions(+) create mode 100644 tests/integration/_support/responses_replay.py create mode 100644 tests/integration/providers/test_responses_bridge_history_replay_wire.py create mode 100644 tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py diff --git a/tests/integration/_support/responses_replay.py b/tests/integration/_support/responses_replay.py new file mode 100644 index 00000000000..c4e73ffbf2b --- /dev/null +++ b/tests/integration/_support/responses_replay.py @@ -0,0 +1,214 @@ +import json +import uuid +from collections.abc import Callable, Sequence +from itertools import chain +from typing import Final + +from integration._support.client import list_value, object_value +from integration._support.wire import Reply, Request, Wire +from pydantic import JsonValue + +MODEL: Final = "gpt-6.1-sol" +PLAN: Final = "I am starting by choosing which city to check." +PICK: Final = "My pick is Tokyo, so I will check its weather." +PATCH_NOTE: Final = "I will create hello.txt containing hi." +ANSWER: Final = "It is raining in Tokyo." + +WEATHER_TOOL: Final[dict[str, JsonValue]] = { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get the weather for a city.", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + }, +} +PATCH_TOOL: Final[dict[str, JsonValue]] = { + "type": "custom", + "custom": {"name": "apply_patch", "description": "Apply a V4A patch to the workspace"}, +} + + +def _message(item_id: str, text: str, phase: str) -> dict[str, JsonValue]: + return { + "type": "message", + "id": item_id, + "status": "completed", + "role": "assistant", + "phase": phase, + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + + +def _reasoning(item_id: str, summary: str) -> dict[str, JsonValue]: + return { + "type": "reasoning", + "id": item_id, + "summary": [{"type": "summary_text", "text": summary}], + "encrypted_content": f"enc_{item_id}", + } + + +_WEATHER_CALL: Final[dict[str, JsonValue]] = { + "type": "function_call", + "id": "fc_weather", + "call_id": "call_weather", + "name": "get_weather", + "arguments": '{"city":"Tokyo"}', + "status": "completed", +} +WEATHER_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _message("msg_plan", PLAN, "commentary"), + _reasoning("rs_pick", "Tokyo is often rainy this time of year."), + _message("msg_pick", PICK, "commentary"), + _WEATHER_CALL, +) +TWO_THOUGHTS_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _reasoning("rs_scan", "Paris and Tokyo are the two candidates."), + _reasoning("rs_pick", "Tokyo is often rainy this time of year."), + _message("msg_pick", PICK, "commentary"), + _WEATHER_CALL, +) +CALL_ONLY_TURN: Final[tuple[dict[str, JsonValue], ...]] = (_WEATHER_CALL,) +PATCH_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _message("msg_patch", PATCH_NOTE, "commentary"), + { + "type": "custom_tool_call", + "id": "ctc_patch", + "call_id": "call_patch", + "name": "apply_patch", + "input": "*** Begin Patch\n*** Add File: hello.txt\n+hi\n*** End Patch", + "status": "completed", + }, +) +WEATHER_REPLAY: Final = ( + ("message", "assistant", "commentary", PLAN), + ("reasoning", "rs_pick"), + ("message", "assistant", "commentary", PICK), + ("function_call", "call_weather", "get_weather"), + ("function_call_output", "call_weather"), +) +TWO_THOUGHTS_REPLAY: Final = ( + ("reasoning", "rs_scan"), + ("reasoning", "rs_pick"), + ("message", "assistant", "commentary", PICK), + ("function_call", "call_weather", "get_weather"), + ("function_call_output", "call_weather"), +) + + +def _response(identity: str, output: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + return { + "id": f"resp_{identity}", + "object": "response", + "created_at": 1789788253, + "status": "completed", + "model": MODEL, + "output": list(output), + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + } + + +def _event(payload: dict[str, JsonValue]) -> bytes: + return f"event: {payload['type']}\ndata: {json.dumps(payload)}\n\n".encode() + + +def _opened(item: dict[str, JsonValue]) -> dict[str, JsonValue]: + match item["type"]: + case "message": + return {**item, "status": "in_progress", "content": []} + case "function_call": + return {**item, "status": "in_progress", "arguments": ""} + case _: + return item + + +def _deltas(index: int, item: dict[str, JsonValue]) -> tuple[dict[str, JsonValue], ...]: + match item["type"]: + case "message": + text: Final = object_value(list_value(item["content"])[0])["text"] + return ( + { + "type": "response.output_text.delta", + "item_id": item["id"], + "output_index": index, + "content_index": 0, + "delta": text, + }, + ) + case "function_call": + return ( + { + "type": "response.function_call_arguments.delta", + "item_id": item["id"], + "output_index": index, + "delta": item["arguments"], + }, + ) + case _: + return () + + +def _item_events(index: int, item: dict[str, JsonValue]) -> tuple[bytes, ...]: + return ( + _event({"type": "response.output_item.added", "output_index": index, "item": _opened(item)}), + *(_event(delta) for delta in _deltas(index, item)), + _event({"type": "response.output_item.done", "output_index": index, "item": item}), + ) + + +def _stream(identity: str, output: Sequence[dict[str, JsonValue]]) -> tuple[bytes, ...]: + response: Final = _response(identity, output) + return ( + _event({"type": "response.created", "response": {**response, "status": "in_progress", "output": []}}), + *chain.from_iterable(_item_events(index, item) for index, item in enumerate(output)), + _event({"type": "response.completed", "response": response}), + ) + + +def _answered(body: dict[str, JsonValue]) -> bool: + return any( + object_value(item).get("type") in {"function_call_output", "custom_tool_call_output"} + for item in list_value(body["input"]) + ) + + +def responses_upstream(turn_one: Sequence[dict[str, JsonValue]]) -> Callable[[Request], Reply]: + def respond(request: Request) -> Reply: + if request.method == "GET": + return Reply(body=b'{"object":"list","data":[]}') + assert request.target == "/responses", request.target + body: Final = object_value(json.loads(request.body)) + identity: Final = uuid.uuid4().hex + output: Final = (_message("msg_answer", ANSWER, "final_answer"),) if _answered(body) else turn_one + if body.get("stream"): + return Reply(content_type="text/event-stream", chunks=_stream(identity, output)) + return Reply(body=json.dumps(_response(identity, output)).encode()) + + return respond + + +def _text(content: JsonValue) -> JsonValue: + if isinstance(content, str): + return content + return "".join(str(object_value(part).get("text")) for part in list_value(content)) + + +def _shape(item: JsonValue) -> tuple[JsonValue, ...]: + fields: Final = object_value(item) + match fields.get("type"): + case "message": + return ("message", fields.get("role"), fields.get("phase"), _text(fields.get("content"))) + case "reasoning": + return ("reasoning", fields.get("id")) + case "function_call" | "custom_tool_call" as kind: + return (kind, fields.get("call_id"), fields.get("name")) + case kind: + return (kind, fields.get("call_id")) + + +def posted_inputs(wire: Wire) -> tuple[tuple[tuple[JsonValue, ...], ...], ...]: + return tuple( + tuple(_shape(item) for item in list_value(object_value(json.loads(request.body))["input"])) + for request in wire.drain() + if request.method == "POST" + ) diff --git a/tests/integration/providers/test_responses_bridge_history_replay_wire.py b/tests/integration/providers/test_responses_bridge_history_replay_wire.py new file mode 100644 index 00000000000..f85dca67be4 --- /dev/null +++ b/tests/integration/providers/test_responses_bridge_history_replay_wire.py @@ -0,0 +1,233 @@ +import json +import uuid +from collections.abc import Sequence +from functools import reduce +from itertools import chain +from typing import Final + +import httpx +import pytest +from integration._support.client import Gateway, list_value, object_value +from integration._support.responses_replay import ( + CALL_ONLY_TURN, + MODEL, + PATCH_NOTE, + PATCH_TOOL, + PATCH_TURN, + PICK, + PLAN, + TWO_THOUGHTS_REPLAY, + TWO_THOUGHTS_TURN, + WEATHER_REPLAY, + WEATHER_TOOL, + WEATHER_TURN, + posted_inputs, + responses_upstream, +) +from integration._support.wire import wire_server +from pydantic import JsonValue + +_API_KEY: Final = "synthetic-openai-key" + + +def _chunks(response: httpx.Response) -> list[dict[str, JsonValue]]: + return [ + object_value(json.loads(line[6:])) + for line in response.iter_lines() + if line.startswith("data: ") and line != "data: [DONE]" + ] + + +def _tool_call(index: JsonValue, calls: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + parts: Final = [call for call in calls if call["index"] == index] + functions: Final = [object_value(part["function"]) for part in parts] + return { + "id": next(part["id"] for part in parts if part.get("id")), + "type": "function", + "function": { + "name": next(function["name"] for function in functions if function.get("name")), + "arguments": "".join(str(function.get("arguments") or "") for function in functions), + }, + } + + +def _deltas(chunks: Sequence[dict[str, JsonValue]]) -> list[dict[str, JsonValue]]: + return [ + object_value(object_value(choice)["delta"]) + for choice in chain.from_iterable(list_value(chunk["choices"]) for chunk in chunks) + ] + + +def _assembled(chunks: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + deltas: Final = _deltas(chunks) + calls: Final = [object_value(call) for call in chain.from_iterable(d.get("tool_calls") or [] for d in deltas)] + fields: Final = reduce( + lambda merged, delta: {**merged, **object_value(delta.get("provider_specific_fields") or {})}, deltas, {} + ) + return { + "role": "assistant", + "content": "".join(str(delta["content"]) for delta in deltas if delta.get("content")) or None, + "tool_calls": [_tool_call(index, calls) for index in sorted({int(str(call["index"])) for call in calls})], + "reasoning_items": list(chain.from_iterable(list_value(d.get("reasoning_items") or []) for d in deltas)), + "provider_specific_fields": fields, + } + + +def _first_turn( + gateway: Gateway, model: str, prompt: str, tools: Sequence[dict[str, JsonValue]], *, stream: bool +) -> dict[str, JsonValue]: + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": list(tools), + "stream": stream, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == 200, response.text + if stream: + return _assembled(_chunks(response)) + return object_value(object_value(list_value(response.json()["choices"])[0])["message"]) + + +def _next_turn( + gateway: Gateway, + model: str, + prompt: str, + tools: Sequence[dict[str, JsonValue]], + message: dict[str, JsonValue], + **extra: JsonValue, +) -> httpx.Response: + call_id: Final = object_value(list_value(message["tool_calls"])[0])["id"] + return gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [ + {"role": "user", "content": prompt}, + message, + {"role": "tool", "tool_call_id": call_id, "content": "Rain, 18C"}, + ], + "tools": list(tools), + "cache": {"no-cache": True}, + **extra, + }, + ) + + +def _with_phase(message: dict[str, JsonValue], phase: str) -> dict[str, JsonValue]: + fields: Final = object_value(message.get("provider_specific_fields") or {}) + return {**message, "provider_specific_fields": {**fields, "phase": phase}} + + +def _assistant_messages(items: Sequence[tuple[JsonValue, ...]]) -> list[tuple[JsonValue, ...]]: + return [item for item in items if item[:2] == ("message", "assistant")] + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_next_turn_replays_responses_items_in_the_order_openai_produced(gateway: Gateway, stream: bool) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=stream) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert message["content"] == PLAN + PICK, message + assert object_value(message["provider_specific_fields"])["phase"] == "commentary", message + assert inputs[1] == (("message", "user", None, prompt), *WEATHER_REPLAY), inputs + + +def test_chat_next_turn_sends_the_phase_the_app_set_on_the_assistant_message(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], _with_phase(message, "final_answer")) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert [item[2] for item in _assistant_messages(inputs[1])] == ["final_answer"], inputs + + +def test_chat_next_turn_rejects_an_undefined_phase_unless_drop_params_is_set(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _with_phase(_first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False), "analysis") + rejected: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + dropped: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message, drop_params=True) + inputs: Final = posted_inputs(wire) + assert rejected.status_code == 400, rejected.text + assert "'analysis'" in rejected.text, rejected.text + assert dropped.status_code == 200, dropped.text + assert len(inputs) == 2, inputs + assert [item[2] for item in _assistant_messages(inputs[1])] == [None], inputs + + +def test_chat_next_turn_sends_the_text_the_app_edited_instead_of_the_recorded_items(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + edited_text: Final = "I picked Tokyo." + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], {**message, "content": edited_text}) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert _assistant_messages(inputs[1]) == [("message", "assistant", "commentary", edited_text)], inputs + + +def test_chat_next_turn_replays_a_custom_tool_call_as_a_custom_tool_call(gateway: Gateway) -> None: + prompt: Final = f"Create hello.txt containing hi {uuid.uuid4().hex}" + with wire_server(responses_upstream(PATCH_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [PATCH_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [PATCH_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert inputs[1] == ( + ("message", "user", None, prompt), + ("message", "assistant", "commentary", PATCH_NOTE), + ("custom_tool_call", "call_patch", "apply_patch"), + ("custom_tool_call_output", "call_patch"), + ), inputs + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_reply_keeps_every_reasoning_item_and_the_next_turn_replays_each_one( + gateway: Gateway, stream: bool +) -> None: + prompt: Final = f"Weigh two cities, then check whether it is raining in your pick {uuid.uuid4().hex}" + with wire_server(responses_upstream(TWO_THOUGHTS_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=stream) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert [object_value(item)["id"] for item in list_value(message["reasoning_items"])] == ["rs_scan", "rs_pick"] + assert inputs[1] == (("message", "user", None, prompt), *TWO_THOUGHTS_REPLAY), inputs + + +def test_chat_stream_of_a_tool_call_only_turn_carries_null_content_like_the_json_reply(gateway: Gateway) -> None: + prompt: Final = f"Check whether it is raining in Tokyo {uuid.uuid4().hex}" + with wire_server(responses_upstream(CALL_ONLY_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + streamed: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": [WEATHER_TOOL], + "stream": True, + "cache": {"no-cache": True}, + }, + ) + assert streamed.status_code == 200, streamed.text + deltas: Final = _deltas(_chunks(streamed)) + assert message["content"] is None, message + assert [delta["content"] for delta in deltas if delta.get("content") is not None] == [], deltas diff --git a/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py b/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py new file mode 100644 index 00000000000..0c58545182b --- /dev/null +++ b/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py @@ -0,0 +1,37 @@ +import uuid +from typing import Final, cast + +from integration._support.responses_replay import ( + ANSWER, + MODEL, + WEATHER_REPLAY, + WEATHER_TOOL, + WEATHER_TURN, + posted_inputs, + responses_upstream, +) +from integration._support.wire import wire_server + +import litellm +from litellm.types.llms.openai import ChatCompletionToolMessage, ChatCompletionToolParam +from litellm.types.utils import ChatCompletionMessageToolCall + + +def _rain(call: ChatCompletionMessageToolCall) -> ChatCompletionToolMessage: + return {"role": "tool", "tool_call_id": call.id, "content": "Rain, 18C"} + + +def test_run_tool_loop_replays_responses_items_in_the_order_openai_produced() -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire: + content: Final = litellm.run_tool_loop( + model=f"openai/responses/{MODEL}", + messages=[{"role": "user", "content": prompt}], + tools=[cast(ChatCompletionToolParam, WEATHER_TOOL)], + execute_tool=_rain, + api_base=wire.url, + api_key="synthetic-openai-key", + ) + inputs: Final = posted_inputs(wire) + assert content == ANSWER + assert inputs[1] == (("message", "user", None, prompt), *WEATHER_REPLAY), inputs