Responses support (#32689)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: Harmya Bhatt <harmyacs@gmail.com>
Co-authored-by: harmya <harmya@modal.com>
Co-authored-by: Xinyuan <xinyuan@radixark.com>
Co-authored-by: Xinyuan Tong <xinyuantong.cs@gmail.com>
Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com>

python/sglang/srt/entrypoints/context.py

@@ -79,6 +79,7 @@ class HarmonyContext(ConversationContext):
         self.num_cached_tokens = 0
         self.num_output_tokens = 0
         self.num_reasoning_tokens = 0
+        self.finish_reason = None
 
     def append_output(self, output) -> None:
         if isinstance(output, dict) and "output_ids" in output:
@@ -97,12 +98,20 @@ class HarmonyContext(ConversationContext):
                     self.num_cached_tokens = meta_info["cached_tokens"]
                 if "completion_tokens" in meta_info:
                     self.num_output_tokens += meta_info["completion_tokens"]
+                self._record_finish_reason(meta_info)
 
         else:
             output_msgs = output
 
         self._messages.extend(output_msgs)
 
+    def _record_finish_reason(self, meta_info: dict) -> None:
+        # Last non-null wins: a builtin-tool continuation turn supersedes the
+        # reason recorded for the turn before it.
+        reason = meta_info.get("finish_reason")
+        if reason is not None:
+            self.finish_reason = reason
+
     @property
     def messages(self) -> list:
         return self._messages
@@ -210,6 +219,8 @@ class StreamingHarmonyContext(HarmonyContext):
                 new_token_ids = output_token_ids
                 self.num_processed_tokens += len(output_token_ids)
 
+            self._record_finish_reason(meta_info)
+
             for token_id in new_token_ids:
                 self.parser.process(token_id)
 

python/sglang/srt/entrypoints/harmony_utils.py

@@ -5,7 +5,7 @@
 import datetime
 import logging
 from collections.abc import Iterable
-from typing import Literal, Optional, Union
+from typing import Optional, Union
 
 import orjson
 from openai.types.responses import (
@@ -40,7 +40,10 @@ from openai_harmony import (
     load_harmony_encoding,
 )
 
-from sglang.srt.entrypoints.openai.protocol import ResponseInputOutputItem
+from sglang.srt.entrypoints.openai.protocol import (
+    ReasoningEffortTier,
+    ResponseInputOutputItem,
+)
 from sglang.srt.utils import random_uuid
 
 logger = logging.getLogger(__name__)
@@ -49,6 +52,11 @@ REASONING_EFFORT = {
     "high": ReasoningEffort.HIGH,
     "medium": ReasoningEffort.MEDIUM,
     "low": ReasoningEffort.LOW,
+    # harmony has only these three, so outer tiers clamp inward. "none" is left
+    # out so it falls through to the harmony default.
+    "minimal": ReasoningEffort.LOW,
+    "xhigh": ReasoningEffort.HIGH,
+    "max": ReasoningEffort.HIGH,
 }
 
 _harmony_encoding = None
@@ -63,7 +71,7 @@ def get_encoding():
 
 def get_system_message(
     model_identity: Optional[str] = None,
-    reasoning_effort: Optional[Literal["high", "medium", "low"]] = None,
+    reasoning_effort: Optional[ReasoningEffortTier] = None,
     start_date: Optional[str] = None,
     browser_description: Optional[str] = None,
     python_description: Optional[str] = None,
@@ -72,9 +80,9 @@ def get_system_message(
     if model_identity is not None:
         sys_msg_content = sys_msg_content.with_model_identity(model_identity)
     if reasoning_effort is not None:
-        sys_msg_content = sys_msg_content.with_reasoning_effort(
+        effort = REASONING_EFFORT.get(reasoning_effort)
-            REASONING_EFFORT[reasoning_effort]
+        if effort is not None:
-        )
+            sys_msg_content = sys_msg_content.with_reasoning_effort(effort)
     if start_date is None:
         start_date = datetime.datetime.now().strftime("%Y-%m-%d")
     sys_msg_content = sys_msg_content.with_conversation_start_date(start_date)

python/sglang/srt/entrypoints/openai/protocol.py

@@ -42,12 +42,18 @@ from openai.types.responses import (
     ResponseOutputMessage,
     ResponseOutputText,
     ResponseReasoningItem,
+    ResponseTextConfig,
 )
 from openai.types.responses.response import ToolChoice
+from openai.types.responses.response_format_text_json_schema_config import (
+    ResponseFormatTextJSONSchemaConfig,
+)
+from openai.types.shared.response_format_json_object import ResponseFormatJSONObject
 from pydantic import (
     BaseModel,
     ConfigDict,
     Field,
+    field_serializer,
     field_validator,
     model_serializer,
     model_validator,
@@ -710,6 +716,9 @@ class ToolChoice(BaseModel):
 ReasoningEffortTier = Literal[
     "none", "minimal", "low", "medium", "high", "xhigh", "max"
 ]
+# The typed /v1/responses stream events validate against OpenAI's narrower
+# ``Reasoning.effort``; echoing a tier outside this set kills the stream.
+ECHOABLE_REASONING_EFFORTS = frozenset({"minimal", "low", "medium", "high"})
 # Chat Completions and /v1/tokenize additionally accept a fine-grained float in
 # [0.0, 0.99] as an sglang extension (not part of the OpenAI schema, so the
 # /v1/responses surface deliberately keeps the string tiers only). Single-sourced
@@ -1509,7 +1518,8 @@ class ResponsesRequest(BaseModel):
     store: Optional[bool] = True
     stream: Optional[bool] = False
     temperature: Optional[float] = None
-    tool_choice: Literal["auto", "required", "none"] = "auto"
+    text: Optional[ResponseTextConfig] = None
+    tool_choice: Union[Literal["auto", "required", "none"], Dict[str, Any]] = "auto"
     tools: List[ResponseTool] = Field(default_factory=list)
     top_logprobs: Optional[int] = 0
     top_p: Optional[float] = None
@@ -1517,6 +1527,7 @@ class ResponsesRequest(BaseModel):
     user: Optional[str] = None
 
     # Extra SGLang parameters
+    chat_template_kwargs: Optional[Dict[str, Any]] = None
     request_id: str = Field(
         default_factory=lambda: f"resp_{uuid.uuid4().hex}",
         description="The request_id related to this request. If the caller does not set it, a random uuid will be generated.",
@@ -1548,6 +1559,34 @@ class ResponsesRequest(BaseModel):
         "repetition_penalty": 1.0,
     }
 
+    @model_validator(mode="before")
+    @classmethod
+    def normalize_reasoning_to_thinking(cls, values):
+        """Turn reasoning.effort == "none" into a thinking toggle, as
+        ChatCompletionRequest does. Both keys are set: families differ
+        ("thinking" for deepseek/kimi, "enable_thinking" for qwen3/glm)."""
+        if not isinstance(values, dict):
+            return values
+        # mode="before", so reasoning is still raw: a dict from JSON, or a
+        # built param object when constructed in Python.
+        r = values.get("reasoning")
+        effort = None
+        if isinstance(r, dict):
+            effort = r.get("effort") or r.get("reasoning_effort")
+        elif isinstance(r, ResponseReasoningParam):
+            effort = r.effort
+
+        if effort == "none":
+            existing = values.get("chat_template_kwargs")
+            existing = existing if isinstance(existing, dict) else {}
+            # existing last: an explicit caller value wins.
+            values["chat_template_kwargs"] = {
+                "thinking": False,
+                "enable_thinking": False,
+                **existing,
+            }
+        return values
+
     @model_validator(mode="before")
     @classmethod
     def normalize_responses_input(cls, values):
@@ -1569,11 +1608,17 @@ class ResponsesRequest(BaseModel):
         if not isinstance(item, dict):
             return item
 
+        # an output item replayed into input carries a string id; without this it'd
+        # be read as an item-reference, drop its content, and fail as an empty {}.
+        # input-item ids aren't resolved server-side, so just drop it.
+        if isinstance(item.get("id"), str) and item.get("content") is not None:
+            item = {k: v for k, v in item.items() if k != "id"}
+
         content = item.get("content")
         if not isinstance(content, list):
             return item
 
-        item = item.copy()
+        item = dict(item)
         item["content"] = [
             ResponsesRequest._normalize_content_part_for_validation(part)
             for part in content
@@ -1593,6 +1638,40 @@ class ResponsesRequest(BaseModel):
         part["detail"] = "auto"
         return part
 
+    @staticmethod
+    def _json_schema_from_text_format(
+        text: Optional[ResponseTextConfig],
+    ) -> Optional[str]:
+        """Map a Responses ``text.format`` to a json_schema string, or None when
+        no JSON constraint applies (``text`` and anything unrecognized)."""
+        response_format = text.format if text is not None else None
+        if isinstance(response_format, ResponseFormatJSONObject):
+            return '{"type": "object"}'
+        if not isinstance(response_format, ResponseFormatTextJSONSchemaConfig):
+            return None
+        schema = response_format.schema_
+        return convert_json_schema_to_str(schema) if schema is not None else None
+
+    def is_include_output_logprobs(self) -> bool:
+        return bool(self.include and "message.output_text.logprobs" in self.include)
+
+    def has_json_schema_constraint(self) -> bool:
+        return self._json_schema_from_text_format(self.text) is not None
+
+    def effective_tool_choice(self) -> Union[str, Dict[str, Any]]:
+        """``tool_choice`` reduced to what the server can actually honor: of the
+        object forms only a named ``function`` survives, the rest (web_search,
+        mcp, ...) can't be forced through the tool-call parser."""
+        tool_choice = self.tool_choice
+        if not isinstance(tool_choice, dict):
+            return tool_choice
+        name = tool_choice.get("name") or (tool_choice.get("function") or {}).get(
+            "name"
+        )
+        if tool_choice.get("type") == "function" and name:
+            return {"type": "function", "name": name}
+        return "auto"
+
     def to_sampling_params(
         self,
         default_max_tokens: int,
@@ -1645,6 +1724,10 @@ class ResponsesRequest(BaseModel):
             if key not in params or params[key] is None:
                 params[key] = value
 
+        json_schema = self._json_schema_from_text_format(self.text)
+        if json_schema is not None:
+            params["json_schema"] = json_schema
+
         has_existing_constraints = (
             params.get("regex")
             or params.get("ebnf")
@@ -1655,7 +1738,8 @@ class ResponsesRequest(BaseModel):
             # Refuse rather than silently drop the tool-call grammar.
             raise ValueError(
                 "Cannot combine tool calls with constrained decoding "
-                "(regex / ebnf / structural_tag / json_schema). Remove one."
+                "(text.format / regex / ebnf / structural_tag / json_schema). "
+                "Remove one."
             )
         if tool_call_constraint:
             constraint_type, constraint_value = tool_call_constraint
@@ -1688,10 +1772,12 @@ class ResponsesResponse(BaseModel):
     output: List[
         Union[ResponseOutputItem, ResponseReasoningItem, ResponseFunctionToolCall]
     ] = Field(default_factory=list)
-    status: Literal["queued", "in_progress", "completed", "failed", "cancelled"]
+    status: Literal[
+        "queued", "in_progress", "completed", "incomplete", "failed", "cancelled"
+    ]
     usage: Optional[UsageInfo] = None
     parallel_tool_calls: bool = True
-    tool_choice: str = "auto"
+    tool_choice: Union[str, Dict[str, Any]] = "auto"
     tools: List[ResponseTool] = Field(default_factory=list)
 
     # OpenAI compatibility fields. not all are used at the moment.
@@ -1714,6 +1800,28 @@ class ResponsesResponse(BaseModel):
     user: Optional[str] = None
     metadata: Optional[Dict[str, Any]] = None
 
+    @field_serializer("usage")
+    def _serialize_usage(self, usage: Optional[UsageInfo], _info):
+        """Emit the Responses usage shape, not the chat one UsageInfo carries."""
+        if usage is None:
+            return None
+        cached = (
+            usage.prompt_tokens_details.cached_tokens
+            if usage.prompt_tokens_details
+            else 0
+        )
+        return {
+            "input_tokens": usage.prompt_tokens,
+            "input_tokens_details": {
+                "cached_tokens": cached,
+                # required (no default) in the SDK's InputTokensDetails model
+                "cache_write_tokens": 0,
+            },
+            "output_tokens": usage.completion_tokens or 0,
+            "output_tokens_details": {"reasoning_tokens": usage.reasoning_tokens or 0},
+            "total_tokens": usage.total_tokens,
+        }
+
     @classmethod
     def from_request(
         cls,
@@ -1761,7 +1869,13 @@ class ResponsesResponse(BaseModel):
                     return False
             return True
 
-        text_format = {"format": {"type": "text"}} if _is_text_only(output) else None
+        if request.text is not None:
+            # by_alias keeps the wire key "schema" rather than the attr schema_.
+            text_format = request.text.model_dump(by_alias=True, exclude_none=True)
+        else:
+            text_format = (
+                {"format": {"type": "text"}} if _is_text_only(output) else None
+            )
 
         return cls(
             id=request.request_id,
@@ -1775,16 +1889,23 @@ class ResponsesResponse(BaseModel):
                 if request.parallel_tool_calls is not None
                 else True
             ),
-            tool_choice=request.tool_choice,
+            tool_choice=request.effective_tool_choice(),
             tools=request.tools,
             # fields for parity with v1/responses
             error=None,
-            incomplete_details=None,
+            incomplete_details=(
+                {"reason": "max_output_tokens"} if status == "incomplete" else None
+            ),
             instructions=request.instructions,
             max_output_tokens=request.max_output_tokens,
             previous_response_id=request.previous_response_id,  # TODO(v): ensure this is propagated if retrieved from store
             reasoning={
-                "effort": request.reasoning.effort if request.reasoning else None,
+                "effort": (
+                    request.reasoning.effort
+                    if request.reasoning
+                    and request.reasoning.effort in ECHOABLE_REASONING_EFFORTS
+                    else None
+                ),
                 "summary": None,  # unused
             },
             store=request.store,
@@ -1814,6 +1935,7 @@ class MessageProcessingResult:
     modalities: List[str]
     stop: List[str]
     tool_call_constraint: Optional[ToolCallConstraint] = None
+    skip_special_tokens: bool = True
     require_reasoning: bool = False
 
 

python/sglang/srt/entrypoints/openai/serving_responses.py

@@ -23,6 +23,7 @@ from openai.types.responses import (
     ResponseReasoningItem,
 )
 from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
+from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
 from openai.types.responses.response_reasoning_item import (
     Content as ResponseReasoningTextContent,
 )
@@ -67,6 +68,7 @@ from sglang.srt.entrypoints.openai.protocol import (
 )
 from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat
 from sglang.srt.entrypoints.openai.tool_server import MCPToolServer, ToolServer
+from sglang.srt.entrypoints.openai.utils import to_openai_style_logprobs
 from sglang.srt.function_call.function_call_parser import FunctionCallParser
 from sglang.srt.function_call.json_array_parser import JsonArrayParser
 from sglang.srt.managers.io_struct import GenerateReqInput
@@ -84,6 +86,54 @@ class _MediaInputValidationError(ValueError):
     pass
 
 
+def _build_output_text_logprobs(meta_info: dict) -> list[Logprob]:
+    """Reshape decoded ``meta_info`` logprobs into the Responses logprob type,
+    covering every generated token."""
+    decoded = to_openai_style_logprobs(
+        output_token_logprobs=meta_info.get("output_token_logprobs"),
+        output_top_logprobs=meta_info.get("output_top_logprobs"),
+    )
+    top_lists = decoded.top_logprobs or []
+    logprobs: list[Logprob] = []
+    for index, (token, logprob) in enumerate(
+        zip(decoded.tokens, decoded.token_logprobs)
+    ):
+        top_entry = top_lists[index] if index < len(top_lists) else None
+        top_logprobs = [
+            LogprobTopLogprob(
+                token=top_token,
+                logprob=top_logprob,
+                bytes=list(top_token.encode("utf-8")),
+            )
+            for top_token, top_logprob in (top_entry or {}).items()
+        ]
+        logprobs.append(
+            Logprob(
+                token=token,
+                logprob=logprob,
+                bytes=list(token.encode("utf-8")),
+                top_logprobs=top_logprobs,
+            )
+        )
+    return logprobs
+
+
+def _should_emit_normal_text_as_message(
+    text: str, *, any_tool_call_in_progress: bool
+) -> bool:
+    """Whether ``text`` should open / extend a user-visible message item.
+
+    qwen3-coder separates adjacent tool-call blocks with ``\\n``, which the
+    streaming detector cannot tell from real content -- so whitespace arriving
+    while a call is open is treated as a separator.
+    """
+    if not text:
+        return False
+    if any_tool_call_in_progress and not text.strip():
+        return False
+    return True
+
+
 class OpenAIServingResponses(OpenAIServingChat):
     """Handler for /v1/responses requests"""
 
@@ -202,6 +252,23 @@ class OpenAIServingResponses(OpenAIServingChat):
                 'type="function"; other built-in tool types cannot be forced.'
             )
 
+        # harmony emits raw tokens; per-token logprobs aren't wired there.
+        if self.use_harmony and request.is_include_output_logprobs():
+            return self.create_error_response(
+                "logprobs are not supported with gpt-oss models", param="logprobs"
+            )
+        # streaming skips the logprobs build path; reject so the include doesn't silently no-op.
+        if request.stream and request.is_include_output_logprobs():
+            return self.create_error_response(
+                "logprobs are not supported in streaming mode", param="logprobs"
+            )
+        # harmony output opens with <|channel|>analysis<|message|>, so a whole-output
+        # json_schema forces "{" at the first token and the harmony parse then fails.
+        if self.use_harmony and request.has_json_schema_constraint():
+            return self.create_error_response(
+                "structured output (text.format) is not supported with gpt-oss models",
+                param="text",
+            )
         if (
             self.use_harmony
             and self._has_response_tool(request, "web_search", "web_search_preview")
@@ -328,6 +395,13 @@ class OpenAIServingResponses(OpenAIServingChat):
                         ),
                     )
 
+                    # _process_messages set skip_special_tokens on a chat_request
+                    # we then discard, so re-apply it to the engine sampling dict.
+                    if processed_messages is not None and (
+                        not processed_messages.skip_special_tokens
+                    ):
+                        sampling_params["skip_special_tokens"] = False
+
                     context: ConversationContext
                     if self.use_harmony:
                         if request.stream:
@@ -343,8 +417,20 @@ class OpenAIServingResponses(OpenAIServingChat):
                     else:
                         prompt_kwargs = {"input_ids": engine_prompt}
 
+                    logprob_kwargs = (
+                        {
+                            "return_logprob": True,
+                            "logprob_start_len": -1,
+                            "top_logprobs_num": request.top_logprobs or 0,
+                            "return_text_in_logprobs": True,
+                        }
+                        if request.is_include_output_logprobs()
+                        else {}
+                    )
+
                     adapted_request = GenerateReqInput(
                         **prompt_kwargs,
+                        **logprob_kwargs,
                         image_data=(
                             processed_messages.image_data
                             if processed_messages
@@ -370,7 +456,8 @@ class OpenAIServingResponses(OpenAIServingChat):
                         rid=request.request_id,
                         session_id=request.session_id,
                         extra_key=self._compute_extra_key(request),
-                        background=request.background,
+                        # background+stream streams on this connection, so don't detach.
+                        background=request.background and not request.stream,
                         require_reasoning=require_reasoning,
                     )
 
@@ -394,7 +481,7 @@ class OpenAIServingResponses(OpenAIServingChat):
             if request.store:
                 self.msg_store[request.request_id] = messages
 
-            if request.background:
+            if request.background and not request.stream:
                 created_time = int(time.time())
                 response = ResponsesResponse.from_request(
                     request,
@@ -484,7 +571,11 @@ class OpenAIServingResponses(OpenAIServingChat):
             messages=messages,
             stream=request.stream,
             tools=chat_tools or None,
-            tool_choice=request.tool_choice if chat_tools else "none",
+            tool_choice=(
+                self._chat_tool_choice(request.effective_tool_choice())
+                if chat_tools
+                else "none"
+            ),
             parallel_tool_calls=(
                 request.parallel_tool_calls
                 if request.parallel_tool_calls is not None
@@ -492,6 +583,7 @@ class OpenAIServingResponses(OpenAIServingChat):
             ),
             stop=request.stop,
             reasoning_effort=(request.reasoning.effort if request.reasoning else None),
+            chat_template_kwargs=request.chat_template_kwargs,
         )
 
         media_error = self._validate_media_content(chat_request)
@@ -500,6 +592,7 @@ class OpenAIServingResponses(OpenAIServingChat):
 
         is_multimodal = self.tokenizer_manager.model_config.is_multimodal
         processed_messages = self._process_messages(chat_request, is_multimodal)
+        processed_messages.skip_special_tokens = chat_request.skip_special_tokens
 
         if is_multimodal:
             request_prompts = [processed_messages.prompt]
@@ -548,6 +641,7 @@ class OpenAIServingResponses(OpenAIServingChat):
         except ValueError as e:
             return self.create_error_response(str(e))
 
+        status = "completed"
         if self.use_harmony:
             assert isinstance(context, HarmonyContext)
             output = self._make_response_output_items_with_harmony(context)
@@ -556,19 +650,12 @@ class OpenAIServingResponses(OpenAIServingChat):
             num_generated_tokens = context.num_output_tokens
             num_cached_tokens = context.num_cached_tokens
             num_reasoning_tokens = context.num_reasoning_tokens
+            status = self._status_from_finish_reason(context.finish_reason)
         else:
             assert isinstance(context, SimpleContext)
             final_res = context.last_output
             assert final_res is not None
 
-            output = self._make_response_output_items(
-                request,
-                final_res["text"],
-                tokenizer,
-                require_reasoning=require_reasoning,
-            )
-
-            # Calculate usage from actual output
             num_reasoning_tokens = 0
             meta_info = None
             if isinstance(final_res, dict) and isinstance(
@@ -578,11 +665,25 @@ class OpenAIServingResponses(OpenAIServingChat):
             elif hasattr(final_res, "meta_info"):
                 meta_info = final_res.meta_info
 
+            output_logprobs = (
+                _build_output_text_logprobs(meta_info)
+                if request.is_include_output_logprobs() and isinstance(meta_info, dict)
+                else None
+            )
+            output = self._make_response_output_items(
+                request,
+                final_res["text"],
+                tokenizer,
+                output_logprobs=output_logprobs,
+                require_reasoning=require_reasoning,
+            )
+
             if meta_info is not None:
                 num_prompt_tokens = meta_info.get("prompt_tokens", 0)
                 num_generated_tokens = meta_info.get("completion_tokens", 0)
                 num_cached_tokens = meta_info.get("cached_tokens", 0)
                 num_reasoning_tokens = meta_info.get("reasoning_tokens", 0)
+                status = self._status_from_finish_reason(meta_info.get("finish_reason"))
             elif isinstance(final_res, dict) and (
                 final_res.get("prompt_token_ids") is not None
                 or final_res.get("output_ids") is not None
@@ -630,7 +731,7 @@ class OpenAIServingResponses(OpenAIServingChat):
             model_name=model_name,
             created_time=created_time,
             output=output,
-            status="completed",
+            status=status,
             usage=usage,
         )
 
@@ -647,9 +748,27 @@ class OpenAIServingResponses(OpenAIServingChat):
     def _wants_reasoning_summary(request: ResponsesRequest) -> bool:
         return request.reasoning is not None and request.reasoning.summary is not None
 
+    @staticmethod
+    def _status_from_finish_reason(finish_reason: Any) -> str:
+        """Only a length-capped generation is ``incomplete``; anything that got
+        here otherwise finished normally."""
+        reason = None
+        if isinstance(finish_reason, dict):
+            reason = finish_reason.get("type")
+        elif isinstance(finish_reason, str):
+            reason = finish_reason
+        return "incomplete" if reason == "length" else "completed"
+
     def _is_thinking_enabled_for_request(self, request: ResponsesRequest) -> bool:
         if not self.reasoning_parser:
             return False
+        # an explicit toggle wins; the key differs by family (enable_thinking vs thinking).
+        ctk = request.chat_template_kwargs or {}
+        thinking_toggles = (ctk.get("enable_thinking"), ctk.get("thinking"))
+        if any(toggle is False for toggle in thinking_toggles):
+            return False
+        if any(toggle is True for toggle in thinking_toggles):
+            return True
         effort = request.reasoning.effort if request.reasoning is not None else None
         if self.reasoning_parser == "hunyuan":
             return effort not in (None, "none", "no_think")
@@ -685,6 +804,7 @@ class OpenAIServingResponses(OpenAIServingChat):
         request: ResponsesRequest,
         final_output: Any,
         tokenizer: Any,
+        output_logprobs: Optional[list] = None,
         *,
         require_reasoning: bool,
     ):
@@ -692,7 +812,11 @@ class OpenAIServingResponses(OpenAIServingChat):
             reasoning_parser = ReasoningParser(
                 model_type=self.reasoning_parser,
                 stream_reasoning=False,
-                force_reasoning=require_reasoning,
+                # A template that prefills <think> forces the parser open even
+                # when the request itself did not ask for reasoning, same as chat.
+                force_reasoning=(
+                    self.template_manager.force_reasoning or require_reasoning
+                ),
                 request=request,
                 tokenizer=self.tokenizer_manager.tokenizer,
             )
@@ -794,7 +918,8 @@ class OpenAIServingResponses(OpenAIServingChat):
                 text=content,
                 annotations=[],  # TODO
                 type="output_text",
-                logprobs=None,  # TODO
+                # logprobs cover all generated tokens, not just the stripped content.
+                logprobs=output_logprobs,
             )
             message = ResponseOutputMessage(
                 id=f"msg_{random_uuid()}",
@@ -821,6 +946,14 @@ class OpenAIServingResponses(OpenAIServingChat):
             output_items.extend(last_items)
         return output_items
 
+    @staticmethod
+    def _chat_tool_choice(tool_choice: Any) -> Any:
+        """Nest an ``effective_tool_choice()`` result the way chat expects:
+        ``{"type":"function","name":X}`` -> ``{...,"function":{"name":X}}``."""
+        if not isinstance(tool_choice, dict):
+            return tool_choice
+        return {"type": "function", "function": {"name": tool_choice["name"]}}
+
     @staticmethod
     def _response_tools_to_chat_tools(request: ResponsesRequest) -> list[Tool]:
         # Only ``function`` tools flow to chat; built-ins go through harmony.
@@ -929,10 +1062,15 @@ class OpenAIServingResponses(OpenAIServingChat):
                 ],
             }
         if msg_type == "function_call_output":
+            # ``output`` may be a string or an array of content parts (OpenAI
+            # allows both); the chat tool message needs a string, so flatten.
+            out = message.get("output", "")
+            if isinstance(out, list):
+                out = "".join(p.get("text", "") for p in out if isinstance(p, dict))
             return {
                 "role": "tool",
                 "tool_call_id": message.get("call_id"),
-                "content": message.get("output", ""),
+                "content": out,
             }
         # Reasoning items render as {role: assistant, reasoning_content};
         # empty ones drop instead of injecting an empty assistant block.
@@ -1274,10 +1412,8 @@ class OpenAIServingResponses(OpenAIServingChat):
 
             prev_status = response.status
             if prev_status not in ("queued", "in_progress"):
-                return self.create_error_response(
+                # already terminal; a second cancel is a no-op, return as-is.
-                    err_type="invalid_request_error",
+                return response
-                    message="Cannot cancel a synchronous response.",
-                )
 
             # Update the status to "cancelled"
             response.status = "cancelled"
@@ -1374,7 +1510,6 @@ class OpenAIServingResponses(OpenAIServingChat):
         )
 
         async for ctx in result_generator:
-
             # Only process context objects that implement the `is_expecting_start()` method,
             # which indicates they support per-turn streaming (e.g., StreamingHarmonyContext).
             # Contexts without this method are skipped, as they do not represent a new turn
@@ -1731,26 +1866,10 @@ class OpenAIServingResponses(OpenAIServingChat):
             created_time=created_time,
             require_reasoning=require_reasoning,
         )
-        # Convert final_response to the format expected by ResponseCompletedEvent
         response_dict = final_response.model_dump()
         # OpenAI SDK's Tool union may not know extended types; drop echo.
         response_dict["tools"] = []
 
-        # Convert UsageInfo to ResponseUsage format
-        if response_dict.get("usage"):
-            usage_info = response_dict["usage"]
-            response_dict["usage"] = {
-                "input_tokens": usage_info.get("prompt_tokens", 0),
-                "input_tokens_details": {
-                    "cached_tokens": usage_info.get("cached_tokens", 0)
-                },
-                "output_tokens": usage_info.get("completion_tokens", 0),
-                "output_tokens_details": {
-                    "reasoning_tokens": usage_info.get("reasoning_tokens", 0)
-                },
-                "total_tokens": usage_info.get("total_tokens", 0),
-            }
-
         yield _send_event(
             openai_responses_types.ResponseCompletedEvent(
                 type="response.completed",
@@ -1851,7 +1970,11 @@ class OpenAIServingResponses(OpenAIServingChat):
             reasoning_parser_obj = ReasoningParser(
                 model_type=self.reasoning_parser,
                 stream_reasoning=True,
-                force_reasoning=require_reasoning,
+                # A template that prefills <think> forces the parser open even
+                # when the request itself did not ask for reasoning, same as chat.
+                force_reasoning=(
+                    self.template_manager.force_reasoning or require_reasoning
+                ),
                 request=request,
                 tokenizer=self.tokenizer_manager.tokenizer,
             )
@@ -2166,9 +2289,75 @@ class OpenAIServingResponses(OpenAIServingChat):
                 else:
                     normal_text, tool_calls = delta, []
 
-                # Close any open tool-call item before opening a message so
+                def _emit_tool_calls(calls):
-                # ``output_item.done`` lands before the next ``added``.
+                    nonlocal current_output_index
-                if normal_text:
+                    if calls:
+                        if reasoning_state["open"]:
+                            for ev in _close_reasoning_item():
+                                yield ev
+                        if message_state["open"]:
+                            for ev in _close_message_item():
+                                yield ev
+
+                    for call in calls:
+                        tool_index = call.tool_index
+                        state = tool_call_states.get(tool_index)
+                        if state is None or state.get("done"):
+                            # Close other open calls first, so their
+                            # output_item.done precedes the next added.
+                            for other_index in list(tool_call_states):
+                                if other_index != tool_index:
+                                    for ev in _close_tool_call_state(other_index):
+                                        yield ev
+                            current_output_index += 1
+                            item_id = f"fc_{random_uuid()[:8]}"
+                            call_id = f"call_{random_uuid()[:24]}"
+                            state = {
+                                "item_id": item_id,
+                                "call_id": call_id,
+                                "output_index": current_output_index,
+                                "name": call.name or "",
+                                "arguments": "",
+                                "added": False,
+                                "done": False,
+                            }
+                            tool_call_states[tool_index] = state
+                        if not state["added"]:
+                            state["added"] = True
+                            yield _send_event(
+                                openai_responses_types.ResponseOutputItemAddedEvent(
+                                    type="response.output_item.added",
+                                    sequence_number=-1,
+                                    output_index=state["output_index"],
+                                    item=ResponseFunctionToolCall(
+                                        arguments="",
+                                        call_id=state["call_id"],
+                                        name=state["name"],
+                                        type="function_call",
+                                        id=state["item_id"],
+                                        status="in_progress",
+                                    ),
+                                )
+                            )
+                        if call.parameters:
+                            state["arguments"] += call.parameters
+                            yield _send_event(
+                                openai_responses_types.ResponseFunctionCallArgumentsDeltaEvent(
+                                    type="response.function_call_arguments.delta",
+                                    sequence_number=-1,
+                                    item_id=state["item_id"],
+                                    output_index=state["output_index"],
+                                    delta=call.parameters,
+                                )
+                            )
+
+                def _emit_normal_text():
+                    if normal_text and _should_emit_normal_text_as_message(
+                        normal_text,
+                        any_tool_call_in_progress=any(
+                            not s.get("done") for s in tool_call_states.values()
+                        ),
+                    ):
                         if reasoning_state["open"]:
                             for ev in _close_reasoning_item():
                                 yield ev
@@ -2219,65 +2408,23 @@ class OpenAIServingResponses(OpenAIServingChat):
                             )
                         )
 
-                if not tool_calls:
+                # The parser's (text, calls) tuple is unordered, but positions
-                    continue
+                # are recoverable: continuing arguments precede this delta's
+                # text, a newly opened call follows it. Classify first --
+                # emitting mutates tool_call_states.
+                def _is_continuing(call):
+                    state = tool_call_states.get(call.tool_index)
+                    return state is not None and not state.get("done")
 
-                if reasoning_state["open"]:
+                continuing = [c for c in tool_calls if _is_continuing(c)]
-                    for ev in _close_reasoning_item():
+                opening = [c for c in tool_calls if not _is_continuing(c)]
-                        yield ev
-                if message_state["open"]:
-                    for ev in _close_message_item():
-                        yield ev
 
-                for call in tool_calls:
+                for ev in _emit_tool_calls(continuing):
-                    tool_index = call.tool_index
+                    yield ev
-                    state = tool_call_states.get(tool_index)
+                for ev in _emit_normal_text():
-                    if state is None or state.get("done"):
+                    yield ev
-                        current_output_index += 1
+                for ev in _emit_tool_calls(opening):
-                        item_id = f"fc_{random_uuid()[:8]}"
+                    yield ev
-                        call_id = f"call_{random_uuid()[:24]}"
-                        state = {
-                            "item_id": item_id,
-                            "call_id": call_id,
-                            "output_index": current_output_index,
-                            "name": call.name or "",
-                            "arguments": "",
-                            "added": False,
-                            "done": False,
-                        }
-                        tool_call_states[tool_index] = state
-                    if not state["added"]:
-                        state["added"] = True
-                        # Capture ``call.name`` before the ``added`` event so
-                        # the name is set on the first emitted item.
-                        if call.name and not state["name"]:
-                            state["name"] = call.name
-                        yield _send_event(
-                            openai_responses_types.ResponseOutputItemAddedEvent(
-                                type="response.output_item.added",
-                                sequence_number=-1,
-                                output_index=state["output_index"],
-                                item=ResponseFunctionToolCall(
-                                    arguments="",
-                                    call_id=state["call_id"],
-                                    name=state["name"],
-                                    type="function_call",
-                                    id=state["item_id"],
-                                    status="in_progress",
-                                ),
-                            )
-                        )
-                    if call.parameters:
-                        state["arguments"] += call.parameters
-                        yield _send_event(
-                            openai_responses_types.ResponseFunctionCallArgumentsDeltaEvent(
-                                type="response.function_call_arguments.delta",
-                                sequence_number=-1,
-                                item_id=state["item_id"],
-                                output_index=state["output_index"],
-                                delta=call.parameters,
-                            )
-                        )
         except Exception:
             logger.exception("Error while streaming /v1/responses")
             failed = _sanitize_response_dict(
@@ -2328,7 +2475,7 @@ class OpenAIServingResponses(OpenAIServingChat):
             model_name=model_name,
             created_time=created_time,
             output=final_output_items,
-            status="completed",
+            status=self._status_from_finish_reason(finish_reason),
             usage=usage,
         )
         if request.store:
@@ -2338,19 +2485,6 @@ class OpenAIServingResponses(OpenAIServingChat):
                     self.response_store[final_response.id] = final_response
 
         response_dict = _sanitize_response_dict(final_response.model_dump())
-        if response_dict.get("usage"):
-            usage_info = response_dict["usage"]
-            response_dict["usage"] = {
-                "input_tokens": usage_info.get("prompt_tokens", 0),
-                "input_tokens_details": {
-                    "cached_tokens": cached_tokens,
-                },
-                "output_tokens": usage_info.get("completion_tokens", 0),
-                "output_tokens_details": {
-                    "reasoning_tokens": reasoning_tokens_meta,
-                },
-                "total_tokens": usage_info.get("total_tokens", 0),
-            }
 
         yield _send_event(
             openai_responses_types.ResponseCompletedEvent(

test/registered/openai_server/basic/test_openai_server.py

@@ -786,19 +786,23 @@ class TestOpenAIServerv1Responses(CustomTestCase):
         assert isinstance(resp.output, list)
         assert resp.status in (
             "completed",
+            "incomplete",
             "in_progress",
             "queued",
             "failed",
             "cancelled",
         )
-        if resp.status == "completed":
+        if resp.status in ("completed", "incomplete"):
             assert resp.usage is not None
-            assert resp.usage.prompt_tokens >= 0
+            assert resp.usage.input_tokens >= 0
-            assert resp.usage.completion_tokens >= 0
+            assert resp.usage.output_tokens >= 0
             assert resp.usage.total_tokens >= 0
         if hasattr(resp, "error"):
             assert resp.error is None
         if hasattr(resp, "incomplete_details"):
+            if resp.status == "incomplete":
+                assert resp.incomplete_details.reason == "max_output_tokens"
+            else:
                 assert resp.incomplete_details is None
         if getattr(resp, "text", None):
             fmt = resp.text.get("format") if isinstance(resp.text, dict) else None
@@ -818,8 +822,8 @@ class TestOpenAIServerv1Responses(CustomTestCase):
 
     def test_response_completion(self):
         resp = self.run_response(temperature=0, max_output_tokens=16)
-        assert resp.status in ("completed", "in_progress", "queued")
+        assert resp.status in ("completed", "incomplete", "in_progress", "queued")
-        if resp.status == "completed":
+        if resp.status in ("completed", "incomplete"):
             assert resp.usage is not None
             assert resp.usage.total_tokens >= 0
 
@@ -900,8 +904,9 @@ class TestOpenAIServerv1Responses(CustomTestCase):
         self.assertEqual(body.get("object"), "response")
         self.assertIn("output", body)
         self.assertIn("status", body)
-        if "usage" in body:
+        self.assertIn("usage", body)
-            self.assertIn("prompt_tokens", body["usage"])
+        self.assertIn("input_tokens", body["usage"])
+        self.assertIn("output_tokens", body["usage"])
         self.assertIn("total_tokens", body["usage"])
 
     def test_response_prefill(self):

test/registered/unit/entrypoints/openai/test_responses_protocol.py

@@ -1,14 +1,33 @@
+import json
 import unittest
 
 from utils import make_serving  # noqa: F401 — bootstrap import
 
-from sglang.srt.entrypoints.openai.protocol import ResponsesRequest, UsageInfo
+from sglang.srt.entrypoints.openai.protocol import (
+    PromptTokensDetails,
+    ResponsesRequest,
+    ResponsesResponse,
+    UsageInfo,
+)
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
 
-register_cpu_ci(est_time=4, suite="base-a-test-cpu")
+register_cpu_ci(est_time=7, suite="base-a-test-cpu")
 
 
-class ResponsesRequestTestCase(unittest.TestCase):
+def _in_progress_response(request: ResponsesRequest) -> ResponsesResponse:
+    return ResponsesResponse.from_request(
+        request,
+        sampling_params={},
+        model_name="x",
+        created_time=0,
+        output=[],
+        status="in_progress",
+        usage=None,
+    )
+
+
+class ResponsesRequestTestCase(CustomTestCase):
     def test_function_tool_accepted(self):
         request = ResponsesRequest(
             model="x",
@@ -88,7 +107,7 @@ class ResponsesRequestTestCase(unittest.TestCase):
         self.assertEqual(request.tools[0].tools[0]["name"], "apply_patch")
 
 
-class ResponsesSamplingParamsTestCase(unittest.TestCase):
+class ResponsesSamplingParamsTestCase(CustomTestCase):
     def test_processed_stop_and_tool_constraint_propagate(self):
         request = ResponsesRequest(model="x", input="call the tool", store=False)
         params = request.to_sampling_params(
@@ -122,11 +141,118 @@ class ResponsesSamplingParamsTestCase(unittest.TestCase):
         )
         self.assertEqual(params["structural_tag"], '{"type": "structural_tag"}')
 
+    def test_text_format_maps_to_json_schema_constraint(self):
+        schema = {"type": "object", "properties": {"age": {"type": "integer"}}}
+        for fmt, expected in [
+            ({"type": "json_schema", "name": "p", "schema": schema}, schema),
+            ({"type": "json_object"}, {"type": "object"}),
+        ]:
+            req = ResponsesRequest(
+                model="x", input="hi", store=False, text={"format": fmt}
+            )
+            params = req.to_sampling_params(default_max_tokens=128, default_params={})
+            self.assertEqual(json.loads(params["json_schema"]), expected, fmt)
+        plain = ResponsesRequest(
+            model="x", input="hi", store=False, text={"format": {"type": "text"}}
+        ).to_sampling_params(default_max_tokens=128, default_params={})
+        self.assertNotIn("json_schema", plain)
+
+    def test_text_format_conflicts_with_tool_constraint(self):
+        request = ResponsesRequest(
+            model="x",
+            input="hi",
+            store=False,
+            text={
+                "format": {
+                    "type": "json_schema",
+                    "name": "p",
+                    "schema": {"type": "object"},
+                }
+            },
+        )
+        # The message must name text.format: it is the only source of the
+        # conflict a /v1/responses caller can actually set.
+        with self.assertRaisesRegex(ValueError, r"text\.format"):
+            request.to_sampling_params(
+                default_max_tokens=128,
+                default_params={},
+                tool_call_constraint=("json_schema", {"type": "object"}),
+            )
+
+
+class IncludeOutputLogprobsTestCase(CustomTestCase):
+    def test_detected_only_for_logprobs_include(self):
+        def has(include):
+            return ResponsesRequest(
+                model="x", input="hi", store=False, include=include
+            ).is_include_output_logprobs()
+
+        self.assertTrue(has(["message.output_text.logprobs"]))
+        self.assertFalse(has(None))
+        self.assertFalse(has(["reasoning.encrypted_content"]))
+
+
+class ThinkingControlTestCase(CustomTestCase):
+    def test_effort_none_disables_thinking(self):
+        ctk = ResponsesRequest(
+            model="x", input="hi", store=False, reasoning={"effort": "none"}
+        ).chat_template_kwargs
+        self.assertEqual((ctk["enable_thinking"], ctk["thinking"]), (False, False))
+
+    def test_thinking_untouched_otherwise(self):
+        # grammar-constrained requests keep thinking on; ReasonerGrammarBackend
+        # defers the grammar past </think> so the two coexist.
+        for kw in (
+            {"reasoning": {"effort": "medium"}},
+            {"text": {"format": {"type": "text"}}},
+            {
+                "text": {
+                    "format": {
+                        "type": "json_schema",
+                        "name": "p",
+                        "schema": {"type": "object"},
+                    }
+                }
+            },
+            {"text": {"format": {"type": "json_object"}}},
+            {"tool_choice": "required"},
+            {"tool_choice": {"type": "function", "name": "f"}},
+            {},
+        ):
+            req = ResponsesRequest(model="x", input="hi", store=False, **kw)
+            self.assertIsNone(req.chat_template_kwargs, kw)
+
+    def test_explicit_chat_template_kwargs_preserved(self):
+        req = ResponsesRequest(
+            model="x",
+            input="hi",
+            store=False,
+            chat_template_kwargs={"enable_thinking": True},
+        )
+        self.assertTrue(req.chat_template_kwargs["enable_thinking"])
+
+
+class ResponsesResponseFromRequestTestCase(CustomTestCase):
+    def test_requested_text_format_is_echoed(self):
+        schema = {"type": "object", "properties": {"x": {"type": "integer"}}}
+        request = ResponsesRequest(
+            model="x",
+            input="hi",
+            store=False,
+            text={"format": {"type": "json_schema", "name": "p", "schema": schema}},
+        )
+        response = ResponsesResponse.from_request(
+            request,
+            sampling_params={},
+            model_name="x",
+            created_time=0,
+            output=[],
+            status="completed",
+            usage=UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2),
+        )
+        self.assertEqual(response.text["format"]["type"], "json_schema")
 
-class ResponsesResponseFromRequestTestCase(unittest.TestCase):
     def test_parallel_tool_calls_false_preserved(self):
-        from sglang.srt.entrypoints.openai.protocol import ResponsesResponse
-
         request = ResponsesRequest(
             model="x", input="hi", parallel_tool_calls=False, store=False
         )
@@ -141,6 +267,144 @@ class ResponsesResponseFromRequestTestCase(unittest.TestCase):
         )
         self.assertFalse(response.parallel_tool_calls)
 
+    def test_incomplete_status_sets_incomplete_details(self):
+        request = ResponsesRequest(model="x", input="hi", store=False)
+        incomplete = ResponsesResponse.from_request(
+            request,
+            sampling_params={},
+            model_name="x",
+            created_time=0,
+            output=[],
+            status="incomplete",
+            usage=UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2),
+        )
+        self.assertEqual(incomplete.status, "incomplete")
+        self.assertEqual(incomplete.incomplete_details, {"reason": "max_output_tokens"})
+
+        completed = ResponsesResponse.from_request(
+            request,
+            sampling_params={},
+            model_name="x",
+            created_time=0,
+            output=[],
+            status="completed",
+            usage=UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2),
+        )
+        self.assertIsNone(completed.incomplete_details)
+
+    def test_usage_serialized_in_responses_shape(self):
+        request = ResponsesRequest(model="x", input="hi", store=False)
+        resp = ResponsesResponse.from_request(
+            request,
+            sampling_params={},
+            model_name="x",
+            created_time=0,
+            output=[],
+            status="completed",
+            usage=UsageInfo(
+                prompt_tokens=11,
+                completion_tokens=102,
+                total_tokens=113,
+                reasoning_tokens=7,
+                prompt_tokens_details=PromptTokensDetails(cached_tokens=3),
+            ),
+        )
+        usage = resp.model_dump()["usage"]
+        self.assertEqual(usage["input_tokens"], 11)
+        self.assertEqual(usage["output_tokens"], 102)
+        self.assertEqual(usage["total_tokens"], 113)
+        self.assertEqual(usage["output_tokens_details"]["reasoning_tokens"], 7)
+        self.assertEqual(usage["input_tokens_details"]["cached_tokens"], 3)
+        # Chat-style keys must be gone.
+        self.assertNotIn("prompt_tokens", usage)
+        self.assertNotIn("completion_tokens", usage)
+
+    def test_only_sdk_known_efforts_echoed_so_streaming_event_validates(self):
+        import openai.types.responses as ort
+
+        # OpenAI's Reasoning.effort literal is narrower than our tier list:
+        # "none" is a request-side extension and xhigh/max postdate it. Echoing
+        # one of those raises a ValidationError in the typed event below, which
+        # the stream generator builds before its try block -- the connection
+        # then dies without a single SSE byte.
+        for effort in ("none", "minimal", "low", "medium", "high", "xhigh", "max"):
+            resp = _in_progress_response(
+                ResponsesRequest(
+                    model="x", input="hi", store=False, reasoning={"effort": effort}
+                )
+            )
+            expected = (
+                effort if effort in ("minimal", "low", "medium", "high") else None
+            )
+            self.assertEqual(resp.reasoning["effort"], expected, effort)
+            ort.ResponseCreatedEvent(
+                type="response.created", sequence_number=0, response=resp.model_dump()
+            )
+
+
+class InputItemStringIdTestCase(CustomTestCase):
+    """A response.output item replayed into input (string id + content) must
+    keep its content rather than collapse to an item-reference."""
+
+    def test_string_id_dropped_only_for_content_items(self):
+        norm = ResponsesRequest._normalize_input_item_for_validation
+        kept = norm(
+            {
+                "id": "msg_x",
+                "role": "assistant",
+                "content": [{"type": "output_text", "text": "hi"}],
+            }
+        )
+        self.assertNotIn("id", kept)
+        self.assertTrue(kept["content"])
+        # int ids and bare item-references are left alone
+        self.assertEqual(norm({"id": 123, "content": [{"type": "text"}]})["id"], 123)
+        ref = {"type": "item_reference", "id": "msg_ref"}
+        self.assertEqual(norm(ref), ref)
+
+    def test_request_accepts_replayed_output_item(self):
+        # Construction must not raise (the bug returned 400).
+        ResponsesRequest(
+            model="x",
+            store=False,
+            input=[
+                {
+                    "id": "msg_x",
+                    "role": "assistant",
+                    "content": [{"type": "output_text", "text": "hi"}],
+                }
+            ],
+        )
+
+
+class ToolChoiceObjectFormTestCase(CustomTestCase):
+    def test_echoed_choice_is_what_the_server_honors(self):
+        import openai.types.responses as ort
+
+        named = {"type": "function", "name": "get_weather"}
+        # Object forms other than a named function cannot be forced through the
+        # tool-call parser, so the response must echo the "auto" we actually
+        # run -- which also keeps it inside the SDK's ToolChoice union that the
+        # typed event below validates against.
+        for tool_choice, expected in (
+            ("auto", "auto"),
+            ("required", "required"),
+            ("none", "none"),
+            (named, named),
+            ({"type": "function", "function": {"name": "get_weather"}}, named),
+            ({"type": "web_search"}, "auto"),
+            ({"type": "mcp", "server_label": "s"}, "auto"),
+        ):
+            req = ResponsesRequest(
+                model="x", input="hi", store=False, tool_choice=tool_choice
+            )
+            self.assertEqual(req.effective_tool_choice(), expected, tool_choice)
+            resp = _in_progress_response(req)
+            self.assertEqual(resp.tool_choice, expected, tool_choice)
+            ort.ResponseCreatedEvent(
+                type="response.created", sequence_number=0, response=resp.model_dump()
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/entrypoints/openai/test_serving_responses.py

@@ -16,15 +16,20 @@ from sglang.srt.entrypoints.openai.protocol import (
     RequestResponseMetadata,
     ResponsesRequest,
 )
-from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses
+from sglang.srt.entrypoints.openai.serving_responses import (
+    OpenAIServingResponses,
+    _build_output_text_logprobs,
+    _should_emit_normal_text_as_message,
+)
 from sglang.srt.function_call.core_types import ToolCallItem
 from sglang.srt.parser.template_detection import ReasoningToggleConfig
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
 
-register_cpu_ci(est_time=8, suite="base-a-test-cpu")
+register_cpu_ci(est_time=7, suite="base-a-test-cpu")
 
 
-class InputMessageConstructionTestCase(unittest.TestCase):
+class InputMessageConstructionTestCase(CustomTestCase):
     def test_previous_response_replays_assistant_text_not_instructions(self):
         serving = make_serving()
         prev_response = Mock(id="resp_prev")
@@ -146,7 +151,7 @@ class InputMessageConstructionTestCase(unittest.TestCase):
             pass
 
 
-class ChatToolForwardingTestCase(unittest.TestCase):
+class ChatToolForwardingTestCase(CustomTestCase):
     def test_make_request_passes_function_tools_to_chat_processing(self):
         serving = make_serving()
         seen = {}
@@ -306,7 +311,7 @@ class ReasoningRequestForwardingTestCase(unittest.TestCase):
         self.assertFalse(parser_cls.call_args.kwargs["force_reasoning"])
 
 
-class InputItemNormalizationTestCase(unittest.TestCase):
+class InputItemNormalizationTestCase(CustomTestCase):
     def test_function_call_becomes_assistant_tool_call(self):
         normalized = OpenAIServingResponses._normalize_response_message_for_chat(
             {
@@ -361,7 +366,7 @@ class InputItemNormalizationTestCase(unittest.TestCase):
             )
 
 
-class FullResponseUsageTestCase(unittest.TestCase):
+class FullResponseUsageTestCase(CustomTestCase):
     def test_full_response_uses_dict_meta_info_for_usage(self):
         serving = make_serving()
         context = SimpleContext()
@@ -403,7 +408,7 @@ class FullResponseUsageTestCase(unittest.TestCase):
         self.assertEqual(metadata.final_usage_info, response.usage)
 
 
-class MultimodalRequestTestCase(unittest.TestCase):
+class MultimodalRequestTestCase(CustomTestCase):
     def test_text_only_create_responses_rejects_media_before_generation(self):
         serving = make_serving()
         serving._process_messages = Mock()
@@ -500,7 +505,13 @@ class MultimodalRequestTestCase(unittest.TestCase):
         self.assertEqual(captured["adapted_request"].modalities, ["image"])
 
 
-class OutputItemsTestCase(unittest.TestCase):
+class OutputItemsTestCase(CustomTestCase):
+    def setUp(self):
+        # qwen3_coder is the default for this class; the one no-native-parser
+        # case overrides it.
+        self.serving = make_serving()
+        self.serving.tool_call_parser = "qwen3_coder"
+
     def _function_tool_request(self):
         return ResponsesRequest(
             model="x",
@@ -517,8 +528,7 @@ class OutputItemsTestCase(unittest.TestCase):
         )
 
     def test_function_tool_call_extracted_via_parser(self):
-        serving = make_serving()
+        serving = self.serving
-        serving.tool_call_parser = "qwen3_coder"
         fake_call = ToolCallItem(
             tool_index=0, name="get_weather", parameters='{"city": "Beijing"}'
         )
@@ -550,8 +560,7 @@ class OutputItemsTestCase(unittest.TestCase):
         self.assertEqual(message_items[0].content[0].text, "trailing text")
 
     def test_prose_emitted_before_tool_call_item(self):
-        serving = make_serving()
+        serving = self.serving
-        serving.tool_call_parser = "qwen3_coder"
         fake_call = ToolCallItem(
             tool_index=0, name="get_weather", parameters='{"city": "Beijing"}'
         )
@@ -576,7 +585,7 @@ class OutputItemsTestCase(unittest.TestCase):
         self.assertEqual(types, ["ResponseOutputMessage", "ResponseFunctionToolCall"])
 
     def test_required_tool_choice_parses_json_array_without_native_parser(self):
-        serving = make_serving()
+        serving = self.serving
         serving.tool_call_parser = None
         request = ResponsesRequest(
             model="x",
@@ -609,8 +618,7 @@ class OutputItemsTestCase(unittest.TestCase):
         )
 
     def test_no_tool_call_extraction_when_tool_choice_none(self):
-        serving = make_serving()
+        serving = self.serving
-        serving.tool_call_parser = "qwen3_coder"
         request = ResponsesRequest(
             model="x",
             input="hi",
@@ -640,7 +648,7 @@ class OutputItemsTestCase(unittest.TestCase):
         self.assertIsInstance(output_items[0], ResponseOutputMessage)
 
 
-class HarmonyResponsesTestCase(unittest.TestCase):
+class HarmonyResponsesTestCase(CustomTestCase):
     def test_developer_message_skips_unsupported_tool_types(self):
         from sglang.srt.entrypoints.harmony_utils import get_developer_message
         from sglang.srt.entrypoints.openai.protocol import ResponseTool
@@ -660,5 +668,214 @@ class HarmonyResponsesTestCase(unittest.TestCase):
         self.assertIsNotNone(msg)
 
 
+class StatusFromFinishReasonTestCase(CustomTestCase):
+    def test_only_length_maps_to_incomplete(self):
+        fn = OpenAIServingResponses._status_from_finish_reason
+        self.assertEqual(fn({"type": "length"}), "incomplete")
+        self.assertEqual(fn("length"), "incomplete")
+        for other in ({"type": "stop"}, {"type": "tool_calls"}, "stop", None):
+            self.assertEqual(fn(other), "completed", other)
+
+
+class BuildOutputTextLogprobsTestCase(CustomTestCase):
+    def test_tokens_and_top_logprobs_are_converted(self):
+        meta_info = {
+            "output_token_logprobs": [(-0.1, 10, "Hello"), (-0.2, 11, " world")],
+            "output_top_logprobs": [
+                [(-0.1, 10, "Hello"), (-2.0, 12, "Hi")],
+                [(-0.2, 11, " world"), (-3.0, 13, " earth")],
+            ],
+        }
+        out = _build_output_text_logprobs(meta_info)
+        self.assertEqual(len(out), 2)
+        self.assertEqual(out[0].token, "Hello")
+        self.assertEqual(out[0].logprob, -0.1)
+        self.assertEqual(out[0].bytes, list("Hello".encode("utf-8")))
+        self.assertEqual(len(out[0].top_logprobs), 2)
+        self.assertEqual(out[0].top_logprobs[0].token, "Hello")
+        self.assertEqual(out[1].token, " world")
+
+    def test_no_top_logprobs_yields_empty_lists(self):
+        meta_info = {
+            "output_token_logprobs": [(-0.5, 7, "hi")],
+            "output_top_logprobs": None,
+        }
+        out = _build_output_text_logprobs(meta_info)
+        self.assertEqual(len(out), 1)
+        self.assertEqual(out[0].top_logprobs, [])
+
+
+class ChatToolChoiceConversionTestCase(CustomTestCase):
+    def test_conversion(self):
+        fn = OpenAIServingResponses._chat_tool_choice
+        for s in ("auto", "required", "none"):
+            self.assertEqual(fn(s), s)
+        # Input is an effective_tool_choice() result, so the only object form
+        # reaching here is a named function; degrading the rest to "auto"
+        # happens there, once, so the echoed and the honored value agree.
+        self.assertEqual(
+            fn({"type": "function", "name": "get_weather"}),
+            {"type": "function", "function": {"name": "get_weather"}},
+        )
+
+
+class ShouldEmitNormalTextTestCase(CustomTestCase):
+    def test_whitespace_suppressed_only_while_a_tool_is_open(self):
+        emit = _should_emit_normal_text_as_message
+        self.assertFalse(emit("", any_tool_call_in_progress=False))
+        # whitespace between tool blocks is an inter-call separator, not content
+        self.assertFalse(emit("\n", any_tool_call_in_progress=True))
+        self.assertTrue(emit("\n", any_tool_call_in_progress=False))
+        self.assertTrue(emit("hello", any_tool_call_in_progress=True))
+
+
+class EnginePassthroughTestCase(CustomTestCase):
+    """Both flags cross hops with no type contract, and dropping either fails
+    silently."""
+
+    def _capture(self, serving, request):
+        # Let the real _process_messages run: it is the hop that turns
+        # skip_special_tokens off, so mocking it would make that assertion vacuous.
+        # chat_template_name=None routes it through the tokenizer's template
+        # (mocked) instead of the conversation registry, which has no fixture entry.
+        serving.default_chat_template_kwargs = {}
+        serving.template_manager.chat_template_name = None
+        captured = {}
+
+        async def fake_generate(
+            request_id,
+            request_prompt,
+            adapted_request,
+            sampling_params,
+            context,
+            **kwargs,
+        ):
+            captured["adapted_request"] = adapted_request
+            captured["sampling_params"] = sampling_params
+            context.append_output(
+                {
+                    "text": "ok",
+                    "meta_info": {
+                        "prompt_tokens": 1,
+                        "completion_tokens": 1,
+                        "cached_tokens": 0,
+                    },
+                }
+            )
+            yield context
+
+        serving._generate_with_builtin_tools = fake_generate
+        asyncio.run(serving.create_responses(request))
+        return captured
+
+    def test_require_reasoning_forwarded_when_reasoning_parser_configured(self):
+        serving = make_serving()
+        serving.reasoning_parser = "deepseek-r1"
+        serving.template_manager.reasoning_config = ReasoningToggleConfig(
+            toggle_param="thinking", default_enabled=True
+        )
+
+        captured = self._capture(
+            serving, ResponsesRequest(model="x", input="hi", store=False)
+        )
+
+        self.assertTrue(captured["adapted_request"].require_reasoning)
+
+    def test_prefilled_think_template_opens_the_parser(self):
+        """``force_reasoning`` is a template property, not a request one, so it
+        drives the parser but never the engine flag -- as on the chat path."""
+        serving = make_serving()
+        serving.reasoning_parser = "deepseek-r1"
+        serving.template_manager.force_reasoning = True
+
+        with patch(
+            "sglang.srt.entrypoints.openai.serving_responses.ReasoningParser"
+        ) as parser_cls:
+            parser_cls.return_value.parse_non_stream.return_value = (None, "hi")
+            serving._make_response_output_items(
+                ResponsesRequest(model="x", input="hi", store=False),
+                "hi",
+                tokenizer=Mock(),
+                require_reasoning=False,
+            )
+
+        self.assertTrue(parser_cls.call_args.kwargs["force_reasoning"])
+
+    def test_require_reasoning_false_without_reasoning_parser(self):
+        serving = make_serving()
+        serving.reasoning_parser = None
+
+        captured = self._capture(
+            serving, ResponsesRequest(model="x", input="hi", store=False)
+        )
+
+        self.assertFalse(captured["adapted_request"].require_reasoning)
+
+    def test_skip_special_tokens_disabled_for_tool_requests(self):
+        # _process_messages turns it off so tool-call markers survive detokenize;
+        # create_responses must re-apply it to the engine sampling dict.
+        serving = make_serving()
+        serving.tool_call_parser = "qwen25"
+
+        captured = self._capture(
+            serving,
+            ResponsesRequest(
+                model="x",
+                input="weather",
+                store=False,
+                tools=[
+                    {
+                        "type": "function",
+                        "name": "get_weather",
+                        "parameters": {"type": "object"},
+                    }
+                ],
+            ),
+        )
+
+        self.assertFalse(captured["sampling_params"]["skip_special_tokens"])
+
+
+class CancelIdempotencyTestCase(CustomTestCase):
+    def test_cancelling_a_terminal_response_returns_it_not_an_error(self):
+        from sglang.srt.entrypoints.openai.protocol import ResponsesResponse
+
+        for status in ("cancelled", "completed"):
+            serving = make_serving()
+            resp = ResponsesResponse.from_request(
+                ResponsesRequest(model="x", input="hi", store=False),
+                sampling_params={},
+                model_name="x",
+                created_time=0,
+                output=[],
+                status=status,
+                usage=None,
+            )
+            serving.response_store[resp.id] = resp
+
+            out = asyncio.run(serving.cancel_responses(resp.id))
+
+            self.assertIs(out, resp, status)
+            self.assertEqual(out.status, status)
+
+
+class StreamingLogprobsRejectionTestCase(CustomTestCase):
+    def test_stream_with_logprobs_include_rejected(self):
+        import orjson
+
+        serving = make_serving()
+        request = ResponsesRequest(
+            model="x",
+            input="hi",
+            store=False,
+            stream=True,
+            include=["message.output_text.logprobs"],
+        )
+        result = asyncio.run(serving.create_responses(request))
+        self.assertEqual(result.status_code, 400)
+        body = orjson.loads(result.body)
+        self.assertIn("streaming mode", body["error"]["message"])
+
+
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/entrypoints/openai/test_serving_responses_stream.py

@@ -1,67 +1,23 @@
-import asyncio
 import unittest
-from unittest.mock import Mock, patch
+from unittest.mock import patch
 
 from utils import (
-    collect_stream_events,
+    StreamFixture,
+    engine_chunk,
     event_payloads,
     event_types,
     find_completed_event,
     make_serving,
 )
 
-from sglang.srt.entrypoints.openai.protocol import (
+from sglang.srt.entrypoints.openai.protocol import ResponsesRequest
-    RequestResponseMetadata,
-    ResponsesRequest,
-)
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
 
-register_cpu_ci(est_time=4, suite="base-a-test-cpu")
+register_cpu_ci(est_time=7, suite="base-a-test-cpu")
 
 
-class _StreamFixture:
+class NonHarmonyStreamTestCase(CustomTestCase):
-    def __init__(self, serving, request, *, require_reasoning=False):
-        self.serving = serving
-        self.request = request
-        self.require_reasoning = require_reasoning
-        self.request_metadata = RequestResponseMetadata(request_id=request.request_id)
-
-    def run(self, chunks):
-        async def gen():
-            for ch in chunks:
-                yield ch
-
-        async def collect():
-            return await collect_stream_events(
-                self.serving.responses_stream_generator_non_harmony(
-                    self.request,
-                    sampling_params={},
-                    result_generator=gen(),
-                    model_name="x",
-                    tokenizer=Mock(),
-                    request_metadata=self.request_metadata,
-                    require_reasoning=self.require_reasoning,
-                )
-            )
-
-        return asyncio.run(collect())
-
-
-def _engine_chunk(text, completion_tokens, *, finish=False):
-    return {
-        "text": text,
-        "meta_info": {
-            "id": "rid",
-            "prompt_tokens": 5,
-            "completion_tokens": completion_tokens,
-            "cached_tokens": 0,
-            "reasoning_tokens": 0,
-            "finish_reason": {"type": "stop"} if finish else None,
-        },
-    }
-
-
-class NonHarmonyStreamTestCase(unittest.TestCase):
     def test_reasoning_parser_uses_processed_reasoning_state(self):
         serving = make_serving()
         serving.reasoning_parser = "deepseek-r1"
@@ -71,8 +27,8 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
             "sglang.srt.entrypoints.openai.serving_responses.ReasoningParser"
         ) as parser_cls:
             parser_cls.return_value.parse_stream_chunk.return_value = (None, "done")
-            fixture = _StreamFixture(serving, request, require_reasoning=True)
+            fixture = StreamFixture(serving, request, require_reasoning=True)
-            fixture.run([_engine_chunk("done", 1, finish=True)])
+            fixture.run([engine_chunk("done", 1, finish=True)])
 
         self.assertTrue(parser_cls.call_args.kwargs["force_reasoning"])
 
@@ -82,12 +38,12 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
         serving.tool_call_parser = None
 
         request = ResponsesRequest(model="x", input="hi", stream=True, store=False)
-        fixture = _StreamFixture(serving, request)
+        fixture = StreamFixture(serving, request)
         events = fixture.run(
             [
-                _engine_chunk("Hel", 1),
+                engine_chunk("Hel", 1),
-                _engine_chunk("Hello", 2),
+                engine_chunk("Hello", 2),
-                _engine_chunk("Hello world", 4, finish=True),
+                engine_chunk("Hello world", 4, finish=True),
             ]
         )
 
@@ -134,10 +90,10 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
         while sent < len(payload):
             sent += min(8, len(payload) - sent)
             chunks.append(
-                _engine_chunk(payload[:sent], sent, finish=sent == len(payload))
+                engine_chunk(payload[:sent], sent, finish=sent == len(payload))
             )
 
-        fixture = _StreamFixture(serving, request)
+        fixture = StreamFixture(serving, request)
         events = fixture.run(chunks)
         types = event_types(events)
 
@@ -193,9 +149,9 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
             StreamingParseResult(normal_text="It's sunny.", calls=[]),
         ]
         chunks = [
-            _engine_chunk(" " * 3, 3),
+            engine_chunk(" " * 3, 3),
-            _engine_chunk(" " * 10, 10),
+            engine_chunk(" " * 10, 10),
-            _engine_chunk(" " * 14, 14, finish=True),
+            engine_chunk(" " * 14, 14, finish=True),
         ]
 
         script_iter = iter(scripted)
@@ -211,7 +167,7 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
             parser_cls.return_value.parse_stream_chunk.side_effect = (
                 fake_parse_stream_chunk
             )
-            fixture = _StreamFixture(serving, request)
+            fixture = StreamFixture(serving, request)
             events = fixture.run(chunks)
 
         completed = find_completed_event(events)
@@ -223,5 +179,109 @@ class NonHarmonyStreamTestCase(unittest.TestCase):
         self.assertEqual(output[2]["content"][0]["text"], "It's sunny.")
 
 
+class MultiToolCallStreamingOrderTestCase(CustomTestCase):
+    """The wire order of message / function_call items across tool-call deltas."""
+
+    def setUp(self):
+        from sglang.srt.function_call.qwen3_coder_detector import Qwen3CoderDetector
+
+        self.serving = make_serving()
+        self.serving.tool_call_parser = "qwen3_coder"
+        self.serving.reasoning_parser = None
+
+        det = Qwen3CoderDetector()
+        s, e = det.tool_call_start_token, det.tool_call_end_token
+        fp, fe = det.tool_call_prefix, det.function_end_token
+        pp, pe = det.parameter_prefix, det.parameter_end_token
+        self.weather = f"{s}{fp}get_weather>{pp}city>Beijing{pe}{fe}{e}"
+        self.time = f"{s}{fp}get_time>{pp}tz>UTC{pe}{fe}{e}"
+        # a prefix of ``weather`` that stops mid-arguments
+        self.weather_head = f"{s}{fp}get_weather>{pp}city>Beij"
+
+    def _seq(self, texts, *names):
+        """Stream cumulative ``texts`` (last one final) and return (type, payload)."""
+        request = ResponsesRequest(
+            model="x",
+            input="weather and time",
+            store=False,
+            tools=[
+                {"type": "function", "name": n, "parameters": {"type": "object"}}
+                for n in names
+            ],
+        )
+        chunks = [engine_chunk(t) for t in texts]
+        chunks.append(engine_chunk(texts[-1], finish=True))
+        return StreamFixture(self.serving, request).run_seq(chunks)
+
+    @staticmethod
+    def _added(seq):
+        return [
+            (p["output_index"], p["item"].get("type"))
+            for t, p in seq
+            if t == "response.output_item.added"
+        ]
+
+    @staticmethod
+    def _done_calls(seq):
+        return [
+            p["item"]
+            for t, p in seq
+            if t == "response.output_item.done"
+            and p["item"].get("type") == "function_call"
+        ]
+
+    def test_prior_tool_call_done_before_next_added(self):
+        full = self.weather + "\n" + self.time
+        seq = self._seq(
+            [self.weather, self.weather + "\n", full], "get_weather", "get_time"
+        )
+
+        def position(pred):
+            return next(i for i, (t, p) in enumerate(seq) if pred(t, p))
+
+        done0 = position(
+            lambda t, p: t == "response.output_item.done" and p["output_index"] == 0
+        )
+        added1 = position(
+            lambda t, p: t == "response.output_item.added" and p["output_index"] == 1
+        )
+        self.assertLess(done0, added1)
+
+        items = self._done_calls(seq)
+        self.assertEqual(sorted(i["name"] for i in items), ["get_time", "get_weather"])
+
+    def test_prose_before_tool_call_keeps_message_first(self):
+        """Prose and a tool-call start in one delta: the message item must come
+        first, since the prose preceded the call."""
+        # One delta spanning prose + the whole call, as spec decoding or
+        # --stream-interval > 1 produces.
+        seq = self._seq(["Let me check." + self.weather], "get_weather")
+
+        added = self._added(seq)
+        message_index = next(i for i, kind in added if kind == "message")
+        call_index = next(i for i, kind in added if kind == "function_call")
+        self.assertLess(message_index, call_index)
+
+        # The call must not be split across two items by the reordering.
+        self.assertEqual(len([k for _, k in added if k == "function_call"]), 1)
+
+    def test_call_tail_prose_and_next_call_in_one_delta(self):
+        """One delta closing tool1, carrying prose, and opening tool2 needs both
+        orders at once: tool1's trailing "}" must be drained before the prose
+        closes every open item, and tool2 must land after the message."""
+        seq = self._seq(
+            [self.weather_head, self.weather + "Here you go." + self.time],
+            "get_weather",
+            "get_time",
+        )
+
+        items = self._done_calls(seq)
+        # No duplicate item invented for the already-closed call, and no call
+        # left nameless by being reopened from an args-only fragment.
+        self.assertEqual(len(items), 2)
+        self.assertTrue(all(i["name"] for i in items))
+        self.assertEqual(items[0]["arguments"], '{"city": "Beijing"}')
+
+
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/entrypoints/openai/utils.py

@@ -20,10 +20,12 @@ from sglang.test.test_utils import maybe_stub_sgl_kernel
 
 maybe_stub_sgl_kernel()
 
+import asyncio
 import json
 from typing import AsyncIterator
 from unittest.mock import Mock
 
+from sglang.srt.entrypoints.openai.protocol import RequestResponseMetadata
 from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses
 from sglang.test.ci.ci_register import register_cpu_ci
 
@@ -115,3 +117,52 @@ def find_completed_event(events: list[str]) -> dict:
         if lines and lines[0] == "event: response.completed":
             return json.loads(lines[1][len("data: ") :])
     raise AssertionError("response.completed event missing from stream")
+
+
+def engine_chunk(text, completion_tokens=1, *, finish=False):
+    return {
+        "text": text,
+        "meta_info": {
+            "id": "rid",
+            "prompt_tokens": 5,
+            "completion_tokens": completion_tokens,
+            "cached_tokens": 0,
+            "reasoning_tokens": 0,
+            "finish_reason": {"type": "stop"} if finish else None,
+        },
+    }
+
+
+class StreamFixture:
+    """Drives ``responses_stream_generator_non_harmony`` over a chunk list."""
+
+    def __init__(self, serving, request, *, require_reasoning=False):
+        self.serving = serving
+        self.request = request
+        self.require_reasoning = require_reasoning
+        self.request_metadata = RequestResponseMetadata(request_id=request.request_id)
+
+    def run(self, chunks) -> list[str]:
+        async def gen():
+            for ch in chunks:
+                yield ch
+
+        async def collect():
+            return await collect_stream_events(
+                self.serving.responses_stream_generator_non_harmony(
+                    self.request,
+                    sampling_params={},
+                    result_generator=gen(),
+                    model_name="x",
+                    tokenizer=Mock(),
+                    request_metadata=self.request_metadata,
+                    require_reasoning=self.require_reasoning,
+                )
+            )
+
+        return asyncio.run(collect())
+
+    def run_seq(self, chunks) -> list[tuple]:
+        """``run`` plus (event type, payload) pairing, the common assertion shape."""
+        events = self.run(chunks)
+        return list(zip(event_types(events), event_payloads(events)))