test(responses): add black box tests for chat turn history replay

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"
+    )

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

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