Unify image references and guided UI/API workflows

Sync GitHub commit 07374badda564e1a143363df710bb35476fcf6e3; publish only the image input and API workflow changes.

README.md

@@ -200,11 +200,11 @@ Fluxus 与 ImageGen 共享同一 Git 历史，Fluxus 当前版本只调整了品
 
 ## 支持能力
 
-- 任务：文生图、图生图、局部重绘、扩图、高清修复。
+- 任务：文生图、参考图编辑 / 融合、图生图（整体重绘）、局部重绘、扩图、高清修复。
 - 模型：由 `yaml/model_list.yaml` 驱动，当前包含 30 类架构、92 个显示模型。
 - 扩展：LoRA、ControlNet、IP-Adapter、Embedding、区域提示、多图编辑、VAE、PiD 等。
 - 输出：PNG 内写入生成参数和完整 ComfyUI 工作流元数据。
-- 中文 Prompt：默认原样传递，不做隐式翻译；模型说明会区分中文自然语言和英文标签型模型。
+- 中文 Prompt：不做隐式翻译或扩写；只统一显式图片编号，模型说明会区分中文自然语言和英文标签型模型。
 
 ## 批量、多图与模型 PK
 
@@ -219,7 +219,11 @@ Fluxus 与 ImageGen 共享同一 Git 历史，Fluxus 当前版本只调整了品
 | 多图融合 | 多张图共同作为一组参考 | 由兼容的编辑模型融合为 1–4 张结果 |
 
 - 多图独立目前支持图生图、扩图和高清修复；局部重绘需要逐张图片配对蒙版，暂不做批量。
-- 多图融合内部固定走“文生图 + 模型专属参考链”，不会与图生图 base latent 混用。Prompt 可按“参考图 1 / 2 / 3”说明各图角色。
+- 按指令改图请选择“参考图编辑 / 融合”：一张图编辑，多张图整组参考。内部走 `txt2img + 模型专属参考链`，不与整体重绘的源图 latent 混用；兼容旧的“文生图 + 多图融合”选择。
+- 图片列表按显示顺序从 1 编号，可拖动排序；有效图片预览中的 `图1 = img1 = image1` 指向同一张图，送入模型时统一为 `image 1`。空槽位不编号；重排/删除后编号随预览更新。编号只是输入对应关系，不保证每种模型都能精确执行空间或身份指令。
+- 提示词中的 `图1`、`图 1`、`参考图1`、`图片1`、`img1`、`IMG1`、`image 1` 均可使用；不存在的编号会在模型下载 / GPU 生成前报错。双引号、中文引号或反引号内的编号视为需要显示的文字，不解析；文件名和标识符不替换。
+- 高层图片列表是当前任务的唯一图片来源，不合并下方旧参考槽位。切换任务/模型会保留隐藏素材供切回，但隐藏或不兼容的原生参考图不参加生成。普通文生图仍兼容下方原生参考槽位；多个原生参考链同时使用时不能用统一编号引用，推荐高层入口。
+- 多图独立 / 图片×模型共用一条 Prompt，请写“当前图片”，不能用 `img2` 跨图引用；需要两图同时参与时改用参考图编辑 / 融合。超出张数限制明确报错，不静默丢图。
 - 多图融合是生成式参考，不保证无损拼接、角色逐像素保留或确定性元素替换。
 - PK 默认采用各模型推荐采样参数；关闭后才严格复用当前步数、CFG、采样器和调度器。同一 Seed 在不同架构之间只是尽量控制变量，不表示初始噪声数学等价。
 - PK V1 会关闭 LoRA、ControlNet、IP-Adapter、参考链、自定义 VAE 和 PiD，只比较所有基础 checkpoint 都具备的 Prompt / 源图能力，避免某个模型偷偷多一层条件。
@@ -314,8 +318,69 @@ COMFYUI_PATH=/absolute/path/to/ComfyUI python app.py
 
 `get_feature_list()` 空参默认返回 Fluxus 兼容的完整 schema；如需节省 MCP token，传 `compact=true` 获取摘要。也可以传 `feature_name`，或使用 `get_chain_schema(chain_type)` 查询单项完整 schema。
 
+`run` 会在排队前校验基础参数、chain 结构和模型扩展能力；请求未知或明显不兼容的 chain 会明确失败。数值字符串会先归一化，不要求调用方严格区分 JSON 数字与表单字符串。
+
 所有 UI 原子事件均隐藏，不作为公共 API 暴露。
 
+### AI 调用图片编辑与批量任务
+
+推荐流程：`get_task_list()` → `get_model_list()` → `get_model_features(model)` → `run(json_params)` → 轮询 `get_task_status(task_id)`。`get_model_features.image_workflows` 提供具体模型的运行方式、参考用途、图片上限、增量参数 schema 和使用规则；AI 不需要猜 `injector_type`。不传 `steps/cfg/sampler/scheduler` 时服务端沿用模型推荐值，负面提示词是否生效仍取决于模型和 CFG。
+
+编辑两张参考图的 JSON 模板（图片值需替换为实际可访问的 URL 或完整 Base64 Data URI）：
+
+```json
+{
+  "task_type": "reference",
+  "model": "lightx2v/Qwen-Image-Edit-2511-Lightning",
+  "prompt": "保留图1的主体，参考 img2 的背景和光线。",
+  "images": ["https://example.com/subject.png", "https://example.com/background.png"],
+  "reference_role": "auto",
+  "width": 1024,
+  "height": 1024,
+  "async_execution": true
+}
+```
+
+也可用顶层 `"img1": "...", "图2": "..."` 替代 `images`，按数值排序而非 JSON 字段顺序。编号必须连续且唯一；`img1` 与 `图1` 重复、跳号、空项、`images` 与 `image` / 原生参考 `chain` 混用都会报 `INVALID_PARAMS`。输入图片支持公网 HTTP(S) URL 和 Base64 Data URI，不接受服务端本地路径。
+
+| 意图 | 最小图片参数 |
+|---|---|
+| 一图指令编辑 / 多图融合 | `task_type=reference`、`images`、`width`、`height`；默认 `run_mode=multi_reference` |
+| 普通整体重绘 | `task_type=img2img`、`image`（或单元素 `images`）；可设 `denoise` |
+| 多图逐张重绘 | `task_type=img2img`、`run_mode=multi_independent`、`images`；Prompt 共用 |
+| 图片×模型 | 同上，改为 `run_mode=multi_model_grid` 并加 `extra_models` |
+| 模型 PK | 原任务参数加 `run_mode=model_pk`、`extra_models`；只比较基础模型 |
+
+旧的 `image`、`txt2img + chain`、同步 / 异步调用、`run_imagegen` 保留。参考用途默认 `auto`；`identity` / `style` 只在模型 discovery 明确返回时使用，不能跨模型猜测。
+
+Gradio Client 传参和异步轮询方式（`params` 使用上面的 JSON 内容）：
+
+```python
+import json
+import time
+from gradio_client import Client
+
+client = Client("BlueSkyXN/ImageGen-Studio")
+task = client.predict(json_params=json.dumps(params, ensure_ascii=False), api_name="/run")
+if "error" in task:
+    raise RuntimeError(task["error"])
+while task["status"] in {"queued", "processing"}:
+    time.sleep(2)
+    task = client.predict(task_id=task["task_id"], api_name="/get_task_status")
+if task["status"] == "failed":
+    raise RuntimeError(task["error"])
+print(task["result"])
+```
+
+轮询超时应重查同一 `task_id`，不要重复提交生成。`result.images` 仍是图片 URL 列表；新增 `image_references`、`effective_prompt`、`captions`、`jobs`、`partial_success` 和 `summary` 用于检查实际编号、送入模型的 Prompt、各子任务结果。批量任务的尺寸/seed 以 `jobs` 为准，顶层字段兼容保留首任务值。匿名 HF 调用仍受 ZeroGPU 额度限制，已有授权客户端可按自身认证方式传入 HF token。
+
+### 配置入口
+
+- 模型清单：`yaml/model_list.yaml`；模型推荐采样参数：`yaml/model_defaults.yaml`，UI 重置按钮和 API 默认值沿用此配置。
+- 架构扩展：`yaml/image_gen_features.yaml`；具体 checkpoint 的编辑能力限制：`core/model_capabilities.py`。
+- 原生参考图张数：`core/reference_inputs.py` 的 `REFERENCE_CHAIN_SPECS` 与 injector 容量一致；有效上限还受 `IMAGEGEN_MAX_MULTI_IMAGES`、`IMAGEGEN_MAX_REFERENCE_IMAGES` 限制，修改后重启生效。仅提高环境变量不能突破模型容量。
+- 像素和任务预算沿用上表的 `IMAGEGEN_MAX_*` 环境变量；界面、执行计划与 API discovery 共用，不需维护另一套 API 参数表。
+
 ## 验证
 
 ```bash
@@ -323,7 +388,7 @@ python -m compileall -q .
 python -m unittest discover -s tests -v
 ```
 
-界面烟测会以 stub ComfyUI 构建完整 Gradio 配置，因此不需要下载模型或占用 GPU。
+界面烟测在安装 Gradio 后会以 stub ComfyUI 构建完整配置，因此不需要下载模型或占用 GPU；未安装 Gradio 时相关用例会跳过。
 
 ## 目录
 

core/execution_plan.py

@@ -17,11 +17,18 @@ from typing import Any
 
 from PIL import Image
 
-from core.model_capabilities import supports_chain_for_model
+from core.reference_inputs import (
+    ImageInputError as ExecutionPlanError,
+    REFERENCE_CHAIN_SPECS,
+    SOURCE_IMAGE_KEYS,
+    clear_native_references,
+    prepare_image_bindings,
+    reference_choices,
+    validate_image_budget,
+)
 from core.runtime_config import CONFIG
 from core.settings import (
     ARCHITECTURES_CONFIG,
-    FEATURES_CONFIG,
     MODEL_DEFAULTS_CONFIG,
     MODEL_MAP_CHECKPOINT,
     MODEL_TYPE_MAP,
@@ -71,21 +78,15 @@ COMPARISON_CHAIN_INPUT_KEYS = (
     "reference_image_data",
 )
 
-# (chain name, pipeline input key, maximum images supported by its injector)
-REFERENCE_CHAIN_SPECS = {
-    "qwen_image_edit": ("qwen_image_edit_data", 3),
-    "joyai_image": ("joyai_reference_data", 2),
-    "boogu_image_edit": ("boogu_edit_data", 10),
-    "reference_image": ("reference_image_data", 10),
-    "reference_latent": ("reference_latent_data", 10),
-    "hidream_o1_reference": ("hidream_o1_reference_data", 10),
-    "krea2_identity_edit": ("krea2_identity_edit_data", 2),
-    "krea2_style_reference": ("krea2_reference_edit_data", 3),
-}
-
-
-class ExecutionPlanError(ValueError):
-    pass
+def available_run_modes(task_type: str, model_name: str) -> list[tuple[str, str]]:
+    if task_type == "reference":
+        return [("整组参考图一起编辑", MODE_MULTI_REFERENCE)] if reference_choices(model_name) else []
+    modes = {MODE_SINGLE, MODE_MODEL_PK}
+    if task_type in INDEPENDENT_IMAGE_TASK_KEYS:
+        modes.update({MODE_MULTI_INDEPENDENT, MODE_MULTI_MODEL_GRID})
+    if task_type == "txt2img" and reference_choices(model_name):
+        modes.add(MODE_MULTI_REFERENCE)
+    return [choice for choice in RUN_MODE_CHOICES if choice[1] in modes]
 
 
 @dataclass(frozen=True)
@@ -219,42 +220,12 @@ def load_uploaded_images(uploaded_files: Sequence[Any] | None) -> list[Image.Ima
 
 
 def _pick_reference_chain(model_name: str, role: str) -> tuple[str, str, int]:
-    workflow_type = _workflow_type(model_name)
-    enabled = set(
-        FEATURES_CONFIG.get(workflow_type, {}).get("enabled_chains", [])
-    )
-
-    if role == "identity":
-        order = ["krea2_identity_edit"]
-    elif role == "style":
-        # The generic FLUX style injector has a different image/weight schema;
-        # keep this high-level path limited to the validated Krea reference chain.
-        order = ["krea2_style_reference"]
-    else:
-        order = [
-            "qwen_image_edit",
-            "joyai_image",
-            "boogu_image_edit",
-            "reference_image",
-            "reference_latent",
-            "hidream_o1_reference",
-            "krea2_identity_edit",
-            "krea2_style_reference",
-        ]
-
-    for chain_name in order:
-        if (
-            chain_name in enabled
-            and chain_name in REFERENCE_CHAIN_SPECS
-            and supports_chain_for_model(model_name, chain_name)
-        ):
-            input_key, maximum = REFERENCE_CHAIN_SPECS[chain_name]
-            return chain_name, input_key, maximum
-
-    if role in {"identity", "style"}:
-        raise ExecutionPlanError(
-            f"模型“{model_name}”不支持所选的{('身份' if role == 'identity' else '风格')}参考方式。"
-        )
+    choices = reference_choices(model_name)
+    if role in choices:
+        selected = choices[role]
+        return selected["chain"], selected["input_key"], selected["max_images"]
+    if choices:
+        raise ExecutionPlanError(f"模型“{model_name}”不支持参考用途 {role}；可选：{', '.join(choices)}。")
     raise ExecutionPlanError(
         f"模型“{model_name}”没有可自动使用的多图参考链；请换用编辑/多模态模型。"
     )
@@ -273,6 +244,14 @@ def build_execution_plan(
     if mode not in {choice[1] for choice in RUN_MODE_CHOICES}:
         raise ExecutionPlanError(f"未知运行模式：{mode}")
 
+    base_inputs = dict(base_inputs)
+    if base_inputs.get("task_type") == "reference":
+        if mode == MODE_SINGLE:
+            mode = MODE_MULTI_REFERENCE
+        if mode != MODE_MULTI_REFERENCE:
+            raise ExecutionPlanError("参考图编辑 / 融合只支持整组参考一起处理；逐张重绘请选图生图。")
+        base_inputs["task_type"] = "txt2img"
+
     base_model = str(base_inputs.get("model_display_name") or "")
     if base_model not in MODEL_MAP_CHECKPOINT:
         raise ExecutionPlanError("请先选择有效模型。")
@@ -293,6 +272,18 @@ def build_execution_plan(
         raise ExecutionPlanError("模型 PK 至少需要再选择 1 个对比模型。")
 
     source_images = list(images or [])
+    if len(source_images) > CONFIG.max_multi_images:
+        raise ExecutionPlanError(f"一次最多输入 {CONFIG.max_multi_images} 张图片；当前为 {len(source_images)} 张。")
+    if any(image is None for image in source_images):
+        raise ExecutionPlanError("图片列表不能包含空项；请按图1、图2的顺序连续提供图片。")
+    validate_image_budget(source_images)
+    if source_images and mode in {MODE_SINGLE, MODE_MODEL_PK}:
+        key = SOURCE_IMAGE_KEYS.get(base_inputs.get("task_type"))
+        if not key or len(source_images) != 1:
+            raise ExecutionPlanError("普通生成 / 模型 PK 只接受一张源图；多图请显式选择逐图处理或参考图编辑 / 融合。")
+        if base_inputs.get(key) is not None:
+            raise ExecutionPlanError("源图与图片列表不能同时提供，请只使用一个图片入口。")
+        base_inputs[key] = source_images[0]
     shared_seed = base_inputs.get("seed", -1)
     try:
         shared_seed = int(shared_seed)
@@ -337,12 +328,13 @@ def build_execution_plan(
                     _make_fair_comparison(values)
                 values["seed"] = shared_seed
                 values[task_input_key] = image
+                values["_source_image_index"] = image_index
                 caption = _caption(f"输入 {image_index} · {model_name}", values)
                 plan.append(PlannedGeneration(values, caption))
 
     elif mode == MODE_MULTI_REFERENCE:
         if str(base_inputs.get("task_type")) != "txt2img":
-            raise ExecutionPlanError("多图融合请把任务切换为“文生图”；参考图会直接进入编辑模型。")
+            raise ExecutionPlanError("多图融合请选“参考图编辑 / 融合”（也兼容文生图）；普通图生图是逐张重绘。")
         if not source_images:
             raise ExecutionPlanError("多图融合需要上传至少 1 张参考图。")
         chain_name, input_key, maximum = _pick_reference_chain(
@@ -353,8 +345,9 @@ def build_execution_plan(
                 f"当前模型的 {chain_name} 最多支持 {maximum} 张参考图。"
             )
         values = _for_model(base_inputs, base_model, False)
-        existing = [value for value in values.get(input_key, []) if value is not None]
-        values[input_key] = [*existing, *source_images][:maximum]
+        # 高层入口的文件列表是唯一素材源；隐藏的旧槽位保留在 UI，不并入本次任务。
+        clear_native_references(values)
+        values[input_key] = source_images
         values["seed"] = shared_seed
         plan.append(
             PlannedGeneration(
@@ -377,6 +370,11 @@ def build_execution_plan(
             f"预计输出 {estimated_outputs} 张，超过上限 {CONFIG.max_plan_outputs}；"
             "请减少模型、输入图片或单次生成数量。"
         )
+    for item in plan:
+        prepare_image_bindings(
+            item.inputs, independent=mode in {MODE_MULTI_INDEPENDENT, MODE_MULTI_MODEL_GRID},
+        )
+
     # Release Comfy's global model state only at an actual model boundary.  The
     # final model remains warm for a likely follow-up generation.
     for current, following in pairwise(plan):

core/model_capabilities.py

@@ -20,3 +20,17 @@ def supports_chain_for_model(model_name: str, chain_name: str) -> bool:
     if chain_name in _EDIT_ONLY_CHAINS:
         return "edit" in str(model_name).casefold()
     return True
+
+
+def enabled_chains_for_model(model_name: str) -> list[str]:
+    """以具体 checkpoint 为准，供界面、计划与 API 共用。"""
+    from core.settings import ARCHITECTURES_CONFIG, FEATURES_CONFIG, MODEL_TYPE_MAP
+
+    architecture = MODEL_TYPE_MAP.get(model_name, "SDXL")
+    model_type = ARCHITECTURES_CONFIG.get("architectures", {}).get(
+        architecture, {}
+    ).get("model_type", architecture.lower().replace(" ", "").replace(".", ""))
+    return [
+        chain for chain in FEATURES_CONFIG.get(model_type, {}).get("enabled_chains", [])
+        if supports_chain_for_model(model_name, chain)
+    ]

core/pipelines/pipeline_input_processor.py

@@ -7,6 +7,7 @@ from typing import Dict, Any, List
 
 from core.settings import INPUT_DIR, MULTIPLIERS_MAP, LORA_DIR, EMBEDDING_DIR, VAE_DIR
 from core.runtime_config import CONFIG
+from core.reference_inputs import REFERENCE_CHAIN_SPECS
 from imagegen_utils.app_utils import (
     sanitize_filename,
     get_lora_path,
@@ -32,14 +33,7 @@ REFERENCE_IMAGE_LIMITS = {
     "flux1_ipadapter_data": 5,
     "sd3_ipadapter_chain": 5,
     "style_data": 5,
-    "reference_latent_data": 10,
-    "hidream_o1_reference_data": 10,
-    "joyai_reference_data": 2,
-    "krea2_identity_edit_data": 2,
-    "krea2_reference_edit_data": 3,
-    "qwen_image_edit_data": 3,
-    "boogu_edit_data": 10,
-    "reference_image_data": 10,
+    **{key: limit for key, limit in REFERENCE_CHAIN_SPECS.values()},
 }
 
 

core/pipelines/sd_image_pipeline.py

@@ -10,6 +10,7 @@ from typing import List, Dict, Any
 from .base_pipeline import BasePipeline
 from core.settings import *
 from core.model_capabilities import supports_chain_for_model
+from core.reference_inputs import ImageInputError, prepare_image_bindings
 from imagegen_utils.app_utils import sanitize_prompt
 from core.workflow_assembler import WorkflowAssembler
 from core.runtime_config import CONFIG, estimate_gpu_duration
@@ -114,17 +115,6 @@ class SdImagePipeline(BasePipeline):
         enabled_chains = set(
             FEATURES_CONFIG.get(workflow_model_type, {}).get("enabled_chains", [])
         )
-        if task_type != "txt2img":
-            has_native_references = any(
-                any(ui_inputs.get(input_key) or [])
-                for chain_name in self.NATIVE_REFERENCE_CHAINS
-                for input_key in self.CHAIN_INPUT_KEYS[chain_name]
-            )
-            if has_native_references:
-                raise gr.Error(
-                    "原生多图参考编辑目前只支持“文生图 / 多图融合”。"
-                    "普通逐张重绘请使用图生图；不要同时叠加源图 latent 与参考图链。"
-                )
         for chain_name, input_keys in self.CHAIN_INPUT_KEYS.items():
             if (
                 chain_name not in enabled_chains
@@ -132,6 +122,10 @@ class SdImagePipeline(BasePipeline):
             ):
                 for input_key in input_keys:
                     ui_inputs[input_key] = []
+        try:
+            prepare_image_bindings(ui_inputs)
+        except ImageInputError as exc:
+            raise gr.Error(str(exc)) from exc
         if "pid" not in enabled_chains:
             ui_inputs["pid_settings"] = "OFF"
         if "vae" not in enabled_chains:

core/reference_inputs.py

@@ -0,0 +1,141 @@
+"""图片顺序、模型参考图能力和提示词编号的共享契约；不下载或改写图片。"""
+
+from __future__ import annotations
+
+import re
+from PIL import Image
+
+from core.model_capabilities import enabled_chains_for_model
+from core.runtime_config import CONFIG
+
+
+class ImageInputError(ValueError):
+    pass
+
+
+# (pipeline input key, injector capacity)。应用上限仍受 RuntimeConfig 约束。
+REFERENCE_CHAIN_SPECS = {
+    "qwen_image_edit": ("qwen_image_edit_data", 3),
+    "joyai_image": ("joyai_reference_data", 2),
+    "boogu_image_edit": ("boogu_edit_data", 10),
+    "reference_image": ("reference_image_data", 10),
+    "reference_latent": ("reference_latent_data", 10),
+    "hidream_o1_reference": ("hidream_o1_reference_data", 10),
+    "krea2_identity_edit": ("krea2_identity_edit_data", 2),
+    "krea2_style_reference": ("krea2_reference_edit_data", 3),
+}
+SOURCE_IMAGE_KEYS = {
+    "img2img": "img2img_image",
+    "inpaint": "inpaint_image",
+    "outpaint": "outpaint_image",
+    "hires_fix": "hires_image",
+}
+
+# 不匹配文件名/标识符；引号和反引号内的文字作为画面文字保留。
+# 裸“图”不从构图、地图、草图等常用复合词中截取；保留完整的“参考图 / 图片”别名。
+_REFERENCE = re.compile(
+    r"(?:(?<![A-Za-z_./\\-])(?:参考图|图片|(?<![构地蓝位视插草绘截贴])图)|(?<![A-Za-z0-9_./\\-])(?:img|image))[ \t]*([0-9０-９]+)"
+    r"(?![0-9０-９A-Za-z_./\\-])", re.IGNORECASE,
+)
+_LITERAL = re.compile(r'("[^"\n]*"|“[^”\n]*”|「[^」\n]*」|`[^`\n]*`)')
+
+
+def reference_choices(model: str) -> dict[str, dict]:
+    enabled = enabled_chains_for_model(model)
+    available = [name for name in REFERENCE_CHAIN_SPECS if name in enabled]
+    roles = {}
+    for role, order in (
+        ("auto", available),
+        ("identity", ["krea2_identity_edit"]),
+        ("style", ["krea2_style_reference"]),
+    ):
+        chain = next((name for name in order if name in available), None)
+        if chain:
+            key, limit = REFERENCE_CHAIN_SPECS[chain]
+            roles[role] = {
+                "chain": chain, "input_key": key,
+                "max_images": min(limit, CONFIG.max_multi_images, CONFIG.max_reference_images),
+            }
+    return roles
+
+
+def clear_native_references(values: dict) -> None:
+    for key, _ in REFERENCE_CHAIN_SPECS.values():
+        values[key] = []
+
+
+def active_reference_groups(values: dict) -> dict[str, list]:
+    enabled = enabled_chains_for_model(values.get("model_display_name", ""))
+    return {
+        chain: [image for image in values.get(key, []) if image is not None]
+        for chain, (key, _) in REFERENCE_CHAIN_SPECS.items()
+        if chain in enabled and any(image is not None for image in values.get(key, []))
+    }
+
+
+def image_bindings(count: int) -> list[dict]:
+    return [
+        {"id": f"img{i}", "index": i, "aliases": [f"图{i}", f"img{i}", f"image{i}"],
+         "model_reference": f"image {i}"}
+        for i in range(1, count + 1)
+    ]
+
+
+def validate_image_budget(images: list) -> None:
+    total = 0.0
+    for i, image in enumerate(images, 1):
+        if not isinstance(image, Image.Image):
+            continue  # API 排队前可用未下载的 URL 校验结构。
+        megapixels = image.width * image.height / 1_000_000
+        if megapixels > CONFIG.max_input_megapixels:
+            raise ImageInputError(f"图{i} 为 {megapixels:.1f} MP，超过单图 {CONFIG.max_input_megapixels:g} MP 上限。")
+        total += megapixels
+    if total > CONFIG.max_reference_megapixels:
+        raise ImageInputError(f"图片累计 {total:.1f} MP，超过 {CONFIG.max_reference_megapixels:g} MP 上限；请减少或缩小图片。")
+
+
+def normalize_image_references(prompt: str, count: int, *, independent=False, ambiguous=False) -> str:
+    """仅统一显式编号；不是翻译器，也不承诺所有模型都懂位置指令。"""
+    def replace(match):
+        if independent:
+            raise ImageInputError("多图独立/图片×模型共用提示词，不能跨图引用编号；请写“当前图片”，或改用参考图编辑 / 融合。")
+        if ambiguous:
+            raise ImageInputError("多个原生参考链同时启用时图片编号不明确；请只保留一个参考链，或使用参考图编辑 / 融合入口。")
+        index = int(match.group(1))
+        if not 1 <= index <= count:
+            raise ImageInputError(f"提示词引用了 {match.group(0)}，但本次只有 {count} 张有效图片；编号从 1 开始。")
+        separator = " " if _REFERENCE.match(match.string, match.end()) else ""
+        return f"image {index}{separator}"
+
+    parts = _LITERAL.split(prompt or "")
+    return "".join(part if index % 2 else _REFERENCE.sub(replace, part) for index, part in enumerate(parts))
+
+
+def prepare_image_bindings(values: dict, *, independent=False) -> None:
+    """在下载模型前处理当前有效输入；调用方持有自己的输入字典。"""
+    enabled = enabled_chains_for_model(values.get("model_display_name", ""))
+    for chain, (key, _) in REFERENCE_CHAIN_SPECS.items():
+        if chain not in enabled:
+            values[key] = []
+    groups = active_reference_groups(values)
+    task = values.get("task_type")
+    if task != "txt2img" and groups:
+        raise ImageInputError("原生参考图只能用于参考图编辑 / 融合（API 也兼容 txt2img + chain）；普通重绘不要叠加参考链。")
+    for chain, images in groups.items():
+        limit = REFERENCE_CHAIN_SPECS[chain][1]
+        if len(images) > limit:
+            raise ImageInputError(f"{chain} 最多支持 {limit} 张参考图，当前为 {len(images)} 张；不会截断图片。")
+    count = sum(map(len, groups.values()))
+    if count > CONFIG.max_reference_images:
+        raise ImageInputError(f"本次参考图超过 {CONFIG.max_reference_images} 张上限。")
+    validate_image_budget([image for images in groups.values() for image in images])
+    if task in SOURCE_IMAGE_KEYS:
+        source = values.get(SOURCE_IMAGE_KEYS[task])
+        if task == "inpaint" and source is None:
+            source = (values.get("inpaint_image_dict") or {}).get("background")
+        count = int(source is not None)
+    for key in ("positive_prompt", "negative_prompt"):
+        values[key] = normalize_image_references(
+            values.get(key, ""), count, independent=independent, ambiguous=len(groups) > 1,
+        )
+    values["_image_references"] = image_bindings(count) if len(groups) <= 1 else []

mcp_tools/__init__.py

@@ -1,4 +1,18 @@
+_GRADIO_INTEGRATION_EXPORTS = {
+    "register_high_level_mcp_apis",
+    "cleanup_dependencies_api_names",
+    "patch_gradio_api_suppression",
+    "HIGH_LEVEL_MCP_API_NAMES",
+}
+
+
 def __getattr__(name):
+    if name in _GRADIO_INTEGRATION_EXPORTS:
+        # Keep the transport-independent handlers importable in validation and
+        # CLI environments where Gradio is intentionally not installed.
+        from . import mcp_gradio_integration
+
+        return getattr(mcp_gradio_integration, name)
     if name in ("types", "server", "client", "shared"):
         raise ImportError(f"No module named 'mcp.{name}' in local mcp package")
     raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
@@ -12,13 +26,6 @@ from .run import handle_run
 from .get_task_status import handle_get_task_status
 from .get_chain_schema import handle_get_chain_schema
 from .error_schema import make_error, make_validation_error, make_not_found_error
-from .mcp_gradio_integration import (
-    register_high_level_mcp_apis,
-    cleanup_dependencies_api_names,
-    patch_gradio_api_suppression,
-    HIGH_LEVEL_MCP_API_NAMES,
-)
-
 MCP_FUNCTIONS = [
     handle_get_task_list,
     handle_get_model_architecture_list,

mcp_tools/common.py

@@ -202,9 +202,17 @@ def _load_yaml(filepath: str) -> dict:
 _COMMON_OPTIONAL_INPUTS = [
     "steps", "cfg", "sampler", "scheduler", "seed",
     "negative_prompt", "batch_size", "chain", "async_execution",
+    "images", "run_mode", "extra_models", "reference_role", "use_model_defaults",
 ]
 
 _TASK_DEFINITIONS = [
+    {
+        "task_type": "reference",
+        "display_name": "参考图编辑 / 融合",
+        "description": "一张图按指令编辑，多张图整组参考。images 按数组顺序从 1 编号；图1 / img1 / image1 等价。先查询 get_model_features.image_workflows 的用途和张数上限，不需要填写 chain。",
+        "required_inputs": ["prompt", "images", "width", "height"],
+        "optional_inputs": [field for field in _COMMON_OPTIONAL_INPUTS if field != "images"],
+    },
     {
         "task_type": "txt2img",
         "display_name": "Text-to-Image",
@@ -215,9 +223,9 @@ _TASK_DEFINITIONS = [
     {
         "task_type": "img2img",
         "display_name": "Image-to-Image",
-        "description": "Perform global repaint and style transfer based on a source image. Denoise strength must be specified.",
-        "required_inputs": ["prompt", "image", "denoise"],
-        "optional_inputs": _COMMON_OPTIONAL_INPUTS,
+        "description": "Perform global repaint and style transfer based on a source image. Denoise defaults to 0.7 when omitted.",
+        "required_inputs": ["prompt", "image"],
+        "optional_inputs": ["denoise"] + _COMMON_OPTIONAL_INPUTS,
     },
     {
         "task_type": "inpaint",
@@ -242,6 +250,65 @@ _TASK_DEFINITIONS = [
     },
 ]
 
+
+def _find_model_record(model: str) -> tuple[str | None, dict | None]:
+    """Return the registry architecture and record for a display name."""
+
+    model_list = _load_yaml(_MODEL_LIST_PATH)
+    for architecture, architecture_data in (model_list.get("Checkpoint", {}) or {}).items():
+        if not isinstance(architecture_data, dict):
+            continue
+        for record in architecture_data.get("models", []) or []:
+            if isinstance(record, dict) and record.get("display_name") == model:
+                return architecture, record
+    return None, None
+
+
+def _resolve_feature_name(chain_type: str) -> str | None:
+    """Resolve a canonical chain feature from its name or configured alias."""
+
+    chain_features = _load_yaml(_CHAIN_FEATURES_PATH)
+    if chain_type in chain_features:
+        return chain_type
+    for feature_name, feature_data in chain_features.items():
+        aliases = feature_data.get("chains") if isinstance(feature_data, dict) else None
+        if aliases == chain_type or (isinstance(aliases, list) and chain_type in aliases):
+            return feature_name
+    return None
+
+
+_FEATURE_ADAPTER_ALIASES = {
+    "boogu_edit": "boogu_image_edit",
+}
+
+
+def _canonicalize_chain_type(chain_type: str) -> str:
+    """Normalize explicitly supported legacy request aliases."""
+
+    return _FEATURE_ADAPTER_ALIASES.get(chain_type, chain_type)
+
+
+def _get_model_feature_context(model: str) -> dict | None:
+    """Return registry-backed architecture and enabled-chain information."""
+
+    architecture, _ = _find_model_record(model)
+    if not architecture:
+        return None
+    architecture_config = _load_yaml(_MODEL_ARCHITECTURES_PATH).get("architectures", {}) or {}
+    architecture_info = architecture_config.get(architecture, {}) or {}
+    model_type = architecture_info.get("model_type", architecture.lower())
+    features_config = _load_yaml(_IMAGE_GEN_FEATURES_PATH)
+    enabled = features_config.get(model_type, {}) or {}
+    from core.model_capabilities import supports_chain_for_model
+
+    return {
+        "architecture": architecture,
+        "model_type": model_type,
+        "enabled_chains": set(enabled.get("enabled_chains", []) or []),
+        "supports_chain": lambda chain: supports_chain_for_model(model, chain),
+    }
+
+
 _TASKS_DB: Dict[str, Dict[str, Any]] = {}
 _TASKS_LOCK = threading.RLock()
 
@@ -351,9 +418,9 @@ def _execute_imagegen_pipeline(task_id: str, params: dict):
             "denoise": params.get("denoise", 1.0),
         }
 
-        if "image" in params and params["image"]:
-            pil_img = _parse_image_param(params["image"])
-            if pil_img:
+        if task_type in {"img2img", "inpaint", "outpaint", "hires_fix"}:
+            pil_img = _parse_image_param(params.get("image"))
+            if pil_img is not None or params.get("images"):
                 if task_type == "img2img":
                     ui_inputs["img2img_image"] = pil_img
                     ui_inputs["img2img_denoise"] = params.get("denoise", 0.7)
@@ -735,8 +802,40 @@ def _execute_imagegen_pipeline(task_id: str, params: dict):
 
         _update_task(task_id, progress=50)
 
-        # Execute Pipeline
-        output = sd_image_pipeline.run(ui_inputs=ui_inputs, progress=DummyProgress())
+        from core.execution_plan import build_execution_plan, execute_generation_plan
+
+        plan = build_execution_plan(
+            ui_inputs, mode=params.get("run_mode", "single"),
+            images=[_parse_image_param(value) for value in params.get("images", [])],
+            extra_models=params.get("extra_models"),
+            reference_role=params.get("reference_role", "auto"),
+            use_model_defaults=params.get("use_model_defaults", True),
+        )
+        jobs = []
+
+        def generate(values, progress):
+            job = {"model": values["model_display_name"], "task_type": values["task_type"], "status": "processing"}
+            if "_source_image_index" in values:
+                job["source_index"] = values["_source_image_index"]
+            jobs.append(job)
+            try:
+                result = sd_image_pipeline.run(ui_inputs=values, progress=progress)
+                if result is None or (isinstance(result, list) and not result):
+                    raise ValueError("生成器没有返回图片。")
+                job.update(status="completed", output_count=len(result) if isinstance(result, list) else int(result is not None))
+                return result
+            except Exception as exc:
+                job.update(status="failed", error=str(exc))
+                raise
+            finally:
+                job.update(
+                    seed=values.get("seed"), width=values.get("width"), height=values.get("height"),
+                    effective_prompt=values.get("positive_prompt", ""),
+                    image_references=values.get("_image_references", []),
+                )
+
+        output, summary = execute_generation_plan(plan, generate, progress=DummyProgress())
+        ui_inputs = plan[0].inputs
 
         try:
             from core.settings import OUTPUT_DIR
@@ -753,8 +852,9 @@ def _execute_imagegen_pipeline(task_id: str, params: dict):
 
         base_url = _get_public_base_url()
         images = []
-        raw_list = output if isinstance(output, list) else ([output] if output else [])
-        for idx, item in enumerate(raw_list):
+        captions = []
+        for idx, (item, caption) in enumerate(output):
+            captions.append(caption)
             target_path = None
             if hasattr(item, "save"):  # PIL Image
                 filename = f"mcp_{task_id}_{idx}.png"
@@ -789,6 +889,12 @@ def _execute_imagegen_pipeline(task_id: str, params: dict):
                 "width": ui_inputs.get("width", params.get("width", 1024)),
                 "height": ui_inputs.get("height", params.get("height", 1024)),
                 "execution_time_seconds": execution_time,
+                "effective_prompt": ui_inputs.get("positive_prompt", ""),
+                "image_references": ui_inputs.get("_image_references", []),
+                "captions": captions,
+                "jobs": jobs,
+                "partial_success": any(job["status"] == "failed" for job in jobs),
+                "summary": summary,
             },
         )
 

mcp_tools/get_model_features.py

@@ -14,6 +14,7 @@ from .common import (
 )
 from .error_schema import make_validation_error, make_not_found_error
 from core.model_capabilities import supports_chain_for_model
+from .image_requests import image_workflow_contract
 
 
 def handle_get_model_features(model: str) -> dict:
@@ -80,7 +81,11 @@ def handle_get_model_features(model: str) -> dict:
         "cfg": merged_defaults.get("cfg", 1.0),
     }
 
-    supported_tasks = [t["task_type"] for t in _TASK_DEFINITIONS]
+    image_workflows = image_workflow_contract(model)
+    supported_tasks = [
+        t["task_type"] for t in _TASK_DEFINITIONS
+        if t["task_type"] != "reference" or image_workflows["reference_supported"]
+    ]
 
     result = {
         "name": model,
@@ -88,6 +93,7 @@ def handle_get_model_features(model: str) -> dict:
         "supported_tasks": supported_tasks,
         "supported_features": supported_features,
         "default_parameter": default_parameter,
+        "image_workflows": image_workflows,
     }
 
     default_pos = model_specific_defaults.get(

mcp_tools/get_task_list.py

@@ -4,8 +4,16 @@ Get a list of all supported image generation task types along with their require
 """
 
 from .common import _TASK_DEFINITIONS
+import copy
 
 
 def handle_get_task_list() -> list:
     """Get a list of all supported image generation task types along with their required/optional parameter lists."""
-    return _TASK_DEFINITIONS
+    tasks = copy.deepcopy(_TASK_DEFINITIONS)
+    for task in tasks:
+        if "image" in task["required_inputs"] or "images" in task["required_inputs"]:
+            task["image_input_alternatives"] = {
+                "choose_exactly_one": ["image", "images", "numbered fields (img1 / 图1 / image1, ...)"],
+                "description": "image is a single source; images is an ordered list. Numbered fields replace images. See get_model_features.image_workflows for task/run_mode constraints. Never send multiple forms together.",
+            }
+    return tasks

mcp_tools/image_requests.py

@@ -0,0 +1,143 @@
+"""高层 API 图片参数兼容层；顺序与能力校验复用 UI 的执行计划。"""
+
+import re
+from PIL import Image
+
+from core.execution_plan import (
+    MODE_MODEL_PK, MODE_MULTI_MODEL_GRID, MODE_MULTI_REFERENCE, MODE_SINGLE,
+    available_run_modes, build_execution_plan,
+)
+from core.reference_inputs import ImageInputError, REFERENCE_CHAIN_SPECS, SOURCE_IMAGE_KEYS, reference_choices
+from core.runtime_config import CONFIG
+
+
+NATIVE_CHAIN_ALIASES = {
+    "reference_edit": "reference_latent",
+    "mage_flow_reference_edit": "reference_image",
+    "joyai_reference_edit": "joyai_image",
+    "boogu_edit": "boogu_image_edit",
+}
+_IMAGE_KEY = re.compile(r"(?:图|img|image)([0-9０-９]+)", re.IGNORECASE)
+
+
+def normalize_image_request(params: dict) -> None:
+    """只修改 handle_run 的副本。显式混用入口时拒绝，不猜优先级。"""
+    numbered = {}
+    for key in list(params):
+        match = _IMAGE_KEY.fullmatch(key) if isinstance(key, str) else None
+        if not match:
+            continue
+        index = int(match.group(1))
+        if index in numbered:
+            raise ImageInputError(f"图片编号 {index} 重复；图{index} / img{index} / image{index} 只能提供一个字段。")
+        numbered[index] = params.pop(key)
+    if numbered:
+        if "images" in params or params.get("image") is not None:
+            raise ImageInputError("编号字段、images 列表和 image 源图不能混用，请只保留一个入口。")
+        if sorted(numbered) != list(range(1, len(numbered) + 1)):
+            raise ImageInputError("图片编号必须从 1 开始且连续，不能缺少 img1 或跳号。")
+        params["images"] = [numbered[i] for i in range(1, len(numbered) + 1)]
+    if "images" in params:
+        images = params["images"]
+        if not isinstance(images, list) or not images:
+            raise ImageInputError("images 必须是非空、有序图片列表。")
+        if any(not isinstance(image, (str, Image.Image)) or (isinstance(image, str) and not image.strip()) for image in images):
+            raise ImageInputError("images 不允许空图片；删除空项后重新按顺序编号。")
+        if params.get("image") is not None:
+            raise ImageInputError("image 与 images 不能混用。")
+    if params["task_type"] == "reference":
+        params.setdefault("run_mode", MODE_MULTI_REFERENCE)
+        if params.get("image") is not None:
+            if not isinstance(params["image"], (str, Image.Image)):
+                raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。")
+            params["images"] = [params.pop("image")]
+    elif params["task_type"] == "txt2img" and params.get("image") is not None:
+        raise ImageInputError("txt2img 不使用 image 源图；参考图编辑请用 task_type=reference 和 images。")
+    if params.get("image") is not None and not isinstance(params["image"], (str, Image.Image)):
+        raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。")
+    for field in ("run_mode", "reference_role"):
+        if field in params and not isinstance(params[field], str):
+            raise ImageInputError(f"{field} 必须是字符串。")
+    if "extra_models" in params and (
+        not isinstance(params["extra_models"], list)
+        or not all(isinstance(model, str) for model in params["extra_models"])
+    ):
+        raise ImageInputError("extra_models 必须是模型名称列表。")
+    if "use_model_defaults" in params and not isinstance(params["use_model_defaults"], bool):
+        raise ImageInputError("use_model_defaults 必须是 boolean。")
+
+
+def validate_image_plan(params: dict) -> None:
+    """不下载图片、不分配 GPU，提交队列前校验完整调用语义。"""
+    values = {
+        "task_type": params["task_type"], "model_display_name": params["model"],
+        "positive_prompt": params["prompt"], "negative_prompt": params.get("negative_prompt") or "",
+        "seed": params.get("seed", -1), "batch_size": params.get("batch_size", 1),
+    }
+    task = params["task_type"]
+    if task in SOURCE_IMAGE_KEYS and params.get("image") is not None:
+        values[SOURCE_IMAGE_KEYS[task]] = params["image"]
+    for item in params.get("chain", []):
+        chain = NATIVE_CHAIN_ALIASES.get(item["injector_type"], item["injector_type"])
+        if chain in REFERENCE_CHAIN_SPECS:
+            if params.get("images"):
+                raise ImageInputError("images 与原生参考 chain 不能混用；自动路由用 images，旧调用只用 chain。")
+            if item.get("image") is None:
+                raise ImageInputError(f"{chain} 缺少 image。")
+            if not isinstance(item["image"], (str, Image.Image)) or (isinstance(item["image"], str) and not item["image"].strip()):
+                raise ImageInputError(f"{chain}.image 必须是图片 URL 或 Base64 Data URI。")
+            values.setdefault(REFERENCE_CHAIN_SPECS[chain][0], []).append(item["image"])
+    mode = params.get("run_mode", MODE_SINGLE)
+    if mode not in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and params.get("extra_models"):
+        raise ImageInputError("extra_models 只用于 model_pk / multi_model_grid。")
+    if mode != MODE_MULTI_REFERENCE and params.get("reference_role", "auto") != "auto":
+        raise ImageInputError("reference_role 只用于 reference / multi_reference。")
+    if mode in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and (
+        params.get("chain") or params.get("vae_id") or params.get("pid") or params.get("pid_settings")
+    ):
+        raise ImageInputError("模型对比仅使用基础模型，不支持 chain / VAE / PiD 覆盖。")
+    build_execution_plan(
+        values, mode=mode, images=params.get("images"),
+        extra_models=params.get("extra_models"), reference_role=params.get("reference_role", "auto"),
+        use_model_defaults=params.get("use_model_defaults", True),
+    )
+
+
+def image_workflow_contract(model: str) -> dict:
+    """在现有 discovery 上增量公开高层图片契约，不要求 AI 猜 injector。"""
+    roles = reference_choices(model)
+    return {
+        "reference_supported": bool(roles),
+        "reference_roles": {
+            role: {"chain": spec["chain"], "max_images": spec["max_images"]}
+            for role, spec in roles.items()
+        },
+        "run_modes_by_task": {
+            task: [mode for _, mode in available_run_modes(task, model)]
+            for task in ["txt2img", *SOURCE_IMAGE_KEYS, "reference"]
+        },
+        "parameters_schema": {
+            "type": "object",
+            "properties": {
+                "images": {"type": "array", "minItems": 1, "maxItems": CONFIG.max_multi_images,
+                           "items": {"type": "string", "description": "Public HTTP(S) image URL or Base64 Data URI; not a local path."}},
+                "run_mode": {"type": "string", "enum": ["single", "model_pk", "multi_independent", "multi_model_grid", "multi_reference"],
+                             "description": "Default single; task_type=reference defaults to multi_reference. Use run_modes_by_task."},
+                "reference_role": {"type": "string", "enum": list(roles) or ["auto"], "default": "auto"},
+                "extra_models": {"type": "array", "items": {"type": "string"}, "description": "Only for model_pk / multi_model_grid; current model is included automatically."},
+                "use_model_defaults": {"type": "boolean", "default": True},
+            },
+            "patternProperties": {r"^(图|img|image)[1-9][0-9]*$": {"type": "string", "description": "Alternative to images; indices must start at 1, be contiguous and unique."}},
+        },
+        "rules": [
+            "Use task_type=reference with images, prompt, width, height for instruction editing / fusion; no native chain needed. Model must support reference.",
+            "images order is 1-based: 图1 = img1 = image1. Numbered prompt references become image 1; quoted/backticked text is literal. No auto translation or prompt expansion.",
+            "Do not mix images, numbered fields, image, or native reference chain. Legacy txt2img + chain and single-image image calls remain supported.",
+            "multi_independent / multi_model_grid use one shared prompt per image. Say 当前图片 / the current image; cross-image numbered references are rejected.",
+            "Leave steps/cfg/sampler/scheduler omitted to use model defaults. Negative prompts depend on model/CFG and may be unused at CFG=1.",
+            "Prefer async_execution=true; poll the same task_id with get_task_status until completed/failed; do not resubmit on a polling timeout.",
+            "Read result.image_references, effective_prompt, captions and jobs. Existing result.images remains a flat URL list; partial_success signals mixed results.",
+        ],
+        "limits": {"max_images": CONFIG.max_multi_images, "max_pk_models": CONFIG.max_pk_models,
+                   "max_jobs": CONFIG.max_plan_jobs, "max_outputs": CONFIG.max_plan_outputs},
+    }

mcp_tools/mcp_gradio_integration.py

@@ -100,11 +100,11 @@ def register_high_level_mcp_apis(demo):
         )
 
     def get_model_features(model: str = "") -> dict:
-        """Query metadata for the specified model, including supported task types, extended features, and official default inference parameters (steps, cfg, sampler, scheduler). This tool MUST be called when explicitly obtaining a model's optimal default hyperparameters (Path 2). Guessing or fabricating hyperparameters without querying is strictly prohibited."""
+        """Query the concrete model's task types, features, recommended sampling defaults and image_workflows. For editing/multi-image use image_workflows: supported run modes, reference roles, image count limits, parameter schema, aliases and usage rules. Do not infer editing support from architecture alone."""
         return sanitize_keys(handle_get_model_features(model.strip()))
 
     def run(json_params: str = "{}") -> dict:
-        """[Recommended Discovery Flow Step 4] Unified image generation task execution interface. Supports txt2img, img2img, and other tasks with chainable extended features. [IMPORTANT PARAMETER RULES] Do NOT guess or fabricate inference hyperparameters such as steps, cfg, sampler, scheduler! Path 1 (Recommended): Pass only required parameters (task_type, model, prompt, width, height), leave optional hyperparams empty (server uses optimal defaults). Path 2: If explicit hyperparams are needed, you MUST first call get_model_features to obtain official defaults before passing them."""
+        """Submit a JSON object encoded as json_params. First discover get_task_list and get_model_features(model). Text generation: task_type=txt2img, model, prompt, width, height. Instruction editing/fusion: task_type=reference, model, prompt, width, height, images=[URL_or_DataURI,...]; reference_role defaults to auto. images order starts at 1: 图1 / img1 / image1 in prompt refer to the first image. Alternatively use top-level img1,img2 (or 图1,图2) fields; do not mix input forms. Ordinary repaint: task_type=img2img and image; batch repaint: images plus run_mode=multi_independent. See image_workflows for model-specific limits and other modes. Omit steps/cfg/sampler/scheduler to use recommended defaults. Prefer async_execution=true; poll the returned task_id via get_task_status (do not resubmit). Legacy chain and run_imagegen calls remain available."""
         try:
             if isinstance(json_params, dict):
                 params = json_params

mcp_tools/run.py

@@ -10,19 +10,31 @@ from core.runtime_config import CONFIG
 from .common import (
     _load_yaml,
     _MODEL_LIST_PATH,
+    _CHAIN_FEATURES_PATH,
     _TASK_DEFINITIONS,
     _TASKS_DB,
     _TASKS_LOCK,
     _get_task_snapshot,
     _execute_imagegen_pipeline,
+    _get_model_feature_context,
+    _canonicalize_chain_type,
+    _resolve_feature_name,
 )
 from .error_schema import make_error, make_validation_error, make_not_found_error
+from .image_requests import normalize_image_request, validate_image_plan
+from core.reference_inputs import ImageInputError
 
 
 def handle_run(params: dict) -> dict:
     """Unified image generation task execution interface."""
     if not isinstance(params, dict):
         return make_validation_error("Request params must be an object.")
+    params = params.copy()
+    if isinstance(params.get("chain"), list):
+        params["chain"] = [
+            item.copy() if isinstance(item, dict) else item
+            for item in params["chain"]
+        ]
 
     missing = []
     for req_field in ["task_type", "model", "prompt"]:
@@ -35,6 +47,25 @@ def handle_run(params: dict) -> dict:
         )
 
     task_type = params["task_type"]
+    if (
+        not isinstance(task_type, str)
+        or not isinstance(params["model"], str)
+        or not isinstance(params["prompt"], str)
+        or not task_type.strip()
+        or not params["model"].strip()
+        or not params["prompt"].strip()
+    ):
+        return make_validation_error(
+            "task_type, model, and prompt must be strings.",
+            invalid_fields={
+                key: "Expected a non-empty string"
+                for key in ("task_type", "model", "prompt")
+                if not isinstance(params.get(key), str) or not params.get(key, "").strip()
+            },
+        )
+    task_type = task_type.strip()
+    params["task_type"] = task_type
+    params["model"] = params["model"].strip()
     valid_tasks = [t["task_type"] for t in _TASK_DEFINITIONS]
     if task_type not in valid_tasks:
         return make_validation_error(
@@ -53,8 +84,14 @@ def handle_run(params: dict) -> dict:
     if params["model"] not in all_models:
         return make_not_found_error("model", params["model"])
 
+    try:
+        normalize_image_request(params)
+    except ImageInputError as exc:
+        return make_validation_error(str(exc), invalid_fields={"images": str(exc)})
+
     task_requirements = {
         "txt2img": ("width", "height"),
+        "reference": ("width", "height", "images"),
         "img2img": ("image",),
         "inpaint": ("image",),
         "outpaint": ("image", "pad_left", "pad_right", "pad_top", "pad_bottom"),
@@ -63,6 +100,7 @@ def handle_run(params: dict) -> dict:
     missing_task_fields = [
         field
         for field in task_requirements.get(task_type, ())
+        if not (field == "image" and params.get("images"))
         if field not in params or params[field] is None or params[field] == ""
     ]
     if missing_task_fields:
@@ -71,10 +109,123 @@ def handle_run(params: dict) -> dict:
             missing_fields=missing_task_fields,
         )
 
+    invalid_fields = {}
+
+    chain_features = _load_yaml(_CHAIN_FEATURES_PATH)
+
+    def _number(name, *, integer=False, minimum=None, maximum=None):
+        if name not in params:
+            return
+        raw_value = params[name]
+        if raw_value is None or (isinstance(raw_value, str) and not raw_value.strip()):
+            if name in task_requirements.get(task_type, ()):
+                invalid_fields[name] = "Expected a non-empty number"
+            else:
+                # 可选数值的空值等同于未提供，统一沿用执行层的任务 / 模型默认值。
+                params.pop(name)
+            return
+        if isinstance(raw_value, bool):
+            invalid_fields[name] = "Expected a number" if not integer else "Expected an integer"
+            return
+        try:
+            numeric_value = float(raw_value)
+        except (TypeError, ValueError):
+            invalid_fields[name] = "Expected a number" if not integer else "Expected an integer"
+            return
+        if numeric_value != numeric_value or numeric_value in (float("inf"), float("-inf")):
+            invalid_fields[name] = "Must be finite"
+            return
+        if integer and not numeric_value.is_integer():
+            invalid_fields[name] = "Expected an integer"
+            return
+        value = int(numeric_value) if integer else numeric_value
+        if minimum is not None and value < minimum:
+            invalid_fields[name] = f"Must be >= {minimum}"
+        elif maximum is not None and value > maximum:
+            invalid_fields[name] = f"Must be <= {maximum}"
+        else:
+            params[name] = value
+
+    for field in ("width", "height"):
+        if task_type == "txt2img" or field in params:
+            _number(field, integer=True, minimum=1, maximum=8192)
+    for field in ("pad_left", "pad_right", "pad_top", "pad_bottom"):
+        _number(field, integer=True, minimum=0, maximum=4096)
+    _number("upscale_by", minimum=0.01, maximum=4.0)
+    _number("denoise", minimum=0.0, maximum=1.0)
+    _number("steps", integer=True, minimum=1, maximum=100)
+    _number("cfg", minimum=0.0, maximum=100.0)
+    _number("batch_size", integer=True, minimum=1, maximum=CONFIG.max_batch_size)
+    _number("seed", integer=True, minimum=-1)
+
+    for field in ("sampler", "scheduler", "negative_prompt"):
+        if field in params and params[field] is not None and not isinstance(params[field], str):
+            invalid_fields[field] = "Expected a string"
+
+    if "async_execution" in params:
+        async_execution = params["async_execution"]
+        if isinstance(async_execution, str):
+            normalized = async_execution.strip().casefold()
+            if normalized in {"true", "1"}:
+                params["async_execution"] = True
+            elif normalized in {"false", "0"}:
+                params["async_execution"] = False
+            else:
+                invalid_fields["async_execution"] = "Expected true/false or 1/0"
+        elif not isinstance(async_execution, bool):
+            invalid_fields["async_execution"] = "Expected a boolean"
+
+    if "chain" in params:
+        chain = params["chain"]
+        if not isinstance(chain, list):
+            invalid_fields["chain"] = "Expected a list of objects"
+        else:
+            context = _get_model_feature_context(params["model"])
+            for index, item in enumerate(chain):
+                field_name = f"chain[{index}]"
+                if not isinstance(item, dict):
+                    invalid_fields[field_name] = "Expected an object"
+                    continue
+                injector_type = item.get("injector_type")
+                if not isinstance(injector_type, str) or not injector_type.strip():
+                    invalid_fields[field_name] = "injector_type must be a non-empty string"
+                    continue
+                injector_type = injector_type.strip()
+                injector_type = _canonicalize_chain_type(injector_type)
+                item["injector_type"] = injector_type
+                feature_name = _resolve_feature_name(injector_type)
+                if feature_name is None:
+                    return make_validation_error(
+                        f"Unknown chain/injector type '{injector_type}'.",
+                        invalid_fields={field_name: "Unknown injector_type"},
+                    )
+                feature_data = chain_features.get(feature_name, {})
+                aliases = feature_data.get("chains", feature_name)
+                aliases = aliases if isinstance(aliases, list) else [aliases]
+                enabled = context and context["enabled_chains"]
+                supports = context and context["supports_chain"]
+                if not enabled or not any(alias in enabled for alias in [feature_name, *aliases]):
+                    return make_error(
+                        "FEATURE_NOT_SUPPORTED",
+                        f"Model '{params['model']}' does not support chain '{injector_type}'.",
+                        {"model": params["model"], "chain": injector_type},
+                    )
+                if supports and not any(supports(alias) for alias in [injector_type, feature_name, *aliases]):
+                    return make_error(
+                        "FEATURE_NOT_SUPPORTED",
+                        f"Model '{params['model']}' does not support chain '{injector_type}'.",
+                        {"model": params["model"], "chain": injector_type},
+                    )
+
+    if invalid_fields:
+        return make_validation_error("Request contains invalid parameters.", invalid_fields=invalid_fields)
+
     try:
-        batch_size = int(params.get("batch_size", 1))
-    except (TypeError, ValueError):
-        batch_size = 0
+        validate_image_plan(params)
+    except ImageInputError as exc:
+        return make_validation_error(str(exc), invalid_fields={"image_plan": str(exc)})
+
+    batch_size = params.get("batch_size", 1)
     if not 1 <= batch_size <= CONFIG.max_batch_size:
         return make_validation_error(
             f"batch_size must be between 1 and {CONFIG.max_batch_size}.",

tests/test_gpu_boundary.py

@@ -168,6 +168,20 @@ class GpuBoundaryTests(unittest.TestCase):
         self.assertNotIn("_cancel_event", self.payloads[0]["ui_inputs"])
         self.release_models.assert_not_called()
 
+    def test_unsupported_stale_reference_does_not_block_img2img(self):
+        from PIL import Image
+
+        self.pipeline.run(self.inputs(task_type="img2img", img2img_image=Image.new("RGB", (64, 64)), qwen_image_edit_data=[Image.new("RGB", (32, 32))], positive_prompt="保留图1"), _LockedProgress())
+        self.download.assert_called_once()
+        self.assertEqual(self.payloads[0]["ui_inputs"]["qwen_image_edit_data"], [])
+        self.assertEqual(self.payloads[0]["ui_inputs"]["positive_prompt"], "保留image 1")
+
+    def test_invalid_numbered_reference_rejected_before_download_and_gpu(self):
+        with self.assertRaisesRegex(RuntimeError, "本次只有 0"):
+            self.pipeline.run(self.inputs(positive_prompt="img2"), _LockedProgress())
+        self.download.assert_not_called()
+        self.assertEqual(self.payloads, [])
+
     def test_cancellation_after_download_stays_on_cpu(self):
         cancellation = threading.Event()
         self.download.side_effect = lambda *args, **kwargs: cancellation.set()

tests/test_image_inputs.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import unittest
+import itertools
+import types
+from unittest import mock
+
+from PIL import Image
+
+from core.execution_plan import available_run_modes, build_execution_plan, ExecutionPlanError
+from core.reference_inputs import normalize_image_references, reference_choices
+from mcp_tools.get_model_features import handle_get_model_features
+from mcp_tools.run import handle_run
+from tests.test_execution_plan import base_inputs
+
+
+EDIT = "lightx2v/Qwen-Image-Edit-2511-Lightning"
+ANIMA = "circlestone-labs/Anima-Turbo-v1.0"
+
+
+class ImageInputContractTests(unittest.TestCase):
+    def test_aliases_are_one_based_and_idempotent(self):
+        prompt = "将图1图2合并；img1 在左、IMG2在右；image 1 保留主体，参考图２提供背景。"
+        expected = "将image 1 image 2合并；image 1 在左、image 2在右；image 1 保留主体，image 2提供背景。"
+        self.assertEqual(normalize_image_references(prompt, 2), expected)
+        self.assertEqual(normalize_image_references(expected, 2), expected)
+
+    def test_filenames_identifiers_and_quoted_literals_are_not_rewritten(self):
+        prompt = 'img1.png /images/img2 myimg3 img4_suffix "图9" “img8” `image7` 「图6」'
+        self.assertEqual(normalize_image_references(prompt, 0), prompt)
+
+    def test_chinese_compound_words_are_not_image_aliases(self):
+        prompts = (
+            "电影感，横向构图16:9，日落风景",
+            "横向构图 １６：９，地图1:10000，草图2张，插图3幅",
+            "蓝图1份，位图2张，视图3个，绘图4幅，截图5张，贴图6张",
+        )
+        for prompt in prompts:
+            for count in (0, 2):
+                with self.subTest(prompt=prompt, count=count):
+                    self.assertEqual(normalize_image_references(prompt, count), prompt)
+            plan = build_execution_plan(base_inputs(positive_prompt=prompt))
+            self.assertEqual(plan[0].inputs["positive_prompt"], prompt)
+
+    def test_explicit_aliases_next_to_chinese_text_and_numeric_details_still_work(self):
+        prompt = "横向构图16:9；保留图1，图1图2合并；图1：2只猫，参考图2：1只狗，图片1和img2。"
+        expected = "横向构图16:9；保留image 1，image 1 image 2合并；image 1：2只猫，image 2：1只狗，image 1和image 2。"
+        self.assertEqual(normalize_image_references(prompt, 2), expected)
+        self.assertEqual(normalize_image_references(expected, 2), expected)
+        with self.assertRaisesRegex(ExecutionPlanError, "本次只有 2"):
+            normalize_image_references("横向构图16:9；保留图3", 2)
+
+    def test_missing_and_cross_image_references_fail_early(self):
+        for prompt in ("图0", "img0", "图3", "IMG3", "image 3"):
+            with self.subTest(prompt=prompt), self.assertRaisesRegex(ExecutionPlanError, "本次只有 2"):
+                normalize_image_references(prompt, 2)
+        with self.assertRaisesRegex(ExecutionPlanError, "当前图片"):
+            normalize_image_references("保留图1", 1, independent=True)
+
+    def test_high_level_list_excludes_stale_slots_without_mutating_them(self):
+        stale = [Image.new("RGB", (8, 8), "black"), Image.new("RGB", (8, 8), "white")]
+        pictures = [Image.new("RGB", (8, 8), "red"), Image.new("RGB", (8, 8), "blue")]
+        values = base_inputs(model_display_name=EDIT, task_type="reference", positive_prompt="图1左、img2右", qwen_image_edit_data=stale)
+        plan = build_execution_plan(values, images=pictures)
+        self.assertEqual(values["qwen_image_edit_data"], stale)
+        self.assertEqual(values["positive_prompt"], "图1左、img2右")
+        self.assertEqual(plan[0].inputs["qwen_image_edit_data"], pictures)
+        self.assertEqual(plan[0].inputs["positive_prompt"], "image 1左、image 2右")
+        self.assertEqual(plan[0].inputs["task_type"], "txt2img")
+        self.assertIn("2 张参考图", plan[0].caption)
+        self.assertEqual([item["id"] for item in plan[0].inputs["_image_references"]], ["img1", "img2"])
+        reordered = build_execution_plan(values, images=list(reversed(pictures)))
+        self.assertIs(reordered[0].inputs["qwen_image_edit_data"][0], pictures[1])
+
+    def test_reference_limit_never_truncates(self):
+        with self.assertRaisesRegex(ExecutionPlanError, "3 张"):
+            build_execution_plan(base_inputs(task_type="reference", model_display_name=EDIT), images=["a", "b", "c", "d"])
+        with self.assertRaisesRegex(ExecutionPlanError, "不会截断"):
+            build_execution_plan(base_inputs(model_display_name=EDIT, qwen_image_edit_data=["a", "b", "c", "d"]))
+
+    def test_supported_native_slots_compact_and_unsupported_slots_do_not_count(self):
+        values = base_inputs(model_display_name=EDIT, positive_prompt="图1", qwen_image_edit_data=[None, "a", None])
+        plan = build_execution_plan(values)
+        self.assertEqual(len(plan[0].inputs["_image_references"]), 1)
+        plan = build_execution_plan(base_inputs(task_type="img2img", model_display_name=ANIMA, img2img_image="source", positive_prompt="图1", qwen_image_edit_data=["old"]))
+        self.assertEqual(plan[0].inputs["qwen_image_edit_data"], [])
+        self.assertEqual(plan[0].inputs["positive_prompt"], "image 1")
+
+    def test_discovery_matches_checkpoint_limits_and_modes(self):
+        edit = handle_get_model_features(EDIT)["image_workflows"]
+        self.assertEqual(edit["reference_roles"]["auto"]["max_images"], 3)
+        self.assertNotIn("style", edit["reference_roles"])
+        regular = handle_get_model_features("Qwen-Image")["image_workflows"]
+        self.assertFalse(regular["reference_supported"])
+        self.assertEqual(regular["run_modes_by_task"]["reference"], [])
+        self.assertNotIn("multi_reference", [mode for _, mode in available_run_modes("img2img", EDIT)])
+        self.assertNotIn("multi_independent", [mode for _, mode in available_run_modes("inpaint", EDIT)])
+        self.assertIn("identity", reference_choices("Krea-2-Turbo"))
+
+    def test_qwen_injector_wires_ordered_images_to_both_prompt_nodes(self):
+        from chain_injectors.qwen_image_edit_injector import inject
+
+        ids = itertools.count(10)
+        assembler = types.SimpleNamespace(
+            node_map={"ksampler": "1", "pos_prompt": "2", "neg_prompt": "3", "vae_loader": "4"},
+            workflow={
+                "1": {"class_type": "KSampler", "inputs": {"model": ["5", 0]}},
+                "2": {"class_type": "TextEncodeQwenImageEditPlus", "inputs": {"prompt": "image 1左，image 2右"}},
+                "3": {"class_type": "TextEncodeQwenImageEditPlus", "inputs": {"prompt": ""}},
+                "4": {"class_type": "VAELoader", "inputs": {}},
+            },
+            _get_unique_id=lambda: str(next(ids)),
+            _get_node_template=lambda kind: {"class_type": kind, "inputs": {}, "_meta": {}},
+        )
+        inject(assembler, {}, ["blue.png", "red.png"])
+        for node_id in ("2", "3"):
+            for i, filename in enumerate(("blue.png", "red.png"), 1):
+                scale_id = assembler.workflow[node_id]["inputs"][f"image{i}"][0]
+                load_id = assembler.workflow[scale_id]["inputs"]["image"][0]
+                self.assertEqual(assembler.workflow[load_id]["inputs"]["image"], filename)
+
+    def test_multiple_native_namespaces_cannot_share_numbered_prompt(self):
+        with self.assertRaisesRegex(ExecutionPlanError, "多个原生参考链"):
+            build_execution_plan(base_inputs(positive_prompt="图1", krea2_identity_edit_data=["a"], krea2_reference_edit_data=["b"]))
+
+
+class ApiImageRequestTests(unittest.TestCase):
+    def request(self, **overrides):
+        return dict(task_type="reference", model=EDIT, prompt="图1左，img2右", width=512, height=512, async_execution=True, **overrides)
+
+    def test_numbered_fields_sort_numerically_and_keep_caller_data(self):
+        request = self.request(**{"图2": "blue", "IMG1": "red"})
+        with mock.patch("mcp_tools.run.submit_background") as submit:
+            result = handle_run(request)
+        self.assertEqual(result["status"], "queued")
+        self.assertEqual(submit.call_args.args[2]["images"], ["red", "blue"])
+        self.assertEqual(submit.call_args.args[2]["run_mode"], "multi_reference")
+        self.assertEqual(request["图2"], "blue")
+        self.assertNotIn("images", request)
+
+    def test_invalid_image_contract_never_enters_queue(self):
+        cases = [
+            {"images": ["a"], "img1": "b"}, {"image": "a", "images": ["b"]},
+            {"img1": "a", "图1": "b"}, {"img2": "a"}, {"img0": "a"},
+            {"images": []}, {"images": [None]}, {"images": [1]},
+            {"images": ["a"]}, {"images": ["a", "b", "c", "d"]},
+            {"images": ["a", "b"], "reference_role": "style"},
+            {"images": ["a", "b"], "chain": [{"injector_type": "qwen_image_edit", "image": "c"}]},
+            {"images": ["a", "b"], "run_mode": "nonsense"},
+        ]
+        for case in cases:
+            with self.subTest(case=case), mock.patch("mcp_tools.run.submit_background") as submit:
+                result = handle_run(self.request(**case))
+                self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
+                submit.assert_not_called()
+
+    def test_legacy_chain_and_single_image_calls_still_queue(self):
+        cases = [
+            {"task_type": "txt2img", "model": EDIT, "prompt": "img1左，图2右", "width": 512, "height": 512,
+             "chain": [{"injector_type": "qwen_image_edit", "image": "a"}, {"injector_type": "qwen_image_edit", "image": "b"}]},
+            {"task_type": "img2img", "model": ANIMA, "prompt": "图1改为水彩", "image": "a"},
+            {"task_type": "img2img", "model": ANIMA, "prompt": "当前图片转水彩", "images": ["a", "b"], "run_mode": "multi_independent"},
+        ]
+        for params in cases:
+            with self.subTest(params=params), mock.patch("mcp_tools.run.submit_background") as submit:
+                result = handle_run({**params, "async_execution": True})
+                self.assertEqual(result["status"], "queued")
+                submit.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_validation.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import unittest
+from unittest import mock
 
 from PIL import Image
 
@@ -12,10 +13,28 @@ from mcp_tools.run import handle_run
 
 
 class McpValidationTests(unittest.TestCase):
+    @staticmethod
+    def _request(**overrides):
+        params = {
+            "task_type": "txt2img",
+            "model": "Krea-2-Turbo",
+            "prompt": "测试",
+            "width": 1024,
+            "height": 1024,
+        }
+        params.update(overrides)
+        return params
+
     def test_requires_common_fields(self):
         result = handle_run({})
         self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
 
+        blank_prompt = handle_run(self._request(prompt="   "))
+        self.assertEqual(blank_prompt["error"]["code"], "INVALID_PARAMS")
+        self.assertIn(
+            "prompt", blank_prompt["error"]["details"]["invalid_fields"]
+        )
+
     def test_requires_task_specific_fields_before_execution(self):
         result = handle_run(
             {
@@ -28,18 +47,164 @@ class McpValidationTests(unittest.TestCase):
         self.assertEqual(set(result["error"]["details"]["missing_fields"]), {"width", "height"})
 
     def test_batch_limit_is_enforced(self):
-        result = handle_run(
-            {
-                "task_type": "txt2img",
-                "model": "Krea-2-Turbo",
-                "prompt": "测试",
-                "width": 1024,
-                "height": 1024,
-                "batch_size": 999,
-            }
-        )
+        result = handle_run(self._request(batch_size=999))
         self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
 
+    def test_async_execution_accepts_common_boolean_strings(self):
+        for raw_value in ("true", "1", " TRUE "):
+            with self.subTest(raw_value=raw_value), mock.patch(
+                "mcp_tools.run.submit_background"
+            ) as submit:
+                result = handle_run(
+                    self._request(async_execution=raw_value)
+                )
+                self.assertEqual(result["status"], "queued")
+                self.assertIs(submit.call_args.args[2]["async_execution"], True)
+
+        for raw_value in ("false", "0", " FALSE "):
+            with self.subTest(raw_value=raw_value), mock.patch(
+                "mcp_tools.run._execute_imagegen_pipeline"
+            ) as execute:
+                handle_run(self._request(async_execution=raw_value))
+                self.assertIs(execute.call_args.args[1]["async_execution"], False)
+
+        result = handle_run(self._request(async_execution="yes"))
+        self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
+        self.assertIn(
+            "async_execution", result["error"]["details"]["invalid_fields"]
+        )
+
+    def test_trimmed_task_type_is_forwarded_to_execution(self):
+        with mock.patch("mcp_tools.run.submit_background") as submit:
+            result = handle_run(
+                self._request(task_type=" txt2img ", async_execution=True)
+            )
+        self.assertEqual(result["status"], "queued")
+        submitted_params = submit.call_args.args[2]
+        self.assertEqual(submitted_params["task_type"], "txt2img")
+
+    def test_numeric_fields_are_normalized_and_bounded(self):
+        with mock.patch("mcp_tools.run.submit_background") as submit:
+            accepted = handle_run(
+                self._request(
+                    width="1024",
+                    height=1024.0,
+                    steps=20.0,
+                    batch_size="2",
+                    async_execution=True,
+                )
+            )
+        self.assertEqual(accepted["status"], "queued")
+        submitted = submit.call_args.args[2]
+        self.assertEqual(submitted["width"], 1024)
+        self.assertEqual(submitted["height"], 1024)
+        self.assertEqual(submitted["steps"], 20)
+        self.assertEqual(submitted["batch_size"], 2)
+
+        for field, value in (
+            ("height", 0),
+            ("denoise", 1.1),
+            ("seed", True),
+            ("steps", 1.5),
+        ):
+            with self.subTest(field=field):
+                with mock.patch("mcp_tools.run._execute_imagegen_pipeline"):
+                    result = handle_run(self._request(**{field: value}))
+                self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
+                self.assertIn(field, result["error"]["details"]["invalid_fields"])
+
+    def test_empty_optional_numbers_behave_like_omitted_fields(self):
+        fields = ("steps", "cfg", "denoise", "batch_size", "seed")
+        for empty in (None, "", " \t "):
+            for asynchronous in (True, False):
+                with self.subTest(empty=empty, asynchronous=asynchronous), mock.patch(
+                    "mcp_tools.run.submit_background"
+                ) as submit, mock.patch("mcp_tools.run._execute_imagegen_pipeline") as execute:
+                    request = self._request(**dict.fromkeys(fields, empty), async_execution=asynchronous)
+                    result = handle_run(request)
+                    if asynchronous:
+                        self.assertEqual(result["status"], "queued")
+                        submitted = submit.call_args.args[2]
+                    else:
+                        execute.assert_called_once()
+                        submitted = execute.call_args.args[1]
+                    for field in fields:
+                        self.assertNotIn(field, submitted)
+                        self.assertEqual(request[field], empty)
+
+    def test_empty_required_numbers_remain_invalid(self):
+        requests = (
+            ("width", self._request()),
+            ("height", self._request()),
+            ("pad_left", self._request(task_type="outpaint", image="opaque-source", pad_left=0, pad_right=0, pad_top=0, pad_bottom=0)),
+            ("upscale_by", self._request(task_type="hires_fix", image="opaque-source", upscale_by=1.5)),
+        )
+        for field, request in requests:
+            for empty in (None, "", " \t "):
+                with self.subTest(field=field, empty=empty), mock.patch("mcp_tools.run.submit_background") as submit:
+                    result = handle_run({**request, field: empty, "async_execution": True})
+                    self.assertEqual(result["error"]["code"], "INVALID_PARAMS")
+                    details = result["error"]["details"]
+                    self.assertIn(field, [*details.get("missing_fields", []), *details.get("invalid_fields", {})])
+                    submit.assert_not_called()
+
+    def test_zero_optional_numbers_are_not_treated_as_empty(self):
+        with mock.patch("mcp_tools.run.submit_background") as submit:
+            result = handle_run(self._request(cfg=0, denoise=0, seed=0, async_execution=True))
+        self.assertEqual(result["status"], "queued")
+        submitted = submit.call_args.args[2]
+        self.assertEqual({field: submitted[field] for field in ("cfg", "denoise", "seed")}, {"cfg": 0, "denoise": 0, "seed": 0})
+        rejected = handle_run(self._request(batch_size=0))
+        self.assertEqual(rejected["error"]["code"], "INVALID_PARAMS")
+
+    def test_chain_shape_and_model_capability_are_validated(self):
+        invalid_shape = handle_run(self._request(chain={"injector_type": "lora"}))
+        self.assertEqual(invalid_shape["error"]["code"], "INVALID_PARAMS")
+
+        invalid_item = handle_run(self._request(chain=["lora"]))
+        self.assertEqual(invalid_item["error"]["code"], "INVALID_PARAMS")
+
+        unknown = handle_run(
+            self._request(chain=[{"injector_type": "does_not_exist"}])
+        )
+        self.assertEqual(unknown["error"]["code"], "INVALID_PARAMS")
+
+        unsupported = handle_run(
+            self._request(chain=[{"injector_type": "qwen_image_edit"}])
+        )
+        self.assertEqual(unsupported["error"]["code"], "FEATURE_NOT_SUPPORTED")
+
+    def test_supported_chain_is_forwarded_without_schema_policing(self):
+        request = self._request(
+            chain=[
+                {
+                    "injector_type": "lora",
+                    "source": "Civitai",
+                    "lora_value": "123",
+                    "custom_field": "kept-for-forward-compatibility",
+                }
+            ],
+            async_execution=True,
+        )
+        with mock.patch("mcp_tools.run.submit_background") as submit:
+            result = handle_run(request)
+
+        self.assertEqual(result["status"], "queued")
+        submitted_item = submit.call_args.args[2]["chain"][0]
+        self.assertEqual(
+            submitted_item["custom_field"], "kept-for-forward-compatibility"
+        )
+
+    def test_legacy_boogu_alias_is_canonicalized(self):
+        request = self._request(
+            model="Boogu-Image-Edit-Turbo",
+            chain=[{"injector_type": "boogu_edit", "image": "opaque-reference"}],
+        )
+        with mock.patch("mcp_tools.run.submit_background", return_value=None):
+            result = handle_run({**request, "async_execution": True})
+        self.assertEqual(result["status"], "queued")
+        self.assertEqual(request["chain"][0]["injector_type"], "boogu_edit")
+
     def test_private_image_urls_are_rejected(self):
         with self.assertRaises(ValueError):
             _validate_public_image_url("http://127.0.0.1/private.png")
@@ -76,6 +241,30 @@ class McpValidationTests(unittest.TestCase):
         self.assertNotIn("qwen_image_edit", regular["supported_features"])
         self.assertIn("qwen_image_edit", editing["supported_features"])
 
+    def test_model_features_keep_legacy_task_list(self):
+        result = handle_get_model_features("Krea-2-Turbo")
+        self.assertNotIn("task_capabilities_known", result)
+        self.assertIn("txt2img", result["supported_tasks"])
+
+    def test_transport_handlers_import_without_gradio(self):
+        import subprocess
+        import sys
+
+        command = (
+            "import builtins; real_import = builtins.__import__; "
+            "builtins.__import__ = lambda name, *a, **k: "
+            "(_ for _ in ()).throw(ImportError('blocked')) "
+            "if name == 'gradio' else real_import(name, *a, **k); "
+            "import mcp_tools; assert callable(mcp_tools.handle_run)"
+        )
+        result = subprocess.run(
+            [sys.executable, "-c", command],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_ui_smoke.py

@@ -4,6 +4,7 @@ import os
 import sys
 import types
 import unittest
+import tempfile
 from unittest import mock
 
 os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
@@ -62,6 +63,82 @@ class UnifiedUiSmokeTests(unittest.TestCase):
                 "get_chain_schema",
             },
         )
+        input_update = next(fn for fn in app.fns.values() if getattr(fn.fn, "__qualname__", "") == "attach_input_events.<locals>.update")
+        args = [component.value for component in input_update.inputs]
+        args[:4] = ["reference", "lightx2v/Qwen-Image-Edit-2511-Lightning", "single", "style"]
+        updates = input_update.fn(*args)
+        self.assertEqual(updates[9]["value"], "multi_reference")
+        self.assertEqual(updates[10]["value"], "auto")
+        self.assertNotIn("value", updates[15])  # 切任务/模型只改引导，不覆盖用户 Prompt。
+        self.assertFalse(updates[1]["visible"])
+        self.assertFalse(updates[14]["visible"])
+        run = next(fn for fn in app.fns.values() if getattr(fn.fn, "__qualname__", "") == "create_unified_run_event.<locals>.execute")
+        args = [component.value for component in run.inputs]
+        args[0] = "reference"
+        args[1] = "lightx2v/Qwen-Image-Edit-2511-Lightning"
+        failure = run.fn(types.SimpleNamespace(session_hash="test-invalid-images"), lambda *_args, **_kwargs: None, *args)
+        self.assertEqual(failure[0], gradio.update())
+        self.assertIn("本次未完成", failure[1])
+        self.assertIn("不是本次新生成", failure[1])
+        app.close()
+
+    def test_input_context_uses_only_visible_inputs_and_current_model(self):
+        from ui.events.input_handlers import input_context
+        from ui.guidance import model_hint, prompt_guidance
+
+        edit = "lightx2v/Qwen-Image-Edit-2511-Lightning"
+        ref = input_context("reference", edit, "single", "style", ["new-a", "new-b"], "old-source", {"qwen_image_edit_data": ["old-ref"]})
+        self.assertEqual(ref["mode"], "multi_reference")
+        self.assertEqual(ref["role"], "auto")
+        self.assertEqual([image for image, _ in ref["preview"]], ["new-a", "new-b"])
+        self.assertFalse(ref["native"])
+        unsupported = input_context("reference", "Qwen-Image", "multi_reference", "auto")
+        self.assertFalse(unsupported["can_run"])
+        self.assertNotIn("Qwen 图片编辑", model_hint("Qwen-Image"))
+        independent = input_context("img2img", edit, "multi_independent", "auto", ["new"], "old")
+        self.assertEqual(independent["preview"][0][0], "new")
+        self.assertIn("当前图片", independent["prompt"]["help"])
+        self.assertNotEqual(prompt_guidance("txt2img", "single", edit)["examples"], prompt_guidance("reference", "multi_reference", edit)["examples"])
+
+    def test_api_reference_legacy_and_batch_execution_use_shared_plan(self):
+        from PIL import Image
+        import gradio.processing_utils as pu
+        from core import settings
+        from mcp_tools import common
+        from mcp_tools.run import handle_run
+
+        red, blue = Image.new("RGB", (32, 32), "red"), Image.new("RGB", (48, 32), "blue")
+        cases = [
+            {"task_type": "reference", "img1": red, "图2": blue, "prompt": "图1左，img2右"},
+            {"task_type": "txt2img", "chain": [{"injector_type": "qwen_image_edit", "image": red}, {"injector_type": "qwen_image_edit", "image": blue}], "prompt": "图1左，img2右"},
+            {"task_type": "img2img", "images": [red, blue], "run_mode": "multi_independent", "prompt": "当前图片转水彩"},
+            {"task_type": "img2img", "image": red, "prompt": "图1转水彩"},
+        ]
+        for case in cases:
+            called = []
+
+            def generate(ui_inputs, progress):
+                called.append(ui_inputs)
+                return [red.copy()]
+
+            with self.subTest(task=case["task_type"]), tempfile.TemporaryDirectory() as tmp, mock.patch.object(settings, "OUTPUT_DIR", tmp), mock.patch.object(
+                sys.modules["core.generation_logic"], "sd_image_pipeline", types.SimpleNamespace(run=generate), create=True
+            ), mock.patch.object(pu, "save_file_to_cache", side_effect=lambda path, cache_dir: path), mock.patch.dict(common._TASKS_DB, {}, clear=True):
+                response = handle_run({"model": "lightx2v/Qwen-Image-Edit-2511-Lightning", "width": 512, "height": 512, "seed": 123, **case})
+                self.assertEqual(response["status"], "completed", response)
+                result = response["result"]
+                self.assertEqual(len(result["images"]), len(called))
+                self.assertFalse(result["partial_success"])
+                self.assertEqual(result["seed"], 123)
+                if case["task_type"] in {"reference", "txt2img"}:
+                    self.assertEqual(len(called), 1)
+                    self.assertEqual(called[0]["positive_prompt"], "image 1左，image 2右")
+                    self.assertEqual([img.getpixel((0, 0)) for img in called[0]["qwen_image_edit_data"]], [(255, 0, 0), (0, 0, 255)])
+                    self.assertEqual(len(result["image_references"]), 2)
+                elif case.get("images"):
+                    self.assertEqual(len(called), 2)
+                    self.assertEqual([values["img2img_image"].size for values in called], [(32, 32), (48, 32)])
+                    self.assertEqual(called[0]["img2img_denoise"], 0.7)
 
     def test_reference_budget_rejects_oversized_advanced_image(self):
         from PIL import Image
@@ -75,6 +152,26 @@ class UnifiedUiSmokeTests(unittest.TestCase):
                 {"reference_latent_data": [Image.new("1", (3000, 2000))]}
             )
 
+    def test_api_partial_result_keeps_source_mapping_and_explicit_failure(self):
+        from PIL import Image
+        import gradio.processing_utils as pu
+        from core import settings
+        from mcp_tools import common
+        from mcp_tools.run import handle_run
+
+        def generate(ui_inputs, progress):
+            return [ui_inputs["img2img_image"]] if ui_inputs["_source_image_index"] == 1 else []
+
+        with tempfile.TemporaryDirectory() as tmp, mock.patch.object(settings, "OUTPUT_DIR", tmp), mock.patch.object(
+            sys.modules["core.generation_logic"], "sd_image_pipeline", types.SimpleNamespace(run=generate), create=True
+        ), mock.patch.object(pu, "save_file_to_cache", side_effect=lambda path, cache_dir: path), mock.patch.dict(common._TASKS_DB, {}, clear=True), self.assertLogs("core.execution_plan", level="ERROR"):
+            response = handle_run({"task_type": "img2img", "model": "Krea-2-Turbo", "prompt": "当前图片转水彩", "run_mode": "multi_independent", "images": [Image.new("RGB", (32, 32)), Image.new("RGB", (32, 32))]})
+            self.assertEqual(response["status"], "completed")
+            self.assertEqual(len(response["result"]["images"]), 1)
+            self.assertTrue(response["result"]["partial_success"])
+            self.assertEqual([job["source_index"] for job in response["result"]["jobs"]], [1, 2])
+            self.assertEqual(response["result"]["jobs"][1]["status"], "failed")
+
     def test_model_download_preflight_preserves_disk_reserve(self):
         from imagegen_utils.app_utils import _assert_download_space
 

ui/events/input_handlers.py

@@ -0,0 +1,132 @@
+"""任务、模型、运行方式和有效图片预览使用同一个输入上下文。"""
+
+import gradio as gr
+
+from core.execution_plan import available_run_modes
+from core.reference_inputs import active_reference_groups, reference_choices
+from ui.guidance import prompt_guidance, task_help
+
+
+REFERENCE_COMPONENTS = {
+    "reference_latent_data": "all_reference_latent_components_flat",
+    "hidream_o1_reference_data": "all_hidream_o1_reference_components_flat",
+    "joyai_reference_data": "all_joyai_reference_components_flat",
+    "krea2_identity_edit_data": "all_krea2_identity_edit_components_flat",
+    "krea2_reference_edit_data": "all_krea2_reference_edit_components_flat",
+    "qwen_image_edit_data": "all_qwen_image_edit_components_flat",
+    "boogu_edit_data": "all_boogu_edit_components_flat",
+    "reference_image_data": "all_reference_image_components_flat",
+}
+ROLE_LABELS = {"auto": "自动匹配模型能力", "identity": "人物 / 身份参考", "style": "风格参考"}
+RUN_LABELS = {"img2img": "开始重绘", "inpaint": "开始局部重绘", "outpaint": "开始扩图", "hires_fix": "开始高清修复"}
+
+
+def input_context(task, model, mode, role, files=None, source=None, native=None):
+    """纯状态计算，供事件和测试共用；不清空用户控件的图片或提示词。"""
+    choices = available_run_modes(task, model)
+    allowed = [value for _, value in choices]
+    selected = mode if mode in allowed else (allowed[0] if allowed else "multi_reference")
+    reference = selected == "multi_reference"
+    independent = selected in {"multi_independent", "multi_model_grid"}
+    pk = selected in {"model_pk", "multi_model_grid"}
+    uses_files = reference or independent
+    native_visible = task == "txt2img" and selected == "single"
+    roles = reference_choices(model)
+    selected_role = role if role in roles else "auto"
+    info = roles.get(selected_role)
+    pictures = []
+    if uses_files:
+        pictures = [(getattr(value, "name", value), "") for value in (files or [])]
+    elif task in {"img2img", "inpaint", "outpaint", "hires_fix"} and source is not None:
+        pictures = [(source, "")]
+    elif native_visible:
+        groups = active_reference_groups({"model_display_name": model, **(native or {})})
+        pictures = [(value, chain) for chain, values in groups.items() for value in values]
+
+    preview = [
+        (value, f"输入 {i}（单独处理）" if independent else f"图{i} / img{i}" + (f" · {chain}" if chain else ""))
+        for i, (value, chain) in enumerate(pictures, 1)
+    ]
+    if reference:
+        hint = (
+            f"**本次有效参考图：{len(pictures)} 张 / 上限 {info['max_images']} 张**。"
+            "只使用本列表，按显示顺序编号；拖动可排序。下方旧槽位不参与。"
+            if info else "**当前模型不支持参考图编辑 / 融合。** 请选用 Qwen-Image-Edit 等编辑模型；不会自动改用普通重绘。"
+        )
+        if info and len(pictures) > info["max_images"]:
+            hint += " **已超限，请移除图片；不会静默丢图。**"
+    elif independent:
+        hint = f"**本次输入：{len(pictures)} 张**。逐张处理、共用提示词，不会让某张图参考其他图片。"
+    else:
+        hint = f"**本次有效图片：{len(pictures)} 张**。" + (
+            "图片编号按预览为准，空槽位不编号。" if pictures else "没有参考图；需要编辑素材时选择“参考图编辑 / 融合”。"
+        )
+    if pk:
+        hint += " 模型 PK 仅比较基础模型，忽略高级扩展 / VAE / PiD；Prompt、尺寸、种子共用。"
+    return {
+        "mode": selected, "choices": choices, "role": selected_role, "roles": roles,
+        "reference": reference, "independent": independent, "pk": pk,
+        "files": uses_files, "native": native_visible, "preview": preview, "hint": hint,
+        "can_run": not reference or bool(info),
+        "prompt": prompt_guidance(task, selected, model),
+    }
+
+
+def attach_input_events(prefix, components, demo):
+    def component(name):
+        return components[f"{name}_{prefix}"]
+
+    native_lists = [(key, components.get(f"{name}_{prefix}", [])) for key, name in REFERENCE_COMPONENTS.items()]
+    inputs = [component(name) for name in ("task_type", "base_model", "run_mode", "reference_role", "batch_images", "source_image", "inpaint_image_dict")]
+    inputs.extend(value for _, values in native_lists for value in values)
+
+    def update(task, model, mode, role, files, source, inpaint, *native_values):
+        index = 0
+        native = {}
+        for key, values in native_lists:
+            native[key] = list(native_values[index:index + len(values)])
+            index += len(values)
+        state = input_context(task, model, mode, role, files, (inpaint or {}).get("background") if task == "inpaint" else source, native)
+        prompt = state["prompt"]
+        mode_update = {"choices": state["choices"] or [("请选择支持参考图的模型", "multi_reference")]}
+        if mode != state["mode"]:
+            mode_update["value"] = state["mode"]
+        role_update = {
+            "choices": [(ROLE_LABELS[key], key) for key in state["roles"]] or [(ROLE_LABELS["auto"], "auto")],
+            "visible": state["reference"],
+        }
+        if role != state["role"]:
+            role_update["value"] = state["role"]
+        return (
+            task_help("reference" if state["reference"] else task),
+            gr.update(visible=task in {"img2img", "outpaint", "hires_fix"} and not state["files"]),
+            gr.update(label="源图片（图1 / img1）"),
+            gr.update(visible=task == "inpaint"),
+            gr.update(visible=task == "img2img"),
+            gr.update(visible=task == "outpaint"),
+            gr.update(visible=task == "hires_fix"),
+            gr.update(visible=task == "txt2img" or state["reference"]),
+            gr.update(value="开始编辑 / 融合" if state["reference"] else RUN_LABELS.get(task, "开始生成"), interactive=state["can_run"]),
+            gr.update(**mode_update), gr.update(**role_update),
+            gr.update(visible=state["pk"]), gr.update(visible=state["files"]),
+            gr.update(visible=state["pk"]), gr.update(visible=state["native"]),
+            gr.update(label=prompt["label"], placeholder=prompt["placeholder"]),
+            prompt["help"], gr.update(samples=prompt["examples"]),
+            state["hint"], gr.update(value=state["preview"], visible=bool(state["preview"])),
+        )
+
+    outputs = [component(name) for name in (
+        "task_help", "source_panel", "source_image", "inpaint_panel", "img2img_panel",
+        "outpaint_panel", "hires_panel", "size_panel", "run", "run_mode", "reference_role",
+        "pk_models", "batch_images", "pk_model_defaults", "native_reference_panel",
+        "prompt", "prompt_help", "prompt_examples", "input_hint", "input_preview",
+    )]
+    gr.on(
+        triggers=[value.change for value in inputs], fn=update, inputs=inputs, outputs=outputs,
+        queue=False, show_progress="hidden", api_name=False, show_api=False,
+    )
+    demo.load(update, inputs, outputs, queue=False, show_progress="hidden", api_name=False, show_api=False)
+    component("prompt_examples").click(
+        lambda values: values[0], [component("prompt_examples")], [component("prompt")],
+        queue=False, show_progress="hidden", api_name=False, show_api=False,
+    )

ui/events/main.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import gradio as gr
 
-from ui.guidance import model_hint, recommended_params, task_help
+from ui.guidance import model_hint, recommended_params
 from .chain_handlers import (
     create_anima_controlnet_lllite_event_handlers,
     create_boogu_edit_event_handlers,
@@ -33,6 +33,7 @@ from .change_handlers import (
     run_on_load,
 )
 from .run_handlers import create_run_event, create_unified_run_event
+from .input_handlers import attach_input_events
 
 
 CHAIN_EVENT_FACTORIES = (
@@ -63,33 +64,6 @@ def _append_if(target, component):
         target.append(component)
 
 
-def _task_switch_values(task_type):
-    uses_source = task_type in {"img2img", "outpaint", "hires_fix"}
-    labels = {
-        "img2img": "源图片（图生图）",
-        "outpaint": "源图片（扩图）",
-        "hires_fix": "源图片（高清修复）",
-    }
-    run_labels = {
-        "txt2img": "开始生成",
-        "img2img": "开始重绘",
-        "inpaint": "开始局部重绘",
-        "outpaint": "开始扩图",
-        "hires_fix": "开始高清修复",
-    }
-    return (
-        task_help(task_type),
-        gr.update(visible=uses_source),
-        gr.update(label=labels.get(task_type, "源图片")),
-        gr.update(visible=task_type == "inpaint"),
-        gr.update(visible=task_type == "img2img"),
-        gr.update(visible=task_type == "outpaint"),
-        gr.update(visible=task_type == "hires_fix"),
-        gr.update(visible=task_type == "txt2img"),
-        gr.update(value=run_labels.get(task_type, "开始生成")),
-    )
-
-
 def _select_quick_preset(preset):
     if preset == "__manual__":
         return gr.update()
@@ -283,25 +257,7 @@ def attach_event_handlers(ui_components, demo):
             )
 
         if task_type is None:
-            task_component = ui_components[f"task_type_{prefix}"]
-            task_component.change(
-                _task_switch_values,
-                [task_component],
-                [
-                    ui_components[f"task_help_{prefix}"],
-                    ui_components[f"source_panel_{prefix}"],
-                    ui_components[f"source_image_{prefix}"],
-                    ui_components[f"inpaint_panel_{prefix}"],
-                    ui_components[f"img2img_panel_{prefix}"],
-                    ui_components[f"outpaint_panel_{prefix}"],
-                    ui_components[f"hires_panel_{prefix}"],
-                    ui_components[f"size_panel_{prefix}"],
-                    ui_components[f"run_{prefix}"],
-                ],
-                show_progress="hidden",
-                api_name=False,
-                show_api=False,
-            )
+            attach_input_events(prefix, ui_components, demo)
             preset = ui_components[f"quick_preset_{prefix}"]
             preset.change(
                 _select_quick_preset,

ui/events/run_handlers.py

@@ -14,6 +14,7 @@ from core.execution_plan import (
 from core.generation_logic import generate_image_wrapper
 from core.runtime_config import CONFIG
 from core.task_scheduler import TaskCancelledError
+from core.reference_inputs import clear_native_references
 
 
 _CANCEL_LOCK = threading.Lock()
@@ -222,6 +223,8 @@ def create_unified_run_event(prefix: str, ui_components: dict):
 
         task_type = values["task_type"]
         run_mode = values.get("_run_mode", "single")
+        if task_type != "txt2img" or run_mode != "single":
+            clear_native_references(values)
         source_image = values.pop("source_image", None)
         batch_driven = run_mode in {MODE_MULTI_INDEPENDENT, MODE_MULTI_MODEL_GRID}
         if task_type == "img2img" and not batch_driven:
@@ -246,35 +249,6 @@ def create_unified_run_event(prefix: str, ui_components: dict):
     result_gallery = ui_components[f"result_{prefix}"]
     run_summary = ui_components[f"run_summary_{prefix}"]
 
-    def change_run_mode(mode):
-        is_pk = mode in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID}
-        uses_files = mode in {
-            MODE_MULTI_INDEPENDENT,
-            MODE_MULTI_MODEL_GRID,
-            MODE_MULTI_REFERENCE,
-        }
-        return (
-            gr.update(visible=is_pk),
-            gr.update(visible=uses_files),
-            gr.update(visible=mode == MODE_MULTI_REFERENCE),
-            gr.update(visible=is_pk),
-        )
-
-    ui_components[f"run_mode_{prefix}"].change(
-        fn=change_run_mode,
-        inputs=[ui_components[f"run_mode_{prefix}"]],
-        outputs=[
-            ui_components[f"pk_models_{prefix}"],
-            ui_components[f"batch_images_{prefix}"],
-            ui_components[f"reference_role_{prefix}"],
-            ui_components[f"pk_model_defaults_{prefix}"],
-        ],
-        queue=False,
-        show_progress="hidden",
-        api_name=False,
-        show_api=False,
-    )
-
     def execute(
         request: gr.Request,
         progress=gr.Progress(track_tqdm=True),
@@ -288,6 +262,8 @@ def create_unified_run_event(prefix: str, ui_components: dict):
             values = create_inputs(*args)
             values["_cancel_event"] = cancel_event
             run_mode = values.pop("_run_mode", "single")
+            if values["task_type"] == "reference":
+                run_mode = MODE_MULTI_REFERENCE
             pk_models = values.pop("_pk_models", []) or []
             uploaded_files = values.pop("_batch_images", []) or []
             reference_role = values.pop("_reference_role", "auto") or "auto"
@@ -317,7 +293,10 @@ def create_unified_run_event(prefix: str, ui_components: dict):
                 cancel_event=cancel_event,
             )
         except (TaskCancelledError, ExecutionPlanError) as exc:
-            raise gr.Error(str(exc)) from exc
+            return gr.update(), (
+                f"### 本次未完成\n\n{exc}\n\n"
+                "已保留之前的成功结果（如有）；它们不是本次新生成的图片。"
+            )
         finally:
             with _CANCEL_LOCK:
                 if _CANCEL_EVENTS.get(key) is cancel_event:

ui/guidance.py

@@ -4,15 +4,16 @@ from __future__ import annotations
 
 from core.settings import (
     ARCHITECTURES_CONFIG,
-    FEATURES_CONFIG,
     MODEL_DEFAULTS_CONFIG,
     MODEL_MAP_CHECKPOINT,
     MODEL_TYPE_MAP,
 )
 from imagegen_utils.app_utils import get_model_generation_defaults
+from core.model_capabilities import enabled_chains_for_model
 
 
 TASK_HELP = {
+    "reference": "**参考图编辑 / 融合**：一张图按指令修改，多张图整组参考；选择编辑模型，按顺序上传图片，再描述各图用途和要保留、修改的内容。",
     "txt2img": "**文生图**：选择模型，写清主体、动作、构图、风格和光线，再点击生成。",
     "img2img": "**图生图**：上传原图；重绘幅度越低越接近原图，越高改动越大。",
     "inpaint": "**局部重绘**：上传图片并涂抹需要修改的区域；提示词只描述希望出现的新内容。",
@@ -73,7 +74,7 @@ def model_hint(model_name: str) -> str:
     defaults = get_model_generation_defaults(
         model_name, workflow_type, MODEL_DEFAULTS_CONFIG
     )
-    features = FEATURES_CONFIG.get(workflow_type, {}).get("enabled_chains", [])
+    features = enabled_chains_for_model(model_name)
     feature_text = "、".join(FEATURE_NAMES.get(item, item) for item in features[:6]) or "基础生成"
     if len(features) > 6:
         feature_text += f" 等 {len(features)} 项"
@@ -88,7 +89,7 @@ def model_hint(model_name: str) -> str:
 
     edit_tip = ""
     if "edit" in model_name.lower():
-        edit_tip = " 这是编辑向模型，配合下方自动出现的参考图能力使用。"
+        edit_tip = " 这是编辑向模型，推荐选择“参考图编辑 / 融合”任务。"
 
     return (
         f"**{architecture} · 推荐参数**：{defaults.get('steps', 20)} 步，"
@@ -120,3 +121,39 @@ def recommended_params(model_name: str) -> tuple:
         defaults.get("sampler_name", "euler"),
         defaults.get("scheduler", "simple"),
     )
+
+
+def prompt_guidance(task_type: str, mode: str, model_name: str) -> dict:
+    reference = task_type == "reference" or mode == "multi_reference"
+    independent = mode in {"multi_independent", "multi_model_grid"}
+    if reference:
+        label = "3. 说明每张图的用途和修改要求"
+        placeholder = "保留图1的主体，参考图2的背景；写清哪些不变、哪些修改。只有一张图时只写图1。"
+        examples = [
+            ["保留图1的主体、构图和光线，将背景改成纯白色。"],
+            ["保留 img1 的主体，参考 img2 的背景和光线，生成自然融合的新画面。"],
+        ]
+        help_text = "图1 = img1 = image1，对应有效图片预览顺序；编号统一成 image 1 送入模型。第二条示例需要两张图。"
+    elif task_type == "txt2img":
+        label, placeholder = "3. 描述你想要的画面", "主体 + 动作 + 场景 + 构图 + 风格 + 光线"
+        if MODEL_TYPE_MAP.get(model_name) in {"Anima", "SDXL", "SD1.5", "NewBie-Image"}:
+            examples = [["1girl, silver hair, train station, rainy night, upper body, cinematic lighting, detailed illustration"],
+                        ["cityscape, night, neon lights, wet street, reflections, low angle, depth of field"]]
+        else:
+            examples = [["现代东方客厅，浅木色与米白配色，午后自然光，广角室内摄影"],
+                        ["未来城市夜景，湿润街道倒影，霓虹灯，低机位，电影概念设计"]]
+        help_text = "描述最终画面；要按图片修改或融合，请选择“参考图编辑 / 融合”。"
+    else:
+        label, placeholder = "3. 描述修改后的画面", "要保留的主体和构图 + 希望改变的部分；改动大小还受重绘幅度控制"
+        example = {
+            "img2img": "将当前图片转换为水彩插画，保留主体轮廓和构图，柔和色彩。",
+            "inpaint": "将涂抹区域替换为一束白色鲜花，匹配周围光线与透视。",
+            "outpaint": "向画面外延续现有背景，保持原有透视、色调与光照，衔接自然。",
+            "hires_fix": "保留当前图片的主体、配色和构图，提升细节清晰度，避免新增元素。",
+        }.get(task_type, "保留当前图片的主体，背景改为纯白。")
+        examples = [[example]]
+        help_text = "普通图生图是整体重绘，不等于精确指令编辑；低重绘幅度可能保留你想改变的内容。"
+    if independent:
+        help_text = "每张源图单独执行，共用同一条提示词；请写“当前图片”，不要用图1/img2跨图引用。需要两图同时参与请选参考图编辑 / 融合。"
+    help_text += " 不自动翻译或扩写；引号/反引号内的编号当作画面文字，不解析。"
+    return {"label": label, "placeholder": placeholder, "examples": examples, "help": help_text}

ui/layout.py

@@ -28,7 +28,7 @@ def build_ui(event_handler_function):
             """
             <section id="studio-hero">
               <h1>ImageGen Studio</h1>
-              <p>一套界面完成文生图、图生图、局部重绘、扩图和高清修复。模型与任务切换会保留你的提示词和已上传素材。</p>
+              <p>一套界面完成文生图、参考图编辑 / 融合、整体重绘、扩图和修复。切换任务与模型会保留提示词和素材，仅当前有效输入参与生成。</p>
             </section>
             <section class="studio-steps" aria-label="快速开始">
               <div class="studio-step"><b>① 选任务</b><span>先决定生成、编辑，还是修复图片</span></div>

ui/shared/studio_ui.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import gradio as gr
 
 from comfy_integration.nodes import SAMPLER_CHOICES, SCHEDULER_CHOICES
-from core.execution_plan import RUN_MODE_CHOICES
+from core.execution_plan import available_run_modes
 from core.runtime_config import CONFIG
 from core.settings import (
     ARCHITECTURES_CONFIG,
@@ -14,7 +14,7 @@ from core.settings import (
     MODEL_TYPE_MAP,
     RESOLUTION_MAP,
 )
-from ui.guidance import QUICK_PRESETS, model_hint, task_help
+from ui.guidance import QUICK_PRESETS, model_hint, prompt_guidance, task_help
 from .ui_components import (
     create_anima_controlnet_lllite_ui,
     create_boogu_edit_ui,
@@ -44,24 +44,19 @@ from .ui_components import (
 
 TASK_CHOICES = [
     ("✨ 文生图", "txt2img"),
-    ("🖼️ 图生图", "img2img"),
+    ("参考图编辑 / 融合", "reference"),
+    ("🖼️ 图生图（整体重绘）", "img2img"),
     ("🖌️ 局部重绘", "inpaint"),
     ("↔️ 扩图", "outpaint"),
     ("🔎 高清修复", "hires_fix"),
 ]
 
-PROMPT_EXAMPLES = [
-    ["一位银发少女站在雨夜车站，半身构图，电影感侧光，细腻插画，高质量"],
-    ["现代东方客厅，浅木色与米白配色，午后自然光，广角室内摄影"],
-    ["未来城市夜景，湿润街道倒影，霓虹灯，低机位，强烈纵深，电影概念设计"],
-]
-
-
 def create_ui() -> dict:
     prefix = "studio"
     components: dict = {"_task_prefixes": [(prefix, None)]}
     default_values = MODEL_DEFAULTS_CONFIG.get("Default", {})
     default_model = next(iter(MODEL_MAP_CHECKPOINT), None)
+    prompt_info = prompt_guidance("txt2img", "single", default_model)
     default_architecture = MODEL_TYPE_MAP.get(default_model, "SDXL")
     default_architecture_info = ARCHITECTURES_CONFIG.get("architectures", {}).get(
         default_architecture, {}
@@ -110,9 +105,9 @@ def create_ui() -> dict:
                     elem_classes="model-hint",
                 )
 
-            with gr.Accordion("批量、多图与模型 PK（可选）", open=False):
+            with gr.Accordion("图片输入与运行方式", open=True):
                 components[f"run_mode_{prefix}"] = gr.Dropdown(
-                    choices=RUN_MODE_CHOICES,
+                    choices=available_run_modes("txt2img", default_model),
                     value="single",
                     label="运行方式",
                     info="所有模式都顺序执行，不会同时把多个大模型塞进显存。",
@@ -132,7 +127,7 @@ def create_ui() -> dict:
                     file_types=["image"],
                     type="filepath",
                     allow_reordering=True,
-                    label="批量输入 / 多图参考",
+                    label="图片列表（按顺序编号，可拖动排序）",
                     visible=False,
                 )
                 with gr.Row():
@@ -152,29 +147,36 @@ def create_ui() -> dict:
                         info="Prompt、尺寸与种子保持一致；步数和 CFG 按模型推荐值。",
                         visible=False,
                     )
-                gr.Markdown(
+                components[f"input_hint_{prefix}"] = gr.Markdown(
                     "**怎么选：** 多图独立是 A→A′、B→B′；多图×多模型会执行图片和模型的组合；"
-                    "多图融合是把整组图片同时交给 Qwen/Mage-Flow/JoyAI/Boogu/Krea/Flux 等兼容模型。"
+                    "需要按指令修改或融合素材，请选择上方“参考图编辑 / 融合”。"
+                )
+                components[f"input_preview_{prefix}"] = gr.Gallery(
+                    label="本次有效图片（空槽位不编号）", columns=4, rows=1, height=200,
+                    object_fit="contain", interactive=False, visible=False,
+                    elem_id="input-image-preview",
                 )
 
             with gr.Group(elem_classes="prompt-card"):
                 components[f"prompt_{prefix}"] = gr.Textbox(
-                    label="3. 描述你想要的画面",
-                    placeholder="主体 + 动作 + 场景 + 构图 + 风格 + 光线",
+                    label=prompt_info["label"],
+                    placeholder=prompt_info["placeholder"],
                     lines=4,
                     max_lines=8,
                     value=default_values.get("positive_prompt", ""),
                 )
+                components[f"prompt_help_{prefix}"] = gr.Markdown(prompt_info["help"])
                 with gr.Accordion("负面提示词（可选）", open=False):
                     components[f"neg_prompt_{prefix}"] = gr.Textbox(
                         label="不希望出现的内容",
                         placeholder="例如：低质量、模糊、多余手指、水印",
                         lines=2,
                         value=default_values.get("negative_prompt", ""),
+                        info="是否生效取决于模型和 CFG；CFG=1 的常见采样路径通常不使用负面分支。",
                     )
-                gr.Examples(
-                    examples=PROMPT_EXAMPLES,
-                    inputs=[components[f"prompt_{prefix}"]],
+                components[f"prompt_examples_{prefix}"] = gr.Dataset(
+                    samples=prompt_info["examples"],
+                    components=["textbox"], type="values",
                     label="点一下填入示例",
                 )
 
@@ -351,14 +353,17 @@ def create_ui() -> dict:
     components.update(create_embedding_ui(prefix))
     components.update(create_style_ui(prefix))
     components.update(create_conditioning_ui(prefix))
-    components.update(create_reference_latent_ui(prefix))
-    components.update(create_hidream_o1_reference_ui(prefix))
-    components.update(create_joyai_reference_ui(prefix))
-    components.update(create_krea2_identity_edit_ui(prefix))
-    components.update(create_krea2_reference_edit_ui(prefix))
-    components.update(create_qwen_image_edit_ui(prefix))
-    components.update(create_boogu_edit_ui(prefix))
-    components.update(create_reference_image_ui(prefix))
+    with gr.Group() as native_reference_panel:
+        gr.Markdown("原生参考槽位（兼容旧用法）：仅普通文生图使用。参考图编辑 / 融合任务只读取上方图片列表；切换任务保留素材，但不混入本次生成。编号按有效图片预览为准。")
+        components.update(create_reference_latent_ui(prefix))
+        components.update(create_hidream_o1_reference_ui(prefix))
+        components.update(create_joyai_reference_ui(prefix))
+        components.update(create_krea2_identity_edit_ui(prefix))
+        components.update(create_krea2_reference_edit_ui(prefix))
+        components.update(create_qwen_image_edit_ui(prefix))
+        components.update(create_boogu_edit_ui(prefix))
+        components.update(create_reference_image_ui(prefix))
+    components[f"native_reference_panel_{prefix}"] = native_reference_panel
     components.update(create_vae_override_ui(prefix))
     components.update(create_pid_ui(prefix))
 

ui/theme.py

@@ -12,6 +12,11 @@ APP_CSS = r"""
   margin: 0 auto !important;
 }
 
+#input-image-preview .grid-container {
+  grid-template-rows: 160px;
+  grid-auto-rows: 160px;
+}
+
 #studio-hero {
   padding: 28px 30px;
   margin: 14px 0 18px;