guidellm.mock_server.handlers

class ChatCompletionsHandler:
    """
    Handles OpenAI Chat Completions API requests with realistic LLM simulation.

    Implements the /v1/chat/completions endpoint behavior including request validation,
    response generation, and timing simulation. Supports both streaming and
    non-streaming modes with configurable latency characteristics for comprehensive
    benchmarking. Uses either a mock tokenizer or a real tokenizer for accurate token
    counting and realistic text generation.

    Example:
    ::
        config = MockServerConfig(ttft_ms=100, itl_ms=50)
        handler = ChatCompletionsHandler(config)
        response = await handler.handle(request)
    """

    def __init__(self, config: MockServerConfig) -> None:
        """
        Initialize the Chat Completions handler with server configuration.

        :param config: Mock server configuration containing timing and behavior settings
        """
        self.config = config
        self.tokenizer = (
            MockTokenizer()
            if config.processor is None
            else PreTrainedTokenizer.from_pretrained(config.processor)
        )

    async def handle(self, request: Request) -> HTTPResponse:
        """
        Process incoming chat completion requests with validation and routing.

        Validates the request payload, handles errors gracefully, and routes to
        appropriate streaming or non-streaming response handlers based on the
        request configuration.

        :param request: Sanic HTTP request containing chat completion parameters
        :return: HTTP response with completion data or error information
        :raises ValidationError: When request payload fails validation
        :raises JSONDecodeError: When request contains invalid JSON
        """
        try:
            # Parse and validate request
            req_data = ChatCompletionsRequest(**request.json)
        except ValidationError as exc:
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message=f"Invalid request: {str(exc)}",
                        type="invalid_request_error",
                        code="invalid_request",
                    )
                ).model_dump(),
                status=400,
            )
        except (json.JSONDecodeError, TypeError):
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message="Invalid JSON in request body",
                        type="invalid_request_error",
                        code="invalid_json",
                    )
                ).model_dump(),
                status=400,
            )

        # Handle streaming vs non-streaming
        if req_data.stream:
            return await self._handle_stream(req_data)
        else:
            return await self._handle_non_stream(req_data)

    def _should_return_tool_calls(self, req: ChatCompletionsRequest) -> bool:
        """Check if this request expects a tool call response.

        Returns True when tool_choice is "required", which is what GuideLLM
        sets on client_tool_call turns.
        """
        return req.tool_choice == "required" and bool(req.tools)

    def _build_tool_calls(self, req: ChatCompletionsRequest) -> list[dict[str, Any]]:
        """Build a mock tool_calls array from the request's tool definitions.

        Uses the first tool's function name and returns empty arguments.
        """
        func_name = "mock_function"
        if req.tools:
            first_tool = req.tools[0]
            if "function" in first_tool:
                func_name = first_tool["function"].get("name", func_name)

        return [
            {
                "id": f"call_{uuid.uuid4().hex[:24]}",
                "type": "function",
                "function": {"name": func_name, "arguments": "{}"},
            }
        ]

    async def _handle_non_stream(self, req: ChatCompletionsRequest) -> HTTPResponse:
        """
        Generate complete non-streaming chat completion response.

        Simulates realistic LLM behavior with TTFT and ITL delays, generates
        appropriate token counts, and returns a complete response with usage
        statistics and generated content. When tool_choice is "required",
        returns tool_calls instead of text content.

        :param req: Validated chat completion request parameters
        :return: Complete HTTP response with generated completion data
        """
        # TTFT delay
        await asyncio.sleep(
            sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
        )

        # Token counts
        prompt_text = self.tokenizer.apply_chat_template(req.messages)
        prompt_tokens = len(self.tokenizer(prompt_text))  # type: ignore[arg-type]
        max_tokens = req.max_completion_tokens or req.max_tokens or math.inf
        completion_tokens_count = min(
            sample_number(self.config.output_tokens, self.config.output_tokens_std),
            max_tokens,
        )

        # ITL delay
        itl_delay = 0.0
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))
        for _ in range(int(completion_tokens_count) - 1):
            itl_delay += next(delays_iter)
        await asyncio.sleep(itl_delay / 1000.0)

        # Build message and finish_reason based on whether this is a tool call turn
        if self._should_return_tool_calls(req):
            choice = ChatCompletionChoice(
                index=0,
                message=ChatMessage(
                    role="assistant",
                    content=None,
                    tool_calls=self._build_tool_calls(req),
                ),
                finish_reason="tool_calls",
            )
        else:
            choice = ChatCompletionChoice(
                index=0,
                message=ChatMessage(
                    role="assistant",
                    content=create_fake_text(
                        int(completion_tokens_count), self.tokenizer
                    ),
                ),
                finish_reason="stop",
            )

        chat_response = ChatCompletionsResponse(
            id=f"chatcmpl-{uuid.uuid4().hex[:29]}",
            model=req.model,
            choices=[choice],
            usage=Usage(
                prompt_tokens=prompt_tokens,
                completion_tokens=int(completion_tokens_count),
            ),
            system_fingerprint=f"fp_{uuid.uuid4().hex[:10]}",
        )

        return response.json(chat_response.model_dump())

    async def _handle_stream(self, req: ChatCompletionsRequest) -> HTTPResponse:
        """
        Generate streaming chat completion response with real-time token delivery.

        Creates a streaming response that delivers tokens incrementally with
        realistic timing delays. Supports optional usage statistics in the final
        stream chunk when requested via stream_options. When tool_choice is
        "required", streams tool call deltas instead of content.

        :param req: Validated chat completion request with streaming enabled
        :return: Streaming HTTP response delivering tokens with proper timing
        """
        is_tool_call = self._should_return_tool_calls(req)

        async def generate_stream(stream_response):
            completion_id = f"chatcmpl-{uuid.uuid4().hex[:29]}"

            # TTFT delay
            await asyncio.sleep(
                sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
            )

            # Token counts
            prompt_text = self.tokenizer.apply_chat_template(req.messages)
            prompt_tokens = len(self.tokenizer(prompt_text))  # type: ignore[arg-type]
            max_tokens = req.max_completion_tokens or req.max_tokens or math.inf
            completion_tokens_count = int(
                min(
                    sample_number(
                        self.config.output_tokens, self.config.output_tokens_std
                    ),
                    max_tokens,
                )
            )

            if is_tool_call:
                await self._stream_tool_call(
                    stream_response,
                    req,
                    completion_id,
                    prompt_tokens,
                    completion_tokens_count,
                )
            else:
                await self._stream_content(
                    stream_response,
                    req,
                    completion_id,
                    prompt_tokens,
                    completion_tokens_count,
                )

            # End stream
            await stream_response.write("data: [DONE]\n\n")

        return ResponseStream(  # type: ignore[return-value]
            generate_stream,
            content_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

    async def _stream_content(
        self,
        stream_response,
        req,
        completion_id,
        prompt_tokens,
        completion_tokens_count,
    ):
        """Stream text content tokens with ITL delays."""
        tokens = create_fake_tokens_str(completion_tokens_count, self.tokenizer)
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))

        for index, token in enumerate(tokens):
            if index > 0:
                itl_delay = next(delays_iter)
                await asyncio.sleep(itl_delay / 1000.0)

            chunk_data = {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": req.model,
                "choices": [
                    {
                        "index": 0,
                        "delta": {"content": token},
                        "finish_reason": None,
                    }
                ],
            }
            await stream_response.write(f"data: {json.dumps(chunk_data)}\n\n")

        # Send final chunk with finish reason
        final_chunk = {
            "id": completion_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": req.model,
            "choices": [
                {
                    "index": 0,
                    "delta": {},
                    "finish_reason": "stop",
                }
            ],
        }
        await stream_response.write(f"data: {json.dumps(final_chunk)}\n\n")

        # Send usage if requested
        if req.stream_options and req.stream_options.include_usage:
            usage_chunk = {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": req.model,
                "choices": [],
                "usage": {
                    "prompt_tokens": prompt_tokens,
                    "completion_tokens": completion_tokens_count,
                    "total_tokens": prompt_tokens + completion_tokens_count,
                },
            }
            await stream_response.write(f"data: {json.dumps(usage_chunk)}\n\n")

    async def _stream_tool_call(
        self,
        stream_response,
        req,
        completion_id,
        prompt_tokens,
        completion_tokens_count,
    ):
        """Stream tool call deltas matching the OpenAI streaming tool call format.

        Emits: first chunk with tool call id/name, argument chunks with ITL
        delays, final chunk with finish_reason="tool_calls", and optional usage.
        """
        tool_calls = self._build_tool_calls(req)
        tool_call = tool_calls[0]
        func_name = tool_call["function"]["name"]
        arguments = tool_call["function"]["arguments"]

        # Split arguments into character chunks to simulate streaming
        # Use at least 2 chunks for realistic ITL measurement
        arg_chunks = list(arguments) if len(arguments) > 1 else [arguments]

        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))

        # First chunk: tool call with id, type, and function name
        first_chunk = {
            "id": completion_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": req.model,
            "choices": [
                {
                    "index": 0,
                    "delta": {
                        "tool_calls": [
                            {
                                "index": 0,
                                "id": tool_call["id"],
                                "type": "function",
                                "function": {
                                    "name": func_name,
                                    "arguments": "",
                                },
                            }
                        ]
                    },
                    "finish_reason": None,
                }
            ],
        }
        await stream_response.write(f"data: {json.dumps(first_chunk)}\n\n")

        # Argument chunks with ITL delays
        for arg_piece in arg_chunks:
            itl_delay = next(delays_iter)
            await asyncio.sleep(itl_delay / 1000.0)

            arg_chunk = {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": req.model,
                "choices": [
                    {
                        "index": 0,
                        "delta": {
                            "tool_calls": [
                                {
                                    "index": 0,
                                    "function": {"arguments": arg_piece},
                                }
                            ]
                        },
                        "finish_reason": None,
                    }
                ],
            }
            await stream_response.write(f"data: {json.dumps(arg_chunk)}\n\n")

        # Final chunk with finish reason
        final_chunk = {
            "id": completion_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": req.model,
            "choices": [
                {
                    "index": 0,
                    "delta": {},
                    "finish_reason": "tool_calls",
                }
            ],
        }
        await stream_response.write(f"data: {json.dumps(final_chunk)}\n\n")

        # Send usage if requested
        if req.stream_options and req.stream_options.include_usage:
            usage_chunk = {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": req.model,
                "choices": [],
                "usage": {
                    "prompt_tokens": prompt_tokens,
                    "completion_tokens": completion_tokens_count,
                    "total_tokens": prompt_tokens + completion_tokens_count,
                },
            }
            await stream_response.write(f"data: {json.dumps(usage_chunk)}\n\n")

class CompletionsHandler:
    """
    Handler for the OpenAI /v1/completions endpoint in the mock server.

    This handler simulates the legacy OpenAI completions API by processing incoming
    requests and generating responses with configurable timing and token generation
    patterns. It supports both streaming and non-streaming modes, applying realistic
    timing delays (TTFT and ITL) to mimic actual LLM behavior for benchmarking.

    Example:
    ::
        config = MockServerConfig(ttft_ms=100, itl_ms=50)
        handler = CompletionsHandler(config)
        response = await handler.handle(sanic_request)
    """

    def __init__(self, config: MockServerConfig) -> None:
        """
        Initialize the completions handler with configuration settings.

        :param config: Mock server configuration containing timing parameters
            and tokenizer settings
        """
        self.config = config
        self.tokenizer = (
            MockTokenizer()
            if config.processor is None
            else PreTrainedTokenizer.from_pretrained(config.processor)
        )

    async def handle(self, request: Request) -> HTTPResponse:
        """
        Process a completions request and return the appropriate response.

        Validates the incoming request, determines whether to use streaming or
        non-streaming mode, and delegates to the appropriate handler method.

        :param request: Sanic request object containing the completions request data
        :return: HTTP response with completion data or error information
        :raises ValidationError: When request validation fails
        :raises json.JSONDecodeError: When request JSON is malformed
        """
        try:
            # Parse and validate request
            req_data = CompletionsRequest(**request.json)
        except ValidationError as e:
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message=f"Invalid request: {str(e)}",
                        type="invalid_request_error",
                        code="invalid_request",
                    )
                ).model_dump(),
                status=400,
            )
        except (json.JSONDecodeError, TypeError):
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message="Invalid JSON in request body",
                        type="invalid_request_error",
                        code="invalid_json",
                    )
                ).model_dump(),
                status=400,
            )

        # Handle streaming vs non-streaming
        if req_data.stream:
            return await self._handle_stream(req_data)
        else:
            return await self._handle_non_stream(req_data)

    async def _handle_non_stream(self, req: CompletionsRequest) -> HTTPResponse:
        """
        Generate a non-streaming completion response.

        Simulates TTFT and ITL delays, generates appropriate token counts, and returns
        a complete response with the generated text and usage statistics.

        :param req: Validated completions request containing prompt and parameters
        :return: JSON HTTP response with completion text and usage data
        :raises NotImplementedError: When batch processing is requested
        """
        if isinstance(req.prompt, list):
            raise NotImplementedError("Batch processing is not supported.")

        # TTFT delay
        await asyncio.sleep(
            sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
        )

        # Token counts
        prompt_tokens = len(self.tokenizer(req.prompt))
        max_tokens = req.max_tokens or math.inf
        completion_tokens_count = int(
            min(
                sample_number(self.config.output_tokens, self.config.output_tokens_std),
                max_tokens,
            )
            if req.stop
            else max_tokens
        )

        # ITL delay
        itl_delay = 0.0
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))
        for _ in range(int(completion_tokens_count) - 1):
            itl_delay += next(delays_iter)
        await asyncio.sleep(itl_delay / 1000.0)

        # Response
        completion_response = CompletionsResponse(
            id=f"cmpl-{uuid.uuid4().hex[:29]}",
            model=req.model,
            choices=[
                CompletionChoice(
                    text=create_fake_text(completion_tokens_count, self.tokenizer),
                    index=0,
                    finish_reason="stop",
                )
            ],
            usage=Usage(
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens_count,
            ),
            system_fingerprint=f"fp_{uuid.uuid4().hex[:10]}",
        )

        return response.json(completion_response.model_dump())

    async def _handle_stream(self, req: CompletionsRequest) -> HTTPResponse:
        """
        Generate a streaming completion response.

        Creates a server-sent events stream that delivers tokens incrementally with
        realistic timing delays between each token. Includes usage statistics if
        requested and properly terminates the stream.

        :param req: Validated completions request containing prompt and streaming
            options
        :return: ResponseStream object that generates server-sent events
        """

        async def generate_stream(stream_response):
            completion_id = f"cmpl-{uuid.uuid4().hex[:29]}"

            # TTFT delay
            await asyncio.sleep(
                sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
            )

            # Token counts
            prompt_tokens = len(self.tokenizer(req.prompt))
            max_tokens = req.max_tokens or math.inf
            completion_tokens_count = int(
                min(
                    sample_number(
                        self.config.output_tokens, self.config.output_tokens_std
                    ),
                    max_tokens,
                )
                if req.stop
                else max_tokens
            )

            # Send tokens
            tokens = create_fake_tokens_str(completion_tokens_count, self.tokenizer)
            delays_iter = iter(
                times_generator(self.config.itl_ms, self.config.itl_ms_std)
            )

            for index, token in enumerate(tokens):
                if index > 0:
                    itl_delay = next(delays_iter)
                    await asyncio.sleep(itl_delay / 1000.0)

                chunk_data = {
                    "id": completion_id,
                    "object": "text_completion",
                    "created": int(time.time()),
                    "model": req.model,
                    "choices": [
                        {
                            "text": token,
                            "index": index,
                            "finish_reason": None,
                        }
                    ],
                }
                await stream_response.write(f"data: {json.dumps(chunk_data)}\n\n")

            # Send final chunk with finish reason
            final_chunk = {
                "id": completion_id,
                "object": "text_completion",
                "created": int(time.time()),
                "model": req.model,
                "choices": [
                    {
                        "text": "",
                        "index": index,
                        "finish_reason": "stop",
                    }
                ],
            }
            await stream_response.write(f"data: {json.dumps(final_chunk)}\n\n")

            # Send usage if requested
            if req.stream_options and req.stream_options.include_usage:
                usage_chunk = {
                    "id": completion_id,
                    "object": "text_completion",
                    "created": int(time.time()),
                    "model": req.model,
                    "choices": [],
                    "usage": {
                        "prompt_tokens": prompt_tokens,
                        "completion_tokens": completion_tokens_count,
                        "total_tokens": prompt_tokens + completion_tokens_count,
                    },
                }
                await stream_response.write(f"data: {json.dumps(usage_chunk)}\n\n")

            # End stream
            await stream_response.write("data: [DONE]\n\n")

        return ResponseStream(  # type: ignore[return-value]
            generate_stream,
            content_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

class ResponsesHandler:
    """
    Handles OpenAI Responses API requests with realistic LLM simulation.

    Implements the /v1/responses endpoint behavior including request validation,
    response generation, and timing simulation. Supports both streaming and
    non-streaming modes with configurable latency characteristics.
    """

    def __init__(self, config: MockServerConfig) -> None:
        self.config = config
        self.tokenizer = (
            MockTokenizer()
            if config.processor is None
            else PreTrainedTokenizer.from_pretrained(config.processor)
        )

    def _extract_input_text(self, req: ResponsesRequest) -> str:
        """Extract plain text from the input field for tokenization."""
        parts: list[str] = []
        if req.instructions:
            parts.append(req.instructions)

        if isinstance(req.input, str):
            parts.append(req.input)
        else:
            for item in req.input:
                content = item.get("content", "")
                if isinstance(content, str):
                    parts.append(content)
                elif isinstance(content, list):
                    for part in content:
                        if isinstance(part, dict) and part.get("type") in (
                            "input_text",
                            "text",
                        ):
                            parts.append(part.get("text", ""))
        return " ".join(parts)

    def _build_response_object(
        self,
        *,
        response_id: str,
        model: str,
        status: str,
        output: list,
        usage: ResponsesUsage | None = None,
    ) -> dict:
        return {
            "id": response_id,
            "object": "response",
            "created_at": int(time.time()),
            "status": status,
            "model": model,
            "output": output,
            "usage": usage.model_dump() if usage else None,
        }

    def _should_return_tool_calls(self, req: ResponsesRequest) -> bool:
        """Check if this request expects a function_call response.

        Returns True when tool_choice is "required", which is what GuideLLM
        sets on client_tool_call turns.
        """
        return req.tool_choice == "required" and bool(req.tools)

    def _build_function_call_item(
        self,
        req: ResponsesRequest,
    ) -> dict[str, Any]:
        """Build a mock function_call output item from the request's tools.

        Uses the first tool's name and returns empty arguments.
        """
        func_name = "mock_function"
        if req.tools:
            first_tool = req.tools[0]
            func_name = first_tool.get("name", func_name)

        call_id = f"call_{uuid.uuid4().hex[:24]}"
        return {
            "type": "function_call",
            "id": f"fc_{uuid.uuid4().hex[:24]}",
            "call_id": call_id,
            "name": func_name,
            "arguments": "{}",
            "status": "completed",
        }

    async def handle(self, request: Request) -> HTTPResponse:
        try:
            req_data = ResponsesRequest(**request.json)
        except ValidationError as exc:
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message=f"Invalid request: {str(exc)}",
                        type="invalid_request_error",
                        code="invalid_request",
                    )
                ).model_dump(),
                status=400,
            )
        except (json.JSONDecodeError, TypeError):
            return response.json(
                ErrorResponse(
                    error=ErrorDetail(
                        message="Invalid JSON in request body",
                        type="invalid_request_error",
                        code="invalid_json",
                    )
                ).model_dump(),
                status=400,
            )

        if req_data.stream:
            return await self._handle_stream(req_data)
        else:
            return await self._handle_non_stream(req_data)

    async def _handle_non_stream(self, req: ResponsesRequest) -> HTTPResponse:
        await asyncio.sleep(
            sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
        )

        input_text = self._extract_input_text(req)
        input_tokens = len(self.tokenizer(input_text))  # type: ignore[arg-type]
        max_tokens = req.max_output_tokens or math.inf
        output_tokens_count = min(
            sample_number(self.config.output_tokens, self.config.output_tokens_std),
            max_tokens,
        )

        itl_delay = 0.0
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))
        for _ in range(int(output_tokens_count) - 1):
            itl_delay += next(delays_iter)
        await asyncio.sleep(itl_delay / 1000.0)

        response_id = f"resp_{uuid.uuid4().hex[:29]}"

        if self._should_return_tool_calls(req):
            output = [self._build_function_call_item(req)]
        else:
            msg_id = f"msg_{uuid.uuid4().hex[:29]}"
            generated_text = create_fake_text(int(output_tokens_count), self.tokenizer)
            output = [
                {
                    "type": "message",
                    "id": msg_id,
                    "role": "assistant",
                    "status": "completed",
                    "content": [
                        {"type": "output_text", "text": generated_text},
                    ],
                }
            ]

        resp = ResponsesResponse(
            id=response_id,
            status="completed",
            model=req.model,
            output=output,
            usage=ResponsesUsage(
                input_tokens=input_tokens,
                output_tokens=int(output_tokens_count),
            ),
        )

        return response.json(resp.model_dump())

    async def _handle_stream(self, req: ResponsesRequest) -> HTTPResponse:
        is_tool_call = self._should_return_tool_calls(req)

        async def generate_stream(stream_response):
            response_id = f"resp_{uuid.uuid4().hex[:29]}"

            await asyncio.sleep(
                sample_number(self.config.ttft_ms, self.config.ttft_ms_std) / 1000.0
            )

            input_text = self._extract_input_text(req)
            input_tokens = len(self.tokenizer(input_text))  # type: ignore[arg-type]
            max_tokens = req.max_output_tokens or math.inf
            output_tokens_count = int(
                min(
                    sample_number(
                        self.config.output_tokens, self.config.output_tokens_std
                    ),
                    max_tokens,
                )
            )

            seq = 0

            def _write_event(event_type: str, data: dict) -> str:
                return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"

            empty_response = self._build_response_object(
                response_id=response_id,
                model=req.model,
                status="in_progress",
                output=[],
            )

            await stream_response.write(
                _write_event(
                    "response.created",
                    {
                        "type": "response.created",
                        "response": empty_response,
                        "sequence_number": seq,
                    },
                )
            )
            seq += 1

            await stream_response.write(
                _write_event(
                    "response.in_progress",
                    {
                        "type": "response.in_progress",
                        "response": empty_response,
                        "sequence_number": seq,
                    },
                )
            )
            seq += 1

            if is_tool_call:
                done_item, seq = await self._stream_function_call(
                    stream_response,
                    req,
                    _write_event,
                    seq,
                )
            else:
                done_item, seq = await self._stream_text_content(
                    stream_response,
                    _write_event,
                    seq,
                    output_tokens_count,
                )

            usage = ResponsesUsage(
                input_tokens=input_tokens,
                output_tokens=output_tokens_count,
            )
            completed_response = self._build_response_object(
                response_id=response_id,
                model=req.model,
                status="completed",
                output=[done_item],
                usage=usage,
            )
            await stream_response.write(
                _write_event(
                    "response.completed",
                    {
                        "type": "response.completed",
                        "response": completed_response,
                        "sequence_number": seq,
                    },
                )
            )

        return ResponseStream(  # type: ignore[return-value]
            generate_stream,
            content_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

    async def _stream_text_content(
        self,
        stream_response,
        _write_event,
        seq,
        output_tokens_count,
    ) -> tuple[dict, int]:
        """Stream text content tokens and return the completed message item."""
        msg_id = f"msg_{uuid.uuid4().hex[:29]}"

        message_item = {
            "type": "message",
            "id": msg_id,
            "role": "assistant",
            "status": "in_progress",
            "content": [],
        }
        await stream_response.write(
            _write_event(
                "response.output_item.added",
                {
                    "type": "response.output_item.added",
                    "output_index": 0,
                    "item": message_item,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        content_part = {"type": "output_text", "text": ""}
        await stream_response.write(
            _write_event(
                "response.content_part.added",
                {
                    "type": "response.content_part.added",
                    "item_id": msg_id,
                    "output_index": 0,
                    "content_index": 0,
                    "part": content_part,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        tokens = create_fake_tokens_str(output_tokens_count, self.tokenizer)
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))

        for index, token in enumerate(tokens):
            if index > 0:
                itl_delay = next(delays_iter)
                await asyncio.sleep(itl_delay / 1000.0)

            await stream_response.write(
                _write_event(
                    "response.output_text.delta",
                    {
                        "type": "response.output_text.delta",
                        "item_id": msg_id,
                        "output_index": 0,
                        "content_index": 0,
                        "delta": token,
                        "sequence_number": seq,
                    },
                )
            )
            seq += 1

        full_text = "".join(tokens)
        await stream_response.write(
            _write_event(
                "response.output_text.done",
                {
                    "type": "response.output_text.done",
                    "item_id": msg_id,
                    "output_index": 0,
                    "content_index": 0,
                    "text": full_text,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        done_message = {
            "type": "message",
            "id": msg_id,
            "role": "assistant",
            "status": "completed",
            "content": [{"type": "output_text", "text": full_text}],
        }
        await stream_response.write(
            _write_event(
                "response.output_item.done",
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": done_message,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        return done_message, seq

    async def _stream_function_call(
        self,
        stream_response,
        req,
        _write_event,
        seq,
    ) -> tuple[dict, int]:
        """Stream a function_call item with argument deltas and ITL delays."""
        fc_item = self._build_function_call_item(req)
        arguments = fc_item["arguments"]

        # Emit output_item.added with the function_call item (empty arguments)
        added_item = {
            "type": "function_call",
            "id": fc_item["id"],
            "call_id": fc_item["call_id"],
            "name": fc_item["name"],
            "arguments": "",
            "status": "in_progress",
        }
        await stream_response.write(
            _write_event(
                "response.output_item.added",
                {
                    "type": "response.output_item.added",
                    "output_index": 0,
                    "item": added_item,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        # Stream argument chunks with ITL delays
        arg_chunks = list(arguments) if len(arguments) > 1 else [arguments]
        delays_iter = iter(times_generator(self.config.itl_ms, self.config.itl_ms_std))

        for chunk in arg_chunks:
            itl_delay = next(delays_iter)
            await asyncio.sleep(itl_delay / 1000.0)

            await stream_response.write(
                _write_event(
                    "response.function_call_arguments.delta",
                    {
                        "type": "response.function_call_arguments.delta",
                        "output_index": 0,
                        "delta": chunk,
                        "sequence_number": seq,
                    },
                )
            )
            seq += 1

        # Emit arguments done
        await stream_response.write(
            _write_event(
                "response.function_call_arguments.done",
                {
                    "type": "response.function_call_arguments.done",
                    "output_index": 0,
                    "arguments": arguments,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        # Emit output_item.done with the completed function_call item
        done_item = {
            "type": "function_call",
            "id": fc_item["id"],
            "call_id": fc_item["call_id"],
            "name": fc_item["name"],
            "arguments": arguments,
            "status": "completed",
        }
        await stream_response.write(
            _write_event(
                "response.output_item.done",
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": done_item,
                    "sequence_number": seq,
                },
            )
        )
        seq += 1

        return done_item, seq