vLLM/vllm/entrypoints/openai/responses/serving.py

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

import asyncio
import time
from collections import deque
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping, Sequence
from contextlib import AsyncExitStack
from http import HTTPStatus
from typing import Any, Final, cast

from fastapi import Request
from openai.types.responses import (
    ResponseOutputItem,
    ResponseOutputItemDoneEvent,
    ResponseOutputMessage,
    ResponseOutputText,
    ResponseStatus,
)
from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
from pydantic import TypeAdapter

from vllm import envs
from vllm.config.utils import replace
from vllm.engine.protocol import EngineClient
from vllm.entrypoints.generate.base.protocol import (
    DeltaMessage,
    RequestResponseMetadata,
)
from vllm.entrypoints.generate.base.serving import (
    GenerateBaseServing,
    build_per_request_timing_metrics,
)
from vllm.entrypoints.mcp.tool_server import ToolServer
from vllm.entrypoints.openai.models.serving import OpenAIServingModels
from vllm.entrypoints.openai.responses.context import (
    ConversationContext,
    HarmonyContext,
    ParsableContext,
    SimpleContext,
)
from vllm.entrypoints.openai.responses.harmony import harmony_to_response_output
from vllm.entrypoints.openai.responses.protocol import (
    InputTokensDetails,
    OutputTokensDetails,
    ResponseCompletedEvent,
    ResponseCreatedEvent,
    ResponseIncompleteEvent,
    ResponseInProgressEvent,
    ResponseInputOutputItem,
    ResponseInputOutputMessage,
    ResponsesRequest,
    ResponsesResponse,
    ResponseUsage,
    StreamingResponsesResponse,
)
from vllm.entrypoints.openai.responses.streaming_events import (
    SimpleStreamingEventProcessor,
    StreamingState,
    _StateType,
    emit_content_delta_events,
    emit_previous_item_done_events,
    emit_tool_action_events,
    split_delta,
)
from vllm.entrypoints.openai.responses.utils import (
    build_response_output_items,
    extract_function_tool_names,
    extract_tool_types,
    response_item_status,
    reuse_streamed_item_ids,
)
from vllm.entrypoints.serve.engine.protocol import ErrorResponse
from vllm.entrypoints.serve.utils.api_utils import get_max_tokens
from vllm.entrypoints.serve.utils.request_logger import RequestLogger
from vllm.entrypoints.serve.utils.tool_calls_utils import (
    maybe_filter_parallel_tool_calls,
)
from vllm.exceptions import GenerationError, VLLMValidationError
from vllm.inputs import EngineInput
from vllm.logger import init_logger
from vllm.logprobs import Logprob as SampleLogprob
from vllm.logprobs import SampleLogprobs
from vllm.lora.request import LoRARequest
from vllm.outputs import CompletionOutput
from vllm.parser import Parser, ParserManager
from vllm.renderers import TokenizeParams
from vllm.renderers.chat_utils import (
    ChatTemplateContentFormatOption,
)
from vllm.renderers.online_renderer import (
    OnlineRenderer,
    ResponsesPreviousMessages,
    ResponsesRenderResult,
)
from vllm.sampling_params import SamplingParams, StructuredOutputsParams
from vllm.tokenizers import TokenizerLike
from vllm.utils import random_uuid
from vllm.utils.collection_utils import as_list

logger = init_logger(__name__)


class OpenAIServingResponses(GenerateBaseServing):
    def __init__(
        self,
        engine_client: EngineClient,
        models: OpenAIServingModels,
        online_renderer: OnlineRenderer,
        *,
        request_logger: RequestLogger | None,
        chat_template: str | None,
        chat_template_content_format: ChatTemplateContentFormatOption,
        return_tokens_as_token_ids: bool = False,
        reasoning_parser: str = "",
        enable_auto_tools: bool = False,
        tool_parser: str | None = None,
        tool_strict_level: str | None = None,
        tool_server: ToolServer | None = None,
        enable_prompt_tokens_details: bool = False,
        enable_force_include_usage: bool = False,
        enable_per_request_metrics: bool = False,
        enable_log_outputs: bool = False,
        default_chat_template_kwargs: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(
            engine_client=engine_client,
            models=models,
            request_logger=request_logger,
            return_tokens_as_token_ids=return_tokens_as_token_ids,
        )

        self.online_renderer = online_renderer
        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
        self.chat_template_kwargs = default_chat_template_kwargs or {}
        self.enable_log_outputs = enable_log_outputs

        # Set up the unified parser - either a unified parser or fall back to
        # separate parsers accessed through the parser interface
        self.parser = ParserManager.get_parser(
            tool_parser_name=tool_parser,
            reasoning_parser_name=reasoning_parser,
            enable_auto_tools=enable_auto_tools,
            tool_strict_level=tool_strict_level,
            model_name=self.model_config.model,
            is_harmony=self.model_config.hf_config.model_type == "gpt_oss",
            tokenizer=self.renderer.tokenizer,
        )
        self.enable_prompt_tokens_details = enable_prompt_tokens_details
        self.enable_force_include_usage = enable_force_include_usage
        self.enable_per_request_metrics = enable_per_request_metrics

        self.default_sampling_params = self.model_config.get_diff_sampling_param()
        mc = self.model_config
        self.override_max_tokens = (
            self.default_sampling_params.get("max_tokens")
            if mc.generation_config not in ("auto", "vllm")
            else getattr(mc, "override_generation_config", {}).get("max_new_tokens")
        )

        # If False (default), the "store" option is (silently) ignored and the
        # response is not stored. If True, the response is stored in memory.
        # NOTE(woosuk): This may not be intuitive for users, as the default
        # behavior in OpenAI's Responses API is to store the response, but
        # vLLM's default behavior is not.
        self.enable_store = envs.VLLM_ENABLE_RESPONSES_API_STORE
        if self.enable_store:
            logger.warning_once(
                "`VLLM_ENABLE_RESPONSES_API_STORE` is enabled. This may "
                "cause a memory leak since we never remove responses from "
                "the store."
            )

        self.use_harmony = self.model_config.hf_config.model_type == "gpt_oss"
        if self.use_harmony:
            logger.warning(
                "For gpt-oss, we ignore --enable-auto-tool-choice "
                "and always enable tool use."
            )
        self.enable_auto_tools = enable_auto_tools
        # HACK(woosuk): This is a hack. We should use a better store.
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove responses from the store.
        self.response_store: dict[str, ResponsesResponse] = {}
        self.response_store_lock = asyncio.Lock()

        # HACK(woosuk): This is a hack. We should use a better store.
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove messages from the store.
        self.msg_store: dict[str, ResponsesPreviousMessages] = {}

        # HACK(wuhang): This is a hack. We should use a better store.
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove events from the store.
        self.event_store: dict[
            str, tuple[deque[StreamingResponsesResponse], asyncio.Event]
        ] = {}

        self.background_tasks: dict[str, asyncio.Task] = {}

        self.tool_server = tool_server

    def _effective_chat_template_kwargs(
        self, request: ResponsesRequest
    ) -> dict[str, Any]:
        return (
            request.build_chat_params(
                self.chat_template,
                self.chat_template_content_format,
            )
            .with_defaults(self.chat_template_kwargs)
            .chat_template_kwargs
        )

    def _make_response_parser(
        self,
        request: ResponsesRequest,
        tokenizer: TokenizerLike,
        chat_template_kwargs: dict[str, Any],
    ) -> Parser | None:
        if self.parser is None:
            return None
        return self.parser(
            tokenizer,
            request.tools,
            chat_template_kwargs=chat_template_kwargs,
            model_config=self.model_config,
        )

    def _validate_generator_input(
        self,
        engine_input: EngineInput,
    ) -> ErrorResponse | None:
        """Add validations to the input to the generator here."""
        prompt_len = self._extract_prompt_len(engine_input)
        max_model_len = self.model_config.max_model_len

        if prompt_len >= max_model_len:
            error_message = (
                f"The engine prompt length {prompt_len} "
                f"exceeds the max_model_len {max_model_len}. "
                "Please reduce prompt."
            )
            return self.create_error_response(
                err_type="invalid_request_error",
                message=error_message,
                status_code=HTTPStatus.BAD_REQUEST,
                param="input",
            )

        return None

    def _validate_create_responses_input(
        self, request: ResponsesRequest
    ) -> ErrorResponse | None:
        if self.use_harmony and request.is_include_output_logprobs():
            return self.create_error_response(
                err_type="invalid_request_error",
                message="logprobs are not supported with gpt-oss models",
                status_code=HTTPStatus.BAD_REQUEST,
                param="logprobs",
            )
        if request.store and not self.enable_store and request.background:
            return self.create_error_response(
                err_type="invalid_request_error",
                message=(
                    "This vLLM engine does not support `store=True` and "
                    "therefore does not support the background mode. To "
                    "enable these features, set the environment variable "
                    "`VLLM_ENABLE_RESPONSES_API_STORE=1` when launching "
                    "the vLLM server."
                ),
                status_code=HTTPStatus.BAD_REQUEST,
                param="background",
            )
        if request.previous_input_messages and request.previous_response_id:
            return self.create_error_response(
                err_type="invalid_request_error",
                message="Only one of `previous_input_messages` and "
                "`previous_response_id` can be set.",
                status_code=HTTPStatus.BAD_REQUEST,
                param="previous_response_id",
            )
        return None

    async def _render_resolved_response_inputs(
        self,
        request: ResponsesRequest,
        prev_response: ResponsesResponse | None,
        *,
        skip_mm_cache: bool = False,
    ) -> ResponsesRenderResult | ErrorResponse:
        previous_messages: ResponsesPreviousMessages | None = None
        previous_response_outputs: list[ResponseOutputItem] | None = None
        if prev_response is not None:
            stored_messages = (
                self.msg_store[prev_response.id]
                if self.use_harmony
                else self.msg_store.get(prev_response.id)
            )
            if stored_messages is not None:
                previous_messages = stored_messages.copy()
            previous_response_outputs = list(prev_response.output)

        return await self.online_renderer.render_responses(
            request,
            previous_messages=previous_messages,
            previous_response_outputs=previous_response_outputs,
            tool_server=self.tool_server,
            skip_mm_cache=skip_mm_cache,
        )

    async def create_responses(
        self,
        request: ResponsesRequest,
        raw_request: Request | None = None,
    ) -> (
        AsyncGenerator[StreamingResponsesResponse, None]
        | ResponsesResponse
        | ErrorResponse
    ):
        return await self._with_kv_transfer_rejection_cleanup(
            self._create_responses(request, raw_request), request, raw_request
        )

    async def _create_responses(
        self, request: ResponsesRequest, raw_request: Request | None = None
    ) -> (
        AsyncGenerator[StreamingResponsesResponse, None]
        | ResponsesResponse
        | ErrorResponse
    ):
        error_check_ret = await self._check_model(request)
        if error_check_ret is not None:
            logger.error("Error with model %s", error_check_ret)
            return error_check_ret
        maybe_validation_error = self._validate_create_responses_input(request)
        if maybe_validation_error is not None:
            return maybe_validation_error

        maybe_template_error = self.online_renderer.validate_chat_template(
            request_chat_template=None,
            chat_template_kwargs=request.chat_template_kwargs,
            trust_request_chat_template=(
                self.online_renderer.trust_request_chat_template
            ),
        )
        if maybe_template_error is not None:
            return maybe_template_error

        self._preflight()

        if request.store and not self.enable_store:
            # Disable the store option.
            # NOTE(woosuk): Although returning an error is possible, we opted
            # to implicitly disable store and process the request anyway, as
            # we assume most users do not intend to actually store the response
            # (i.e., their request's `store=True` just because it's the default
            # value).
            request.store = False

        # Handle the previous response ID.
        prev_response_id = request.previous_response_id
        if prev_response_id is not None:
            async with self.response_store_lock:
                prev_response = self.response_store.get(prev_response_id)
            if prev_response is None:
                return self._make_not_found_error(prev_response_id)
        else:
            prev_response = None

        lora_request = self._maybe_get_adapters(request)
        model_name = self.models.model_name(lora_request)

        render_result = await self._render_resolved_response_inputs(
            request,
            prev_response,
        )
        if isinstance(render_result, ErrorResponse):
            return render_result
        messages = render_result.messages
        engine_inputs = [render_result.engine_input]

        request_metadata = RequestResponseMetadata(request_id=request.request_id)
        if raw_request:
            raw_request.state.request_metadata = request_metadata

        # Schedule the request and get the result generator.
        max_model_len = self.model_config.max_model_len
        generators: list[AsyncGenerator[ConversationContext, None]] = []

        # Only include builtin tools that the request actually asked for.
        # Without this filter, tools registered on the server (e.g. via
        # --tool-server demo) would be available for execution even when
        # the request didn't enable them.
        requested_tool_types = extract_tool_types(request.tools)
        builtin_tool_list: list[str] = []
        if self.tool_server is not None:
            if (
                self.tool_server.has_tool("browser")
                and "web_search_preview" in requested_tool_types
            ):
                builtin_tool_list.append("browser")
            if (
                self.tool_server.has_tool("python")
                and "code_interpreter" in requested_tool_types
            ):
                builtin_tool_list.append("python")
            if (
                self.tool_server.has_tool("container")
                and "container" in requested_tool_types
            ):
                builtin_tool_list.append("container")

        if self.tool_server is not None:
            available_tools = builtin_tool_list
        else:
            assert len(builtin_tool_list) == 0
            available_tools = []
        tokenizer = self.renderer.get_tokenizer()

        for engine_input in engine_inputs:
            maybe_error = self._validate_generator_input(engine_input)
            if maybe_error is not None:
                return maybe_error

            default_max_tokens = get_max_tokens(
                max_model_len,
                request.max_output_tokens,
                self._extract_prompt_len(engine_input),
                self.default_sampling_params,
                self.override_max_tokens,
                truncate_prompt_tokens=(
                    -1 if request.truncation != "disabled" else None
                ),
            )

            sampling_params = request.to_sampling_params(
                default_max_tokens, self.default_sampling_params
            )

            trace_headers = (
                None
                if raw_request is None
                else await self._get_trace_headers(raw_request.headers)
            )
            session_id = self._get_session_id(request, raw_request)

            chat_template_kwargs = self._effective_chat_template_kwargs(request)
            response_parser = self._make_response_parser(
                request, tokenizer, chat_template_kwargs
            )
            if response_parser is not None:
                prompt_token_ids = self._extract_prompt_components(
                    engine_input
                ).token_ids
                if prompt_token_ids is not None:
                    response_parser.set_prompt_token_ids(prompt_token_ids)

            context: ConversationContext
            function_tool_names = extract_function_tool_names(request.tools)
            if self.use_harmony:
                context = HarmonyContext(
                    messages,
                    available_tools,
                    function_tool_names,
                    response_parser=response_parser,
                    request=request,
                )
            else:
                if envs.VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT:
                    # This is a feature in development for parsing
                    # tokens during generation instead of at the end
                    context = ParsableContext(
                        response_messages=messages,
                        tokenizer=tokenizer,
                        parser_cls=self.parser,
                        request=request,
                        response_parser=response_parser,
                        available_tools=available_tools,
                        chat_template=self.chat_template,
                        chat_template_content_format=self.chat_template_content_format,
                        enable_auto_tools=self.enable_auto_tools,
                    )
                else:
                    context = SimpleContext(
                        response_parser=response_parser,
                    )

            reasoning_parser_kwargs = None
            if (
                context.response_parser is not None
                and context.response_parser.reasoning_parser is not None
            ):
                reasoning_parser_kwargs = {
                    "chat_template_kwargs": chat_template_kwargs,
                }
                if (
                    isinstance(
                        struct_out := sampling_params.structured_outputs,
                        StructuredOutputsParams,
                    )
                    and struct_out.all_non_structural_tag_constraints_none()
                ):
                    sampling_params.structured_outputs = replace(
                        struct_out,
                        structural_tag=(
                            context.response_parser.reasoning_parser.prepare_structured_tag(
                                struct_out.structural_tag, self.tool_server
                            )
                        ),
                    )
            generator = self._generate_with_builtin_tools(
                request_id=request.request_id,
                engine_input=engine_input,
                sampling_params=sampling_params,
                context=context,
                tok_params=(
                    request.build_tok_params(self.model_config)
                    if isinstance(context, HarmonyContext)
                    else None
                ),
                lora_request=lora_request,
                priority=self._get_priority(request, raw_request),
                trace_headers=trace_headers,
                session_id=session_id,
                reasoning_parser_kwargs=reasoning_parser_kwargs,
            )
            generators.append(generator)

        assert len(generators) == 1
        (result_generator,) = generators

        # Store the input messages.
        if request.store:
            self.msg_store[request.request_id] = messages

        if request.background:
            created_time = int(time.time())
            response = ResponsesResponse.from_request(
                request,
                sampling_params,
                model_name=model_name,
                created_time=created_time,
                output=[],
                status="queued",
                usage=None,
            )
            async with self.response_store_lock:
                self.response_store[response.id] = response

            # Run the request in the background.
            if request.stream:
                task = asyncio.create_task(
                    self._run_background_request_stream(
                        request,
                        sampling_params,
                        result_generator,
                        context,
                        model_name,
                        tokenizer,
                        request_metadata,
                        created_time,
                    ),
                    name=f"create_{request.request_id}",
                )
            else:
                task = asyncio.create_task(
                    self._run_background_request(
                        request,
                        sampling_params,
                        result_generator,
                        context,
                        model_name,
                        tokenizer,
                        request_metadata,
                        created_time,
                    ),
                    name=f"create_{response.id}",
                )

            # For cleanup.
            response_id = response.id
            self.background_tasks[response_id] = task
            task.add_done_callback(
                lambda _: self.background_tasks.pop(response_id, None)
            )

            if request.stream:
                return self.responses_background_stream_generator(request.request_id)
            return response

        if request.stream:
            return self.responses_stream_generator(
                request,
                sampling_params,
                result_generator,
                context,
                model_name,
                tokenizer,
                request_metadata,
            )

        return await self.responses_full_generator(
            request,
            sampling_params,
            result_generator,
            context,
            model_name,
            tokenizer,
            request_metadata,
        )

    async def _render_next_turn(
        self,
        request: ResponsesRequest,
        messages: list[ResponseInputOutputItem],
    ) -> list[EngineInput]:
        render_request = request.model_copy(
            update={
                "input": list(messages),
                "instructions": None,
                "previous_input_messages": None,
                "previous_response_id": None,
            }
        )
        render_result = await self.online_renderer.render_responses(
            render_request,
            previous_messages=None,
            previous_response_outputs=None,
            tool_server=self.tool_server,
        )
        if isinstance(render_result, ErrorResponse):
            raise VLLMValidationError(render_result.error.message)
        return [render_result.engine_input]

    async def _generate_with_builtin_tools(
        self,
        request_id: str,
        engine_input: EngineInput,
        sampling_params: SamplingParams,
        context: ConversationContext,
        tok_params: TokenizeParams | None = None,
        lora_request: LoRARequest | None = None,
        priority: int = 0,
        trace_headers: Mapping[str, str] | None = None,
        session_id: str | None = None,
        reasoning_parser_kwargs: dict[str, Any] | None = None,
    ):
        max_model_len = self.model_config.max_model_len
        cache_salt = cast(str | None, engine_input.get("cache_salt"))

        orig_priority = priority
        sub_request = 0
        while True:
            # Ensure that each sub-request has a unique request id.
            sub_request_id = f"{request_id}_{sub_request}"

            self._log_inputs(
                sub_request_id,
                engine_input,
                params=sampling_params,
                lora_request=lora_request,
            )

            generator = self.engine_client.generate(
                engine_input,
                sampling_params,
                sub_request_id,
                lora_request=lora_request,
                trace_headers=trace_headers,
                priority=priority,
                session_id=session_id,
                reasoning_parser_kwargs=reasoning_parser_kwargs,
            )

            async for res in generator:
                context.request_metrics = res.metrics
                context.append_output(res)
                # NOTE(woosuk): The stop condition is handled by the engine.
                yield context

            if not context.need_builtin_tool_call():
                # The model did not ask for a tool call, so we're done.
                break

            # Call the tool and update the context with the result.
            context.request_metrics_cover_all_generation_turns = False
            tool_output = await context.call_tool()
            context.append_tool_output(tool_output)

            # TODO: uncomment this and enable tool output streaming
            # yield context

            # Create inputs for the next turn.
            # Render the next prompt token ids and update sampling_params.
            if isinstance(context, HarmonyContext):
                engine_input = self.online_renderer.render_responses_harmony_messages(
                    context.messages,
                    cache_salt=cache_salt,
                    tok_params=tok_params,
                )

                sampling_params.max_tokens = get_max_tokens(
                    max_model_len,
                    context.request.max_output_tokens if context.request else None,
                    self._extract_prompt_len(engine_input),
                    self.default_sampling_params,
                    self.override_max_tokens,
                )
            elif isinstance(context, ParsableContext):
                (engine_input,) = await self._render_next_turn(
                    context.request,
                    context.response_messages,
                )
                if context.response_parser is not None:
                    prompt_token_ids = self._extract_prompt_components(
                        engine_input
                    ).token_ids
                    if prompt_token_ids is not None:
                        context.response_parser.set_prompt_token_ids(prompt_token_ids)

                sampling_params.max_tokens = get_max_tokens(
                    max_model_len,
                    context.request.max_output_tokens,
                    self._extract_prompt_len(engine_input),
                    self.default_sampling_params,  # type: ignore
                    self.override_max_tokens,  # type: ignore
                    truncate_prompt_tokens=(
                        -1 if context.request.truncation != "disabled" else None
                    ),
                )

            # OPTIMIZATION
            priority = orig_priority - 1
            sub_request += 1

    async def _initialize_tool_sessions(
        self,
        request: ResponsesRequest,
        context: ConversationContext,
        exit_stack: AsyncExitStack,
    ):
        # we should only initialize the tool session if the request needs tools
        if len(request.tools) == 0:
            return
        mcp_tools = {
            tool.server_label: tool for tool in request.tools if tool.type == "mcp"
        }
        await context.init_tool_sessions(
            self.tool_server, exit_stack, request.request_id, mcp_tools
        )

    async def responses_full_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
        result_generator: AsyncIterator[ConversationContext],
        context: ConversationContext,
        model_name: str,
        tokenizer: TokenizerLike,
        request_metadata: RequestResponseMetadata,
        created_time: int | None = None,
    ) -> ErrorResponse | ResponsesResponse:
        if created_time is None:
            created_time = int(time.time())

        async with AsyncExitStack() as exit_stack:
            try:
                await self._initialize_tool_sessions(request, context, exit_stack)
                async for _ in result_generator:
                    pass
            except asyncio.CancelledError:
                return self.create_error_response("Client disconnected")

        # NOTE: Implementation of status is still WIP, but for now
        # we guarantee that if the status is not "completed", it is accurate.
        # "completed" is implemented as the "catch-all" for now.
        status: ResponseStatus = "completed"

        input_messages: ResponseInputOutputMessage | None = None
        output_messages: ResponseInputOutputMessage | None = None
        if self.use_harmony:
            assert isinstance(context, HarmonyContext)
            output = []
            harmony_msgs = context.messages[context.num_init_messages :]
            # Streamed messages are a subsequence of harmony_msgs, in order.
            streamed = iter(context.streamed_items_by_message)
            next_streamed = next(streamed, None)
            for i, msg in enumerate(harmony_msgs):
                items = harmony_to_response_output(
                    msg,
                    context.function_tool_names,
                    incomplete=(
                        i == len(harmony_msgs) - 1 and context.last_append_flush_status
                    ),
                )
                if next_streamed is not None and next_streamed[0] is msg:
                    reuse_streamed_item_ids(items, next_streamed[1])
                    next_streamed = next(streamed, None)
                output.extend(items)

            if request.enable_response_messages:
                input_messages = context.messages[: context.num_init_messages]
                output_messages = context.messages[context.num_init_messages :]
            num_tool_output_tokens = context.num_tool_output_tokens
            if len(output) > 0:
                if context.finish_reason == "length":
                    status = "incomplete"
                elif context.finish_reason == "abort":
                    status = "cancelled"
                else:
                    self._raise_if_error(context.finish_reason, request.request_id)
            else:
                status = "incomplete"
        elif isinstance(context, ParsableContext):
            output = context.make_response_output_items()

            if request.enable_response_messages:
                input_messages = context.input_messages
                output_messages = context.output_messages

            # TODO: Calculate usage.
            # assert final_res.prompt_token_ids is not None
            num_tool_output_tokens = 0

            # Check finish reason from the parser
            if context.finish_reason == "length":
                status = "incomplete"
        else:
            assert isinstance(context, SimpleContext)
            # Use final_output which has accumulated text/token_ids/logprobs
            final_res = context.final_output
            assert final_res is not None
            assert len(final_res.outputs) == 1
            final_output = final_res.outputs[0]

            # finish_reason='error' indicates retryable internal error
            self._raise_if_error(final_output.finish_reason, request.request_id)

            # Check if generation was stopped due to max_tokens
            if final_output.finish_reason == "length":
                status = "incomplete"

            streamed_items = context.streamed_output_items
            self._log_final_output(
                request, final_output, is_streaming=streamed_items is not None
            )
            if streamed_items is not None:
                output = streamed_items
            else:
                output = self._make_response_output_items(
                    request,
                    final_output,
                    tokenizer,
                    parser=context.response_parser,
                )

            if request.enable_response_messages:
                input_messages = context.input_messages
                output_messages = context.output_messages

            # Calculate usage.
            assert final_res.prompt_token_ids is not None
            num_tool_output_tokens = 0

        assert isinstance(context, (SimpleContext, HarmonyContext, ParsableContext))
        num_prompt_tokens = context.num_prompt_tokens
        num_generated_tokens = context.num_output_tokens
        num_cached_tokens = context.num_cached_tokens
        num_reasoning_tokens = context.num_reasoning_tokens
        # HarmonyContext and ParsableContext count reasoning tokens as each
        # round is appended. SimpleContext is single-round but accumulates
        # streaming deltas, so count its full output once here.
        if isinstance(context, SimpleContext) and context.response_parser is not None:
            num_reasoning_tokens = context.response_parser.count_reasoning_tokens(
                context._accumulated_token_ids
            )

        usage = ResponseUsage(
            input_tokens=num_prompt_tokens,
            output_tokens=num_generated_tokens,
            total_tokens=num_prompt_tokens + num_generated_tokens,
            input_tokens_details=InputTokensDetails(
                cache_write_tokens=context.num_cache_creation_tokens,
                cached_tokens=num_cached_tokens,
                input_tokens_per_turn=[
                    turn.input_tokens for turn in context.all_turn_metrics
                ],
                cached_tokens_per_turn=[
                    turn.cached_input_tokens for turn in context.all_turn_metrics
                ],
            ),
            output_tokens_details=OutputTokensDetails(
                reasoning_tokens=num_reasoning_tokens,
                tool_output_tokens=num_tool_output_tokens,
                output_tokens_per_turn=[
                    turn.output_tokens for turn in context.all_turn_metrics
                ],
                tool_output_tokens_per_turn=[
                    turn.tool_output_tokens for turn in context.all_turn_metrics
                ],
            ),
        )
        per_request_metrics = None
        if (
            self.enable_per_request_metrics
            and context.request_metrics_cover_all_generation_turns
        ):
            per_request_metrics = build_per_request_timing_metrics(
                context.request_metrics, num_generated_tokens
            )
        response = ResponsesResponse.from_request(
            request,
            sampling_params,
            input_messages=input_messages,
            output_messages=output_messages,
            model_name=model_name,
            created_time=created_time,
            output=output,
            status=status,
            usage=usage,
            metrics=per_request_metrics,
            kv_transfer_params=context.kv_transfer_params,
            ec_transfer_params=context.ec_transfer_params,
        )

        if request.store:
            async with self.response_store_lock:
                stored_response = self.response_store.get(response.id)
                # If the response is already cancelled, don't update it.
                if stored_response is None or stored_response.status != "cancelled":
                    self.response_store[response.id] = response
        return response

    def _topk_logprobs(
        self,
        logprobs: dict[int, SampleLogprob],
        top_logprobs: int,
        tokenizer: TokenizerLike,
    ) -> list[LogprobTopLogprob]:
        """Returns the top-k logprobs from the logprobs dictionary."""
        out = []
        for i, (token_id, _logprob) in enumerate(logprobs.items()):
            if i >= top_logprobs:
                break
            text = self._get_decoded_token(
                logprob=_logprob,
                token_id=token_id,
                tokenizer=tokenizer,
                return_as_token_id=self.return_tokens_as_token_ids,
            )
            out.append(
                LogprobTopLogprob(
                    token=text,
                    logprob=max(_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
                )
            )
        return out

    def _create_response_logprobs(
        self,
        token_ids: Sequence[int],
        logprobs: SampleLogprobs | None,
        tokenizer: TokenizerLike,
        top_logprobs: int | None = None,
    ) -> list[Logprob]:
        assert logprobs is not None, "logprobs must be provided"
        assert len(token_ids) == len(logprobs), (
            "token_ids and logprobs.token_ids must have the same length"
        )
        out = []
        for i, token_id in enumerate(token_ids):
            logprob = logprobs[i]
            token_logprob = logprob[token_id]
            text = self._get_decoded_token(
                logprob=token_logprob,
                token_id=token_id,
                tokenizer=tokenizer,
                return_as_token_id=self.return_tokens_as_token_ids,
            )
            out.append(
                Logprob(
                    token=text,
                    logprob=max(token_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
                    top_logprobs=(
                        self._topk_logprobs(
                            logprob, top_logprobs=top_logprobs, tokenizer=tokenizer
                        )
                        if top_logprobs
                        else []
                    ),
                )
            )
        return out

    def _log_final_output(
        self,
        request: ResponsesRequest,
        final_output: CompletionOutput,
        is_streaming: bool,
    ) -> None:
        if self.enable_log_outputs and self.request_logger:
            self.request_logger.log_outputs(
                request_id=request.request_id,
                outputs=final_output.text,
                output_token_ids=final_output.token_ids,
                finish_reason=final_output.finish_reason,
                is_streaming=is_streaming,
                delta=False,
            )

    def _make_response_output_items(
        self,
        request: ResponsesRequest,
        final_output: CompletionOutput,
        tokenizer: TokenizerLike,
        parser: Parser | None = None,
    ) -> list[ResponseOutputItem]:
        # Compute logprobs if requested
        logprobs = None
        if request.is_include_output_logprobs() and final_output.logprobs:
            logprobs = self._create_response_logprobs(
                token_ids=final_output.token_ids,
                logprobs=final_output.logprobs,
                tokenizer=tokenizer,
                top_logprobs=request.top_logprobs,
            )

        # Use parser to extract reasoning, content, and tool calls
        if parser:
            reasoning, content, tool_calls = parser.parse(
                final_output.text,
                request,
                enable_auto_tools=self.enable_auto_tools,
                model_output_token_ids=final_output.token_ids,
            )
            if not request.include_reasoning:
                reasoning = None
                logprobs = None
            return build_response_output_items(
                reasoning=reasoning,
                content=content,
                tool_calls=maybe_filter_parallel_tool_calls(tool_calls or [], request),
                logprobs=logprobs,
                tools=request.tools,
                finish_reason=final_output.finish_reason,
            )

        # Fallback when no parser is configured
        return [
            ResponseOutputMessage(
                id=f"msg_{random_uuid()}",
                content=[
                    ResponseOutputText(
                        text=final_output.text,
                        annotations=[],
                        type="output_text",
                        logprobs=logprobs,
                    )
                ]
                if final_output.text
                else [],
                role="assistant",
                status=response_item_status(final_output.finish_reason),
                type="message",
            )
        ]

    async def _run_background_request_stream(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
        event_deque: deque[StreamingResponsesResponse] = deque()
        new_event_signal = asyncio.Event()
        self.event_store[request.request_id] = (event_deque, new_event_signal)
        generator = self.responses_stream_generator(request, *args, **kwargs)
        try:
            async for event in generator:
                event_deque.append(event)
                new_event_signal.set()  # Signal new event available
        finally:
            new_event_signal.set()

    async def _run_background_request(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
        response = await self.responses_full_generator(request, *args, **kwargs)

        if isinstance(response, ErrorResponse):
            # If the request has failed, update the status to "failed".
            response_id = request.request_id
            async with self.response_store_lock:
                stored_response = self.response_store.get(response_id)
                assert stored_response is not None
                if stored_response.status not in ("completed", "cancelled"):
                    stored_response.status = "failed"

    async def responses_background_stream_generator(
        self,
        response_id: str,
        starting_after: int | None = None,
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
        if response_id not in self.event_store:
            raise VLLMValidationError(
                f"Unknown response_id: {response_id}",
                parameter="response_id",
                value=response_id,
            )

        event_deque, new_event_signal = self.event_store[response_id]
        start_index = 0 if starting_after is None else starting_after + 1
        current_index = start_index

        while True:
            new_event_signal.clear()

            # Yield existing events from start_index
            while current_index < len(event_deque):
                event = event_deque[current_index]
                yield event
                if getattr(event, "type", "unknown") in (
                    "response.completed",
                    "response.incomplete",
                ):
                    return
                current_index += 1

            await new_event_signal.wait()

    async def retrieve_responses(
        self,
        response_id: str,
        starting_after: int | None,
        stream: bool | None,
    ) -> (
        ErrorResponse
        | ResponsesResponse
        | AsyncGenerator[StreamingResponsesResponse, None]
    ):
        async with self.response_store_lock:
            response = self.response_store.get(response_id)

        if response is None:
            return self._make_not_found_error(response_id)

        if stream:
            return self.responses_background_stream_generator(
                response_id,
                starting_after,
            )
        return response

    async def cancel_responses(
        self,
        response_id: str,
    ) -> ErrorResponse | ResponsesResponse:
        async with self.response_store_lock:
            response = self.response_store.get(response_id)
            if response is None:
                return self._make_not_found_error(response_id)

            prev_status = response.status
            if prev_status not in ("queued", "in_progress"):
                return self.create_error_response(
                    err_type="invalid_request_error",
                    message="Cannot cancel a synchronous response.",
                    param="response_id",
                )

            # Update the status to "cancelled".
            response.status = "cancelled"

        # Abort the request.
        if task := self.background_tasks.get(response_id):
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                logger.exception("Background task for %s was cancelled", response_id)
        return response

    def _make_not_found_error(self, response_id: str) -> ErrorResponse:
        return self.create_error_response(
            err_type="invalid_request_error",
            message=f"Response with id '{response_id}' not found.",
            status_code=HTTPStatus.NOT_FOUND,
            param="response_id",
        )

    async def _process_simple_streaming_events(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
        result_generator: AsyncIterator[ConversationContext | None],
        context: ConversationContext,
        model_name: str,
        tokenizer: TokenizerLike,
        request_metadata: RequestResponseMetadata,
        created_time: int,
        _increment_sequence_number_and_return: Callable[
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
        processor = SimpleStreamingEventProcessor(tools=request.tools)

        hide_stream_metadata = (
            not request.include_reasoning and context.response_parser is not None
        )

        def _get_logprobs(output: CompletionOutput) -> list[Logprob]:
            if not request.is_include_output_logprobs():
                return []
            if hide_stream_metadata:
                return []
            return self._create_response_logprobs(
                token_ids=output.token_ids,
                logprobs=output.logprobs,
                tokenizer=tokenizer,
                top_logprobs=request.top_logprobs,
            )

        finish_reason: str | None = None
        async for ctx in result_generator:
            assert isinstance(ctx, SimpleContext)
            if ctx.last_output is None or not ctx.last_output.outputs:
                continue

            output = ctx.last_output.outputs[0]
            self._raise_if_error(output.finish_reason, request.request_id)
            finish_reason = output.finish_reason
            delta_text = output.text
            delta_token_ids = as_list(output.token_ids)

            if ctx.response_parser:
                delta_message = ctx.response_parser.parse_delta(
                    delta_text=delta_text,
                    delta_token_ids=delta_token_ids,
                    request=request,
                    prompt_token_ids=ctx.last_output.prompt_token_ids,
                    finished=output.finish_reason is not None,
                )
            else:
                delta_message = DeltaMessage(content=output.text)

            if not delta_message:
                continue
            delta_message = maybe_filter_parallel_tool_calls(delta_message, request)

            for dm in split_delta(delta_message):
                target_state, tool_call = processor.resolve_target_state(dm)
                if target_state == _StateType.NONE:
                    continue

                if processor.needs_transition(target_state, tool_call):
                    for event in processor.close_current():
                        yield _increment_sequence_number_and_return(event)
                    for event in processor.open(target_state, tool_call):
                        yield _increment_sequence_number_and_return(event)

                for event in processor.emit_delta(dm, output, _get_logprobs):
                    yield _increment_sequence_number_and_return(event)

        for event in processor.close_current(response_item_status(finish_reason)):
            yield _increment_sequence_number_and_return(event)

        assert isinstance(context, SimpleContext)
        context.streamed_output_items = processor.output_items

    async def _process_harmony_streaming_events(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
        result_generator: AsyncIterator[ConversationContext | None],
        context: ConversationContext,
        model_name: str,
        tokenizer: TokenizerLike,
        request_metadata: RequestResponseMetadata,
        created_time: int,
        _increment_sequence_number_and_return: Callable[
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
        state = StreamingState()

        async for ctx in result_generator:
            assert isinstance(ctx, HarmonyContext)

            # finish_reason='error' indicates a retryable error
            self._raise_if_error(ctx.finish_reason, request.request_id)

            for segment in ctx.last_append_segments:
                if segment.delta:
                    for event in emit_content_delta_events(
                        segment, state, ctx.function_tool_names
                    ):
                        yield _increment_sequence_number_and_return(event)

                elif completed_message := segment.completed_message:
                    done_items: list[ResponseOutputItem] = []
                    # TODO: Fix browser emitted as MCP calls
                    for event in emit_previous_item_done_events(
                        completed_message, state, ctx.function_tool_names
                    ):
                        if isinstance(event, ResponseOutputItemDoneEvent):
                            done_items.append(event.item)
                        yield _increment_sequence_number_and_return(event)

                    for event in emit_tool_action_events(
                        completed_message, state, self.tool_server
                    ):
                        if isinstance(event, ResponseOutputItemDoneEvent):
                            done_items.append(event.item)
                        yield _increment_sequence_number_and_return(event)
                    ctx.streamed_items_by_message.append(
                        (completed_message, done_items)
                    )
                    state.reset_for_new_item()

    async def responses_stream_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
        result_generator: AsyncIterator[ConversationContext | None],
        context: ConversationContext,
        model_name: str,
        tokenizer: TokenizerLike,
        request_metadata: RequestResponseMetadata,
        created_time: int | None = None,
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
        # TODO:
        # 1. Handle disconnect

        created_time = created_time or int(time.time())

        sequence_number = 0

        def _increment_sequence_number_and_return(
            event: StreamingResponsesResponse,
        ) -> StreamingResponsesResponse:
            nonlocal sequence_number
            # Set sequence_number if the event has this attribute
            if hasattr(event, "sequence_number"):
                event.sequence_number = sequence_number
            sequence_number += 1
            return event

        async with AsyncExitStack() as exit_stack:
            if self.use_harmony:
                # TODO: in streaming, we noticed this bug:
                # https://github.com/vllm-project/vllm/issues/25697
                await self._initialize_tool_sessions(request, context, exit_stack)
                processor = self._process_harmony_streaming_events
            else:
                processor = self._process_simple_streaming_events
            # TODO Hanchen make sampling params to include the structural tag

            initial_response = ResponsesResponse.from_request(
                request,
                sampling_params,
                model_name=model_name,
                created_time=created_time,
                output=[],
                status="in_progress",
                usage=None,
            ).model_dump(mode="json", by_alias=True)
            yield _increment_sequence_number_and_return(
                ResponseCreatedEvent(
                    type="response.created",
                    sequence_number=-1,
                    response=initial_response,
                )
            )
            yield _increment_sequence_number_and_return(
                ResponseInProgressEvent(
                    type="response.in_progress",
                    sequence_number=-1,
                    response=initial_response,
                )
            )

            try:
                async for event_data in processor(
                    request,
                    sampling_params,
                    result_generator,
                    context,
                    model_name,
                    tokenizer,
                    request_metadata,
                    created_time,
                    _increment_sequence_number_and_return,
                ):
                    yield event_data
            except GenerationError as e:
                error_json = self._convert_generation_error_to_streaming_response(e)
                yield _increment_sequence_number_and_return(
                    TypeAdapter(StreamingResponsesResponse).validate_json(error_json)
                )
                return

            async def empty_async_generator():
                # A hack to trick Python to think this is a generator but
                # in fact it immediately returns.
                if False:
                    yield

            final_response = await self.responses_full_generator(
                request,
                sampling_params,
                empty_async_generator(),
                context,
                model_name,
                tokenizer,
                request_metadata,
                created_time=created_time,
            )
            terminal_event: StreamingResponsesResponse
            if (
                isinstance(final_response, ResponsesResponse)
                and final_response.status == "incomplete"
            ):
                terminal_event = ResponseIncompleteEvent(
                    type="response.incomplete",
                    sequence_number=-1,
                    response=final_response,
                )
            else:
                terminal_event = ResponseCompletedEvent(
                    type="response.completed",
                    sequence_number=-1,
                    response=final_response,
                )
            yield _increment_sequence_number_and_return(terminal_event)