import asyncio
import json
import time
from collections.abc import AsyncIterator, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NamedTuple, cast, get_args
from uuid import uuid4

import fastapi
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from starlette.websockets import WebSocket, WebSocketDisconnect

from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import ModifyResponseException
from litellm.llms.base_llm.guardrail_translation.utils import (
    blocked_responses_api_usage as _blocked_responses_api_usage,
)
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import (
    UserAPIKeyAuth,
    user_api_key_auth,
    user_api_key_auth_websocket,
)
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.http_parsing_utils import (
    _read_request_body,
    _safe_set_request_parsed_body,
)
from litellm.types.llms.openai import REASONING_EFFORT, ResponsesAPIResponse
from litellm.types.responses.main import DeleteResponseResult

if TYPE_CHECKING:
    from litellm.router import Router

router: Final = APIRouter()

_user_api_key_auth_dep: Final = Depends(user_api_key_auth)
_RESPONSES_TAGS: Final = ["responses"]  # mutable-ok: fastapi's route signature requires List[str] tags

_TOOL_PAYLOAD_KEYS: Final[Mapping[str, tuple[str, ...]]] = MappingProxyType(
    {
        "custom": ("name", "description", "format"),
        "function": ("name", "description", "parameters", "strict"),
    }
)
_EMPTY_TOOL_PAYLOAD: Final[Mapping[str, Any]] = MappingProxyType({})


def _convert_tool_payload_value(key: str, value: object, *, to_chat: bool) -> object:
    if key != "format" or not isinstance(value, dict):
        return value
    from litellm.litellm_core_utils.prompt_templates.common_utils import (
        convert_custom_tool_format_to_chat_shape,
        convert_custom_tool_format_to_responses_shape,
    )

    convert = convert_custom_tool_format_to_chat_shape if to_chat else convert_custom_tool_format_to_responses_shape
    return convert(value)


def _convert_tool_envelope(obj: object, *, to_chat: bool) -> object:
    if not isinstance(obj, dict):
        return obj
    tool_type: Final = obj.get("type")
    payload_keys: Final = _TOOL_PAYLOAD_KEYS.get(tool_type)
    if payload_keys is None:
        return obj
    nested: Final = obj.get(tool_type)
    nested_source: Final = nested if isinstance(nested, dict) else _EMPTY_TOOL_PAYLOAD
    payload: Final = {  # mutable-ok: tool entries are embedded verbatim in the JSON request body
        key: _convert_tool_payload_value(key, nested_source[key] if key in nested_source else obj[key], to_chat=to_chat)
        for key in payload_keys
        if key in nested_source or key in obj
    }
    if "name" not in payload:
        return obj
    return {"type": tool_type, tool_type: payload} if to_chat else {"type": tool_type, **payload}  # mutable-ok: same


def _normalize_tool_dialect(
    data: dict, *, to_chat: bool
) -> dict:  # mutable-ok: the parsed request body contract is a plain dict
    tools: Final = data.get("tools")
    tool_choice: Final = data.get("tool_choice")
    normalized_tools: Final = (
        [
            _convert_tool_envelope(tool, to_chat=to_chat) for tool in tools
        ]  # mutable-ok: body's tools stays a plain JSON list
        if isinstance(tools, list)
        else tools
    )
    normalized_choice: Final = _convert_tool_envelope(tool_choice, to_chat=to_chat)
    if normalized_tools == tools and normalized_choice == tool_choice:
        return data
    replaceable: Final = (("tools", normalized_tools), ("tool_choice", normalized_choice))
    return {**data, **{key: value for key, value in replaceable if key in data}}  # mutable-ok: plain body dict


def _is_chat_completions_body(data: Mapping[str, Any]) -> bool:
    messages: Final = data.get("messages")
    if isinstance(messages, list) and messages:
        return True
    return "messages" in data and "input" not in data


_CURSOR_THINKING_SEPARATOR: Final = "-thinking-"
_CURSOR_FAST_SUFFIX: Final = "-fast"
_CURSOR_THINKING_LEVELS: Final[frozenset[str]] = frozenset(get_args(REASONING_EFFORT))


class _CursorModelVariant(NamedTuple):
    base_model: str
    reasoning_effort: str | None


def _parse_cursor_model_variant(model: str) -> _CursorModelVariant:
    stripped: Final = model.removesuffix(_CURSOR_FAST_SUFFIX)
    base, separator, level = stripped.rpartition(_CURSOR_THINKING_SEPARATOR)
    if separator and base and level in _CURSOR_THINKING_LEVELS:
        return _CursorModelVariant(base, level)
    return _CursorModelVariant(stripped, None)


def _router_can_serve(model: str, llm_router: "Router | None") -> bool:
    if llm_router is None:
        return False
    if llm_router.is_recognized_model(model):
        return True
    if model in llm_router.team_public_model_names:
        return True
    return bool(llm_router.pattern_router.get_pattern(model))


def _resolve_cursor_model_variant(
    data: dict, llm_router: "Router | None"
) -> dict:  # mutable-ok: the parsed request body contract is a plain dict
    model: Final = data.get("model")
    if not isinstance(model, str) or _router_can_serve(model, llm_router):
        return data
    variant: Final = _parse_cursor_model_variant(model)
    if variant.base_model == model or not _router_can_serve(variant.base_model, llm_router):
        return data
    resolved: Final = {**data, "model": variant.base_model}  # mutable-ok: plain body dict
    if variant.reasoning_effort is None:
        return resolved
    if _is_chat_completions_body(data):
        if "reasoning_effort" in data:
            return resolved
        return {**resolved, "reasoning_effort": variant.reasoning_effort}  # mutable-ok: plain body dict
    reasoning: Final = data.get("reasoning")
    if isinstance(reasoning, dict):
        if reasoning.get("effort"):
            return resolved
        return {**resolved, "reasoning": {**reasoning, "effort": variant.reasoning_effort}}  # mutable-ok: same
    return {**resolved, "reasoning": {"effort": variant.reasoning_effort}}  # mutable-ok: plain body dict


async def _resolve_cursor_model_variant_before_auth(request: Request) -> None:
    from litellm.proxy.proxy_server import llm_router

    try:
        raw_body: Final = await _read_request_body(request=request)
    except (json.JSONDecodeError, ProxyException):
        return
    resolved: Final = _resolve_cursor_model_variant(raw_body, llm_router)
    if resolved is not raw_body:
        _safe_set_request_parsed_body(request=request, parsed_body=resolved)


@router.post(
    "/v1/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/openai/v1/responses",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def responses_api(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses

    Supports background mode with polling_via_cache for partial response retrieval.
    When background=true and polling_via_cache is enabled, returns a polling_id immediately
    and streams the response in the background, updating Redis cache.

    ```bash
    # Normal request
    curl -X POST http://localhost:4000/v1/responses \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer sk-1234" \
    -d '{
        "model": "gpt-4o",
        "input": "Tell me about AI"
    }'

    # Background request with polling
    curl -X POST http://localhost:4000/v1/responses \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer sk-1234" \
    -d '{
        "model": "gpt-4o",
        "input": "Tell me about AI",
        "background": true
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        native_background_mode,
        polling_cache_ttl,
        polling_via_cache_enabled,
        proxy_config,
        proxy_logging_obj,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data = await _read_request_body(request=request)

    # Check if polling via cache should be used for this request
    from litellm.proxy.response_polling.polling_handler import (
        should_use_polling_for_request,
    )

    should_use_polling: Final = should_use_polling_for_request(
        background_mode=data.get("background", False),
        polling_via_cache_enabled=polling_via_cache_enabled,
        redis_cache=redis_usage_cache,
        model=data.get("model", ""),
        llm_router=llm_router,
        native_background_mode=native_background_mode,
    )

    # If polling is enabled, use polling mode
    if should_use_polling:
        from litellm.proxy.response_polling.background_streaming import (
            background_streaming_task,
        )
        from litellm.proxy.response_polling.polling_handler import (
            ResponsePollingHandler,
        )

        verbose_proxy_logger.info("Starting background response with polling for model=%s", data.get("model"))

        # Run pre-call checks (rate limits, guardrails, budget) BEFORE creating
        # polling ID. This ensures rate-limited requests get a synchronous 429
        # instead of a polling ID that immediately fails in the background task.
        processor = ProxyBaseLLMRequestProcessing(data=data)
        try:
            data, _logging_obj = await processor.common_processing_pre_call_logic(
                request=request,
                general_settings=general_settings,
                proxy_logging_obj=proxy_logging_obj,
                user_api_key_dict=user_api_key_dict,
                version=version,
                proxy_config=proxy_config,
                user_model=user_model,
                user_temperature=user_temperature,
                user_request_timeout=user_request_timeout,
                user_max_tokens=user_max_tokens,
                user_api_base=user_api_base,
                model=None,
                route_type="aresponses",
                llm_router=llm_router,
            )
        except Exception as e:
            raise await processor._handle_llm_api_exception(
                e=e,
                user_api_key_dict=user_api_key_dict,
                proxy_logging_obj=proxy_logging_obj,
                version=version,
            )

        # Initialize polling handler with configured TTL (from global config)
        polling_handler: Final = ResponsePollingHandler(
            redis_cache=redis_usage_cache,
            ttl=polling_cache_ttl,  # Global var set at startup
        )

        # Generate polling ID
        polling_id: Final = ResponsePollingHandler.generate_polling_id()

        # Create initial state in Redis
        initial_state: Final = await polling_handler.create_initial_state(
            polling_id=polling_id,
            request_data=data,
        )

        # Start background task to stream and update cache.
        # Pass pre-processed data so the background task skips pre-call logic
        # (rate limits, guardrails already checked above).
        asyncio.create_task(
            background_streaming_task(
                polling_id=polling_id,
                data=data.copy(),
                polling_handler=polling_handler,
                request=request,
                fastapi_response=fastapi_response,
                user_api_key_dict=user_api_key_dict,
                general_settings=general_settings,
                llm_router=llm_router,
                proxy_config=proxy_config,
                proxy_logging_obj=proxy_logging_obj,
                select_data_generator=select_data_generator,
                user_model=user_model,
                user_temperature=user_temperature,
                user_request_timeout=user_request_timeout,
                user_max_tokens=user_max_tokens,
                user_api_base=user_api_base,
                version=version,
            )
        )

        # Return OpenAI Response object format (initial state)
        # https://platform.openai.com/docs/api-reference/responses/object
        return initial_state

    # Normal response flow
    processor = ProxyBaseLLMRequestProcessing(data=data)
    try:
        response: Final = await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aresponses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )

        # Store in managed objects table if background mode is enabled
        if data.get("background") and isinstance(response, ResponsesAPIResponse):
            if response.status in ["queued", "in_progress"]:
                from litellm_enterprise.proxy.hooks.managed_files import (
                    _PROXY_LiteLLMManagedFiles,
                )

                managed_files_obj: Final = cast(
                    _PROXY_LiteLLMManagedFiles | None,
                    proxy_logging_obj.get_proxy_hook("managed_files"),
                )

                if managed_files_obj and llm_router:
                    try:
                        # Get the actual deployment model_id from hidden params
                        hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
                        model_id: Final = hidden_params.get("model_id", None)

                        if not model_id:
                            verbose_proxy_logger.warning(
                                "No model_id found in response hidden params for response %s, skipping managed object storage",
                                response.id,
                            )
                            raise Exception("No model_id found in response hidden params")
                        # Store in managed objects table
                        await managed_files_obj.store_unified_object_id(
                            unified_object_id=response.id,
                            file_object=response,
                            litellm_parent_otel_span=None,
                            model_object_id=response.id,
                            file_purpose="response",
                            user_api_key_dict=user_api_key_dict,
                        )

                        verbose_proxy_logger.info(
                            "Stored background response %s in managed objects table with unified_id=%s",
                            response.id,
                            response.id,
                        )
                    except Exception as e:
                        verbose_proxy_logger.error(
                            "Failed to store background response in managed objects table: %s", e
                        )

        return response
    except ModifyResponseException as e:
        # Guardrail passthrough: return violation message in Responses API format (200)
        _data: Final = e.request_data
        await proxy_logging_obj.post_call_failure_hook(
            user_api_key_dict=user_api_key_dict,
            original_exception=e,
            request_data=_data,
        )

        violation_text: Final = e.message
        response_obj: Final = ResponsesAPIResponse(
            id=f"resp_{uuid4()}",
            object="response",
            created_at=int(time.time()),
            model=e.model or data.get("model"),
            output=cast(Any, [{"content": [{"type": "text", "text": violation_text}]}]),
            status="completed",
            usage=_blocked_responses_api_usage(e.original_response),
        )
        return response_obj
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/cursor/models",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
@router.get(
    "/cursor/v1/models",
    dependencies=(_user_api_key_auth_dep,),
    tags=_RESPONSES_TAGS,
)
async def cursor_model_list(
    user_api_key_dict: UserAPIKeyAuth = _user_api_key_auth_dep,
):
    """
    OpenAI-compatible model listing for the Cursor BYOK base URL.

    Clients pointed at `<proxy>/cursor` as an OpenAI-compatible base URL resolve and
    verify models via `GET {base}/models` (the OpenAI SDK contract). Without this
    route those requests fall through to the Cursor Cloud Agents passthrough, which
    demands a Cursor API key and 401s, so key verification silently fails before any
    chat request is ever sent. Delegates to the standard `/v1/models` handler.
    """
    from litellm.proxy.proxy_server import model_list

    return await model_list(user_api_key_dict=user_api_key_dict)


@router.post(
    "/cursor/chat/completions",
    dependencies=[
        Depends(_resolve_cursor_model_variant_before_auth),
        Depends(user_api_key_auth),
    ],
    tags=["responses"],
)
async def cursor_chat_completions(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Cursor BYOK endpoint. Accepts both request shapes Cursor sends to its OpenAI-compatible
    base URL and always answers in chat completions format.

    Cursor agent mode sends Responses API format bodies (`input`, flat tool defs, `reasoning`,
    custom tools) to the chat/completions path while expecting chat completions responses;
    those are routed through the Responses API pipeline and converted back. Genuine chat
    completions bodies (`messages` present) are routed through the standard chat completions
    pipeline, after normalizing each level of the `tools` array and `tool_choice` to the chat
    completions shapes OpenAI requires. Cursor mixes Responses API shapes into chat bodies
    per level, independently: a flat tool def (`{"type": "custom", "name": "ApplyPatch", ...}`)
    gets nested under `custom`, and a flat grammar format
    (`{"type": "grammar", "definition", "syntax"}`) gets wrapped as
    `{"type": "grammar", "grammar": {...}}` wherever it appears, including inside tool defs
    Cursor already sent pre-nested.

    ```bash
    curl -X POST http://localhost:4000/cursor/chat/completions \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer sk-1234" \
    -d '{
        "model": "gpt-4o",
        "input": [{"role": "user", "content": "Hello"}]
    }'
    Responds back in chat completions format.
    ```
    """
    from litellm.completion_extras.litellm_responses_transformation.handler import (
        responses_api_bridge,
    )
    from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
    from litellm.proxy.proxy_server import (
        async_data_generator,
        chat_completion,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.types.llms.openai import ResponsesAPIResponse
    from litellm.types.utils import ModelResponse

    raw_body: Final = await _read_request_body(request=request)

    if _is_chat_completions_body(raw_body):
        # Genuine chat completions body (Cursor sends these for models whose BYOK it
        # already fixed); delegate so behavior matches /chat/completions exactly.
        # Keyed on messages CONTENT, not key presence: Cursor can send a null or
        # empty messages stub alongside a real agent-mode input array
        normalized: Final = _normalize_tool_dialect(raw_body, to_chat=True)
        if normalized is not raw_body:
            _safe_set_request_parsed_body(request=request, parsed_body=normalized)
        return await chat_completion(
            request=request,
            fastapi_response=fastapi_response,
            model=None,
            user_api_key_dict=user_api_key_dict,
        )

    # OpenAI's Responses API rejects chat-completions-only stream_options
    # (Cursor sends include_usage); usage arrives via response.completed anyway.
    # Rebuild rather than pop: _read_request_body can return the request-scope
    # cached parsed-body dict itself, and removing keys from it corrupts the
    # cache's key snapshot so later readers get an empty body
    body_without_stream_options: Final = {  # mutable-ok: base_process_llm_request mutates the body dict in place
        key: value for key, value in raw_body.items() if key != "stream_options"
    }

    data: Final = _normalize_tool_dialect(body_without_stream_options, to_chat=False)

    processor: Final = ProxyBaseLLMRequestProcessing(data=data)

    def cursor_data_generator(response, user_api_key_dict, request_data, request=None):
        """
        Custom generator that transforms Responses API streaming chunks to chat completion chunks.

        This generator is used for the cursor endpoint to convert Responses API format responses
        to chat completion format that Cursor IDE expects.

        Args:
            response: The streaming Responses API event iterator (router-wrapped or not)
            user_api_key_dict: User API key authentication dict
            request_data: Request data containing model, logging_obj, etc.
            request: The originating FastAPI request, forwarded for disconnect handling

        Returns:
            Async generator that yields SSE-formatted chat completion chunks
        """
        # Any async-iterable here is a Responses API event stream needing conversion.
        # Class-identity checks miss router-wrapped streams (e.g.
        # HiddenParamsAsyncIteratorWrapper around LiteLLMCompletionStreamingIterator),
        # which previously leaked raw Responses events to the client.
        if hasattr(response, "__anext__"):
            # Transform Responses API iterator to chat completion iterator
            # Cast to AsyncIterator[str] since the stream implements __aiter__/__anext__
            completion_stream: Final = responses_api_bridge.transformation_handler.get_model_response_iterator(
                streaming_response=cast(AsyncIterator[str], response),
                sync_stream=False,
                json_mode=False,
            )
            # Wrap in CustomStreamWrapper to get the async generator
            logging_obj: Final = request_data.get("litellm_logging_obj")
            streamwrapper: Final = CustomStreamWrapper(
                completion_stream=completion_stream,
                model=request_data.get("model", ""),
                custom_llm_provider=None,
                logging_obj=logging_obj,
            )
            # Use async_data_generator to format as SSE
            return async_data_generator(
                response=streamwrapper,
                user_api_key_dict=user_api_key_dict,
                request_data=request_data,
                request=request,
            )
        # Otherwise, use the default generator
        return async_data_generator(
            response=response,
            user_api_key_dict=user_api_key_dict,
            request_data=request_data,
            request=request,
        )

    try:
        response: Final = await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aresponses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=cursor_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )

        # Transform non-streaming Responses API response to chat completions format
        if isinstance(response, ResponsesAPIResponse):
            logging_obj: Final = processor.data.get("litellm_logging_obj")
            transformed_response: Final = responses_api_bridge.transformation_handler.transform_response(
                model=processor.data.get("model", ""),
                raw_response=response,
                model_response=ModelResponse(),
                logging_obj=cast(Any, logging_obj),
                request_data=processor.data,
                messages=processor.data.get("input", []),
                optional_params={},
                litellm_params={},
                encoding=None,
                api_key=None,
                json_mode=None,
            )
            return transformed_response

        # Streaming responses are already transformed by cursor_select_data_generator
        return response
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.get(
    "/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.get(
    "/openai/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def get_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Get a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Returns cumulative cached content from background responses
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/get
    
    ```bash
    # Get polling response
    curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123 \
    -H "Authorization: Bearer sk-1234"
    
    # Get provider response
    curl -X GET http://localhost:4000/v1/responses/resp_abc123 \
    -H "Authorization: Bearer sk-1234"
    ```
    """
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response
        if not redis_usage_cache:
            raise HTTPException(
                status_code=500,
                detail="Redis cache not configured. Polling requires Redis.",
            )

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get current state from cache
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(
                status_code=404,
                detail=f"Polling response {response_id} not found or expired",
            )

        # Return the whole state directly (OpenAI Response object format)
        # https://platform.openai.com/docs/api-reference/responses/object
        return state

    # Normal provider response flow
    data: Final = await _read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="aget_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.delete(
    "/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.delete(
    "/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.delete(
    "/openai/v1/responses/{response_id}",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def delete_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Delete a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Deletes from Redis cache
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/delete
    
    ```bash
    curl -X DELETE http://localhost:4000/v1/responses/resp_abc123 \
    -H "Authorization: Bearer sk-1234"
    ```
    """
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response deletion
        if not redis_usage_cache:
            raise HTTPException(status_code=500, detail="Redis cache not configured.")

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get state to verify access
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(status_code=404, detail=f"Polling response {response_id} not found")

        # Delete from cache
        success: Final = await polling_handler.delete_polling(response_id)

        if success:
            return DeleteResponseResult(id=response_id, object="response", deleted=True)
        else:
            raise HTTPException(status_code=500, detail="Failed to delete polling response")

    # Normal provider response flow
    data: Final = await _read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="adelete_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.get(
    "/v1/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.get(
    "/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.get(
    "/openai/v1/responses/{response_id}/input_items",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def get_response_input_items(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """List input items for a response."""
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data: Final = await _read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="alist_input_items",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.post(
    "/v1/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/openai/v1/responses/compact",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def compact_response(
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Compact a response by running a compaction pass over a conversation.
    
    Returns encrypted, opaque items that can be used to reduce context size.
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/compact
    
    ```bash
    curl -X POST http://localhost:4000/v1/responses/compact \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer sk-1234" \
    -d '{
        "model": "gpt-4o",
        "input": [{"role": "user", "content": "Hello"}]
    }'
    ```
    """
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )

    data: Final = await _read_request_body(request=request)
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="acompact_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


@router.post(
    "/v1/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
@router.post(
    "/openai/v1/responses/{response_id}/cancel",
    dependencies=[Depends(user_api_key_auth)],
    tags=["responses"],
)
async def cancel_response(
    response_id: str,
    request: Request,
    fastapi_response: Response,
    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
    """
    Cancel a response by ID.
    
    Supports both:
    - Polling IDs (litellm_poll_*): Cancels background response and updates status in Redis
    - Provider response IDs: Passes through to provider API
    
    Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/cancel
    
    ```bash
    # Cancel polling response
    curl -X POST http://localhost:4000/v1/responses/litellm_poll_abc123/cancel \
    -H "Authorization: Bearer sk-1234"
    
    # Cancel provider response
    curl -X POST http://localhost:4000/v1/responses/resp_abc123/cancel \
    -H "Authorization: Bearer sk-1234"
    ```
    """
    from litellm.proxy.proxy_server import (
        _read_request_body,
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        redis_usage_cache,
        select_data_generator,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler

    # Check if this is a polling ID
    if ResponsePollingHandler.is_polling_id(response_id):
        # Handle polling response cancellation
        if not redis_usage_cache:
            raise HTTPException(status_code=500, detail="Redis cache not configured.")

        polling_handler: Final = ResponsePollingHandler(redis_cache=redis_usage_cache)

        # Get current state to verify it exists
        state: Final = await polling_handler.get_state(response_id)

        if not state:
            raise HTTPException(status_code=404, detail=f"Polling response {response_id} not found")

        # Cancel the polling response (sets status to "cancelled")
        success: Final = await polling_handler.cancel_polling(response_id)

        if success:
            # Fetch the updated state with cancelled status
            updated_state: Final = await polling_handler.get_state(response_id)

            # Return the whole state directly (now with status="cancelled")
            return updated_state
        else:
            raise HTTPException(status_code=500, detail="Failed to cancel polling response")

    # Normal provider response flow
    data: Final = await _read_request_body(request=request)
    data["response_id"] = response_id
    processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        return await processor.base_process_llm_request(
            request=request,
            fastapi_response=fastapi_response,
            user_api_key_dict=user_api_key_dict,
            route_type="acancel_responses",
            proxy_logging_obj=proxy_logging_obj,
            llm_router=llm_router,
            general_settings=general_settings,
            proxy_config=proxy_config,
            select_data_generator=select_data_generator,
            model=None,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            version=version,
        )
    except Exception as e:
        raise await processor._handle_llm_api_exception(
            e=e,
            user_api_key_dict=user_api_key_dict,
            proxy_logging_obj=proxy_logging_obj,
            version=version,
        )


async def _read_ws_model_from_first_frame(
    websocket: WebSocket,
) -> tuple | None:
    """Read the first WS frame and return (model, raw_message), or None on error.

    Sends an appropriate error frame and closes the socket before returning None.
    """
    try:
        first_message: Final = await asyncio.wait_for(websocket.receive_text(), timeout=30)
    except asyncio.TimeoutError:
        await websocket.close(code=1008, reason="Timed out waiting for first message")
        return None
    except WebSocketDisconnect:
        return None
    except Exception:
        verbose_proxy_logger.exception("Responses WebSocket error reading first message")
        await websocket.close(code=1011, reason="Internal server error")
        return None

    try:
        first_event: Final = json.loads(first_message)
    except json.JSONDecodeError:
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "First message is not valid JSON.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="Invalid JSON in first message")
        return None

    if not isinstance(first_event, dict) or first_event.get("type") != "response.create":
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "First message must be a response.create JSON object.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="Invalid first message")
        return None

    model: Final = _extract_model_from_first_ws_event(first_event)
    if not model:
        await websocket.send_text(
            json.dumps(
                {
                    "type": "error",
                    "error": {
                        "type": "invalid_request_error",
                        "message": "No model provided. Supply ?model=<model> in the URL or include 'model' in the first response.create event.",
                    },
                }
            )
        )
        await websocket.close(code=1008, reason="No model provided")
        return None

    return model, first_message


def _extract_model_from_first_ws_event(first_event: Any) -> str | None:
    """Extract model from a response.create WS event, handling flat and nested formats.

    Flat:   {"type": "response.create", "model": "gpt-4o", ...}
    Nested: {"type": "response.create", "response": {"model": "gpt-4o", ...}}
    """
    if not isinstance(first_event, dict):
        return None
    nested: Final = first_event.get("response")
    return (nested.get("model") if isinstance(nested, dict) else None) or first_event.get("model")


async def _enforce_responses_ws_first_frame_model_auth(
    request: Request,
    model: str,
    user_api_key_dict: UserAPIKeyAuth,
    llm_router: Any | None,
) -> None:
    from litellm.proxy.auth.user_api_key_auth import (
        _enforce_key_and_fallback_model_access,
        _run_centralized_common_checks,
    )
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_model_list,
        master_key,
        user_custom_auth,
    )

    request_data: Final = {"model": model}
    route: Final = request.scope.get("path") or "/v1/responses"
    if master_key is None and not (
        general_settings.get("enable_jwt_auth", False)
        or general_settings.get("enable_oauth2_auth", False)
        or general_settings.get("enable_oauth2_proxy_auth", False)
    ):
        return
    if user_custom_auth is not None and not general_settings.get("custom_auth_run_common_checks", False):
        return
    await _enforce_key_and_fallback_model_access(
        valid_token=user_api_key_dict,
        request_data=request_data,
        route=route,
        request=request,
        llm_model_list=llm_model_list,
        llm_router=llm_router,
    )
    await _run_centralized_common_checks(
        user_api_key_auth_obj=user_api_key_dict,
        request=request,
        request_data=request_data,
        route=route,
    )


@router.websocket("/v1/responses")
@router.websocket("/responses")
async def responses_websocket_endpoint(
    websocket: WebSocket,
    model: str | None = fastapi.Query(None, description="The model to use for the responses WebSocket session."),
    user_api_key_dict=Depends(user_api_key_auth_websocket),
):
    """
    Responses API WebSocket mode endpoint.

    Keeps a persistent WebSocket connection for response.create events,
    enabling lower-latency agentic workflows with many tool-call round trips.

    Follows the OpenAI split: the bearer token is validated at connection time
    (before accept); the model is resolved either from the ?model= query param
    or from the first response.create frame, whichever is present.

    See: https://developers.openai.com/api/docs/guides/websocket-mode/
    """
    from litellm.proxy.proxy_server import (
        general_settings,
        llm_router,
        proxy_config,
        proxy_logging_obj,
        user_api_base,
        user_max_tokens,
        user_model,
        user_request_timeout,
        user_temperature,
        version,
    )
    from litellm.proxy.route_llm_request import route_request

    # Accept the WebSocket handshake. Key was already validated by the Depends
    # above; we can safely accept regardless of whether ?model= was supplied.
    requested_protocols: Final = [
        p.strip() for p in (websocket.headers.get("sec-websocket-protocol") or "").split(",") if p.strip()
    ]
    accept_kwargs: Final[dict] = {}
    if requested_protocols:
        accept_kwargs["subprotocol"] = requested_protocols[0]
    await websocket.accept(**accept_kwargs)

    first_message: str | None = None
    if not model:
        result: Final = await _read_ws_model_from_first_frame(websocket)
        if result is None:
            return
        model, first_message = result

    data: dict[str, Any] = {
        "model": model,
        "websocket": websocket,
    }
    if first_message is not None:
        data["first_message"] = first_message

    # Construct a synthetic Request for pre-call processing
    headers_list: Final = list(websocket.scope.get("headers") or [])
    scope: Final[dict[str, Any]] = {
        "type": "http",
        "method": "POST",
        "path": "/v1/responses",
        "headers": headers_list,
    }
    request: Final = Request(scope=scope)
    request._url = websocket.url

    _body_bytes: Final = json.dumps({"model": model}).encode()

    async def return_body():
        return _body_bytes

    request.body = return_body

    # Phase 1: pre-call processing (auth, guardrails, rate limits)
    base_llm_response_processor: Final = ProxyBaseLLMRequestProcessing(data=data)
    try:
        if first_message is not None:
            await _enforce_responses_ws_first_frame_model_auth(
                request=request,
                model=model,
                user_api_key_dict=user_api_key_dict,
                llm_router=llm_router,
            )
        (
            data,
            litellm_logging_obj,
        ) = await base_llm_response_processor.common_processing_pre_call_logic(
            request=request,
            general_settings=general_settings,
            user_api_key_dict=user_api_key_dict,
            version=version,
            proxy_logging_obj=proxy_logging_obj,
            proxy_config=proxy_config,
            user_model=user_model,
            user_temperature=user_temperature,
            user_request_timeout=user_request_timeout,
            user_max_tokens=user_max_tokens,
            user_api_base=user_api_base,
            model=model,
            route_type="_aresponses_websocket",
        )
    except Exception as e:
        verbose_proxy_logger.exception("Responses WebSocket pre-call error")
        try:
            await websocket.send_text(
                json.dumps(
                    {
                        "type": "error",
                        "error": {
                            "type": "invalid_request_error",
                            "message": str(e),
                        },
                    }
                )
            )
        except Exception:
            pass
        await websocket.close(code=1008, reason="Pre-call error")
        return

    # Phase 2: route to upstream provider
    try:
        data["user_api_key_dict"] = user_api_key_dict
        llm_call: Final = await route_request(
            data=data,
            route_type="_aresponses_websocket",
            llm_router=llm_router,
            user_model=user_model,
        )
        await llm_call
    except Exception:
        verbose_proxy_logger.exception("Responses WebSocket error")
        await websocket.close(code=1011, reason="Internal server error")