"""xAI Grok-Imagine video generation backend.

Surface: text-to-video, image-to-video, and reference-to-video through the
unified video provider. xAI edit/extend are exposed through separate tools.

Originally salvaged from PR #10600 by @Jaaneek; reshaped into the
:class:`VideoGenProvider` plugin interface and trimmed to the
generate-only surface.

Authentication: xAI Grok OAuth tokens (preferred — billed against the
user's SuperGrok or X Premium+ subscription) or ``XAI_API_KEY``. Both routes are
resolved through ``tools.xai_http.resolve_xai_http_credentials`` so a
single login covers chat + TTS + image gen + video gen + transcription.
When xAI storage is enabled, the primary ``video`` / ``public_url`` fields are the
stored files-cdn HTTPS link. Pass that public MP4 URL as ``video_url`` for
edit/extend; it is sent to xAI as ``video.url``.
"""

from __future__ import annotations

import asyncio
import base64
import logging
import mimetypes
import os
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import httpx

from agent.video_gen_provider import (
    VideoGenProvider,
    error_response,
    success_response,
)

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

DEFAULT_XAI_BASE_URL = "https://api.x.ai/v1"
DEFAULT_TEXT_TO_VIDEO_MODEL = "grok-imagine-video"
DEFAULT_IMAGE_TO_VIDEO_MODEL = "grok-imagine-video-1.5"
DEFAULT_MODEL = DEFAULT_TEXT_TO_VIDEO_MODEL
DEFAULT_DURATION = 8
DEFAULT_ASPECT_RATIO = "16:9"
DEFAULT_RESOLUTION = "720p"
DEFAULT_TIMEOUT_SECONDS = 240
DEFAULT_POLL_INTERVAL_SECONDS = 5
DEFAULT_EXTEND_DURATION = 6

VALID_ASPECT_RATIOS = {"1:1", "16:9", "9:16", "4:3", "3:4", "3:2", "2:3"}
VALID_RESOLUTIONS = {"480p", "720p"}
MAX_REFERENCE_IMAGES = 7


_MODELS: Dict[str, Dict[str, Any]] = {
    "grok-imagine-video": {
        "display": "Grok Imagine Video",
        "speed": "~60-240s",
        "strengths": "Text-to-video; legacy image-to-video fallback.",
        "price": "see https://docs.x.ai/developers/models/grok-imagine-video",
        "modalities": ["text", "image"],
    },
    "grok-imagine-video-1.5": {
        "display": "Grok Imagine Video 1.5",
        "speed": "~60-240s",
        "strengths": "Latest xAI image-to-video model.",
        "price": "see https://docs.x.ai/developers/pricing",
        "modalities": ["image"],
    },
}

_IMAGE_TO_VIDEO_COMPAT_MODEL_IDS = {
    "grok-imagine-video-1.5-preview",
    "grok-imagine-video-1.5-2026-05-30",
}


# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------


def _resolve_xai_credentials() -> Tuple[str, str]:
    """Return ``(api_key, base_url)`` from the shared xAI credential resolver.

    Order: runtime provider (xai-oauth pool entry) → singleton ``auth.json``
    OAuth tokens → ``XAI_API_KEY`` env var. ``api_key`` is empty when no
    credential source is available; callers must check before using it.
    """
    try:
        from tools.xai_http import resolve_xai_http_credentials

        creds = resolve_xai_http_credentials() or {}
    except Exception as exc:
        logger.debug("xAI credential resolver failed: %s", exc)
        creds = {}

    api_key = str(creds.get("api_key") or os.getenv("XAI_API_KEY", "")).strip()
    base_url = str(
        creds.get("base_url")
        or os.getenv("XAI_BASE_URL")
        or DEFAULT_XAI_BASE_URL
    ).strip().rstrip("/")
    return api_key, base_url


def _xai_user_agent() -> str:
    try:
        from tools.xai_http import hermes_xai_user_agent

        return hermes_xai_user_agent()
    except Exception:
        return "hermes-agent/video_gen"


def _xai_headers(api_key: str) -> Dict[str, str]:
    return {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "User-Agent": _xai_user_agent(),
    }


def _raise_if_blocked_local_input(ref: str) -> None:
    """Refuse to read a local media path that Hermes' read deny-list blocks.

    Thin wrapper over the shared ``agent.file_safety.raise_if_read_blocked``
    chokepoint so xAI video inputs enforce the same credential-store guard as
    the image providers. Fails open if the guard machinery is unavailable
    (defense-in-depth, per the denylist's own framing).
    """
    try:
        from agent.file_safety import raise_if_read_blocked
    except Exception as exc:  # noqa: BLE001 - guard must never break loading
        logger.debug("xAI media input read guard unavailable: %s", exc)
        return
    raise_if_read_blocked(ref)


def _image_ref_to_xai_url(value: str) -> str:
    """Return a URL/data URI accepted by xAI for image inputs."""
    ref = (value or "").strip()
    if not ref:
        return ""
    lower = ref.lower()
    if lower.startswith(("http://", "https://", "data:image/")):
        return ref

    path = Path(ref).expanduser()
    if not path.is_file():
        return ref

    _raise_if_blocked_local_input(ref)

    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    if not mime.startswith("image/"):
        return ref

    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
    return f"data:{mime};base64,{encoded}"


def _image_ref_to_xai_input(value: str) -> Optional[Dict[str, str]]:
    ref = _image_ref_to_xai_url(value)
    if not ref:
        return None
    lower = ref.lower()
    if lower.startswith(("http://", "https://", "data:image/")):
        return {"url": ref}
    return None


def _xai_video_output_urls(
    video: Dict[str, Any],
) -> Tuple[str, Optional[str], Optional[str]]:
    """Return ``(public_video_url, temporary_url, stored_public_url)``.

    ``public_video_url`` is the stored files-cdn HTTPS MP4 (``public_url``) when
    storage is enabled; otherwise xAI's temporary ``video.url``. Pass this value
    as ``video_url`` for edit/extend chaining.
    """
    file_output = video.get("file_output") if isinstance(video.get("file_output"), dict) else {}
    file_output = file_output or {}
    stored_public = file_output.get("public_url")
    stored_public = stored_public.strip() if isinstance(stored_public, str) else None
    temporary = video.get("url")
    temporary = temporary.strip() if isinstance(temporary, str) else None
    public_video_url = stored_public or temporary or ""
    temporary_out = (
        temporary
        if temporary and stored_public and temporary != stored_public
        else None
    )
    return public_video_url, temporary_out, stored_public


def _video_ref_to_xai_url(value: str) -> str:
    """Return a URL/data URI accepted by xAI for video inputs."""
    ref = (value or "").strip()
    if not ref:
        return ""
    lower = ref.lower()
    if lower.startswith(("http://", "https://", "data:video/")):
        return ref

    path = Path(ref).expanduser()
    if not path.is_file():
        return ref

    _raise_if_blocked_local_input(ref)

    mime = mimetypes.guess_type(path.name)[0] or "video/mp4"
    if not mime.startswith("video/"):
        return ref

    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
    return f"data:{mime};base64,{encoded}"


async def _video_input_from_public_url(
    value: str,
    *,
    api_key: str,
    base_url: str,
) -> Optional[Dict[str, str]]:
    """Build xAI ``video`` input using a public HTTPS URL (``url`` field only)."""
    ref = (value or "").strip()
    if not ref:
        return None

    path = Path(ref).expanduser()
    if path.is_file():
        data_ref = _video_ref_to_xai_url(ref)
        return {"url": data_ref} if data_ref else None

    lower = ref.lower()
    if not lower.startswith(("http://", "https://")):
        return None

    return {"url": ref}


def _normalize_reference_images(
    reference_image_urls: Optional[List[str]],
) -> Tuple[Optional[List[Dict[str, str]]], Optional[str]]:
    refs: List[Dict[str, str]] = []
    for url in reference_image_urls or []:
        cleaned = (url or "").strip()
        if not cleaned:
            continue
        normalized = _image_ref_to_xai_input(cleaned)
        if not normalized:
            return None, (
                "reference_image_urls must be public HTTPS URLs or data URIs "
                "(e.g. the `image`/`public_url` from a prior Imagine result)"
            )
        refs.append(normalized)
    return (refs if refs else None), None


def _clamp_duration(
    duration: Optional[int],
    *,
    has_reference_images: bool = False,
    max_seconds: int = 15,
    default: int = DEFAULT_DURATION,
) -> int:
    value = duration if duration is not None else default
    if value < 1:
        value = 1
    if value > max_seconds:
        value = max_seconds
    if has_reference_images and value > 10:
        value = 10
    return value


def _resolve_model_for_modality(
    model: Optional[str],
    *,
    modality: str,
    explicit_model: bool,
) -> str:
    """Select xAI's text/video model without treating config as a prompt override.

    ``grok-imagine-video-1.5`` currently rejects text-only video
    generation, but it is the desired image-to-video backend. Explicit tool
    ``model=`` still wins for users who intentionally request another model.
    """
    requested = (model or "").strip()
    if explicit_model and requested:
        return requested
    if modality == "image":
        return DEFAULT_IMAGE_TO_VIDEO_MODEL
    if requested == DEFAULT_IMAGE_TO_VIDEO_MODEL or requested in _IMAGE_TO_VIDEO_COMPAT_MODEL_IDS:
        return DEFAULT_TEXT_TO_VIDEO_MODEL
    return requested or DEFAULT_TEXT_TO_VIDEO_MODEL


async def _submit(
    client: httpx.AsyncClient,
    payload: Dict[str, Any],
    *,
    api_key: str,
    base_url: str,
    endpoint: str = "generations",
) -> str:
    """POST to one of xAI's async video endpoints and return request_id."""
    response = await client.post(
        f"{base_url}/videos/{endpoint}",
        headers={**_xai_headers(api_key), "x-idempotency-key": str(uuid.uuid4())},
        json=payload,
        timeout=60,
    )
    response.raise_for_status()
    body = response.json()
    request_id = body.get("request_id")
    if not request_id:
        raise RuntimeError("xAI video response did not include request_id")
    return request_id


async def _poll(
    client: httpx.AsyncClient,
    request_id: str,
    *,
    api_key: str,
    base_url: str,
    timeout_seconds: int,
    poll_interval: int,
) -> Dict[str, Any]:
    elapsed = 0.0
    last_status = "queued"
    while elapsed < timeout_seconds:
        response = await client.get(
            f"{base_url}/videos/{request_id}",
            headers=_xai_headers(api_key),
            timeout=30,
        )
        response.raise_for_status()
        body = response.json()
        last_status = (body.get("status") or "").lower()

        if last_status == "done":
            return {"status": "done", "body": body}
        if last_status in {"failed", "error", "expired", "cancelled"}:
            return {"status": last_status, "body": body}

        await asyncio.sleep(poll_interval)
        elapsed += poll_interval

    return {"status": "timeout", "body": {"status": last_status}}


# ---------------------------------------------------------------------------
# Provider
# ---------------------------------------------------------------------------


class XAIVideoGenProvider(VideoGenProvider):
    """xAI Grok Imagine video backend."""

    @property
    def name(self) -> str:
        return "xai"

    @property
    def display_name(self) -> str:
        return "xAI"

    def is_available(self) -> bool:
        api_key, _ = _resolve_xai_credentials()
        return bool(api_key)

    def list_models(self) -> List[Dict[str, Any]]:
        return [{"id": mid, **meta} for mid, meta in _MODELS.items()]

    def default_model(self) -> Optional[str]:
        return DEFAULT_MODEL

    def get_setup_schema(self) -> Dict[str, Any]:
        # Auth resolution lives entirely in the shared ``xai_grok`` post_setup
        # hook (``hermes_cli/tools_config.py``) so the picker doesn't blindly
        # prompt for an API key when the user is already signed in via xAI
        # Grok OAuth (SuperGrok / Premium+) — TTS / image gen / video gen
        # all share the same credential resolver. The hook offers an
        # OAuth-vs-API-key choice when neither is configured.
        try:
            from tools.xai_http import xai_storage_notice_text

            storage_notice = xai_storage_notice_text("video_gen")
        except Exception:
            storage_notice = ""
        tag = (
            "grok-imagine-video for text/reference; "
            "grok-imagine-video-1.5 for image-to-video; "
            "edit/extend: pass the stored public HTTPS MP4 (`video` / "
            "`public_url` from a prior Imagine result); uses xAI Grok OAuth "
            "or XAI_API_KEY"
        )
        if storage_notice:
            tag += f". {storage_notice}"
        return {
            "name": "xAI Grok Imagine",
            "badge": "paid",
            "tag": tag,
            "env_vars": [],
            "post_setup": "xai_grok",
        }

    def capabilities(self) -> Dict[str, Any]:
        return {
            "modalities": ["text", "image"],
            "aspect_ratios": sorted(VALID_ASPECT_RATIOS),
            "resolutions": sorted(VALID_RESOLUTIONS),
            "max_duration": 15,
            "min_duration": 1,
            "supports_audio": False,
            "supports_negative_prompt": False,
            "max_reference_images": MAX_REFERENCE_IMAGES,
        }

    def generate(
        self,
        prompt: str,
        *,
        model: Optional[str] = None,
        image_url: Optional[str] = None,
        reference_image_urls: Optional[List[str]] = None,
        duration: Optional[int] = None,
        aspect_ratio: str = DEFAULT_ASPECT_RATIO,
        resolution: str = DEFAULT_RESOLUTION,
        negative_prompt: Optional[str] = None,
        audio: Optional[bool] = None,
        seed: Optional[int] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        return run_xai_video_generation(
            prompt=prompt,
            model=model,
            explicit_model=bool(kwargs.get("_model_override_explicit")),
            image_url=image_url,
            reference_image_urls=reference_image_urls,
            duration=duration,
            aspect_ratio=aspect_ratio,
            resolution=resolution,
        )


def has_xai_video_credentials() -> bool:
    api_key, _ = _resolve_xai_credentials()
    return bool(api_key)


def run_xai_video_generation(
    *,
    prompt: str,
    model: Optional[str],
    explicit_model: bool,
    image_url: Optional[str],
    reference_image_urls: Optional[List[str]],
    duration: Optional[int],
    aspect_ratio: str,
    resolution: str,
) -> Dict[str, Any]:
    return _run_xai_video_coroutine(
        _generate_xai_video_async(
            prompt=prompt,
            model=model,
            explicit_model=explicit_model,
            image_url=image_url,
            reference_image_urls=reference_image_urls,
            duration=duration,
            aspect_ratio=aspect_ratio,
            resolution=resolution,
        ),
        operation_label="generation",
        model=model,
        prompt=prompt,
        aspect_ratio=aspect_ratio,
    )


def run_xai_video_edit(
    *,
    prompt: str,
    video_url: str,
    model: Optional[str] = None,
) -> Dict[str, Any]:
    return _run_xai_video_coroutine(
        _edit_xai_video_async(prompt=prompt, video_url=video_url, model=model),
        operation_label="edit",
        model=model,
        prompt=prompt,
        aspect_ratio=DEFAULT_ASPECT_RATIO,
    )


def run_xai_video_extend(
    *,
    prompt: str,
    video_url: str,
    duration: Optional[int] = None,
    model: Optional[str] = None,
) -> Dict[str, Any]:
    return _run_xai_video_coroutine(
        _extend_xai_video_async(
            prompt=prompt,
            video_url=video_url,
            duration=duration,
            model=model,
        ),
        operation_label="extend",
        model=model,
        prompt=prompt,
        aspect_ratio=DEFAULT_ASPECT_RATIO,
    )


def _run_xai_video_coroutine(
    coro,
    *,
    operation_label: str,
    model: Optional[str],
    prompt: str,
    aspect_ratio: str,
) -> Dict[str, Any]:
    try:
        loop = asyncio.new_event_loop()
        try:
            return loop.run_until_complete(coro)
        finally:
            loop.close()
    except Exception as exc:
        logger.warning("xAI video %s unexpected failure: %s", operation_label, exc, exc_info=True)
        return error_response(
            error=f"xAI video {operation_label} failed: {exc}",
            error_type="api_error",
            provider="xai",
            model=model or DEFAULT_MODEL,
            prompt=prompt,
            aspect_ratio=aspect_ratio,
        )


async def _generate_xai_video_async(
    *,
    prompt: str,
    model: Optional[str],
    explicit_model: bool,
    image_url: Optional[str],
    reference_image_urls: Optional[List[str]],
    duration: Optional[int],
    aspect_ratio: str,
    resolution: str,
) -> Dict[str, Any]:
    api_key, base_url = _resolve_xai_credentials()
    if not api_key:
        return _auth_required_response(prompt)

    prompt = (prompt or "").strip()
    image_input = None
    if (image_url or "").strip():
        image_input = _image_ref_to_xai_input(image_url)
        if not image_input:
            return error_response(
                error=(
                    "image_url must be a public HTTPS URL or data URI "
                    "(e.g. the `image`/`public_url` from a prior Imagine result)"
                ),
                error_type="invalid_image_url",
                provider="xai",
                prompt=prompt,
            )
    normalized_aspect_ratio = (aspect_ratio or DEFAULT_ASPECT_RATIO).strip()
    normalized_resolution = (resolution or DEFAULT_RESOLUTION).strip().lower()
    refs, refs_error = _normalize_reference_images(reference_image_urls)
    if refs_error:
        return error_response(
            error=refs_error,
            error_type="invalid_reference_image_urls",
            provider="xai",
            prompt=prompt,
        )

    if not prompt:
        return error_response(
            error="prompt is required for xAI video generation",
            error_type="missing_prompt",
            provider="xai", prompt=prompt,
        )
    if refs and len(refs) > MAX_REFERENCE_IMAGES:
        return error_response(
            error=f"reference_image_urls supports at most {MAX_REFERENCE_IMAGES} images on xAI",
            error_type="too_many_references",
            provider="xai", prompt=prompt,
        )
    if image_input and refs:
        return error_response(
            error="image_url and reference_image_urls cannot be combined on xAI",
            error_type="conflicting_inputs",
            provider="xai", prompt=prompt,
        )

    if normalized_aspect_ratio not in VALID_ASPECT_RATIOS:
        normalized_aspect_ratio = DEFAULT_ASPECT_RATIO
    if normalized_resolution not in VALID_RESOLUTIONS:
        normalized_resolution = DEFAULT_RESOLUTION

    modality_used = "reference" if refs else ("image" if image_input else "text")
    resolved_model = _resolve_model_for_modality(
        model,
        modality=modality_used,
        explicit_model=explicit_model,
    )
    if refs and resolved_model != DEFAULT_TEXT_TO_VIDEO_MODEL:
        if explicit_model:
            return error_response(
                error=(
                    "xAI reference-to-video requires "
                    f"{DEFAULT_TEXT_TO_VIDEO_MODEL}; got {resolved_model}"
                ),
                error_type="unsupported_model",
                provider="xai",
                model=resolved_model,
                prompt=prompt,
            )
        resolved_model = DEFAULT_TEXT_TO_VIDEO_MODEL

    clamped_duration = _clamp_duration(duration, has_reference_images=bool(refs))
    payload = {
        "model": resolved_model,
        "prompt": prompt,
        "duration": clamped_duration,
        "aspect_ratio": normalized_aspect_ratio,
        "resolution": normalized_resolution,
    }
    if image_input:
        payload["image"] = image_input
    if refs:
        payload["reference_images"] = refs

    return await _submit_xai_video_payload(
        api_key=api_key,
        base_url=base_url,
        endpoint="generations",
        payload=payload,
        prompt=prompt,
        resolved_model=resolved_model,
        modality=modality_used,
        aspect_ratio=normalized_aspect_ratio,
        duration=clamped_duration,
        operation="generate",
        resolution=normalized_resolution,
    )


async def _run_xai_video_mutation(
    *,
    prompt: str,
    video_url: str,
    model: Optional[str],
    endpoint: str,
    operation: str,
    duration: int,
) -> Dict[str, Any]:
    """Edit or extend using a public HTTPS ``video_url`` input (``url`` on the wire)."""
    api_key, base_url = _resolve_xai_credentials()
    if not api_key:
        return _auth_required_response(prompt)

    prompt = (prompt or "").strip()
    video_input = await _video_input_from_public_url(
        video_url or "",
        api_key=api_key,
        base_url=base_url,
    )
    if not prompt:
        return error_response(
            error="prompt is required for xAI video edit/extend",
            error_type="missing_prompt",
            provider="xai",
            prompt=prompt,
        )
    if not video_input:
        return error_response(
            error=(
                "video_url must be a public HTTPS MP4 URL "
                "(the `video`/`public_url` from a prior Imagine result)"
            ),
            error_type="missing_video",
            provider="xai",
            prompt=prompt,
        )

    resolved_model = _resolve_model_for_modality(
        model,
        modality="text",
        explicit_model=bool(model),
    )
    payload: Dict[str, Any] = {
        "model": resolved_model,
        "prompt": prompt,
        "video": video_input,
    }
    if endpoint == "extensions":
        payload["duration"] = duration

    return await _submit_xai_video_payload(
        api_key=api_key,
        base_url=base_url,
        endpoint=endpoint,
        payload=payload,
        prompt=prompt,
        resolved_model=resolved_model,
        modality=operation,
        aspect_ratio=DEFAULT_ASPECT_RATIO,
        duration=duration,
        operation=operation,
    )


async def _edit_xai_video_async(
    *,
    prompt: str,
    video_url: str,
    model: Optional[str],
) -> Dict[str, Any]:
    return await _run_xai_video_mutation(
        prompt=prompt,
        video_url=video_url,
        model=model,
        endpoint="edits",
        operation="edit",
        duration=DEFAULT_DURATION,
    )


async def _extend_xai_video_async(
    *,
    prompt: str,
    video_url: str,
    duration: Optional[int],
    model: Optional[str],
) -> Dict[str, Any]:
    clamped_duration = _clamp_duration(
        duration,
        max_seconds=10,
        default=DEFAULT_EXTEND_DURATION,
    )
    return await _run_xai_video_mutation(
        prompt=prompt,
        video_url=video_url,
        model=model,
        endpoint="extensions",
        operation="extend",
        duration=clamped_duration,
    )


def _auth_required_response(prompt: str) -> Dict[str, Any]:
    return error_response(
        error=(
            "No xAI credentials found. Sign in via `hermes auth add xai-oauth` "
            "(SuperGrok / Premium+) or set XAI_API_KEY from "
            "https://console.x.ai/."
        ),
        error_type="auth_required",
        provider="xai", prompt=prompt,
    )


async def _submit_xai_video_payload(
    *,
    api_key: str,
    base_url: str,
    endpoint: str,
    payload: Dict[str, Any],
    prompt: str,
    resolved_model: str,
    modality: str,
    aspect_ratio: str,
    duration: int,
    operation: str,
    resolution: Optional[str] = None,
) -> Dict[str, Any]:
    try:
        from tools.xai_http import (
            build_xai_storage_options,
            maybe_mark_xai_storage_notice_seen,
            read_xai_imagine_storage_config,
        )

        storage_options = build_xai_storage_options(
            "video_gen",
            filename_prefix="hermes-xai-video",
            extension="mp4",
        )
        storage_notice = maybe_mark_xai_storage_notice_seen("video_gen")
        storage_cfg = read_xai_imagine_storage_config("video_gen")
    except Exception:
        storage_options = None
        storage_notice = None
        storage_cfg = {"enabled": False}
    if storage_options is not None:
        payload["storage_options"] = storage_options

    async with httpx.AsyncClient() as client:
        try:
            request_id = await _submit(
                client, payload, api_key=api_key, base_url=base_url,
                endpoint=endpoint,
            )
        except httpx.HTTPStatusError as exc:
            detail = ""
            try:
                detail = exc.response.text[:500]
            except Exception:
                pass
            return error_response(
                error=f"xAI submit failed ({exc.response.status_code}): {detail or exc}",
                error_type="api_error",
                provider="xai",
                model=resolved_model,
                prompt=prompt,
            )

        poll_result = await _poll(
            client, request_id,
            api_key=api_key, base_url=base_url,
            timeout_seconds=DEFAULT_TIMEOUT_SECONDS,
            poll_interval=DEFAULT_POLL_INTERVAL_SECONDS,
        )

    status = poll_result["status"]
    body = poll_result["body"]

    if status == "done":
        video = body.get("video") or {}
        if not isinstance(video, dict):
            video = {}
        file_output = video.get("file_output") if isinstance(video.get("file_output"), dict) else {}
        file_output = file_output or {}
        public_video_url, temporary_url, stored_public_url = _xai_video_output_urls(video)
        if not public_video_url:
            return error_response(
                error="xAI video request completed without a video URL",
                error_type="empty_response",
                provider="xai",
                model=body.get("model") or resolved_model,
                prompt=prompt,
            )
        extra: Dict[str, Any] = {
            "request_id": request_id,
            "operation": operation,
            "storage_enabled": bool(storage_cfg.get("enabled")),
        }
        if resolution:
            extra["resolution"] = resolution
        if storage_notice:
            extra["storage_notice"] = storage_notice
        if stored_public_url:
            extra["public_url"] = stored_public_url
        if temporary_url:
            extra["temporary_url"] = temporary_url
        if file_output:
            for key in (
                "filename",
                "expires_at",
                "public_url_expires_at",
                "public_url_error",
                "storage_error",
            ):
                if key in file_output:
                    extra[key] = file_output[key]
        if body.get("usage"):
            extra["usage"] = body["usage"]
        return success_response(
            video=public_video_url,
            model=body.get("model") or resolved_model,
            prompt=prompt,
            modality=modality,
            aspect_ratio=aspect_ratio,
            duration=video.get("duration") or duration,
            provider="xai",
            extra=extra,
        )

    if status == "timeout":
        return error_response(
            error=f"Timed out waiting for xAI video request after {DEFAULT_TIMEOUT_SECONDS}s",
            error_type="timeout",
            provider="xai",
            model=resolved_model,
            prompt=prompt,
        )

    message = (
        (body.get("error", {}) or {}).get("message")
        or body.get("message")
        or f"xAI video request ended with status '{status}'"
    )
    return error_response(
        error=message,
        error_type=f"xai_{status}",
        provider="xai",
        model=resolved_model,
        prompt=prompt,
    )


# ---------------------------------------------------------------------------
# Plugin entry point
# ---------------------------------------------------------------------------


def register(ctx) -> None:
    """Plugin entry point — wire ``XAIVideoGenProvider`` into the registry."""
    ctx.register_video_gen_provider(XAIVideoGenProvider())
