Add custom Image Inversion job API

README.md

@@ -13,3 +13,36 @@ hf_oauth: true
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+
+## 自定义任务 API
+
+该 Space 在保留原 Gradio 页面功能的同时提供受共享密钥保护的异步标签分析接口：
+
+- `POST /api/jobs`：创建任务，返回 `job_id`、`status_url` 与 `poll_after_seconds`。
+- `GET /api/jobs/{job_id}`：运行中返回 HTTP 202；成功返回标签展示字段；失败返回脱敏错误码。
+- 两个接口都必须携带 `X-API-Key`。创建接口支持最长 200 字符的 `Idempotency-Key`。
+
+创建请求 JSON：
+
+```json
+{
+  "input_image_url": "https://example.com/input.png",
+  "general_threshold": 0.35,
+  "character_threshold": 0.85,
+  "show_confidence": true,
+  "show_general": true,
+  "show_character": true,
+  "show_ip": true,
+  "separator": "comma",
+  "show_chinese": true
+}
+```
+
+`separator` 仅接受 `comma`、`newline` 或 `space`。远程图片仅允许来自 `JOB_IMAGE_ALLOWED_HOSTS` 配置的精确 HTTPS 主机。
+
+部署时必须设置：
+
+- Secret：`JOB_API_KEY`（至少 32 字符）。
+- Variable：`JOB_IMAGE_ALLOWED_HOSTS`（逗号分隔的精确域名，不支持通配符）。
+
+可选 Variables：`JOB_RESULT_TTL_SECONDS`、`JOB_POLL_AFTER_SECONDS`、`JOB_IMAGE_FETCH_TIMEOUT_SECONDS`、`JOB_IMAGE_MAX_BYTES`、`JOB_IMAGE_MAX_PIXELS`、`JOB_IMAGE_MAX_DIMENSION`、`JOB_MAX_RECORDS`、`SPACE_HOST`。

app.py

@@ -1,7 +1,12 @@
 import os
+
+# 模型仓库均为公开资源；必须在任何 Hub 间接依赖导入前禁用隐式私有 token。
+os.environ.setdefault("HF_HUB_DISABLE_IMPLICIT_TOKEN", "1")
+
 import json
 import time
 import shutil
+import threading
 import warnings
 from html import escape
 from pathlib import Path
@@ -12,6 +17,12 @@ from huggingface_hub import snapshot_download
 from PIL import Image, ImageFile
 
 from handler import EndpointHandler
+from image_inversion_job_api import (
+    ImageInversionJobAPI,
+    ImageInversionJobRequest,
+    ImageInversionJobSettings,
+    create_job_api_lifespan,
+)
 from translator import translate_texts
 
 # ------------------------------------------------------------------
@@ -129,13 +140,6 @@ ASSETS_REPO_ID = os.environ.get("ASSETS_REPO_ID", "pixai-labs/pixai-tagger-v0.9"
 ASSETS_REVISION = os.environ.get("ASSETS_REVISION")
 MODEL_DIR = os.environ.get("MODEL_DIR", "./assets")
 
-HF_TOKEN = (
-    os.environ.get("HUGGINGFACE_HUB_TOKEN")
-    or os.environ.get("HF_TOKEN")
-    or os.environ.get("HUGGINGFACE_TOKEN")
-    or os.environ.get("HUGGINGFACEHUB_API_TOKEN")
-)
-
 REQUIRED_FILES = [
     "model_v0.9.pth",
     "tags_v0.9_13k.json",
@@ -159,7 +163,7 @@ def ensure_assets(repo_id: str, revision: Optional[str], target_dir: str) -> Non
         repo_id=repo_id,
         revision=revision,
         allow_patterns=REQUIRED_FILES,
-        token=HF_TOKEN,
+        token=False,
     )
 
     for fname in REQUIRED_FILES:
@@ -313,6 +317,38 @@ DEVICE_LABEL = (
     else "设备：UNKNOWN"
 )
 
+# UI 与自定义任务 API 共用单推理门闩，避免 CPU 模型被并发调用导致资源争用。
+INFERENCE_SLOT = threading.Lock()
+
+
+def predict_tags_with_shared_slot(
+    image: Image.Image,
+    general_threshold: float,
+    character_threshold: float,
+):
+    """在共享单推理门闩内调用现有标签模型。
+
+    Args:
+        image: 已完成安全校验与 RGB 转换的输入图片。
+        general_threshold: 通用标签阈值。
+        character_threshold: 角色标签阈值。
+
+    Returns:
+        Tagger.predict 返回的标签、翻译顺序和元数据三元组。
+    """
+    if tagger_instance is None:
+        raise RuntimeError("标签分析器尚未成功初始化。")
+    if not INFERENCE_SLOT.acquire(blocking=False):
+        raise RuntimeError("标签分析服务正忙，请稍后重试。")
+    try:
+        return tagger_instance.predict(
+            image,
+            general_threshold,
+            character_threshold,
+        )
+    finally:
+        INFERENCE_SLOT.release()
+
 # ------------------------------------------------------------------
 # Gradio UI
 # ------------------------------------------------------------------
@@ -639,7 +675,11 @@ with gr.Blocks(theme=gr.themes.Soft(), title="AI 图像标签分析器", css=cus
 
         try:
             img = validate_and_open_image(image_path)
-            res, tag_categories_original_order, meta = tagger_instance.predict(img, g_th, c_th)
+            res, tag_categories_original_order, meta = predict_tags_with_shared_slot(
+                img,
+                g_th,
+                c_th,
+            )
 
             all_tags_to_translate = []
             for cat_key in ["general", "characters", "ips"]:
@@ -802,7 +842,113 @@ with gr.Blocks(theme=gr.themes.Soft(), title="AI 图像标签分析器", css=cus
         )
 
 
+def execute_image_inversion_job(
+    payload: ImageInversionJobRequest,
+    input_image: Image.Image,
+    job_id: str,
+) -> dict:
+    """复用现有 Tagger、翻译与展示格式生成自定义 API 结果。
+
+    Args:
+        payload: 已通过公开 Schema 校验的标签分析参数。
+        input_image: 已由任务 API 安全抓取并转换为 RGB 的图片。
+        job_id: 当前异步任务标识，仅用于隔离任务上下文。
+
+    Returns:
+        包含现有前端所需六个展示字段的 JSON 对象。
+    """
+    del job_id
+    if tagger_instance is None:
+        raise RuntimeError("标签分析器尚未成功初始化。")
+
+    # ImageInversionJobAPI 已持有 INFERENCE_SLOT，此处不能重复加锁。
+    result, tag_order, metadata = tagger_instance.predict(
+        input_image,
+        payload.general_threshold,
+        payload.character_threshold,
+    )
+
+    all_tags: list[str] = []
+    for category_name in ("general", "characters", "ips"):
+        all_tags.extend(tag_order.get(category_name, []))
+
+    translated_tags: list[str] = []
+    if all_tags:
+        try:
+            translated_tags = translate_texts(
+                all_tags,
+                src_lang="auto",
+                tgt_lang="zh",
+            )
+        except Exception as exc:
+            print(f"标签翻译失败，将仅返回英文标签：{type(exc).__name__}")
+            translated_tags = [""] * len(all_tags)
+
+    translations: dict[str, list[str]] = {}
+    offset = 0
+    for category_name in ("general", "characters", "ips"):
+        category_tags = tag_order.get(category_name, [])
+        tag_count = len(category_tags)
+        translations[category_name] = translated_tags[offset : offset + tag_count]
+        offset += tag_count
+
+    separator_names = {
+        "comma": "逗号",
+        "newline": "换行",
+        "space": "空格",
+    }
+    summary_text = generate_summary_text_content(
+        result,
+        translations,
+        payload.show_general,
+        payload.show_character,
+        payload.show_ip,
+        separator_names[payload.separator],
+        payload.show_chinese,
+    )
+
+    return {
+        "status_markdown": "✅ 分析完成!",
+        "general_tags_html": format_tags_html(
+            result.get("general", {}),
+            translations.get("general", []),
+            "general",
+            payload.show_confidence,
+            True,
+        ),
+        "character_tags_html": format_tags_html(
+            result.get("characters", {}),
+            translations.get("characters", []),
+            "characters",
+            payload.show_confidence,
+            True,
+        ),
+        "ip_tags_html": format_tags_html(
+            result.get("ips", {}),
+            translations.get("ips", []),
+            "ips",
+            payload.show_confidence,
+            True,
+        ),
+        "summary_text": summary_text,
+        "metadata": metadata,
+    }
+
+
+IMAGE_INVERSION_JOB_SETTINGS = ImageInversionJobSettings.from_env()
+IMAGE_INVERSION_JOB_API = ImageInversionJobAPI(
+    settings=IMAGE_INVERSION_JOB_SETTINGS,
+    executor=execute_image_inversion_job,
+    inference_slot=INFERENCE_SLOT,
+)
+JOB_API_LIFESPAN = create_job_api_lifespan(IMAGE_INVERSION_JOB_API)
+
+
 if __name__ == "__main__":
     if tagger_instance is None:
         print("CRITICAL: Tagger 未能初始化，应用功能将受限。请检查之前的错误信息。")
-    demo.queue(max_size=8).launch(server_name="0.0.0.0", server_port=7860)
+    demo.queue(max_size=8).launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        app_kwargs={"lifespan": JOB_API_LIFESPAN},
+    )

image_inversion_job_api.py

@@ -0,0 +1,1328 @@
+from __future__ import annotations
+
+import hashlib
+import ipaddress
+import io
+import json
+import logging
+import os
+import secrets
+import socket
+import threading
+import time
+import warnings
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Any, Callable, Literal
+from urllib.parse import SplitResult, urlsplit
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request
+from fastapi import status as http_status
+from fastapi.responses import JSONResponse
+from PIL import Image, ImageOps, UnidentifiedImageError
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+LOGGER = logging.getLogger(__name__)
+DNS_RESOLVER_SLOTS = threading.BoundedSemaphore(4)
+IMAGE_FETCH_SLOTS = threading.BoundedSemaphore(4)
+MIN_IMAGE_DIMENSION = 32
+DEFAULT_MAX_IMAGE_BYTES = 8 * 1024 * 1024
+MAX_CONFIGURABLE_IMAGE_BYTES = 20 * 1024 * 1024
+DEFAULT_MAX_IMAGE_PIXELS = 4_194_304
+MAX_CONFIGURABLE_IMAGE_PIXELS = 20_000_000
+DEFAULT_MAX_IMAGE_DIMENSION = 4096
+MAX_CONFIGURABLE_IMAGE_DIMENSION = 8192
+DEFAULT_MAX_RECORDS = 256
+MIN_MAX_RECORDS = 8
+MAX_MAX_RECORDS = 4096
+
+JobStatus = Literal["queued", "running", "succeeded", "failed"]
+SummarySeparator = Literal["comma", "newline", "space"]
+AnalysisExecutor = Callable[
+    ["ImageInversionJobRequest", Image.Image, str],
+    dict[str, Any],
+]
+ImageFetcher = Callable[[str, "ImageInversionJobSettings"], Image.Image]
+
+
+def _read_int_env(name: str, default: int) -> int:
+    """读取整数环境变量。
+
+    Args:
+        name: 环境变量名称。
+        default: 变量为空时采用的默认值。
+
+    Returns:
+        解析完成的整数。
+    """
+    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:
+    """读取有限浮点环境变量。
+
+    Args:
+        name: 环境变量名称。
+        default: 变量为空时采用的默认值。
+
+    Returns:
+        解析完成的有限浮点数。
+    """
+    raw_value = os.getenv(name)
+    if raw_value is None or not raw_value.strip():
+        return default
+    try:
+        parsed_value = float(raw_value)
+    except ValueError as exc:
+        raise RuntimeError(f"{name} must be a number") from exc
+    if not float("-inf") < parsed_value < float("inf"):
+        raise RuntimeError(f"{name} must be finite")
+    return parsed_value
+
+
+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 ImageInversionJobSettings:
+    """保存 Image Inversion 自定义任务 API 的部署边界与资源限制。"""
+
+    api_key: str
+    allowed_hosts: frozenset[str]
+    result_ttl_seconds: int = 1800
+    poll_after_seconds: int = 2
+    fetch_timeout_seconds: float = 15.0
+    max_image_bytes: int = DEFAULT_MAX_IMAGE_BYTES
+    max_image_pixels: int = DEFAULT_MAX_IMAGE_PIXELS
+    max_image_dimension: int = DEFAULT_MAX_IMAGE_DIMENSION
+    max_records: int = DEFAULT_MAX_RECORDS
+    space_host: str = ""
+
+    @classmethod
+    def from_env(cls) -> "ImageInversionJobSettings":
+        """从 Space Secret 与 Variables 读取并校验任务 API 配置。
+
+        Args:
+            此方法不接收参数，配置统一从当前进程环境读取。
+
+        Returns:
+            已完成密钥、网络及图片资源边界校验的设置。
+        """
+        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",
+            DEFAULT_MAX_IMAGE_BYTES,
+        )
+        max_image_pixels = _read_int_env(
+            "JOB_IMAGE_MAX_PIXELS",
+            DEFAULT_MAX_IMAGE_PIXELS,
+        )
+        max_image_dimension = _read_int_env(
+            "JOB_IMAGE_MAX_DIMENSION",
+            DEFAULT_MAX_IMAGE_DIMENSION,
+        )
+        max_records = _read_int_env("JOB_MAX_RECORDS", DEFAULT_MAX_RECORDS)
+
+        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 <= MAX_CONFIGURABLE_IMAGE_BYTES:
+            raise RuntimeError("JOB_IMAGE_MAX_BYTES must be between 1 KiB and 20 MiB")
+        if not 1 <= max_image_pixels <= MAX_CONFIGURABLE_IMAGE_PIXELS:
+            raise RuntimeError(
+                "JOB_IMAGE_MAX_PIXELS must be between 1 and 20000000"
+            )
+        if not MIN_IMAGE_DIMENSION <= max_image_dimension <= MAX_CONFIGURABLE_IMAGE_DIMENSION:
+            raise RuntimeError(
+                "JOB_IMAGE_MAX_DIMENSION must be between 32 and 8192"
+            )
+        if not MIN_MAX_RECORDS <= max_records <= MAX_MAX_RECORDS:
+            raise RuntimeError("JOB_MAX_RECORDS must be between 8 and 4096")
+
+        return cls(
+            api_key=api_key,
+            allowed_hosts=normalize_allowed_hosts(
+                os.getenv("JOB_IMAGE_ALLOWED_HOSTS", "")
+            ),
+            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,
+            max_image_dimension=max_image_dimension,
+            max_records=max_records,
+            space_host=os.getenv("SPACE_HOST", "").strip().rstrip("/"),
+        )
+
+
+class ImageInversionJobRequest(BaseModel):
+    """定义 Image Inversion 标签分析任务的公开请求参数。"""
+
+    model_config = ConfigDict(
+        extra="forbid",
+        str_strip_whitespace=True,
+        allow_inf_nan=False,
+    )
+
+    input_image_url: str = Field(min_length=1, max_length=4096)
+    general_threshold: float = Field(default=0.35, ge=0.0, le=1.0)
+    character_threshold: float = Field(default=0.85, ge=0.0, le=1.0)
+    show_confidence: bool = True
+    show_general: bool = True
+    show_character: bool = True
+    show_ip: bool = True
+    separator: SummarySeparator = "comma"
+    show_chinese: bool = True
+
+    @field_validator("input_image_url")
+    @classmethod
+    def require_https_image_url(cls, value: str) -> str:
+        """在 Schema 阶段拒绝非 HTTPS、凭据、片段和非标准端口。
+
+        Args:
+            value: 调用方提交的远程图片 URL。
+
+        Returns:
+            通过基础 HTTPS 语法检查的原始 URL。
+        """
+        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) -> None:
+        """创建携带脱敏公开信息的图片来源错误。
+
+        Args:
+            status_code: API 应返回的 HTTP 状态码。
+            public_message: 不包含 URL、凭据或底层网络细节的公开文案。
+
+        Returns:
+            此初始化方法不返回数据。
+        """
+        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]:
+    """在硬期限内解析白名单主机的地址。
+
+    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="image-inversion-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: ImageInversionJobSettings,
+) -> 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: ImageInversionJobSettings,
+) -> Image.Image:
+    """校验远程内容并解码为脱离底层流的 RGB 图片。
+
+    Args:
+        image_bytes: 在字节上限内完整读取的图片内容。
+        content_type: 远程响应声明的 MIME 类型，仅为接口兼容保留。
+        settings: 图片字节、像素及单边尺寸限制。
+
+    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.",
+        )
+
+    # 远端可合法返回 application/octet-stream；只信任 Pillow 识别的真实格式。
+    _ = content_type
+    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 < MIN_IMAGE_DIMENSION or height < MIN_IMAGE_DIMENSION:
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "The remote image dimensions are not supported.",
+                    )
+                if (
+                    width > settings.max_image_dimension
+                    or height > settings.max_image_dimension
+                    or width * height > settings.max_image_pixels
+                ):
+                    raise ImageSourceError(
+                        http_status.HTTP_400_BAD_REQUEST,
+                        "The remote image exceeds the configured dimension 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 _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_inner(
+    url: str,
+    settings: ImageInversionJobSettings,
+) -> Image.Image:
+    """从白名单公网主机流式抓取并安全解码图片。
+
+    Args:
+        url: 调用方提交的图片 HTTPS URL。
+        settings: 图片白名单、超时、体积、像素与尺寸边界。
+
+    Returns:
+        可直接交给标签模型的 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 fetch_remote_image(
+    url: str,
+    settings: ImageInversionJobSettings,
+) -> Image.Image:
+    """以有界后台抓取器执行下载并强制墙钟总期限。
+
+    Args:
+        url: 调用方提交的图片 HTTPS URL。
+        settings: 白名单、硬超时、字节、像素与尺寸边界。
+
+    Returns:
+        在总期限内完成验证、EXIF 转正和 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:
+                fetched_image.close()
+                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="image-inversion-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:
+    """以字节形式比较不可信文本。
+
+    Args:
+        supplied: 请求携带的待校验文本。
+        expected: 服务端保存的期望文本。
+
+    Returns:
+        两个 UTF-8 字节序列完全一致时返回 True。
+    """
+    return secrets.compare_digest(
+        supplied.encode("utf-8", errors="surrogatepass"),
+        expected.encode("utf-8", errors="surrogatepass"),
+    )
+
+
+def fingerprint_request(payload: ImageInversionJobRequest) -> 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 ImageInversionJobRecord:
+    """保存单个标签分析任务在当前 Space 进程内的生命周期数据。"""
+
+    job_id: str
+    payload: ImageInversionJobRequest
+    request_fingerprint: str
+    input_image: Image.Image | None
+    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: dict[str, Any] | None = None
+    error_type: str | None = None
+
+
+class ImageInversionJobAPI:
+    """实现 Image Inversion 异步 JSON 任务 API 状态机。"""
+
+    def __init__(
+        self,
+        settings: ImageInversionJobSettings,
+        executor: AnalysisExecutor,
+        inference_slot: threading.Lock,
+        image_fetcher: ImageFetcher = fetch_remote_image,
+    ) -> None:
+        """创建任务服务并注册 Image Inversion 自定义路由。
+
+        Args:
+            settings: 已校验的部署配置和资源限制。
+            executor: 复用现有标签分析链路的执行适配函数。
+            inference_slot: 与 UI 共用的非阻塞单推理门闩。
+            image_fetcher: 可替换的安全远程图片抓取器。
+
+        Returns:
+            此初始化方法不返回数据。
+        """
+        self.settings = settings
+        self.executor = executor
+        self.inference_slot = inference_slot
+        self.image_fetcher = image_fetcher
+        self.jobs: dict[str, ImageInversionJobRecord] = {}
+        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="image_inversion_create_job",
+        )
+        self.router.add_api_route(
+            "/api/jobs/{job_id}",
+            self.get_job_status,
+            methods=["GET"],
+            name="image_inversion_get_job_status",
+        )
+
+    def install_on_app(self, app: Any) -> None:
+        """把自定义任务路由安装到 Gradio FastAPI 应用前部。
+
+        Args:
+            app: Gradio 在 launch 阶段创建的 FastAPI 应用。
+
+        Returns:
+            路由已存在或安装完成后不返回数据。
+        """
+        if getattr(app.state, "image_inversion_job_api_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.image_inversion_job_api_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: ImageInversionJobRecord,
+        now: float | None = None,
+    ) -> bool:
+        current_time = time.time() if now is None else now
+        if record.completed_at is None:
+            return False
+        return current_time - record.completed_at > self.settings.result_ttl_seconds
+
+    @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 an Image Inversion job image: %s",
+                        type(exc).__name__,
+                    )
+
+    @staticmethod
+    def _close_job_image(record: ImageInversionJobRecord | None) -> None:
+        if record is None:
+            return
+        ImageInversionJobAPI._close_images(record.input_image)
+        record.input_image = None
+
+    def _remove_record_locked(self, record: ImageInversionJobRecord) -> 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_expired_jobs(self) -> None:
+        """清理当前所有过期终态任务及其输入图片引用。
+
+        Args:
+            此方法不接收参数。
+
+        Returns:
+            清理完成后不返回数据。
+        """
+        expired_records: list[ImageInversionJobRecord] = []
+        now = time.time()
+        with self.jobs_lock:
+            for record in tuple(self.jobs.values()):
+                if self._is_expired(record, now):
+                    self._remove_record_locked(record)
+                    expired_records.append(record)
+        for expired_record in expired_records:
+            self._close_job_image(expired_record)
+
+    def _ensure_capacity_for_new_record(self) -> None:
+        """为一个新任务腾出有限记录容量。
+
+        Args:
+            此方法不接收���数。
+
+        Returns:
+            容量已经可用时不返回数据；无法安全腾出容量时抛出 HTTP 503。
+        """
+        evicted_records: list[ImageInversionJobRecord] = []
+        with self.jobs_lock:
+            now = time.time()
+            for record in tuple(self.jobs.values()):
+                if self._is_expired(record, now):
+                    self._remove_record_locked(record)
+                    evicted_records.append(record)
+
+            while len(self.jobs) >= self.settings.max_records:
+                terminal_records = [
+                    record
+                    for record in self.jobs.values()
+                    if record.status in {"succeeded", "failed"}
+                    and record.completed_at is not None
+                ]
+                if not terminal_records:
+                    raise HTTPException(
+                        status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE,
+                        detail="The prediction job store is full.",
+                        headers={"Retry-After": "5"},
+                    )
+                # 仅淘汰最老终态；绝不删除 queued/running 任务。
+                oldest_terminal = min(
+                    terminal_records,
+                    key=lambda record: record.completed_at or record.created_at,
+                )
+                self._remove_record_locked(oldest_terminal)
+                evicted_records.append(oldest_terminal)
+
+        for evicted_record in evicted_records:
+            self._close_job_image(evicted_record)
+
+    def _job_or_404(self, job_id: str) -> ImageInversionJobRecord:
+        expired_record: ImageInversionJobRecord | 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
+        self._close_job_image(expired_record)
+        if record is None:
+            raise HTTPException(
+                status_code=http_status.HTTP_404_NOT_FOUND,
+                detail="Job not found or result expired.",
+            )
+        return record
+
+    def _status_response(
+        self,
+        request: Request,
+        record: ImageInversionJobRecord,
+    ) -> JSONResponse:
+        status_url = self._public_route_url(
+            request,
+            "image_inversion_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,
+    ) -> ImageInversionJobRecord | None:
+        if idempotency_key is None:
+            return 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)
+                record = None
+            if record is None:
+                if existing_job_id is not None:
+                    self.idempotency_jobs.pop(idempotency_key, None)
+                return None
+            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
+
+    def create_job(
+        self,
+        payload: ImageInversionJobRequest,
+        request: Request,
+    ) -> JSONResponse:
+        """验证远程图片并创建立即执行的异步标签分析任务。
+
+        Args:
+            payload: 已通过 Pydantic 校验的命名分析参数。
+            request: 包含共享密钥的 FastAPI 请求。
+
+        Returns:
+            HTTP 202 任务标识、状态 URL 与建议轮询间隔。
+        """
+        self._require_api_key(request)
+        self.remove_expired_jobs()
+
+        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 = self._lookup_idempotent_job(
+            idempotency_key,
+            request_fingerprint,
+        )
+        if existing_record is not None:
+            return self._status_response(request, existing_record)
+        self._ensure_capacity_for_new_record()
+
+        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 prediction service is busy.",
+                headers={"Retry-After": "5"},
+            )
+
+        slot_owned_by_request = False
+        record: ImageInversionJobRecord | None = None
+        replay_record: ImageInversionJobRecord | None = None
+        input_image: Image.Image | None = None
+        try:
+            try:
+                input_image = self.image_fetcher(
+                    payload.input_image_url,
+                    self.settings,
+                )
+            except ImageSourceError as exc:
+                raise HTTPException(
+                    status_code=exc.status_code,
+                    detail=exc.public_message,
+                ) from exc
+
+            with self.jobs_lock:
+                # 相同幂等键可并发抓图；只有二次判定有权创建唯一任务。
+                replay_record = self._lookup_idempotent_job(
+                    idempotency_key,
+                    request_fingerprint,
+                )
+                if replay_record is None:
+                    self._ensure_capacity_for_new_record()
+                    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 prediction 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 prediction service is busy.",
+                            headers={"Retry-After": "5"},
+                        )
+                    slot_owned_by_request = True
+
+                    job_id = secrets.token_urlsafe(24)
+                    record = ImageInversionJobRecord(
+                        job_id=job_id,
+                        payload=payload,
+                        request_fingerprint=request_fingerprint,
+                        input_image=input_image,
+                        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
+
+            if replay_record is not None:
+                self._close_images(input_image)
+                input_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"image-inversion-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)
+                if record is not None and self.active_job_id == record.job_id:
+                    self.active_job_id = None
+            self._close_job_image(record)
+            if record is None:
+                self._close_images(input_image)
+            if slot_owned_by_request:
+                self.inference_slot.release()
+            raise
+
+    def _execute_job(self, record: ImageInversionJobRecord) -> 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.")
+            result = self.executor(
+                record.payload,
+                record.input_image,
+                record.job_id,
+            )
+            if not isinstance(result, dict):
+                raise RuntimeError("The prediction executor did not return an object.")
+            # 任务结果必须在后台线程内完成 JSON 校验，避免轮询阶段才暴露序列化错误。
+            normalized_result = json.loads(
+                json.dumps(result, ensure_ascii=False, allow_nan=False)
+            )
+
+            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 = normalized_result
+                    record.completed_at = time.time()
+        except Exception as exc:
+            LOGGER.error(
+                "Image Inversion job %s failed with %s",
+                record.job_id[:8],
+                type(exc).__name__,
+            )
+            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:
+                if self.active_job_id == record.job_id:
+                    self.active_job_id = None
+            self._close_job_image(record)
+            self.inference_slot.release()
+            self.remove_expired_jobs()
+
+    def get_job_status(self, job_id: str, request: Request) -> JSONResponse:
+        """读取标签分析任务状态并在成功后返回 JSON 结果。
+
+        Args:
+            job_id: POST 创建任务时返回的自定义任务标识。
+            request: 用于鉴权的 FastAPI 请求。
+
+        Returns:
+            运行中返回 202，成功返回 result，失败返回脱敏 500。
+        """
+        self._require_api_key(request)
+        record = self._job_or_404(job_id)
+        with self.jobs_lock:
+            status_value = record.status
+            result_value = record.result
+
+        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": "PREDICTION_FAILED"}},
+            )
+        if result_value is None:
+            raise RuntimeError("A succeeded prediction job has no result.")
+        return JSONResponse(
+            content=result_value,
+            headers={"Cache-Control": "no-store"},
+        )
+
+
+def create_job_api_lifespan(job_api: ImageInversionJobAPI):
+    """创建供 Gradio launch 组合使用的自定义路由 lifespan。
+
+    Args:
+        job_api: 已配置执行器和共享推理锁的标签分析任务服务。
+
+    Returns:
+        可传给 Gradio 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

requirements.txt

@@ -1,4 +1,5 @@
-gradio>=4.31.0
+gradio==5.29.0
+fastapi>=0.115.2,<1
 huggingface_hub>=0.24.0
 numpy
 pandas
@@ -7,4 +8,6 @@ onnxruntime
 torch
 torchvision
 timm
-requests
+requests
+httpx==0.28.1
+pydantic>=2,<2.12

tests/test_image_inversion_job_api.py

@@ -0,0 +1,772 @@
+from __future__ import annotations
+
+import ast
+import io
+import logging
+import os
+import socket
+import tempfile
+import threading
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest.mock import patch
+
+import httpx
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from PIL import Image
+from pydantic import ValidationError
+
+from image_inversion_job_api import (
+    ImageInversionJobAPI,
+    ImageInversionJobRecord,
+    ImageInversionJobRequest,
+    ImageInversionJobSettings,
+    ImageSourceError,
+    create_job_api_lifespan,
+    decode_image_bytes,
+    fetch_remote_image,
+    fingerprint_request,
+    validate_image_url,
+)
+
+
+API_KEY = "k" * 48
+EXPECTED_RESULT = {
+    "status_markdown": "done",
+    "general_tags_html": "<p>general</p>",
+    "character_tags_html": "<p>character</p>",
+    "ip_tags_html": "<p>ip</p>",
+    "summary_text": "tag one, tag two",
+    "metadata": {"device": "cpu", "latency_s_total": 0.1},
+}
+
+
+def _make_png_bytes(size: tuple[int, int] = (96, 80)) -> bytes:
+    """生成远程抓图测试使用的有效 PNG。
+
+    Args:
+        size: PNG 的宽高。
+
+    Returns:
+        完整 PNG 文件字节。
+    """
+    image_buffer = io.BytesIO()
+    with Image.new("RGB", size, "green") as image:
+        image.save(image_buffer, format="PNG")
+    return image_buffer.getvalue()
+
+
+PNG_BYTES = _make_png_bytes()
+
+
+class _TrackedImage:
+    """记录任务输入图是否被 API 生命周期可靠关闭。"""
+
+    def __init__(self) -> None:
+        self.closed = False
+
+    def close(self) -> None:
+        """标记当前测试图片已经关闭。"""
+        self.closed = True
+
+
+class _FakeStreamResponse:
+    """提供 httpx 流式响应所需的最小测试替身。"""
+
+    def __init__(
+        self,
+        status_code: int,
+        body: bytes,
+        *,
+        peer_address: str = "93.184.216.34",
+        extra_headers: dict[str, str] | None = None,
+    ) -> None:
+        self.status_code = status_code
+        self.body = body
+        self.headers = {
+            "content-type": "image/png",
+            "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:
+        """退出响应上下文且不吞掉异常。"""
+        del exc_type, exc, traceback
+        return False
+
+    def iter_bytes(self):
+        """以单个分块返回完整响应体。"""
+        yield self.body
+
+
+class ImageInversionJobAPITest(unittest.TestCase):
+    """验证 Image Inversion 自定义任务 API 的公开契约与安全边界。"""
+
+    def setUp(self) -> None:
+        """登记每个测试需要在结束时关闭的客户端和线程事件。"""
+        self.clients: list[TestClient] = []
+        self.release_events: list[threading.Event] = []
+
+    def tearDown(self) -> None:
+        """释放测试线程和 TestClient，不创建持久化测试产物。"""
+        for release_event in self.release_events:
+            release_event.set()
+        time.sleep(0.02)
+        for client in self.clients:
+            client.close()
+
+    @staticmethod
+    def _settings(**overrides) -> ImageInversionJobSettings:
+        """构造不依赖进程环境的隔离任务设置。
+
+        Args:
+            overrides: 要覆盖的 ImageInversionJobSettings 字段。
+
+        Returns:
+            可供单元测试直接注入的设置。
+        """
+        values = {
+            "api_key": API_KEY,
+            "allowed_hosts": frozenset({"allowed.example"}),
+            "result_ttl_seconds": 1800,
+            "poll_after_seconds": 1,
+            "fetch_timeout_seconds": 15.0,
+            "max_image_bytes": 8 * 1024 * 1024,
+            "max_image_pixels": 4_194_304,
+            "max_image_dimension": 4096,
+            "max_records": 256,
+            "space_host": "https://space.example",
+        }
+        values.update(overrides)
+        return ImageInversionJobSettings(**values)
+
+    @staticmethod
+    def _payload(**overrides) -> dict:
+        """生成可直接提交给创建接口的合法请求体。
+
+        Args:
+            overrides: 要覆盖或增加的 JSON 字段。
+
+        Returns:
+            包含远程输入图和全部展示参数的请求字典。
+        """
+        payload = {
+            "input_image_url": "https://allowed.example/input.png?source=private",
+            "general_threshold": 0.35,
+            "character_threshold": 0.85,
+            "show_confidence": True,
+            "show_general": True,
+            "show_character": True,
+            "show_ip": True,
+            "separator": "comma",
+            "show_chinese": True,
+        }
+        payload.update(overrides)
+        return payload
+
+    @staticmethod
+    def _headers(**overrides) -> dict[str, str]:
+        """生成包含共享密钥的请求头。
+
+        Args:
+            overrides: 要覆盖或追加的请求头。
+
+        Returns:
+            可传给 TestClient 的请求头字典。
+        """
+        headers = {"X-API-Key": API_KEY}
+        headers.update(overrides)
+        return headers
+
+    def _make_service(
+        self,
+        *,
+        executor=None,
+        image_fetcher=None,
+        inference_slot: threading.Lock | None = None,
+        settings: ImageInversionJobSettings | None = None,
+        add_catch_all: bool = False,
+    ) -> tuple[TestClient, ImageInversionJobAPI, list[_TrackedImage], dict[str, int]]:
+        """创建完全隔离且不导入模型的 FastAPI 测试服务。
+
+        Args:
+            executor: 可选假标签执行器。
+            image_fetcher: 可选假远程图片抓取器。
+            inference_slot: 可选 UI/API 共享锁。
+            settings: 可选任务设置。
+            add_catch_all: 是否先注册模拟 Gradio 的宽泛路由。
+
+        Returns:
+            TestClient、API 对象、输入图片列表与调用计数器。
+        """
+        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, job_id):
+                """返回与新接口契约一致的六字段 JSON 结果。"""
+                del payload, input_image, job_id
+                with counter_lock:
+                    counters["execute"] += 1
+                return EXPECTED_RESULT
+
+        api = ImageInversionJobAPI(
+            settings=settings,
+            executor=executor,
+            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"])
+            def catch_all(path: str):
+                """模拟可能抢先匹配 API 请求的 Gradio 宽泛路由。"""
+                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: 当前隔离服务客户端。
+            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_cpu_contract_and_route_precedence(self) -> None:
+        """验证鉴权、CPU 请求契约及自定义路由优先级。"""
+        client, api, _, counters = self._make_service(add_catch_all=True)
+        self.assertTrue(callable(create_job_api_lifespan(api)))
+
+        unauthorized = client.post("/api/jobs", json=self._payload())
+        self.assertEqual(unauthorized.status_code, 401)
+        self.assertEqual(counters["fetch"], 0)
+
+        unknown_field = client.post(
+            "/api/jobs",
+            json=self._payload(unknown=True),
+            headers=self._headers(),
+        )
+        self.assertEqual(unknown_field.status_code, 422)
+        self.assertEqual(counters["fetch"], 0)
+
+        # cpu-basic Space 不依赖 ZeroGPU 的 x-ip-token。
+        accepted = client.post(
+            "/api/jobs",
+            json=self._payload(),
+            headers=self._headers(),
+        )
+        self.assertEqual(accepted.status_code, 202)
+        self.assertNotIn("catch_all", accepted.json())
+        terminal = self._wait_for_terminal(client, accepted.json()["job_id"])
+        self.assertEqual(terminal.status_code, 200)
+        self.assertEqual(terminal.json(), EXPECTED_RESULT)
+
+    def test_success_contract_does_not_retain_caller_credentials(self) -> None:
+        """验证成功链路仅向执行器传业务参数且结果无包装层。"""
+        seen: dict = {}
+
+        def executor(payload, input_image, job_id):
+            """记录后台适配器参数并返回固定六字段结果。"""
+            seen.update({"payload": payload, "input_image": input_image, "job_id": job_id})
+            return EXPECTED_RESULT
+
+        client, api, images, _ = self._make_service(executor=executor)
+        created = client.post(
+            "/api/jobs",
+            json=self._payload(separator="newline"),
+            headers=self._headers(
+                Authorization="Bearer must-not-be-retained",
+                Cookie="must-not-be-retained",
+                **{"X-IP-Token": "not-needed-on-cpu"},
+            ),
+        )
+        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(set(terminal.json()), set(EXPECTED_RESULT))
+        self.assertEqual(seen["job_id"], job_id)
+        self.assertEqual(seen["payload"].separator, "newline")
+        self.assertTrue(all(image.closed for image in images))
+        self.assertFalse(hasattr(api.jobs[job_id], "zero_gpu_headers"))
+
+    def test_idempotent_replay_conflict_and_busy_submission(self) -> None:
+        """验证幂等重放、冲突及单任务繁忙状态。"""
+        release_event = threading.Event()
+        started_event = threading.Event()
+        self.release_events.append(release_event)
+        execution_count = 0
+        count_lock = threading.Lock()
+
+        def blocking_executor(payload, input_image, job_id):
+            """阻塞首个任务，使测试可观察幂等和忙状态。"""
+            nonlocal execution_count
+            del payload, input_image, job_id
+            with count_lock:
+                execution_count += 1
+            started_event.set()
+            release_event.wait(timeout=3)
+            return EXPECTED_RESULT
+
+        client, _, _, counters = self._make_service(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(general_threshold=0.5),
+            headers=headers,
+        )
+        self.assertEqual(conflict.status_code, 409)
+
+        busy = client.post(
+            "/api/jobs",
+            json=self._payload(character_threshold=0.7),
+            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_executes_once(self) -> None:
+        """验证并发同键请求最终只执行一次。"""
+        fetch_barrier = threading.Barrier(2)
+        fetch_count = 0
+        execute_count = 0
+        count_lock = threading.Lock()
+
+        def racing_fetcher(url, settings):
+            """让两个请求都完成第一次幂等检查后再返回图片。"""
+            nonlocal fetch_count
+            del url, settings
+            with count_lock:
+                fetch_count += 1
+            fetch_barrier.wait(timeout=2)
+            return _TrackedImage()
+
+        def counting_executor(payload, input_image, job_id):
+            """统计实际执行次数并返回固定结果。"""
+            nonlocal execute_count
+            del payload, input_image, job_id
+            with count_lock:
+                execute_count += 1
+            return EXPECTED_RESULT
+
+        client, _, _, _ = self._make_service(
+            image_fetcher=racing_fetcher,
+            executor=counting_executor,
+        )
+        headers = self._headers(**{"Idempotency-Key": "concurrent-key"})
+
+        def submit_job(_index: int):
+            """提交一个并发幂等请求。"""
+            return client.post("/api/jobs", json=self._payload(), headers=headers)
+
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            responses = list(pool.map(submit_job, 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_releases_image_and_slot(self) -> None:
+        """验证内部异常不泄密，且图片与共享锁始终释放。"""
+        shared_slot = threading.Lock()
+
+        def failing_executor(payload, input_image, job_id):
+            """模拟包含敏感内部信息的预测错误。"""
+            del payload, input_image, job_id
+            raise RuntimeError("INTERNAL-SECRET-STACK-DATA")
+
+        client, api, images, _ = self._make_service(
+            executor=failing_executor,
+            inference_slot=shared_slot,
+        )
+        with self.assertLogs("image_inversion_job_api", level=logging.ERROR) as logs:
+            created = client.post(
+                "/api/jobs",
+                json=self._payload(),
+                headers=self._headers(Authorization="Bearer sensitive"),
+            )
+            terminal = self._wait_for_terminal(client, created.json()["job_id"])
+
+        self.assertEqual(terminal.status_code, 500)
+        self.assertEqual(terminal.json(), {"error": {"code": "PREDICTION_FAILED"}})
+        combined_logs = "\n".join(logs.output)
+        self.assertNotIn("INTERNAL-SECRET-STACK-DATA", combined_logs)
+        self.assertNotIn(API_KEY, combined_logs)
+        self.assertTrue(all(image.closed for image in images))
+        self.assertTrue(shared_slot.acquire(blocking=False))
+        shared_slot.release()
+        self.assertEqual(api.jobs[created.json()["job_id"]].error_type, "RuntimeError")
+
+    def test_thread_start_failure_rolls_back_and_retry_succeeds(self) -> None:
+        """验证后台线程启动失败时任务、幂等索引和锁均回滚。"""
+        client, api, images, counters = self._make_service()
+        with patch(
+            "image_inversion_job_api.threading.Thread.start",
+            side_effect=RuntimeError("thread-start-failed"),
+        ):
+            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_ttl_removes_all_expired_terminal_records(self) -> None:
+        """验证 TTL 清理会移除全部已过期终态记录。"""
+        settings = self._settings(result_ttl_seconds=60)
+        client, api, _, _ = self._make_service(settings=settings)
+        job_ids: list[str] = []
+        for threshold in (0.2, 0.3):
+            created = client.post(
+                "/api/jobs",
+                json=self._payload(general_threshold=threshold),
+                headers=self._headers(),
+            )
+            job_id = created.json()["job_id"]
+            self.assertEqual(self._wait_for_terminal(client, job_id).status_code, 200)
+            api.jobs[job_id].completed_at = time.time() - 100
+            job_ids.append(job_id)
+
+        api.remove_expired_jobs()
+        self.assertEqual(api.jobs, {})
+        self.assertEqual(len(job_ids), 2)
+
+    def test_capacity_evicts_oldest_terminal_and_preserves_running_records(self) -> None:
+        """验证容量上限只淘汰最老终态且满是运行态时返回 503。"""
+        settings = self._settings(max_records=2)
+        client, api, _, _ = self._make_service(settings=settings)
+        completed_job_ids: list[str] = []
+        for threshold in (0.2, 0.3):
+            created = client.post(
+                "/api/jobs",
+                json=self._payload(general_threshold=threshold),
+                headers=self._headers(),
+            )
+            job_id = created.json()["job_id"]
+            self.assertEqual(self._wait_for_terminal(client, job_id).status_code, 200)
+            completed_job_ids.append(job_id)
+            time.sleep(0.01)
+
+        third = client.post(
+            "/api/jobs",
+            json=self._payload(general_threshold=0.4),
+            headers=self._headers(),
+        )
+        self.assertEqual(third.status_code, 202)
+        self.assertNotIn(completed_job_ids[0], api.jobs)
+        self.assertIn(completed_job_ids[1], api.jobs)
+        self.assertLessEqual(len(api.jobs), 2)
+        self.assertEqual(self._wait_for_terminal(client, third.json()["job_id"]).status_code, 200)
+
+        full_settings = self._settings(max_records=1)
+        full_client, full_api, _, counters = self._make_service(settings=full_settings)
+        running_record = ImageInversionJobRecord(
+            job_id="running-record",
+            payload=ImageInversionJobRequest(**self._payload()),
+            request_fingerprint="fingerprint",
+            input_image=_TrackedImage(),
+            status="running",
+        )
+        full_api.jobs[running_record.job_id] = running_record
+        blocked = full_client.post(
+            "/api/jobs",
+            json=self._payload(general_threshold=0.6),
+            headers=self._headers(),
+        )
+        self.assertEqual(blocked.status_code, 503)
+        self.assertEqual(blocked.headers["Retry-After"], "5")
+        self.assertIn(running_record.job_id, full_api.jobs)
+        self.assertEqual(counters["fetch"], 0)
+
+    def test_settings_schema_and_fingerprint(self) -> None:
+        """验证环境设置、公开 Schema 和请求指纹边界。"""
+        with patch.dict(
+            os.environ,
+            {
+                "JOB_API_KEY": API_KEY,
+                "JOB_IMAGE_ALLOWED_HOSTS": "Allowed.Example,cdn.example",
+            },
+            clear=True,
+        ):
+            settings = ImageInversionJobSettings.from_env()
+        self.assertEqual(
+            settings.allowed_hosts,
+            frozenset({"allowed.example", "cdn.example"}),
+        )
+        self.assertEqual(settings.max_image_bytes, 8 * 1024 * 1024)
+        self.assertEqual(settings.max_records, 256)
+
+        for invalid_max_records in (7, 4097):
+            with self.subTest(max_records=invalid_max_records), patch.dict(
+                os.environ,
+                {
+                    "JOB_API_KEY": API_KEY,
+                    "JOB_IMAGE_ALLOWED_HOSTS": "allowed.example",
+                    "JOB_MAX_RECORDS": str(invalid_max_records),
+                },
+                clear=True,
+            ):
+                with self.assertRaises(RuntimeError):
+                    ImageInversionJobSettings.from_env()
+
+        minimum = ImageInversionJobRequest(
+            input_image_url="https://allowed.example/input.png",
+            general_threshold=0.0,
+            character_threshold=0.0,
+            separator="space",
+        )
+        maximum = ImageInversionJobRequest(
+            input_image_url="https://allowed.example/input.png",
+            general_threshold=1.0,
+            character_threshold=1.0,
+            separator="newline",
+        )
+        self.assertNotEqual(fingerprint_request(minimum), fingerprint_request(maximum))
+        reordered = ImageInversionJobRequest.model_validate(
+            minimum.model_dump(mode="json")
+        )
+        self.assertEqual(fingerprint_request(minimum), fingerprint_request(reordered))
+
+        for invalid_payload in (
+            self._payload(input_image_url="http://allowed.example/input.png"),
+            self._payload(general_threshold=-0.01),
+            self._payload(character_threshold=1.01),
+            self._payload(general_threshold=float("nan")),
+            self._payload(separator="逗号"),
+            self._payload(unknown=True),
+        ):
+            with self.subTest(payload=invalid_payload):
+                with self.assertRaises(ValidationError):
+                    ImageInversionJobRequest(**invalid_payload)
+
+    def test_url_policy_decode_and_fetch_do_not_leak_headers(self) -> None:
+        """验证白名单、SSRF、图片解码和无凭据抓取边界。"""
+        settings = self._settings()
+        public_record = (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("93.184.216.34", 443),
+        )
+        with patch(
+            "image_inversion_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):
+                        validate_image_url(invalid_url, settings)
+
+        private_record = (
+            socket.AF_INET,
+            socket.SOCK_STREAM,
+            socket.IPPROTO_TCP,
+            "",
+            ("127.0.0.1", 443),
+        )
+        with patch(
+            "image_inversion_job_api.socket.getaddrinfo",
+            return_value=[public_record, private_record],
+        ):
+            with self.assertRaises(ImageSourceError):
+                validate_image_url("https://allowed.example/image.png", settings)
+
+        decoded = decode_image_bytes(PNG_BYTES, "application/octet-stream", settings)
+        self.assertEqual(decoded.mode, "RGB")
+        decoded.close()
+
+        captured: dict = {}
+
+        class FakeClient:
+            """捕获 httpx 客户端配置并返回假响应。"""
+
+            def __init__(self, **kwargs) -> None:
+                captured["client_kwargs"] = kwargs
+
+            def __enter__(self):
+                """进入客户端上下文并返回自身。"""
+                return self
+
+            def __exit__(self, exc_type, exc, traceback) -> bool:
+                """退出客户端上下文且不吞掉异常。"""
+                del exc_type, exc, traceback
+                return False
+
+            def close(self) -> None:
+                """模拟关闭客户端。"""
+                captured["closed"] = True
+
+            def stream(self, method, url, headers):
+                """记录公开请求参数并返回假响应。"""
+                captured["method"] = method
+                captured["url"] = url
+                captured["headers"] = dict(headers)
+                return captured["response"]
+
+        with patch(
+            "image_inversion_job_api.socket.getaddrinfo",
+            return_value=[public_record],
+        ), patch("image_inversion_job_api.httpx.Client", FakeClient):
+            captured["response"] = _FakeStreamResponse(200, PNG_BYTES)
+            fetched = fetch_remote_image(
+                "https://allowed.example/image.png?secret=value",
+                settings,
+            )
+            fetched.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 header_name in ("Authorization", "Cookie", "X-API-Key", "X-IP-Token"):
+                self.assertNotIn(header_name, captured["headers"])
+
+            captured["response"] = _FakeStreamResponse(302, b"")
+            with self.assertRaises(ImageSourceError):
+                fetch_remote_image("https://allowed.example/redirect", settings)
+
+            captured["response"] = _FakeStreamResponse(
+                200,
+                PNG_BYTES,
+                peer_address="127.0.0.1",
+            )
+            with self.assertRaises(ImageSourceError):
+                fetch_remote_image("https://allowed.example/rebound", settings)
+
+    def test_app_wires_job_lifespan_and_disables_private_hf_token(self) -> None:
+        """静态验证应用接线、六字段结果与匿名模型下载设置。"""
+        app_path = Path(__file__).resolve().parents[1] / "app.py"
+        app_source = app_path.read_text(encoding="utf-8")
+        app_tree = ast.parse(app_source)
+
+        executor_function = next(
+            node
+            for node in app_tree.body
+            if isinstance(node, ast.FunctionDef)
+            and node.name == "execute_image_inversion_job"
+        )
+        result_return = next(
+            node
+            for node in ast.walk(executor_function)
+            if isinstance(node, ast.Return) and isinstance(node.value, ast.Dict)
+        )
+        result_keys = {
+            key.value
+            for key in result_return.value.keys
+            if isinstance(key, ast.Constant) and isinstance(key.value, str)
+        }
+        self.assertEqual(result_keys, set(EXPECTED_RESULT))
+        self.assertIn('app_kwargs={"lifespan": JOB_API_LIFESPAN}', app_source)
+        self.assertIn('token=False', app_source)
+        self.assertIn('HF_HUB_DISABLE_IMPLICIT_TOKEN', app_source)
+        self.assertNotIn('os.environ.get("HF_TOKEN")', app_source)
+
+
+if __name__ == "__main__":
+    unittest.main()