Add custom video job API

README.md

@@ -7,6 +7,120 @@ sdk: gradio
 sdk_version: 6.0.1
 app_file: app.py
 pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+
+# Wan 2.2 I2V 14B Lightning · ZeroGPU
+
+本 Space 保留 Gradio UI、MCP、`/generate_video` 与 `/extract_frame`，并额外提供一个不依赖 Gradio SSE 的异步视频任务 API。任务状态和幂等索引保存在当前进程内，Space 重启后不会保留；完成结果默认保存 1800 秒。
+
+## Space 配置
+
+必需 Secret：
+
+- `HF_TOKEN`：仅用于读取模型或私有 LoRA，不会替代调用方的 Hugging Face 身份。
+- `JOB_API_KEY`：至少 32 字符的随机共享密钥，仅通过 `X-API-Key` 传入，禁止提交到仓库。
+
+必需 Variable：
+
+- `SPACE_HOST`：公开 Space 根地址，例如 `https://idlecloudx-i2v-1.hf.space`。
+- `JOB_IMAGE_ALLOWED_HOSTS`：允许抓取首图和尾图的精确主机名，以逗号分隔，不支持通配符、协议、端口或路径。
+
+可选 Variable：
+
+- `JOB_RESULT_TTL_SECONDS`：终态结果保留秒数，默认 `1800`。
+- `JOB_POLL_AFTER_SECONDS`：建议轮询间隔，默认 `2`。
+- `JOB_IMAGE_FETCH_TIMEOUT_SECONDS`：单图连接与总抓取期限，默认且最大 `15`。
+- `JOB_IMAGE_MAX_BYTES`：单图字节上限，默认且最大 `20971520`（20 MiB）。
+- `JOB_IMAGE_MAX_PIXELS`：单图像素上限，默认且最大 `40000000`。
+
+修改 Secret 或 Variable 会触发 Space 重启。
+
+## 自定义视频任务 API
+
+所有任务管理请求都需要 `X-API-Key`。`Idempotency-Key` 可省略，正式调用方应提供不超过 200 字符的稳定唯一值。相同键与相同请求体只会生成一次；相同键配合不同请求体返回 `409`。
+
+创建任务：
+
+```bash
+curl -i -X POST "https://idlecloudx-i2v-1.hf.space/api/jobs" \
+  -H "X-API-Key: your-shared-random-key" \
+  -H "Idempotency-Key: caller-generated-unique-value" \
+  -H "Content-Type: application/json" \
+  --data '{
+    "input_image_url": "https://allowed.example/input.png",
+    "last_image_url": null,
+    "prompt": "make this image come alive, cinematic motion, smooth animation",
+    "steps": 6,
+    "negative_prompt": "",
+    "duration_seconds": 3.5,
+    "guidance_scale": 1,
+    "guidance_scale_2": 1,
+    "seed": 42,
+    "randomize_seed": false,
+    "quality": 6,
+    "scheduler": "UniPCMultistep",
+    "flow_shift": 3.0,
+    "frame_multiplier": 16,
+    "safe_mode": true,
+    "lora_groups": []
+  }'
+```
+
+`input_image_url` 必须是白名单精确主机上的 HTTPS URL；`last_image_url` 可空。抓取禁止重定向、用户信息、非默认端口和非公网解析地址，不会向图片主机转发 `Authorization`、Cookie、API key 或 ZeroGPU 请求头。每张图片仅接受实际 PNG/JPEG/WebP。`lora_groups` 中每一项都必须与当前 UI 下拉框显示名称完全一致。
+
+字段边界：
+
+| 字段 | 默认值 | 边界 |
+| --- | --- | --- |
+| `prompt` | 当前 UI 默认文案 | 非空，最长 5000 |
+| `negative_prompt` | 当前 UI 默认文案 | 最长 5000，可空 |
+| `steps` | `6` | 1–30 |
+| `duration_seconds` | `3.5` | 0.5–20.1 |
+| `guidance_scale`, `guidance_scale_2` | `1` | 0–10 |
+| `seed` | `42` | 0–2147483647 |
+| `randomize_seed` | `true` | 布尔值 |
+| `quality` | `6` | 1–10 |
+| `scheduler` | `UniPCMultistep` | UI 当前 7 个调度器之一 |
+| `flow_shift` | `3.0` | 0.5–15 |
+| `frame_multiplier` | `16` | 仅 16/32/64/128 |
+| `safe_mode` | `true` | 布尔值 |
+
+成功创建返回 `202`，并同时设置 `Location` 与 `Cache-Control: no-store`：
+
+```json
+{"job_id":"...","status_url":"https://.../api/jobs/...","poll_after_seconds":2}
+```
+
+使用同一个 API key 轮询 `GET /api/jobs/{job_id}`。排队或运行中返回 `202`、`Retry-After` 和 `{"status":"queued"}` / `{"status":"running"}`。完成后返回：
+
+```json
+{"video_url":"https://.../api/jobs/.../video?token=...","used_seed":42}
+```
+
+`video_url` 自带短期随机凭证，不再需要 API key。它支持 `GET`、`HEAD` 和单区间 `Range`：
+
+```bash
+curl -I "https://.../api/jobs/.../video?token=..."
+curl -H "Range: bytes=0-1023" "https://.../api/jobs/.../video?token=..." -o first-kib.bin
+```
+
+视频返回 `video/mp4`，有效单区间为 `206`；多区间或不可满足范围为 `416`。响应允许跨域播放，缓存策略为 `private, max-age=300`。
+
+## 身份、配额与并发
+
+- 不带 `Authorization` 时，Hugging Face 将请求计入匿名共享 ZeroGPU 配额。
+- 带 `Authorization: Bearer hf_<调用方令牌>` 时，由 Hugging Face 将生成计入调用方自己的 ZeroGPU 配额；应用仅保留代理注入的短效身份信息到任务启动，不保存或转发 Bearer 令牌。
+- 内部模型下载使用的 `HF_TOKEN` 永远不会充当调用方身份。
+- UI 和自定义 API 共用一个非阻塞推理槽位。已有任务运行时，新自定义提交立即返回 `503` 与 `Retry-After: 5`，不会在 Space 内排队。
+
+## 错误约定
+
+- `400`：URL、��片内容或 `Idempotency-Key` 不合法。
+- `401`：API key 或视频下载 token 缺失/错误。
+- `404`：任务不存在或结果 TTL 已过期。
+- `409`：幂等键冲突，或视频结果尚未就绪。
+- `410`：任务元数据仍在但 MP4 已丢失。
+- `422`：JSON Schema 校验失败，包括未知字段。
+- `500`：生成失败；仅返回脱敏的 `GENERATION_FAILED`。
+- `502` / `504`：图片源临时故障或抓取超时。
+- `503`：推理槽位繁忙，或 Hugging Face 未提供 ZeroGPU 身份。

app.py

@@ -1,6 +1,5 @@
-import os; os.system('pip install --upgrade --no-deps spaces')
+import os; os.system('pip install --no-deps spaces==0.51.1')
 import spaces
-import shutil
 import subprocess
 import sys
 import copy
@@ -10,6 +9,7 @@ import warnings
 import time
 import gc
 import uuid
+import threading
 from tqdm import tqdm
 import cv2
 import numpy as np
@@ -19,6 +19,7 @@ from torch.nn import functional as F
 from PIL import Image
 
 import gradio as gr
+from gradio.context import LocalContext
 from diffusers import (
     FlowMatchEulerDiscreteScheduler,
     SASolverScheduler,
@@ -34,14 +35,22 @@ from diffusers.utils.export_utils import export_to_video
 from torchao.quantization import quantize_, Float8DynamicActivationFloat8WeightConfig, Int8WeightOnlyConfig
 import aoti
 import lora_loader
+from video_job_api import (
+    DEFAULT_NEGATIVE_PROMPT,
+    DEFAULT_PROMPT,
+    VideoJobAPI,
+    VideoJobRequest,
+    VideoJobSettings,
+    bind_context_values,
+    create_job_api_lifespan,
+)
 
 os.environ["TOKENIZERS_PARALLELISM"] = "true"
 warnings.filterwarnings("ignore")
-IS_ZERO_GPU = bool(os.getenv("SPACES_ZERO_GPU"))
-
-# if IS_ZERO_GPU:
-#     print("Loading...")
-#     subprocess.run("rm -rf /data-nvme/zerogpu-offload/*", env={}, shell=True)
+VIDEO_JOB_SETTINGS = VideoJobSettings.from_env()
+INFERENCE_SLOT = threading.Lock()
+# UI 外层不能再次自动申请 GPU；唯一 ZeroGPU 边界由 run_inference 的动态装饰器负责。
+spaces.disable_gradio_auto_wrap()
 
 # --- FRAME EXTRACTION JS & LOGIC ---
 
@@ -297,12 +306,6 @@ for i, lora in enumerate(LORA_MODELS):
         print("Failed LoRA:", name_high_tr)
         pipe.unload_lora_weights()
 
-# if os.path.exists(CACHE_DIR):
-#     shutil.rmtree(CACHE_DIR)
-#     print("Deleted Hugging Face cache.")
-# else:
-#     print("No hub cache found.")
-
 quantize_(pipe.text_encoder, Int8WeightOnlyConfig())
 torch._dynamo.reset()
 quantize_(pipe.transformer, Float8DynamicActivationFloat8WeightConfig())
@@ -322,8 +325,8 @@ spaces.aoti_load(
 # pipe.vae.enable_slicing()
 # pipe.vae.enable_tiling()
 
-default_prompt_i2v = "make this image come alive, cinematic motion, smooth animation"
-default_negative_prompt = "色调艳丽, 过曝, 静态, 细节模糊不清, 字幕, 风格, 作品, 画作, 画面, 静止, 整体发灰, 最差质量, 低质量, JPEG压缩残留, 丑陋的, 残缺的, 多余的手指, 画得不好的手部, 画得不好的脸部, 畸形的, 毁容的, 形态畸形的肢体, 手指融合, 静止不动的画面, 杂乱的背景, 三条腿, 背景人很多, 倒着走"
+default_prompt_i2v = DEFAULT_PROMPT
+default_negative_prompt = DEFAULT_NEGATIVE_PROMPT
 
 
 def model_title():
@@ -443,79 +446,137 @@ def run_inference(
     lora_groups=None,
     progress=gr.Progress(track_tqdm=True),
 ):
-    scheduler_class = SCHEDULER_MAP.get(scheduler_name)
-    if scheduler_class.__name__ != pipe.scheduler.config._class_name or flow_shift != pipe.scheduler.config.get("flow_shift", "shift"):
-        config = copy.deepcopy(original_scheduler.config)
-        if scheduler_class == FlowMatchEulerDiscreteScheduler:
-            config['shift'] = flow_shift
-        else:
-            config['flow_shift'] = flow_shift
-        pipe.scheduler = scheduler_class.from_config(config)
+    """在 ZeroGPU 上执行现有 Wan I2V 推理并生成一个临时 MP4。
 
-    clear_vram()
+    Args:
+        resized_image: 已按模型要求缩放的首图。
+        processed_last_image: 已匹配首图尺寸的可选尾图。
+        prompt: 正向提示词。
+        steps: 推理步数。
+        negative_prompt: 负向提示词。
+        num_frames: 模型需要生成的基础帧数。
+        guidance_scale: 高噪声阶段引导强度。
+        guidance_scale_2: 低噪声阶段引导强度。
+        current_seed: 本次实际使用的随机种子。
+        scheduler_name: 现有调度器映射中的名称。
+        flow_shift: 调度器流偏移值。
+        frame_multiplier: 输出目标帧率值。
+        quality: MP4 编码质量。
+        duration_seconds: 用于日志和 ZeroGPU 时长估算的视频秒数。
+        safe_mode: 是否为 ZeroGPU 估时增加安全余量。
+        lora_groups: 要动态加载的 LoRA 精确名称列表。
+        progress: Gradio 进度对象。
 
+    Returns:
+        生成的单个临时 MP4 路径与截短任务标识。
+    """
     task_name = str(uuid.uuid4())[:8]
-    print(f"Generating {num_frames} frames, task: {task_name}, {duration_seconds}, {resized_image.size}, lora={lora_groups}")
-    start = time.time()
-
-    lora_loaded = False
-    if lora_groups:
-        try:
-            for idx, name in enumerate(lora_groups):
-                if name and name != "(None)":
-                    lora_loader.load_lora_to_pipe(pipe, name, adapter_name=f"lora_{idx}")
-            lora_loaded = True
-            print(f"LoRA loaded: {lora_groups}")
-        except Exception as e:
-            print(f"LoRA warning: {e}")
-
-    result = pipe(
-        image=resized_image,
-        last_image=processed_last_image,
-        prompt=prompt,
-        negative_prompt=negative_prompt,
-        height=resized_image.height,
-        width=resized_image.width,
-        num_frames=num_frames,
-        guidance_scale=float(guidance_scale),
-        guidance_scale_2=float(guidance_scale_2),
-        num_inference_steps=int(steps),
-        generator=torch.Generator(device="cuda").manual_seed(current_seed),
-        output_type="np"
-    )
-
-    if lora_loaded:
-        lora_loader.unload_lora(pipe)
-
-    print("gen time passed:", time.time() - start)
-
-    raw_frames_np = result.frames[0]  # Returns (T, H, W, C) float32
-    pipe.scheduler = original_scheduler
+    video_path = None
+    video_ready = False
+    lora_attempted = False
+    result = None
+    raw_frames_np = None
+    final_frames = None
+
+    try:
+        scheduler_class = SCHEDULER_MAP.get(scheduler_name)
+        if scheduler_class is None:
+            raise ValueError(f"Unsupported scheduler: {scheduler_name}")
+        if scheduler_class.__name__ != pipe.scheduler.config._class_name or flow_shift != pipe.scheduler.config.get("flow_shift", "shift"):
+            config = copy.deepcopy(original_scheduler.config)
+            if scheduler_class == FlowMatchEulerDiscreteScheduler:
+                config['shift'] = flow_shift
+            else:
+                config['flow_shift'] = flow_shift
+            pipe.scheduler = scheduler_class.from_config(config)
 
-    frame_factor = frame_multiplier // FIXED_FPS
-    if frame_factor > 1:
+        clear_vram()
+        print(f"Generating {num_frames} frames, task: {task_name}, {duration_seconds}, {resized_image.size}, lora={lora_groups}")
         start = time.time()
-        print(f"Processing frames (RIFE Multiplier: {frame_factor}x)...")
-        rife_model.device()
-        rife_model.flownet = rife_model.flownet.half()
-        final_frames = interpolate_bits(raw_frames_np, multiplier=int(frame_factor))
-        print("Interpolation time passed:", time.time() - start)
-    else:
-        final_frames = list(raw_frames_np)
-
-    final_fps = FIXED_FPS * int(frame_factor)
 
-    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmpfile:
-        video_path = tmpfile.name
+        if lora_groups:
+            # 从第一项开始加载就视为已污染管线，部分加载失败也必须进入 finally 卸载。
+            lora_attempted = True
+            try:
+                for idx, name in enumerate(lora_groups):
+                    if name and name != "(None)":
+                        lora_loader.load_lora_to_pipe(pipe, name, adapter_name=f"lora_{idx}")
+                print(f"LoRA loaded: {lora_groups}")
+            except Exception as exc:
+                print(f"LoRA warning: {type(exc).__name__}")
+                # 保留原 UI 的降级语义，但不能让部分 LoRA 参与本次或后续推理。
+                lora_loader.unload_lora(pipe)
+                lora_attempted = False
+
+        result = pipe(
+            image=resized_image,
+            last_image=processed_last_image,
+            prompt=prompt,
+            negative_prompt=negative_prompt,
+            height=resized_image.height,
+            width=resized_image.width,
+            num_frames=num_frames,
+            guidance_scale=float(guidance_scale),
+            guidance_scale_2=float(guidance_scale_2),
+            num_inference_steps=int(steps),
+            generator=torch.Generator(device="cuda").manual_seed(current_seed),
+            output_type="np"
+        )
+        print("gen time passed:", time.time() - start)
+
+        raw_frames_np = result.frames[0]  # Returns (T, H, W, C) float32
+        frame_factor = frame_multiplier // FIXED_FPS
+        if frame_factor > 1:
+            start = time.time()
+            print(f"Processing frames (RIFE Multiplier: {frame_factor}x)...")
+            rife_model.device()
+            rife_model.flownet = rife_model.flownet.half()
+            final_frames = interpolate_bits(raw_frames_np, multiplier=int(frame_factor))
+            print("Interpolation time passed:", time.time() - start)
+        else:
+            final_frames = list(raw_frames_np)
 
-    start = time.time()
-    with tqdm(total=3, desc="Rendering Media", unit="clip") as pbar:
-        pbar.update(2)
-        export_to_video(final_frames, video_path, fps=final_fps, quality=quality)
-        pbar.update(1)
-    print(f"Export time passed, {final_fps} FPS:", time.time() - start)
+        final_fps = FIXED_FPS * int(frame_factor)
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmpfile:
+            video_path = tmpfile.name
 
-    return video_path, task_name
+        start = time.time()
+        with tqdm(total=3, desc="Rendering Media", unit="clip") as pbar:
+            pbar.update(2)
+            export_to_video(final_frames, video_path, fps=final_fps, quality=quality)
+            pbar.update(1)
+        print(f"Export time passed, {final_fps} FPS:", time.time() - start)
+
+        video_ready = True
+        return video_path, task_name
+    finally:
+        # 无论模型、插帧还是编码在哪一步失败，都恢复可供下一任务复用的全局管线。
+        cleanup_error = None
+        try:
+            if lora_attempted:
+                lora_loader.unload_lora(pipe)
+        except Exception as exc:
+            cleanup_error = exc
+        try:
+            pipe.scheduler = copy.deepcopy(original_scheduler)
+        except Exception as exc:
+            cleanup_error = cleanup_error or exc
+        result = None
+        raw_frames_np = None
+        final_frames = None
+        try:
+            clear_vram()
+        except Exception as exc:
+            cleanup_error = cleanup_error or exc
+        if video_path is not None and (not video_ready or cleanup_error is not None):
+            try:
+                os.remove(video_path)
+            except FileNotFoundError:
+                pass
+            except OSError as exc:
+                print(f"Failed to remove temporary video: {type(exc).__name__}")
+        if cleanup_error is not None:
+            raise RuntimeError("Inference cleanup failed.") from cleanup_error
 
 
 def generate_video(
@@ -618,6 +679,161 @@ def generate_video(
     return (video_path if video_component else None), video_path, current_seed
 
 
+def generate_video_ui(
+    input_image,
+    last_image,
+    prompt,
+    steps=4,
+    negative_prompt=default_negative_prompt,
+    duration_seconds=MAX_DURATION,
+    guidance_scale=1,
+    guidance_scale_2=1,
+    seed=42,
+    randomize_seed=False,
+    quality=5,
+    scheduler="UniPCMultistep",
+    flow_shift=6.0,
+    frame_multiplier=16,
+    safe_mode=False,
+    lora_groups=None,
+    video_component=True,
+    progress=gr.Progress(track_tqdm=True),
+):
+    """让现有 UI 通过与自定义 API 共用的非阻塞推理槽位生成视频。
+
+    Args:
+        input_image: UI 上传的首图。
+        last_image: 可选尾图。
+        prompt: 正向提示词。
+        steps: 推理步数。
+        negative_prompt: 负向提示词。
+        duration_seconds: 目标视频时长。
+        guidance_scale: 高噪声阶段引导强度。
+        guidance_scale_2: 低噪声阶段引导强度。
+        seed: 固定随机种子。
+        randomize_seed: 是否在生成前随机化种子。
+        quality: MP4 编码质量。
+        scheduler: 调度器名称。
+        flow_shift: 调度器流偏移值。
+        frame_multiplier: 输出帧率倍数对应的帧率值。
+        safe_mode: 是否申请额外 ZeroGPU 运行时间。
+        lora_groups: 当前 LoRA 下拉框选择的精确名称列表。
+        video_component: 是否把结果同时显示在视频组件中。
+        progress: Gradio 进度对象。
+
+    Returns:
+        与原 generate_video 一致的视频组件路径、下载路径和实际 seed。
+    """
+    if not INFERENCE_SLOT.acquire(blocking=False):
+        raise gr.Error("The generation service is busy. Please retry shortly.")
+    try:
+        return generate_video(
+            input_image,
+            last_image,
+            prompt,
+            steps,
+            negative_prompt,
+            duration_seconds,
+            guidance_scale,
+            guidance_scale_2,
+            seed,
+            randomize_seed,
+            quality,
+            scheduler,
+            flow_shift,
+            frame_multiplier,
+            safe_mode,
+            lora_groups,
+            video_component,
+            progress,
+        )
+    finally:
+        INFERENCE_SLOT.release()
+
+
+def build_gradio_request(headers: dict[str, str], job_id: str) -> gr.Request:
+    """为后台 ZeroGPU 调用重建最小 Gradio 请求对象。
+
+    Args:
+        headers: 仅含 ZeroGPU 身份所需字段的筛选后请求头。
+        job_id: 用作隔离会话哈希的自定义任务标识。
+
+    Returns:
+        可供 spaces.GPU 装饰器读取身份信息的 Gradio 请求。
+    """
+    return gr.Request(
+        username=headers.get("x-gradio-user"),
+        session_hash=job_id,
+        headers=dict(headers),
+        query_params={},
+        cookies={},
+        path_params={},
+        client={"host": "127.0.0.1", "port": 0},
+        url="",
+    )
+
+
+def execute_video_job(
+    payload: VideoJobRequest,
+    input_image: Image.Image,
+    last_image: Image.Image | None,
+    zero_gpu_headers: dict[str, str],
+    job_id: str,
+) -> tuple[str, int]:
+    """在后台线程恢复 Gradio 上下文并调用现有视频生成链路。
+
+    Args:
+        payload: 已通过公开 Schema 校验的命名任务参数。
+        input_image: 已安全抓取并解码的首图。
+        last_image: 已安全抓取并解码的可选尾图。
+        zero_gpu_headers: 仅含短效 ZeroGPU 身份字段的请求头。
+        job_id: 用于隔离后台 Gradio 请求上下文的任务标识。
+
+    Returns:
+        现有生成函数产生的临时 MP4 明确路径与实际使用的 seed。
+    """
+    request_context = build_gradio_request(zero_gpu_headers, job_id)
+
+    # 通用 ContextVar 绑定器保证成功或异��时都恢复四个 Gradio 本地上下文。
+    with bind_context_values(
+        (
+            (LocalContext.request, request_context),
+            (LocalContext.blocks, demo),
+            (LocalContext.in_event_listener, True),
+            (LocalContext.event_id, None),
+        )
+    ):
+        _, video_path, used_seed = generate_video(
+            input_image=input_image,
+            last_image=last_image,
+            prompt=payload.prompt,
+            steps=payload.steps,
+            negative_prompt=payload.negative_prompt,
+            duration_seconds=payload.duration_seconds,
+            guidance_scale=payload.guidance_scale,
+            guidance_scale_2=payload.guidance_scale_2,
+            seed=payload.seed,
+            randomize_seed=payload.randomize_seed,
+            quality=payload.quality,
+            scheduler=payload.scheduler,
+            flow_shift=payload.flow_shift,
+            frame_multiplier=payload.frame_multiplier,
+            safe_mode=payload.safe_mode,
+            lora_groups=payload.lora_groups,
+            video_component=False,
+        )
+    return video_path, int(used_seed)
+
+
+VIDEO_JOB_API = VideoJobAPI(
+    settings=VIDEO_JOB_SETTINGS,
+    executor=execute_video_job,
+    allowed_loras=set(lora_loader.get_lora_choices()),
+    inference_slot=INFERENCE_SLOT,
+)
+JOB_API_LIFESPAN = create_job_api_lifespan(VIDEO_JOB_API)
+
+
 CSS = """
 #hidden-timestamp {
     opacity: 0;
@@ -696,9 +912,11 @@ with gr.Blocks(delete_cache=(3600, 10800)) as demo:
     ]
     
     generate_button.click(
-        fn=generate_video, 
-        inputs=ui_inputs, 
-        outputs=[video_output, file_output, seed_input]
+        fn=generate_video_ui,
+        inputs=ui_inputs,
+        outputs=[video_output, file_output, seed_input],
+        api_name="generate_video",
+        concurrency_limit=1,
     )
     
     # --- Frame Grabbing Events ---
@@ -718,8 +936,10 @@ with gr.Blocks(delete_cache=(3600, 10800)) as demo:
     )
 
 if __name__ == "__main__":
-    demo.queue().launch(
+    demo.queue(default_concurrency_limit=1).launch(
         mcp_server=True,
         css=CSS,
         show_error=True,
-    )
+        ssr_mode=False,
+        app_kwargs={"lifespan": JOB_API_LIFESPAN},
+    )

lora_loader.py

@@ -195,7 +195,12 @@ def load_lora_to_pipe(pipe, group_name, adapter_name="lora"):
 
 
 def unload_lora(pipe):
-    try:
-        pipe.unload_lora_weights()
-    except:
-        pass
+    """严格卸载动态 LoRA，使调用方能够感知并处理清理失败。
+
+    Args:
+        pipe: 当前全局 Wan 视频推理管线。
+
+    Returns:
+        卸载完成后不返回数据；底层失败会原样抛给调用方。
+    """
+    pipe.unload_lora_weights()

requirements.txt

@@ -9,6 +9,8 @@ imageio
 imageio-ffmpeg
 opencv-python
 torchao==0.17.0
+httpx==0.28.1
+Pillow>=10,<13
 
 numpy>=1.16, <=1.23.5
 # tqdm>=4.35.0

tests/test_video_job_api.py

@@ -0,0 +1,1084 @@
+from __future__ import annotations
+
+import io
+import os
+import socket
+import tempfile
+import threading
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from contextvars import ContextVar
+from pathlib import Path
+from unittest.mock import patch
+from urllib.parse import urlsplit
+
+import httpx
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+from PIL import Image
+
+from video_job_api import (
+    ImageSourceError,
+    VideoJobAPI,
+    VideoJobRequest,
+    VideoJobSettings,
+    _parse_single_byte_range,
+    bind_context_values,
+    decode_image_bytes,
+    fetch_remote_image,
+    validate_image_url,
+)
+
+
+API_KEY = "k" * 48
+VIDEO_BYTES = b"\x00\x00\x00\x18ftypmp42fake-video-payload"
+
+
+class _TrackedImage:
+    def __init__(self, fail_on_close: bool = False) -> None:
+        self.closed = False
+        self.fail_on_close = fail_on_close
+
+    def close(self) -> None:
+        self.closed = True
+        if self.fail_on_close:
+            raise RuntimeError("close failed")
+
+
+class _FakeStreamResponse:
+    def __init__(
+        self,
+        status_code: int,
+        body: bytes,
+        content_type: str = "image/png",
+        extra_headers: dict[str, str] | None = None,
+        peer_address: str = "93.184.216.34",
+    ) -> None:
+        self.status_code = status_code
+        self.body = body
+        self.headers = {
+            "content-type": content_type,
+            "content-length": str(len(body)),
+        }
+        if extra_headers:
+            self.headers.update(extra_headers)
+        self.extensions = {
+            "network_stream": type(
+                "NetworkStreamStub",
+                (),
+                {
+                    "get_extra_info": lambda self, name: (
+                        (peer_address, 443) if name == "server_addr" else None
+                    )
+                },
+            )()
+        }
+
+    def __enter__(self) -> "_FakeStreamResponse":
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> bool:
+        return False
+
+    def iter_bytes(self):
+        yield self.body
+
+
+class VideoJobAPITest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.clients: list[TestClient] = []
+        self.result_dirs: list[Path] = []
+        self.release_events: list[threading.Event] = []
+
+    def tearDown(self) -> None:
+        for release_event in self.release_events:
+            release_event.set()
+        time.sleep(0.02)
+        for client in self.clients:
+            client.close()
+        for result_dir in self.result_dirs:
+            if not result_dir.exists():
+                continue
+            # 测试清理也只对枚举出的明确文件逐一 unlink，不做递归目录删除。
+            for child_path in tuple(result_dir.iterdir()):
+                if child_path.is_file():
+                    child_path.unlink(missing_ok=True)
+            try:
+                result_dir.rmdir()
+            except OSError:
+                pass
+
+    def _new_result_dir(self) -> Path:
+        """创建并登记一个测试专用结果目录。
+
+        Args:
+            此辅助方法不接收参数。
+
+        Returns:
+            当前测试可安全写入的唯一临时目录。
+        """
+        result_dir = Path(tempfile.mkdtemp(prefix="i2v-job-api-test-"))
+        self.result_dirs.append(result_dir)
+        return result_dir
+
+    def _settings(self, **overrides) -> VideoJobSettings:
+        """构造不读取环境变量的隔离任务设置。
+
+        Args:
+            overrides: 要覆盖的 VideoJobSettings 字段。
+
+        Returns:
+            指向测试专用目录的任务设置。
+        """
+        values = {
+            "api_key": API_KEY,
+            "allowed_hosts": frozenset({"allowed.example"}),
+            "result_dir": self._new_result_dir(),
+            "result_ttl_seconds": 1800,
+            "poll_after_seconds": 1,
+            "fetch_timeout_seconds": 15.0,
+            "max_image_bytes": 20 * 1024 * 1024,
+            "max_image_pixels": 40_000_000,
+            "space_host": "https://space.example",
+        }
+        values.update(overrides)
+        return VideoJobSettings(**values)
+
+    @staticmethod
+    def _payload(**overrides) -> dict:
+        """生成最小合法 API 请求体。
+
+        Args:
+            overrides: 要覆盖或增加的 JSON 字段。
+
+        Returns:
+            可直接提交给 POST /api/jobs 的字典。
+        """
+        payload = {
+            "input_image_url": "https://allowed.example/input.png?source=private",
+            "prompt": "animate naturally",
+            "randomize_seed": False,
+            "seed": 42,
+            "lora_groups": [],
+        }
+        payload.update(overrides)
+        return payload
+
+    @staticmethod
+    def _headers(**overrides) -> dict[str, str]:
+        """生成包含 API key 与模拟 ZeroGPU 身份的请求头。
+
+        Args:
+            overrides: 要覆盖或增加的请求头。
+
+        Returns:
+            可直接传给 TestClient 的请求头字典。
+        """
+        headers = {"X-API-Key": API_KEY, "X-IP-Token": "zero-gpu-token"}
+        headers.update(overrides)
+        return headers
+
+    def _make_service(
+        self,
+        *,
+        executor=None,
+        image_fetcher=None,
+        inference_slot: threading.Lock | None = None,
+        settings: VideoJobSettings | None = None,
+        allowed_loras: set[str] | None = None,
+        add_catch_all: bool = False,
+    ) -> tuple[TestClient, VideoJobAPI, list[_TrackedImage], dict[str, int]]:
+        """创建不导入模型的 FastAPI 测试服务。
+
+        Args:
+            executor: 可选假视频执行器。
+            image_fetcher: 可选假图片抓取器。
+            inference_slot: 可选外部共享锁。
+            settings: 可选任务设置。
+            allowed_loras: 可选合法 LoRA 精确名称集合。
+            add_catch_all: 是否先注册一个模拟 Gradio 的通配路由。
+
+        Returns:
+            TestClient、VideoJobAPI、已创建图片列表与执行次数容器。
+        """
+        settings = settings or self._settings()
+        created_images: list[_TrackedImage] = []
+        counters = {"fetch": 0, "execute": 0}
+        counter_lock = threading.Lock()
+
+        if image_fetcher is None:
+            def image_fetcher(url, current_settings):
+                del url, current_settings
+                with counter_lock:
+                    counters["fetch"] += 1
+                image = _TrackedImage()
+                created_images.append(image)
+                return image
+
+        if executor is None:
+            def executor(payload, input_image, last_image, headers, job_id):
+                del input_image, last_image, headers, job_id
+                with counter_lock:
+                    counters["execute"] += 1
+                    execution_index = counters["execute"]
+                source_path = settings.result_dir / f"source-{execution_index}.mp4"
+                source_path.write_bytes(VIDEO_BYTES)
+                return source_path, payload.seed
+
+        api = VideoJobAPI(
+            settings=settings,
+            executor=executor,
+            allowed_loras=allowed_loras or {"Valid LoRA"},
+            inference_slot=inference_slot or threading.Lock(),
+            image_fetcher=image_fetcher,
+        )
+        app = FastAPI()
+        if add_catch_all:
+            @app.api_route("/{path:path}", methods=["GET", "POST", "HEAD"])
+            def catch_all(path: str):
+                return {"catch_all": path}
+
+        api.install_on_app(app)
+        client = TestClient(app, base_url="https://space.example")
+        self.clients.append(client)
+        return client, api, created_images, counters
+
+    def _wait_for_terminal(
+        self,
+        client: TestClient,
+        job_id: str,
+        timeout_seconds: float = 3.0,
+    ):
+        """轮询测试任务直到进入成功或失败终态。
+
+        Args:
+            client: 当前隔离 FastAPI 客户端。
+            job_id: 要轮询的任务标识。
+            timeout_seconds: 测试允许的最长等待时间。
+
+        Returns:
+            第一个非 202 状态响应。
+        """
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            response = client.get(
+                f"/api/jobs/{job_id}",
+                headers={"X-API-Key": API_KEY},
+            )
+            if response.status_code != 202:
+                return response
+            time.sleep(0.01)
+        self.fail(f"job {job_id} did not reach a terminal state")
+
+    def test_auth_schema_zero_gpu_and_route_precedence(self) -> None:
+        client, api, _, counters = self._make_service(add_catch_all=True)
+
+        unauthorized = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers={"X-IP-Token": "token"},
+        )
+        self.assertEqual(unauthorized.status_code, 401)
+        self.assertEqual(counters["fetch"], 0)
+
+        unknown_field = client.post(
+            "/api/jobs",
+            json=self._payload(video_component=False),
+            headers=self._headers(),
+        )
+        self.assertEqual(unknown_field.status_code, 422)
+
+        for bad_payload in (
+            self._payload(steps=0),
+            self._payload(duration_seconds=20.2),
+            self._payload(frame_multiplier=24),
+            self._payload(prompt="   "),
+            self._payload(seed=2**31),
+            self._payload(scheduler="unknown"),
+            self._payload(input_image_url="http://allowed.example/input.png"),
+        ):
+            with self.subTest(payload=bad_payload):
+                response = client.post(
+                    "/api/jobs",
+                    json=bad_payload,
+                    headers=self._headers(),
+                )
+                self.assertEqual(response.status_code, 422)
+
+        missing_identity = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers={"X-API-Key": API_KEY},
+        )
+        self.assertEqual(missing_identity.status_code, 503)
+        self.assertEqual(missing_identity.headers["Retry-After"], "3")
+
+        with self.assertRaises(HTTPException) as unicode_key_error:
+            fake_request = type(
+                "RequestStub",
+                (),
+                {"headers": {"x-api-key": "密钥"}},
+            )()
+            api._require_api_key(fake_request)
+        self.assertEqual(unicode_key_error.exception.status_code, 401)
+
+    def test_schema_defaults_and_exact_lora_validation(self) -> None:
+        model = VideoJobRequest(input_image_url="https://allowed.example/a.png")
+        self.assertEqual(model.steps, 6)
+        self.assertEqual(model.duration_seconds, 3.5)
+        self.assertEqual(model.quality, 6)
+        self.assertEqual(model.frame_multiplier, 16)
+        self.assertTrue(model.randomize_seed)
+        self.assertTrue(model.safe_mode)
+
+        client, _, _, _ = self._make_service(allowed_loras={"Exact (HIGH only)"})
+        rejected = client.post(
+            "/api/jobs",
+            json=self._payload(lora_groups=["Exact"]),
+            headers=self._headers(),
+        )
+        self.assertEqual(rejected.status_code, 422)
+
+        accepted = client.post(
+            "/api/jobs",
+            json=self._payload(lora_groups=["Exact (HIGH only)"]),
+            headers=self._headers(),
+        )
+        self.assertEqual(accepted.status_code, 202)
+        self._wait_for_terminal(client, accepted.json()["job_id"])
+
+    def test_url_policy_requires_exact_public_https_host(self) -> None:
+        settings = self._settings()
+        public_record = (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("93.184.216.34", 443),
+        )
+        with patch("video_job_api.socket.getaddrinfo", return_value=[public_record]):
+            validated = validate_image_url(
+                "https://allowed.example/image.png?private=query",
+                settings,
+            )
+            self.assertEqual(validated.host, "allowed.example")
+
+            for invalid_url in (
+                "http://allowed.example/image.png",
+                "https://user:password@allowed.example/image.png",
+                "https://allowed.example:444/image.png",
+                "https://sub.allowed.example/image.png",
+                "https://allowed.example/image.png#fragment",
+            ):
+                with self.subTest(url=invalid_url):
+                    with self.assertRaises(ImageSourceError) as invalid_error:
+                        validate_image_url(invalid_url, settings)
+                    self.assertEqual(invalid_error.exception.status_code, 400)
+
+        private_record = (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("127.0.0.1", 443),
+        )
+        mixed_records = [public_record, private_record]
+        with patch("video_job_api.socket.getaddrinfo", return_value=mixed_records):
+            with self.assertRaises(ImageSourceError) as private_error:
+                validate_image_url("https://allowed.example/image.png", settings)
+        self.assertEqual(private_error.exception.status_code, 400)
+
+        with patch(
+            "video_job_api.socket.getaddrinfo",
+            side_effect=socket.gaierror("secret resolver detail"),
+        ):
+            with self.assertRaises(ImageSourceError) as dns_error:
+                validate_image_url("https://allowed.example/image.png", settings)
+        self.assertEqual(dns_error.exception.status_code, 502)
+        self.assertNotIn("secret", dns_error.exception.public_message)
+
+        def slow_resolver(*args, **kwargs):
+            del args, kwargs
+            time.sleep(0.2)
+            return [public_record]
+
+        short_dns_settings = self._settings(fetch_timeout_seconds=0.02)
+        started_at = time.monotonic()
+        with patch("video_job_api.socket.getaddrinfo", side_effect=slow_resolver):
+            with self.assertRaises(ImageSourceError) as dns_timeout:
+                validate_image_url(
+                    "https://allowed.example/image.png",
+                    short_dns_settings,
+                )
+        self.assertEqual(dns_timeout.exception.status_code, 504)
+        self.assertLess(time.monotonic() - started_at, 0.15)
+
+    def test_image_decode_uses_actual_format_and_enforces_limits(self) -> None:
+        settings = self._settings(max_image_pixels=4)
+        png_buffer = io.BytesIO()
+        Image.new("RGB", (2, 2), "red").save(png_buffer, format="PNG")
+        png_bytes = png_buffer.getvalue()
+
+        decoded = decode_image_bytes(png_bytes, "application/octet-stream", settings)
+        self.assertEqual(decoded.mode, "RGB")
+        self.assertEqual(decoded.size, (2, 2))
+        decoded.close()
+
+        with self.assertRaises(ImageSourceError):
+            decode_image_bytes(
+                png_bytes,
+                "image/png",
+                self._settings(max_image_bytes=len(png_bytes) - 1),
+            )
+
+        with self.assertRaises(ImageSourceError):
+            decode_image_bytes(
+                png_bytes,
+                "image/png",
+                self._settings(max_image_pixels=3),
+            )
+
+        gif_buffer = io.BytesIO()
+        Image.new("RGB", (1, 1), "blue").save(gif_buffer, format="GIF")
+        with self.assertRaises(ImageSourceError):
+            decode_image_bytes(gif_buffer.getvalue(), "image/png", settings)
+
+        with self.assertRaises(ImageSourceError):
+            decode_image_bytes(b"not-an-image", "image/png", settings)
+
+    def test_fetch_has_fresh_headers_no_redirect_and_public_error_mapping(self) -> None:
+        settings = self._settings()
+        png_buffer = io.BytesIO()
+        Image.new("RGB", (1, 1), "green").save(png_buffer, format="PNG")
+        png_bytes = png_buffer.getvalue()
+        public_record = (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("93.184.216.34", 443),
+        )
+        captured: dict = {}
+
+        class FakeClient:
+            def __init__(self, **kwargs):
+                captured["client_kwargs"] = kwargs
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, traceback):
+                return False
+
+            def close(self):
+                captured["closed"] = True
+                response = captured.get("response")
+                close_event = getattr(response, "close_event", None)
+                if close_event is not None:
+                    close_event.set()
+
+            def stream(self, method, url, headers):
+                captured["method"] = method
+                captured["url"] = url
+                captured["headers"] = dict(headers)
+                return captured["response"]
+
+        with patch("video_job_api.socket.getaddrinfo", return_value=[public_record]), patch(
+            "video_job_api.httpx.Client",
+            FakeClient,
+        ):
+            captured["response"] = _FakeStreamResponse(200, png_bytes)
+            image = fetch_remote_image(
+                "https://allowed.example/image.png?do-not-log=this",
+                settings,
+            )
+            image.close()
+            self.assertFalse(captured["client_kwargs"]["follow_redirects"])
+            self.assertFalse(captured["client_kwargs"]["trust_env"])
+            self.assertEqual(
+                captured["headers"],
+                {"Accept": "image/png,image/jpeg,image/webp"},
+            )
+            for sensitive_header in (
+                "Authorization",
+                "Cookie",
+                "X-API-Key",
+                "X-IP-Token",
+            ):
+                self.assertNotIn(sensitive_header, captured["headers"])
+
+            captured["response"] = _FakeStreamResponse(302, b"")
+            with self.assertRaises(ImageSourceError) as redirect_error:
+                fetch_remote_image("https://allowed.example/redirect", settings)
+            self.assertEqual(redirect_error.exception.status_code, 400)
+
+            captured["response"] = _FakeStreamResponse(503, b"")
+            with self.assertRaises(ImageSourceError) as upstream_error:
+                fetch_remote_image("https://allowed.example/unavailable", settings)
+            self.assertEqual(upstream_error.exception.status_code, 502)
+
+            captured["response"] = _FakeStreamResponse(
+                200,
+                png_bytes,
+                peer_address="127.0.0.1",
+            )
+            with self.assertRaises(ImageSourceError) as rebound_error:
+                fetch_remote_image("https://allowed.example/rebound", settings)
+            self.assertEqual(rebound_error.exception.status_code, 400)
+
+            captured["response"] = _FakeStreamResponse(
+                200,
+                png_bytes,
+                extra_headers={"content-length": str(settings.max_image_bytes + 1)},
+            )
+            with self.assertRaises(ImageSourceError) as size_error:
+                fetch_remote_image("https://allowed.example/large", settings)
+            self.assertEqual(size_error.exception.status_code, 400)
+
+            streamed_limit_settings = self._settings(
+                max_image_bytes=len(png_bytes) - 1,
+            )
+            captured["response"] = _FakeStreamResponse(200, png_bytes)
+            captured["response"].headers.pop("content-length")
+            with self.assertRaises(ImageSourceError) as streamed_size_error:
+                fetch_remote_image(
+                    "https://allowed.example/streamed-large",
+                    streamed_limit_settings,
+                )
+            self.assertEqual(streamed_size_error.exception.status_code, 400)
+
+            captured["response"] = _FakeStreamResponse(200, png_bytes)
+            with patch("video_job_api.time.monotonic", side_effect=[0.0, 0.0, 16.0]):
+                with self.assertRaises(ImageSourceError) as timeout_error:
+                    fetch_remote_image("https://allowed.example/slow", settings)
+            self.assertEqual(timeout_error.exception.status_code, 504)
+
+            class DeadlineResponse(_FakeStreamResponse):
+                def __init__(self):
+                    super().__init__(200, png_bytes)
+                    self.close_event = threading.Event()
+
+                def iter_bytes(self):
+                    self.close_event.wait(timeout=0.5)
+                    raise httpx.ReadError("closed by deadline")
+
+            deadline_settings = self._settings(fetch_timeout_seconds=0.05)
+            captured["response"] = DeadlineResponse()
+            deadline_started = time.monotonic()
+            with self.assertRaises(ImageSourceError) as hard_timeout_error:
+                fetch_remote_image(
+                    "https://allowed.example/hard-timeout",
+                    deadline_settings,
+                )
+            self.assertEqual(hard_timeout_error.exception.status_code, 504)
+            self.assertLess(time.monotonic() - deadline_started, 0.2)
+
+    def test_second_image_fetch_failure_closes_first_and_releases_slot(self) -> None:
+        shared_slot = threading.Lock()
+        first_image = _TrackedImage(fail_on_close=True)
+        fetch_count = 0
+
+        def fetcher(url, settings):
+            nonlocal fetch_count
+            del url, settings
+            fetch_count += 1
+            if fetch_count == 1:
+                return first_image
+            raise ImageSourceError(504, "The image download timed out.")
+
+        client, _, _, _ = self._make_service(
+            image_fetcher=fetcher,
+            inference_slot=shared_slot,
+        )
+        response = client.post(
+            "/api/jobs",
+            json=self._payload(
+                last_image_url="https://allowed.example/last.png",
+            ),
+            headers=self._headers(),
+        )
+        self.assertEqual(response.status_code, 504)
+        self.assertTrue(first_image.closed)
+        self.assertTrue(shared_slot.acquire(blocking=False))
+        shared_slot.release()
+
+    def test_success_dual_image_status_download_head_and_ranges(self) -> None:
+        seen: dict = {}
+
+        def executor(payload, input_image, last_image, headers, job_id):
+            seen.update(
+                {
+                    "payload": payload,
+                    "input_image": input_image,
+                    "last_image": last_image,
+                    "headers": dict(headers),
+                    "job_id": job_id,
+                }
+            )
+            source_path = settings.result_dir / "source.mp4"
+            source_path.write_bytes(VIDEO_BYTES)
+            return source_path, 77
+
+        settings = self._settings()
+        client, api, images, _ = self._make_service(
+            settings=settings,
+            executor=executor,
+        )
+        created = client.post(
+            "/api/jobs",
+            json=self._payload(
+                last_image_url="https://allowed.example/last.webp",
+                seed=77,
+            ),
+            headers=self._headers(
+                **{
+                    "X-Gradio-User": "caller",
+                    "Authorization": "Bearer must-not-be-retained",
+                    "Cookie": "must-not-be-retained",
+                }
+            ),
+        )
+        self.assertEqual(created.status_code, 202)
+        self.assertEqual(created.headers["Cache-Control"], "no-store")
+        self.assertEqual(created.headers["Location"], created.json()["status_url"])
+        job_id = created.json()["job_id"]
+
+        terminal = self._wait_for_terminal(client, job_id)
+        self.assertEqual(terminal.status_code, 200)
+        self.assertEqual(terminal.json()["used_seed"], 77)
+        self.assertEqual(seen["job_id"], job_id)
+        self.assertEqual(
+            seen["headers"],
+            {"x-ip-token": "zero-gpu-token", "x-gradio-user": "caller"},
+        )
+        self.assertIsNotNone(seen["last_image"])
+        self.assertTrue(all(image.closed for image in images))
+        self.assertEqual(api.jobs[job_id].zero_gpu_headers, {})
+
+        video_url = terminal.json()["video_url"]
+        parsed_video_url = urlsplit(video_url)
+        video_path = parsed_video_url.path
+        token = parsed_video_url.query.split("=", 1)[1]
+
+        self.assertEqual(client.get(video_path).status_code, 401)
+        self.assertEqual(
+            client.get(video_path, params={"token": "令牌"}).status_code,
+            401,
+        )
+
+        head = client.head(video_path, params={"token": token}, headers={"Range": "bytes=0-0"})
+        self.assertEqual(head.status_code, 200)
+        self.assertEqual(head.headers["Content-Type"], "video/mp4")
+        self.assertEqual(int(head.headers["Content-Length"]), len(VIDEO_BYTES))
+        self.assertEqual(head.headers["Accept-Ranges"], "bytes")
+        self.assertEqual(head.headers["Access-Control-Allow-Origin"], "*")
+
+        complete = client.get(video_path, params={"token": token})
+        self.assertEqual(complete.status_code, 200)
+        self.assertEqual(complete.content, VIDEO_BYTES)
+        self.assertEqual(complete.headers["Cache-Control"], "private, max-age=300")
+
+        for range_value, expected_content, expected_range in (
+            ("bytes=0-0", VIDEO_BYTES[:1], f"bytes 0-0/{len(VIDEO_BYTES)}"),
+            ("bytes=5-", VIDEO_BYTES[5:], f"bytes 5-{len(VIDEO_BYTES)-1}/{len(VIDEO_BYTES)}"),
+            ("bytes=-3", VIDEO_BYTES[-3:], f"bytes {len(VIDEO_BYTES)-3}-{len(VIDEO_BYTES)-1}/{len(VIDEO_BYTES)}"),
+            ("bytes=0-999", VIDEO_BYTES, f"bytes 0-{len(VIDEO_BYTES)-1}/{len(VIDEO_BYTES)}"),
+        ):
+            with self.subTest(range_value=range_value):
+                ranged = client.get(
+                    video_path,
+                    params={"token": token},
+                    headers={"Range": range_value},
+                )
+                self.assertEqual(ranged.status_code, 206)
+                self.assertEqual(ranged.content, expected_content)
+                self.assertEqual(ranged.headers["Content-Range"], expected_range)
+                self.assertEqual(
+                    int(ranged.headers["Content-Length"]),
+                    len(expected_content),
+                )
+
+        for invalid_range in (
+            "",
+            "items=0-1",
+            "bytes=0-1,3-4",
+            "bytes=-0",
+            f"bytes={len(VIDEO_BYTES)}-",
+        ):
+            with self.subTest(invalid_range=invalid_range):
+                rejected = client.get(
+                    video_path,
+                    params={"token": token},
+                    headers={"Range": invalid_range},
+                )
+                self.assertEqual(rejected.status_code, 416)
+                self.assertEqual(
+                    rejected.headers["Content-Range"],
+                    f"bytes */{len(VIDEO_BYTES)}",
+                )
+                self.assertEqual(rejected.headers["Access-Control-Allow-Origin"], "*")
+        self.assertEqual(api.jobs[job_id].active_downloads, 0)
+
+    def test_idempotent_replay_conflict_and_busy_submission(self) -> None:
+        release_event = threading.Event()
+        started_event = threading.Event()
+        self.release_events.append(release_event)
+        settings = self._settings()
+        execution_count = 0
+        count_lock = threading.Lock()
+
+        def blocking_executor(payload, input_image, last_image, headers, job_id):
+            nonlocal execution_count
+            del input_image, last_image, headers, job_id
+            with count_lock:
+                execution_count += 1
+            started_event.set()
+            release_event.wait(timeout=3)
+            source_path = settings.result_dir / "blocking-source.mp4"
+            source_path.write_bytes(VIDEO_BYTES)
+            return source_path, payload.seed
+
+        client, _, _, counters = self._make_service(
+            settings=settings,
+            executor=blocking_executor,
+        )
+        headers = self._headers(**{"Idempotency-Key": "same-request"})
+        first = client.post("/api/jobs", json=self._payload(), headers=headers)
+        self.assertEqual(first.status_code, 202)
+        self.assertTrue(started_event.wait(timeout=1))
+
+        replay = client.post("/api/jobs", json=self._payload(), headers=headers)
+        self.assertEqual(replay.status_code, 202)
+        self.assertEqual(replay.json()["job_id"], first.json()["job_id"])
+        self.assertEqual(counters["fetch"], 1)
+
+        conflict = client.post(
+            "/api/jobs",
+            json=self._payload(seed=43),
+            headers=headers,
+        )
+        self.assertEqual(conflict.status_code, 409)
+
+        busy = client.post(
+            "/api/jobs",
+            json=self._payload(seed=44),
+            headers=self._headers(**{"Idempotency-Key": "different-request"}),
+        )
+        self.assertEqual(busy.status_code, 503)
+        self.assertEqual(busy.headers["Retry-After"], "5")
+        self.assertEqual(counters["fetch"], 1)
+
+        release_event.set()
+        terminal = self._wait_for_terminal(client, first.json()["job_id"])
+        self.assertEqual(terminal.status_code, 200)
+        self.assertEqual(execution_count, 1)
+
+    def test_concurrent_same_idempotency_key_creates_one_job(self) -> None:
+        fetch_barrier = threading.Barrier(2)
+        fetch_count = 0
+        execute_count = 0
+        count_lock = threading.Lock()
+        settings = self._settings()
+
+        def racing_fetcher(url, current_settings):
+            nonlocal fetch_count
+            del url, current_settings
+            with count_lock:
+                fetch_count += 1
+            fetch_barrier.wait(timeout=2)
+            return _TrackedImage()
+
+        def counting_executor(payload, input_image, last_image, headers, job_id):
+            nonlocal execute_count
+            del input_image, last_image, headers, job_id
+            with count_lock:
+                execute_count += 1
+                current_count = execute_count
+            source_path = settings.result_dir / f"race-source-{current_count}.mp4"
+            source_path.write_bytes(VIDEO_BYTES)
+            return source_path, payload.seed
+
+        client, _, _, _ = self._make_service(
+            settings=settings,
+            image_fetcher=racing_fetcher,
+            executor=counting_executor,
+        )
+        headers = self._headers(**{"Idempotency-Key": "concurrent-key"})
+
+        def submit():
+            return client.post("/api/jobs", json=self._payload(), headers=headers)
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            responses = list(pool.map(lambda _: submit(), range(2)))
+        self.assertEqual([response.status_code for response in responses], [202, 202])
+        self.assertEqual(responses[0].json()["job_id"], responses[1].json()["job_id"])
+        terminal = self._wait_for_terminal(client, responses[0].json()["job_id"])
+        self.assertEqual(terminal.status_code, 200)
+        self.assertEqual(fetch_count, 2)
+        self.assertEqual(execute_count, 1)
+
+    def test_failure_is_sanitized_and_always_releases_images_and_slot(self) -> None:
+        shared_slot = threading.Lock()
+        secret_prompt = "PROMPT-MUST-NOT-APPEAR"
+
+        def failing_executor(payload, input_image, last_image, headers, job_id):
+            del payload, input_image, last_image, headers, job_id
+            raise RuntimeError("INTERNAL-SECRET-STACK-DATA")
+
+        client, api, images, _ = self._make_service(
+            executor=failing_executor,
+            inference_slot=shared_slot,
+        )
+        with self.assertLogs("video_job_api", level="ERROR") as captured_logs:
+            created = client.post(
+                "/api/jobs",
+                json=self._payload(prompt=secret_prompt),
+                headers=self._headers(),
+            )
+            terminal = self._wait_for_terminal(client, created.json()["job_id"])
+
+        self.assertEqual(terminal.status_code, 500)
+        self.assertEqual(
+            terminal.json(),
+            {"error": {"code": "GENERATION_FAILED"}},
+        )
+        combined_logs = "\n".join(captured_logs.output)
+        self.assertNotIn(secret_prompt, combined_logs)
+        self.assertNotIn("private", combined_logs)
+        self.assertNotIn("INTERNAL-SECRET-STACK-DATA", combined_logs)
+        self.assertTrue(all(image.closed for image in images))
+        self.assertEqual(api.jobs[created.json()["job_id"]].zero_gpu_headers, {})
+        self.assertTrue(shared_slot.acquire(blocking=False))
+        shared_slot.release()
+
+    def test_image_close_error_cannot_leak_shared_slot(self) -> None:
+        shared_slot = threading.Lock()
+        returned_image = _TrackedImage(fail_on_close=True)
+
+        def fetcher(url, settings):
+            del url, settings
+            return returned_image
+
+        client, _, _, _ = self._make_service(
+            image_fetcher=fetcher,
+            inference_slot=shared_slot,
+        )
+        created = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers=self._headers(),
+        )
+        terminal = self._wait_for_terminal(client, created.json()["job_id"])
+        self.assertEqual(terminal.status_code, 200)
+        self.assertTrue(returned_image.closed)
+        self.assertTrue(shared_slot.acquire(blocking=False))
+        shared_slot.release()
+
+    def test_result_not_ready_missing_file_and_one_at_a_time_ttl_cleanup(self) -> None:
+        release_event = threading.Event()
+        started_event = threading.Event()
+        self.release_events.append(release_event)
+        settings = self._settings(result_ttl_seconds=60)
+
+        def blocking_executor(payload, input_image, last_image, headers, job_id):
+            del input_image, last_image, headers, job_id
+            started_event.set()
+            release_event.wait(timeout=3)
+            source_path = settings.result_dir / f"source-{payload.seed}.mp4"
+            source_path.write_bytes(VIDEO_BYTES)
+            return source_path, payload.seed
+
+        client, api, _, _ = self._make_service(
+            settings=settings,
+            executor=blocking_executor,
+        )
+        first = client.post(
+            "/api/jobs",
+            json=self._payload(seed=1),
+            headers=self._headers(),
+        )
+        first_job_id = first.json()["job_id"]
+        self.assertTrue(started_event.wait(timeout=1))
+        first_token = api.jobs[first_job_id].download_token
+        not_ready = client.get(
+            f"/api/jobs/{first_job_id}/video",
+            params={"token": first_token},
+        )
+        self.assertEqual(not_ready.status_code, 409)
+
+        release_event.set()
+        self.assertEqual(self._wait_for_terminal(client, first_job_id).status_code, 200)
+        first_path = api.jobs[first_job_id].result_path
+        self.assertIsNotNone(first_path)
+
+        started_event.clear()
+        release_event.clear()
+        second = client.post(
+            "/api/jobs",
+            json=self._payload(seed=2),
+            headers=self._headers(),
+        )
+        second_job_id = second.json()["job_id"]
+        self.assertTrue(started_event.wait(timeout=1))
+        release_event.set()
+        self.assertEqual(self._wait_for_terminal(client, second_job_id).status_code, 200)
+        second_path = api.jobs[second_job_id].result_path
+        self.assertIsNotNone(second_path)
+
+        assert first_path is not None and second_path is not None
+        first_path.unlink()
+        missing = client.get(
+            f"/api/jobs/{first_job_id}/video",
+            params={"token": first_token},
+        )
+        self.assertEqual(missing.status_code, 410)
+
+        # 恢复一个明确文件，随后把两个记录同时标为过期，验证一次只移除其中一个。
+        first_path.write_bytes(VIDEO_BYTES)
+        api.jobs[first_job_id].completed_at = time.time() - 100
+        api.jobs[second_job_id].completed_at = time.time() - 100
+        api.remove_one_expired_job()
+        self.assertEqual(len(api.jobs), 1)
+        remaining_record = next(iter(api.jobs.values()))
+        self.assertIsNotNone(remaining_record.result_path)
+        self.assertTrue(remaining_record.result_path.is_file())
+
+    def test_external_shared_slot_returns_busy_without_fetching(self) -> None:
+        shared_slot = threading.Lock()
+        shared_slot.acquire()
+        try:
+            client, _, _, counters = self._make_service(inference_slot=shared_slot)
+            response = client.post(
+                "/api/jobs",
+                json=self._payload(),
+                headers=self._headers(),
+            )
+            self.assertEqual(response.status_code, 503)
+            self.assertEqual(response.headers["Retry-After"], "5")
+            self.assertEqual(counters["fetch"], 0)
+        finally:
+            shared_slot.release()
+
+    def test_active_download_lease_defers_and_self_heals_ttl_cleanup(self) -> None:
+        settings = self._settings(result_ttl_seconds=60)
+        client, api, _, _ = self._make_service(settings=settings)
+        created = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers=self._headers(),
+        )
+        job_id = created.json()["job_id"]
+        self.assertEqual(self._wait_for_terminal(client, job_id).status_code, 200)
+        record = api.jobs[job_id]
+        result_path = record.result_path
+        self.assertIsNotNone(result_path)
+
+        api._download_record(job_id, record.download_token)
+        record.completed_at = time.time() - 100
+        api.remove_one_expired_job()
+        self.assertIn(job_id, api.jobs)
+        assert result_path is not None
+        self.assertTrue(result_path.is_file())
+
+        record.download_lease_expires_at = time.time() - 1
+        api.remove_one_expired_job()
+        self.assertNotIn(job_id, api.jobs)
+        self.assertFalse(result_path.exists())
+
+    def test_thread_start_failure_rolls_back_and_next_job_can_run(self) -> None:
+        client, api, images, counters = self._make_service()
+        with patch("video_job_api.threading.Thread.start", side_effect=RuntimeError("start")):
+            with self.assertRaises(RuntimeError):
+                client.post(
+                    "/api/jobs",
+                    json=self._payload(),
+                    headers=self._headers(**{"Idempotency-Key": "rollback-key"}),
+                )
+
+        self.assertEqual(api.jobs, {})
+        self.assertEqual(api.idempotency_jobs, {})
+        self.assertTrue(all(image.closed for image in images))
+        self.assertFalse(api.inference_slot.locked())
+
+        retry = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers=self._headers(**{"Idempotency-Key": "rollback-key"}),
+        )
+        self.assertEqual(retry.status_code, 202)
+        self.assertEqual(self._wait_for_terminal(client, retry.json()["job_id"]).status_code, 200)
+        self.assertEqual(counters["execute"], 1)
+
+    def test_range_parser_contract(self) -> None:
+        self.assertEqual(_parse_single_byte_range("bytes=0-0", 10), (0, 0))
+        self.assertEqual(_parse_single_byte_range("bytes=5-", 10), (5, 9))
+        self.assertEqual(_parse_single_byte_range("bytes=-3", 10), (7, 9))
+        self.assertEqual(_parse_single_byte_range("bytes=0-99", 10), (0, 9))
+        for invalid_range in ("", "items=0-1", "bytes=1-0", "bytes=-0", "bytes=0-1,2-3"):
+            with self.subTest(invalid_range=invalid_range):
+                with self.assertRaises(ValueError):
+                    _parse_single_byte_range(invalid_range, 10)
+
+    def test_app_source_preserves_gradio_and_shared_context_contract(self) -> None:
+        app_source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
+        required_fragments = (
+            "spaces.disable_gradio_auto_wrap()",
+            "if not INFERENCE_SLOT.acquire(blocking=False):",
+            "inference_slot=INFERENCE_SLOT",
+            "with bind_context_values(",
+            "(LocalContext.request, request_context)",
+            "(LocalContext.blocks, demo)",
+            "(LocalContext.in_event_listener, True)",
+            "(LocalContext.event_id, None)",
+            "lora_attempted = True",
+            "lora_loader.unload_lora(pipe)",
+            "pipe.scheduler = copy.deepcopy(original_scheduler)",
+            "if video_path is not None and (not video_ready or cleanup_error is not None):",
+            "fn=generate_video_ui",
+            'api_name="generate_video"',
+            "demo.queue(default_concurrency_limit=1).launch(",
+            'mcp_server=True',
+            'ssr_mode=False',
+            'app_kwargs={"lifespan": JOB_API_LIFESPAN}',
+        )
+        for fragment in required_fragments:
+            with self.subTest(fragment=fragment):
+                self.assertIn(fragment, app_source)
+
+        lora_source = (
+            Path(__file__).parents[1] / "lora_loader.py"
+        ).read_text(encoding="utf-8")
+        strict_unload_body = lora_source.split("def unload_lora(pipe):", 1)[1]
+        self.assertIn("pipe.unload_lora_weights()", strict_unload_body)
+        self.assertNotIn("except:", strict_unload_body)
+
+    def test_context_bindings_restore_all_values_after_exception(self) -> None:
+        request_context = ContextVar("request_context", default="request-before")
+        blocks_context = ContextVar("blocks_context", default="blocks-before")
+        listener_context = ContextVar("listener_context", default=False)
+        event_context = ContextVar("event_context", default="event-before")
+
+        with self.assertRaises(RuntimeError):
+            with bind_context_values(
+                (
+                    (request_context, "request-during"),
+                    (blocks_context, "blocks-during"),
+                    (listener_context, True),
+                    (event_context, None),
+                )
+            ):
+                self.assertEqual(request_context.get(), "request-during")
+                self.assertEqual(blocks_context.get(), "blocks-during")
+                self.assertTrue(listener_context.get())
+                self.assertIsNone(event_context.get())
+                raise RuntimeError("executor failed")
+
+        self.assertEqual(request_context.get(), "request-before")
+        self.assertEqual(blocks_context.get(), "blocks-before")
+        self.assertFalse(listener_context.get())
+        self.assertEqual(event_context.get(), "event-before")
+
+
+if __name__ == "__main__":
+    unittest.main()

video_job_api.py

@@ -0,0 +1,1666 @@
+from __future__ import annotations
+
+import hashlib
+import ipaddress
+import io
+import json
+import logging
+import os
+import secrets
+import socket
+import tempfile
+import threading
+import time
+import warnings
+from collections.abc import Iterator
+from contextlib import ExitStack, asynccontextmanager, contextmanager
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Literal
+from urllib.parse import SplitResult, urlsplit
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request, Response
+from fastapi import status as http_status
+from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
+from PIL import Image, ImageOps, UnidentifiedImageError
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from starlette.background import BackgroundTask
+
+
+LOGGER = logging.getLogger(__name__)
+DNS_RESOLVER_SLOTS = threading.BoundedSemaphore(4)
+IMAGE_FETCH_SLOTS = threading.BoundedSemaphore(4)
+DOWNLOAD_LEASE_SECONDS = 600
+
+DEFAULT_PROMPT = "make this image come alive, cinematic motion, smooth animation"
+DEFAULT_NEGATIVE_PROMPT = (
+    "色调艳丽, 过曝, 静态, 细节模糊不清, 字幕, 风格, 作品, 画作, 画面, 静止, "
+    "整体发灰, 最差质量, 低质量, JPEG压缩残留, 丑陋的, 残缺的, 多余的手指, "
+    "画得不好的手部, 画得不好的脸部, 畸形的, 毁容的, 形态畸形的肢体, "
+    "手指融合, 静止不动的画面, 杂乱的背景, 三条腿, 背景人很多, 倒着走"
+)
+
+SchedulerName = Literal[
+    "FlowMatchEulerDiscrete",
+    "SASolver",
+    "DEISMultistep",
+    "DPMSolverMultistepInverse",
+    "UniPCMultistep",
+    "DPMSolverMultistep",
+    "DPMSolverSinglestep",
+]
+FrameMultiplier = Literal[16, 32, 64, 128]
+JobStatus = Literal["queued", "running", "succeeded", "failed"]
+
+VideoExecutor = Callable[
+    ["VideoJobRequest", Image.Image, Image.Image | None, dict[str, str], str],
+    tuple[str | os.PathLike[str], int],
+]
+ImageFetcher = Callable[[str, "VideoJobSettings"], Image.Image]
+
+
+@contextmanager
+def bind_context_values(bindings: tuple[tuple[Any, Any], ...]):
+    """临时设置一组 ContextVar，并保证成功或异常后按逆序恢复。
+
+    Args:
+        bindings: 按设置顺序提供的 ContextVar 与临时值二元组。
+
+    Returns:
+        进入上下文后不返回业务值，退出时恢复所有原值。
+    """
+    with ExitStack() as context_stack:
+        for context_variable, temporary_value in bindings:
+            token = context_variable.set(temporary_value)
+            context_stack.callback(context_variable.reset, token)
+        yield
+
+
+def _read_int_env(name: str, default: int) -> int:
+    raw_value = os.getenv(name)
+    if raw_value is None or not raw_value.strip():
+        return default
+    try:
+        return int(raw_value)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} must be an integer") from exc
+
+
+def _read_float_env(name: str, default: float) -> float:
+    raw_value = os.getenv(name)
+    if raw_value is None or not raw_value.strip():
+        return default
+    try:
+        return float(raw_value)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} must be a number") from exc
+
+
+def normalize_allowed_hosts(raw_hosts: str) -> frozenset[str]:
+    """规范化图片抓取白名单中的精确主机名。
+
+    Args:
+        raw_hosts: 以逗号分隔的主机名，不允许包含协议、端口、路径或通配符。
+
+    Returns:
+        小写并完成 IDNA 转换的不可变主机名集合。
+    """
+    normalized_hosts: set[str] = set()
+    for raw_host in raw_hosts.split(","):
+        host = raw_host.strip().rstrip(".")
+        if not host:
+            continue
+        if any(marker in host for marker in ("://", "/", "?", "#", "*", "@", ":")):
+            raise RuntimeError(
+                "JOB_IMAGE_ALLOWED_HOSTS must contain exact hostnames only"
+            )
+        try:
+            normalized_hosts.add(host.encode("idna").decode("ascii").lower())
+        except UnicodeError as exc:
+            raise RuntimeError(
+                "JOB_IMAGE_ALLOWED_HOSTS contains an invalid hostname"
+            ) from exc
+    if not normalized_hosts:
+        raise RuntimeError("JOB_IMAGE_ALLOWED_HOSTS must contain at least one hostname")
+    return frozenset(normalized_hosts)
+
+
+@dataclass(frozen=True, slots=True)
+class VideoJobSettings:
+    """保存自定义视频任务 API 的部署边界与资源限制。"""
+
+    api_key: str
+    allowed_hosts: frozenset[str]
+    result_dir: Path
+    result_ttl_seconds: int = 1800
+    poll_after_seconds: int = 2
+    fetch_timeout_seconds: float = 15.0
+    max_image_bytes: int = 20 * 1024 * 1024
+    max_image_pixels: int = 40_000_000
+    space_host: str = ""
+
+    @classmethod
+    def from_env(cls) -> "VideoJobSettings":
+        """从 Space Secret 与 Variables 读取并校验任务 API 配置。
+
+        Args:
+            此方法不接��参数，配置统一从当前进程环境读取。
+
+        Returns:
+            已完成边界校验的任务 API 设置。
+        """
+        api_key = os.getenv("JOB_API_KEY", "").strip()
+        if len(api_key) < 32:
+            raise RuntimeError(
+                "JOB_API_KEY must be configured as a Space Secret with at least 32 characters"
+            )
+
+        result_ttl_seconds = _read_int_env("JOB_RESULT_TTL_SECONDS", 1800)
+        poll_after_seconds = _read_int_env("JOB_POLL_AFTER_SECONDS", 2)
+        fetch_timeout_seconds = _read_float_env(
+            "JOB_IMAGE_FETCH_TIMEOUT_SECONDS", 15.0
+        )
+        max_image_bytes = _read_int_env(
+            "JOB_IMAGE_MAX_BYTES", 20 * 1024 * 1024
+        )
+        max_image_pixels = _read_int_env("JOB_IMAGE_MAX_PIXELS", 40_000_000)
+
+        if not 60 <= result_ttl_seconds <= 86400:
+            raise RuntimeError("JOB_RESULT_TTL_SECONDS must be between 60 and 86400")
+        if not 1 <= poll_after_seconds <= 30:
+            raise RuntimeError("JOB_POLL_AFTER_SECONDS must be between 1 and 30")
+        if not 1.0 <= fetch_timeout_seconds <= 15.0:
+            raise RuntimeError(
+                "JOB_IMAGE_FETCH_TIMEOUT_SECONDS must be between 1 and 15"
+            )
+        if not 1024 <= max_image_bytes <= 20 * 1024 * 1024:
+            raise RuntimeError("JOB_IMAGE_MAX_BYTES must be between 1 KiB and 20 MiB")
+        if not 1 <= max_image_pixels <= 40_000_000:
+            raise RuntimeError(
+                "JOB_IMAGE_MAX_PIXELS must be between 1 and 40000000"
+            )
+
+        result_dir = Path(tempfile.gettempdir()) / "i2v_job_results"
+        result_dir.mkdir(parents=True, exist_ok=True)
+        return cls(
+            api_key=api_key,
+            allowed_hosts=normalize_allowed_hosts(
+                os.getenv("JOB_IMAGE_ALLOWED_HOSTS", "")
+            ),
+            result_dir=result_dir,
+            result_ttl_seconds=result_ttl_seconds,
+            poll_after_seconds=poll_after_seconds,
+            fetch_timeout_seconds=fetch_timeout_seconds,
+            max_image_bytes=max_image_bytes,
+            max_image_pixels=max_image_pixels,
+            space_host=os.getenv("SPACE_HOST", "").strip().rstrip("/"),
+        )
+
+
+class VideoJobRequest(BaseModel):
+    """定义视频任务的命名参数、默认值和公开校验边界。"""
+
+    model_config = ConfigDict(
+        extra="forbid",
+        str_strip_whitespace=True,
+        allow_inf_nan=False,
+    )
+
+    input_image_url: str = Field(min_length=1, max_length=4096)
+    last_image_url: str | None = Field(default=None, max_length=4096)
+    prompt: str = Field(default=DEFAULT_PROMPT, min_length=1, max_length=5000)
+    steps: int = Field(default=6, ge=1, le=30)
+    negative_prompt: str = Field(default=DEFAULT_NEGATIVE_PROMPT, max_length=5000)
+    duration_seconds: float = Field(default=3.5, ge=0.5, le=20.1)
+    guidance_scale: float = Field(default=1.0, ge=0.0, le=10.0)
+    guidance_scale_2: float = Field(default=1.0, ge=0.0, le=10.0)
+    seed: int = Field(default=42, ge=0, le=2**31 - 1)
+    randomize_seed: bool = True
+    quality: int = Field(default=6, ge=1, le=10)
+    scheduler: SchedulerName = "UniPCMultistep"
+    flow_shift: float = Field(default=3.0, ge=0.5, le=15.0)
+    frame_multiplier: FrameMultiplier = 16
+    safe_mode: bool = True
+    lora_groups: list[str] = Field(default_factory=list)
+
+    @field_validator("prompt")
+    @classmethod
+    def reject_blank_prompt(cls, value: str) -> str:
+        """拒绝清理空白后没有内容的正向提示词。
+
+        Args:
+            value: Pydantic 已完成基础类型检查的提示词。
+
+        Returns:
+            去除两端空白后的有效提示词。
+        """
+        stripped_value = value.strip()
+        if not stripped_value:
+            raise ValueError("prompt cannot be blank")
+        return stripped_value
+
+    @field_validator("negative_prompt")
+    @classmethod
+    def strip_negative_prompt(cls, value: str) -> str:
+        """清理负向提示词两端空白。
+
+        Args:
+            value: 原始负向提示词。
+
+        Returns:
+            去除两端空白后的负向提示词。
+        """
+        return value.strip()
+
+    @field_validator("last_image_url", mode="before")
+    @classmethod
+    def normalize_optional_url(cls, value: Any) -> Any:
+        """把空字符串形式的可选尾图 URL 规范化为空值。
+
+        Args:
+            value: 调用方提交的尾图 URL 原始值。
+
+        Returns:
+            空白字符串返回 None，其余值保持不变。
+        """
+        if isinstance(value, str) and not value.strip():
+            return None
+        return value
+
+    @field_validator("input_image_url", "last_image_url")
+    @classmethod
+    def require_https_image_url(cls, value: str | None) -> str | None:
+        """在 Schema 阶段拒绝非 HTTPS 或携带凭据的图片 URL。
+
+        Args:
+            value: 首图或已规范化的可选尾图 URL。
+
+        Returns:
+            通过基础 HTTPS 语法检查的原始 URL，或可选字段的 None。
+        """
+        if value is None:
+            return None
+        try:
+            parsed = urlsplit(value)
+            port = parsed.port
+        except ValueError as exc:
+            raise ValueError("image URL is invalid") from exc
+        if (
+            parsed.scheme.lower() != "https"
+            or not parsed.hostname
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.fragment
+            or port not in {None, 443}
+        ):
+            raise ValueError(
+                "image URL must use credential-free HTTPS on the default port"
+            )
+        return value
+
+
+@dataclass(frozen=True, slots=True)
+class ValidatedImageURL:
+    """保存通过协议、主机与公网地址检查的图片 URL。"""
+
+    value: str
+    host: str
+
+
+class ImageSourceError(RuntimeError):
+    """表示图片来源不合法或暂时无法抓取。"""
+
+    def __init__(self, status_code: int, public_message: str):
+        super().__init__(public_message)
+        self.status_code = status_code
+        self.public_message = public_message
+
+
+def _resolve_host_addresses(host: str, timeout_seconds: float) -> list[Any]:
+    """在硬期限内解析主机地址，避免同步 DNS 无限阻塞请求线程。
+
+    Args:
+        host: 已通过精确白名单校验的规范化主机名。
+        timeout_seconds: DNS 解析允许占用的最长墙钟秒数。
+
+    Returns:
+        socket.getaddrinfo 返回的非空地址记录列表。
+    """
+    completed = threading.Event()
+    result: dict[str, Any] = {}
+    if not DNS_RESOLVER_SLOTS.acquire(blocking=False):
+        raise ImageSourceError(
+            http_status.HTTP_504_GATEWAY_TIMEOUT,
+            "The image host resolution is temporarily busy.",
+        )
+
+    def resolve() -> None:
+        try:
+            result["records"] = socket.getaddrinfo(
+                host,
+                443,
+                type=socket.SOCK_STREAM,
+            )
+        except Exception as exc:
+            result["error"] = exc
+        finally:
+            DNS_RESOLVER_SLOTS.release()
+            completed.set()
+
+    resolver = threading.Thread(
+        target=resolve,
+        name="i2v-image-dns",
+        daemon=True,
+    )
+    try:
+        resolver.start()
+    except Exception:
+        DNS_RESOLVER_SLOTS.release()
+        raise
+    if not completed.wait(timeout=max(0.001, timeout_seconds)):
+        raise ImageSourceError(
+            http_status.HTTP_504_GATEWAY_TIMEOUT,
+            "The image host resolution timed out.",
+        )
+    resolution_error = result.get("error")
+    if resolution_error is not None:
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image host could not be resolved.",
+        ) from resolution_error
+    address_records = result.get("records")
+    if not address_records:
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image host did not return an address.",
+        )
+    return address_records
+
+
+def validate_image_url(url: str, settings: VideoJobSettings) -> ValidatedImageURL:
+    """校验远程图片 URL 只指向白名单内的公网 HTTPS 主机。
+
+    Args:
+        url: 调用方提交的完整图片 URL。
+        settings: 包含白名单与网络资源限制的任务 API 设置。
+
+    Returns:
+        已规范化主机名并可安全用于抓取的 URL 描述。
+    """
+    try:
+        parsed: SplitResult = urlsplit(url)
+        port = parsed.port
+    except ValueError as exc:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST, "The image URL is invalid."
+        ) from exc
+
+    if parsed.scheme.lower() != "https" or not parsed.hostname:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "Image URLs must use HTTPS.",
+        )
+    if parsed.username is not None or parsed.password is not None or parsed.fragment:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The image URL contains unsupported credentials or fragments.",
+        )
+    if port not in {None, 443}:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The image URL must use the default HTTPS port.",
+        )
+
+    try:
+        normalized_host = (
+            parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
+        )
+    except UnicodeError as exc:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST, "The image URL hostname is invalid."
+        ) from exc
+
+    if normalized_host not in settings.allowed_hosts:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The image URL host is not allowed.",
+        )
+
+    address_records = _resolve_host_addresses(
+        normalized_host,
+        settings.fetch_timeout_seconds,
+    )
+
+    # 白名单只能解决预期主机边界；逐个拒绝非公网解析结果可阻断错误配置和常见 DNS 绕过。
+    for address_record in address_records:
+        raw_address = address_record[4][0].split("%", 1)[0]
+        try:
+            resolved_ip = ipaddress.ip_address(raw_address)
+        except ValueError as exc:
+            raise ImageSourceError(
+                http_status.HTTP_400_BAD_REQUEST,
+                "The image host resolved to an invalid address.",
+            ) from exc
+        if not resolved_ip.is_global:
+            raise ImageSourceError(
+                http_status.HTTP_400_BAD_REQUEST,
+                "The image host must resolve only to public addresses.",
+            )
+
+    return ValidatedImageURL(value=url, host=normalized_host)
+
+
+def _validate_connected_peer(response: httpx.Response) -> None:
+    """确认 HTTP 客户端实际连接的对端仍是公网 IP。
+
+    Args:
+        response: 已建立 TLS 连接并收到响应头的 httpx 响应。
+
+    Returns:
+        对端地址可验证且属于公网时不返回数据。
+    """
+    network_stream = response.extensions.get("network_stream")
+    if network_stream is None:
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image connection peer could not be verified.",
+        )
+    try:
+        peer_address = network_stream.get_extra_info("server_addr")
+        raw_address = peer_address[0].split("%", 1)[0]
+        peer_ip = ipaddress.ip_address(raw_address)
+    except (AttributeError, IndexError, TypeError, ValueError) as exc:
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image connection peer could not be verified.",
+        ) from exc
+    if not peer_ip.is_global:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The image connection peer must be a public address.",
+        )
+
+
+def decode_image_bytes(
+    image_bytes: bytes,
+    content_type: str,
+    settings: VideoJobSettings,
+) -> Image.Image:
+    """校验远程内容并解码为已脱离底层流的 RGB 图片。
+
+    Args:
+        image_bytes: 受大小上限约束后完整读取的图片字节。
+        content_type: 远程响应声明的 MIME 类型。
+        settings: 包含图片字节数与像素数上限的任务 API 设置。
+
+    Returns:
+        完成 EXIF 方向修正并复制到内存的 RGB PIL 图片。
+    """
+    if not image_bytes or len(image_bytes) > settings.max_image_bytes:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The remote image is empty or exceeds the configured size limit.",
+        )
+
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", Image.DecompressionBombWarning)
+            with Image.open(io.BytesIO(image_bytes)) as source_image:
+                if source_image.format not in {"JPEG", "PNG", "WEBP"}:
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "The remote image format is not supported.",
+                    )
+                width, height = source_image.size
+                if width <= 0 or height <= 0 or width * height > settings.max_image_pixels:
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "The remote image exceeds the configured pixel limit.",
+                    )
+                source_image.load()
+                return ImageOps.exif_transpose(source_image).convert("RGB").copy()
+    except ImageSourceError:
+        raise
+    except (
+        UnidentifiedImageError,
+        OSError,
+        ValueError,
+        Image.DecompressionBombError,
+        Image.DecompressionBombWarning,
+    ) as exc:
+        raise ImageSourceError(
+            http_status.HTTP_400_BAD_REQUEST,
+            "The remote image could not be decoded safely.",
+        ) from exc
+
+
+def _fetch_remote_image_inner(url: str, settings: VideoJobSettings) -> Image.Image:
+    """从白名单公网主机流式抓取并安全解码一张图片。
+
+    Args:
+        url: 调用方提交的首图或尾图 HTTPS URL。
+        settings: 图片白名单、超时、体积和像素限制。
+
+    Returns:
+        可直接交给现有 I2V 预处理逻辑的 RGB PIL 图片。
+    """
+    started_at = time.monotonic()
+    validated_url = validate_image_url(url, settings)
+    remaining_seconds = settings.fetch_timeout_seconds - (
+        time.monotonic() - started_at
+    )
+    if remaining_seconds <= 0:
+        raise ImageSourceError(
+            http_status.HTTP_504_GATEWAY_TIMEOUT,
+            "The image download timed out.",
+        )
+    timeout = httpx.Timeout(
+        remaining_seconds,
+        connect=min(5.0, remaining_seconds),
+    )
+    deadline_reached = threading.Event()
+    client = httpx.Client(
+        timeout=timeout,
+        follow_redirects=False,
+        trust_env=False,
+    )
+
+    def stop_at_deadline() -> None:
+        deadline_reached.set()
+        try:
+            client.close()
+        except Exception:
+            pass
+
+    deadline_timer = threading.Timer(remaining_seconds, stop_at_deadline)
+    deadline_timer.daemon = True
+    try:
+        with client:
+            deadline_timer.start()
+            with client.stream(
+                "GET",
+                validated_url.value,
+                headers={"Accept": "image/png,image/jpeg,image/webp"},
+            ) as response:
+                if time.monotonic() - started_at > settings.fetch_timeout_seconds:
+                    raise ImageSourceError(
+                        http_status.HTTP_504_GATEWAY_TIMEOUT,
+                        "The image download timed out.",
+                    )
+                _validate_connected_peer(response)
+                if 300 <= response.status_code < 400:
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "Image URL redirects are not allowed.",
+                    )
+                if response.status_code == http_status.HTTP_408_REQUEST_TIMEOUT:
+                    raise ImageSourceError(
+                        http_status.HTTP_504_GATEWAY_TIMEOUT,
+                        "The image download timed out.",
+                    )
+                if response.status_code == 429 or response.status_code >= 500:
+                    raise ImageSourceError(
+                        http_status.HTTP_502_BAD_GATEWAY,
+                        "The image host is temporarily unavailable.",
+                    )
+                if response.status_code != http_status.HTTP_200_OK:
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "The image URL did not return a readable resource.",
+                    )
+
+                declared_length = response.headers.get("content-length")
+                if declared_length:
+                    try:
+                        parsed_length = int(declared_length)
+                        if parsed_length < 0:
+                            raise ValueError("negative content length")
+                        if parsed_length > settings.max_image_bytes:
+                            raise ImageSourceError(
+                                http_status.HTTP_400_BAD_REQUEST,
+                                "The remote image exceeds the configured size limit.",
+                            )
+                    except ValueError as exc:
+                        raise ImageSourceError(
+                            http_status.HTTP_400_BAD_REQUEST,
+                            "The image host returned an invalid content length.",
+                        ) from exc
+
+                image_buffer = bytearray()
+                for chunk in response.iter_bytes():
+                    if time.monotonic() - started_at > settings.fetch_timeout_seconds:
+                        raise ImageSourceError(
+                            http_status.HTTP_504_GATEWAY_TIMEOUT,
+                            "The image download timed out.",
+                        )
+                    image_buffer.extend(chunk)
+                    if len(image_buffer) > settings.max_image_bytes:
+                        raise ImageSourceError(
+                            http_status.HTTP_400_BAD_REQUEST,
+                            "The remote image exceeds the configured size limit.",
+                        )
+                decoded_image = decode_image_bytes(
+                    bytes(image_buffer),
+                    response.headers.get("content-type", ""),
+                    settings,
+                )
+                deadline_timer.cancel()
+                if (
+                    deadline_reached.is_set()
+                    or time.monotonic() - started_at > settings.fetch_timeout_seconds
+                ):
+                    decoded_image.close()
+                    raise ImageSourceError(
+                        http_status.HTTP_504_GATEWAY_TIMEOUT,
+                        "The image download timed out.",
+                    )
+                return decoded_image
+    except ImageSourceError:
+        raise
+    except (httpx.RequestError, RuntimeError) as exc:
+        raise _map_image_request_error(exc, deadline_reached) from exc
+    finally:
+        deadline_timer.cancel()
+
+
+def _map_image_request_error(
+    exc: Exception,
+    deadline_reached: threading.Event,
+) -> ImageSourceError:
+    """把底层 HTTP 错误映射为不含远程 URL 的公开错误。
+
+    Args:
+        exc: httpx 或客户端关闭路径产生的底层异常。
+        deadline_reached: 硬期限看门狗是否已经触发。
+
+    Returns:
+        可由 API 层安全公开的图片来源错误。
+    """
+    if deadline_reached.is_set() or isinstance(exc, httpx.TimeoutException):
+        return ImageSourceError(
+            http_status.HTTP_504_GATEWAY_TIMEOUT,
+            "The image download timed out.",
+        )
+    return ImageSourceError(
+        http_status.HTTP_502_BAD_GATEWAY,
+        "The image host is temporarily unavailable.",
+    )
+
+
+def fetch_remote_image(url: str, settings: VideoJobSettings) -> Image.Image:
+    """以有界后台抓取器执行单图下载，并强制墙钟总期限。
+
+    Args:
+        url: 调用方提交的首图或尾图 HTTPS URL。
+        settings: 白名单、硬超时、体积与像素边界。
+
+    Returns:
+        在总期限内完成验证和解码的 RGB PIL 图片。
+    """
+    if not IMAGE_FETCH_SLOTS.acquire(blocking=False):
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image fetch service is temporarily busy.",
+        )
+
+    completed = threading.Event()
+    state_lock = threading.Lock()
+    state: dict[str, Any] = {"abandoned": False}
+
+    def fetch() -> None:
+        fetched_image: Image.Image | None = None
+        try:
+            fetched_image = _fetch_remote_image_inner(url, settings)
+            should_close = False
+            with state_lock:
+                if state["abandoned"]:
+                    should_close = True
+                else:
+                    state["image"] = fetched_image
+                    fetched_image = None
+            if should_close and fetched_image is not None:
+                try:
+                    fetched_image.close()
+                finally:
+                    fetched_image = None
+        except Exception as exc:
+            with state_lock:
+                if not state["abandoned"]:
+                    state["error"] = exc
+        finally:
+            if fetched_image is not None:
+                try:
+                    fetched_image.close()
+                except Exception as exc:
+                    LOGGER.warning(
+                        "Failed to close an abandoned fetched image: %s",
+                        type(exc).__name__,
+                    )
+            IMAGE_FETCH_SLOTS.release()
+            completed.set()
+
+    fetch_thread = threading.Thread(
+        target=fetch,
+        name="i2v-image-fetch",
+        daemon=True,
+    )
+    try:
+        fetch_thread.start()
+    except Exception as exc:
+        IMAGE_FETCH_SLOTS.release()
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image fetch service could not start.",
+        ) from exc
+
+    if not completed.wait(timeout=settings.fetch_timeout_seconds):
+        abandoned_image: Image.Image | None = None
+        with state_lock:
+            state["abandoned"] = True
+            abandoned_image = state.pop("image", None)
+        if abandoned_image is not None:
+            try:
+                abandoned_image.close()
+            except Exception as exc:
+                LOGGER.warning(
+                    "Failed to close a timed-out fetched image: %s",
+                    type(exc).__name__,
+                )
+        raise ImageSourceError(
+            http_status.HTTP_504_GATEWAY_TIMEOUT,
+            "The image download timed out.",
+        )
+
+    with state_lock:
+        fetch_error = state.get("error")
+        fetched_image = state.get("image")
+    if fetch_error is not None:
+        if isinstance(fetch_error, ImageSourceError):
+            raise fetch_error
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image host is temporarily unavailable.",
+        ) from fetch_error
+    if fetched_image is None:
+        raise ImageSourceError(
+            http_status.HTTP_502_BAD_GATEWAY,
+            "The image response could not be completed.",
+        )
+    return fetched_image
+
+
+def _constant_time_text_equal(supplied: str, expected: str) -> bool:
+    """以字节形式比较不可信文本，避免非 ASCII 输入触发类型异常。
+
+    Args:
+        supplied: 请求携带的待校验文本。
+        expected: 服务端保存的期望文本。
+
+    Returns:
+        两个 UTF-8 字节序列完全一致时返回 True。
+    """
+    return secrets.compare_digest(
+        supplied.encode("utf-8", errors="surrogatepass"),
+        expected.encode("utf-8", errors="surrogatepass"),
+    )
+
+
+def _parse_single_byte_range(range_header: str, file_size: int) -> tuple[int, int]:
+    """解析一个 HTTP bytes 区间并拒绝多区间或不可满足范围。
+
+    Args:
+        range_header: 客户端发送的完整 Range 请求头。
+        file_size: 当前 MP4 文件的总字节数。
+
+    Returns:
+        已裁剪到文件边界内的闭区间起止偏移。
+    """
+    normalized_header = range_header.strip()
+    if file_size <= 0 or "=" not in normalized_header:
+        raise ValueError("invalid byte range")
+    unit, raw_spec = normalized_header.split("=", 1)
+    if unit.strip().lower() != "bytes" or "," in raw_spec:
+        raise ValueError("only one bytes range is supported")
+
+    range_spec = raw_spec.strip()
+    if range_spec.count("-") != 1:
+        raise ValueError("invalid byte range")
+    raw_start, raw_end = range_spec.split("-", 1)
+    if raw_start:
+        if not raw_start.isdigit() or (raw_end and not raw_end.isdigit()):
+            raise ValueError("invalid byte range")
+        start = int(raw_start)
+        if start >= file_size:
+            raise ValueError("byte range is not satisfiable")
+        end = file_size - 1 if not raw_end else min(int(raw_end), file_size - 1)
+        if end < start:
+            raise ValueError("byte range is not satisfiable")
+        return start, end
+
+    if not raw_end.isdigit():
+        raise ValueError("invalid suffix byte range")
+    suffix_length = int(raw_end)
+    if suffix_length <= 0:
+        raise ValueError("invalid suffix byte range")
+    return max(0, file_size - suffix_length), file_size - 1
+
+
+def _iter_file_range(path: Path, start: int, end: int) -> Iterator[bytes]:
+    """按固定小块读取一个明确 MP4 文件的闭区间。
+
+    Args:
+        path: 要读取的单个结果文件路径。
+        start: 起始字节偏移。
+        end: 结束字节偏移，包含该字节。
+
+    Returns:
+        依次产生不超过目标闭区间的二进制块迭代器。
+    """
+    remaining = end - start + 1
+    with path.open("rb") as video_file:
+        video_file.seek(start)
+        while remaining > 0:
+            chunk = video_file.read(min(1024 * 1024, remaining))
+            if not chunk:
+                break
+            remaining -= len(chunk)
+            yield chunk
+
+
+def fingerprint_request(payload: VideoJobRequest) -> str:
+    """生成与 JSON 字段顺序无关的任务请求指纹。
+
+    Args:
+        payload: 已通过 Pydantic 校验的任务参数。
+
+    Returns:
+        用于幂等键复用校验的 SHA-256 摘要。
+    """
+    canonical_payload = json.dumps(
+        payload.model_dump(mode="json"),
+        ensure_ascii=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+    return hashlib.sha256(canonical_payload.encode("utf-8")).hexdigest()
+
+
+@dataclass(slots=True)
+class VideoJobRecord:
+    """保存单个视频任务在当前 Space 进程内的生命周期数据。"""
+
+    job_id: str
+    payload: VideoJobRequest
+    request_fingerprint: str
+    input_image: Image.Image | None
+    last_image: Image.Image | None
+    zero_gpu_headers: dict[str, str]
+    download_token: str
+    idempotency_key: str | None = None
+    status: JobStatus = "queued"
+    created_at: float = field(default_factory=time.time)
+    started_at: float | None = None
+    completed_at: float | None = None
+    result_path: Path | None = None
+    used_seed: int | None = None
+    error_type: str | None = None
+    active_downloads: int = 0
+    download_lease_expires_at: float | None = None
+
+
+class VideoJobAPI:
+    """实现与 Krea2 一致的异步视频任务 API 状态机。"""
+
+    def __init__(
+        self,
+        settings: VideoJobSettings,
+        executor: VideoExecutor,
+        allowed_loras: set[str] | frozenset[str],
+        inference_slot: threading.Lock,
+        image_fetcher: ImageFetcher = fetch_remote_image,
+    ) -> None:
+        """创建视频任务服务并注册路由定义。
+
+        Args:
+            settings: 已校验的部署配置和资源限制。
+            executor: 调用现有 Gradio/ZeroGPU 生成链路的适配函数。
+            allowed_loras: 当前 UI 可选择的 LoRA 名称集合。
+            inference_slot: 与 UI 共用的非阻塞单推理门闩。
+            image_fetcher: 可替换的远程图片抓取函数，便于隔离测试。
+
+        Returns:
+            此初始化方法不返回数据。
+        """
+        self.settings = settings
+        self.executor = executor
+        self.allowed_loras = frozenset(allowed_loras)
+        self.inference_slot = inference_slot
+        self.image_fetcher = image_fetcher
+        self.jobs: dict[str, VideoJobRecord] = {}
+        self.idempotency_jobs: dict[str, str] = {}
+        self.jobs_lock = threading.RLock()
+        self.active_job_id: str | None = None
+
+        self.router = APIRouter(tags=["jobs"])
+        self.router.add_api_route(
+            "/api/jobs",
+            self.create_job,
+            methods=["POST"],
+            name="i2v_create_job",
+        )
+        self.router.add_api_route(
+            "/api/jobs/{job_id}",
+            self.get_job_status,
+            methods=["GET"],
+            name="i2v_get_job_status",
+        )
+        self.router.add_api_route(
+            "/api/jobs/{job_id}/video",
+            self.get_job_video,
+            methods=["GET"],
+            name="i2v_get_job_video",
+        )
+        self.router.add_api_route(
+            "/api/jobs/{job_id}/video",
+            self.head_job_video,
+            methods=["HEAD"],
+            name="i2v_head_job_video",
+        )
+
+    def install_on_app(self, app: Any) -> None:
+        """把自定义任务路由安装到 Gradio 创建的 FastAPI 应用前部。
+
+        Args:
+            app: Gradio 在 launch 阶段创建的 FastAPI 应用。
+
+        Returns:
+            路由已存在或安装完成后不返回数据。
+        """
+        if getattr(app.state, "i2v_job_api_router_registered", False):
+            return
+        original_route_count = len(app.router.routes)
+        app.include_router(self.router)
+        added_routes = app.router.routes[original_route_count:]
+        original_routes = app.router.routes[:original_route_count]
+        # Gradio 带有宽泛路由，自定义 API 必须排在其前面才能稳定命中。
+        app.router.routes[:] = [*added_routes, *original_routes]
+        app.state.i2v_job_api_router_registered = True
+
+    def _require_api_key(self, request: Request) -> None:
+        supplied_key = request.headers.get("x-api-key", "")
+        if not supplied_key or not _constant_time_text_equal(
+            supplied_key,
+            self.settings.api_key,
+        ):
+            raise HTTPException(
+                status_code=http_status.HTTP_401_UNAUTHORIZED,
+                detail="A valid X-API-Key header is required.",
+            )
+
+    def _public_route_url(
+        self,
+        request: Request,
+        route_name: str,
+        **path_params: str,
+    ) -> str:
+        path = str(request.app.url_path_for(route_name, **path_params))
+        if self.settings.space_host:
+            space_host = self.settings.space_host
+            if not space_host.startswith(("http://", "https://")):
+                space_host = f"https://{space_host}"
+            return f"{space_host}{path}"
+        return str(request.url_for(route_name, **path_params))
+
+    def _is_expired(
+        self,
+        record: VideoJobRecord,
+        now: float | None = None,
+    ) -> bool:
+        current_time = time.time() if now is None else now
+        if record.active_downloads > 0:
+            lease_deadline = record.download_lease_expires_at
+            if lease_deadline is None or current_time < lease_deadline:
+                return False
+            # 响应异常可能跳过 BackgroundTask；过期租约必须能自行恢复。
+            record.active_downloads = 0
+            record.download_lease_expires_at = None
+        if record.completed_at is None:
+            return False
+        return current_time - record.completed_at > self.settings.result_ttl_seconds
+
+    @staticmethod
+    def _unlink_one(path: Path | None) -> None:
+        if path is None:
+            return
+        try:
+            path.unlink(missing_ok=True)
+        except OSError as exc:
+            LOGGER.warning(
+                "Failed to remove video job file %s: %s",
+                path.stem[:8],
+                type(exc).__name__,
+            )
+
+    @staticmethod
+    def _close_images(*images: Image.Image | None) -> None:
+        """逐一关闭已知图片并隔离单个清理异常。
+
+        Args:
+            images: 要释放的零个或多个明确图片对象。
+
+        Returns:
+            清理完成后不返回数据；关闭错误仅写入脱敏日志。
+        """
+        for image in images:
+            if image is not None:
+                try:
+                    image.close()
+                except Exception as exc:
+                    LOGGER.warning(
+                        "Failed to close a video job image: %s",
+                        type(exc).__name__,
+                    )
+
+    @staticmethod
+    def _close_job_images(record: VideoJobRecord | None) -> None:
+        if record is None:
+            return
+        VideoJobAPI._close_images(record.input_image, record.last_image)
+        record.input_image = None
+        record.last_image = None
+
+    def _remove_record_locked(self, record: VideoJobRecord) -> None:
+        self.jobs.pop(record.job_id, None)
+        if record.idempotency_key is not None:
+            if self.idempotency_jobs.get(record.idempotency_key) == record.job_id:
+                self.idempotency_jobs.pop(record.idempotency_key, None)
+
+    def remove_one_expired_job(self) -> None:
+        """每次仅清理一个过期终态任务及其明确结果文件。
+
+        Args:
+            此方法不接收参数。
+
+        Returns:
+            清理完成后不返回数据。
+        """
+        expired_record: VideoJobRecord | None = None
+        now = time.time()
+        with self.jobs_lock:
+            for record in self.jobs.values():
+                if self._is_expired(record, now):
+                    expired_record = record
+                    self._remove_record_locked(record)
+                    break
+        if expired_record is not None:
+            self._close_job_images(expired_record)
+            self._unlink_one(expired_record.result_path)
+
+    def _job_or_404(self, job_id: str) -> VideoJobRecord:
+        expired_record: VideoJobRecord | None = None
+        with self.jobs_lock:
+            record = self.jobs.get(job_id)
+            if record is not None and self._is_expired(record):
+                expired_record = record
+                self._remove_record_locked(record)
+                record = None
+        if expired_record is not None:
+            self._close_job_images(expired_record)
+            self._unlink_one(expired_record.result_path)
+        if record is None:
+            raise HTTPException(
+                status_code=http_status.HTTP_404_NOT_FOUND,
+                detail="Job not found or result expired.",
+            )
+        return record
+
+    def _validate_loras(self, lora_groups: list[str]) -> None:
+        if len(lora_groups) != len(set(lora_groups)):
+            raise HTTPException(
+                status_code=http_status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="lora_groups cannot contain duplicate choices.",
+            )
+        unknown_loras = [
+            lora_name
+            for lora_name in lora_groups
+            if lora_name not in self.allowed_loras
+        ]
+        if unknown_loras:
+            raise HTTPException(
+                status_code=http_status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail="lora_groups contains an unsupported choice.",
+            )
+
+    def _status_response(
+        self,
+        request: Request,
+        record: VideoJobRecord,
+    ) -> JSONResponse:
+        status_url = self._public_route_url(
+            request,
+            "i2v_get_job_status",
+            job_id=record.job_id,
+        )
+        return JSONResponse(
+            status_code=http_status.HTTP_202_ACCEPTED,
+            headers={
+                "Location": status_url,
+                "Cache-Control": "no-store",
+            },
+            content={
+                "job_id": record.job_id,
+                "status_url": status_url,
+                "poll_after_seconds": self.settings.poll_after_seconds,
+            },
+        )
+
+    def _lookup_idempotent_job(
+        self,
+        idempotency_key: str | None,
+        request_fingerprint: str,
+    ) -> tuple[VideoJobRecord | None, Path | None]:
+        if idempotency_key is None:
+            return None, None
+        expired_path: Path | None = None
+        with self.jobs_lock:
+            existing_job_id = self.idempotency_jobs.get(idempotency_key)
+            record = self.jobs.get(existing_job_id or "")
+            if record is not None and self._is_expired(record):
+                self._remove_record_locked(record)
+                expired_path = record.result_path
+                record = None
+            if record is None:
+                if existing_job_id is not None:
+                    self.idempotency_jobs.pop(idempotency_key, None)
+                return None, expired_path
+            if not secrets.compare_digest(
+                record.request_fingerprint,
+                request_fingerprint,
+            ):
+                raise HTTPException(
+                    status_code=http_status.HTTP_409_CONFLICT,
+                    detail=(
+                        "Idempotency-Key was already used with a different "
+                        "request body."
+                    ),
+                )
+            return record, expired_path
+
+    def create_job(
+        self,
+        payload: VideoJobRequest,
+        request: Request,
+    ) -> JSONResponse:
+        """验证远程图片并创建立即执行的异步视频任务。
+
+        Args:
+            payload: 已通过 Pydantic 校验的命名视频参数。
+            request: 包含共享密钥与 ZeroGPU 身份头的 FastAPI 请求。
+
+        Returns:
+            HTTP 202 任务标识、状态 URL 与建议轮询间隔。
+        """
+        self._require_api_key(request)
+        self.remove_one_expired_job()
+        self._validate_loras(payload.lora_groups)
+
+        idempotency_key = request.headers.get("idempotency-key", "").strip() or None
+        if idempotency_key is not None and len(idempotency_key) > 200:
+            raise HTTPException(
+                status_code=http_status.HTTP_400_BAD_REQUEST,
+                detail="Idempotency-Key must be 200 characters or fewer.",
+            )
+        request_fingerprint = fingerprint_request(payload)
+        existing_record, expired_path = self._lookup_idempotent_job(
+            idempotency_key,
+            request_fingerprint,
+        )
+        self._unlink_one(expired_path)
+        if existing_record is not None:
+            return self._status_response(request, existing_record)
+
+        x_ip_token = request.headers.get("x-ip-token")
+        if not x_ip_token:
+            raise HTTPException(
+                status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=(
+                    "The ZeroGPU identity header is unavailable. "
+                    "Retry through the Space URL."
+                ),
+                headers={"Retry-After": "3"},
+            )
+
+        with self.jobs_lock:
+            active_record = self.jobs.get(self.active_job_id or "")
+            api_job_active = active_record is not None and active_record.status in {
+                "queued",
+                "running",
+            }
+        # 抓图前先做快速忙检查；抓图后仍会在同一状态锁内做一次权威检查。
+        if api_job_active or self.inference_slot.locked():
+            raise HTTPException(
+                status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="The generation service is busy.",
+                headers={"Retry-After": "5"},
+            )
+
+        slot_owned_by_request = False
+        record: VideoJobRecord | None = None
+        replay_record: VideoJobRecord | None = None
+        input_image: Image.Image | None = None
+        last_image: Image.Image | None = None
+        second_expired_path: Path | None = None
+        try:
+            try:
+                input_image = self.image_fetcher(payload.input_image_url, self.settings)
+                last_image = (
+                    self.image_fetcher(payload.last_image_url, self.settings)
+                    if payload.last_image_url is not None
+                    else None
+                )
+            except ImageSourceError as exc:
+                raise HTTPException(
+                    status_code=exc.status_code,
+                    detail=exc.public_message,
+                ) from exc
+
+            with self.jobs_lock:
+                # 两个相同幂等键可并发抓图；只有此处的二次判定有权创建任务。
+                replay_record, second_expired_path = self._lookup_idempotent_job(
+                    idempotency_key,
+                    request_fingerprint,
+                )
+                if replay_record is None:
+                    active_record = self.jobs.get(self.active_job_id or "")
+                    if active_record is not None and active_record.status in {
+                        "queued",
+                        "running",
+                    }:
+                        raise HTTPException(
+                            status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE,
+                            detail="The generation service is busy.",
+                            headers={"Retry-After": "5"},
+                        )
+                    if not self.inference_slot.acquire(blocking=False):
+                        raise HTTPException(
+                            status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE,
+                            detail="The generation service is busy.",
+                            headers={"Retry-After": "5"},
+                        )
+                    slot_owned_by_request = True
+
+                    job_id = secrets.token_urlsafe(24)
+                    zero_gpu_headers = {"x-ip-token": x_ip_token}
+                    x_gradio_user = request.headers.get("x-gradio-user")
+                    if x_gradio_user:
+                        zero_gpu_headers["x-gradio-user"] = x_gradio_user
+
+                    record = VideoJobRecord(
+                        job_id=job_id,
+                        payload=payload,
+                        request_fingerprint=request_fingerprint,
+                        input_image=input_image,
+                        last_image=last_image,
+                        zero_gpu_headers=zero_gpu_headers,
+                        download_token=secrets.token_urlsafe(32),
+                        idempotency_key=idempotency_key,
+                    )
+                    self.jobs[job_id] = record
+                    self.active_job_id = job_id
+                    if idempotency_key is not None:
+                        self.idempotency_jobs[idempotency_key] = job_id
+
+            self._unlink_one(second_expired_path)
+            second_expired_path = None
+            if replay_record is not None:
+                self._close_images(input_image, last_image)
+                input_image = None
+                last_image = None
+                return self._status_response(request, replay_record)
+
+            if record is None:
+                raise RuntimeError("Job initialization did not produce a record.")
+
+            response = self._status_response(request, record)
+            worker = threading.Thread(
+                target=self._execute_job,
+                args=(record,),
+                name=f"i2v-job-{record.job_id[:8]}",
+                daemon=True,
+            )
+            worker.start()
+            # 线程成功启动后，推理槽位所有权转交给 worker 最外层 finally。
+            slot_owned_by_request = False
+            return response
+        except Exception:
+            with self.jobs_lock:
+                if record is not None:
+                    self._remove_record_locked(record)
+                    record.zero_gpu_headers.clear()
+                if record is not None and self.active_job_id == record.job_id:
+                    self.active_job_id = None
+            self._close_job_images(record)
+            if record is None:
+                # 尾图抓取、竞争判定或任务记录创建失败时，关闭已下载图片。
+                self._close_images(input_image, last_image)
+            if slot_owned_by_request:
+                self.inference_slot.release()
+            self._unlink_one(second_expired_path)
+            raise
+
+    def _execute_job(self, record: VideoJobRecord) -> None:
+        generated_path: Path | None = None
+        final_path: Path | None = None
+        try:
+            with self.jobs_lock:
+                current_record = self.jobs.get(record.job_id)
+                if current_record is not record or record.status != "queued":
+                    return
+                record.status = "running"
+                record.started_at = time.time()
+
+            if record.input_image is None:
+                raise RuntimeError("The job input image is unavailable.")
+            generated_value, used_seed = self.executor(
+                record.payload,
+                record.input_image,
+                record.last_image,
+                record.zero_gpu_headers,
+                record.job_id,
+            )
+            generated_path = Path(generated_value)
+            if not generated_path.is_file() or generated_path.stat().st_size <= 0:
+                raise RuntimeError("The video generator did not produce a readable MP4.")
+
+            final_path = self.settings.result_dir / f"{record.job_id}.mp4"
+            os.replace(generated_path, final_path)
+            generated_path = None
+
+            with self.jobs_lock:
+                current_record = self.jobs.get(record.job_id)
+                if current_record is record and record.status == "running":
+                    record.status = "succeeded"
+                    record.result_path = final_path
+                    record.used_seed = int(used_seed)
+                    record.completed_at = time.time()
+                else:
+                    self._unlink_one(final_path)
+        except Exception as exc:
+            LOGGER.error(
+                "Video job %s failed with %s",
+                record.job_id[:8],
+                type(exc).__name__,
+            )
+            self._unlink_one(generated_path)
+            self._unlink_one(final_path)
+            with self.jobs_lock:
+                current_record = self.jobs.get(record.job_id)
+                if current_record is record and record.status in {"queued", "running"}:
+                    record.status = "failed"
+                    record.error_type = type(exc).__name__
+                    record.completed_at = time.time()
+        finally:
+            with self.jobs_lock:
+                record.zero_gpu_headers.clear()
+                if self.active_job_id == record.job_id:
+                    self.active_job_id = None
+            self._close_job_images(record)
+            self.inference_slot.release()
+            self.remove_one_expired_job()
+
+    def get_job_status(self, job_id: str, request: Request) -> JSONResponse:
+        """读取视频任务状态并在成功后返回签名视频 URL。
+
+        Args:
+            job_id: POST 创建任务时返回的自定义任务标识。
+            request: 用于鉴权和生成公网路由地址的 FastAPI 请求。
+
+        Returns:
+            运行中返回 202，成功返回视频 URL 与实际 seed，失败返回 500。
+        """
+        self._require_api_key(request)
+        record = self._job_or_404(job_id)
+        with self.jobs_lock:
+            status_value = record.status
+            used_seed = record.used_seed
+            download_token = record.download_token
+
+        if status_value in {"queued", "running"}:
+            return JSONResponse(
+                status_code=http_status.HTTP_202_ACCEPTED,
+                headers={
+                    "Retry-After": str(self.settings.poll_after_seconds),
+                    "Cache-Control": "no-store",
+                },
+                content={"status": status_value},
+            )
+        if status_value == "failed":
+            return JSONResponse(
+                status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
+                headers={"Cache-Control": "no-store"},
+                content={
+                    "error": {
+                        "code": "GENERATION_FAILED",
+                    }
+                },
+            )
+
+        video_url = self._public_route_url(
+            request,
+            "i2v_get_job_video",
+            job_id=job_id,
+        )
+        return JSONResponse(
+            content={
+                "video_url": f"{video_url}?token={download_token}",
+                "used_seed": used_seed,
+            },
+            headers={"Cache-Control": "no-store"},
+        )
+
+    def _download_record(
+        self,
+        job_id: str,
+        token: str | None,
+    ) -> tuple[VideoJobRecord, Path, int]:
+        expired_record: VideoJobRecord | None = None
+        with self.jobs_lock:
+            record = self.jobs.get(job_id)
+            if record is not None and self._is_expired(record):
+                expired_record = record
+                self._remove_record_locked(record)
+                record = None
+            if record is not None:
+                if not token or not _constant_time_text_equal(
+                    token,
+                    record.download_token,
+                ):
+                    raise HTTPException(
+                        status_code=http_status.HTTP_401_UNAUTHORIZED,
+                        detail="A valid video download token is required.",
+                    )
+                if record.status != "succeeded" or record.result_path is None:
+                    raise HTTPException(
+                        status_code=http_status.HTTP_409_CONFLICT,
+                        detail="The job has not produced a video.",
+                    )
+                try:
+                    file_size = record.result_path.stat().st_size
+                except OSError as exc:
+                    raise HTTPException(
+                        status_code=http_status.HTTP_410_GONE,
+                        detail="The result file is no longer available.",
+                    ) from exc
+                # 下载租约使 TTL 清理在响应完成前跳过该明确文件。
+                record.active_downloads += 1
+                lease_deadline = time.time() + DOWNLOAD_LEASE_SECONDS
+                record.download_lease_expires_at = max(
+                    record.download_lease_expires_at or 0.0,
+                    lease_deadline,
+                )
+                return record, record.result_path, file_size
+
+        if expired_record is not None:
+            self._close_job_images(expired_record)
+            self._unlink_one(expired_record.result_path)
+        raise HTTPException(
+            status_code=http_status.HTTP_404_NOT_FOUND,
+            detail="Job not found or result expired.",
+        )
+
+    def _release_download(self, job_id: str) -> None:
+        """释放一个视频响应租约并按需清理刚过期的单个文件。
+
+        Args:
+            job_id: 当前下载响应对应的任务标识。
+
+        Returns:
+            租约释放和可选单文件清理完成后不返回数据。
+        """
+        expired_record: VideoJobRecord | None = None
+        with self.jobs_lock:
+            record = self.jobs.get(job_id)
+            if record is None:
+                return
+            if record.active_downloads > 0:
+                record.active_downloads -= 1
+            if record.active_downloads == 0:
+                record.download_lease_expires_at = None
+            if self._is_expired(record):
+                expired_record = record
+                self._remove_record_locked(record)
+        if expired_record is not None:
+            self._close_job_images(expired_record)
+            self._unlink_one(expired_record.result_path)
+
+    @staticmethod
+    def _video_headers() -> dict[str, str]:
+        return {
+            "Cache-Control": "private, max-age=300",
+            "Referrer-Policy": "no-referrer",
+            "Accept-Ranges": "bytes",
+            "Access-Control-Allow-Origin": "*",
+            "Access-Control-Expose-Headers": (
+                "Accept-Ranges, Content-Length, Content-Range"
+            ),
+        }
+
+    def get_job_video(
+        self,
+        job_id: str,
+        request: Request,
+        token: str | None = None,
+    ) -> Response:
+        """通过随机下载令牌返回支持 Range 的 MP4 文件。
+
+        Args:
+            job_id: 自定义视频任务标识。
+            request: 用于读取可选单区间 Range 请求头的 FastAPI 请求。
+            token: 成功状态响应中视频 URL 携带的随机下载令牌。
+
+        Returns:
+            支持浏览器跨域播放和区间读取的 MP4 FileResponse。
+        """
+        record, result_path, file_size = self._download_record(job_id, token)
+        background = BackgroundTask(self._release_download, job_id)
+        range_header = request.headers.get("range")
+        if range_header is not None:
+            try:
+                range_start, range_end = _parse_single_byte_range(
+                    range_header,
+                    file_size,
+                )
+            except ValueError as exc:
+                error_headers = self._video_headers()
+                error_headers["Content-Range"] = f"bytes */{file_size}"
+                self._release_download(job_id)
+                raise HTTPException(
+                    status_code=http_status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
+                    detail="The requested byte range is not satisfiable.",
+                    headers=error_headers,
+                ) from exc
+
+            response_headers = self._video_headers()
+            response_headers.update(
+                {
+                    "Content-Length": str(range_end - range_start + 1),
+                    "Content-Range": (
+                        f"bytes {range_start}-{range_end}/{file_size}"
+                    ),
+                    "Content-Disposition": f'inline; filename="{job_id}.mp4"',
+                }
+            )
+            return StreamingResponse(
+                _iter_file_range(result_path, range_start, range_end),
+                status_code=http_status.HTTP_206_PARTIAL_CONTENT,
+                media_type="video/mp4",
+                headers=response_headers,
+                background=background,
+            )
+
+        return FileResponse(
+            result_path,
+            media_type="video/mp4",
+            filename=f"{job_id}.mp4",
+            content_disposition_type="inline",
+            headers=self._video_headers(),
+            background=background,
+        )
+
+    def head_job_video(
+        self,
+        job_id: str,
+        token: str | None = None,
+    ) -> Response:
+        """验证签名视频链接并仅返回文件元数据。
+
+        Args:
+            job_id: 自定义视频任务标识。
+            token: 成功状态响应中视频 URL 携带的随机下载令牌。
+
+        Returns:
+            包含 MP4 类型、长度、Range 与跨域头的空响应。
+        """
+        _, _, file_size = self._download_record(job_id, token)
+        try:
+            headers = self._video_headers()
+            headers.update(
+                {
+                    "Content-Length": str(file_size),
+                    "Accept-Ranges": "bytes",
+                }
+            )
+            return Response(
+                status_code=http_status.HTTP_200_OK,
+                media_type="video/mp4",
+                headers=headers,
+            )
+        finally:
+            self._release_download(job_id)
+
+
+def create_job_api_lifespan(job_api: VideoJobAPI):
+    """创建供 Gradio launch 组合使用的自定义路由 lifespan。
+
+    Args:
+        job_api: 已配置执行器和共享推理锁的视频任务服务。
+
+    Returns:
+        可传给 FastAPI app_kwargs 的异步 lifespan 上下文管理器。
+    """
+
+    @asynccontextmanager
+    async def job_api_lifespan(app: Any):
+        """在 Gradio 应用开始接收请求前安装自定义任务路由。
+
+        Args:
+            app: Gradio 创建并传入 lifespan 的 FastAPI 应用。
+
+        Returns:
+            lifespan 启动和关闭阶段不返回业务数据。
+        """
+        job_api.install_on_app(app)
+        yield
+
+    return job_api_lifespan