ai-video-fullstack/backend/routes/openai_responses.py

"""Public stateful OpenAI Responses-compatible text endpoints."""

from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator
from typing import Annotated, Any, Literal

import settings
from db.session import SessionLocal
from fastapi import APIRouter, Depends, Header, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse
from loguru import logger
from services.openai_api import (
    DeletedResource,
    OpenAIErrorResponse,
    PUBLIC_API_BEARER,
    PUBLIC_ERROR_RESPONSES,
    REQUEST_ID_RESPONSE_HEADERS,
    openai_error_response,
)
from services.openai_realtime.auth import (
    RealtimeAuthError,
    RealtimeCredential,
    authenticate_bearer,
    bearer_from_authorization,
)
from services.openai_responses.protocol import (
    ResponseRequest,
    ResponsesAPIError,
    new_id,
    normalize_idempotency_key,
    parse_request,
    request_fingerprint,
    response_payload,
)
from services.openai_responses.runner import ResponseRunError, ResponsesTextRunner
from services.openai_responses.schemas import (
    ResponseCreateRequest,
    ResponseInputItemList,
    ResponseObject,
)
from services.openai_responses.state import (
    PreparedResponse,
    delete_response,
    finish_response,
    get_idempotent_response,
    get_response,
    prepare_response,
)
from services.openai_responses.wire import (
    SSEEventSequence,
    assistant_text_item,
    build_output_items,
    output_lifecycle_events,
    stored_input_items,
)


router = APIRouter(
    prefix="/v1/responses",
    tags=["openai-responses"],
    dependencies=[Depends(PUBLIC_API_BEARER)],
)


async def _credential(request: Request) -> RealtimeCredential:
    token = bearer_from_authorization(request.headers.get("authorization"))
    async with SessionLocal() as db:
        credential = await authenticate_bearer(db, token)
    if credential.ephemeral:
        raise RealtimeAuthError("A long-lived API key is required")
    return credential


def _json_error(error: ResponsesAPIError) -> JSONResponse:
    return openai_error_response(
        error.message,
        code=error.code,
        param=error.param,
        status_code=error.status_code,
    )


async def _run_prepared(
    prepared: PreparedResponse,
    *,
    on_delta=None,
    text_item_id: str,
):
    try:
        async with asyncio.timeout(settings.RESPONSES_REQUEST_TIMEOUT_SECONDS):
            result = await ResponsesTextRunner().run(
                prepared.config,
                history_messages=prepared.history_messages,
                current_messages=prepared.current_messages,
                runtime_state=prepared.runtime_state,
                client_tool_results=prepared.client_tool_results,
                on_delta=on_delta,
            )
        return await finish_response(
            prepared.row.id,
            status="completed",
            output_items=build_output_items(
                text=result.text,
                text_item_id=text_item_id,
                function_calls=result.function_calls,
            ),
            context_messages=result.context_messages,
            dynamic_variables=result.dynamic_variables,
            runtime_state=result.runtime_state,
            usage=result.usage,
        )
    except TimeoutError as exc:
        await finish_response(
            prepared.row.id,
            status="failed",
            output_items=[],
            error={"code": "response_timeout", "message": "Response timed out"},
        )
        raise ResponsesAPIError(
            "Response timed out", code="response_timeout", status_code=504
        ) from exc
    except asyncio.CancelledError:
        await finish_response(
            prepared.row.id,
            status="cancelled",
            output_items=[],
            error={"code": "client_disconnected", "message": "Client disconnected"},
        )
        raise
    except ResponseRunError as exc:
        logger.warning(
            f"Responses pipeline failed for {prepared.row.id}: {exc}"
        )
        await finish_response(
            prepared.row.id,
            status="failed",
            output_items=[],
            error={
                "code": "response_failed",
                "message": "Response generation failed",
            },
        )
        raise ResponsesAPIError(
            "Response generation failed",
            code="response_failed",
            status_code=500,
        ) from exc
    except Exception as exc:  # noqa: BLE001
        logger.exception(f"Responses generation failed: {exc}")
        await finish_response(
            prepared.row.id,
            status="failed",
            output_items=[],
            error={"code": "internal_error", "message": "Response generation failed"},
        )
        raise ResponsesAPIError(
            "Response generation failed", code="internal_error", status_code=500
        ) from exc


async def _stream_response(
    prepared: PreparedResponse,
    *,
    text_item_id: str,
) -> AsyncIterator[bytes]:
    queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
    started_text_items: set[str] = set()
    sequence = SSEEventSequence()

    async def on_delta(delta: str) -> None:
        await queue.put({"kind": "delta", "delta": delta})

    async def generate() -> None:
        try:
            row = await _run_prepared(
                prepared,
                on_delta=on_delta,
                text_item_id=text_item_id,
            )
            await queue.put({"kind": "completed", "row": row})
        except ResponsesAPIError as exc:
            await queue.put({"kind": "failed", "error": exc})
        finally:
            await queue.put(None)

    task = asyncio.create_task(generate(), name=f"stream-{prepared.row.id}")
    try:
        created = response_payload(prepared.row)
        yield sequence.encode("response.created", response=created)
        yield sequence.encode("response.in_progress", response=created)
        while True:
            event = await queue.get()
            if event is None:
                break
            if event["kind"] == "delta":
                if text_item_id not in started_text_items:
                    started_text_items.add(text_item_id)
                    pending_item = assistant_text_item(text_item_id, "")
                    pending_item["status"] = "in_progress"
                    yield sequence.encode(
                        "response.output_item.added",
                        output_index=0,
                        item=pending_item,
                    )
                    yield sequence.encode(
                        "response.content_part.added",
                        item_id=text_item_id,
                        output_index=0,
                        content_index=0,
                        part={
                            "type": "output_text",
                            "text": "",
                            "annotations": [],
                        },
                    )
                yield sequence.encode(
                    "response.output_text.delta",
                    item_id=text_item_id,
                    output_index=0,
                    content_index=0,
                    delta=event["delta"],
                    logprobs=[],
                )
                continue
            if event["kind"] == "failed":
                error: ResponsesAPIError = event["error"]
                yield sequence.encode(
                    "error",
                    code=error.code,
                    message=error.message,
                    param=error.param,
                )
                failed = await get_response(
                    prepared.row.id, api_key_id=prepared.row.api_key_id
                )
                yield sequence.encode(
                    "response.failed",
                    response=response_payload(failed),
                )
                continue

            row = event["row"]
            payload = response_payload(row)
            for output_event in output_lifecycle_events(
                row,
                started_text_items=started_text_items,
                replay_text=False,
            ):
                event_type = output_event.pop("type")
                yield sequence.encode(event_type, **output_event)
            yield sequence.encode("response.completed", response=payload)
    finally:
        if not task.done():
            task.cancel()
        await asyncio.gather(task, return_exceptions=True)


async def _stream_stored_response(row) -> AsyncIterator[bytes]:
    """Replay a persisted terminal response for an idempotent streaming call."""

    sequence = SSEEventSequence()
    final_payload = response_payload(row)
    initial_payload = {
        **final_payload,
        "status": "in_progress",
        "completed_at": None,
        "error": None,
        "output": [],
    }
    yield sequence.encode("response.created", response=initial_payload)
    yield sequence.encode("response.in_progress", response=initial_payload)
    if row.status == "completed":
        for output_event in output_lifecycle_events(
            row,
            started_text_items=set(),
            replay_text=True,
        ):
            event_type = output_event.pop("type")
            yield sequence.encode(event_type, **output_event)
        yield sequence.encode("response.completed", response=final_payload)
        return

    error = row.error or {
        "code": "response_cancelled",
        "message": "Response was cancelled",
    }
    yield sequence.encode(
        "error",
        code=str(error.get("code") or "response_failed"),
        message=str(error.get("message") or "Response generation failed"),
        param=None,
    )
    yield sequence.encode("response.failed", response=final_payload)


def _idempotent_error(row) -> JSONResponse | None:
    if row.status == "completed":
        return None
    error = row.error or {}
    code = str(error.get("code") or "response_failed")
    status_code = 504 if code == "response_timeout" else 500
    if row.status == "cancelled":
        code = str(error.get("code") or "response_cancelled")
        status_code = 409
    return openai_error_response(
        str(error.get("message") or f"Response is {row.status}"),
        code=code,
        status_code=status_code,
        headers={"X-Idempotent-Replayed": "true"},
    )


@router.post(
    "",
    response_model=ResponseObject,
    responses={
        200: {
            "description": "A completed Response or a stream of Responses events.",
            "headers": {
                **REQUEST_ID_RESPONSE_HEADERS,
                "X-Idempotent-Replayed": {
                    "description": "Present and true when a stored result is replayed.",
                    "schema": {"type": "string", "enum": ["true"]},
                },
            },
            "content": {
                "text/event-stream": {
                    "schema": {"type": "string"},
                    "example": (
                        "event: response.created\\n"
                        'data: {"type":"response.created",...}\\n\\n'
                    ),
                }
            },
        },
        **PUBLIC_ERROR_RESPONSES,
    },
    summary="Create a stateful model response",
)
async def create_response(
    body: ResponseCreateRequest,
    request: Request,
    idempotency_key_header: Annotated[
        str | None,
        Header(
            alias="Idempotency-Key",
            description="Retry key scoped to the authenticated API key.",
        ),
    ] = None,
):
    try:
        credential = await _credential(request)
        body_value = body.model_dump(exclude_unset=True)
        parsed: ResponseRequest = parse_request(body_value)
        idempotency_key = normalize_idempotency_key(idempotency_key_header)
        fingerprint = request_fingerprint(body_value)
        stored = await get_idempotent_response(
            api_key_id=credential.api_key_id,
            idempotency_key=idempotency_key,
            request_hash=fingerprint,
        )
        if stored is not None:
            if parsed.stream:
                return StreamingResponse(
                    _stream_stored_response(stored),
                    media_type="text/event-stream",
                    headers={
                        "Cache-Control": "no-cache",
                        "X-Accel-Buffering": "no",
                        "X-Idempotent-Replayed": "true",
                    },
                )
            replay_error = _idempotent_error(stored)
            if replay_error is not None:
                return replay_error
            return JSONResponse(
                content=response_payload(stored),
                headers={"X-Idempotent-Replayed": "true"},
            )
        prepared = await prepare_response(
            parsed,
            api_key_id=credential.api_key_id,
            idempotency_key=idempotency_key,
            request_hash=fingerprint if idempotency_key else None,
        )
        text_item_id = new_id("msg")
        if parsed.stream:
            return StreamingResponse(
                _stream_response(prepared, text_item_id=text_item_id),
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        row = await _run_prepared(prepared, text_item_id=text_item_id)
        return response_payload(row)
    except RealtimeAuthError as exc:
        return _json_error(
            ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401)
        )
    except ResponsesAPIError as exc:
        return _json_error(exc)
    except ValueError as exc:
        return _json_error(ResponsesAPIError(str(exc)))


@router.get(
    "/{response_id}",
    response_model=ResponseObject,
    responses={
        200: {"headers": REQUEST_ID_RESPONSE_HEADERS},
        **PUBLIC_ERROR_RESPONSES,
    },
    summary="Retrieve a stored response",
)
async def retrieve_response(response_id: str, request: Request):
    try:
        credential = await _credential(request)
        row = await get_response(response_id, api_key_id=credential.api_key_id)
        return response_payload(row)
    except RealtimeAuthError as exc:
        return _json_error(
            ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401)
        )
    except ResponsesAPIError as exc:
        return _json_error(exc)


@router.get(
    "/{response_id}/input_items",
    response_model=ResponseInputItemList,
    responses={
        200: {"headers": REQUEST_ID_RESPONSE_HEADERS},
        **PUBLIC_ERROR_RESPONSES,
    },
    summary="List input items for a stored response",
)
async def list_response_input_items(
    response_id: str,
    request: Request,
    after: Annotated[str | None, Query()] = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    order: Annotated[Literal["asc", "desc"], Query()] = "desc",
    include: Annotated[list[str] | None, Query()] = None,
):
    try:
        credential = await _credential(request)
        if include:
            raise ResponsesAPIError(
                "include is not supported by this API subset",
                code="unsupported_parameter",
                param="include",
            )
        row = await get_response(response_id, api_key_id=credential.api_key_id)
        items = stored_input_items(row)
        ordered = items if order == "asc" else list(reversed(items))
        start = 0
        if after is not None:
            try:
                start = next(
                    index + 1
                    for index, item in enumerate(ordered)
                    if item["id"] == after
                )
            except StopIteration as exc:
                raise ResponsesAPIError(
                    "after does not identify an input item in this response",
                    code="invalid_cursor",
                    param="after",
                ) from exc
        data = ordered[start : start + limit]
        return {
            "object": "list",
            "data": data,
            "first_id": data[0]["id"] if data else None,
            "last_id": data[-1]["id"] if data else None,
            "has_more": start + len(data) < len(ordered),
        }
    except RealtimeAuthError as exc:
        return _json_error(
            ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401)
        )
    except ResponsesAPIError as exc:
        return _json_error(exc)


@router.delete(
    "/{response_id}",
    response_model=DeletedResource,
    responses={
        200: {"headers": REQUEST_ID_RESPONSE_HEADERS},
        **PUBLIC_ERROR_RESPONSES,
    },
    summary="Delete a stored response",
)
async def remove_response(response_id: str, request: Request):
    try:
        credential = await _credential(request)
        await delete_response(response_id, api_key_id=credential.api_key_id)
        return {"id": response_id, "object": "response", "deleted": True}
    except RealtimeAuthError as exc:
        return _json_error(
            ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401)
        )
    except ResponsesAPIError as exc:
        return _json_error(exc)


@router.post(
    "/{response_id}/cancel",
    response_model=OpenAIErrorResponse,
    status_code=400,
    responses=PUBLIC_ERROR_RESPONSES,
    summary="Reject background cancellation in subset v1",
)
async def reject_response_cancellation(response_id: str, request: Request):
    del response_id
    try:
        await _credential(request)
        raise ResponsesAPIError(
            "Response cancellation requires background mode, which is not supported",
            code="unsupported_operation",
            param="background",
        )
    except RealtimeAuthError as exc:
        return _json_error(
            ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401)
        )
    except ResponsesAPIError as exc:
        return _json_error(exc)