| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import logging |
| 7 | from collections.abc import Callable |
| 8 | from typing import Any |
| 9 | |
| 10 | import pytest |
| 11 | import pytest_asyncio |
| 12 | |
| 13 | from tests.utils import RemoteOpenAIServer |
| 14 | |
| 15 | logger = logging.getLogger(__name__) |
| 16 | |
| 17 | BASE_TEST_ENV = { |
| 18 | # The day vLLM said "hello world" on arxiv 🚀 |
| 19 | "VLLM_SYSTEM_START_DATE": "2023-09-12", |
| 20 | } |
| 21 | DEFAULT_MAX_RETRIES = 3 |
| 22 | |
| 23 | |
| 24 | @pytest.fixture |
| 25 | def pairs_of_event_types() -> dict[str, str]: |
| 26 | """Links the 'done' event type with the corresponding 'start' event type. |
| 27 | |
| 28 | This mapping should link all done <-> start events; if tests mean to |
| 29 | restrict the allowed events, they should filter this fixture to avoid |
| 30 | copy + paste errors in the mappings or unexpected KeyErrors due to missing |
| 31 | events. |
| 32 | """ |
| 33 | # fmt: off |
| 34 | event_pairs = { |
| 35 | "response.completed": "response.created", |
| 36 | "response.output_item.done": "response.output_item.added", |
| 37 | "response.content_part.done": "response.content_part.added", |
| 38 | "response.output_text.done": "response.output_text.delta", |
| 39 | "response.reasoning_text.done": "response.reasoning_text.delta", |
| 40 | "response.mcp_call_arguments.done": "response.mcp_call_arguments.delta", |
| 41 | "response.mcp_call.completed": "response.mcp_call.in_progress", |
| 42 | "response.function_call_arguments.done": "response.function_call_arguments.delta", # noqa: E501 |
| 43 | "response.code_interpreter_call_code.done": "response.code_interpreter_call_code.delta", # noqa: E501 |
| 44 | "response.code_interpreter_call.completed": "response.code_interpreter_call.in_progress", # noqa: E501 |
| 45 | "response.web_search_call.completed": "response.web_search_call.in_progress", |
| 46 | } |
| 47 | # fmt: on |
| 48 | return event_pairs |
| 49 | |
| 50 | |
| 51 | async def retry_for_tool_call( |
| 52 | client, |
| 53 | *, |
| 54 | model: str, |
| 55 | expected_tool_type: str, |
| 56 | max_retries: int = DEFAULT_MAX_RETRIES, |
| 57 | **create_kwargs: Any, |
| 58 | ): |
| 59 | """Call ``client.responses.create`` up to *max_retries* times, returning |
| 60 | the first response that contains an output item of *expected_tool_type*. |
| 61 | |
| 62 | Returns the **last** response if none match so the caller's assertions |
| 63 | fire with a clear diagnostic. |
| 64 | """ |
| 65 | last_response = None |
| 66 | for attempt in range(max_retries): |
| 67 | response = await client.responses.create(model=model, **create_kwargs) |
| 68 | last_response = response |
| 69 | if any( |
| 70 | getattr(item, "type", None) == expected_tool_type |
| 71 | for item in response.output |
| 72 | ): |
| 73 | return response |
| 74 | assert last_response is not None |
| 75 | return last_response |
| 76 | |
| 77 | |
| 78 | async def retry_streaming_for( |
| 79 | client, |
| 80 | *, |
| 81 | model: str, |
| 82 | validate_events: Callable[[list], bool], |
| 83 | max_retries: int = DEFAULT_MAX_RETRIES, |
| 84 | **create_kwargs: Any, |
| 85 | ) -> list: |
| 86 | """Call ``client.responses.create(stream=True)`` up to *max_retries* |
| 87 | times, returning the first event list where *validate_events* returns |
| 88 | ``True``. |
| 89 | """ |
| 90 | last_events: list = [] |
| 91 | for attempt in range(max_retries): |
| 92 | stream = await client.responses.create( |
| 93 | model=model, stream=True, **create_kwargs |
| 94 | ) |
| 95 | events: list = [] |
| 96 | async for event in stream: |
| 97 | events.append(event) |
| 98 | last_events = events |
| 99 | if validate_events(events): |
| 100 | return events |
| 101 | return last_events |
| 102 | |
| 103 | |
| 104 | def has_output_type(response, type_name: str) -> bool: |
| 105 | """Return True if *response* has at least one output item of *type_name*.""" |
| 106 | return any(getattr(item, "type", None) == type_name for item in response.output) |
| 107 | |
| 108 | |
| 109 | def events_contain_type(events: list, type_substring: str) -> bool: |
| 110 | """Return True if any event's type contains *type_substring*.""" |
| 111 | return any(type_substring in getattr(e, "type", "") for e in events) |
| 112 | |
| 113 | |
| 114 | def _validate_event_pairing(events: list, pairs_of_event_types: dict[str, str]) -> None: |
| 115 | """Validate that streaming events are properly nested/paired. |
| 116 | |
| 117 | Derives push/pop sets from *pairs_of_event_types* so that every |
| 118 | start/end pair in the dict is handled automatically. |
| 119 | """ |
| 120 | start_events = set(pairs_of_event_types.values()) |
| 121 | end_events = set(pairs_of_event_types.keys()) |
| 122 | |
| 123 | stack: list[str] = [] |
| 124 | for event in events: |
| 125 | etype = event.type |
| 126 | if etype in end_events: |
| 127 | expected_start = pairs_of_event_types[etype] |
| 128 | # Deltas are optional, e.g. a function call with no arguments. |
| 129 | if expected_start.endswith(".delta") and ( |
| 130 | not stack or stack[-1] != expected_start |
| 131 | ): |
| 132 | continue |
| 133 | assert stack and stack[-1] == expected_start, ( |
| 134 | f"Stack mismatch for {etype}: " |
| 135 | f"expected {expected_start}, " |
| 136 | f"got {stack[-1] if stack else '<empty>'}" |
| 137 | ) |
| 138 | stack.pop() |
| 139 | elif etype in start_events: |
| 140 | # Consecutive deltas of the same type share a single stack slot. |
| 141 | if etype.endswith("delta") and stack and stack[-1] == etype: |
| 142 | continue |
| 143 | stack.append(etype) |
| 144 | # else: passthrough event (e.g. response.in_progress, |
| 145 | # web_search_call.searching, code_interpreter_call.interpreting) |
| 146 | assert len(stack) == 0, f"Unclosed events on stack: {stack}" |
| 147 | |
| 148 | |
| 149 | def _validate_event_ordering(events: list) -> None: |
| 150 | """Validate that envelope events appear in the correct positions.""" |
| 151 | assert len(events) >= 2, f"Expected at least 2 events, got {len(events)}" |
| 152 | |
| 153 | # First event must be response.created |
| 154 | assert events[0].type == "response.created", ( |
| 155 | f"First event must be response.created, got {events[0].type}" |
| 156 | ) |
| 157 | # Last event must be response.completed |
| 158 | assert events[-1].type == "response.completed", ( |
| 159 | f"Last event must be response.completed, got {events[-1].type}" |
| 160 | ) |
| 161 | |
| 162 | # response.in_progress, if present, must be the second event |
| 163 | in_progress_indices = [ |
| 164 | i for i, e in enumerate(events) if e.type == "response.in_progress" |
| 165 | ] |
| 166 | if in_progress_indices: |
| 167 | assert in_progress_indices == [1], ( |
| 168 | f"response.in_progress must be the second event, " |
| 169 | f"found at indices {in_progress_indices}" |
| 170 | ) |
| 171 | |
| 172 | # Exactly one created and one completed |
| 173 | created_count = sum(1 for e in events if e.type == "response.created") |
| 174 | completed_count = sum(1 for e in events if e.type == "response.completed") |
| 175 | assert created_count == 1, ( |
| 176 | f"Expected exactly 1 response.created, got {created_count}" |
| 177 | ) |
| 178 | assert completed_count == 1, ( |
| 179 | f"Expected exactly 1 response.completed, got {completed_count}" |
| 180 | ) |
| 181 | |
| 182 | |
| 183 | def _validate_field_consistency(events: list) -> None: |
| 184 | """Validate item_id, output_index, and content_index consistency. |
| 185 | |
| 186 | Tracks the active output item established by ``output_item.added`` |
| 187 | and verifies that all subsequent events for that item carry matching |
| 188 | identifiers until ``output_item.done`` closes it. |
| 189 | """ |
| 190 | _SESSION_EVENTS = { |
| 191 | "response.created", |
| 192 | "response.in_progress", |
| 193 | "response.completed", |
| 194 | } |
| 195 | |
| 196 | active_item_id: str | None = None |
| 197 | active_output_index: int | None = None |
| 198 | last_output_index: int = -1 |
| 199 | active_content_index: int | None = None |
| 200 | |
| 201 | for event in events: |
| 202 | etype = event.type |
| 203 | |
| 204 | if etype in _SESSION_EVENTS: |
| 205 | continue |
| 206 | |
| 207 | # --- output_item.added: opens a new item ------------------ |
| 208 | if etype == "response.output_item.added": |
| 209 | item = getattr(event, "item", None) |
| 210 | output_index = getattr(event, "output_index", None) |
| 211 | |
| 212 | assert item is not None, "output_item.added must have an item" |
| 213 | item_id = getattr(item, "id", None) |
| 214 | assert item_id, "output_item.added item must have an id" |
| 215 | |
| 216 | # output_index must be non-decreasing across items |
| 217 | if output_index is not None: |
| 218 | assert output_index >= last_output_index, ( |
| 219 | f"output_index went backwards: {output_index} < {last_output_index}" |
| 220 | ) |
| 221 | last_output_index = output_index |
| 222 | |
| 223 | active_item_id = item_id |
| 224 | active_output_index = output_index |
| 225 | active_content_index = None |
| 226 | continue |
| 227 | |
| 228 | # --- output_item.done: closes the active item ------------- |
| 229 | if etype == "response.output_item.done": |
| 230 | item = getattr(event, "item", None) |
| 231 | output_index = getattr(event, "output_index", None) |
| 232 | |
| 233 | assert item is not None, "output_item.done must have an item" |
| 234 | done_item_id = getattr(item, "id", None) |
| 235 | |
| 236 | if active_item_id is not None and done_item_id: |
| 237 | assert done_item_id == active_item_id, ( |
| 238 | f"output_item.done item.id mismatch: " |
| 239 | f"expected {active_item_id}, got {done_item_id}" |
| 240 | ) |
| 241 | if active_output_index is not None and output_index is not None: |
| 242 | assert output_index == active_output_index, ( |
| 243 | f"output_item.done output_index mismatch: " |
| 244 | f"expected {active_output_index}, got {output_index}" |
| 245 | ) |
| 246 | |
| 247 | active_item_id = None |
| 248 | active_output_index = None |
| 249 | active_content_index = None |
| 250 | continue |
| 251 | |
| 252 | # --- content_part added: sets content_index |
| 253 | if etype == "response.content_part.added": |
| 254 | _assert_item_fields(event, etype, active_item_id, active_output_index) |
| 255 | content_index = getattr(event, "content_index", None) |
| 256 | if active_content_index is None: |
| 257 | assert content_index == 0, ( |
| 258 | f"{etype} for a new item must start at content_index 0, " |
| 259 | f"got {content_index}" |
| 260 | ) |
| 261 | active_content_index = content_index |
| 262 | continue |
| 263 | |
| 264 | # --- all other item-level events -------------------------- |
| 265 | _assert_item_fields(event, etype, active_item_id, active_output_index) |
| 266 | |
| 267 | # content_index (only meaningful on events that carry it) |
| 268 | content_index = getattr(event, "content_index", None) |
| 269 | if content_index is not None and active_content_index is not None: |
| 270 | assert content_index == active_content_index, ( |
| 271 | f"{etype} content_index mismatch: " |
| 272 | f"expected {active_content_index}, got {content_index}" |
| 273 | ) |
| 274 | |
| 275 | |
| 276 | def _assert_item_fields( |
| 277 | event, |
| 278 | etype: str, |
| 279 | active_item_id: str | None, |
| 280 | active_output_index: int | None, |
| 281 | ) -> None: |
| 282 | """Check that *event*'s item_id and output_index match the active item.""" |
| 283 | event_item_id = getattr(event, "item_id", None) |
| 284 | output_index = getattr(event, "output_index", None) |
| 285 | |
| 286 | if active_item_id is not None and event_item_id is not None: |
| 287 | assert event_item_id == active_item_id, ( |
| 288 | f"{etype} item_id mismatch: expected {active_item_id}, got {event_item_id}" |
| 289 | ) |
| 290 | if active_output_index is not None and output_index is not None: |
| 291 | assert output_index == active_output_index, ( |
| 292 | f"{etype} output_index mismatch: " |
| 293 | f"expected {active_output_index}, got {output_index}" |
| 294 | ) |
| 295 | |
| 296 | |
| 297 | def validate_streaming_event_stack( |
| 298 | events: list, pairs_of_event_types: dict[str, str] |
| 299 | ) -> None: |
| 300 | """Validate streaming events: pairing, ordering, and field consistency. |
| 301 | |
| 302 | Checks three aspects: |
| 303 | 1. **Event pairing** — start/end events are properly nested |
| 304 | (stack-based matching derived from *pairs_of_event_types*). |
| 305 | 2. **Event ordering** — envelope events (``created``, |
| 306 | ``in_progress``, ``completed``) appear at the correct positions. |
| 307 | 3. **Field consistency** — ``item_id``, ``output_index``, and |
| 308 | ``content_index`` are consistent across related events within |
| 309 | each output item's lifecycle. |
| 310 | """ |
| 311 | _validate_event_pairing(events, pairs_of_event_types) |
| 312 | _validate_event_ordering(events) |
| 313 | _validate_field_consistency(events) |
| 314 | |
| 315 | |
| 316 | def log_response_diagnostics( |
| 317 | response, |
| 318 | *, |
| 319 | label: str = "Response Diagnostics", |
| 320 | ) -> dict[str, Any]: |
| 321 | """Extract and log diagnostic info from a Responses API response. |
| 322 | |
| 323 | Logs reasoning, tool-call attempts, MCP items, and output types so |
| 324 | that CI output (``pytest -s`` or ``--log-cli-level=INFO``) gives |
| 325 | full visibility into model behaviour even on passing runs. |
| 326 | |
| 327 | Returns the extracted data so callers can make additional assertions |
| 328 | if needed. |
| 329 | """ |
| 330 | reasoning_texts = [ |
| 331 | text |
| 332 | for item in response.output |
| 333 | if getattr(item, "type", None) == "reasoning" |
| 334 | for content in getattr(item, "content", []) |
| 335 | if (text := getattr(content, "text", None)) |
| 336 | ] |
| 337 | |
| 338 | tool_call_attempts = [ |
| 339 | { |
| 340 | "recipient": msg.get("recipient"), |
| 341 | "channel": msg.get("channel"), |
| 342 | } |
| 343 | for msg in response.output_messages |
| 344 | if (msg.get("recipient") or "").startswith("python") |
| 345 | ] |
| 346 | |
| 347 | mcp_items = [ |
| 348 | { |
| 349 | "name": getattr(item, "name", None), |
| 350 | "status": getattr(item, "status", None), |
| 351 | } |
| 352 | for item in response.output |
| 353 | if getattr(item, "type", None) == "mcp_call" |
| 354 | ] |
| 355 | |
| 356 | output_types = [getattr(o, "type", None) for o in response.output] |
| 357 | |
| 358 | diagnostics = { |
| 359 | "model_attempted_tool_calls": bool(tool_call_attempts), |
| 360 | "tool_call_attempts": tool_call_attempts, |
| 361 | "mcp_items": mcp_items, |
| 362 | "reasoning": reasoning_texts, |
| 363 | "output_text": response.output_text, |
| 364 | "output_types": output_types, |
| 365 | } |
| 366 | |
| 367 | logger.info( |
| 368 | "\n====== %s ======\n%s\n==============================", |
| 369 | label, |
| 370 | json.dumps(diagnostics, indent=2, default=str), |
| 371 | ) |
| 372 | |
| 373 | return diagnostics |
| 374 | |
| 375 | |
| 376 | @pytest.fixture(scope="module") |
| 377 | def default_server_args(): |
| 378 | return [ |
| 379 | "--max-model-len", |
| 380 | "18192", |
| 381 | "--enforce-eager", # For faster startup. |
| 382 | "--enable-auto-tool-choice", |
| 383 | "--structured-outputs-config.backend", |
| 384 | "xgrammar", |
| 385 | "--tool-call-parser", |
| 386 | "hermes", |
| 387 | "--reasoning-parser", |
| 388 | "qwen3", |
| 389 | ] |
| 390 | |
| 391 | |
| 392 | @pytest.fixture(scope="module") |
| 393 | def server_with_store(default_server_args): |
| 394 | with RemoteOpenAIServer( |
| 395 | "Qwen/Qwen3-1.7B", |
| 396 | default_server_args, |
| 397 | env_dict={ |
| 398 | "VLLM_ENABLE_RESPONSES_API_STORE": "1", |
| 399 | "VLLM_SERVER_DEV_MODE": "1", |
| 400 | }, |
| 401 | ) as remote_server: |
| 402 | yield remote_server |
| 403 | |
| 404 | |
| 405 | @pytest_asyncio.fixture |
| 406 | async def client(server_with_store): |
| 407 | async with server_with_store.get_async_client() as async_client: |
| 408 | yield async_client |