"""Workflow Studio: local conditioning and generation in one Space and API call."""
from __future__ import annotations

import base64
import faulthandler
import gc
import hashlib
import json
import os
import shutil
import signal
import sys
import threading
import time
import uuid
from pathlib import Path

# Import before anything that can initialize CUDA.
try:
    import spaces
except ImportError:
    spaces = None

import gradio as gr
from canvas_bridge import (get_oauth_available, get_oauth_scopes, get_token, make_api_fn, make_call_fn,
                           make_media_fn, make_palette_fn)
import media_tools
import worker_probe
from workflow_templates import function_palette, list_templates, list_workflow_templates
from preset_core import list_presets, plan_reference_request, plan_request
from studio_service import (initialization_blockers, reservation_seconds,
                            conditioning_reservation_seconds, validate_local_conditioning,
                            conditioning_text_rows_bound, workflow_reservation_seconds,
                            reference_conditioning_seconds, reference_generation_seconds,
                            reference_rows, reference_text_rows_bound, reference_workflow_seconds,
                            COLD_DOWNLOAD_ALLOWANCE_S, REFERENCE_IMAGE_SHORT_EDGE)

MODEL_REPO = "MiniMaxAI/MiniMax-H3"
MODEL_REVISION = "42ed227ee7df40d41602854ae760620d6eb651fe"
ADAPTER_REVISIONS = {
    "larryvrh/MiniMax-H3-Turbo-Lora": "43a74557ac3f6539db8e0f2a959d03feb7a81480",
    "lightx2v/Minimax-h3-Turbo": "30f485ed874aa44c8183348ad361c5eabad318da",
}
OUTPUT_DIR = Path(os.environ.get("WORKFLOW_OUTPUT_DIR", "outputs")).resolve()
MODEL_CACHE_DIR = Path(os.environ.get("WORKFLOW_MODEL_CACHE", ".workflow-models")).resolve()
MODEL_STORE = None
LOAD_ERROR = None
RUN_LOCK = threading.Lock()


def blockers() -> list[str]:
    result = initialization_blockers(os.environ)
    if not os.environ.get("SPACE_ID") or spaces is None:
        result.append("hosted_zerogpu_required")
    if LOAD_ERROR:
        result.append(LOAD_ERROR)
    return result


def health() -> dict:
    """Readiness is not an inference or quality claim."""
    reasons = blockers()
    return {
        "schema_version": 2, "status": "ready" if MODEL_STORE is not None and not reasons else "not_ready",
        "runtime_initialized": MODEL_STORE is not None,
        "model_loaded": False, "model_loading": "staged_local", "blockers": reasons,
        "execution": "eager", "compiled": False, "inference_verified": False,
        "model_repo": MODEL_REPO, "model_revision": MODEL_REVISION,
        "conditioning_transport": "in_process",
        "workflow_components": ["qwen3_vl_conditioning", "h3_fl2va_generation"],
        "busy": RUN_LOCK.locked(),
    }


def catalog() -> dict:
    return {"schema_version": 2, "presets": list_presets(), "runtime": health()}


def plan(preset_id: str, prompt: str, canvas: str = "", duration: float = 5,
         seed: int = 42, has_first: bool = False, has_last: bool = False) -> dict:
    try:
        return plan_request(preset_id, prompt, canvas, duration, seed, has_first, has_last)
    except ValueError as error:
        # Planner errors are fixed validation messages, safe to expose without
        # enabling verbose provider/runtime exception reporting.
        raise gr.Error(str(error)) from None


def initialize() -> None:
    global MODEL_STORE, LOAD_ERROR
    if blockers():
        return
    try:
        # Import actual runtime dependencies, but never download or pack weights
        # during startup. Each stage owns its own bounded, temporary checkpoint.
        from h3_split_blocks import MiniMaxH3ConditionerBlocks, MiniMaxH3GeneratorBlocks
        from local_runtime import StageStore
        os.environ.setdefault("HF_XET_CHUNK_CACHE_SIZE_BYTES", "0")
        MODEL_STORE = StageStore(MODEL_CACHE_DIR, MODEL_REPO, MODEL_REVISION)
    except Exception as error:
        LOAD_ERROR = "runtime_init_failed_" + type(error).__name__
        print(LOAD_ERROR, flush=True)


# ZeroGPU reuses an idle worker process for the next call of the same GPU
# function. Every GPU job ends its worker shortly after returning, and the
# parent holds RUN_LOCK until then, so each job forks a fresh worker with no
# pooled connections, model references or staged files left by an earlier job.
_PARENT_PID = os.getpid()
_WORKER_RETIRE_DELAY_S = 1.0
# Only hosted ZeroGPU forks workers; elsewhere nothing needs to wait or watch.
_HOSTED_ZEROGPU = spaces is not None and bool(os.environ.get("SPACE_ID"))
_WORKER_RETIRE_HOLD_S = 2.5 if _HOSTED_ZEROGPU else 0
# While a GPU call is pending the parent logs its workers' /proc state and
# SIGUSR1-dumps a stalled one (worker_probe).
_WORKER_WATCH_S = 45 if _HOSTED_ZEROGPU else 0


def _suppress_core_dumps():
    """A crashed worker otherwise spends its GPU allocation writing a core dump.

    A shot-2 worker crashed seconds into a job and sat in vfs_coredump for ~300 s
    until ZeroGPU aborted it (2026-10-23). Limit 1, not 0: the kernel also skips
    piped core handlers (core_pattern "|...") at exactly 1.
    """
    import resource
    _, hard = resource.getrlimit(resource.RLIMIT_CORE)
    limit = 1 if hard == resource.RLIM_INFINITY or hard >= 1 else 0
    resource.setrlimit(resource.RLIMIT_CORE, (limit, hard))
    return limit


if _HOSTED_ZEROGPU:
    # Forked workers inherit all three: a fatal signal prints every Python
    # thread's stack, SIGUSR1 does the same on demand, and no core is written.
    faulthandler.enable(file=sys.stderr, all_threads=True)
    faulthandler.register(signal.SIGUSR1, file=sys.stderr, all_threads=True)
    print(f"core_dump_limit={_suppress_core_dumps()}", flush=True)


def _retire_worker():
    """In a forked ZeroGPU worker, exit just after the result is handed back."""
    if os.getpid() == _PARENT_PID:
        return

    def exit_soon():
        time.sleep(_WORKER_RETIRE_DELAY_S)
        os._exit(0)
    threading.Thread(target=exit_soon, daemon=True).start()


def _sleep(seconds):
    time.sleep(seconds)


def _workflow_reservation(plan_data, image_path, last_image_path):
    return workflow_reservation_seconds(plan_data, bool(image_path), bool(last_image_path))


def _encode_gpu(plan_data, image_path, last_image_path, model_path):
    from local_runtime import encode_local
    return encode_local(plan_data, image_path, last_image_path, model_path)


def _execute_gpu(plan_data, embeds, tags, image_path, last_image_path, model_path, adapter_path):
    from local_runtime import generate_local
    return generate_local(plan_data, embeds, tags, image_path, last_image_path, model_path, adapter_path)


def _workflow_gpu(plan_data, image_path, last_image_path):
    """One allocation owns both downloads and model stages, never the parent lock."""
    try:
        return _workflow_stages(plan_data, image_path, last_image_path)
    finally:
        _retire_worker()


def _workflow_stages(plan_data, image_path, last_image_path):
    started = time.monotonic()
    has_first, has_last = bool(image_path), bool(last_image_path)
    rows_bound = conditioning_text_rows_bound(plan_data, has_first, has_last)
    metrics = {
        "gpu_allocations": 1, "downloads_inside_gpu_allocation": True,
        "cold_download_allowance_s": COLD_DOWNLOAD_ALLOWANCE_S,
        "conditioning_reservation_s": conditioning_reservation_seconds(plan_data, has_first, has_last),
        "generation_reservation_s": reservation_seconds(plan_data, rows_bound),
    }
    print("workflow_stage=conditioning_download", flush=True)
    with MODEL_STORE.stage("conditioner") as model_path:
        downloaded = time.monotonic()
        metrics["conditioning_download_s"] = round(downloaded - started, 3)
        print("workflow_stage=conditioning_gpu", flush=True)
        embeds, tags, height, width, num_frames = _encode_gpu(
            plan_data, image_path, last_image_path, str(model_path))
        metrics["conditioning_gpu_s"] = round(time.monotonic() - downloaded, 3)
        validate_local_conditioning(plan_data, embeds, tags, height, width, num_frames)
    metrics["conditioning_elapsed_s"] = round(time.monotonic() - started, 3)
    if int(tags.numel()) > rows_bound:
        raise ValueError("local conditioning exceeds the preallocated text-row bound")
    metrics["generation_actual_reservation_s"] = reservation_seconds(plan_data, int(tags.numel()))

    generation_started = time.monotonic()
    print("workflow_stage=generator_download", flush=True)
    with MODEL_STORE.stage("generator") as model_path:
        downloaded = time.monotonic()
        metrics["generator_download_s"] = round(downloaded - generation_started, 3)
        adapter_path = None
        if plan_data["adapter"] != "off":
            from huggingface_hub import hf_hub_download
            from h3_adapter_spec import PROFILES
            profile = PROFILES[plan_data["adapter"]]
            adapter_path = hf_hub_download(
                profile.repo, profile.file, revision=ADAPTER_REVISIONS[profile.repo],
                token=False, local_dir=Path(model_path) / "adapters")
        execution_started = time.monotonic()
        metrics["adapter_download_s"] = round(execution_started - downloaded, 3)
        print("workflow_stage=generator_gpu", flush=True)
        frames, audio, sr = _execute_gpu(plan_data, embeds, tags, image_path,
                                       last_image_path, str(model_path), adapter_path)
        metrics["generation_gpu_s"] = round(time.monotonic() - execution_started, 3)
    metrics["generation_elapsed_s"] = round(time.monotonic() - generation_started, 3)
    # GPU-body elapsed excludes scheduler/worker IPC overhead, and is not billing.
    metrics["gpu_elapsed_s"] = round(time.monotonic() - started, 3)
    return frames, audio, sr, metrics


def _reference_reservation(plan_data, references, metas):
    return reference_workflow_seconds(plan_data, metas)


def _encode_reference_gpu(plan_data, references, model_path):
    from local_runtime import encode_local_ref2va
    return encode_local_ref2va(plan_data, references, model_path, REFERENCE_IMAGE_SHORT_EDGE)


def _execute_reference_gpu(plan_data, embeds, tags, references, model_path):
    from local_runtime import generate_local_ref2va
    return generate_local_ref2va(plan_data, embeds, tags, references, model_path, REFERENCE_IMAGE_SHORT_EDGE)


def _reference_gpu(plan_data, references, metas):
    """Ref2VA in one allocation: conditioner stage, then the transformer_ref stage."""
    try:
        return _reference_stages(plan_data, references, metas)
    finally:
        _retire_worker()


def _reference_stages(plan_data, references, metas):
    started = time.monotonic()
    rows_bound = reference_text_rows_bound(plan_data, metas)
    ref_rows = reference_rows(metas, plan_data["num_frames"])
    metrics = {
        "gpu_allocations": 1, "downloads_inside_gpu_allocation": True,
        "cold_download_allowance_s": COLD_DOWNLOAD_ALLOWANCE_S, "reference_rows": ref_rows,
        "conditioning_reservation_s": reference_conditioning_seconds(plan_data),
        "generation_reservation_s": reference_generation_seconds(plan_data, rows_bound, ref_rows),
    }
    print("workflow_stage=conditioning_download", flush=True)
    with MODEL_STORE.stage("conditioner") as model_path:
        downloaded = time.monotonic()
        metrics["conditioning_download_s"] = round(downloaded - started, 3)
        print("workflow_stage=conditioning_gpu", flush=True)
        embeds, tags, height, width, num_frames = _encode_reference_gpu(plan_data, references, str(model_path))
        metrics["conditioning_gpu_s"] = round(time.monotonic() - downloaded, 3)
        validate_local_conditioning(plan_data, embeds, tags, height, width, num_frames)
    metrics["conditioning_elapsed_s"] = round(time.monotonic() - started, 3)
    # Calibration evidence: the real conditioning rows against the reserved bound.
    metrics["text_rows"] = int(tags.numel())
    print(f"reference_rows text={metrics['text_rows']} bound={rows_bound} reference={ref_rows}", flush=True)
    if metrics["text_rows"] > rows_bound:
        raise ValueError("reference conditioning exceeds the preallocated text-row bound; "
                         "use fewer or shorter references")

    generation_started = time.monotonic()
    print("workflow_stage=generator_download", flush=True)
    with MODEL_STORE.stage("generator_ref") as model_path:
        downloaded = time.monotonic()
        metrics["generator_download_s"] = round(downloaded - generation_started, 3)
        print("workflow_stage=generator_gpu", flush=True)
        frames, audio, sr = _execute_reference_gpu(plan_data, embeds, tags, references, str(model_path))
        metrics["generation_gpu_s"] = round(time.monotonic() - downloaded, 3)
    metrics["generation_elapsed_s"] = round(time.monotonic() - generation_started, 3)
    metrics["gpu_elapsed_s"] = round(time.monotonic() - started, 3)
    return frames, audio, sr, metrics


def _encode_output(frames, audio, sr, target):
    from diffusers.utils import encode_video
    encode_video(frames, fps=24, output_path=str(target), audio=audio, audio_sample_rate=sr)


if spaces is not None and os.environ.get("SPACE_ID"):
    _workflow_gpu = spaces.GPU(duration=_workflow_reservation, size="xlarge")(_workflow_gpu)
    _reference_gpu = spaces.GPU(duration=_reference_reservation, size="xlarge")(_reference_gpu)


# ZeroGPU schedules with the caller's X-IP-Token. A page left open for hours, or
# a stale client, carries an expired one that the scheduler rejects with a 401.
_IDENTITY_MIN_TTL_S = 3
_IDENTITY_EXPIRED = ("zerogpu_session_expired; reload the page (or reconnect your API client) "
                     "and run again. Nothing was submitted.")


def _identity_ttl():
    """Seconds left on the caller's ZeroGPU identity token, or None without one."""
    from gradio.context import LocalContext
    request = LocalContext.request.get(None)
    token = (getattr(request, "headers", None) or {}).get("x-ip-token")
    if not token:
        return None
    try:
        payload = token.split(".")[1]
        claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
        return float(claims["exp"]) - time.time()
    except Exception:
        return None


def _admit():
    """Readiness and caller identity, before the run lock or any allocation."""
    if MODEL_STORE is None or blockers():
        raise gr.Error("runtime_not_ready: " + ", ".join(blockers() or ["runtime_not_initialized"]))
    ttl = _identity_ttl()
    print(f"zerogpu_identity_ttl_s={'none' if ttl is None else round(ttl)}", flush=True)
    if ttl is not None and ttl < _IDENTITY_MIN_TTL_S:
        raise gr.Error(_IDENTITY_EXPIRED)


def _run_locked(plan_data, reserve, run_gpu):
    """Serialize one GPU workflow, encode its MP4 on CPU, and report it."""
    if not RUN_LOCK.acquire(blocking=False):
        raise gr.Error("runtime_busy; another job is running and nothing was submitted; "
                       "retry after it finishes")
    gpu_finished = None
    try:
        OUTPUT_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
        if shutil.disk_usage(OUTPUT_DIR).free < 3 * 1024**3:
            raise gr.Error("insufficient_output_storage")
        # Retain only our own expired outputs. Active output paths never preexist.
        for old in OUTPUT_DIR.glob("workflow-*.mp4"):
            if old.is_file() and not old.is_symlink() and time.time() - old.stat().st_mtime > 7200:
                old.unlink()
        started = time.monotonic()
        # Admission precedes all downloads. Schedule once while the original
        # request's proxy token is fresh; cold downloads consume GPU allocation.
        reservation = reserve()
        watch_stop = threading.Event()
        if _WORKER_WATCH_S:
            threading.Thread(target=worker_probe.watch, args=(watch_stop, _WORKER_WATCH_S),
                             name="worker-probe", daemon=True).start()
        # The worker is forked inside run_gpu() and must never garbage-collect
        # parent objects: a shot-2 worker segfaulted in its first collection,
        # freeing garbage the parent still held from the previous job (2026-10-23).
        # Collect here, then freeze what is left out of every later collection.
        gc.collect()
        gc.freeze()
        try:
            frames, audio, sr, metrics = run_gpu()
        finally:
            gc.unfreeze()
            watch_stop.set()
            gpu_finished = time.monotonic()
        print("workflow_stage=output_encoding", flush=True)
        encoding_started = time.monotonic()
        target = OUTPUT_DIR / f"workflow-{uuid.uuid4().hex}.mp4"
        _encode_output(frames, audio, sr, target)
        metrics["output_encoding_s"] = round(time.monotonic() - encoding_started, 3)
        target.chmod(0o600)
        with target.open("rb") as handle:
            digest = hashlib.file_digest(handle, "sha256").hexdigest()
        report = {
            **plan_data, "schema_version": 2, "status": "generated_unreviewed", "model_revision": MODEL_REVISION,
            "conditioning_transport": "in_process", "model_loading": "staged_local",
            **metrics,
            "elapsed_s": round(time.monotonic() - started, 3), "reservation_s": reservation,
            "output_sha256": digest, "output_bytes": target.stat().st_size,
            "compiled": False, "quality_reviewed": False,
        }
        return str(target), report
    except gr.Error as error:
        # gr.Error subclasses ValueError, so it must be handled first.
        if "Expired ZeroGPU proxy token" in str(error.message):
            raise gr.Error(_IDENTITY_EXPIRED) from None
        raise
    except ValueError as error:
        # Validation messages contain no credentials; remote/provider exceptions
        # are deliberately not copied into the public API response.
        raise gr.Error(str(error)) from None
    except Exception as error:
        print("generation_error_" + type(error).__name__, flush=True)
        raise gr.Error("generation_outcome_unknown; reconcile before retrying") from None
    finally:
        # Let the retiring worker exit before another job can be scheduled.
        if gpu_finished is not None and _WORKER_RETIRE_HOLD_S:
            remaining = _WORKER_RETIRE_HOLD_S - (time.monotonic() - gpu_finished)
            if remaining > 0:
                _sleep(remaining)
        RUN_LOCK.release()


def generate(preset_id, prompt, image_path=None, last_image_path=None, canvas="", duration=5,
             seed=42):
    plan_data = plan(preset_id, prompt, canvas, duration, seed, bool(image_path), bool(last_image_path))
    _admit()
    return _run_locked(plan_data, lambda: _workflow_reservation(plan_data, image_path, last_image_path),
                       lambda: _workflow_gpu(plan_data, image_path, last_image_path))


REFERENCE_VIDEO_SECONDS = (2.0, 15.0)


def _media_seconds(stream, container):
    import av
    if stream.duration is not None and stream.time_base is not None:
        return float(stream.duration * stream.time_base)
    return None if container.duration is None else container.duration / av.time_base


def _probe_reference(kind, path):
    """Metadata the Ref2VA budget needs; no decoding of pixels or samples."""
    if kind == "image":
        from PIL import Image
        with Image.open(path) as image:
            width, height = image.size
        return {"kind": "image", "width": width, "height": height}
    import av
    with av.open(path) as container:
        video = container.streams.video[0] if container.streams.video else None
        audio = container.streams.audio[0] if container.streams.audio else None
        audio_seconds = _media_seconds(audio, container) if audio else None
        if kind == "audio":
            if not audio_seconds:
                raise ValueError("the reference audio has no audio stream")
            return {"kind": "audio", "seconds": audio_seconds}
        if video is None:
            raise ValueError("the reference video has no video stream")
        seconds = _media_seconds(video, container) or 0.0
        low, high = REFERENCE_VIDEO_SECONDS
        if not low <= seconds <= high:
            raise ValueError(f"the reference video must be {low:g}-{high:g} seconds long")
        return {"kind": "video", "width": video.width, "height": video.height,
                "seconds": seconds, "audio_seconds": audio_seconds}


def reference_to_video(prompt, image_1=None, image_2=None, image_3=None, audio=None, video=None,
                       canvas="", duration=5, seed=42):
    """Ref2VA: images, one soundtrack and one video read in that order, then the prompt."""
    references = [("image", path) for path in (image_1, image_2, image_3) if path]
    references += [(kind, path) for kind, path in (("audio", audio), ("video", video)) if path]
    try:
        plan_data = plan_reference_request(prompt, [kind for kind, _ in references], canvas, duration, seed)
    except ValueError as error:
        raise gr.Error(str(error)) from None
    try:
        metas = [_probe_reference(kind, path) for kind, path in references]
    except ValueError as error:
        raise gr.Error(str(error)) from None
    except Exception:
        raise gr.Error("a reference file could not be read") from None
    _admit()
    return _run_locked(plan_data, lambda: _reference_reservation(plan_data, references, metas),
                       lambda: _reference_gpu(plan_data, references, metas))


# Card backgrounds for the canvas Templates modal, one per model family.
_CARD_GRADIENTS = {
    "base": ("#2dd4bf", "radial-gradient(ellipse at 15% 85%, #2dd4bf 0%, #0d9488 40%, transparent 70%), "
             "radial-gradient(ellipse at 80% 10%, #38bdf8 0%, #0284c7 45%, transparent 70%), #0f766e"),
    "keyframes": ("#a3e635", "radial-gradient(ellipse at 20% 20%, #a3e635 0%, #65a30d 40%, transparent 70%), "
                  "radial-gradient(ellipse at 85% 80%, #34d399 0%, #059669 45%, transparent 70%), #166534"),
    "larry": ("#c084fc", "radial-gradient(ellipse at 15% 80%, #c084fc 0%, #7c3aed 40%, transparent 70%), "
              "radial-gradient(ellipse at 85% 15%, #f472b6 0%, #db2777 45%, transparent 70%), #4c1d95"),
    "lightx": ("#fb923c", "radial-gradient(ellipse at 80% 85%, #fbbf24 0%, #d97706 40%, transparent 70%), "
               "radial-gradient(ellipse at 15% 15%, #fb923c 0%, #ea580c 45%, transparent 70%), #7c2d12"),
    "ref2va": ("#f472b6", "radial-gradient(ellipse at 20% 80%, #f472b6 0%, #be185d 40%, transparent 70%), "
               "radial-gradient(ellipse at 80% 15%, #fb7185 0%, #e11d48 45%, transparent 70%), #831843"),
    "extend": ("#818cf8", "radial-gradient(ellipse at 15% 20%, #818cf8 0%, #4f46e5 40%, transparent 70%), "
               "radial-gradient(ellipse at 85% 85%, #22d3ee 0%, #0891b2 45%, transparent 70%), #312e81"),
}
_WORKFLOW_CARD_TEXT = {
    "ref2va-images": ("Reference to video · 2 images", "Ref2VA · images then prompt · budget checked on Run"),
    "extend-clip": ("Extend a clip · 2 shots", "Text → last frame → image to video → join · 2 GPU runs"),
}
_CARD_MODES = {"t2va": "Text to video", "i2va": "First frame to video", "fl2va": "First + last frame to video"}


def template_cards() -> list[dict]:
    """Templates-modal cards: each preset's local H3 workflow with its admission budget,
    then the Ref2VA and multi-node workflows."""
    presets = {preset["id"]: preset for preset in list_presets()}
    cards = []
    for entry in list_templates():
        preset, mode = presets[entry["preset_id"]], entry["mode"]
        family = ("keyframes" if mode != "t2va" else "base" if preset["adapter"] == "off"
                  else preset["adapter"].split("-")[0])
        accent, gradient = _CARD_GRADIENTS[family]
        admission = entry["admission"]
        budget = (f"{admission['reservation_seconds']} s GPU reserved" if admission["allowed"]
                  else "Blocked at defaults")
        name = preset["label"] if mode == "t2va" else f"{preset['label']} · {_CARD_MODES[mode].split(' to ')[0]}"
        cards.append({
            "id": entry["id"], "name": name, "category": _CARD_MODES[mode],
            "description": f"{_CARD_MODES[mode]} · {preset['default_canvas'].split(' · ')[0]} · {budget}",
            "accent": accent, "gradient": gradient, "workflow": entry["graph"],
        })
    for entry in list_workflow_templates():
        accent, gradient = _CARD_GRADIENTS[entry["mode"]]
        name, description = _WORKFLOW_CARD_TEXT[entry["id"]]
        cards.append({"id": entry["id"], "name": name, "category": "Custom workflow",
                      "description": description, "accent": accent, "gradient": gradient,
                      "workflow": entry["graph"]})
    return cards


initialize()
CARDS = template_cards()
# Gradio appends launch(js=...) verbatim as a <script>, so call the function.
CANVAS_JS = (f"({Path(__file__).with_name('canvas_launch.js').read_text().strip()})"
             f"({json.dumps(CARDS).replace('</', '<' + chr(92) + '/')});")
with gr.Blocks(title="Workflow Studio", analytics_enabled=False, fill_width=True, fill_height=True) as demo:
    # Build the stable API first so the canvas invokes its exact preprocessing,
    # execution context and output registration, without remote inference nodes.
    gr.api(catalog, api_name="presets", queue=False)
    gr.api(plan, api_name="plan", queue=False)
    first = gr.Image(label="First frame", type="filepath", visible=False)
    last = gr.Image(label="Last frame", type="filepath", visible=False)
    video = gr.Video(label="Generated output", visible=False)
    result = gr.JSON(label="Run metadata", visible=False)
    # Preserve raw API scalar types. UI Textbox/Number preprocessing otherwise
    # coerces numeric prompts and rounds fractional or boolean seeds. Keep normal
    # Gradio file preprocessing so upload/cache path protections remain enabled.
    api_defaults = {"preset_id": "stable", "canvas": "", "duration": 5, "seed": 42}
    # JSON components parse string values as JSON; encode scalar defaults once
    # so the public schema retains the same optional arguments as the UI route.
    api_scalars = {name: gr.JSON(value=json.dumps(api_defaults.get(name)), label=name,
                                 visible=False) for name in (
        "preset_id", "prompt", "canvas", "duration", "seed",
    )}
    api_run = gr.Button(visible=False)
    api_run.click(generate, [api_scalars["preset_id"], api_scalars["prompt"], first, last,
                            api_scalars["canvas"], api_scalars["duration"], api_scalars["seed"]],
                  [video, result], api_name="generate", concurrency_limit=None)
    reference_images = [gr.Image(label=f"Reference image {n}", type="filepath", visible=False)
                        for n in (1, 2, 3)]
    reference_audio = gr.Audio(label="Reference audio", type="filepath", visible=False)
    reference_video = gr.Video(label="Reference video", visible=False)
    reference_defaults = {"canvas": "", "duration": 5, "seed": 42}
    reference_scalars = {name: gr.JSON(value=json.dumps(reference_defaults.get(name)), label=name,
                                       visible=False) for name in ("prompt", "canvas", "duration", "seed")}
    gr.Button(visible=False).click(
        reference_to_video,
        [reference_scalars["prompt"], *reference_images, reference_audio, reference_video,
         reference_scalars["canvas"], reference_scalars["duration"], reference_scalars["seed"]],
        [video, result], api_name="reference_to_video", concurrency_limit=None)
    # The canvas has its own sign-in button; this hidden one makes Gradio mount
    # the /login/huggingface OAuth routes it uses.
    gr.LoginButton(visible=False)
    runtime = gr.JSON(visible=False)
    gr.Button(visible=False).click(health, outputs=runtime, api_name="health", queue=False)
    def endpoint_index(api_name):
        return next(i for i, fn in demo.fns.items() if fn.api_name == api_name)

    def queued_node(name, run):
        async def node(args_json: str, request: gr.Request) -> str:
            return await run([name, args_json], request)
        node.__name__ = f"predict_fn_{name}"
        return node

    # Gradio 6.28 canvases submit fn nodes to /predict_fn_<name> through the
    # queue, which carries the viewer's ZeroGPU identity. No route waits behind a
    # GPU job (a wait can outlive that identity); RUN_LOCK answers "busy".
    gpu_nodes = {
        "generate_video": make_call_fn(demo, endpoint_index("generate")),
        "reference_to_video": make_call_fn(demo, endpoint_index("reference_to_video"),
                                           "reference_to_video", 9, (1, 2, 3, 4, 5)),
    }
    for name, run in gpu_nodes.items():
        gr.api(queued_node(name, run), api_name=f"predict_fn_{name}",
               api_visibility="undocumented", concurrency_limit=None)
    for name, tool, arity, mime_type in (("last_frame", media_tools.last_frame, 1, "image/png"),
                                         ("first_frame", media_tools.first_frame, 1, "image/png"),
                                         ("concat_videos", media_tools.concat_videos, 2, "video/mp4")):
        gr.api(make_media_fn(tool, arity, mime_type), api_name=f"predict_fn_{name}",
               api_visibility="undocumented", concurrency_limit=2)
    # Full-page canvas, as in gr.Workflow; the Templates modal lists CARDS.
    workflow_canvas = gr.WorkflowCanvas(
        value=json.dumps(CARDS[0]["workflow"]), elem_id="workflow-canvas",
        server_functions=[get_token, get_oauth_available, get_oauth_scopes,
                          make_palette_fn(function_palette()), make_api_fn(demo, [
            ("/generate", "Generate video"), ("/reference_to_video", "Reference to video"),
            ("/plan", "Plan a request"),
            ("/presets", "Presets and readiness"), ("/health", "Runtime health")])])

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1, max_size=4).launch(
        server_name=os.environ.get("GRADIO_SERVER_NAME", "127.0.0.1"),
        server_port=int(os.environ.get("GRADIO_SERVER_PORT", "7860")),
        show_error=False, max_file_size="1gb", footer_links=["api"],
        js=CANVAS_JS,
        css=Path(__file__).with_name("canvas.css").read_text(),
        blocked_paths=[str(Path.home() / ".cache"), str(MODEL_CACHE_DIR),
                       str(Path(__file__).resolve().parent / "tests")],
    )