app.py

"""LTX-2.5 Space entrypoint.

Own only startup-sensitive runtime assignment, thin callback bridges, top-level
Gradio composition, and event wiring here. Reusable feature logic and concrete
UI sub-surfaces belong under ``ltx/``.
"""
import os

# Must precede torch import. Keep the legacy alias as well as the current name.
os.environ.setdefault("PYTORCH_ALLOC_CONF", "backend:cudaMallocAsync")
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "backend:cudaMallocAsync")

import json
import uuid

import gradio as gr
import spaces


# ZeroGPU startup/build detector anchor. Keep one unconditional module-global
# @spaces.GPU function directly visible in the main app module; production GPU
# callbacks may be owned elsewhere. The detector anchor itself does no work.
@spaces.GPU
def _zerogpu_startup_anchor():
    pass


import torch

from diffusers.pipelines.ltx2.utils import (
    DISTILLED_SIGMA_VALUES,
    LTX2_5_I2V_DEFAULT_SYSTEM_PROMPT,
    LTX2_5_T2V_DEFAULT_SYSTEM_PROMPT,
)

from space_config import (
    A2V_EXAMPLE_SPECS,
    A2V_WIDTH,
    A2V_HEIGHT,
    A2V_FRAMES,
    A2V_ZEROGPU_DURATION_SECONDS,
    BUILTIN_LORAS,
    HISTORY_LIMIT,
    FRAME_RATE,
    STANDARD_MAX_SECONDS,
    EXPERIMENTAL_MAX_SECONDS,
    RESOLUTIONS,
    IC_COLORIZER_LORA_REPO_ID,
    IC_COLORIZER_LORA_REVISION,
    IC_COLORIZER_LORA_WEIGHT_NAME,
    IC_COLORIZER_LORA_STRENGTH,
    IC_COLORIZER_REFERENCE_DOWNSCALE_FACTOR,
    IC_COLORIZER_REFERENCE_STRENGTH,
    IC_COLORIZER_CONDITIONING_ATTENTION_STRENGTH,
    IC_COLORIZER_EXAMPLE_SPECS,
    IC_COLORIZER_DEFAULT_PROFILE,
    IC_COLORIZER_PROFILES,
    IC_PIXEL_UPSCALER_LORA_REPO_ID,
    IC_PIXEL_UPSCALER_LORA_REVISION,
    IC_PIXEL_UPSCALER_LORA_WEIGHT_NAME,
    IC_PIXEL_UPSCALER_LORA_STRENGTH,
    IC_PIXEL_UPSCALER_REFERENCE_DOWNSCALE_FACTOR,
    IC_PIXEL_UPSCALER_REFERENCE_STRENGTH,
    IC_PIXEL_UPSCALER_CONDITIONING_ATTENTION_STRENGTH,
    IC_PIXEL_UPSCALER_MAX_DURATION_SECONDS,
    IC_PIXEL_UPSCALER_MAX_OUTPUT_SIDE,
    IC_PIXEL_UPSCALER_ZEROGPU_DURATION_SECONDS,
    IC_PIXEL_UPSCALER_EXAMPLE_SPECS,
    IC_INOUTPAINT_LORA_REPO_ID,
    IC_INOUTPAINT_LORA_REVISION,
    IC_INOUTPAINT_LORA_WEIGHT_NAME,
    IC_INOUTPAINT_LORA_STRENGTH,
    IC_INOUTPAINT_REFERENCE_DOWNSCALE_FACTOR,
    IC_INOUTPAINT_REFERENCE_STRENGTH,
    IC_INOUTPAINT_CONDITIONING_ATTENTION_STRENGTH,
    IC_INOUTPAINT_MAX_DURATION_SECONDS,
    IC_INOUTPAINT_MAX_OUTPUT_SIDE,
    IC_INOUTPAINT_ZEROGPU_DURATION_SECONDS,
    IC_INOUTPAINT_EXAMPLE_SPECS,
    DEFAULT_PROMPT,
    MODEL_QUANTIZATION_POLICY,
    MODEL_REPO_ID,
    MODEL_RUNTIME_PROFILE,
    FULL_SFT_TRANSFORMER_PATH,
    FULL_SFT_STAGE2_LORA_WEIGHT_NAME,
    MODEL_REVISION,
    PROMPT_ENHANCER_REPO_ID,
    PROMPT_ENHANCER_REVISION,
    PROMPT_TOOL_ZEROGPU_DURATION_SECONDS,
    PROMPT_TRANSLATE_LTX_ENGLISH_SYSTEM_PROMPT,
    PROMPT_WRITE_FROM_SEED_SYSTEM_PROMPT,
    TEXT_ENCODER_OVERRIDE_PATH,
    TEXT_ENCODER_OVERRIDE_REPO_ID,
    TEXT_ENCODER_OVERRIDE_REVISION,
    TRANSFORMER_OVERRIDE_PATH,
    TRANSFORMER_OVERRIDE_REPO_ID,
    TRANSFORMER_OVERRIDE_REVISION,
)


from ltx import audio_to_video as a2v_backend
from ltx import civitai as civitai_backend
from ltx import generation as generation_backend
from ltx import ic_colorizer as ic_backend
from ltx import ic_pixel_upscaler as pixel_upscaler_backend
from ltx import ic_inoutpaint as inoutpaint_backend
from ltx import hub_search as hub_search_backend
from ltx import lora as lora_backend
from ltx import maintainer_probe as maintainer_probe_backend
from ltx import probe_catalog as maintainer_probe_catalog
from ltx import prompt_tools as prompt_backend
from ltx import research_probe as research_probe_backend
from ltx import probe_artifacts as probe_artifacts
from ltx import runtime as runtime_backend
from ltx import runtime_utils as runtime_utils
from ltx import settings_history as settings_history
from ltx import ui_lora as ui_lora
from ltx import ui_ic_tabs as ui_ic_tabs
from ltx.gradio_policy import INTERNAL_API_VISIBILITY
from ltx.lora_ui_handlers import (
    _all_lora_defs,
    _invalidate_prepared_loras,
    _lora_dropdown,
    _sanitize_lora_defs,
    add_civitai_session_lora,
    add_session_lora,
    apply_civitai_example_prompt,
    apply_civitai_trigger_words,
    apply_hf_full_search_selection,
    apply_hub_lora_search_selection,
    inspect_civitai_gallery_selection,
    inspect_civitai_version_selection,
    inspect_hub_lora_repo,
    load_civitai_example_prompts,
    load_more_civitai_models,
    prepare_selected_loras,
    remove_selected_session_loras,
    resolve_civitai_url,
    search_civitai_models,
    search_hf_lora_models,
)
from ltx.build_info import CANDIDATE_ID
from ltx.app_config import (
    ALLOW_PATTERNS,
    ATTENTION_BACKEND,
    CANONICAL_MODEL_REPO_ID,
    CANONICAL_MODEL_REVISION,
    CONFIG_WARNINGS,
    EFFECTIVE_A2V_ENABLED,
    EFFECTIVE_AUTO_DURATION_ENABLED,
    EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
    EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
    EFFECTIVE_DEFAULT_DURATION_SECONDS,
    EFFECTIVE_DEFAULT_LORA_STRENGTH,
    EFFECTIVE_DEFAULT_RANDOMIZE_SEED,
    EFFECTIVE_DEFAULT_RESOLUTION,
    EFFECTIVE_DEFAULT_SEED,
    EFFECTIVE_DEFAULT_SELECTED_LORAS,
    EFFECTIVE_DEFAULT_USE_AUTO_DURATION,
    EFFECTIVE_DEFAULT_USE_DIFFUSION_DECODER,
    EFFECTIVE_DIFFUSION_DECODER_ENABLED,
    EFFECTIVE_FULL_SFT_STAGE2_LORA_STRENGTH,
    EFFECTIVE_IC_COLORIZER_ENABLED,
    EFFECTIVE_IC_PIXEL_UPSCALER_ENABLED,
    EFFECTIVE_IC_INOUTPAINT_ENABLED,
    EFFECTIVE_PROMPT_ENHANCER_ENABLED,
    EFFECTIVE_ZEROGPU_GPU_SIZE,
    FULL_SFT_STAGE2_ADAPTER_NAME,
    FULL_SFT_STAGE2_LORA_REPO_EFFECTIVE,
    FULL_SFT_STAGE2_LORA_REVISION_EFFECTIVE,
    FULL_SFT_TRANSFORMER_REPO_EFFECTIVE,
    FULL_SFT_TRANSFORMER_REVISION_EFFECTIVE,
    GPU_CONCURRENCY_ID,
    IS_FULL_SFT_PROFILE,
    IS_ZEROGPU,
    PROMPT_ENHANCER_POLICY,
    QUANTIZATION_POLICY,
    RUNTIME_PROFILE,
)
from ltx.app_runtime_support import (
    APP_LOGGER,
    APP_ROOT,
    HF_TOKEN,
    RUNTIME_A2V_EXAMPLE_DIR,
    RUNTIME_IC_EXAMPLE_DIR,
    RUNTIME_IC_PIXEL_UPSCALER_EXAMPLE_DIR,
    RUNTIME_IC_INOUTPAINT_EXAMPLE_DIR,
    SPACE_DEBUG,
    WORKER_ROOT,
    WORKER_UUID,
    cleanup_session_results,
    close_request_logger as _close_request_logger,
    create_request_paths,
    create_session_id,
    disk_state as _disk_state,
    gpu_state as _gpu_state,
    is_uuid_hex as _is_uuid_hex,
    normalize_session_id,
    open_request_logger as _open_request_logger,
    package_identity as _package_identity,
    runtime_environment_identity as _runtime_environment_identity,
    sha256_file as _sha256_file,
)
from ltx.probe_catalog import (
    MAINTAINER_PROBE_CHOICES,
    MAINTAINER_PROBE_CURRENT_GATES,
    MAINTAINER_PROBE_FULL_SFT_30S_STAGE1,
    MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV,
    MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION,
    MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV_FRAMEWISE,
    MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_LATENT,
    MAINTAINER_PROBE_FULL_SFT_30S_UPSAMPLE,
    MAINTAINER_PROBE_FULL_SFT_30S_FLF2V_STAGE1,
    RESEARCH_PROBE_CHOICES,
)
from ltx.ui_controls import (
    bind_custom_resolution_editor as _bind_custom_resolution_editor,
    build_custom_resolution_editor as _build_custom_resolution_editor,
    build_result_panel as _build_result_panel,
    build_seed_controls as _build_seed_controls,
    duration_slider_update as _duration_slider_update,
    frames_from_seconds as _frames_from_seconds,
    load_conditioning_image as _load_conditioning_image,
    parse_resolution_value as _parse_resolution_value,
    resolution as _resolution,
    resolution_duration_controls as _resolution_duration_controls,
    resolution_input_preflight as _resolution_input_preflight,
    sync_resolution_editor_after as _sync_resolution_editor_after,
    supports_experimental_long as _supports_experimental_long,
)
from ltx.conditioning import (
    conditioning_mode_status_markdown as _conditioning_mode_status_markdown,
    mode_from_condition_images as _mode_from_condition_images,
)
from ltx.zerogpu_duration import (
    generation_duration as _duration,
    generation_duration_preflight as _generation_duration_preflight,
    zerogpu_duration_panel as _zerogpu_duration_panel,
)
from ltx import example_runtime
from ltx.ui_theme import CSS
PIPE = None
PIPE_I2V = None
PIPE_CONDITION = None
PIPE_IC = None
UPSAMPLE_PIPE = None
DIFFUSION_DECODE_PIPE = None
PROMPT_ENHANCER_MODEL = None
PROMPT_ENHANCER_PROCESSOR = None
EXAMPLE_ROWS = []
EXAMPLE_STATE = {"status": "not_prepared", "items": []}
IC_EXAMPLE_ROWS = []
IC_EXAMPLE_STATE = {"status": "not_prepared", "items": []}
PIXEL_UPSCALER_EXAMPLE_ROWS = []
PIXEL_UPSCALER_EXAMPLE_STATE = {"status": "not_prepared", "items": []}
INOUTPAINT_EXAMPLE_ROWS = []
INOUTPAINT_EXAMPLE_STATE = {"status": "not_prepared", "items": []}
A2V_EXAMPLE_ROWS = []
A2V_EXAMPLE_STATE = {"status": "not_prepared", "items": []}
PRELOAD_STATE = {"status": "starting", "candidate_id": CANDIDATE_ID}


RUNTIME_BUILD_POLICY = runtime_backend.RuntimeBuildPolicy(
    allow_patterns=tuple(ALLOW_PATTERNS),
    runtime_profile=RUNTIME_PROFILE,
    is_full_sft_profile=IS_FULL_SFT_PROFILE,
    auto_duration_enabled=EFFECTIVE_AUTO_DURATION_ENABLED,
    auto_duration_min_seconds=EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
    auto_duration_max_seconds=EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
    diffusion_decoder_enabled=EFFECTIVE_DIFFUSION_DECODER_ENABLED,
    prompt_enhancer_enabled=EFFECTIVE_PROMPT_ENHANCER_ENABLED,
    ic_colorizer_enabled=EFFECTIVE_IC_COLORIZER_ENABLED,
    ic_pixel_upscaler_enabled=EFFECTIVE_IC_PIXEL_UPSCALER_ENABLED,
    ic_inoutpaint_enabled=EFFECTIVE_IC_INOUTPAINT_ENABLED,
    transformer_override_repo_id=TRANSFORMER_OVERRIDE_REPO_ID,
    transformer_override_path=TRANSFORMER_OVERRIDE_PATH,
    transformer_override_revision=TRANSFORMER_OVERRIDE_REVISION,
    text_encoder_override_repo_id=TEXT_ENCODER_OVERRIDE_REPO_ID,
    text_encoder_override_path=TEXT_ENCODER_OVERRIDE_PATH,
    text_encoder_override_revision=TEXT_ENCODER_OVERRIDE_REVISION,
    prompt_enhancer_repo_id=PROMPT_ENHANCER_REPO_ID,
    prompt_enhancer_revision=PROMPT_ENHANCER_REVISION,
    prompt_enhancer_policy=PROMPT_ENHANCER_POLICY,
    full_sft_transformer_path=FULL_SFT_TRANSFORMER_PATH,
    full_sft_transformer_repo=FULL_SFT_TRANSFORMER_REPO_EFFECTIVE,
    full_sft_transformer_revision=FULL_SFT_TRANSFORMER_REVISION_EFFECTIVE,
    full_sft_stage2_lora_repo=FULL_SFT_STAGE2_LORA_REPO_EFFECTIVE,
    full_sft_stage2_lora_revision=FULL_SFT_STAGE2_LORA_REVISION_EFFECTIVE,
    full_sft_stage2_lora_weight_name=FULL_SFT_STAGE2_LORA_WEIGHT_NAME,
    full_sft_stage2_lora_strength=EFFECTIVE_FULL_SFT_STAGE2_LORA_STRENGTH,
    full_sft_stage2_adapter_name=FULL_SFT_STAGE2_ADAPTER_NAME,
    attention_backend=ATTENTION_BACKEND,
)

RUNTIME_PRELOAD_POLICY = runtime_backend.RuntimePreloadPolicy(
    candidate_id=CANDIDATE_ID,
    model_repo_id=MODEL_REPO_ID,
    model_revision=MODEL_REVISION,
    model_quantization_policy_requested=MODEL_QUANTIZATION_POLICY,
    quantization_policy=QUANTIZATION_POLICY,
    model_runtime_profile_requested=MODEL_RUNTIME_PROFILE,
    runtime_profile=RUNTIME_PROFILE,
    config_warnings=tuple(CONFIG_WARNINGS),
    worker_uuid=WORKER_UUID,
    canonical_model_repo_id=CANONICAL_MODEL_REPO_ID,
    canonical_model_revision=CANONICAL_MODEL_REVISION,
    is_full_sft_profile=IS_FULL_SFT_PROFILE,
    is_zerogpu=IS_ZEROGPU,
    auto_duration_min_seconds=EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
    auto_duration_max_seconds=EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
)


def _preload():
    global PIPE, PIPE_I2V, PIPE_CONDITION, PIPE_IC, UPSAMPLE_PIPE, DIFFUSION_DECODE_PIPE
    global PROMPT_ENHANCER_MODEL, PROMPT_ENHANCER_PROCESSOR, PRELOAD_STATE
    outcome = runtime_backend.preload_runtime(
        token=HF_TOKEN,
        preload_policy=RUNTIME_PRELOAD_POLICY,
        build_policy=RUNTIME_BUILD_POLICY,
        log_fn=APP_LOGGER.info,
        disk_state_fn=_disk_state,
        package_identity_fn=_package_identity,
        environment_identity_fn=_runtime_environment_identity,
    )
    if outcome.runtime is not None:
        artifacts = outcome.runtime
        PIPE = artifacts.pipe
        PIPE_I2V = artifacts.pipe_i2v
        PIPE_CONDITION = artifacts.pipe_condition
        PIPE_IC = artifacts.pipe_ic
        UPSAMPLE_PIPE = artifacts.upsample_pipe
        DIFFUSION_DECODE_PIPE = artifacts.diffusion_decode_pipe
        PROMPT_ENHANCER_MODEL = artifacts.prompt_enhancer_model
        PROMPT_ENHANCER_PROCESSOR = artifacts.prompt_enhancer_processor
    PRELOAD_STATE = outcome.state


def _runtime_model_status_markdown() -> str:
    if PRELOAD_STATE.get("status") != "ready":
        error = PRELOAD_STATE.get("error") or "preload is not ready"
        base_link = runtime_utils.hf_repo_markdown_link("Lightricks/LTX-2.5-Diffusers")
        return (
            "**Runtime model: startup failed.**  \n"
            f"First check access to {base_link}, then confirm the Space Secret `HF_TOKEN` is a read token owned by the **same Hugging Face account** that approved the gated model. Restart the Space after changing access/token.  \n"
            f"Diagnostic: `{error}`"
        )

    sources = PRELOAD_STATE.get("model_sources") or {}
    base = (sources.get("base") or {}).get("effective") or {}
    base_text = runtime_utils.hf_repo_markdown_link(str(base.get('repo_id') or '?')) if base.get('repo_id') else '`?`' 
    if base.get("revision"):
        base_text += f" @ {str(base['revision'])[:12]}…"
    base_suffix = " · **fallback active**" if (sources.get("base") or {}).get("fallback") else ""

    def component_text(name: str) -> str:
        record = sources.get(name) or {}
        effective = record.get("effective") or {}
        kind = effective.get("kind")
        if kind == "disabled":
            value = f"Unavailable · {effective.get('reason') or 'disabled'}"
        elif kind and kind.startswith("override"):
            repo_id = str(effective.get("repo_id") or "?")
            value = runtime_utils.hf_repo_markdown_link(repo_id) if repo_id != "?" else "`?`"
            if effective.get("path"):
                value += f" / {effective['path']}"
            quant = effective.get("quantization")
            if quant:
                value += f" · {quant}"
        elif name == "transformer" and kind == "full_sft_base_component":
            value = f"Full/SFT · `{effective.get('path', 'transformer_full')}` · {effective.get('quantization', 'NF4')}"
        elif name == "diffusion_decoder" and kind == "base_component":
            value = "Live PASS · CPU RAM-ready · GPU-lazy · NATTEN · tiled"
        elif name == "prompt_enhancer" and kind == "dedicated_gemma4":
            value = (
                f"Ready in CPU RAM · GPU-lazy · {effective.get('quantization', 'NF4')} · "
                + runtime_utils.hf_repo_markdown_link(effective.get("repo_id"))
            )
        elif name == "duration_head" and kind == "base_component":
            bounds = effective.get("bounds_seconds") or [EFFECTIVE_AUTO_DURATION_MIN_SECONDS, EFFECTIVE_AUTO_DURATION_MAX_SECONDS]
            value = f"Ready · model-predicted · {float(bounds[0]):.1f}–{float(bounds[1]):.1f}s clamp · 8k+1 grid"
        elif name == "stage2_distilled_lora" and kind == "required_stage2_distilled_lora":
            value = "Required · CPU RAM-ready · Stage-2 GPU-lazy · returned to CPU after refinement"
        else:
            quant = effective.get("quantization") or "base component"
            value = f"Default · {quant}"
        if record.get("fallback"):
            value += " · **requested path failed → fallback/disable**"
        return value

    defaults = [str(x) for x in EFFECTIVE_DEFAULT_SELECTED_LORAS if str(x).strip()]
    lora_text = ", ".join(defaults) if defaults else "None"
    return (
        f"**Runtime model** · Profile: `{RUNTIME_PROFILE}` · restart-required exclusive · Base: {base_text}{base_suffix}  \n"
        f"Transformer: {component_text('transformer')}  \n"
        f"Stage-2 distilled LoRA: {component_text('stage2_distilled_lora')}  \n"
        f"Text encoder: {component_text('text_encoder')}  \n"
        f"Diffusion decoder: {component_text('diffusion_decoder')}  \n"
        f"Prompt enhancer: {component_text('prompt_enhancer')}  \n"
        f"Auto duration: {component_text('duration_head')}  \n"
        f"Default LoRA: {lora_text}"
    )


def _ic_colorizer_config() -> ic_backend.ColorizerConfig:
    return ic_backend.ColorizerConfig(
        enabled=EFFECTIVE_IC_COLORIZER_ENABLED,
        is_full_sft_profile=IS_FULL_SFT_PROFILE,
        runtime_profile=RUNTIME_PROFILE,
        frame_rate=FRAME_RATE,
        hf_token=HF_TOKEN,
        profiles=IC_COLORIZER_PROFILES,
        default_profile=IC_COLORIZER_DEFAULT_PROFILE,
        example_specs=tuple(IC_COLORIZER_EXAMPLE_SPECS or []),
        lora_repo_id=str(IC_COLORIZER_LORA_REPO_ID),
        lora_revision=str(IC_COLORIZER_LORA_REVISION or "").strip() or None,
        lora_weight_name=str(IC_COLORIZER_LORA_WEIGHT_NAME),
        lora_strength=float(IC_COLORIZER_LORA_STRENGTH),
        reference_downscale_factor=int(IC_COLORIZER_REFERENCE_DOWNSCALE_FACTOR),
        reference_strength=float(IC_COLORIZER_REFERENCE_STRENGTH),
        conditioning_attention_strength=float(IC_COLORIZER_CONDITIONING_ATTENTION_STRENGTH),
        distilled_sigmas=tuple(float(x) for x in DISTILLED_SIGMA_VALUES),
        protected_adapter_name=(FULL_SFT_STAGE2_ADAPTER_NAME if IS_FULL_SFT_PROFILE else None),
    )


def _ic_colorizer_runtime() -> ic_backend.ColorizerRuntime:
    return ic_backend.ColorizerRuntime(
        candidate_id=CANDIDATE_ID,
        worker_uuid=WORKER_UUID,
        worker_root=WORKER_ROOT,
        app_root=APP_ROOT,
        example_dir=RUNTIME_IC_EXAMPLE_DIR,
        pipe=PIPE,
        pipe_ic=PIPE_IC,
        preload_state=PRELOAD_STATE,
        example_state=IC_EXAMPLE_STATE,
        app_logger=APP_LOGGER,
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
        gpu_state=_gpu_state,
        sha256_file=_sha256_file,
        is_uuid_hex=_is_uuid_hex,
    )


def _prepare_ic_examples() -> None:
    global IC_EXAMPLE_ROWS, IC_EXAMPLE_STATE
    IC_EXAMPLE_ROWS, IC_EXAMPLE_STATE = ic_backend.prepare_examples(
        _ic_colorizer_config(),
        _ic_colorizer_runtime(),
    )
    PRELOAD_STATE["ic_examples"] = IC_EXAMPLE_STATE


def _ic_example_attribution_markdown() -> str:
    return ic_backend.example_attribution_markdown(IC_EXAMPLE_STATE)


def _pixel_upscaler_config() -> pixel_upscaler_backend.PixelUpscalerConfig:
    return pixel_upscaler_backend.PixelUpscalerConfig(
        enabled=EFFECTIVE_IC_PIXEL_UPSCALER_ENABLED,
        runtime_profile=RUNTIME_PROFILE,
        frame_rate=FRAME_RATE,
        hf_token=HF_TOKEN,
        example_specs=tuple(IC_PIXEL_UPSCALER_EXAMPLE_SPECS or []),
        lora_repo_id=str(IC_PIXEL_UPSCALER_LORA_REPO_ID),
        lora_revision=str(IC_PIXEL_UPSCALER_LORA_REVISION or "").strip() or None,
        lora_weight_name=str(IC_PIXEL_UPSCALER_LORA_WEIGHT_NAME),
        lora_strength=float(IC_PIXEL_UPSCALER_LORA_STRENGTH),
        reference_downscale_factor=int(IC_PIXEL_UPSCALER_REFERENCE_DOWNSCALE_FACTOR),
        reference_strength=float(IC_PIXEL_UPSCALER_REFERENCE_STRENGTH),
        conditioning_attention_strength=float(IC_PIXEL_UPSCALER_CONDITIONING_ATTENTION_STRENGTH),
        max_duration_seconds=float(IC_PIXEL_UPSCALER_MAX_DURATION_SECONDS),
        max_output_side=int(IC_PIXEL_UPSCALER_MAX_OUTPUT_SIDE),
        zerogpu_duration_seconds=int(IC_PIXEL_UPSCALER_ZEROGPU_DURATION_SECONDS),
    )


def _pixel_upscaler_runtime() -> pixel_upscaler_backend.PixelUpscalerRuntime:
    return pixel_upscaler_backend.PixelUpscalerRuntime(
        candidate_id=CANDIDATE_ID,
        worker_uuid=WORKER_UUID,
        worker_root=WORKER_ROOT,
        app_root=APP_ROOT,
        example_dir=RUNTIME_IC_PIXEL_UPSCALER_EXAMPLE_DIR,
        pipe_ic=PIPE_IC,
        preload_state=PRELOAD_STATE,
        example_state=PIXEL_UPSCALER_EXAMPLE_STATE,
        app_logger=APP_LOGGER,
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
        gpu_state=_gpu_state,
        sha256_file=_sha256_file,
        is_uuid_hex=_is_uuid_hex,
    )


def _prepare_pixel_upscaler_examples() -> None:
    global PIXEL_UPSCALER_EXAMPLE_ROWS, PIXEL_UPSCALER_EXAMPLE_STATE
    PIXEL_UPSCALER_EXAMPLE_ROWS, PIXEL_UPSCALER_EXAMPLE_STATE = pixel_upscaler_backend.prepare_examples(
        _pixel_upscaler_config(),
        _pixel_upscaler_runtime(),
    )
    PRELOAD_STATE["ic_pixel_upscaler_examples"] = PIXEL_UPSCALER_EXAMPLE_STATE


def _pixel_upscaler_example_attribution_markdown() -> str:
    return pixel_upscaler_backend.example_attribution_markdown(PIXEL_UPSCALER_EXAMPLE_STATE)


def _inoutpaint_config() -> inoutpaint_backend.InOutpaintConfig:
    return inoutpaint_backend.InOutpaintConfig(
        enabled=EFFECTIVE_IC_INOUTPAINT_ENABLED,
        runtime_profile=RUNTIME_PROFILE,
        frame_rate=FRAME_RATE,
        hf_token=HF_TOKEN,
        example_specs=tuple(IC_INOUTPAINT_EXAMPLE_SPECS or []),
        lora_repo_id=str(IC_INOUTPAINT_LORA_REPO_ID),
        lora_revision=str(IC_INOUTPAINT_LORA_REVISION or "").strip() or None,
        lora_weight_name=str(IC_INOUTPAINT_LORA_WEIGHT_NAME),
        lora_strength=float(IC_INOUTPAINT_LORA_STRENGTH),
        reference_downscale_factor=int(IC_INOUTPAINT_REFERENCE_DOWNSCALE_FACTOR),
        reference_strength=float(IC_INOUTPAINT_REFERENCE_STRENGTH),
        conditioning_attention_strength=float(IC_INOUTPAINT_CONDITIONING_ATTENTION_STRENGTH),
        max_duration_seconds=float(IC_INOUTPAINT_MAX_DURATION_SECONDS),
        max_output_side=int(IC_INOUTPAINT_MAX_OUTPUT_SIDE),
        zerogpu_duration_seconds=int(IC_INOUTPAINT_ZEROGPU_DURATION_SECONDS),
    )


def _inoutpaint_runtime() -> inoutpaint_backend.InOutpaintRuntime:
    return inoutpaint_backend.InOutpaintRuntime(
        candidate_id=CANDIDATE_ID,
        worker_uuid=WORKER_UUID,
        worker_root=WORKER_ROOT,
        app_root=APP_ROOT,
        example_dir=RUNTIME_IC_INOUTPAINT_EXAMPLE_DIR,
        pipe_ic=PIPE_IC,
        preload_state=PRELOAD_STATE,
        example_state=INOUTPAINT_EXAMPLE_STATE,
        app_logger=APP_LOGGER,
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
        gpu_state=_gpu_state,
        sha256_file=_sha256_file,
        is_uuid_hex=_is_uuid_hex,
    )


def _prepare_inoutpaint_examples() -> None:
    global INOUTPAINT_EXAMPLE_ROWS, INOUTPAINT_EXAMPLE_STATE
    INOUTPAINT_EXAMPLE_ROWS, INOUTPAINT_EXAMPLE_STATE = inoutpaint_backend.prepare_examples(
        _inoutpaint_config(), _inoutpaint_runtime()
    )
    PRELOAD_STATE["ic_inoutpaint_examples"] = INOUTPAINT_EXAMPLE_STATE


def _inoutpaint_example_attribution_markdown() -> str:
    return inoutpaint_backend.example_attribution_markdown(INOUTPAINT_EXAMPLE_STATE)


_preload()
EXAMPLE_ROWS, EXAMPLE_STATE = example_runtime.prepare_examples()
PRELOAD_STATE["examples"] = EXAMPLE_STATE
_prepare_ic_examples()
_prepare_pixel_upscaler_examples()
_prepare_inoutpaint_examples()


def _lora_adapter_state() -> dict:
    return lora_backend.adapter_state(PIPE)



def _load_request_loras(selected, custom_loras, prepared_loras, strength: float, request_id: str) -> tuple[list[dict], dict]:
    try:
        return lora_backend.load_request_loras(
            pipe=PIPE,
            pipe_i2v=PIPE_I2V,
            pipe_condition=PIPE_CONDITION,
            pipe_ic=PIPE_IC,
            selected=selected,
            builtin_loras=BUILTIN_LORAS,
            custom_loras=custom_loras,
            prepared_loras=prepared_loras,
            strength=strength,
            request_id=request_id,
            gpu_state=_gpu_state,
        )
    except lora_backend.LoraSelectionError as exc:
        raise gr.Error(str(exc)) from exc



def _cleanup_request_loras(loaded, metrics=None) -> dict:
    return lora_backend.cleanup_request_loras(
        pipe=PIPE,
        loaded=loaded,
        metrics=metrics,
        full_sft_profile=IS_FULL_SFT_PROFILE,
        gpu_state=_gpu_state,
    )


def _settings_snapshot(
    prompt, duration_seconds, experimental_long, resolution_key, seed, randomize_seed,
    selected_loras, lora_strength, custom_loras, use_diffusion_decoder=False, use_auto_duration=False,
    start_image_path=None, middle_image_path=None, end_image_path=None,
) -> dict:
    return settings_history.settings_snapshot(
        prompt, duration_seconds, experimental_long, resolution_key, seed, randomize_seed,
        selected_loras, lora_strength, custom_loras, use_diffusion_decoder, use_auto_duration,
        start_image_path, middle_image_path, end_image_path,
        mode=_mode_from_condition_images(start_image_path, middle_image_path, end_image_path),
        candidate_id=CANDIDATE_ID,
        runtime_profile=RUNTIME_PROFILE,
        model_repo_id=MODEL_REPO_ID,
        model_revision=MODEL_REVISION,
        model_quantization_policy=MODEL_QUANTIZATION_POLICY,
        auto_duration_min_seconds=EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
        auto_duration_max_seconds=EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
        preload_state=PRELOAD_STATE,
        sanitize_lora_defs=_sanitize_lora_defs,
    )


def _write_settings_export(
    prompt, duration_seconds, experimental_long, resolution_key, seed, randomize_seed, selected_loras,
    lora_strength, custom_loras, use_diffusion_decoder, use_auto_duration, start_image_path, middle_image_path, end_image_path, session_id,
):
    session_id = normalize_session_id(session_id)
    path = WORKER_ROOT / session_id / "settings_exports" / f"ltx25_settings_{uuid.uuid4().hex}.json"
    payload = _settings_snapshot(
        prompt, duration_seconds, experimental_long, resolution_key, seed, randomize_seed,
        selected_loras, lora_strength, custom_loras, use_diffusion_decoder, use_auto_duration, start_image_path, middle_image_path, end_image_path,
    )
    return settings_history.write_settings_export(path, payload)


def _apply_settings_payload(
    payload, import_prompt, import_generation, import_seed, import_lora, import_session_loras,
    current_prompt, current_duration, current_experimental, current_resolution, current_seed,
    current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
    current_use_auto_duration,
):
    return settings_history.apply_settings_payload(
        payload, import_prompt, import_generation, import_seed, import_lora, import_session_loras,
        current_prompt, current_duration, current_experimental, current_resolution, current_seed,
        current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
        current_use_auto_duration,
        experimental_max_seconds=EXPERIMENTAL_MAX_SECONDS,
        standard_max_seconds=STANDARD_MAX_SECONDS,
        auto_duration_min_seconds=EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
        auto_duration_max_seconds=EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
        diffusion_decoder_available=DIFFUSION_DECODE_PIPE is not None,
        auto_duration_available=bool(getattr(PIPE, "duration_head", None) is not None),
        parse_resolution_value=_parse_resolution_value,
        supports_experimental_long=_supports_experimental_long,
        duration_slider_update=_duration_slider_update,
        sanitize_lora_defs=_sanitize_lora_defs,
        all_lora_defs=_all_lora_defs,
        lora_dropdown=_lora_dropdown,
    )


def _import_settings(
    settings_file, import_prompt, import_generation, import_seed, import_lora, import_session_loras,
    current_prompt, current_duration, current_experimental, current_resolution, current_seed,
    current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
    current_use_auto_duration,
):
    payload = settings_history.read_settings_payload(settings_file)
    values, warnings = _apply_settings_payload(
        payload, import_prompt, import_generation, import_seed, import_lora, import_session_loras,
        current_prompt, current_duration, current_experimental, current_resolution, current_seed,
        current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
        current_use_auto_duration,
    )
    provenance = payload.get("runtime_provenance") or {}
    requested = provenance.get("requested_model") or {}
    note = (
        "Applied selected portable settings. Runtime provenance/model override fields are read-only and were not imported. "
        f"Source build: `{payload.get('candidate_id', '?')}`; source model: {runtime_utils.hf_repo_markdown_link(requested.get('repo_id'))}."
        + settings_history.warning_note(warnings)
    )
    return (*values, note)


def _history_label(record: dict) -> str:
    return settings_history.history_label(record)


def _history_outputs(history) -> tuple:
    return settings_history.history_outputs(history)


def _append_probe_output(existing, probe_path):
    values = probe_artifacts.append_probe_path(
        existing, probe_path, limit=max(12, int(HISTORY_LIMIT) * 3)
    )
    return values, values


def _refresh_latest_probe_output(existing, session_id):
    found = probe_artifacts.list_session_probe_paths(WORKER_ROOT, normalize_session_id(session_id))
    values = list(existing or [])
    if found:
        values = probe_artifacts.append_probe_path(
            values, found[-1], limit=max(12, int(HISTORY_LIMIT) * 3)
        )
    return values, values


def _clear_generation_result():
    return None, ""


def _update_history(history, record: dict) -> list[dict]:
    return settings_history.update_history(history, record, history_limit=HISTORY_LIMIT)


def _restore_history_settings(
    history, selected_history, current_prompt, current_duration, current_experimental, current_resolution,
    current_seed, current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
    current_use_auto_duration,
):
    history = list(history or [])
    match = next((item for item in history if _history_label(item) == selected_history), None)
    if match is None:
        raise gr.Error("Choose a recent run first.")
    payload = match.get("settings") or {}
    values, warnings = _apply_settings_payload(
        payload, True, True, True, True, True,
        current_prompt, current_duration, current_experimental, current_resolution, current_seed,
        current_randomize, current_selected_loras, current_lora_strength, current_custom_loras, current_use_diffusion_decoder,
        current_use_auto_duration,
    )
    note = (
        f"Restored settings from run `{match.get('request_id')}`. Input images are not restored."
        + settings_history.warning_note(warnings)
    )
    return (*values, note)


def _maintainer_gate_markdown() -> str:
    gate_count = len(MAINTAINER_PROBE_CURRENT_GATES)
    if gate_count == 0:
        return "**Current maintainer gate: 0.** No Probe is required for promotion right now. Optional profiles may still be used for diagnostics."
    runtime_note = ""
    if not IS_FULL_SFT_PROFILE:
        runtime_note = "  \n⚠️ This gate requires the `full_sft_nf4` runtime profile; switch profile and restart before running it."
    return (
        f"**Current maintainer gate: {gate_count}.** Run the preselected profile below to close the current gate. "
        "Other Probe profiles remain optional diagnostics and do not become promotion requirements merely by existing."
        f"{runtime_note}"
    )


def _maintainer_probe_profile_description(profile_id) -> str:
    return maintainer_probe_catalog.profile_description(profile_id)


def _maintainer_probe_profile_ui(profile_id):
    selected = str(profile_id or "")
    supported = selected in {
        MAINTAINER_PROBE_FULL_SFT_30S_FLF2V_STAGE1,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV_FRAMEWISE,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_LATENT,
        MAINTAINER_PROBE_FULL_SFT_30S_UPSAMPLE,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE1,
    }
    if selected == MAINTAINER_PROBE_FULL_SFT_30S_FLF2V_STAGE1:
        backend_ready = PIPE_CONDITION is not None
    else:
        backend_ready = PIPE_I2V is not None and UPSAMPLE_PIPE is not None
        if selected == MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION:
            backend_ready = backend_ready and DIFFUSION_DECODE_PIPE is not None
    runnable = supported and IS_FULL_SFT_PROFILE and backend_ready
    return _maintainer_probe_profile_description(selected), gr.Button(value="Run Probe", variant="primary", interactive=bool(runnable))


def _maintainer_probe_duration(profile_id, session_id, progress=None):
    del session_id, progress
    selected = str(profile_id or "")
    durations = {
        MAINTAINER_PROBE_FULL_SFT_30S_FLF2V_STAGE1: maintainer_probe_backend.FULL_SFT_30S_FLF2V_STAGE1_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION: maintainer_probe_backend.FULL_SFT_30S_STAGE2_DIFFUSION_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV: maintainer_probe_backend.FULL_SFT_30S_STAGE2_CONV_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV_FRAMEWISE: maintainer_probe_backend.FULL_SFT_30S_STAGE2_CONV_FRAMEWISE_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_LATENT: maintainer_probe_backend.FULL_SFT_30S_STAGE2_LATENT_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_UPSAMPLE: maintainer_probe_backend.FULL_SFT_30S_UPSAMPLE_ZEROGPU_DURATION_SECONDS,
        MAINTAINER_PROBE_FULL_SFT_30S_STAGE1: maintainer_probe_backend.FULL_SFT_30S_STAGE1_ZEROGPU_DURATION_SECONDS,
    }
    return int(durations.get(selected, 5))


def _maintainer_probe_cats_example() -> tuple[str, str]:
    items = list((EXAMPLE_STATE or {}).get("items") or [])
    for row, item in zip(EXAMPLE_ROWS, items):
        if str(item.get("label") or "") == "Cats · I2V" and row and row[0] and len(row) >= 3:
            return str(row[0]), str(row[2])
    raise gr.Error("Probe requires the startup Cats · I2V example, but it is unavailable on this worker.")


def _maintainer_probe_flf2v_example() -> tuple[str, str, str]:
    items = list((EXAMPLE_STATE or {}).get("items") or [])
    for row, item in zip(EXAMPLE_ROWS, items):
        if str(item.get("label") or "") == "Blue bird · first + last frame" and row and len(row) >= 3 and row[0] and row[1]:
            return str(row[0]), str(row[1]), str(row[2])
    raise gr.Error("Probe requires the startup Blue bird · first + last frame example, but it is unavailable on this worker.")


def _run_maintainer_backend(profile_id, session_id, progress):
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_FLF2V_STAGE1:
        return maintainer_probe_backend.run_full_sft_30s_flf2v_stage1_scout(
            session_id=session_id,
            progress=progress,
            candidate_id=CANDIDATE_ID,
            runtime=_generation_runtime(),
            hooks=_generation_hooks(),
            flf2v_example=_maintainer_probe_flf2v_example(),
            create_request_paths=create_request_paths,
            open_request_logger=_open_request_logger,
            close_request_logger=_close_request_logger,
        )
    common = dict(
        session_id=session_id,
        progress=progress,
        candidate_id=CANDIDATE_ID,
        runtime=_generation_runtime(),
        hooks=_generation_hooks(),
        cats_example=_maintainer_probe_cats_example(),
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
    )
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION:
        return maintainer_probe_backend.run_full_sft_30s_stage2_diffusion_scout(**common)
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV:
        return maintainer_probe_backend.run_full_sft_30s_stage2_conv_scout(**common)
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_CONV_FRAMEWISE:
        return maintainer_probe_backend.run_full_sft_30s_stage2_conv_framewise_scout(**common)
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_LATENT:
        return maintainer_probe_backend.run_full_sft_30s_stage2_latent_scout(**common)
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_UPSAMPLE:
        return maintainer_probe_backend.run_full_sft_30s_upsample_scout(**common)
    if profile_id == MAINTAINER_PROBE_FULL_SFT_30S_STAGE1:
        return maintainer_probe_backend.run_full_sft_30s_stage1_scout(**common)
    raise gr.Error("Select a supported Maintainer Probe profile first.")


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_maintainer_probe_duration)
def run_maintainer_probe(profile_id, session_id, progress=gr.Progress(track_tqdm=False)):
    return _run_maintainer_backend(str(profile_id or ""), session_id, progress)


def _research_probe_profile_description(profile_id) -> str:
    selected = str(profile_id or "")
    spec = research_probe_backend.RESEARCH_PROFILE_SPECS.get(selected)
    if spec is None:
        return (
            "**Gate-free Research Probe.** Choose any profile whenever useful. Research failures are evidence, not promotion failures; "
            "shared-runtime profiles remain serial until isolation is proven."
        )
    return (
        "**RESEARCH · gate-free · aggressive exploration**  \n"
        f"{spec['label']}  \n"
        f"Fixed contract: Full/SFT · I2V · Cats · 512×512 · **{spec['frames']}f · {spec['steps']} Stage-1 steps** · seed 42 · no user LoRA · latent output only. "
        f"Reservation: **{spec['duration']}s**. Timing is optional; run whenever the information is useful."
    )


def _research_probe_profile_ui(profile_id):
    selected = str(profile_id or "")
    runnable = selected in research_probe_backend.RESEARCH_PROFILE_SPECS and IS_FULL_SFT_PROFILE and PIPE_I2V is not None
    return _research_probe_profile_description(selected), gr.Button(
        value="Run Research Probe", variant="secondary", interactive=bool(runnable)
    )


def _research_probe_duration(profile_id, session_id, progress=None):
    del session_id, progress
    spec = research_probe_backend.RESEARCH_PROFILE_SPECS.get(str(profile_id or ""))
    return int(spec['duration']) if spec else 5


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_research_probe_duration)
def run_research_probe(profile_id, session_id, progress=gr.Progress(track_tqdm=False)):
    return research_probe_backend.run_research_probe(
        str(profile_id or ""),
        session_id=session_id,
        progress=progress,
        candidate_id=CANDIDATE_ID,
        runtime=_generation_runtime(),
        hooks=_generation_hooks(),
        cats_example=_maintainer_probe_cats_example(),
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
    )


def _generation_runtime() -> generation_backend.GenerationRuntime:
    return generation_backend.GenerationRuntime(
        candidate_id=CANDIDATE_ID,
        diffusion_decode_pipe=DIFFUSION_DECODE_PIPE,
        auto_duration_enabled=EFFECTIVE_AUTO_DURATION_ENABLED,
        auto_duration_max_seconds=EFFECTIVE_AUTO_DURATION_MAX_SECONDS,
        auto_duration_min_seconds=EFFECTIVE_AUTO_DURATION_MIN_SECONDS,
        full_sft_stage2_lora_strength=EFFECTIVE_FULL_SFT_STAGE2_LORA_STRENGTH,
        experimental_max_seconds=EXPERIMENTAL_MAX_SECONDS,
        frame_rate=FRAME_RATE,
        full_sft_stage2_adapter_name=FULL_SFT_STAGE2_ADAPTER_NAME,
        is_full_sft_profile=IS_FULL_SFT_PROFILE,
        pipe=PIPE,
        pipe_condition=PIPE_CONDITION,
        pipe_i2v=PIPE_I2V,
        preload_state=PRELOAD_STATE,
        runtime_profile=RUNTIME_PROFILE,
        standard_max_seconds=STANDARD_MAX_SECONDS,
        upsample_pipe=UPSAMPLE_PIPE,
        worker_uuid=WORKER_UUID,
    )


def _generation_hooks() -> generation_backend.GenerationHooks:
    return generation_backend.GenerationHooks(
        cleanup_request_loras=_cleanup_request_loras,
        close_request_logger=_close_request_logger,
        disk_state=_disk_state,
        duration=_duration,
        frames_from_seconds=_frames_from_seconds,
        gpu_state=_gpu_state,
        history_outputs=_history_outputs,
        load_conditioning_image=_load_conditioning_image,
        load_request_loras=_load_request_loras,
        lora_adapter_state=_lora_adapter_state,
        mode_from_images=_mode_from_condition_images,
        open_request_logger=_open_request_logger,
        resolution=_resolution,
        settings_snapshot=_settings_snapshot,
        sha256_file=_sha256_file,
        supports_experimental_long=_supports_experimental_long,
        update_history=_update_history,
        create_request_paths=create_request_paths,
    )


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_duration)
def generate(
    prompt,
    start_image_path,
    middle_image_path,
    end_image_path,
    duration_seconds,
    experimental_long,
    resolution_key,
    seed,
    randomize_seed,
    selected_loras,
    lora_strength,
    custom_loras,
    prepared_loras,
    use_diffusion_decoder,
    use_auto_duration,
    session_id,
    history,
    progress=gr.Progress(track_tqdm=True),
):
    return generation_backend.run(
        prompt, start_image_path, middle_image_path, end_image_path, duration_seconds, experimental_long, resolution_key,
        seed, randomize_seed, selected_loras, lora_strength, custom_loras, prepared_loras,
        use_diffusion_decoder, use_auto_duration, session_id, history, progress,
        runtime=_generation_runtime(),
        hooks=_generation_hooks(),
    )



def _run_prompt_tool_separate(
    tool: str,
    prompt,
    start_image_path,
    middle_image_path,
    end_image_path,
    session_id,
    manual_system_prompt=None,
):
    global PROMPT_ENHANCER_MODEL, PROMPT_ENHANCER_PROCESSOR
    session_id = normalize_session_id(session_id)
    request_id = uuid.uuid4().hex
    mode = _mode_from_condition_images(start_image_path, middle_image_path, end_image_path)
    model_source = ((PRELOAD_STATE.get("model_sources") or {}).get("prompt_enhancer") or {})
    output, status_text, result, disable_global_model = prompt_backend.run(
        tool=tool,
        prompt=prompt,
        start_image_path=start_image_path,
        end_image_path=end_image_path,
        session_id=session_id,
        request_id=request_id,
        candidate_id=CANDIDATE_ID,
        worker_uuid=WORKER_UUID,
        mode=mode,
        pipe=PIPE,
        model=PROMPT_ENHANCER_MODEL,
        processor=PROMPT_ENHANCER_PROCESSOR,
        runtime_ready=(PRELOAD_STATE.get("status") == "ready"),
        model_source=model_source,
        gpu_state=_gpu_state,
        sha256_file=_sha256_file,
        official_t2v_system_prompt=LTX2_5_T2V_DEFAULT_SYSTEM_PROMPT,
        official_i2v_system_prompt=LTX2_5_I2V_DEFAULT_SYSTEM_PROMPT,
        translate_system_prompt=PROMPT_TRANSLATE_LTX_ENGLISH_SYSTEM_PROMPT,
        write_system_prompt=PROMPT_WRITE_FROM_SEED_SYSTEM_PROMPT,
        manual_system_prompt=manual_system_prompt,
        requested_duration_seconds=PROMPT_TOOL_ZEROGPU_DURATION_SECONDS,
    )
    if disable_global_model:
        PROMPT_ENHANCER_MODEL = None
        PROMPT_ENHANCER_PROCESSOR = None
    APP_LOGGER.info("[D1R8P30R0] prompt_tools " + json.dumps(result, sort_keys=True))
    return output, status_text


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=PROMPT_TOOL_ZEROGPU_DURATION_SECONDS)
def enhance_prompt_separate(prompt, start_image_path, middle_image_path, end_image_path, session_id):
    return _run_prompt_tool_separate(
        prompt_backend.TOOL_ENHANCE, prompt, start_image_path, middle_image_path, end_image_path, session_id
    )


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=PROMPT_TOOL_ZEROGPU_DURATION_SECONDS)
def translate_prompt_separate(prompt, manual_enabled, manual_system_prompt, session_id):
    return _run_prompt_tool_separate(
        prompt_backend.TOOL_TRANSLATE,
        prompt,
        None,
        None,
        None,
        session_id,
        manual_system_prompt if manual_enabled else None,
    )


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=PROMPT_TOOL_ZEROGPU_DURATION_SECONDS)
def write_prompt_from_seed_separate(prompt, manual_enabled, manual_system_prompt, session_id):
    return _run_prompt_tool_separate(
        prompt_backend.TOOL_WRITE,
        prompt,
        None,
        None,
        None,
        session_id,
        manual_system_prompt if manual_enabled else None,
    )



def _audio_to_video_config() -> a2v_backend.AudioToVideoConfig:
    audio_vae = getattr(PIPE_I2V, "audio_vae", None) or getattr(PIPE, "audio_vae", None)
    audio_config = getattr(audio_vae, "config", None)
    sample_rate = int(getattr(audio_config, "sample_rate", 16000))
    hop_length = int(getattr(audio_config, "mel_hop_length", 160))
    mel_bins = int(getattr(audio_config, "mel_bins", 64))
    compression = int(
        getattr(
            PIPE_I2V,
            "audio_vae_temporal_compression_ratio",
            getattr(audio_vae, "temporal_compression_ratio", 4),
        )
        or 4
    )
    return a2v_backend.AudioToVideoConfig(
        enabled=EFFECTIVE_A2V_ENABLED,
        is_full_sft_profile=IS_FULL_SFT_PROFILE,
        frame_rate=FRAME_RATE,
        width=int(A2V_WIDTH),
        height=int(A2V_HEIGHT),
        num_frames=int(A2V_FRAMES),
        zerogpu_duration_seconds=int(A2V_ZEROGPU_DURATION_SECONDS),
        audio_sample_rate=sample_rate,
        audio_hop_length=hop_length,
        audio_mel_bins=mel_bins,
        audio_compression_ratio=compression,
        example_specs=tuple(A2V_EXAMPLE_SPECS or []),
    )


def _audio_to_video_runtime() -> a2v_backend.AudioToVideoRuntime:
    return a2v_backend.AudioToVideoRuntime(
        candidate_id=CANDIDATE_ID,
        worker_uuid=WORKER_UUID,
        worker_root=WORKER_ROOT,
        pipe=PIPE,
        pipe_i2v=PIPE_I2V,
        upsample_pipe=UPSAMPLE_PIPE,
        preload_state=PRELOAD_STATE,
        full_sft_stage2_adapter_name=FULL_SFT_STAGE2_ADAPTER_NAME,
        full_sft_stage2_lora_strength=EFFECTIVE_FULL_SFT_STAGE2_LORA_STRENGTH,
        create_request_paths=create_request_paths,
        open_request_logger=_open_request_logger,
        close_request_logger=_close_request_logger,
        load_conditioning_image=_load_conditioning_image,
        gpu_state=_gpu_state,
        sha256_file=_sha256_file,
        lora_adapter_state=_lora_adapter_state,
        normalize_session_id=normalize_session_id,
        example_dir=RUNTIME_A2V_EXAMPLE_DIR,
        app_logger=APP_LOGGER,
        hf_token=HF_TOKEN,
        resolved_revision_from_hub_path=runtime_utils.resolved_revision_from_hub_path,
    )


def _prepare_a2v_examples() -> None:
    global A2V_EXAMPLE_ROWS, A2V_EXAMPLE_STATE
    A2V_EXAMPLE_ROWS, A2V_EXAMPLE_STATE = a2v_backend.prepare_examples(
        _audio_to_video_config(),
        _audio_to_video_runtime(),
    )
    PRELOAD_STATE["a2v_examples"] = A2V_EXAMPLE_STATE


# A2V example localization depends on the helpers above being bound at module startup.
_prepare_a2v_examples()


def prepare_audio_to_video(audio_value, image_value, prompt, seed, randomize_seed, session_id):
    return a2v_backend.prepare(
        audio_value,
        image_value,
        prompt,
        seed,
        randomize_seed,
        session_id,
        config=_audio_to_video_config(),
        runtime=_audio_to_video_runtime(),
    )


def _audio_to_video_duration(prepared, session_id, *args):  # noqa: ARG001
    return a2v_backend.duration(
        prepared,
        session_id,
        default_seconds=int(A2V_ZEROGPU_DURATION_SECONDS),
    )


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_audio_to_video_duration)
def generate_audio_to_video(prepared, session_id, progress=gr.Progress(track_tqdm=False)):
    return a2v_backend.generate(
        prepared,
        session_id,
        progress,
        config=_audio_to_video_config(),
        runtime=_audio_to_video_runtime(),
    )


def prepare_ic_colorizer(video, prompt, profile_key, seed, randomize_seed, session_id):
    return ic_backend.prepare(
        video,
        prompt,
        profile_key,
        seed,
        randomize_seed,
        session_id,
        config=_ic_colorizer_config(),
        runtime=_ic_colorizer_runtime(),
    )


def _ic_colorizer_duration(prepared, session_id, *args):  # noqa: ARG001
    return ic_backend.duration(prepared, config=_ic_colorizer_config())


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_ic_colorizer_duration)
def generate_ic_colorizer(prepared, session_id, progress=gr.Progress(track_tqdm=False)):
    return ic_backend.generate(
        prepared,
        session_id,
        progress,
        config=_ic_colorizer_config(),
        runtime=_ic_colorizer_runtime(),
    )


def prepare_pixel_upscaler(video, prompt, seed, randomize_seed, session_id):
    return pixel_upscaler_backend.prepare(
        video,
        prompt,
        seed,
        randomize_seed,
        session_id,
        config=_pixel_upscaler_config(),
        runtime=_pixel_upscaler_runtime(),
    )


def _pixel_upscaler_duration(prepared, session_id, *args):  # noqa: ARG001
    return pixel_upscaler_backend.duration(prepared, config=_pixel_upscaler_config())


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_pixel_upscaler_duration)
def generate_pixel_upscaler(prepared, session_id, progress=gr.Progress(track_tqdm=False)):
    return pixel_upscaler_backend.generate(
        prepared,
        session_id,
        progress,
        config=_pixel_upscaler_config(),
        runtime=_pixel_upscaler_runtime(),
    )


def prepare_inoutpaint(video, mask, prompt, mode, layout, dilation_px, seed, randomize_seed, session_id):
    return inoutpaint_backend.prepare(
        video, mask, prompt, mode, layout, dilation_px, seed, randomize_seed, session_id,
        config=_inoutpaint_config(), runtime=_inoutpaint_runtime(),
    )


def _inoutpaint_duration(prepared, session_id, *args):  # noqa: ARG001
    return inoutpaint_backend.duration(prepared, config=_inoutpaint_config())


@spaces.GPU(size=EFFECTIVE_ZEROGPU_GPU_SIZE, duration=_inoutpaint_duration)
def generate_inoutpaint(prepared, session_id, progress=gr.Progress(track_tqdm=False)):
    return inoutpaint_backend.generate(
        prepared, session_id, progress, config=_inoutpaint_config(), runtime=_inoutpaint_runtime(),
    )


with gr.Blocks(title="LTX-2.5 Video", delete_cache=(3600, 10800)) as demo:
    session_id_state = gr.State(value=create_session_id, delete_callback=cleanup_session_results)
    custom_lora_state = gr.State(value=lambda: [])
    prepared_lora_state = gr.State(value=lambda: [])
    history_state = gr.State(value=lambda: [])
    probe_files_state = gr.State(value=lambda: [])
    latest_probe_state = gr.State(value=None)
    maintainer_probe_files_state = gr.State(value=lambda: [])
    maintainer_latest_probe_state = gr.State(value=None)
    research_probe_files_state = gr.State(value=lambda: [])
    research_latest_probe_state = gr.State(value=None)
    a2v_prepared_state = gr.State(value=lambda: {})
    a2v_probe_files_state = gr.State(value=lambda: [])
    a2v_latest_probe_state = gr.State(value=None)
    ic_prepared_state = gr.State(value=lambda: {})
    ic_probe_files_state = gr.State(value=lambda: [])
    ic_latest_probe_state = gr.State(value=None)
    pixel_upscaler_prepared_state = gr.State(value=lambda: {})
    pixel_upscaler_probe_files_state = gr.State(value=lambda: [])
    pixel_upscaler_latest_probe_state = gr.State(value=None)
    inoutpaint_prepared_state = gr.State(value=lambda: {})
    inoutpaint_probe_files_state = gr.State(value=lambda: [])
    inoutpaint_latest_probe_state = gr.State(value=None)

    hero_runtime = (
        "Full/SFT guided Stage 1 + distilled-LoRA Stage 2 · synchronized audio · 48GB-oriented runtime"
        if IS_FULL_SFT_PROFILE
        else "Distilled two-stage generation · synchronized audio · 48GB-oriented runtime"
    )
    gr.HTML(
        f'<div id="ltx-hero"><h1>LTX-2.5 Video</h1><p>{hero_runtime}</p></div>'
    )
    if IS_FULL_SFT_PROFILE:
        gr.Markdown(
            "**Runtime profile: Full / SFT · restart-required exclusive.** Stage 1 uses the NF4 full transformer "
            "with 30-step CFG/STG guidance. User LoRAs use the same picker as distilled mode: they apply in Stage 1 "
            "and remain active in Stage 2, where the internal distilled adapter is added automatically. The internal "
            "adapter source/strength is configured only in `space_config.py`; switching profiles requires a restart."
        )
    gr.Markdown(_runtime_model_status_markdown(), elem_id="runtime-model-status")

    with gr.Tabs():
        with gr.Tab("Generate"):
            with gr.Row(equal_height=False):
                with gr.Column(scale=6):
                    prompt_tools_ready = bool(
                        PROMPT_ENHANCER_MODEL is not None and PROMPT_ENHANCER_PROCESSOR is not None
                    )
                    with gr.Group():
                        prompt = gr.Textbox(
                            label="Prompt",
                            lines=4,
                            value=DEFAULT_PROMPT,
                        )
                        with gr.Row(equal_height=True):
                            enhance_prompt_btn = gr.Button(
                                "Enhance Prompt",
                                variant="secondary",
                                interactive=prompt_tools_ready,
                            )
                            translate_prompt_btn = gr.Button(
                                "Translate → LTX English",
                                variant="secondary",
                                interactive=prompt_tools_ready,
                            )
                            write_prompt_btn = gr.Button(
                                "Write from Seed",
                                variant="secondary",
                                interactive=prompt_tools_ready,
                            )
                        prompt_tools_status = gr.Markdown()
                        with gr.Accordion("Prompt Tools · advanced", open=False):
                            gr.Markdown(
                                "All three actions reuse the same dedicated Gemma 4 runtime and replace the Prompt textbox only after a successful result. "
                                "**Enhance Prompt** keeps the official LTX-2.5 prompt-enhancement contract; for any image-conditioned mode it uses the Start image only. "
                                "**Translate** and **Write from Seed** are text-only. Their full system prompts can be overridden below without changing Enhance Prompt."
                            )
                            translate_manual_enabled = gr.Checkbox(
                                False, label="Translate · use full manual system prompt"
                            )
                            translate_manual_prompt = gr.Textbox(
                                label="Translate · full system prompt",
                                value=PROMPT_TRANSLATE_LTX_ENGLISH_SYSTEM_PROMPT,
                                lines=8,
                            )
                            write_manual_enabled = gr.Checkbox(
                                False, label="Write from Seed · use full manual system prompt"
                            )
                            write_manual_prompt = gr.Textbox(
                                label="Write from Seed · full system prompt",
                                value=PROMPT_WRITE_FROM_SEED_SYSTEM_PROMPT,
                                lines=8,
                            )

                    with gr.Accordion("Image conditioning / keyframes · optional", open=False):
                        with gr.Row(equal_height=True):
                            image = gr.Image(
                                label="Start image · optional",
                                type="filepath",
                                sources=["upload", "clipboard"],
                                height=250,
                            )
                            middle_image = gr.Image(
                                label="Middle image · optional · experimental keyframe",
                                type="filepath",
                                sources=["upload", "clipboard"],
                                height=250,
                            )
                            end_image = gr.Image(
                                label="End image · optional · experimental FLF2V",
                                type="filepath",
                                sources=["upload", "clipboard"],
                                height=250,
                            )
                        gr.Markdown(
                            "No image → **T2V** · Start only → **I2V** · Start + End → **experimental FLF2V** · "
                            "Start + Middle (optionally + End) → **experimental timeline keyframe**. "
                            "Middle is snapped to the nearest model-latent midpoint; Middle/End without Start is rejected.",
                            elem_classes=["ltx-subtle"],
                        )
                        conditioning_mode_status = gr.Markdown(
                            _conditioning_mode_status_markdown(None, None, None),
                            elem_classes=["ltx-subtle"],
                        )

                    with gr.Row():
                        duration_seconds = gr.Slider(
                            minimum=1.0,
                            maximum=STANDARD_MAX_SECONDS,
                            step=0.5,
                            value=min(STANDARD_MAX_SECONDS, max(1.0, EFFECTIVE_DEFAULT_DURATION_SECONDS)),
                            label="Duration",
                            info=(
                                "1–15s standard; 15–30s experimental at 512×512 only. "
                                "Rounded to the required 8k+1 frame grid at 24 fps."
                            ),
                        )
                        resolution = gr.Dropdown(
                            choices=list(RESOLUTIONS.keys()),
                            value=EFFECTIVE_DEFAULT_RESOLUTION,
                            label="Resolution",
                            allow_custom_value=True,
                            info="Choose a preset, or use Custom resolution below. Direct WIDTH × HEIGHT entry also remains supported; both dimensions must be multiples of 64.",
                        )
                    custom_resolution_width, custom_resolution_height, apply_custom_resolution_btn, custom_resolution_status = (
                        _build_custom_resolution_editor(EFFECTIVE_DEFAULT_RESOLUTION)
                    )
                    duration_options_label = "Duration / ZeroGPU quota" if IS_ZEROGPU else "Duration options"
                    with gr.Accordion(duration_options_label, open=False):
                        experimental_long = gr.Checkbox(
                            value=False,
                            interactive=_supports_experimental_long(EFFECTIVE_DEFAULT_RESOLUTION),
                            label="Experimental long duration (>15s)",
                            info="Allows 15–30s only at 512×512. The 30s / 721f endpoint is live-passed for distilled T2V and I2V.",
                        )
                        use_auto_duration = gr.Checkbox(
                            value=EFFECTIVE_DEFAULT_USE_AUTO_DURATION,
                            interactive=bool(getattr(PIPE, "duration_head", None) is not None),
                            label="Auto duration · experimental",
                            info=(
                                f"Let the LTX-2.5 duration head choose the shot length within "
                                f"{EFFECTIVE_AUTO_DURATION_MIN_SECONDS:.1f}–{EFFECTIVE_AUTO_DURATION_MAX_SECONDS:.1f}s. "
                                "The manual Duration slider is ignored while enabled."
                            ),
                        )
                        gr.Markdown(
                            "Long duration is explicit opt-in. Full/SFT is limited here to the tested 121f / 5s range at 512×512; Distilled is the recommended deployment profile. Auto Duration stays within its supported range.",
                            elem_classes=["ltx-subtle"],
                        )
                        if IS_ZEROGPU:
                            gr.Markdown(
                                "ZeroGPU checks requested duration before execution; realistic shorter requests can improve queue priority. The estimate below is quota/runtime reservation, not video length. LoRA Hub files are prepared on CPU before GPU reservation.",
                                elem_classes=["ltx-subtle"],
                            )
                            zerogpu_duration_estimate = gr.Markdown(value=_zerogpu_duration_panel(
                                None, None, None, min(STANDARD_MAX_SECONDS, max(1.0, EFFECTIVE_DEFAULT_DURATION_SECONDS)), False,
                                EFFECTIVE_DEFAULT_RESOLUTION, EFFECTIVE_DEFAULT_SELECTED_LORAS,
                                EFFECTIVE_DEFAULT_USE_DIFFUSION_DECODER, EFFECTIVE_DEFAULT_USE_AUTO_DURATION,
                            ))
                            if IS_FULL_SFT_PROFILE:
                                gr.Markdown(
                                    "Full/SFT validation profile: **90s validated 25f floor**; >25f uses the evidence-based estimator, and each selected user LoRA adds **8s** reservation margin.",
                                    elem_classes=["ltx-subtle"],
                                )
                            if PROMPT_ENHANCER_MODEL is not None:
                                gr.Markdown(
                                    f"Prompt Tools use a separate **{PROMPT_TOOL_ZEROGPU_DURATION_SECONDS}s GPU reservation** per action.",
                                    elem_classes=["ltx-subtle"],
                                )

                    lora_ui = ui_lora.build_lora_ui(
                        lora_dropdown_factory=_lora_dropdown,
                        default_strength=EFFECTIVE_DEFAULT_LORA_STRENGTH,
                        hub_backend=hub_search_backend,
                        civitai_backend=civitai_backend,
                    )
                    # Keep the generation/settings names local and concrete. Discovery-only
                    # component details stay owned by ltx.ui_lora.
                    lora_dropdown = lora_ui.dropdown
                    lora_strength = lora_ui.strength
                    lora_prepare_status = lora_ui.prepare_status
                    civitai_api_key = lora_ui.civitai_api_key

                    with gr.Accordion("Advanced settings", open=False):
                        seed, randomize_seed = _build_seed_controls(
                            seed_value=EFFECTIVE_DEFAULT_SEED,
                            randomize_value=EFFECTIVE_DEFAULT_RANDOMIZE_SEED,
                        )

                        use_diffusion_decoder = gr.Checkbox(
                            value=EFFECTIVE_DEFAULT_USE_DIFFUSION_DECODER,
                            interactive=bool(DIFFUSION_DECODE_PIPE is not None),
                            label="Diffusion decoder",
                            info=(
                                "Optional iterative decode. Startup RAM-ready / GPU-lazy, NATTEN + tiled decode. "
                                "Live-passed at 512×512 for 25f and 361f/15s on ZeroGPU large 48GB; Conv VAE remains the default."
                            ),
                        )

                        with gr.Accordion("Settings Export / Import", open=False):
                            gr.Markdown(
                                "Export writes all portable UI settings plus read-only runtime provenance. "
                                "Input media and secrets are never embedded. Import applies only the categories you select."
                            )
                            export_settings_btn = gr.Button("Export all settings", variant="secondary")
                            settings_download = gr.DownloadButton("Download settings JSON", value=None, size="sm")
                            settings_file = gr.File(
                                label="Import settings JSON",
                                file_types=[".json"],
                                file_count="single",
                                type="filepath",
                            )
                            with gr.Row():
                                import_prompt = gr.Checkbox(True, label="Prompt")
                                import_generation = gr.Checkbox(True, label="Generation")
                                import_seed = gr.Checkbox(True, label="Seed")
                            with gr.Row():
                                import_lora = gr.Checkbox(True, label="LoRA selection / strength")
                                import_session_loras = gr.Checkbox(False, label="Session LoRA definitions")
                            import_settings_btn = gr.Button("Apply selected settings", variant="secondary")
                            settings_status = gr.Markdown()

                    go = gr.Button("Generate video", variant="primary", elem_id="generate-btn")
                    generation_preflight_status = gr.Markdown(
                        "Resolution/LoRA CPU preflight runs before any ZeroGPU reservation is requested." if IS_ZEROGPU
                        else "Resolution/LoRA CPU preflight runs before generation starts.", elem_classes=["ltx-subtle"],
                    )

                with gr.Column(scale=7):
                    out, used_seed, probe_files = _build_result_panel(
                        result_label="Result",
                        probe_note="Probe ZIPs contain run info, diagnostics, request log and SHA256 manifest; video remains separate for visual review.",
                    )

                    with gr.Accordion("Recent history", open=False):
                        history_summary = gr.Markdown("No completed runs in this browser session yet.")
                        history_choice = gr.Dropdown(choices=[], value=None, label="Recent run", interactive=False)
                        history_restore_btn = gr.Button("Restore settings from selected run", variant="secondary")

            if EXAMPLE_ROWS:
                gr.Examples(
                    examples=[[row[0], None, row[1], row[2]] for row in EXAMPLE_ROWS],
                    inputs=[image, middle_image, end_image, prompt],
                    label="Examples · click to fill inputs",
                    cache_examples=False,
                )
                gr.Markdown(example_runtime.attribution_markdown(EXAMPLE_STATE), elem_classes=["ltx-subtle"])

        with gr.Tab("Audio-to-Video"):
            gr.Markdown(
                "### Audio-to-Video\n"
                "The uploaded audio drives video generation while its latent stays frozen in both diffusion stages. "
                "The original decoded input waveform remains the final audio track. Current validated profile: **512×512 / 121f / 24fps**. "
                "This is audio-reactive generation; frame-exact lip sync is not claimed."
            )
            if not IS_FULL_SFT_PROFILE:
                gr.Markdown(
                    "**Unavailable outside `full_sft_nf4`.** Switch `MODEL_RUNTIME_PROFILE` in `space_config.py` and restart."
                )
            else:
                required_audio_seconds = float(A2V_FRAMES) / FRAME_RATE
                gr.Markdown(
                    f"Upload at least **{required_audio_seconds:.3f}s** of audio. CPU preflight freezes the source, decodes it with PyAV and "
                    f"builds the LTX-compatible log-mel cache before the **{int(A2V_ZEROGPU_DURATION_SECONDS)}s** GPU reservation starts."
                )
            with gr.Row(equal_height=False):
                with gr.Column(scale=6):
                    a2v_image = gr.Image(
                        label="Start image",
                        type="filepath",
                        sources=["upload", "clipboard"],
                        height=250,
                    )
                    a2v_audio = gr.Audio(
                        label="Driving audio · upload ≥5.042s",
                        type="filepath",
                        sources=["upload"],
                    )
                    a2v_prompt = gr.Textbox(
                        label="Prompt",
                        lines=5,
                        value="A pair of tabby cats on a bright pink sofa react naturally to the rhythm and dynamics of the supplied audio; coherent motion, stable framing, detailed fur and expressive movement.",
                    )
                    a2v_seed, a2v_randomize_seed = _build_seed_controls(
                        seed_value=42,
                        randomize_value=False,
                    )
                    a2v_go = gr.Button(
                        "Generate audio-driven video",
                        variant="primary",
                        interactive=bool(
                            EFFECTIVE_A2V_ENABLED
                            and IS_FULL_SFT_PROFILE
                            and PIPE_I2V is not None
                            and UPSAMPLE_PIPE is not None
                        ),
                    )
                    a2v_status = gr.Markdown(
                        "CPU preflight runs first; invalid/short audio fails before GPU quota is requested."
                    )
                with gr.Column(scale=7):
                    a2v_out, a2v_used_seed, a2v_probe_files = _build_result_panel(
                        result_label="Audio-to-Video result",
                    )

            if A2V_EXAMPLE_ROWS:
                gr.Examples(
                    examples=A2V_EXAMPLE_ROWS,
                    inputs=[a2v_image, a2v_audio, a2v_prompt],
                    label="Example · image + audio + prompt",
                    cache_examples=False,
                )
                gr.Markdown(
                    "Example media is fetched at startup from the pinned [alexnasa/ltx-2-TURBO](https://huggingface.co/spaces/alexnasa/ltx-2-TURBO) donor revision. "
                    "Only the example media/prompt are reused; generation stays on this Space's LTX-2.5 Full/SFT runtime."
                )

        ic_ui, pixel_upscaler_ui, inoutpaint_ui = ui_ic_tabs.build_ic_product_tabs(
            pipe_ic_ready=PIPE_IC is not None,
            colorizer_examples=IC_EXAMPLE_ROWS,
            colorizer_attribution=_ic_example_attribution_markdown(),
            pixel_upscaler_examples=PIXEL_UPSCALER_EXAMPLE_ROWS,
            pixel_upscaler_attribution=_pixel_upscaler_example_attribution_markdown(),
            inoutpaint_examples=INOUTPAINT_EXAMPLE_ROWS,
            inoutpaint_attribution=_inoutpaint_example_attribution_markdown(),
        )

        # Keep the established app/event names below stable while presentation
        # ownership lives in ltx/ui_ic_tabs.py.
        ic_video = ic_ui.video
        ic_prompt = ic_ui.prompt
        ic_profile = ic_ui.profile
        ic_seed = ic_ui.seed
        ic_randomize_seed = ic_ui.randomize_seed
        ic_go = ic_ui.go
        ic_status = ic_ui.status
        ic_out = ic_ui.output
        ic_used_seed = ic_ui.used_seed
        ic_probe_files = ic_ui.probe_files

        pixel_upscaler_video = pixel_upscaler_ui.video
        pixel_upscaler_prompt = pixel_upscaler_ui.prompt
        pixel_upscaler_seed = pixel_upscaler_ui.seed
        pixel_upscaler_randomize_seed = pixel_upscaler_ui.randomize_seed
        pixel_upscaler_go = pixel_upscaler_ui.go
        pixel_upscaler_status = pixel_upscaler_ui.status
        pixel_upscaler_out = pixel_upscaler_ui.output
        pixel_upscaler_used_seed = pixel_upscaler_ui.used_seed
        pixel_upscaler_probe_files = pixel_upscaler_ui.probe_files

        inoutpaint_mode = inoutpaint_ui.mode
        inoutpaint_video = inoutpaint_ui.video
        inoutpaint_mask = inoutpaint_ui.mask
        inoutpaint_layout = inoutpaint_ui.layout
        inoutpaint_dilation = inoutpaint_ui.dilation
        inoutpaint_prompt = inoutpaint_ui.prompt
        inoutpaint_seed = inoutpaint_ui.seed
        inoutpaint_randomize_seed = inoutpaint_ui.randomize_seed
        inoutpaint_go = inoutpaint_ui.go
        inoutpaint_status = inoutpaint_ui.status
        inoutpaint_out = inoutpaint_ui.output
        inoutpaint_used_seed = inoutpaint_ui.used_seed
        inoutpaint_probe_files = inoutpaint_ui.probe_files


        if SPACE_DEBUG:
            with gr.Tab("Developer / Probes"):
                gr.Markdown(
                    "### Developer / Probes\n"
                    "Maintainer and Research Probe controls are intentionally separated from ordinary product tabs."
                )
                with gr.Accordion("Maintainer Probe", open=False):
                    maintainer_gate_status = gr.Markdown(_maintainer_gate_markdown())
                    _maintainer_initial_profile = MAINTAINER_PROBE_CURRENT_GATES[0] if MAINTAINER_PROBE_CURRENT_GATES else None
                    maintainer_probe_profile = gr.Dropdown(
                        choices=MAINTAINER_PROBE_CHOICES,
                        value=_maintainer_initial_profile,
                        label="Probe profile · optional unless listed as current gate",
                        interactive=True,
                    )
                    maintainer_probe_description = gr.Markdown(
                        _maintainer_probe_profile_description(_maintainer_initial_profile)
                    )
                    maintainer_probe_run = gr.Button(
                        "Run Probe",
                        variant="primary",
                        interactive=bool(
                            _maintainer_initial_profile
                            and IS_FULL_SFT_PROFILE
                            and PIPE_I2V is not None
                            and UPSAMPLE_PIPE is not None
                            and (_maintainer_initial_profile != MAINTAINER_PROBE_FULL_SFT_30S_STAGE2_DIFFUSION or DIFFUSION_DECODE_PIPE is not None)
                        ),
                    )
                    maintainer_probe_status = gr.Markdown(
                        "No Maintainer Probe has run in this browser session."
                    )
                    maintainer_probe_files = gr.File(
                        label="Maintainer Probe artifacts · one ZIP per request",
                        file_count="multiple",
                        interactive=False,
                    )
                    gr.Markdown(
                        f"Runtime: **{RUNTIME_PROFILE}** · ZeroGPU size: **{EFFECTIVE_ZEROGPU_GPU_SIZE}** · Candidate: `{CANDIDATE_ID}`. "
                        "Probe profiles are retained only when their purpose and safety contract remain explicit; profile presence does not imply a promotion requirement.",
                        elem_classes=["ltx-subtle"],
                    )

                with gr.Accordion("Research Probe · free research", open=False):
                    gr.Markdown(
                        "**Gate-free research line.** Profiles here may be run at any time and may fail without affecting LIVE/candidate promotion. "
                        "Default policy is maximum information gain and adventurous exploration; shared mutable runtime is still protected by serialization until a profile is proven isolated."
                    )
                    research_probe_profile = gr.Dropdown(
                        choices=RESEARCH_PROBE_CHOICES,
                        value=None,
                        label="Research profile · never a promotion gate",
                        interactive=True,
                    )
                    research_probe_description = gr.Markdown(_research_probe_profile_description(None))
                    research_probe_run = gr.Button(
                        "Run Research Probe",
                        variant="secondary",
                        interactive=False,
                    )
                    research_probe_status = gr.Markdown("No Research Probe has run in this browser session.")
                    research_probe_files = gr.File(
                        label="Research Probe artifacts · one ZIP per request",
                        file_count="multiple",
                        interactive=False,
                    )
                    gr.Markdown(
                        "Research profiles are intentionally allowed to be aggressive, heavy, risky, or unsuccessful when the expected information value justifies it. "
                        "This initial set uses the shared Full/SFT runtime and therefore remains serial with product/Maintainer GPU work.",
                        elem_classes=["ltx-subtle"],
                    )

    # User-only duration control events avoid function-update `.change` fan-out.
    experimental_long_event = experimental_long.input(
        _duration_slider_update, [experimental_long, duration_seconds, resolution], duration_seconds,
        queue=False, show_progress="hidden",
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    resolution_event = resolution.input(
        _resolution_duration_controls, [resolution, experimental_long, duration_seconds],
        [experimental_long, duration_seconds], queue=False, show_progress="hidden",
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    custom_resolution_duration_event = _bind_custom_resolution_editor(
        resolution_component=resolution, width_input=custom_resolution_width, height_input=custom_resolution_height,
        apply_button=apply_custom_resolution_btn, status_output=custom_resolution_status, experimental_long=experimental_long,
        duration_seconds=duration_seconds, api_visibility=INTERNAL_API_VISIBILITY)

    gr.on(
        triggers=[image.change, middle_image.change, end_image.change],
        fn=_conditioning_mode_status_markdown,
        inputs=[image, middle_image, end_image],
        outputs=conditioning_mode_status,
        queue=False,
        show_progress="hidden",
        api_visibility=INTERNAL_API_VISIBILITY,
    )

    if IS_ZEROGPU:
        quota_inputs = [image, middle_image, end_image, duration_seconds, experimental_long, resolution, lora_dropdown, use_diffusion_decoder, use_auto_duration]
        preview_kwargs = dict(
            fn=_zerogpu_duration_panel, inputs=quota_inputs, outputs=zerogpu_duration_estimate,
            queue=False, show_progress="hidden", api_visibility=INTERNAL_API_VISIBILITY,
        )
        duration_seconds.release(**preview_kwargs)
        gr.on(
            triggers=[image.change, middle_image.change, end_image.change, lora_dropdown.change, use_diffusion_decoder.input, use_auto_duration.input],
            **preview_kwargs,
        )
        experimental_long_event.then(**preview_kwargs)
        resolution_event.then(**preview_kwargs)
        custom_resolution_duration_event.then(**preview_kwargs)

    if SPACE_DEBUG:
        maintainer_probe_profile.change(
            _maintainer_probe_profile_ui,
            inputs=[maintainer_probe_profile],
            outputs=[maintainer_probe_description, maintainer_probe_run],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        maintainer_probe_event = maintainer_probe_run.click(
            run_maintainer_probe,
            inputs=[maintainer_probe_profile, session_id_state],
            outputs=[maintainer_probe_status, maintainer_latest_probe_state],
            api_name="run_maintainer_probe",
            trigger_mode="multiple",
            show_progress="full", show_progress_on=maintainer_probe_status,
            concurrency_limit=1, concurrency_id=GPU_CONCURRENCY_ID,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        maintainer_probe_event.success(
            _append_probe_output,
            inputs=[maintainer_probe_files_state, maintainer_latest_probe_state],
            outputs=[maintainer_probe_files_state, maintainer_probe_files],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        maintainer_probe_event.failure(
            _refresh_latest_probe_output,
            inputs=[maintainer_probe_files_state, session_id_state],
            outputs=[maintainer_probe_files_state, maintainer_probe_files],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )

        research_probe_profile.change(
            _research_probe_profile_ui,
            inputs=[research_probe_profile],
            outputs=[research_probe_description, research_probe_run],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        research_probe_event = research_probe_run.click(
            run_research_probe,
            inputs=[research_probe_profile, session_id_state],
            outputs=[research_probe_status, research_latest_probe_state],
            api_name="run_research_probe",
            trigger_mode="multiple",
            show_progress="full", show_progress_on=research_probe_status,
            concurrency_limit=1, concurrency_id=GPU_CONCURRENCY_ID,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        research_probe_event.success(
            _append_probe_output,
            inputs=[research_probe_files_state, research_latest_probe_state],
            outputs=[research_probe_files_state, research_probe_files],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
        research_probe_event.failure(
            _refresh_latest_probe_output,
            inputs=[research_probe_files_state, session_id_state],
            outputs=[research_probe_files_state, research_probe_files],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )

    enhance_prompt_btn.click(
        enhance_prompt_separate,
        inputs=[prompt, image, middle_image, end_image, session_id_state],
        outputs=[prompt, prompt_tools_status],
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    translate_prompt_btn.click(
        translate_prompt_separate,
        inputs=[prompt, translate_manual_enabled, translate_manual_prompt, session_id_state],
        outputs=[prompt, prompt_tools_status],
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    write_prompt_btn.click(
        write_prompt_from_seed_separate,
        inputs=[prompt, write_manual_enabled, write_manual_prompt, session_id_state],
        outputs=[prompt, prompt_tools_status],
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )

    ui_lora.wire_lora_events(
        lora_ui,
        prompt=prompt,
        custom_lora_state=custom_lora_state,
        prepared_lora_state=prepared_lora_state,
        handlers=ui_lora.LoraHandlers(
            apply_hub_quick_selection=apply_hub_lora_search_selection,
            inspect_hub_repo=inspect_hub_lora_repo,
            search_hf_models=search_hf_lora_models,
            apply_hf_full_selection=apply_hf_full_search_selection,
            search_civitai_models=search_civitai_models,
            load_more_civitai_models=load_more_civitai_models,
            inspect_civitai_gallery_selection=inspect_civitai_gallery_selection,
            resolve_civitai_url=resolve_civitai_url,
            inspect_civitai_version_selection=inspect_civitai_version_selection,
            apply_civitai_trigger_words=apply_civitai_trigger_words,
            load_civitai_example_prompts=load_civitai_example_prompts,
            apply_civitai_example_prompt=apply_civitai_example_prompt,
            add_civitai_session_lora=add_civitai_session_lora,
            add_hf_session_lora=add_session_lora,
            remove_selected_session_loras=remove_selected_session_loras,
            prepare_selected_loras=prepare_selected_loras,
            invalidate_prepared_loras=_invalidate_prepared_loras,
        ),
    )
    export_settings_btn.click(
        _write_settings_export,
        inputs=[
            prompt, duration_seconds, experimental_long, resolution, seed, randomize_seed,
            lora_dropdown, lora_strength, custom_lora_state, use_diffusion_decoder, use_auto_duration, image, middle_image, end_image, session_id_state,
        ],
        outputs=[settings_download, settings_status],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    import_settings_event = import_settings_btn.click(
        _import_settings,
        inputs=[
            settings_file, import_prompt, import_generation, import_seed, import_lora, import_session_loras,
            prompt, duration_seconds, experimental_long, resolution, seed, randomize_seed,
            lora_dropdown, lora_strength, custom_lora_state, use_diffusion_decoder, use_auto_duration,
        ],
        outputs=[
            prompt, duration_seconds, experimental_long, resolution, seed, randomize_seed,
            lora_dropdown, lora_strength, custom_lora_state, use_diffusion_decoder, use_auto_duration, settings_status,
        ],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    history_restore_event = history_restore_btn.click(
        _restore_history_settings,
        inputs=[
            history_state, history_choice, prompt, duration_seconds, experimental_long, resolution,
            seed, randomize_seed, lora_dropdown, lora_strength, custom_lora_state, use_diffusion_decoder, use_auto_duration,
        ],
        outputs=[
            prompt, duration_seconds, experimental_long, resolution, seed, randomize_seed,
            lora_dropdown, lora_strength, custom_lora_state, use_diffusion_decoder, use_auto_duration, settings_status,
        ],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    _sync_resolution_editor_after(
        (import_settings_event, history_restore_event), resolution_component=resolution,
        width_input=custom_resolution_width, height_input=custom_resolution_height, status_output=custom_resolution_status,
        api_visibility=INTERNAL_API_VISIBILITY)

    if IS_ZEROGPU:
        # Programmatic restore paths normalize duration controls, then refresh quota exactly once.
        for _settings_event in (import_settings_event, history_restore_event):
            _settings_event.then(
                _resolution_duration_controls, [resolution, experimental_long, duration_seconds],
                [experimental_long, duration_seconds], queue=False, show_progress="hidden",
                api_visibility=INTERNAL_API_VISIBILITY,
            ).then(**preview_kwargs)

    a2v_prepare_event = a2v_go.click(
        prepare_audio_to_video,
        inputs=[a2v_audio, a2v_image, a2v_prompt, a2v_seed, a2v_randomize_seed, session_id_state],
        outputs=[a2v_prepared_state, a2v_status, a2v_seed],
        api_name="prepare_audio_to_video",
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    a2v_generate_event = a2v_prepare_event.success(
        generate_audio_to_video,
        inputs=[a2v_prepared_state, session_id_state],
        outputs=[a2v_out, a2v_used_seed, a2v_status, a2v_latest_probe_state],
        api_name="generate_audio_to_video",
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    a2v_generate_event.success(
        _append_probe_output,
        inputs=[a2v_probe_files_state, a2v_latest_probe_state],
        outputs=[a2v_probe_files_state, a2v_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    a2v_generate_event.failure(
        _refresh_latest_probe_output,
        inputs=[a2v_probe_files_state, session_id_state],
        outputs=[a2v_probe_files_state, a2v_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )

    ic_prepare_event = ic_go.click(
        prepare_ic_colorizer,
        inputs=[ic_video, ic_prompt, ic_profile, ic_seed, ic_randomize_seed, session_id_state],
        outputs=[ic_prepared_state, ic_status, ic_seed],
        api_name="prepare_ic_colorizer_probe",
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    ic_generate_event = ic_prepare_event.success(
        generate_ic_colorizer,
        inputs=[ic_prepared_state, session_id_state],
        outputs=[ic_out, ic_used_seed, ic_status, ic_latest_probe_state],
        api_name="generate_ic_colorizer_probe",
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    ic_generate_event.success(
        _append_probe_output,
        inputs=[ic_probe_files_state, ic_latest_probe_state],
        outputs=[ic_probe_files_state, ic_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    ic_generate_event.failure(
        _refresh_latest_probe_output,
        inputs=[ic_probe_files_state, session_id_state],
        outputs=[ic_probe_files_state, ic_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )

    pixel_upscaler_prepare_event = pixel_upscaler_go.click(
        prepare_pixel_upscaler,
        inputs=[
            pixel_upscaler_video, pixel_upscaler_prompt, pixel_upscaler_seed,
            pixel_upscaler_randomize_seed, session_id_state,
        ],
        outputs=[pixel_upscaler_prepared_state, pixel_upscaler_status, pixel_upscaler_seed],
        api_name="prepare_pixel_spatial_upscaler",
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    pixel_upscaler_generate_event = pixel_upscaler_prepare_event.success(
        generate_pixel_upscaler,
        inputs=[pixel_upscaler_prepared_state, session_id_state],
        outputs=[
            pixel_upscaler_out, pixel_upscaler_used_seed, pixel_upscaler_status,
            pixel_upscaler_latest_probe_state,
        ],
        api_name="generate_pixel_spatial_upscaler",
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    pixel_upscaler_generate_event.success(
        _append_probe_output,
        inputs=[pixel_upscaler_probe_files_state, pixel_upscaler_latest_probe_state],
        outputs=[pixel_upscaler_probe_files_state, pixel_upscaler_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    pixel_upscaler_generate_event.failure(
        _refresh_latest_probe_output,
        inputs=[pixel_upscaler_probe_files_state, session_id_state],
        outputs=[pixel_upscaler_probe_files_state, pixel_upscaler_probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )

    inoutpaint_prepare_event = inoutpaint_go.click(
        prepare_inoutpaint,
        inputs=[
            inoutpaint_video, inoutpaint_mask, inoutpaint_prompt, inoutpaint_mode,
            inoutpaint_layout, inoutpaint_dilation, inoutpaint_seed, inoutpaint_randomize_seed,
            session_id_state,
        ],
        outputs=[inoutpaint_prepared_state, inoutpaint_status, inoutpaint_seed],
        api_name="prepare_inoutpaint", api_visibility=INTERNAL_API_VISIBILITY,
    )
    inoutpaint_generate_event = inoutpaint_prepare_event.success(
        generate_inoutpaint,
        inputs=[inoutpaint_prepared_state, session_id_state],
        outputs=[inoutpaint_out, inoutpaint_used_seed, inoutpaint_status, inoutpaint_latest_probe_state],
        api_name="generate_inoutpaint", concurrency_limit=1, concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    inoutpaint_generate_event.success(
        _append_probe_output,
        inputs=[inoutpaint_probe_files_state, inoutpaint_latest_probe_state],
        outputs=[inoutpaint_probe_files_state, inoutpaint_probe_files],
        queue=False, api_visibility=INTERNAL_API_VISIBILITY,
    )
    inoutpaint_generate_event.failure(
        _refresh_latest_probe_output,
        inputs=[inoutpaint_probe_files_state, session_id_state],
        outputs=[inoutpaint_probe_files_state, inoutpaint_probe_files],
        queue=False, api_visibility=INTERNAL_API_VISIBILITY,
    )

    clear_generation_event = go.click(
        _clear_generation_result,
        inputs=[],
        outputs=[out, used_seed],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    resolution_preflight_event = clear_generation_event.then(
        _resolution_input_preflight,
        inputs=[resolution],
        outputs=[generation_preflight_status],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    lora_prepare_before_generate = resolution_preflight_event.success(
        prepare_selected_loras,
        inputs=[lora_dropdown, custom_lora_state, civitai_api_key],
        outputs=[prepared_lora_state, lora_prepare_status],
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    generation_ready_event = lora_prepare_before_generate
    if IS_ZEROGPU:
        generation_ready_event = lora_prepare_before_generate.success(
            _generation_duration_preflight,
            inputs=[image, middle_image, end_image, duration_seconds, resolution, lora_dropdown, use_diffusion_decoder, use_auto_duration],
            outputs=[generation_preflight_status],
            queue=False,
            api_visibility=INTERNAL_API_VISIBILITY,
        )
    generation_event = generation_ready_event.success(
        generate,
        inputs=[
            prompt,
            image,
            middle_image,
            end_image,
            duration_seconds,
            experimental_long,
            resolution,
            seed,
            randomize_seed,
            lora_dropdown,
            lora_strength,
            custom_lora_state,
            prepared_lora_state,
            use_diffusion_decoder,
            use_auto_duration,
            session_id_state,
            history_state,
        ],
        outputs=[
            out, seed, used_seed, latest_probe_state,
            history_state, history_choice, history_summary,
        ],
        concurrency_limit=1,
        concurrency_id=GPU_CONCURRENCY_ID,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    generation_event.success(
        _append_probe_output,
        inputs=[probe_files_state, latest_probe_state],
        outputs=[probe_files_state, probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )
    generation_event.failure(
        _refresh_latest_probe_output,
        inputs=[probe_files_state, session_id_state],
        outputs=[probe_files_state, probe_files],
        queue=False,
        api_visibility=INTERNAL_API_VISIBILITY,
    )

demo.queue(default_concurrency_limit=1).launch(css=CSS, ssr_mode=False)