| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Tests for the ``openai-responses`` backend of ``vllm bench serve``. |
| 4 | |
| 5 | The Responses stream differs from Chat Completions in ways the benchmark client |
| 6 | has to get right: every SSE message is labelled with an ``event:`` line, |
| 7 | reasoning and output text arrive as separate token-bearing events, and a |
| 8 | request ends with ``response.completed`` rather than with its last token. These |
| 9 | tests drive the real request function against a local SSE server so the |
| 10 | resulting latency and usage numbers are covered end to end. |
| 11 | """ |
| 12 | |
| 13 | import asyncio |
| 14 | import json |
| 15 | from collections.abc import Sequence |
| 16 | |
| 17 | import aiohttp |
| 18 | import pytest |
| 19 | from aiohttp import web |
| 20 | from aiohttp.test_utils import TestServer |
| 21 | |
| 22 | from vllm.benchmarks.lib.endpoint_request_func import ( |
| 23 | ASYNC_REQUEST_FUNCS, |
| 24 | OPENAI_COMPATIBLE_BACKENDS, |
| 25 | RequestFuncInput, |
| 26 | RequestFuncOutput, |
| 27 | async_request_openai_responses, |
| 28 | ) |
| 29 | |
| 30 | # These tests never touch torch, so skip the GPU cleanup fixture. |
| 31 | pytestmark = pytest.mark.skip_global_cleanup |
| 32 | |
| 33 | MODEL = "openai/gpt-oss-20b" |
| 34 | |
| 35 | # Gaps between scripted events. GAP_TOKEN is much larger than GAP_PREFILL so |
| 36 | # that a time to first token measured at the wrong event is unambiguous. |
| 37 | GAP_PREFILL = 0.15 |
| 38 | GAP_TOKEN = 0.5 |
| 39 | GAP_FINALIZE = 0.3 |
| 40 | |
| 41 | |
| 42 | def sse(event_type: str, payload: dict) -> bytes: |
| 43 | """Frame one event the way ``/v1/responses`` does, with an ``event:`` line. |
| 44 | |
| 45 | ``ensure_ascii=False`` matches the server, whose pydantic |
| 46 | ``model_dump_json`` emits non-ASCII characters unescaped. |
| 47 | """ |
| 48 | data = json.dumps({"type": event_type, **payload}, ensure_ascii=False) |
| 49 | return f"event: {event_type}\ndata: {data}\n\n".encode() |
| 50 | |
| 51 | |
| 52 | def usage_event( |
| 53 | input_tokens: int, |
| 54 | output_tokens: int, |
| 55 | event_type: str = "response.completed", |
| 56 | status: str = "completed", |
| 57 | ): |
| 58 | return sse( |
| 59 | event_type, |
| 60 | { |
| 61 | "response": { |
| 62 | "status": status, |
| 63 | "usage": { |
| 64 | "input_tokens": input_tokens, |
| 65 | "output_tokens": output_tokens, |
| 66 | "total_tokens": input_tokens + output_tokens, |
| 67 | }, |
| 68 | } |
| 69 | }, |
| 70 | ) |
| 71 | |
| 72 | |
| 73 | def text_delta(delta: str) -> bytes: |
| 74 | return sse("response.output_text.delta", {"delta": delta}) |
| 75 | |
| 76 | |
| 77 | def reasoning_delta(delta: str) -> bytes: |
| 78 | return sse("response.reasoning_text.delta", {"delta": delta}) |
| 79 | |
| 80 | |
| 81 | # The metadata events the server sends before any token. None of these may |
| 82 | # start the TTFT clock or contribute an ITL sample. |
| 83 | PREAMBLE = ( |
| 84 | sse("response.created", {"response": {"status": "in_progress"}}) |
| 85 | + sse("response.in_progress", {"response": {"status": "in_progress"}}) |
| 86 | + sse("response.output_item.added", {"output_index": 0}) |
| 87 | + sse("response.reasoning_part.added", {"output_index": 0}) |
| 88 | + sse("response.content_part.added", {"output_index": 0}) |
| 89 | ) |
| 90 | |
| 91 | # The `.done` family the server sends after the last delta. These repeat the |
| 92 | # accumulated text, so timing them would add bogus ITL samples and counting |
| 93 | # their payload would duplicate the output. |
| 94 | DONE_EVENTS = ( |
| 95 | sse("response.reasoning_text.done", {"text": "The user asks "}) |
| 96 | + sse("response.reasoning_part.done", {"output_index": 0}) |
| 97 | + sse("response.output_text.done", {"text": "Rayleigh scattering."}) |
| 98 | + sse("response.content_part.done", {"output_index": 0}) |
| 99 | + sse("response.output_item.done", {"output_index": 0}) |
| 100 | ) |
| 101 | |
| 102 | |
| 103 | async def _serve( |
| 104 | script: Sequence[bytes | float], |
| 105 | *, |
| 106 | status: int = 200, |
| 107 | chunk_size: int | None = None, |
| 108 | seen: dict | None = None, |
| 109 | **request_kwargs, |
| 110 | ) -> RequestFuncOutput: |
| 111 | """Run the request function against a server replaying ``script``. |
| 112 | |
| 113 | Args: |
| 114 | script: Byte strings to write, interleaved with floats to sleep for. |
| 115 | status: HTTP status to respond with. Non-200 skips the script. |
| 116 | chunk_size: If set, split every write into chunks of this many bytes to |
| 117 | emulate fragmented transport. |
| 118 | seen: If given, populated with the request payload. |
| 119 | request_kwargs: Extra fields for the ``RequestFuncInput``. |
| 120 | |
| 121 | """ |
| 122 | |
| 123 | async def handler(request: web.Request) -> web.StreamResponse: |
| 124 | if seen is not None: |
| 125 | seen["payload"] = await request.json() |
| 126 | seen["headers"] = dict(request.headers) |
| 127 | if status != 200: |
| 128 | return web.Response(status=status, reason="Bad Request") |
| 129 | |
| 130 | response = web.StreamResponse(headers={"Content-Type": "text/event-stream"}) |
| 131 | await response.prepare(request) |
| 132 | for item in script: |
| 133 | if isinstance(item, float): |
| 134 | await asyncio.sleep(item) |
| 135 | elif chunk_size is None: |
| 136 | await response.write(item) |
| 137 | else: |
| 138 | for i in range(0, len(item), chunk_size): |
| 139 | await response.write(item[i : i + chunk_size]) |
| 140 | await response.write_eof() |
| 141 | return response |
| 142 | |
| 143 | app = web.Application() |
| 144 | app.router.add_post("/v1/responses", handler) |
| 145 | server = TestServer(app) |
| 146 | await server.start_server() |
| 147 | try: |
| 148 | request_input = RequestFuncInput( |
| 149 | prompt="Why is the sky blue?", |
| 150 | api_url=str(server.make_url("/v1/responses")), |
| 151 | prompt_len=5, |
| 152 | output_len=16, |
| 153 | model=MODEL, |
| 154 | **request_kwargs, |
| 155 | ) |
| 156 | async with aiohttp.ClientSession() as session: |
| 157 | return await async_request_openai_responses(request_input, session) |
| 158 | finally: |
| 159 | await server.close() |
| 160 | |
| 161 | |
| 162 | def run(script: Sequence[bytes | float], **kwargs) -> RequestFuncOutput: |
| 163 | return asyncio.run(_serve(script, **kwargs)) |
| 164 | |
| 165 | |
| 166 | def test_backend_is_registered_as_openai_compatible(): |
| 167 | assert ASYNC_REQUEST_FUNCS["openai-responses"] is async_request_openai_responses |
| 168 | # Membership enables --ignore-eos on random datasets and the sampling |
| 169 | # parameter flags, all of which ResponsesRequest accepts. |
| 170 | assert "openai-responses" in OPENAI_COMPATIBLE_BACKENDS |
| 171 | |
| 172 | |
| 173 | def test_request_payload_is_accepted_by_the_server_schema(): |
| 174 | """Every field sent must be a real ResponsesRequest field. |
| 175 | |
| 176 | This is what makes the backend safe to list in OPENAI_COMPATIBLE_BACKENDS: |
| 177 | the flags that membership enables have to reach the model, and the server |
| 178 | drops unknown fields without complaining. |
| 179 | """ |
| 180 | from vllm.entrypoints.openai.responses.protocol import ResponsesRequest |
| 181 | |
| 182 | sampling = { |
| 183 | "temperature": 0.7, |
| 184 | "top_p": 0.9, |
| 185 | "top_k": 20, |
| 186 | "frequency_penalty": 0.2, |
| 187 | "presence_penalty": 0.1, |
| 188 | "repetition_penalty": 1.05, |
| 189 | } |
| 190 | seen: dict = {} |
| 191 | output = run( |
| 192 | [PREAMBLE, text_delta("hi"), usage_event(5, 1)], |
| 193 | seen=seen, |
| 194 | extra_body=sampling, |
| 195 | ignore_eos=True, |
| 196 | model_name="served-alias", |
| 197 | request_id="bench-7", |
| 198 | extra_headers={"x-team": "perf"}, |
| 199 | ) |
| 200 | |
| 201 | assert output.success |
| 202 | # --served-model-name wins over --model, as on the other backends. |
| 203 | assert seen["payload"]["model"] == "served-alias" |
| 204 | # The route is behind validate_json_request, which 415s a missing type. |
| 205 | assert seen["headers"]["Content-Type"] == "application/json" |
| 206 | assert seen["headers"]["x-request-id"] == "bench-7" |
| 207 | assert seen["headers"]["x-team"] == "perf" |
| 208 | assert seen["payload"] == { |
| 209 | "model": "served-alias", |
| 210 | "input": "Why is the sky blue?", |
| 211 | "max_output_tokens": 16, |
| 212 | "stream": True, |
| 213 | "store": False, |
| 214 | "ignore_eos": True, |
| 215 | **sampling, |
| 216 | } |
| 217 | assert set(seen["payload"]) <= set(ResponsesRequest.model_fields) |
| 218 | ResponsesRequest.model_validate(seen["payload"]) |
| 219 | |
| 220 | # `vllm bench serve` rejects --min-p for this backend for the same reason. |
| 221 | assert "min_p" not in ResponsesRequest.model_fields |
| 222 | |
| 223 | |
| 224 | def test_latency_accounting_matches_the_other_endpoints(): |
| 225 | """`latency - ttft` must equal `sum(itl)`. |
| 226 | |
| 227 | `serve.py` derives TPOT as `(latency - ttft) / (output_len - 1)`, which only |
| 228 | means average inter-token latency when that identity holds. #55508 fixed the |
| 229 | same accounting on the chat and audio backends; the trailing events after |
| 230 | the last token must not extend the request. |
| 231 | """ |
| 232 | output = run( |
| 233 | [ |
| 234 | PREAMBLE, |
| 235 | GAP_PREFILL / 2, |
| 236 | # A keep-alive comment sent during prefill is not a token. |
| 237 | b": keep-alive\n\n", |
| 238 | GAP_PREFILL / 2, |
| 239 | reasoning_delta("The user asks "), |
| 240 | GAP_TOKEN, |
| 241 | text_delta("Rayleigh"), |
| 242 | text_delta(" scattering."), |
| 243 | DONE_EVENTS, |
| 244 | GAP_FINALIZE, |
| 245 | usage_event(5, 8), |
| 246 | ] |
| 247 | ) |
| 248 | |
| 249 | assert output.success |
| 250 | # Reasoning is timed but not returned, matching how `openai-chat` treats |
| 251 | # `DeltaMessage.reasoning`. |
| 252 | assert output.generated_text == "Rayleigh scattering." |
| 253 | |
| 254 | # Neither the response.created/in_progress/output_item.added preamble nor |
| 255 | # the keep-alive comment may start the clock, and the reasoning delta must, |
| 256 | # so TTFT lands on the prefill gap rather than near zero or out at the |
| 257 | # first output text delta. |
| 258 | assert GAP_PREFILL * 0.8 <= output.ttft < GAP_PREFILL + GAP_TOKEN * 0.5 |
| 259 | |
| 260 | # One ITL per token event after the first: the gap before "Rayleigh" and |
| 261 | # the back-to-back delta that follows it. |
| 262 | assert len(output.itl) == 2 |
| 263 | assert output.itl[0] >= GAP_TOKEN * 0.5 |
| 264 | assert output.itl[1] < GAP_TOKEN * 0.5 |
| 265 | |
| 266 | # The identity is exact, so no wall-clock margin is involved. The .done |
| 267 | # family and the GAP_FINALIZE wait before response.completed sit outside |
| 268 | # the request, exactly as the trailing usage chunk does on chat. |
| 269 | assert output.latency - output.ttft == pytest.approx(sum(output.itl), abs=1e-9) |
| 270 | |
| 271 | assert output.start_time > 0.0 |
| 272 | |
| 273 | |
| 274 | def test_usage_is_taken_from_the_terminal_event(): |
| 275 | output = run([PREAMBLE, text_delta("blue"), usage_event(17, 4)]) |
| 276 | |
| 277 | assert output.success |
| 278 | assert output.prompt_len == 17 |
| 279 | assert output.output_tokens == 4 |
| 280 | |
| 281 | |
| 282 | @pytest.mark.parametrize("event_type", ["response.completed", "response.incomplete"]) |
| 283 | def test_incomplete_response_is_a_successful_request(event_type: str): |
| 284 | """Hitting max_output_tokens is the Responses form of finish_reason=length. |
| 285 | |
| 286 | vLLM reports it as `response.completed` carrying `status: "incomplete"`. |
| 287 | The spec's dedicated `response.incomplete` event is accepted as well. |
| 288 | """ |
| 289 | output = run( |
| 290 | [ |
| 291 | PREAMBLE, |
| 292 | text_delta("truncated"), |
| 293 | usage_event(17, 16, event_type=event_type, status="incomplete"), |
| 294 | ] |
| 295 | ) |
| 296 | |
| 297 | assert output.success |
| 298 | assert output.generated_text == "truncated" |
| 299 | assert output.output_tokens == 16 |
| 300 | |
| 301 | |
| 302 | @pytest.mark.parametrize("chunk_size", [1, 3, 7, 64]) |
| 303 | def test_output_survives_transport_chunk_boundaries(chunk_size: int): |
| 304 | """Splits inside the event line, the JSON, or a multibyte character.""" |
| 305 | text = " leading 中 trailing " |
| 306 | output = run( |
| 307 | [PREAMBLE, text_delta(text), usage_event(5, 6)], |
| 308 | chunk_size=chunk_size, |
| 309 | ) |
| 310 | |
| 311 | assert output.success |
| 312 | assert output.generated_text == text |
| 313 | assert output.output_tokens == 6 |
| 314 | |
| 315 | |
| 316 | @pytest.mark.parametrize( |
| 317 | "char", ["\u2028", "\u2029", "\u0085"], ids=["u2028", "u2029", "u0085"] |
| 318 | ) |
| 319 | def test_unicode_line_separators_do_not_truncate_the_payload(char: str): |
| 320 | """The server emits these unescaped and they are legal inside a JSON string. |
| 321 | |
| 322 | `str.splitlines()` treats them as line breaks, which would cut the `data:` |
| 323 | line short and fail the request on valid model output. |
| 324 | """ |
| 325 | text = f"before{char}after" |
| 326 | output = run([PREAMBLE, text_delta(text), usage_event(5, 3)]) |
| 327 | |
| 328 | assert output.success, output.error |
| 329 | assert output.generated_text == text |
| 330 | |
| 331 | |
| 332 | def test_empty_data_line_is_skipped(): |
| 333 | """A bare `data:` line carries no JSON and must not fail the request.""" |
| 334 | output = run([PREAMBLE, b"data:\n\n", text_delta("ok"), usage_event(5, 1)]) |
| 335 | |
| 336 | assert output.success, output.error |
| 337 | assert output.generated_text == "ok" |
| 338 | |
| 339 | |
| 340 | def test_prompt_len_falls_back_when_usage_is_absent(): |
| 341 | """A terminal event without usage keeps the locally computed prompt length.""" |
| 342 | output = run( |
| 343 | [ |
| 344 | PREAMBLE, |
| 345 | text_delta("hi"), |
| 346 | sse("response.completed", {"response": {"status": "completed"}}), |
| 347 | ] |
| 348 | ) |
| 349 | |
| 350 | assert output.success, output.error |
| 351 | assert output.prompt_len == 5 |
| 352 | |
| 353 | |
| 354 | def test_multi_line_data_payload_is_joined_with_newlines(): |
| 355 | """The SSE spec allows one payload to span several `data:` lines.""" |
| 356 | script: list[bytes | float] = [ |
| 357 | PREAMBLE, |
| 358 | b"event: response.output_text.delta\n" |
| 359 | b'data: {"type": "response.output_text.delta",\n' |
| 360 | b'data: "delta": "ok"}\n\n', |
| 361 | usage_event(5, 1), |
| 362 | ] |
| 363 | output = run(script) |
| 364 | |
| 365 | assert output.success, output.error |
| 366 | assert output.generated_text == "ok" |
| 367 | |
| 368 | |
| 369 | def test_done_sentinel_is_skipped(): |
| 370 | """Other Responses servers send `[DONE]`, but vLLM does not.""" |
| 371 | output = run([PREAMBLE, text_delta("ok"), usage_event(5, 1), b"data: [DONE]\n\n"]) |
| 372 | |
| 373 | assert output.success, output.error |
| 374 | assert output.generated_text == "ok" |
| 375 | |
| 376 | |
| 377 | def test_error_before_any_token_reports_the_server_error(): |
| 378 | """The server's message must win over the generic no-TTFT message.""" |
| 379 | output = run( |
| 380 | [PREAMBLE, b'event: error\ndata: {"type": "error", "message": "boom"}\n\n'] |
| 381 | ) |
| 382 | |
| 383 | assert not output.success |
| 384 | assert "boom" in output.error |
| 385 | |
| 386 | |
| 387 | def test_http_error_is_not_reported_as_success(): |
| 388 | output = run([], status=400) |
| 389 | |
| 390 | assert not output.success |
| 391 | assert output.error |
| 392 | |
| 393 | |
| 394 | def test_stream_without_tokens_is_not_reported_as_success(): |
| 395 | """A stream that finishes without generating anything has no TTFT.""" |
| 396 | output = run([PREAMBLE, usage_event(5, 0)]) |
| 397 | |
| 398 | assert not output.success |
| 399 | assert "TTFT" in output.error |
| 400 | |
| 401 | |
| 402 | def test_truncated_stream_is_not_reported_as_success(): |
| 403 | """A stream cut short after a token has no usage and an understated E2EL.""" |
| 404 | output = run([PREAMBLE, text_delta("partial")]) |
| 405 | |
| 406 | assert not output.success |
| 407 | assert "terminal event" in output.error |
| 408 | |
| 409 | |
| 410 | def test_mid_stream_error_is_not_reported_as_success(): |
| 411 | """An error payload mid-stream must not be counted as a success. |
| 412 | |
| 413 | vLLM cannot emit this today: its streaming error path fails to validate |
| 414 | against the event union, so a generation failure truncates the stream |
| 415 | instead. Proxies and other Responses servers do send it. |
| 416 | """ |
| 417 | output = run( |
| 418 | [ |
| 419 | PREAMBLE, |
| 420 | text_delta("partial"), |
| 421 | b'event: error\ndata: {"error": {"message": "boom", "code": 500}}\n\n', |
| 422 | ] |
| 423 | ) |
| 424 | |
| 425 | assert not output.success |
| 426 | assert "boom" in output.error |
| 427 | |
| 428 | |
| 429 | def test_data_only_framing_is_still_parsed(): |
| 430 | """Some proxies drop the ``event:`` line and forward only ``data:``.""" |
| 431 | script: list[bytes | float] = [ |
| 432 | b'data: {"type": "response.output_text.delta", "delta": "ok"}\n\n', |
| 433 | usage_event(5, 1), |
| 434 | ] |
| 435 | output = run(script) |
| 436 | |
| 437 | assert output.success |
| 438 | assert output.generated_text == "ok" |
| 439 | |
| 440 | |
| 441 | def test_endpoint_path_is_validated(): |
| 442 | request_input = RequestFuncInput( |
| 443 | prompt="hi", |
| 444 | api_url="http://localhost:8000/v1/chat/completions", |
| 445 | prompt_len=1, |
| 446 | output_len=1, |
| 447 | model=MODEL, |
| 448 | ) |
| 449 | |
| 450 | async def call() -> None: |
| 451 | async with aiohttp.ClientSession() as session: |
| 452 | await async_request_openai_responses(request_input, session) |
| 453 | |
| 454 | with pytest.raises(ValueError, match="responses"): |
| 455 | asyncio.run(call()) |