From 30d8b1251276a051130d0d69acbcf0a83247c529 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 21 Sep 2026 22:32:56 -0700 Subject: [PATCH] test(e2e): add conversational matrix across chat, messages and responses (#42359) * test(e2e): add conversational matrix across chat, messages and responses Parameterizes one behavioral contract (reply, stream, cost log, tool call, tool round trip) across /v1/chat/completions, /v1/messages and /v1/responses, OpenAI and Anthropic models, and env-ref vs stored-credential auth, with record/replay fixtures. Adds general_settings.disable_model_info_refresh so the proxy fronting a replay fixture does not poll every OpenAI-compatible deployment's /v1/models in the background, which otherwise leaves unconsumed interactions in the recorded bundle. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(e2e): force the weather tool on the first turn and rename Provider to Deployment Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: mateo Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/proxy/proxy_server.py | 30 +- .../coverage_registry/llm_conversational.yaml | 12 + tests/e2e/gateway/record_replay_ci_config.yml | 1 + .../llm_translation/conversational_matrix.py | 544 ++++++++++++++++++ .../test_conversational_matrix_e2e.py | 163 ++++++ .../proxy/proxy_server/test_lifecycle.py | 29 + 6 files changed, 767 insertions(+), 12 deletions(-) create mode 100644 tests/e2e/llm_translation/conversational_matrix.py create mode 100644 tests/e2e/llm_translation/test_conversational_matrix_e2e.py diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 77246804dcd..f8945694051 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1494,23 +1494,29 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[None, None]: ## Initialize shared aiohttp session for connection reuse shared_aiohttp_session = await _initialize_shared_aiohttp_session() - model_info_scheduler: Final = scheduler if scheduler is not None else AsyncIOScheduler() - model_info_scheduler.add_job( - ProxyStartupEvent.refresh_model_info, - "interval", - seconds=MODEL_INFO_REFRESH_SECONDS, - id="refresh_model_info", - next_run_time=datetime.now(timezone.utc), - max_instances=1, - replace_existing=True, + model_info_refresh_disabled: Final = ( + "disable_model_info_refresh" in general_settings and general_settings["disable_model_info_refresh"] is True ) - if not model_info_scheduler.running: - model_info_scheduler.start() + model_info_scheduler: Final = ( + None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler() + ) + if model_info_scheduler is not None: + model_info_scheduler.add_job( + ProxyStartupEvent.refresh_model_info, + "interval", + seconds=MODEL_INFO_REFRESH_SECONDS, + id="refresh_model_info", + next_run_time=datetime.now(timezone.utc), + max_instances=1, + replace_existing=True, + ) + if not model_info_scheduler.running: + model_info_scheduler.start() # End of startup event yield - if model_info_scheduler.running: + if model_info_scheduler is not None and model_info_scheduler.running: model_info_scheduler.remove_job("refresh_model_info") if model_info_scheduler is not scheduler: model_info_scheduler.shutdown(wait=False) diff --git a/tests/e2e/coverage_registry/llm_conversational.yaml b/tests/e2e/coverage_registry/llm_conversational.yaml index 49d4d92ff0b..e4e1ac2c7b6 100644 --- a/tests/e2e/coverage_registry/llm_conversational.yaml +++ b/tests/e2e/coverage_registry/llm_conversational.yaml @@ -94,3 +94,15 @@ - {id: llm.messages.together_ai.basic.stream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together over /v1/messages streaming"} - {id: llm.messages.together_ai.tool_use.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: tool_use, streaming: nonstream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together tool calls over /v1/messages"} - {id: llm.messages.together_ai.multi_turn.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: together_ai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_together_ai_e2e.py", rationale: "Together tool result round trip over /v1/messages"} +- {id: llm.chat_completions.anthropic.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: chat_completions, route: anthropic, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /chat/completions: cost header and spend row agree"} +- {id: llm.chat_completions.anthropic.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic tool result round trip over /chat/completions"} +- {id: llm.messages.anthropic.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic tool result round trip over /v1/messages"} +- {id: llm.messages.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI models served on the Anthropic Messages contract"} +- {id: llm.messages.openai.basic.stream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI over /v1/messages streams the Anthropic event grammar"} +- {id: llm.messages.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI over /v1/messages: cost header and spend row agree"} +- {id: llm.messages.openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI tool calls translated to Anthropic tool_use blocks"} +- {id: llm.messages.openai.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: openai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI tool result round trip over /v1/messages"} +- {id: llm.responses.openai.multi_turn.nonstream.works, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "OpenAI function_call_output round trip over /v1/responses"} +- {id: llm.responses.anthropic.basic.stream.works, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: basic, streaming: stream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /v1/responses streams Responses events"} +- {id: llm.responses.anthropic.basic.nonstream.cost_logged, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic over /v1/responses: cost header and spend row agree"} +- {id: llm.responses.anthropic.multi_turn.nonstream.works, module: llm, tier: P1, subject_endpoint: responses, route: anthropic, capability: multi_turn, streaming: nonstream, assertions: [works], source: "llm_translation/test_conversational_matrix_e2e.py", rationale: "Anthropic function_call_output round trip over /v1/responses"} diff --git a/tests/e2e/gateway/record_replay_ci_config.yml b/tests/e2e/gateway/record_replay_ci_config.yml index 08972969cf0..5b3db0ff530 100644 --- a/tests/e2e/gateway/record_replay_ci_config.yml +++ b/tests/e2e/gateway/record_replay_ci_config.yml @@ -1,3 +1,4 @@ general_settings: master_key: os.environ/LITELLM_MASTER_KEY store_model_in_db: true + disable_model_info_refresh: true diff --git a/tests/e2e/llm_translation/conversational_matrix.py b/tests/e2e/llm_translation/conversational_matrix.py new file mode 100644 index 00000000000..0d6f6ed3d4e --- /dev/null +++ b/tests/e2e/llm_translation/conversational_matrix.py @@ -0,0 +1,544 @@ +"""The endpoint x deployment x auth matrix behind test_conversational_matrix_e2e.py. + +One conversation, three wire formats. Each `Surface` speaks its own API through +the customer SDK (chat completions and Responses through the OpenAI SDK, Messages +through the Anthropic SDK) and folds what came back into the surface-neutral +`Reply` / `StreamedReply`, so a single behavior test asserts the same contract on +every cell. A new model, from an existing or a new provider, is one `Deployment` row +in DEPLOYMENTS; a new way of handing the proxy a provider credential is one +`AuthMethod`. +""" + +from __future__ import annotations + +import json +import os +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from types import MappingProxyType +from typing import Final, Literal, Protocol + +import anthropic +import openai +import pytest +from _pytest.mark.structures import ParameterSet +from anthropic.types import ( + MessageParam, + RawMessageStreamEvent, + TextBlock, + ToolChoiceToolParam, + ToolParam, + ToolResultBlockParam, + ToolUseBlock, + ToolUseBlockParam, +) +from e2e_config import provider_edge_base, unique_marker +from lifecycle import ResourceManager +from llm_translation.sdk_clients import NO_PROXY_CACHE, SdkClients, response_header +from models import CredentialCreateBody, LiteLLMParamsBody +from openai.types.chat import ( + ChatCompletionAssistantMessageParam, + ChatCompletionChunk, + ChatCompletionMessageFunctionToolCallParam, + ChatCompletionMessageParam, + ChatCompletionNamedToolChoiceParam, + ChatCompletionToolMessageParam, + ChatCompletionToolParam, +) +from openai.types.chat.chat_completion_message_function_tool_call import ChatCompletionMessageFunctionToolCall +from openai.types.responses import ( + FunctionToolParam, + ResponseFunctionToolCall, + ResponseFunctionToolCallParam, + ResponseInputParam, + ResponseStreamEvent, + ToolChoiceFunctionParam, +) +from openai.types.responses.response_input_param import FunctionCallOutput +from proxy_client import ProxyClient +from pydantic import BaseModel + +SurfaceName = Literal["chat_completions", "messages", "responses"] +AuthMethod = Literal["env_ref", "stored_credential"] +Capability = Literal["basic", "tool_use", "multi_turn"] +Streaming = Literal["stream", "nonstream"] +Assertion = Literal["works", "cost_logged"] +ToolMode = Literal["none", "forced", "offered"] + +SURFACES: Final[tuple[SurfaceName, ...]] = ("chat_completions", "messages", "responses") +AUTH_METHODS: Final[tuple[AuthMethod, ...]] = ("env_ref", "stored_credential") + +MAX_OUTPUT_TOKENS: Final = 512 +INSTRUCTIONS: Final = "You are a terse assistant. Answer in one short sentence." +GREETING_PROMPT: Final = "Say hello." +WEATHER_PROMPT: Final = "What is the weather in Paris right now? Use the get_weather tool." +WEATHER_REPORT: Final = "Paris: 22 degrees Celsius, clear skies" +WEATHER_TOOL_NAME: Final = "get_weather" +WEATHER_TOOL_DESCRIPTION: Final = "Current weather for a city" +WEATHER_TOOL_SCHEMA: Final[Mapping[str, object]] = MappingProxyType( + { + "type": "object", + "properties": {"location": {"type": "string", "description": "City name"}}, + "required": ["location"], + } +) + + +@dataclass(frozen=True, slots=True) +class Deployment: + """One deployment target: the litellm backend string plus how to wire it.""" + + route: Literal["openai", "anthropic"] + label: str + backend: str + api_key_env: str + edge_mount: str + edge_suffix: str + + def api_base(self) -> str | None: + base: Final = provider_edge_base(self.edge_mount) + return None if base is None else f"{base}{self.edge_suffix}" + + def api_key(self) -> str: + key: Final = os.environ.get(self.api_key_env, "") + assert key, f"{self.api_key_env} is not set in the test process environment" + return key + + +DEPLOYMENTS: Final[tuple[Deployment, ...]] = ( + Deployment( + route="openai", + label="gpt-4o-mini", + backend="openai/gpt-4o-mini", + api_key_env="OPENAI_API_KEY", + edge_mount="openai", + edge_suffix="/v1", + ), + Deployment( + route="openai", + label="gpt-5.4-mini", + backend="openai/gpt-5.4-mini", + api_key_env="OPENAI_API_KEY", + edge_mount="openai", + edge_suffix="/v1", + ), + Deployment( + route="anthropic", + label="claude-haiku-4-5", + backend="anthropic/claude-haiku-4-5", + api_key_env="ANTHROPIC_API_KEY", + edge_mount="anthropic", + edge_suffix="", + ), +) + + +@dataclass(frozen=True, slots=True) +class Cell: + surface: SurfaceName + deployment: Deployment + auth: AuthMethod + + @property + def id(self) -> str: + return f"{self.surface}-{self.deployment.label}-{self.auth}" + + def registry_id(self, capability: Capability, streaming: Streaming, assertion: Assertion) -> str: + return f"llm.{self.surface}.{self.deployment.route}.{capability}.{streaming}.{assertion}" + + +CELLS: Final[tuple[Cell, ...]] = tuple( + Cell(surface=surface, deployment=deployment, auth=auth) + for surface in SURFACES + for deployment in DEPLOYMENTS + for auth in AUTH_METHODS +) + + +def cells_covering(capability: Capability, streaming: Streaming, assertion: Assertion) -> tuple[ParameterSet, ...]: + """Every cell as a pytest param carrying the registry id its test proves.""" + return tuple( + pytest.param(cell, id=cell.id, marks=pytest.mark.covers(cell.registry_id(capability, streaming, assertion))) + for cell in CELLS + ) + + +DeploymentKey = tuple[str, AuthMethod] + + +@dataclass(frozen=True, slots=True) +class Deployments: + """Model aliases registered on the proxy, one per (deployment, auth).""" + + aliases: Mapping[DeploymentKey, str] + + def alias(self, cell: Cell) -> str: + return self.aliases[(cell.deployment.label, cell.auth)] + + +def _litellm_params(deployment: Deployment, auth: AuthMethod, credential_name: str) -> LiteLLMParamsBody: + match auth: + case "env_ref": + return LiteLLMParamsBody( + model=deployment.backend, api_key=f"os.environ/{deployment.api_key_env}", api_base=deployment.api_base() + ) + case "stored_credential": + return LiteLLMParamsBody( + model=deployment.backend, litellm_credential_name=credential_name, api_base=deployment.api_base() + ) + + +def _register(proxy: ProxyClient, resources: ResourceManager, deployment: Deployment, auth: AuthMethod) -> str: + marker: Final = unique_marker() + credential_name: Final = f"e2e-matrix-{deployment.label}-{marker}" + if auth == "stored_credential": + proxy.create_credential( + CredentialCreateBody(credential_name=credential_name, credential_values={"api_key": deployment.api_key()}) + ) + resources.defer(lambda: proxy.delete_credential(credential_name)) + alias: Final = f"e2e-matrix-{deployment.label}-{auth}-{marker}" + model_id: Final = proxy.create_model(alias, _litellm_params(deployment, auth, credential_name)) + resources.defer(lambda: proxy.delete_model(model_id)) + return alias + + +def register_deployments(proxy: ProxyClient) -> Iterator[Deployments]: + resources: Final = ResourceManager(client=proxy) + try: + yield Deployments( + aliases=MappingProxyType( + { + (deployment.label, auth): _register(proxy, resources, deployment, auth) + for deployment in DEPLOYMENTS + for auth in AUTH_METHODS + } + ) + ) + finally: + resources.teardown() + + +class WeatherArgs(BaseModel): + location: str + + +@dataclass(frozen=True, slots=True) +class ToolCall: + call_id: str + name: str + arguments: str + + def parsed(self) -> WeatherArgs: + return WeatherArgs.model_validate_json(self.arguments) + + +@dataclass(frozen=True, slots=True) +class Usage: + input_tokens: int + output_tokens: int + + +@dataclass(frozen=True, slots=True) +class Reply: + """What every surface owes the caller for one non-streamed turn.""" + + response_id: str + text: str + tool_calls: tuple[ToolCall, ...] + usage: Usage | None + call_id_header: str | None + cost_header: str | None + + +@dataclass(frozen=True, slots=True) +class StreamedReply: + """The reassembled stream: its text, whether the surface's own terminal event + arrived, and whether usage was reported anywhere in the stream.""" + + text: str + finished: bool + usage_reported: bool + event_count: int + + +class Surface(Protocol): + @property + def name(self) -> SurfaceName: ... + + def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply: ... + + def stream(self, key: str, model: str, prompt: str) -> StreamedReply: ... + + def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply: ... + + +def _chat_tool() -> ChatCompletionToolParam: + return { + "type": "function", + "function": { + "name": WEATHER_TOOL_NAME, + "description": WEATHER_TOOL_DESCRIPTION, + "parameters": dict(WEATHER_TOOL_SCHEMA), + }, + } + + +def _messages_tool() -> ToolParam: + return { + "name": WEATHER_TOOL_NAME, + "description": WEATHER_TOOL_DESCRIPTION, + "input_schema": dict(WEATHER_TOOL_SCHEMA), + } + + +def _responses_tool() -> FunctionToolParam: + return { + "type": "function", + "name": WEATHER_TOOL_NAME, + "description": WEATHER_TOOL_DESCRIPTION, + "parameters": dict(WEATHER_TOOL_SCHEMA), + "strict": False, + } + + +def _chat_tool_choice() -> ChatCompletionNamedToolChoiceParam: + return {"type": "function", "function": {"name": WEATHER_TOOL_NAME}} + + +def _messages_tool_choice() -> ToolChoiceToolParam: + return {"type": "tool", "name": WEATHER_TOOL_NAME, "disable_parallel_tool_use": True} + + +def _responses_tool_choice() -> ToolChoiceFunctionParam: + return {"type": "function", "name": WEATHER_TOOL_NAME} + + +def _usage(input_tokens: int | None, output_tokens: int | None) -> Usage | None: + if input_tokens is None or output_tokens is None: + return None + return Usage(input_tokens=input_tokens, output_tokens=output_tokens) + + +@dataclass(frozen=True, slots=True) +class ChatCompletionsSurface: + sdk: SdkClients + name: SurfaceName = "chat_completions" + + def _turn(self, key: str, model: str, messages: Sequence[ChatCompletionMessageParam], tool: ToolMode) -> Reply: + raw: Final = self.sdk.openai(key).chat.completions.with_raw_response.create( + model=model, + messages=list(messages), + max_completion_tokens=MAX_OUTPUT_TOKENS, + tools=openai.omit if tool == "none" else [_chat_tool()], + tool_choice=_chat_tool_choice() if tool == "forced" else openai.omit, + parallel_tool_calls=False if tool == "forced" else openai.omit, + extra_body=NO_PROXY_CACHE, + ) + completion: Final = raw.parse() + message: Final = completion.choices[0].message + calls: Final = tuple( + ToolCall(call_id=call.id, name=call.function.name, arguments=call.function.arguments) + for call in message.tool_calls or () + if isinstance(call, ChatCompletionMessageFunctionToolCall) + ) + return Reply( + response_id=completion.id, + text=message.content or "", + tool_calls=calls, + usage=None + if completion.usage is None + else _usage(completion.usage.prompt_tokens, completion.usage.completion_tokens), + call_id_header=response_header(raw.headers, "x-litellm-call-id"), + cost_header=response_header(raw.headers, "x-litellm-response-cost"), + ) + + def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply: + return self._turn(key, model, _chat_history(prompt), "forced" if with_tool else "none") + + def stream(self, key: str, model: str, prompt: str) -> StreamedReply: + chunks: Final[tuple[ChatCompletionChunk, ...]] = tuple( + self.sdk.openai(key).chat.completions.create( + model=model, + messages=_chat_history(prompt), + max_completion_tokens=MAX_OUTPUT_TOKENS, + stream=True, + stream_options={"include_usage": True}, + extra_body=NO_PROXY_CACHE, + ) + ) + return StreamedReply( + text="".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices), + finished=any(chunk.choices[0].finish_reason is not None for chunk in chunks if chunk.choices), + usage_reported=any(chunk.usage is not None for chunk in chunks), + event_count=len(chunks), + ) + + def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply: + tool_call: Final[ChatCompletionMessageFunctionToolCallParam] = { + "id": call.call_id, + "type": "function", + "function": {"name": call.name, "arguments": call.arguments}, + } + assistant: Final[ChatCompletionAssistantMessageParam] = {"role": "assistant", "tool_calls": [tool_call]} + tool_result: Final[ChatCompletionToolMessageParam] = { + "role": "tool", + "tool_call_id": call.call_id, + "content": result, + } + return self._turn(key, model, (*_chat_history(prompt), assistant, tool_result), "offered") + + +def _chat_history(prompt: str) -> tuple[ChatCompletionMessageParam, ...]: + return ({"role": "system", "content": INSTRUCTIONS}, {"role": "user", "content": prompt}) + + +@dataclass(frozen=True, slots=True) +class MessagesSurface: + sdk: SdkClients + name: SurfaceName = "messages" + + def _turn(self, key: str, model: str, messages: Sequence[MessageParam], tool: ToolMode) -> Reply: + raw: Final = self.sdk.anthropic(key).messages.with_raw_response.create( + model=model, + max_tokens=MAX_OUTPUT_TOKENS, + system=INSTRUCTIONS, + messages=list(messages), + tools=anthropic.omit if tool == "none" else [_messages_tool()], + tool_choice=_messages_tool_choice() if tool == "forced" else anthropic.omit, + extra_body=NO_PROXY_CACHE, + ) + message: Final = raw.parse() + return Reply( + response_id=message.id, + text="".join(block.text for block in message.content if isinstance(block, TextBlock)), + tool_calls=tuple( + ToolCall(call_id=block.id, name=block.name, arguments=json.dumps(block.input)) + for block in message.content + if isinstance(block, ToolUseBlock) + ), + usage=_usage(message.usage.input_tokens, message.usage.output_tokens), + call_id_header=response_header(raw.headers, "x-litellm-call-id"), + cost_header=response_header(raw.headers, "x-litellm-response-cost"), + ) + + def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply: + return self._turn(key, model, ({"role": "user", "content": prompt},), "forced" if with_tool else "none") + + def stream(self, key: str, model: str, prompt: str) -> StreamedReply: + events: Final[tuple[RawMessageStreamEvent, ...]] = tuple( + self.sdk.anthropic(key).messages.create( + model=model, + max_tokens=MAX_OUTPUT_TOKENS, + system=INSTRUCTIONS, + messages=[{"role": "user", "content": prompt}], + stream=True, + extra_body=NO_PROXY_CACHE, + ) + ) + return StreamedReply( + text="".join( + event.delta.text + for event in events + if event.type == "content_block_delta" and event.delta.type == "text_delta" + ), + finished=any(event.type == "message_stop" for event in events), + usage_reported=any(event.type == "message_delta" and event.usage.output_tokens > 0 for event in events), + event_count=len(events), + ) + + def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply: + tool_use: Final[ToolUseBlockParam] = { + "type": "tool_use", + "id": call.call_id, + "name": call.name, + "input": call.parsed().model_dump(), + } + tool_result: Final[ToolResultBlockParam] = { + "type": "tool_result", + "tool_use_id": call.call_id, + "content": result, + } + history: Final[tuple[MessageParam, ...]] = ( + {"role": "user", "content": prompt}, + {"role": "assistant", "content": [tool_use]}, + {"role": "user", "content": [tool_result]}, + ) + return self._turn(key, model, history, "offered") + + +@dataclass(frozen=True, slots=True) +class ResponsesSurface: + sdk: SdkClients + name: SurfaceName = "responses" + + def _turn(self, key: str, model: str, history: ResponseInputParam, tool: ToolMode) -> Reply: + raw: Final = self.sdk.openai(key).responses.with_raw_response.create( + model=model, + input=history, + instructions=INSTRUCTIONS, + max_output_tokens=MAX_OUTPUT_TOKENS, + tools=openai.omit if tool == "none" else [_responses_tool()], + tool_choice=_responses_tool_choice() if tool == "forced" else openai.omit, + parallel_tool_calls=False if tool == "forced" else openai.omit, + extra_body=NO_PROXY_CACHE, + ) + response: Final = raw.parse() + return Reply( + response_id=response.id, + text=response.output_text, + tool_calls=tuple( + ToolCall(call_id=item.call_id, name=item.name, arguments=item.arguments) + for item in response.output + if isinstance(item, ResponseFunctionToolCall) + ), + usage=None if response.usage is None else _usage(response.usage.input_tokens, response.usage.output_tokens), + call_id_header=response_header(raw.headers, "x-litellm-call-id"), + cost_header=response_header(raw.headers, "x-litellm-response-cost"), + ) + + def reply(self, key: str, model: str, prompt: str, *, with_tool: bool = False) -> Reply: + return self._turn(key, model, [{"role": "user", "content": prompt}], "forced" if with_tool else "none") + + def stream(self, key: str, model: str, prompt: str) -> StreamedReply: + events: Final[tuple[ResponseStreamEvent, ...]] = tuple( + self.sdk.openai(key).responses.create( + model=model, + input=prompt, + instructions=INSTRUCTIONS, + max_output_tokens=MAX_OUTPUT_TOKENS, + stream=True, + extra_body=NO_PROXY_CACHE, + ) + ) + return StreamedReply( + text="".join(event.delta for event in events if event.type == "response.output_text.delta"), + finished=bool(events) and events[-1].type == "response.completed", + usage_reported=any( + event.type == "response.completed" and event.response.usage is not None for event in events + ), + event_count=len(events), + ) + + def reply_to_tool_result(self, key: str, model: str, prompt: str, call: ToolCall, result: str) -> Reply: + function_call: Final[ResponseFunctionToolCallParam] = { + "type": "function_call", + "call_id": call.call_id, + "name": call.name, + "arguments": call.arguments, + } + output: Final[FunctionCallOutput] = { + "type": "function_call_output", + "call_id": call.call_id, + "output": result, + } + return self._turn(key, model, [{"role": "user", "content": prompt}, function_call, output], "offered") + + +def build_surfaces(sdk: SdkClients) -> Mapping[SurfaceName, Surface]: + return MappingProxyType[SurfaceName, Surface]( + { + "chat_completions": ChatCompletionsSurface(sdk), + "messages": MessagesSurface(sdk), + "responses": ResponsesSurface(sdk), + } + ) diff --git a/tests/e2e/llm_translation/test_conversational_matrix_e2e.py b/tests/e2e/llm_translation/test_conversational_matrix_e2e.py new file mode 100644 index 00000000000..a5cfc9ce70b --- /dev/null +++ b/tests/e2e/llm_translation/test_conversational_matrix_e2e.py @@ -0,0 +1,163 @@ +"""The same conversation contract on every (endpoint, deployment, auth) cell. +A deployment is one provider model (openai/gpt-4o-mini, anthropic/claude-haiku-4-5, ...). + +/chat/completions, /v1/messages and /v1/responses each have their own +translation code in the proxy, so a bug fixed on one surface tends to survive +on the others. Every test here runs once per cell in `CELLS` +(conversational_matrix.py), so a change to a shared helper is proven against all +surfaces and providers at once, and a new model or provider is one row in `DEPLOYMENTS`. + +Edge-wired: OpenAI and Anthropic traffic goes through the provider edge in +record and replay, so the whole matrix replays with zero provider calls. +""" + +from __future__ import annotations + +from collections.abc import Iterator, Mapping +from typing import Final + +import pytest +from e2e_config import unique_marker +from lifecycle import ResourceManager +from llm_translation.conversational_matrix import ( + GREETING_PROMPT, + WEATHER_PROMPT, + WEATHER_REPORT, + WEATHER_TOOL_NAME, + Cell, + Deployments, + Surface, + SurfaceName, + ToolCall, + build_surfaces, + cells_covering, + register_deployments, +) +from llm_translation.sdk_clients import SdkClients +from models import SpendLogRow +from proxy_client import ProxyClient + +pytestmark = [pytest.mark.e2e, pytest.mark.replayable] + + +def _approx_equal(actual: float, expected: float) -> bool: + return abs(actual - expected) <= max(1e-9, abs(expected) * 1e-2) + + +@pytest.fixture(scope="module") +def deployments(proxy: ProxyClient) -> Iterator[Deployments]: + yield from register_deployments(proxy) + + +@pytest.fixture(scope="module") +def surfaces(sdk: SdkClients) -> Mapping[SurfaceName, Surface]: + return build_surfaces(sdk) + + +def _weather_call(surface: Surface, key: str, model: str) -> ToolCall: + first: Final = surface.reply(key, model, WEATHER_PROMPT, with_tool=True) + assert len(first.tool_calls) == 1, ( + f"{surface.name} forced tool_choice={WEATHER_TOOL_NAME} with parallel calls off, " + f"got {len(first.tool_calls)} tool call(s): {first.tool_calls} text={first.text!r}" + ) + call: Final = first.tool_calls[0] + assert call.name == WEATHER_TOOL_NAME, f"{surface.name} called {call.name!r}, not the forced {WEATHER_TOOL_NAME!r}" + assert call.call_id, f"{surface.name} tool call has no id, so the caller cannot answer it: {call}" + assert "paris" in call.parsed().location.lower(), f"{surface.name} tool arguments lost the location: {call}" + return call + + +class TestConversationalMatrix: + @pytest.mark.parametrize("cell", cells_covering("basic", "nonstream", "works")) + def test_reply_carries_assistant_text_and_usage( + self, + cell: Cell, + deployments: Deployments, + surfaces: Mapping[SurfaceName, Surface], + resources: ResourceManager, + ) -> None: + surface: Final = surfaces[cell.surface] + reply: Final = surface.reply(resources.key(), deployments.alias(cell), GREETING_PROMPT) + + assert reply.response_id, f"{cell.id}: response has no id" + assert reply.text.strip(), f"{cell.id}: response carried no assistant text" + assert reply.usage is not None and reply.usage.input_tokens > 0 and reply.usage.output_tokens > 0, ( + f"{cell.id}: usage missing or zero, so the caller cannot account for this call: {reply.usage}" + ) + assert reply.call_id_header, f"{cell.id}: x-litellm-call-id header missing" + + @pytest.mark.parametrize("cell", cells_covering("basic", "stream", "works")) + def test_stream_delivers_text_usage_and_a_terminal_event( + self, + cell: Cell, + deployments: Deployments, + surfaces: Mapping[SurfaceName, Surface], + resources: ResourceManager, + ) -> None: + surface: Final = surfaces[cell.surface] + streamed: Final = surface.stream(resources.key(), deployments.alias(cell), GREETING_PROMPT) + + assert streamed.event_count > 1, f"{cell.id}: stream arrived as {streamed.event_count} event(s), not a stream" + assert streamed.text.strip(), f"{cell.id}: stream carried no text deltas" + assert streamed.finished, f"{cell.id}: stream never sent its terminal event" + assert streamed.usage_reported, f"{cell.id}: stream never reported usage" + + @pytest.mark.parametrize("cell", cells_covering("basic", "nonstream", "cost_logged")) + def test_cost_header_matches_the_spend_log( + self, + cell: Cell, + deployments: Deployments, + surfaces: Mapping[SurfaceName, Surface], + resources: ResourceManager, + proxy: ProxyClient, + ) -> None: + key: Final = resources.key() + surface: Final = surfaces[cell.surface] + reply: Final = surface.reply(key, deployments.alias(cell), f"{GREETING_PROMPT} {unique_marker()}") + + assert reply.cost_header is not None, f"{cell.id}: x-litellm-response-cost header missing" + header_cost: Final = float(reply.cost_header) + assert header_cost > 0, f"{cell.id}: x-litellm-response-cost is not positive: {header_cost}" + + def _priced(rows: list[SpendLogRow]) -> bool: + return any(row.spend is not None and row.spend > 0 for row in rows) + + rows: Final = proxy.poll_logs_for_key(key, predicate=_priced) + priced: Final = tuple(row for row in rows if row.spend is not None and row.spend > 0) + assert len(priced) == 1, f"{cell.id}: expected exactly one priced spend row for a fresh key, got {rows}" + row: Final = priced[0] + assert (row.prompt_tokens or 0) > 0 and (row.completion_tokens or 0) > 0, ( + f"{cell.id}: spend row has no token counts, so the cost is not real usage: {row}" + ) + assert row.spend is not None and _approx_equal(row.spend, header_cost), ( + f"{cell.id}: logged spend {row.spend} disagrees with x-litellm-response-cost {header_cost}" + ) + assert row.model and cell.deployment.backend.endswith(row.model), ( + f"{cell.id}: spend row logged model {row.model!r}, not the deployment's {cell.deployment.backend!r}" + ) + + @pytest.mark.parametrize("cell", cells_covering("tool_use", "nonstream", "works")) + def test_tool_call_is_returned_named_and_addressable( + self, + cell: Cell, + deployments: Deployments, + surfaces: Mapping[SurfaceName, Surface], + resources: ResourceManager, + ) -> None: + _weather_call(surfaces[cell.surface], resources.key(), deployments.alias(cell)) + + @pytest.mark.parametrize("cell", cells_covering("multi_turn", "nonstream", "works")) + def test_tool_result_round_trip_reaches_the_model( + self, + cell: Cell, + deployments: Deployments, + surfaces: Mapping[SurfaceName, Surface], + resources: ResourceManager, + ) -> None: + key: Final = resources.key() + model: Final = deployments.alias(cell) + surface: Final = surfaces[cell.surface] + call: Final = _weather_call(surface, key, model) + + answer: Final = surface.reply_to_tool_result(key, model, WEATHER_PROMPT, call, WEATHER_REPORT) + assert "22" in answer.text, f"{cell.id}: the model never saw the tool result: {answer.text!r}" diff --git a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py index e9695685ca5..9954351fa2e 100644 --- a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py +++ b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py @@ -818,6 +818,35 @@ async def test_proxy_startup_event_prunes_dead_workers_live_gauges(tmp_path): assert counter.exists() +@pytest.mark.asyncio +@pytest.mark.parametrize(("disable_model_info_refresh", "job_scheduled"), [(True, False), (False, True)]) +async def test_proxy_startup_event_honors_disable_model_info_refresh( + disable_model_info_refresh: bool, job_scheduled: bool +) -> None: + """``general_settings.disable_model_info_refresh: true`` keeps the proxy from polling every + OpenAI-compatible deployment's ``/v1/models`` in the background, so a proxy fronting a replay + fixture (or a metered upstream) makes only the calls its clients asked for.""" + scheduler = AsyncIOScheduler() + clean_env = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL")} | { + "LITELLM_DANGEROUSLY_PERMIT_WEAK_OR_UNSET_MASTER_KEY": "true" + } + with ( + patch.dict(os.environ, clean_env, clear=True), + patch.object(ps, "scheduler", scheduler), + patch.dict(ps.general_settings, {"disable_model_info_refresh": disable_model_info_refresh}), + ): + try: + async with proxy_startup_event(app=None): + job = scheduler.get_job("refresh_model_info") + finally: + if scheduler.running: + scheduler.shutdown(wait=False) + + assert (job is not None) is job_scheduled, ( + f"disable_model_info_refresh={disable_model_info_refresh} but refresh_model_info job is {job}" + ) + + def test_otel_global_provider_published_after_callback_init(): """The OTel V2 global-provider publish must run after callback initialization in ``proxy_startup_event``.