feat(workflows): UI→API 转换器 + 批量导入 + 元数据轻量列表 (HEL-681 S2)

- 新增 comfyui/ui_convert.py：离线确定性转换（reroute/bypass/mute 透传、
  子图展开为 packed:inner 限定 id、Set/Get 变量解析、控件三种映射来源，
  无法可靠映射的文件进入需人工确认，绝不静默猜测）
- 新增 convert / import-batch / import-scan 端点：dry-run 逐文件
  ok/warn/fail 报告 + 建议元数据；导入端冲突跳过、重复执行幂等
- workflows 表加列 category/model_family/media_io/tags/favorite/
  last_used_at/source/source_path（纯加列迁移，旧行为零影响）
- GET /api/workflows?view=meta 轻量列表不含全量 JSON；旧接口行为不变
- 元数据建议：分类/模型家族（主加载器加权+目录回退，规避 qwen_image_vae
  误标）/输入输出媒介/MarkdownNote 描述素材
- enqueue 时记录 last_used_at；PATCH 支持 favorite/tags 等字段
- 测试 23 项新增（转换语义单测+146 文件语料+对拍+端点/迁移/幂等）

Co-authored-by: multica-agent <github@multica.ai>

calliope-backend/scripts/build_widget_orders.py

@@ -0,0 +1,128 @@
+"""Regenerate ``src/calliope/comfyui/ui_widget_orders.py`` (HEL-681).
+
+Offline widget-order registry for the UI→API converter. Sources, in priority:
+
+1. ``widgets_values_named`` instances across ``example_ComfyUI_workflows``
+   (including subgraph inner nodes) — the frontend serializes named widgets
+   in widget order, so the key order IS the serialization order;
+2. the repo's 10 ground-truth API exports (root ``*_API.json``) — scalar
+   input names in export order (control_after_generate entries are absent
+   there; the converter's walk skips them positionally).
+
+Run from anywhere inside the repo:
+    python scripts/build_widget_orders.py
+"""
+
+from __future__ import annotations
+
+import collections
+import json
+from pathlib import Path
+
+BACKEND = Path(__file__).resolve().parents[1]
+REPO = BACKEND.parent
+EXAMPLES = REPO / "example_ComfyUI_workflows"
+OUT = BACKEND / "src" / "calliope" / "comfyui" / "ui_widget_orders.py"
+
+BUTTONS = {"preview", "audioUI", "compare_view", "ToggleableColorPicker", "imagebrowser"}
+MAX_SIGNATURES = 3
+MAX_ORDER_LEN = 48
+
+# Hand-curated fallbacks for node classes the corpus never serializes with
+# names (classic core/custom widgets whose order is stable and public).
+CURATED: dict[str, list[list[str]]] = {
+    "PrimitiveNode": [["value"]],
+    "Seed (rgthree)": [["seed"]],
+}
+
+
+def _iter_ui_nodes() -> list[dict]:
+    nodes: list[dict] = []
+    for path in sorted(EXAMPLES.rglob("*.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            continue
+        if not (isinstance(data, dict) and isinstance(data.get("nodes"), list)):
+            continue
+        nodes.extend(n for n in data["nodes"] if isinstance(n, dict))
+        for sub in data.get("definitions", {}).get("subgraphs", []) or []:
+            if isinstance(sub, dict):
+                nodes.extend(n for n in sub.get("nodes", []) or [] if isinstance(n, dict))
+    return nodes
+
+
+def _iter_api_workflows() -> list[dict]:
+    out: list[dict] = []
+    for path in sorted(EXAMPLES.glob("*_API.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            continue
+        if isinstance(data, dict) and "nodes" not in data:
+            out.append(data)
+    return out
+
+
+def main() -> None:
+    learned: dict[str, collections.Counter[tuple[str, ...]]] = collections.defaultdict(
+        collections.Counter
+    )
+    for node in _iter_ui_nodes():
+        named = node.get("widgets_values_named")
+        if not isinstance(named, dict) or not named:
+            continue
+        keys = tuple(
+            str(k) for k in named.keys() if k and k not in BUTTONS
+        )
+        if keys and len(keys) <= MAX_ORDER_LEN:
+            learned[str(node.get("type") or "")][keys] += 1
+
+    for workflow in _iter_api_workflows():
+        for node in workflow.values():
+            if not isinstance(node, dict):
+                continue
+            names = [
+                str(k)
+                for k, v in (node.get("inputs") or {}).items()
+                if not isinstance(v, list) and k and k not in BUTTONS
+            ]
+            if len(names) > MAX_ORDER_LEN:
+                continue
+            # Display nodes (PreviewAny…) carry zero scalar inputs — that
+            # empty order is real signal: their positional arrays are pure
+            # button leftovers and map to {}.
+            learned[str(node.get("class_type") or "")][tuple(names)] += 1
+
+    registry: dict[str, list[list[str]]] = {}
+    for node_type, counter in sorted(learned.items()):
+        if not node_type:
+            continue
+        orders: list[list[str]] = []
+        for signature, _count in counter.most_common(MAX_SIGNATURES):
+            orders.append(list(signature))
+        if orders:
+            registry[node_type] = orders
+    for node_type, orders in CURATED.items():
+        registry.setdefault(node_type, orders)
+
+    body = json.dumps(registry, ensure_ascii=False, indent=1, sort_keys=True)
+    OUT.write_text(
+        '"""GENERATED by scripts/build_widget_orders.py — do not edit by hand.\n'
+        "\n"
+        "Widget-order registry for the UI→API converter: class_type → candidate\n"
+        "widget name orders, learned from the repo's named-widget UI files and\n"
+        "ground-truth API exports (HEL-681).\n"
+        '"""\n'
+        "\n"
+        "from __future__ import annotations\n"
+        "\n"
+        f"WIDGET_ORDERS: dict[str, list[list[str]]] = {body}\n",
+        encoding="utf-8",
+    )
+    total = sum(len(o) for orders in registry.values() for o in orders)
+    print(f"wrote {OUT.name}: {len(registry)} node types, {total} orders total")
+
+
+if __name__ == "__main__":
+    main()

calliope-backend/scripts/ui_import_dryrun.py

@@ -0,0 +1,170 @@
+"""Generate the HEL-681 dry-run report over the repo's UI workflow library.
+
+Walks ``example_ComfyUI_workflows`` (or ``--root``), runs every file through
+the production converter + lint + parser pipeline (exactly what
+``POST /api/workflows/convert`` does, minus HTTP), and writes a JSON detail
+file plus a human-readable Markdown summary. Read-only: nothing is imported,
+no DB is touched.
+
+Usage:
+    python scripts/ui_import_dryrun.py [--root DIR] [--out PREFIX]
+"""
+
+from __future__ import annotations
+
+import argparse
+import collections
+import json
+import sys
+from pathlib import Path
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND / "src"))
+
+from calliope.comfyui.lint import workflow_warnings  # noqa: E402
+from calliope.comfyui.metadata import suggested_metadata  # noqa: E402
+from calliope.comfyui.parser import classify_usage  # noqa: E402
+from calliope.comfyui.ui_convert import convert_ui_workflow  # noqa: E402
+
+REPO = BACKEND.parent
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--root", default=str(REPO / "example_ComfyUI_workflows"))
+    parser.add_argument("--out", default=str(REPO / "dryrun_report"))
+    args = parser.parse_args()
+    root = Path(args.root)
+    out_prefix = Path(args.out)
+
+    entries = []
+    for path in sorted(root.rglob("*.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError) as exc:
+            entries.append(
+                {
+                    "source_path": str(path.relative_to(root)),
+                    "status": "fail",
+                    "issues": [{"code": "unreadable", "message": str(exc)}],
+                }
+            )
+            continue
+        if not (isinstance(data, dict) and isinstance(data.get("nodes"), list)):
+            continue  # API-format / non-workflow files are out of scope
+        rel = str(path.relative_to(root))
+        result = convert_ui_workflow(data)
+        entry: dict = {
+            "source_path": rel,
+            "status": result.status,
+            "needs_confirmation": result.needs_confirmation,
+            "issues": [
+                {"node_id": i.node_id, "code": i.code, "level": i.level, "message": i.message}
+                for i in result.issues
+            ],
+        }
+        if result.workflow is not None:
+            entry["node_count"] = len(result.workflow)
+            entry["suggested"] = suggested_metadata(result.workflow, rel, data)
+            errors = [w for w in workflow_warnings(result.workflow) if w["level"] == "error"]
+            entry["lint_errors"] = [e["code"] for e in errors]
+            entry["importable"] = not result.needs_confirmation and not errors
+        else:
+            entry["importable"] = False
+        entries.append(entry)
+
+    counts = collections.Counter(e["status"] for e in entries)
+    code_counts = collections.Counter(
+        i["code"] for e in entries for i in e["issues"]
+    )
+    category_counts = collections.Counter(
+        (e.get("suggested") or {}).get("category") or "?" for e in entries
+    )
+    family_counts = collections.Counter(
+        (e.get("suggested") or {}).get("model_family") or "?" for e in entries
+    )
+    importable = sum(1 for e in entries if e.get("importable"))
+    needs_confirmation = sum(1 for e in entries if e["needs_confirmation"])
+    lint_blocked = [e for e in entries if e.get("lint_errors")]
+
+    detail = {
+        "root": str(root),
+        "total": len(entries),
+        "status_counts": dict(counts),
+        "issue_code_counts": dict(code_counts),
+        "importable": importable,
+        "needs_confirmation": needs_confirmation,
+        "lint_blocked": [e["source_path"] for e in lint_blocked],
+        "files": entries,
+    }
+    out_prefix.parent.mkdir(parents=True, exist_ok=True)
+    (out_prefix.with_suffix(".json")).write_text(
+        json.dumps(detail, ensure_ascii=False, indent=1), encoding="utf-8"
+    )
+
+    lines = [
+        "# HEL-681 工作流 UI→API 转换 dry-run 报告",
+        "",
+        f"- 扫描目录：`{root}`（根目录 10 个 API 格式旧文件不在本管线范围）",
+        f"- UI 格式文件总数：**{len(entries)}**",
+        f"- 转换结果：ok **{counts.get('ok', 0)}** / warn **{counts.get('warn', 0)}** / fail **{counts.get('fail', 0)}**",
+        f"- 其中「需人工确认」（控件无法可靠映射，禁止直接导入）：**{needs_confirmation}**",
+        f"- 通过转换＋lint 后可直接导入：**{importable}**",
+        f"- 转换成功但被现有 lint 拦截（导入时按阻止处理）：**{len(lint_blocked)}**",
+        "",
+        "## 分类分布（自动提取）",
+        "",
+        "| 分类 | 数量 |",
+        "|---|---|",
+    ]
+    for cat, n in sorted(category_counts.items()):
+        lines.append(f"| {cat} | {n} |")
+    lines += ["", "## 模型家族分布（自动提取）", "", "| 家族 | 数量 |", "|---|---|"]
+    for fam, n in sorted(family_counts.items(), key=lambda kv: -kv[1]):
+        lines.append(f"| {fam} | {n} |")
+    lines += ["", "## 问题代码统计", "", "| 代码 | 次数 | 含义 |", "|---|---|---|"]
+    meaning = {
+        "muted_input_dropped": "上游节点被静音，输入按未连接处理",
+        "variable_upstream_dropped": "Set/Get 变量上游被旁路，取值按未连接处理",
+        "get_without_set": "GetNode 没有同名 SetNode",
+        "widget_unmapped": "控件值无法可靠映射到输入名（需人工确认）",
+        "widget_order_mismatch": "控件数量与已知顺序不符（需人工确认）",
+        "no_output_node": "图中无已知输出节点，全部活跃节点保留待确认",
+        "cyclic_variable": "Set/Get 循环（FOR 循环流），该连线按未连接处理",
+        "dangling_link": "连线指向不存在的节点",
+    }
+    for code, n in code_counts.most_common():
+        lines.append(f"| `{code}` | {n} | {meaning.get(code, '')} |")
+    lines += ["", "## 需人工确认文件清单", ""]
+    for e in entries:
+        if e["needs_confirmation"]:
+            reasons = sorted({i["code"] for i in e["issues"]})[:3]
+            lines.append(f"- `{e['source_path']}`（{', '.join(reasons)}）")
+    lines += ["", "## lint 拦截文件清单", ""]
+    for e in lint_blocked:
+        lines.append(f"- `{e['source_path']}`（{', '.join(e['lint_errors'])}）")
+    lines += [
+        "",
+        "## 逐文件结果",
+        "",
+        "| 文件 | 状态 | 节点数 | 分类 | 模型家族 | 用途 | 问题 |",
+        "|---|---|---|---|---|---|---|",
+    ]
+    for e in entries:
+        sug = e.get("suggested") or {}
+        issues = collections.Counter(i["code"] for i in e["issues"])
+        issue_text = ", ".join(f"{c}×{n}" for c, n in issues.most_common(2)) or "-"
+        usage = sug.get("usage") or ("待定" if sug.get("usage_ambiguous") else "-")
+        lines.append(
+            f"| `{e['source_path']}` | {e['status']} | {e.get('node_count', '-')} "
+            f"| {sug.get('category') or '-'} | {sug.get('model_family') or '-'} "
+            f"| {usage} | {issue_text} |"
+        )
+    (out_prefix.with_suffix(".md")).write_text("\n".join(lines) + "\n", encoding="utf-8")
+    print(f"{len(entries)} files: {dict(counts)}, importable={importable}, nc={needs_confirmation}")
+    print(f"report → {out_prefix}.md / {out_prefix}.json")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

calliope-backend/src/calliope/comfyui/metadata.py

@@ -0,0 +1,160 @@
+"""Suggested workflow metadata for imports (HEL-681 / S2 data layer).
+
+Everything here derives from two honest sources: the repo directory the file
+came from (category) and the converted API graph itself (usage, media in/out,
+model family, note text). Per HEL-675's inventory, clean display names and
+tags stay human work — the suggestion is a starting point, never a guess that
+hides uncertainty.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from calliope.comfyui.parser import (
+    classify_usage,
+    parse_dynamic_inputs,
+    parse_dynamic_outputs,
+)
+
+# Repo top-level directory → (category key, display label). Directory names
+# are mapping sources only — they never surface to users verbatim.
+CATEGORY_DIRS: dict[str, tuple[str, str]] = {
+    "0-基础处理：图像合并-音频分离-视频截取合并-图像视频放大": ("basic", "基础处理"),
+    "1-语音生成": ("audio_tts", "语音生成"),
+    "2-音乐生成": ("music", "音乐生成"),
+    "3-图像生成": ("image", "图像生成"),
+    "4-视频生成": ("video", "视频生成"),
+    "5-数字人": ("digital_human", "数字人"),
+    "6-动作迁移": ("motion_transfer", "动作迁移"),
+}
+
+# Evidence patterns → model family. Shared VAE/text-encoder files must NOT
+# count (HEL-675: Anima/Z-Image graphs reference qwen_image_vae and get
+# mislabeled Qwen — `qwen.?image(?!_vae)` is deliberate).
+_FAMILY_PATTERNS: tuple[tuple[str, str], ...] = (
+    (r"minimax[_\-]?(music|_audio)|MiniMaxMusic", "MiniMax-Music"),
+    (r"minimax.?h3|minimax_h3|h3[-_]realism|fl2v[aa]?", "MiniMax-H3"),
+    (r"ltx[-_.]?2", "LTX-2.5"),
+    (r"ltx", "LTX"),
+    (r"krea", "Krea2"),
+    (r"qwen.?image(?!_vae)|qwen_image_2|qwen2vl|qwen3vl|Qwen3?[-.]?ASR|qwen3?[-_.]?tts", "Qwen"),
+    (r"z[._-]?image", "Z-Image"),
+    (r"flux", "Flux"),
+    (r"\bwan", "Wan"),
+    (r"ace[_-]?step", "ACE-Step"),
+    (r"index[_-]?tts", "IndexTTS2"),
+    (r"anima[-_.]?(turbo|klein)", "Anima"),
+    (r"sdpose", "SDPose"),
+    (r"gemma", "Gemma"),
+    (r"instantmesh|instant_mesh", "InstantMesh"),
+    (r"hidream", "HiDream"),
+)
+_FAMILY_RE = [(re.compile(pat, re.IGNORECASE), fam) for pat, fam in _FAMILY_PATTERNS]
+
+# Main-model loaders dominate the vote: a UNET/checkpoint reference says far
+# more about the workflow's family than a shared VAE or text encoder does.
+_PRIMARY_LOADER_RE = re.compile(r"UNETLoader|CheckpointLoader|unet|diffusion_model", re.IGNORECASE)
+
+# Directory names that carry family evidence the files' own widget values
+# lack (HEL-675: 14 krea2 files reference no "krea" string at all).
+_FAMILY_DIRS = (
+    (re.compile(r"krea2", re.IGNORECASE), "Krea2"),
+    (re.compile(r"Minimax-h3|uncensored", re.IGNORECASE), "MiniMax-H3"),
+    (re.compile(r"Z-Image", re.IGNORECASE), "Z-Image"),
+    (re.compile(r"Qwen-Image", re.IGNORECASE), "Qwen"),
+    (re.compile(r"^Anima$", re.IGNORECASE), "Anima"),
+    (re.compile(r"Flux", re.IGNORECASE), "Flux"),
+)
+
+_NOTE_TYPES = frozenset({"Note", "MarkdownNote"})
+
+
+def category_from_source_path(source_path: str | None) -> tuple[str | None, str | None]:
+    """(category key, label) derived from the file's repo-relative path."""
+    if not source_path:
+        return None, None
+    head = source_path.replace("\\", "/").split("/", 1)[0]
+    return CATEGORY_DIRS.get(head, (None, None))
+
+
+def model_family_from(workflow: dict[str, Any], source_path: str | None = None) -> str | None:
+    """The graph's dominant model family: main-model loader references weigh
+    triple, other node evidence single, and the repo sub-directory adds a
+    strong vote for families the widget values don't spell out."""
+    votes: dict[str, int] = {}
+    for node in workflow.values():
+        if not isinstance(node, dict):
+            continue
+        class_type = str(node.get("class_type") or "")
+        blob = " ".join(
+            [class_type]
+            + [str(v) for v in (node.get("inputs") or {}).values() if isinstance(v, str)]
+        )
+        weight = 3 if _PRIMARY_LOADER_RE.search(blob) else 1
+        for pattern, family in _FAMILY_RE:
+            if pattern.search(blob):
+                votes[family] = votes.get(family, 0) + weight
+                break  # one vote per node: strongest family that node shows
+    if source_path:
+        parts = source_path.replace("\\", "/").split("/")
+        for part in parts[1:2]:
+            for pattern, family in _FAMILY_DIRS:
+                if pattern.search(part):
+                    votes[family] = votes.get(family, 0) + 2
+    if not votes:
+        return None
+    return max(votes.items(), key=lambda kv: kv[1])[0]
+
+
+def note_description(ui_workflow: dict[str, Any], limit: int = 160) -> str | None:
+    """First non-trivial author note as description material (never final copy)."""
+    for node in ui_workflow.get("nodes") or []:
+        if not isinstance(node, dict) or str(node.get("type")) not in _NOTE_TYPES:
+            continue
+        values = node.get("widgets_values")
+        text = values[0] if isinstance(values, list) and values else None
+        if isinstance(values, dict):
+            text = values.get("text") or next(iter(values.values()), None)
+        if isinstance(text, str):
+            text = text.strip()
+            if len(text) >= 8:
+                return text[:limit].strip()
+    return None
+
+
+def suggested_metadata(
+    converted: dict[str, Any],
+    source_path: str | None = None,
+    ui_workflow: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """The import suggestion bundle for one converted workflow."""
+    judgment = classify_usage(converted)
+    media_kinds = ("image", "image_list", "video", "video_list", "audio", "audio_list")
+    media_in = sorted(
+        {
+            entry["kind"].replace("_list", "")
+            for entry in parse_dynamic_inputs(converted)
+            if entry.get("kind") in media_kinds
+        }
+    )
+    media_out = sorted(
+        {
+            str(entry.get("kind"))
+            for entry in parse_dynamic_outputs(converted)
+            if entry.get("kind") in ("image", "video", "audio")
+        }
+    )
+    category, category_label = category_from_source_path(source_path)
+    return {
+        "kind": "video" if "video" in media_out and "image" not in media_out else "image",
+        "usage": judgment["usage"],
+        "usage_ambiguous": judgment["ambiguous"],
+        "category": category,
+        "category_label": category_label,
+        "model_family": model_family_from(converted, source_path),
+        "media_in": media_in,
+        "media_out": media_out,
+        "description": note_description(ui_workflow or {}),
+    }

calliope-backend/src/calliope/comfyui/ui_convert.py

@@ -0,0 +1,801 @@
+"""UI→API converter for ComfyUI workflow JSON (HEL-681 / S2 data layer).
+
+ComfyUI saves two formats: the editor ("UI") format with ``nodes``/``links``
+arrays, and the API ("prompt") format ``{node_id: {class_type, inputs}}`` the
+product executes. This module converts UI→API offline and deterministically:
+
+* links are rewritten through ``Reroute`` chains and bypassed (mode 4) nodes;
+* packed subgraph instances (frontend ≥1.x ``definitions.subgraphs``) expand
+  into their inner nodes under qualified ids ``<packed>:<inner>`` — the same
+  scheme ComfyUI itself exports;
+* widget values map through, in priority order: the node's serialized input
+  list (entries carrying a ``widget`` marker — authoritative, this corpus's
+  frontend version always writes them), the ``widgets_values_named`` dict,
+  then a generated registry of widget orders learned from the corpus and the
+  repo's ground-truth API exports (``ui_widget_orders``).
+
+Anything that cannot be mapped with certainty is reported as an issue and the
+file is flagged ``needs_confirmation`` — never silently guessed: a wrong
+widget order produces a runnable-looking workflow with swapped seed/steps,
+which is worse than an honest「需人工确认」.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from calliope.comfyui.ui_widget_orders import WIDGET_ORDERS
+
+# Values of the frontend-only "control_after_generate" control that serializes
+# right after every seed-like INT widget (and is not a real server input).
+_CONTROL_MODES = frozenset({"fixed", "increment", "decrement", "randomize"})
+
+# Widget names that are pure UI buttons/previews — never serialized in
+# widgets_values (named dicts skip them too) and never API inputs.
+_BUTTON_WIDGETS = frozenset(
+    {"preview", "audioUI", "compare_view", "ToggleableColorPicker", "imagebrowser"}
+)
+
+# Leftover positional values that are provably action buttons, not data:
+# known labels, self-describing ids, or emoji-prefixed actions. Anything
+# else left over keeps the file in the needs-confirmation queue.
+_BUTTON_LEFTOVER_LABELS = frozenset({"Edit", "lanpaint_star_button"})
+_BUTTON_LEFTOVER_EMOJI = ("🖼", "🎬", "🎵", "➕", "🔁", "🗑", "📂", "🔎", "✨")
+
+
+def _is_button_leftover(value: Any) -> bool:
+    if value is None or value == "":
+        return True
+    if isinstance(value, str):
+        return (
+            value in _BUTTON_LEFTOVER_LABELS
+            or "button" in value.lower()
+            or value.startswith(_BUTTON_LEFTOVER_EMOJI)
+        )
+    return False
+
+# Virtual editor-only nodes. Links pass *through* reroute-style ones (see
+# _resolve_source); the rest are dropped from the graph entirely.
+_REROUTE_TYPES = frozenset({"Reroute", "Reroute (rgthree)"})
+# Frontend-virtual variable nodes: SetNode captures a value under a name,
+# GetNode reads it. Neither reaches the server — links through a Get resolve
+# to whatever its same-named Set captured (scope-local).
+_SET_TYPES = frozenset({"SetNode"})
+_GET_TYPES = frozenset({"GetNode"})
+_UI_ONLY_TYPES = frozenset(
+    {
+        "Note",
+        "MarkdownNote",
+        "Image Comparer (rgthree)",
+        "GH_ImageVideoComparer",
+        "Label (rgthree)",
+        "Fast Groups Bypasser (rgthree)",
+        "Fast Groups Muter (rgthree)",
+        "Fast Muter (rgthree)",
+        "FastGroupsBypassSwitch",
+        "Mute / Bypass Repeater (rgthree)",
+    }
+)
+
+# Server-side OUTPUT nodes: the closure roots. Learned from the repo's
+# ground-truth API exports (their unreferenced classes) + the corpus's
+# real save/preview sinks (SaveVideo/SaveImageAdvanced/… custom nodes).
+_OUTPUT_CLASSES = frozenset(
+    {
+        "SaveImage",
+        "PreviewImage",
+        "SaveImageAdvanced",
+        "SaveImageKJ",
+        "SaveAnimatedWEBP",
+        "SaveAnimatedPNG",
+        "SaveAnimatedGIF",
+        "SaveVideo",
+        "SaveAudio",
+        "SaveAudioMP3",
+        "SaveAudioAdvanced",
+        "PreviewAudio",
+        "MaskPreview",
+        "MaskPreview+",
+        "ImageAndMaskPreview",
+        "VHS_VideoCombine",
+        "easy saveVideo",
+        "easy multitrackProjectVideoCombine",
+        "easy multiTrackInfoOutput",
+        "easy multiTrackTaskOutput",
+        "easy compareVideos",
+        "easy compareImages",
+        "easy showText",
+        "easy imageWall",
+        "CR Prompt Text",
+        "PreviewAny",
+        "easy clearGpuUsed",
+        "easy clearCacheAll",
+        "easy showSpentTime",
+    }
+)
+
+MODE_ACTIVE = 0
+MODE_MUTED = 2
+MODE_BYPASSED = 4
+
+_SUBGRAPH_IN = -10  # pseudo-node id of a subgraph's boundary input
+_SUBGRAPH_OUT = -20  # pseudo-node id of a subgraph's boundary output
+
+
+@dataclass
+class ConvertIssue:
+    node_id: str
+    code: str
+    message: str
+    level: str = "warning"  # 'error' | 'warning'
+
+
+@dataclass
+class ConvertResult:
+    workflow: dict[str, Any] | None
+    status: str  # 'ok' | 'warn' | 'fail'
+    issues: list[ConvertIssue] = field(default_factory=list)
+
+    @property
+    def needs_confirmation(self) -> bool:
+        """Widget mapping uncertain — must not be imported as runnable."""
+        return any(
+            i.code in ("widget_unmapped", "widget_order_mismatch") for i in self.issues
+        )
+
+
+def _norm_link(link: Any) -> tuple[int, int, int, int, int, str] | None:
+    """(id, origin_id, origin_slot, target_id, target_slot, type) from either form."""
+    if isinstance(link, dict):
+        try:
+            return (
+                int(link["id"]),
+                int(link["origin_id"]),
+                int(link["origin_slot"]),
+                int(link["target_id"]),
+                int(link["target_slot"]),
+                str(link.get("type") or "*"),
+            )
+        except (KeyError, TypeError, ValueError):
+            return None
+    if isinstance(link, (list, tuple)) and len(link) >= 5:
+        try:
+            return (
+                int(link[0]),
+                int(link[1]),
+                int(link[2]),
+                int(link[3]),
+                int(link[4]),
+                str(link[5]) if len(link) > 5 and link[5] else "*",
+            )
+        except (TypeError, ValueError):
+            return None
+    return None
+
+
+def _is_seed_like(name: str) -> bool:
+    lowered = name.lower()
+    return lowered in ("seed", "noise_seed", "value", "rand_seed", "random_seed") or (
+        lowered.endswith("seed")
+    )
+
+
+def _is_upload_entry(name: str, type_: str) -> bool:
+    return type_.upper().endswith("UPLOAD") or name == "upload" or "UPLOAD" in type_.upper()
+
+
+class _Converter:
+    """One conversion run over a single UI-format document."""
+
+    def __init__(self, ui: dict[str, Any]) -> None:
+        self.ui = ui
+        self.issues: list[ConvertIssue] = []
+        self.subgraphs: dict[str, dict[str, Any]] = {
+            str(s.get("id")): s
+            for s in self.ui.get("definitions", {}).get("subgraphs", []) or []
+            if isinstance(s, dict)
+        }
+        # qualified id → UI node + owning scope
+        self.nodes: dict[str, dict[str, Any]] = {}
+        # scope key ('' for top level) → link table / node-id index / owner
+        self.scope_links: dict[str, dict[int, tuple[int, int, int, int, str]]] = {}
+        self.scope_ids: dict[str, dict[str, str]] = {}
+        self.scope_owner: dict[str, str] = {}
+        self._set_capture_cache: dict[str, dict[str, tuple[str, int]]] = {}
+        self._set_names: dict[str, set[str]] = {}
+        self._depth = 0
+        self._cycle_warned = False
+        self._set_building: set[str] = set()
+
+    # -------------------------------------------------------------- indexing
+    def _index_scope(self, scope: str, links: list[Any], owner: str | None) -> None:
+        table = self.scope_links.setdefault(scope, {})
+        for link in links or []:
+            norm = _norm_link(link)
+            if norm is not None:
+                table[norm[0]] = norm[1:]
+        self.scope_ids.setdefault(scope, {})
+        if owner is not None:
+            self.scope_owner[scope] = owner
+
+    def _flatten(
+        self, node: dict[str, Any], scope: str, prefix: str
+    ) -> None:
+        qualified = f"{prefix}{node.get('id')}"
+        self.nodes[qualified] = {"node": node, "scope": scope}
+        self.scope_ids.setdefault(scope, {})[str(node.get("id"))] = qualified
+        sub = self.subgraphs.get(str(node.get("type") or ""))
+        if sub is None:
+            return
+        if not isinstance(sub.get("nodes"), list):
+            self.issues.append(
+                ConvertIssue(qualified, "subgraph_bad_definition", "子图缺少内部节点定义")
+            )
+            return
+        sub_scope = f"{scope}|{node.get('type')}" if scope else f"{node.get('type')}"
+        if sub_scope not in self.scope_links:
+            self._index_scope(sub_scope, sub.get("links") or [], qualified)
+        for inner in sub["nodes"]:
+            if isinstance(inner, dict):
+                self._flatten(inner, sub_scope, f"{qualified}:")
+
+    # --------------------------------------------------------------- lookups
+    def _node(self, qualified: str) -> dict[str, Any] | None:
+        entry = self.nodes.get(qualified)
+        return entry["node"] if entry else None
+
+    def _scope(self, qualified: str) -> str:
+        entry = self.nodes.get(qualified)
+        return entry["scope"] if entry else ""
+
+    def _link(self, scope: str, link_id: Any) -> tuple[int, int, int, int, str] | None:
+        if link_id is None:
+            return None
+        try:
+            key = int(link_id)
+        except (TypeError, ValueError):
+            return None
+        return self.scope_links.get(scope, {}).get(key)
+
+    def _boundary_source(self, scope: str, link_id: Any) -> tuple[str, int] | None:
+        """(packed qualified id, boundary slot) when this link leaves a
+        subgraph through its -10 input boundary."""
+        link = self._link(scope, link_id)
+        if link is None or link[0] != _SUBGRAPH_IN:
+            return None
+        packed_q = self.scope_owner.get(scope)
+        if packed_q is None:
+            return None
+        return packed_q, link[1]
+
+    def _subgraph_of(self, qualified: str) -> dict[str, Any] | None:
+        node = self._node(qualified)
+        if node is None:
+            return None
+        return self.subgraphs.get(str(node.get("type") or ""))
+
+    @staticmethod
+    def _virtual_name(node: dict[str, Any]) -> str:
+        """A Set/Get node's captured variable name (its first widget value)."""
+        named = node.get("widgets_values_named")
+        if isinstance(named, dict):
+            for key in ("const_name", "name", "value"):
+                if key in named:
+                    return str(named[key])
+        values = node.get("widgets_values")
+        if isinstance(values, list) and values and isinstance(values[0], str):
+            return values[0]
+        return ""
+
+    def _set_captures(self, scope: str) -> dict[str, tuple[str, int]]:
+        """{variable name → resolved source} for the Set nodes in one scope."""
+        table = self._set_capture_cache.get(scope)
+        if table is not None:
+            return table
+        table = {}
+        # Seed before resolving: a Set feeding a same-named Get (FOR-loop
+        # graphs) would otherwise recurse into this builder forever. A miss
+        # while building is not a missing Set — never warn mid-construction.
+        self._set_capture_cache[scope] = table
+        self._set_building.add(scope)
+        try:
+            return self._build_set_captures(scope, table)
+        finally:
+            self._set_building.discard(scope)
+
+    def _build_set_captures(
+        self, scope: str, table: dict[str, tuple[str, int]]
+    ) -> dict[str, tuple[str, int]]:
+        names: set[str] = set()
+        for qualified, entry in self.nodes.items():
+            if entry["scope"] != scope:
+                continue
+            node = entry["node"]
+            if str(node.get("type") or "") not in _SET_TYPES:
+                continue
+            name = self._virtual_name(node)
+            if not name:
+                continue
+            names.add(name)
+            ins = self._inputs_of(node)
+            link = self._link(scope, ins[0].get("link")) if ins else None
+            resolved = self._resolve_link(scope, link)
+            if resolved is not None:
+                table.setdefault(name, resolved)
+        self._set_names[scope] = names
+        self._set_capture_cache[scope] = table
+        return table
+
+    @staticmethod
+    def _inputs_of(node: dict[str, Any]) -> list[dict[str, Any]]:
+        return [i for i in node.get("inputs") or [] if isinstance(i, dict)]
+
+    @staticmethod
+    def _outputs_of(node: dict[str, Any]) -> list[dict[str, Any]]:
+        return [o for o in node.get("outputs") or [] if isinstance(o, dict)]
+
+    # ------------------------------------------------------------- resolution
+    def _resolve_link(
+        self, scope: str, link: tuple[int, int, int, int, str] | None
+    ) -> tuple[str, int] | None:
+        """Resolve one link (in scope) to an API source (qualified id, slot)."""
+        if link is None:
+            return None
+        if self._depth > 200:  # Set/Get cycles (FOR-loop graphs) never unwind
+            if not self._cycle_warned:
+                self._cycle_warned = True
+                self.issues.append(
+                    ConvertIssue(
+                        "",
+                        "cyclic_variable",
+                        "变量/Set-Get 连线存在循环（常见于 FOR 循环流），该连线按未连接处理",
+                    )
+                )
+            return None
+        self._depth += 1
+        try:
+            return self._resolve_link_inner(scope, link)
+        finally:
+            self._depth -= 1
+
+    def _resolve_link_inner(
+        self, scope: str, link: tuple[int, int, int, int, str]
+    ) -> tuple[str, int] | None:
+        origin_id, origin_slot = link[0], link[1]
+        if origin_id == _SUBGRAPH_IN:
+            # Boundary: hand off to the packed instance's own input slot.
+            packed_q = self.scope_owner.get(scope)
+            if packed_q is None:
+                return None
+            node = self._node(packed_q)
+            if node is None:
+                return None
+            entries = self._inputs_of(node)
+            if origin_slot >= len(entries):
+                return None
+            entry = entries[origin_slot]
+            if entry.get("link") is None:
+                return None  # unlinked boundary → inner widget value applies
+            outer = self._scope(packed_q)
+            return self._resolve_link(outer, self._link(outer, entry.get("link")))
+        qualified = self.scope_ids.get(scope, {}).get(str(origin_id))
+        if qualified is None:
+            return None
+        return self._resolve_source(qualified, origin_slot)
+
+    def _resolve_source(self, qualified: str, slot: int) -> tuple[str, int] | None:
+        """What an outgoing edge from (node, output slot) really connects to."""
+        node = self._node(qualified)
+        if node is None:
+            return None
+        node_type = str(node.get("type") or "")
+        mode = node.get("mode") or 0
+        scope = self._scope(qualified)
+
+        if node_type in _UI_ONLY_TYPES:
+            return None
+
+        if node_type in _REROUTE_TYPES:
+            ins = self._inputs_of(node)
+            if not ins:
+                return None
+            return self._resolve_link(scope, self._link(scope, ins[0].get("link")))
+
+        if node_type in _SET_TYPES:
+            ins = self._inputs_of(node)
+            if not ins:
+                return None
+            return self._resolve_link(scope, self._link(scope, ins[0].get("link")))
+
+        if node_type in _GET_TYPES:
+            name = self._virtual_name(node)
+            captured = self._set_captures(scope).get(name)
+            if captured is None and scope not in self._set_building:
+                if name in self._set_names.get(scope, set()):
+                    self.issues.append(
+                        ConvertIssue(
+                            qualified,
+                            "variable_upstream_dropped",
+                            f"变量「{name}」的上游被旁路或未连接，取值按未连接处理",
+                        )
+                    )
+                else:
+                    self.issues.append(
+                        ConvertIssue(
+                            qualified,
+                            "get_without_set",
+                            f"变量节点「{name}」没有对应的 Set 节点",
+                        )
+                    )
+            return captured
+
+        sub = self._subgraph_of(qualified)
+        if sub is not None:
+            outputs = [o for o in sub.get("outputs") or [] if isinstance(o, dict)]
+            if slot >= len(outputs):
+                return None
+            sub_scope = f"{scope}|{node_type}" if scope else node_type
+            for link in self.scope_links.get(sub_scope, {}).values():
+                if link[2] == _SUBGRAPH_OUT and link[3] == slot:
+                    inner_q = self.scope_ids.get(sub_scope, {}).get(str(link[0]))
+                    return self._resolve_source(inner_q, link[1]) if inner_q else None
+            return None  # boundary output with no inner source
+
+        if mode == MODE_MUTED:
+            return None
+        if mode == MODE_BYPASSED:
+            return self._bypass_through(qualified, slot)
+        return (qualified, slot)
+
+    def _bypass_through(self, qualified: str, slot: int) -> tuple[str, int] | None:
+        """Pass-through of a bypassed node: same-name input first, then same-type."""
+        node = self._node(qualified)
+        if node is None:
+            return None
+        outputs = self._outputs_of(node)
+        out = outputs[slot] if slot < len(outputs) else {}
+        out_name = str(out.get("name") or "")
+        out_type = str(out.get("type") or "*")
+        linkable = [
+            e for e in self._inputs_of(node) if not e.get("widget") and e.get("link") is not None
+        ]
+        pick = None
+        if out_name:
+            pick = next((e for e in linkable if str(e.get("name") or "") == out_name), None)
+        if pick is None:
+            pick = next(
+                (
+                    e
+                    for e in linkable
+                    if out_type == "*" or str(e.get("type") or "*") in (out_type, "*")
+                ),
+                None,
+            )
+        if pick is None:
+            return None
+        scope = self._scope(qualified)
+        return self._resolve_link(scope, self._link(scope, pick.get("link")))
+
+    # --------------------------------------------------------- widget mapping
+    def _widget_entries(self, node: dict[str, Any]) -> list[tuple[str, str]]:
+        """[(name, type)] of entries carrying the ``widget`` marker, in order."""
+        out: list[tuple[str, str]] = []
+        for entry in self._inputs_of(node):
+            marker = entry.get("widget")
+            if isinstance(marker, dict):
+                name = str(marker.get("name") or entry.get("name") or "")
+                if name:
+                    out.append((name, str(entry.get("type") or "")))
+        return out
+
+    def _widgets_from_entries(
+        self,
+        node_id: str,
+        entries: list[tuple[str, str]],
+        values: Any,
+    ) -> tuple[dict[str, Any], list[ConvertIssue]]:
+        """Map widget values using the node's serialized input list."""
+        inputs: dict[str, Any] = {}
+        issues: list[ConvertIssue] = []
+        if isinstance(values, dict):
+            for name, type_ in entries:
+                if name in _BUTTON_WIDGETS or self._is_upload(name, type_):
+                    continue
+                if name in values:
+                    inputs[name] = values[name]
+                else:
+                    issues.append(
+                        ConvertIssue(node_id, "widget_unmapped", f"控件「{name}」没有序列化值")
+                    )
+            return inputs, issues
+        if not isinstance(values, list):
+            return inputs, issues
+        if not values:
+            return inputs, issues  # nothing serialized — server defaults apply
+        i = 0
+        for name, type_ in entries:
+            if name in _BUTTON_WIDGETS:
+                continue  # button widgets never serialize
+            if i >= len(values):
+                issues.append(
+                    ConvertIssue(node_id, "widget_order_mismatch", f"控件值数量不足（{name}）")
+                )
+                break
+            value = values[i]
+            i += 1
+            if self._is_upload(name, type_):
+                continue  # consumes a slot, never an API input
+            inputs[name] = value
+            if (
+                type_.upper() == "INT"
+                and _is_seed_like(name)
+                and i < len(values)
+                and values[i] in _CONTROL_MODES
+            ):
+                i += 1  # control_after_generate rides right after its seed
+        leftover_ok = all(_is_button_leftover(v) for v in values[i:])
+        if not issues and i != len(values) and not leftover_ok:
+            issues.append(
+                ConvertIssue(
+                    node_id,
+                    "widget_order_mismatch",
+                    f"有 {len(values)} 个控件值，输入清单只覆盖 {i} 个",
+                )
+            )
+        return inputs, issues
+
+    @staticmethod
+    def _is_upload(name: str, type_: str) -> bool:
+        return _is_upload_entry(name, type_)
+
+    def _widgets_from_registry(
+        self, node: dict[str, Any], values: list[Any]
+    ) -> tuple[dict[str, Any], list[ConvertIssue]]:
+        node_type = str(node.get("type") or "")
+        if not values:
+            return {}, []  # nothing serialized — server defaults apply
+        for order in WIDGET_ORDERS.get(node_type) or []:
+            inputs: dict[str, Any] = {}
+            i = 0
+            for name in order:
+                if name in _BUTTON_WIDGETS:
+                    continue
+                if i >= len(values):
+                    break
+                value = values[i]
+                i += 1
+                if _is_upload_entry(name, ""):
+                    continue
+                inputs[name] = value
+                if _is_seed_like(name) and i < len(values) and values[i] in _CONTROL_MODES:
+                    i += 1
+            if i == len(values) or all(_is_button_leftover(v) for v in values[i:]):
+                return inputs, []  # exact fit, or trailing button leftovers
+        return {}, [
+            ConvertIssue(
+                node.get("id", ""),
+                "widget_unmapped" if not WIDGET_ORDERS.get(node_type) else "widget_order_mismatch",
+                f"「{node_type}」的 {len(values)} 个控件值无法可靠对应到输入名",
+            )
+        ]
+
+    def _packed_widget_values(self, qualified: str) -> Any:
+        """Exposed widget values of a packed instance as {boundary name: value}."""
+        node = self._node(qualified)
+        sub = self._subgraph_of(qualified)
+        if node is None or sub is None:
+            return None
+        boundary = [i for i in sub.get("inputs") or [] if isinstance(i, dict)]
+        named = node.get("widgets_values_named")
+        if isinstance(named, dict):
+            return named
+        values = node.get("widgets_values")
+        if not isinstance(values, list):
+            return None
+        out: dict[str, Any] = {}
+        i = 0
+        for entry in boundary:
+            name = str(entry.get("name") or "")
+            type_ = str(entry.get("type") or "")
+            if name and i < len(values):
+                out[name] = values[i]
+            i += 1
+            if (
+                type_.upper() == "INT"
+                and _is_seed_like(name)
+                and i < len(values)
+                and values[i] in _CONTROL_MODES
+            ):
+                i += 1
+        return out
+
+    # -------------------------------------------------------------- serialize
+    def _serialize_node(self, qualified: str) -> dict[str, Any] | None:
+        node = self._node(qualified)
+        if node is None:
+            return None
+        node_type = str(node.get("type") or "")
+        if (
+            not node_type
+            or node_type in _UI_ONLY_TYPES
+            or node_type in _REROUTE_TYPES
+            or node_type in _SET_TYPES
+            or node_type in _GET_TYPES
+        ):
+            return None
+        scope = self._scope(qualified)
+        inputs: dict[str, Any] = {}
+
+        widget_values = node.get("widgets_values_named")
+        if not isinstance(widget_values, dict):
+            widget_values = node.get("widgets_values")
+        if self._subgraph_of(qualified) is not None:
+            widget_values = self._packed_widget_values(qualified)
+
+        widget_inputs: dict[str, Any] = {}
+        entries = self._widget_entries(node)
+        if widget_values is not None:
+            if entries:
+                widget_inputs, w_issues = self._widgets_from_entries(
+                    qualified, entries, widget_values
+                )
+                self.issues.extend(w_issues)
+            elif isinstance(widget_values, dict):
+                widget_inputs = {
+                    k: v for k, v in widget_values.items() if k not in _BUTTON_WIDGETS
+                }
+            elif isinstance(widget_values, list):
+                widget_inputs, w_issues = self._widgets_from_registry(node, widget_values)
+                self.issues.extend(w_issues)
+
+        for entry in self._inputs_of(node):
+            name = str(entry.get("name") or "")
+            if not name or name in inputs:
+                continue
+            if entry.get("link") is None:
+                continue  # unconnected optional input; widget values applied below
+            # Widget entries fed from the subgraph boundary (-10): when the
+            # packed instance left that boundary slot unlinked, the value is
+            # the packed node's own exposed widget value, not the inner one.
+            boundary = self._boundary_source(scope, entry.get("link"))
+            if boundary is not None:
+                packed_q, boundary_slot = boundary
+                packed_node = self._node(packed_q)
+                packed_entries = self._inputs_of(packed_node) if packed_node else []
+                if boundary_slot < len(packed_entries) and packed_entries[
+                    boundary_slot
+                ].get("link") is None:
+                    values = self._packed_widget_values(packed_q)
+                    boundary_name = str(packed_entries[boundary_slot].get("name") or "")
+                    if isinstance(values, dict) and boundary_name in values:
+                        inputs[name] = values[boundary_name]
+                        continue
+                # linked boundary falls through to normal resolution
+            link = self._link(scope, entry.get("link"))
+            if link is None:
+                self.issues.append(
+                    ConvertIssue(qualified, "dangling_link", f"输入「{name}」的连线不存在")
+                )
+                continue
+            resolved = self._resolve_link(scope, link)
+            if resolved is not None:
+                inputs[name] = [resolved[0], resolved[1]]
+            elif entry.get("widget"):
+                pass  # muted upstream: the widget value below takes over
+            else:
+                self.issues.append(
+                    ConvertIssue(
+                        qualified,
+                        "muted_input_dropped",
+                        f"输入「{name}」上游被静音，已按未连接处理",
+                    )
+                )
+
+        for name, value in widget_inputs.items():
+            inputs.setdefault(name, value)
+
+        api_node: dict[str, Any] = {"class_type": node_type, "inputs": inputs}
+        title = str(node.get("title") or "").strip()
+        if title:
+            api_node["_meta"] = {"title": title}
+        return api_node
+
+    # ------------------------------------------------------------------- main
+    def run(self) -> ConvertResult:
+        nodes = self.ui.get("nodes")
+        if not isinstance(nodes, list) or not nodes:
+            return ConvertResult(
+                None,
+                "fail",
+                [ConvertIssue("", "no_nodes", "文件不含 nodes 数组，不是 UI 格式工作流", "error")],
+            )
+        try:
+            self._index_scope("", self.ui.get("links") or [], None)
+            for node in nodes:
+                if isinstance(node, dict):
+                    self._flatten(node, "", "")
+        except RecursionError:
+            return ConvertResult(
+                None, "fail", [ConvertIssue("", "subgraph_recursion", "子图嵌套过深", "error")]
+            )
+
+        # Closure from known output nodes (ComfyUI's own graphToPrompt roots).
+        roots = [
+            q
+            for q, entry in self.nodes.items()
+            if str(entry["node"].get("type") or "") in _OUTPUT_CLASSES
+            and (entry["node"].get("mode") or 0) == MODE_ACTIVE
+        ]
+        keep: dict[str, None] = {}
+        stack = list(roots)
+        while stack:
+            qualified = stack.pop()
+            if qualified in keep:
+                continue
+            entry = self.nodes.get(qualified)
+            if entry is None:
+                continue
+            keep[qualified] = None
+            node = entry["node"]
+            scope = entry["scope"]
+            for e in self._inputs_of(node):
+                if e.get("link") is None:
+                    continue
+                link = self._link(scope, e.get("link"))
+                resolved = self._resolve_link(scope, link)
+                if resolved is not None:
+                    stack.append(resolved[0])
+
+        if not keep:
+            # No known output node: keep every active computed node and warn —
+            # the import lint will report the missing output separately.
+            for qualified, entry in self.nodes.items():
+                node = entry["node"]
+                node_type = str(node.get("type") or "")
+                if (
+                    node_type in _UI_ONLY_TYPES
+                    or node_type in _REROUTE_TYPES
+                    or (node.get("mode") or 0) != MODE_ACTIVE
+                    or self._subgraph_of(qualified) is not None
+                ):
+                    continue
+                keep[qualified] = None
+            self.issues.append(
+                ConvertIssue(
+                    "",
+                    "no_output_node",
+                    "图中没有已知输出节点（SaveImage / 视频合成等），"
+                    "全部活跃节点已保留，需人工确认",
+                )
+            )
+
+        workflow: dict[str, Any] = {}
+        for qualified in keep:
+            api_node = self._serialize_node(qualified)
+            if api_node is not None and api_node["class_type"]:
+                workflow[qualified] = api_node
+
+        if not workflow:
+            return ConvertResult(
+                None,
+                "fail",
+                self.issues or [ConvertIssue("", "empty", "转换后没有任何节点", "error")],
+            )
+        status = (
+            "fail"
+            if any(i.level == "error" for i in self.issues)
+            else ("ok" if not self.issues else "warn")
+        )
+        return ConvertResult(workflow, status, self.issues)
+
+
+def convert_ui_workflow(ui: dict[str, Any]) -> ConvertResult:
+    """Convert one ComfyUI UI-format workflow dict into API format."""
+    if not isinstance(ui, dict) or not isinstance(ui.get("nodes"), list):
+        return ConvertResult(
+            None,
+            "fail",
+            [ConvertIssue("", "not_ui_format", "不是 UI 格式（缺少 nodes 数组）", "error")],
+        )
+    return _Converter(ui).run()

calliope-backend/src/calliope/db.py

@@ -506,6 +506,27 @@ async def migrate_db(db_path: Path) -> None:
         conn.execute("ALTER TABLE workflows ADD COLUMN output_map TEXT")
     if "output_map_source" not in cols:
         conn.execute("ALTER TABLE workflows ADD COLUMN output_map_source TEXT")
+    # HEL-681 (S2): library metadata for 150+ scale workflow browsing. Purely
+    # additive columns — old rows stay NULL/0 and every legacy path ignores
+    # them; the lightweight list view (?view=meta) is their only reader.
+    if "category" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN category TEXT")
+    if "model_family" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN model_family TEXT")
+    if "media_io" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN media_io TEXT")
+    if "tags" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN tags TEXT")
+    if "favorite" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN favorite INTEGER NOT NULL DEFAULT 0")
+    if "last_used_at" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN last_used_at TIMESTAMP")
+    if "source" not in cols:
+        # where the row came from: 'ui_convert' (HEL-681 batch import) vs
+        # NULL = manual/API import as before.
+        conn.execute("ALTER TABLE workflows ADD COLUMN source TEXT")
+    if "source_path" not in cols:
+        conn.execute("ALTER TABLE workflows ADD COLUMN source_path TEXT")
     # Idempotent backfill: pending rows are re-judged on every boot (parser
     # upgrades can settle them); confirmed rows are never touched.
     _backfill_workflow_usage(conn)

calliope-backend/src/calliope/models/schemas.py

@@ -243,6 +243,55 @@ class WorkflowUpdate(BaseModel):
     # resets the workflow to pending confirmation.
     output_map: list[dict[str, Any]] | None = None
     output_map_source: Literal["auto", "user"] | None = None
+    # HEL-681 (S2): library metadata fields.
+    favorite: bool | None = None
+    tags: list[str] | None = None
+    category: str | None = None
+    model_family: str | None = None
+
+
+class WorkflowUIFileItem(BaseModel):
+    """One UI-format file for the HEL-681 convert/import pipeline."""
+
+    name: str | None = None
+    # repo-relative path (category/model-family evidence); display name
+    # defaults to the file stem.
+    source_path: str | None = None
+    workflow_json: dict[str, Any]
+    # optional human confirmations from the import wizard (S3): a settled
+    # usage and a corrected schema let warn files import enabled.
+    usage: WorkflowUsageKind | None = None
+    input_schema: list[dict[str, Any]] | None = None
+    output_map: list[dict[str, Any]] | None = None
+    description: str | None = None
+    tags: list[str] | None = None
+
+
+class WorkflowConvertRequest(BaseModel):
+    """Dry-run: per-file ok/warn/fail + suggested metadata, nothing stored."""
+
+    files: list[WorkflowUIFileItem]
+    # Include the converted API JSON per file (default on for small batches;
+    # the 146-file report leaves it off to keep the payload sane).
+    include_workflows: bool = False
+    # L2 preflight against a real ComfyUI (opt-in; needs a connection).
+    endpoint_id: str | None = None
+    refresh: bool = False
+
+
+class WorkflowImportBatchRequest(BaseModel):
+    """Commit: convert UI files and insert, skipping conflicts. Idempotent."""
+
+    files: list[WorkflowUIFileItem]
+
+
+class WorkflowImportScanRequest(BaseModel):
+    """Scan a server-side directory of UI JSONs (repo example library)."""
+
+    root: str = "example_ComfyUI_workflows"
+    # Set to preview what would import (same shape as /convert) — the safe
+    # default for a first run against the 146-file library.
+    dry_run: bool = True
 
 
 class WorkflowAnalyze(BaseModel):

calliope-backend/src/calliope/queue/manager.py

@@ -49,6 +49,12 @@ class QueueManager:
                 """,
                 (project_id, scene_id, clip_id, kind, workflow_id, json.dumps(payload), origin),
             )
+            # HEL-681: every enqueued job counts as a use (最近使用).
+            if workflow_id is not None:
+                conn.execute(
+                    "UPDATE workflows SET last_used_at = CURRENT_TIMESTAMP WHERE id = ?",
+                    (workflow_id,),
+                )
             conn.commit()
             row = conn.execute("SELECT * FROM jobs WHERE id = ?", (cur.lastrowid,)).fetchone()
             return row_to_dict(row)

calliope-backend/src/calliope/routers/workflows.py

@@ -7,6 +7,7 @@ from fastapi import APIRouter, HTTPException
 
 from calliope.comfyui.endpoints import client_for, endpoint_snapshot
 from calliope.comfyui.lint import blocking_problem, workflow_warnings
+from calliope.comfyui.metadata import suggested_metadata
 from calliope.comfyui.output_map import (
     auto_output_map,
     load_output_map,
@@ -23,6 +24,7 @@ from calliope.comfyui.parser import (
 from calliope.comfyui.preflight import check_workflow_against
 from calliope.comfyui.profiles import detect_prompt_profile
 from calliope.comfyui.roles import usage_to_kind
+from calliope.comfyui.ui_convert import convert_ui_workflow
 from calliope.comfyui.validate import validate_input_values
 from calliope.config import settings
 from calliope.db import get_db, row_to_dict
@@ -30,8 +32,12 @@ from calliope.events.bus import event_bus
 from calliope.models.schemas import (
     WorkflowAnalyze,
     WorkflowCheckRequest,
+    WorkflowConvertRequest,
     WorkflowCreate,
+    WorkflowImportBatchRequest,
+    WorkflowImportScanRequest,
     WorkflowTestRequest,
+    WorkflowUIFileItem,
     WorkflowUpdate,
 )
 from calliope.queue.manager import queue_manager
@@ -44,6 +50,14 @@ from calliope.queue.workflow_test import (
 
 router = APIRouter()
 
+# HEL-681: metadata columns the lightweight list view exposes (everything
+# except the heavy workflow_json / input_schema / output_map payloads).
+_META_COLUMNS = (
+    "id, name, kind, usage, is_enabled, category, model_family, media_io, tags,"
+    " favorite, last_used_at, description, source, source_path, prompt_profile,"
+    " created_at, endpoint_id, usage_source, output_map_source"
+)
+
 
 def _serialize_workflow(row: Any) -> dict[str, Any]:
     data = row_to_dict(row)
@@ -61,9 +75,21 @@ def _serialize_workflow(row: Any) -> dict[str, Any]:
     data["needs_mapping"] = (
         mapping_problem(data["workflow_json"], raw_map, data.get("output_map_source")) is not None
     )
+    # HEL-681: structured library metadata (raw JSON columns → values).
+    data["media_io"] = _load_json_col(data.get("media_io"), {"in": [], "out": []})
+    data["tags"] = _load_json_col(data.get("tags"), [])
+    data["favorite"] = bool(data.get("favorite"))
     return data
 
 
+def _load_json_col(raw: Any, default: Any) -> Any:
+    try:
+        value = json.loads(raw) if raw else None
+    except (json.JSONDecodeError, TypeError):
+        value = None
+    return value if value is not None else default
+
+
 async def _connection_check(
     endpoint_id: str | None, workflow_json: dict[str, Any], refresh: bool = False
 ) -> dict[str, Any]:
@@ -216,9 +242,31 @@ def _last_test_summary(job: dict[str, Any]) -> dict[str, Any]:
 
 
 @router.get("")
-async def list_workflows() -> list[dict[str, Any]]:
+async def list_workflows(view: str | None = None) -> list[dict[str, Any]]:
     conn = get_db(settings.db_path)
     try:
+        if view == "meta":
+            # HEL-681: metadata-only list for 150+ scale browsing — no
+            # workflow_json, no per-row re-parse. `needs_mapping` here is the
+            # cheap pending check (usage/output_map not yet settled); rows
+            # that never generated keep showing as pending exactly like the
+            # full view reports them.
+            rows = conn.execute(
+                f"SELECT {_META_COLUMNS} FROM workflows ORDER BY favorite DESC,"
+                " last_used_at DESC, id DESC"
+            ).fetchall()
+            out = []
+            for row in rows:
+                data = row_to_dict(row)
+                data["is_enabled"] = bool(data.get("is_enabled"))
+                data["favorite"] = bool(data.get("favorite"))
+                data["media_io"] = _load_json_col(data.get("media_io"), {"in": [], "out": []})
+                data["tags"] = _load_json_col(data.get("tags"), [])
+                data["needs_mapping"] = data.get("usage") is None or data.get(
+                    "output_map_source"
+                ) is None
+                out.append(data)
+            return out
         rows = conn.execute("SELECT * FROM workflows ORDER BY id DESC").fetchall()
         out = [_serialize_workflow(r) for r in rows]
         # 上次测试 state is read live from the newest test job per workflow —
@@ -395,6 +443,12 @@ async def update_workflow(workflow_id: int, payload: WorkflowUpdate) -> dict[str
         data = payload.model_dump(exclude_unset=True)
         if "is_enabled" in data and data["is_enabled"] is not None:
             data["is_enabled"] = 1 if data["is_enabled"] else 0
+        # HEL-681: library metadata patch fields (favorite/tags/category/
+        # model_family) — booleans to ints, lists to JSON text.
+        if "favorite" in data and data["favorite"] is not None:
+            data["favorite"] = 1 if data["favorite"] else 0
+        if data.get("tags") is not None:
+            data["tags"] = json.dumps(data["tags"])
         workflow_json = json.loads(existing["workflow_json"])
         # HEL-638: output_map payload. A non-empty list validates against the
         # stored JSON and becomes the user decision; an explicit empty list
@@ -600,3 +654,280 @@ async def delete_workflow(workflow_id: int) -> dict[str, bool]:
         return {"ok": True}
     finally:
         conn.close()
+
+
+# ---------------------------------------------------------------------------
+# HEL-681 (S2): UI→API conversion dry-run + batch import + repo scan.
+# The convert endpoint never writes; import-batch converts again server-side
+# and skips anything uncertain (needs_confirmation / lint-blocked / conflict)
+# — repeated runs import nothing new, so the batch is idempotent by design.
+# ---------------------------------------------------------------------------
+
+
+def _convert_one(item: WorkflowUIFileItem) -> tuple[dict[str, Any], dict[str, Any] | None]:
+    """(report, converted workflow) for one UI file — shared by every
+    convert/import path so the analysis can never drift between them."""
+    result = convert_ui_workflow(item.workflow_json)
+    stem = (
+        item.source_path.replace("\\", "/").rsplit("/", 1)[-1].removesuffix(".json")
+        if item.source_path
+        else None
+    )
+    name = item.name or stem or "未命名工作流"
+    report: dict[str, Any] = {
+        "name": name,
+        "source_path": item.source_path,
+        "status": result.status,
+        "needs_confirmation": result.needs_confirmation,
+        "issues": [
+            {"node_id": i.node_id, "code": i.code, "level": i.level, "message": i.message}
+            for i in result.issues
+        ],
+    }
+    if result.workflow is None:
+        report["importable"] = False
+        report["skip_reason"] = "转换失败：" + (
+            result.issues[0].message if result.issues else "未知错误"
+        )
+        return report, None
+    suggested = suggested_metadata(result.workflow, item.source_path, item.workflow_json)
+    report["suggested"] = suggested
+    # The import gate re-runs the product's own checks — a converted graph
+    # that the lint would block (missing model link etc.) is reported, not
+    # silently imported.
+    broken = blocking_problem(result.workflow)
+    if broken:
+        report["importable"] = False
+        report["skip_reason"] = broken
+        report["status"] = "warn"
+        return report, result.workflow
+    if result.needs_confirmation:
+        report["importable"] = False
+        report["skip_reason"] = "存在无法可靠映射的控件，需人工确认后重导"
+        return report, result.workflow
+    report["importable"] = True
+    return report, result.workflow
+
+
+@router.post("/convert")
+async def convert_workflows(payload: WorkflowConvertRequest) -> dict[str, Any]:
+    """Dry-run UI→API conversion for a batch of files: per-file ok/warn/fail,
+    issues, suggested metadata and importability. Nothing is stored."""
+    files = []
+    object_info: dict[str, Any] | None = None
+    client = None
+    if payload.endpoint_id:
+        endpoint = settings.resolve_comfyui_endpoint(payload.endpoint_id)
+        if endpoint is not None:
+            client = client_for(endpoint)
+            try:
+                object_info = await client.object_info(force_refresh=payload.refresh)
+            except Exception:
+                object_info = None
+    try:
+        for item in payload.files:
+            report, converted = _convert_one(item)
+            if object_info is not None and report.get("importable") and converted:
+                findings = check_workflow_against(object_info, converted)
+                errors = [f for f in findings if f.get("level") == "error"]
+                report["preflight"] = {
+                    "endpoint_id": payload.endpoint_id,
+                    "ok": not errors,
+                    "findings": findings,
+                }
+                if errors:
+                    report["importable"] = False
+                    report["skip_reason"] = errors[0].get("message") or "preflight 未通过"
+            if payload.include_workflows:
+                report["workflow_json"] = converted
+            files.append(report)
+    finally:
+        if client is not None:
+            await client.close()
+    counts = {key: sum(1 for f in files if f["status"] == key) for key in ("ok", "warn", "fail")}
+    return {
+        "total": len(files),
+        **counts,
+        "needs_confirmation": sum(1 for f in files if f["needs_confirmation"]),
+        "importable": sum(1 for f in files if f.get("importable")),
+        "files": files,
+    }
+
+
+def _insert_converted(
+    conn: Any, item: WorkflowUIFileItem, report: dict[str, Any], converted: dict[str, Any]
+) -> dict[str, Any]:
+    """Insert one converted workflow row (import-batch's write path).
+
+    Mirrors create_workflow's enable gate: usage must be settled (auto or the
+    wizard's confirmation) and unknown outputs need a mapping; anything less
+    lands disabled exactly like a manual import would.
+    """
+    suggested = report["suggested"]
+    judgment = classify_usage(converted)
+    usage = item.usage or judgment["usage"]
+    usage_source = "user" if item.usage else "auto"
+    kind = usage_to_kind(usage) or suggested["kind"]
+    profile = detect_prompt_profile(converted)
+    if item.output_map:
+        problems = validate_output_map(converted, item.output_map)
+        if problems:
+            raise HTTPException(status_code=422, detail="；".join(problems))
+        output_map = json.dumps(item.output_map)
+        output_map_source = "user"
+    else:
+        auto_map = auto_output_map(converted)
+        output_map = json.dumps(auto_map) if auto_map else None
+        output_map_source = "auto" if auto_map else None
+    mapping_ok = not needs_user_output_map(converted) or bool(item.output_map)
+    workflow_text = json.dumps(converted)
+    existing = conn.execute(
+        "SELECT id FROM workflows WHERE name = ? AND kind = ? AND workflow_json = ?",
+        (report["name"], kind, workflow_text),
+    ).fetchone()
+    if existing:
+        return {
+            "result": "skipped",
+            "skip_reason": "conflict",
+            "existing_id": existing["id"],
+        }
+    is_enabled = 1 if (usage and mapping_ok) else 0
+    cur = conn.execute(
+        """
+        INSERT INTO workflows (name, kind, workflow_json, input_schema, output_schema,
+                               description, prompt_profile, endpoint_id, is_enabled,
+                               usage, usage_source, output_map, output_map_source,
+                               category, model_family, media_io, tags, favorite,
+                               source, source_path)
+        VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 'ui_convert', ?)
+        """,
+        (
+            report["name"],
+            kind,
+            workflow_text,
+            json.dumps(
+                item.input_schema if item.input_schema else parse_dynamic_inputs(converted)
+            ),
+            json.dumps(parse_dynamic_outputs(converted)),
+            item.description if item.description is not None else suggested["description"],
+            profile,
+            is_enabled,
+            usage,
+            usage_source,
+            output_map,
+            output_map_source,
+            suggested["category"],
+            suggested["model_family"],
+            json.dumps({"in": suggested["media_in"], "out": suggested["media_out"]}),
+            json.dumps(item.tags or []),
+            item.source_path,
+        ),
+    )
+    return {"result": "imported", "id": cur.lastrowid}
+
+
+async def _run_import(items: list[WorkflowUIFileItem]) -> dict[str, Any]:
+    reports: list[dict[str, Any]] = []
+    conn = get_db(settings.db_path)
+    try:
+        for item in items:
+            report, converted = _convert_one(item)
+            if not report.get("importable"):
+                report["result"] = "skipped"
+                reports.append(report)
+                continue
+            try:
+                outcome = _insert_converted(conn, item, report, converted)
+            except HTTPException as exc:
+                report["result"] = "failed"
+                report["skip_reason"] = str(exc.detail)
+                reports.append(report)
+                continue
+            conn.commit()
+            report.update(outcome)
+            reports.append(report)
+    finally:
+        conn.close()
+    imported = [r for r in reports if r.get("result") == "imported"]
+    skipped = [
+        {"name": r["name"], "source_path": r["source_path"], "reason": r.get("skip_reason")}
+        for r in reports
+        if r.get("result") != "imported"
+    ]
+    return {
+        "total": len(reports),
+        "imported": len(imported),
+        "skipped": len(skipped),
+        "imported_ids": [r["id"] for r in imported],
+        "skipped_files": skipped,
+        "files": reports,
+    }
+
+
+@router.post("/import-batch")
+async def import_workflows_batch(payload: WorkflowImportBatchRequest) -> dict[str, Any]:
+    """Batch-import UI-format workflows: convert server-side, insert what is
+    cleanly convertible, skip conflicts and anything uncertain. Re-running
+    with the same files creates zero new rows."""
+    return await _run_import(payload.files)
+
+
+@router.post("/import-scan")
+async def import_workflows_scan(payload: WorkflowImportScanRequest) -> dict[str, Any]:
+    """Scan a server-side directory for UI-format workflow JSONs and run the
+    convert (dry_run, default) or import pipeline over every file found."""
+    from pathlib import Path
+
+    root = Path(payload.root)
+    if not root.is_absolute():
+        candidates = [
+            Path.cwd() / root,
+            Path(__file__).resolve().parents[4] / root,
+        ]
+        root = next((c for c in candidates if c.is_dir()), root)
+    if not root.is_dir():
+        raise HTTPException(status_code=404, detail=f"目录不存在：{payload.root}")
+    items: list[WorkflowUIFileItem] = []
+    for path in sorted(root.rglob("*.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            continue
+        if not (isinstance(data, dict) and isinstance(data.get("nodes"), list)):
+            continue  # API-format or non-workflow files: not this pipeline's job
+        items.append(
+            WorkflowUIFileItem(
+                source_path=str(path.relative_to(root)), workflow_json=data
+            )
+        )
+    if payload.dry_run:
+        files = [_convert_one(item)[0] for item in items]
+        counts = {
+            key: sum(1 for f in files if f["status"] == key) for key in ("ok", "warn", "fail")
+        }
+        return {
+            "total": len(files),
+            **counts,
+            "needs_confirmation": sum(1 for f in files if f["needs_confirmation"]),
+            "importable": sum(1 for f in files if f.get("importable")),
+            "files": files,
+        }
+    return await _run_import(items)
+
+
+@router.post("/{workflow_id}/touch")
+async def touch_workflow(workflow_id: int) -> dict[str, bool]:
+    """Record a use (最近使用) — called by the generate pipeline on enqueue."""
+    conn = get_db(settings.db_path)
+    try:
+        row = conn.execute("SELECT id FROM workflows WHERE id = ?", (workflow_id,)).fetchone()
+        if not row:
+            raise HTTPException(status_code=404, detail="Workflow not found")
+        conn.execute(
+            "UPDATE workflows SET last_used_at = CURRENT_TIMESTAMP WHERE id = ?",
+            (workflow_id,),
+        )
+        conn.commit()
+        return {"ok": True}
+    finally:
+        conn.close()

calliope-backend/tests/test_ui_convert.py

@@ -0,0 +1,449 @@
+"""HEL-681: UI→API converter unit tests + ground-truth cross-checks.
+
+The synthetic graphs pin the structural semantics (reroute / bypass / mute /
+subgraph expansion / Set-Get variables / control_after_generate pairing);
+the corpus section runs the real 146-file library and the repo's API exports.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+from pathlib import Path
+
+import pytest
+
+from calliope.comfyui.lint import workflow_warnings
+from calliope.comfyui.ui_convert import convert_ui_workflow
+
+EXAMPLES = Path(__file__).resolve().parents[2] / "example_ComfyUI_workflows"
+
+
+def _node(nid, ntype, *, mode=0, inputs=None, outputs=None, wv=None, wvn=None, title=None):
+    node = {"id": nid, "type": ntype, "mode": mode, "inputs": inputs or [], "order": 0}
+    if outputs is not None:
+        node["outputs"] = outputs
+    if wv is not None:
+        node["widgets_values"] = wv
+    if wvn is not None:
+        node["widgets_values_named"] = wvn
+    if title:
+        node["title"] = title
+    return node
+
+
+def _in_slot(name, type_, link=None, widget=None):
+    entry = {"name": name, "type": type_, "link": link}
+    if widget:
+        entry["widget"] = {"name": widget}
+    return entry
+
+
+def _out(name, type_, links=None):
+    return {"name": name, "type": type_, "links": links or [], "slot_index": 0}
+
+
+def _base_graph():
+    """text2image: loader → prompts/latent → KSampler → VAE → SaveImage."""
+    nodes = [
+        _node(
+            1,
+            "CheckpointLoaderSimple",
+            outputs=[_out("MODEL", "MODEL"), _out("CLIP", "CLIP"), _out("VAE", "VAE")],
+            wv=["model.safetensors"],
+            inputs=[_in_slot("ckpt_name", "COMBO", widget="ckpt_name")],
+        ),
+        _node(
+            2,
+            "CLIPTextEncode",
+            inputs=[
+                _in_slot("clip", "CLIP", link=1),
+                _in_slot("text", "STRING", widget="text"),
+            ],
+            outputs=[_out("CONDITIONING", "CONDITIONING")],
+            wv=["a cat"],
+        ),
+        _node(
+            3,
+            "CLIPTextEncode",
+            inputs=[
+                _in_slot("clip", "CLIP", link=2),
+                _in_slot("text", "STRING", widget="text"),
+            ],
+            outputs=[_out("CONDITIONING", "CONDITIONING")],
+            wv=["bad quality"],
+        ),
+        _node(
+            4,
+            "EmptyLatentImage",
+            inputs=[
+                _in_slot("width", "INT", widget="width"),
+                _in_slot("height", "INT", widget="height"),
+                _in_slot("batch_size", "INT", widget="batch_size"),
+            ],
+            outputs=[_out("LATENT", "LATENT")],
+            wv=[512, 512, 1],
+        ),
+        _node(
+            5,
+            "KSampler",
+            inputs=[
+                _in_slot("model", "MODEL", link=3),
+                _in_slot("positive", "CONDITIONING", link=4),
+                _in_slot("negative", "CONDITIONING", link=5),
+                _in_slot("latent_image", "LATENT", link=6),
+                _in_slot("seed", "INT", widget="seed"),
+                _in_slot("steps", "INT", widget="steps"),
+                _in_slot("cfg", "FLOAT", widget="cfg"),
+                _in_slot("sampler_name", "COMBO", widget="sampler_name"),
+                _in_slot("scheduler", "COMBO", widget="scheduler"),
+                _in_slot("denoise", "FLOAT", widget="denoise"),
+            ],
+            outputs=[_out("LATENT", "LATENT")],
+            # seed 42 + control "fixed" rides together, then five widgets
+            wv=[42, "fixed", 20, 7, "euler", "normal", 1.0],
+        ),
+        _node(
+            6,
+            "VAEDecode",
+            inputs=[
+                _in_slot("samples", "LATENT", link=7),
+                _in_slot("vae", "VAE", link=8),
+            ],
+            outputs=[_out("IMAGE", "IMAGE")],
+        ),
+        _node(
+            7,
+            "SaveImage",
+            inputs=[
+                _in_slot("images", "IMAGE", link=9),
+                _in_slot("filename_prefix", "STRING", widget="filename_prefix"),
+            ],
+            wv=["test"],
+        ),
+    ]
+    links = [
+        [1, 1, 1, 2, 0, "CLIP"],
+        [2, 1, 1, 3, 0, "CLIP"],
+        [3, 1, 0, 5, 0, "MODEL"],
+        [4, 2, 0, 5, 1, "CONDITIONING"],
+        [5, 3, 0, 5, 2, "CONDITIONING"],
+        [6, 4, 0, 5, 3, "LATENT"],
+        [7, 5, 0, 6, 0, "LATENT"],
+        [8, 1, 2, 6, 1, "VAE"],
+        [9, 6, 0, 7, 0, "IMAGE"],
+    ]
+    return {"nodes": nodes, "links": links, "last_node_id": 7, "last_link_id": 9}
+
+
+def _convert(graph):
+    return convert_ui_workflow(graph)
+
+
+def test_clean_conversion_maps_everything():
+    result = _convert(_base_graph())
+    assert result.status == "ok"
+    assert result.needs_confirmation is False
+    wf = result.workflow
+    assert set(wf) == {"1", "2", "3", "4", "5", "6", "7"}
+    ks = wf["5"]
+    assert ks["class_type"] == "KSampler"
+    assert ks["inputs"]["seed"] == 42  # control_after_generate consumed silently
+    assert ks["inputs"]["steps"] == 20
+    assert ks["inputs"]["denoise"] == 1.0
+    assert ks["inputs"]["model"] == ["1", 0]
+    assert ks["inputs"]["latent_image"] == ["4", 0]
+    # lint on the converted graph must be error-free
+    assert not [w for w in workflow_warnings(wf) if w["level"] == "error"]
+
+
+def test_upload_button_widgets_are_consumed_but_never_emitted():
+    graph = _base_graph()
+    load = _node(
+        8,
+        "LoadImage",
+        inputs=[
+            _in_slot("image", "COMBO", widget="image"),
+            _in_slot("upload", "IMAGEUPLOAD", widget="upload"),
+        ],
+        outputs=[_out("IMAGE", "IMAGE")],
+        wv=["sample.png", "image"],
+    )
+    save = _node(
+        9,
+        "PreviewImage",
+        inputs=[
+            _in_slot("images", "IMAGE", link=20),
+            _in_slot("filename_prefix", "STRING", widget="filename_prefix"),
+        ],
+        wv=["preview"],
+    )
+    graph["nodes"] += [load, save]
+    graph["links"].append([20, 8, 0, 9, 0, "IMAGE"])
+    result = _convert(graph)
+    assert result.workflow["8"]["inputs"] == {"image": "sample.png"}
+
+
+def test_reroute_pass_through():
+    graph = _base_graph()
+    reroute = _node(8, "Reroute", inputs=[_in_slot("(in)", "*", link=9)])
+    graph["nodes"].append(reroute)
+    # SaveImage now fed by the reroute instead of VAEDecode directly
+    graph["nodes"][6]["inputs"][0]["link"] = 10
+    graph["links"].append([10, 8, 0, 7, 0, "IMAGE"])
+    result = _convert(graph)
+    assert result.workflow["7"]["inputs"]["images"] == ["6", 0]
+    assert "8" not in result.workflow
+
+
+def test_bypass_pass_through_and_mute_drop():
+    graph = _base_graph()
+    # bypass the KSampler: VAEDecode.samples must fall back to the latent
+    graph["nodes"][4]["mode"] = 4
+    result = _convert(graph)
+    assert result.workflow["6"]["inputs"]["samples"] == ["4", 0]
+    # mute the negative prompt: the link input is dropped with a warning
+    graph = _base_graph()
+    graph["nodes"][2]["mode"] = 2
+    result = _convert(graph)
+    assert "negative" not in result.workflow["5"]["inputs"]
+    assert any(i.code == "muted_input_dropped" for i in result.issues)
+
+
+def test_widget_entry_with_muted_upstream_falls_back_to_value():
+    graph = _base_graph()
+    # feed seed from a muted PrimitiveInt via a converted widget input
+    prim = _node(
+        9,
+        "PrimitiveInt",
+        mode=2,
+        inputs=[_in_slot("value", "INT", widget="value"), _in_slot("control", "COMBO", widget="control")],
+        wv=[7],
+    )
+    graph["nodes"].append(prim)
+    graph["nodes"][4]["inputs"][4]["link"] = 11  # seed ← primitive
+    graph["links"].append([11, 9, 0, 5, 4, "INT"])
+    result = _convert(graph)
+    assert result.workflow["5"]["inputs"]["seed"] == 42
+
+
+def test_subgraph_expansion_and_boundary_wiring():
+    graph = _base_graph()
+    # pack the EmptyLatentImage into a subgraph exposing width/height widgets
+    del graph["nodes"][3]
+    graph["nodes"] = [n for n in graph["nodes"] if n["id"] != 4]
+    sub_id = "aaaa-bbbb-cccc-dddd-eeee"
+    packed = _node(
+        20,
+        sub_id,
+        inputs=[
+            _in_slot("width", "INT", widget="width"),
+            _in_slot("height", "INT", widget="height"),
+        ],
+        outputs=[_out("LATENT", "LATENT")],
+        wv=[640, 384],
+    )
+    graph["nodes"].append(packed)
+    # KSampler latent_image ← packed output
+    graph["links"] = [l for l in graph["links"] if l[0] != 6]
+    graph["links"].append([6, 20, 0, 5, 3, "LATENT"])
+    graph["definitions"] = {
+        "subgraphs": [
+            {
+                "id": sub_id,
+                "name": "Latent",
+                "inputs": [
+                    {"name": "width", "type": "INT"},
+                    {"name": "height", "type": "INT"},
+                ],
+                "outputs": [{"name": "LATENT", "type": "LATENT"}],
+                "inputNode": {"id": -10},
+                "outputNode": {"id": -20},
+                "nodes": [
+                    _node(
+                        5,
+                        "EmptyLatentImage",
+                        inputs=[
+                            _in_slot("width", "INT", widget="width", link=101),
+                            _in_slot("height", "INT", widget="height", link=102),
+                            _in_slot("batch_size", "INT", widget="batch_size"),
+                        ],
+                        outputs=[_out("LATENT", "LATENT")],
+                        wv=[512, 512, 1],
+                    ),
+                ],
+                "links": [
+                    {"id": 101, "origin_id": -10, "origin_slot": 0, "target_id": 5, "target_slot": 0, "type": "INT"},
+                    {"id": 102, "origin_id": -10, "origin_slot": 1, "target_id": 5, "target_slot": 1, "type": "INT"},
+                    {"id": 103, "origin_id": 5, "origin_slot": 0, "target_id": -20, "target_slot": 0, "type": "LATENT"},
+                ],
+            }
+        ]
+    }
+    result = _convert(graph)
+    wf = result.workflow
+    assert "20:5" in wf  # qualified packed:inner id, ComfyUI's own scheme
+    inner = wf["20:5"]
+    assert inner["class_type"] == "EmptyLatentImage"
+    # exposed widget values from the packed instance, not the inner defaults
+    assert inner["inputs"]["width"] == 640
+    assert inner["inputs"]["height"] == 384
+    assert inner["inputs"]["batch_size"] == 1
+    assert wf["5"]["inputs"]["latent_image"] == ["20:5", 0]
+    # now wire the packed width slot from outside: boundary becomes a link
+    res_node = _node(
+        21,
+        "ResolutionSelector",
+        inputs=[_in_slot("aspect_ratio", "COMBO", widget="aspect_ratio")],
+        outputs=[_out("width", "INT"), _out("height", "INT")],
+        wv=["1:1"],
+    )
+    graph["nodes"].append(res_node)
+    graph["nodes"][-2]["inputs"][0]["link"] = 12  # packed width ← selector
+    graph["links"].append([12, 21, 0, 20, 0, "INT"])
+    wf2 = _convert(graph).workflow
+    assert wf2["20:5"]["inputs"]["width"] == ["21", 0]
+    assert wf2["20:5"]["inputs"]["height"] == 384
+
+
+def test_set_get_variables_resolve_to_captured_source():
+    graph = _base_graph()
+    setter = _node(8, "SetNode", inputs=[_in_slot("*", "*", link=8)], wv=["the_vae"])
+    getter = _node(9, "GetNode", wv=["the_vae"], outputs=[_out("VAE", "VAE")])
+    graph["nodes"] += [setter, getter]
+    # VAEDecode.vae now fed via the variable
+    graph["nodes"][5]["inputs"][1]["link"] = 13
+    graph["links"] = [l for l in graph["links"] if l[0] != 8]
+    graph["links"] += [
+        [8, 1, 2, 8, 0, "VAE"],
+        [13, 9, 0, 6, 1, "VAE"],
+    ]
+    result = _convert(graph)
+    assert "8" not in result.workflow and "9" not in result.workflow
+    assert result.workflow["6"]["inputs"]["vae"] == ["1", 2]
+
+
+def test_unknown_widget_order_lands_in_needs_confirmation():
+    graph = _base_graph()
+    # an old-frontend node: positional widgets, no widget markers, no registry
+    graph["nodes"].append(
+        _node(10, "MysteryCustomNode", inputs=[], outputs=[_out("IMAGE", "IMAGE")], wv=[1, 2, 3])
+    )
+    save = _node(
+        11,
+        "PreviewImage",
+        inputs=[_in_slot("images", "IMAGE", link=30)],
+        wv=["preview"],
+    )
+    graph["nodes"].append(save)
+    graph["links"].append([30, 10, 0, 11, 0, "IMAGE"])
+    result = _convert(graph)
+    assert result.needs_confirmation is True
+    assert any(i.code == "widget_unmapped" for i in result.issues)
+    # the node is still converted structurally — the report carries the doubt
+    assert "10" in result.workflow
+
+
+def test_not_ui_format_fails_fast():
+    result = convert_ui_workflow({"3": {"class_type": "SaveImage", "inputs": {}}})
+    assert result.status == "fail"
+    assert result.workflow is None
+
+
+# ---------------------------------------------------------------------------
+# Real corpus: the 146 UI files + ground-truth API cross-checks (对拍).
+# ---------------------------------------------------------------------------
+
+pytestmark_convert = pytest.mark.skipif(
+    not EXAMPLES.is_dir(), reason="example_ComfyUI_workflows not checked out"
+)
+
+
+@pytest.mark.skipif(not EXAMPLES.is_dir(), reason="example library missing")
+def test_corpus_146_files_all_convert():
+    ui_files = [
+        p
+        for p in sorted(EXAMPLES.rglob("*.json"))
+        if isinstance(json.loads(p.read_text(encoding="utf-8")), dict)
+        and isinstance(json.loads(p.read_text(encoding="utf-8")).get("nodes"), list)
+    ]
+    assert len(ui_files) == 146
+    statuses = {"ok": 0, "warn": 0, "fail": 0}
+    needs_confirmation = 0
+    for path in ui_files:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        result = convert_ui_workflow(data)
+        statuses[result.status] += 1
+        needs_confirmation += result.needs_confirmation
+        assert result.workflow, f"{path.name} produced no workflow"
+        # structural sanity: no dangling links in any conversion
+        nodes = result.workflow
+        for nid, node in nodes.items():
+            for name, val in node.get("inputs", {}).items():
+                if isinstance(val, list):
+                    assert str(val[0]) in nodes, f"{path.name}: {nid}.{name} dangles"
+    assert statuses["fail"] == 0
+    assert statuses["ok"] + statuses["warn"] == 146
+    # files whose widget order cannot be trusted are flagged, never hidden
+    assert 0 < needs_confirmation < 10
+
+
+@pytest.mark.skipif(not EXAMPLES.is_dir(), reason="example library missing")
+def test_cross_check_against_ground_truth_exports():
+    """The H3 t2v UI export converts to the same node classes the author's
+    own API export carries; shared classes agree on scalar input names."""
+    ui = json.loads(
+        (EXAMPLES / "4-视频生成/Minimax-h3/minimax_h3_t2v_文生视频.json").read_text(encoding="utf-8")
+    )
+    api = json.loads(
+        (EXAMPLES / "minimax_h3-Turbo_Text2Video_20260813_API.json").read_text(encoding="utf-8")
+    )
+    result = convert_ui_workflow(ui)
+    assert result.status in ("ok", "warn")
+    converted = result.workflow
+    api_classes = {n["class_type"] for n in api.values()}
+    # Every scalar name the conversion emits for a shared class must be an
+    # input name the author's own export knows — wiring may differ between
+    # versions (a muted branch falls back to its widget value), but the NAME
+    # mapping must never invent or swap inputs.
+    for cls in api_classes & {n["class_type"] for n in converted.values()}:
+        api_names = {
+            key
+            for n in api.values()
+            if n["class_type"] == cls
+            for key in n.get("inputs", {})
+        }
+        conv_names = {
+            key
+            for n in converted.values()
+            if n["class_type"] == cls
+            for key, v in n.get("inputs", {}).items()
+            if not isinstance(v, list)
+        }
+        assert conv_names <= api_names, f"class {cls} invents inputs: {conv_names - api_names}"
+
+
+@pytest.mark.skipif(not EXAMPLES.is_dir(), reason="example library missing")
+def test_packed_krea2_conversion_shape_matches_export_scheme():
+    """The krea2 turbo file (packed subgraph) converts ok, expands inner nodes
+    under `<packed>:<inner>` ids and maps the sampler widgets exactly — the
+    same scheme the author's own Krea2 API export uses (ids like "171:153")."""
+    ui = json.loads(
+        (EXAMPLES / "3-图像生成/krea2/krea2_turbo_文生图.json").read_text(encoding="utf-8")
+    )
+    api = json.loads(
+        (EXAMPLES / "Krea2_t2i_20260818_API.json").read_text(encoding="utf-8")
+    )
+    result = convert_ui_workflow(ui)
+    assert result.status == "ok"
+    assert any(":" in nid for nid in result.workflow)  # expanded subgraph nodes
+    assert any("171:" in nid for nid in api)  # ground truth uses the same scheme
+    ks = [
+        n
+        for n in result.workflow.values()
+        if n["class_type"] == "KSampler" and not isinstance(n["inputs"].get("seed"), list)
+    ]
+    assert ks, "inner KSampler with widget-mapped seed expected"
+    assert {"seed", "steps", "cfg", "sampler_name", "scheduler", "denoise"} <= set(
+        ks[0]["inputs"]
+    )

calliope-backend/tests/test_workflow_import_batch.py

@@ -0,0 +1,421 @@
+"""HEL-681 (S2): batch import endpoints, meta list view, migration safety.
+
+Everything runs on a temp DB via the shared client fixture — the real
+146-file library is never touched, matching the card's「不导入生产数据」boundary.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+
+from calliope.db import migrate_db
+
+import pytest
+
+
+EXAMPLES = Path(__file__).resolve().parents[2] / "example_ComfyUI_workflows"
+
+
+def _ui_graph():
+    """Minimal clean t2i UI graph (same shape as test_ui_convert's)."""
+    return {
+        "nodes": [
+            {
+                "id": 1,
+                "type": "CheckpointLoaderSimple",
+                "mode": 0,
+                "inputs": [{"name": "ckpt_name", "type": "COMBO", "widget": {"name": "ckpt_name"}, "link": None}],
+                "outputs": [
+                    {"name": "MODEL", "type": "MODEL", "links": [3], "slot_index": 0},
+                    {"name": "CLIP", "type": "CLIP", "links": [1, 2], "slot_index": 1},
+                    {"name": "VAE", "type": "VAE", "links": [8], "slot_index": 2},
+                ],
+                "widgets_values": ["qwen_image_2.safetensors"],
+            },
+            {
+                "id": 2,
+                "type": "CLIPTextEncode",
+                "mode": 0,
+                "inputs": [
+                    {"name": "clip", "type": "CLIP", "link": 1},
+                    {"name": "text", "type": "STRING", "widget": {"name": "text"}, "link": None},
+                ],
+                "outputs": [{"name": "CONDITIONING", "type": "CONDITIONING", "links": [4], "slot_index": 0}],
+                "widgets_values": ["a cat"],
+            },
+            {
+                "id": 3,
+                "type": "CLIPTextEncode",
+                "mode": 0,
+                "inputs": [
+                    {"name": "clip", "type": "CLIP", "link": 2},
+                    {"name": "text", "type": "STRING", "widget": {"name": "text"}, "link": None},
+                ],
+                "outputs": [{"name": "CONDITIONING", "type": "CONDITIONING", "links": [5], "slot_index": 0}],
+                "widgets_values": ["bad"],
+            },
+            {
+                "id": 4,
+                "type": "EmptyLatentImage",
+                "mode": 0,
+                "inputs": [
+                    {"name": "width", "type": "INT", "widget": {"name": "width"}, "link": None},
+                    {"name": "height", "type": "INT", "widget": {"name": "height"}, "link": None},
+                    {"name": "batch_size", "type": "INT", "widget": {"name": "batch_size"}, "link": None},
+                ],
+                "outputs": [{"name": "LATENT", "type": "LATENT", "links": [6], "slot_index": 0}],
+                "widgets_values": [512, 512, 1],
+            },
+            {
+                "id": 5,
+                "type": "KSampler",
+                "mode": 0,
+                "inputs": [
+                    {"name": "model", "type": "MODEL", "link": 3},
+                    {"name": "positive", "type": "CONDITIONING", "link": 4},
+                    {"name": "negative", "type": "CONDITIONING", "link": 5},
+                    {"name": "latent_image", "type": "LATENT", "link": 6},
+                    {"name": "seed", "type": "INT", "widget": {"name": "seed"}, "link": None},
+                    {"name": "steps", "type": "INT", "widget": {"name": "steps"}, "link": None},
+                    {"name": "cfg", "type": "FLOAT", "widget": {"name": "cfg"}, "link": None},
+                    {"name": "sampler_name", "type": "COMBO", "widget": {"name": "sampler_name"}, "link": None},
+                    {"name": "scheduler", "type": "COMBO", "widget": {"name": "scheduler"}, "link": None},
+                    {"name": "denoise", "type": "FLOAT", "widget": {"name": "denoise"}, "link": None},
+                ],
+                "outputs": [{"name": "LATENT", "type": "LATENT", "links": [7], "slot_index": 0}],
+                "widgets_values": [42, "fixed", 20, 7, "euler", "normal", 1.0],
+            },
+            {
+                "id": 6,
+                "type": "VAEDecode",
+                "mode": 0,
+                "inputs": [
+                    {"name": "samples", "type": "LATENT", "link": 7},
+                    {"name": "vae", "type": "VAE", "link": 8},
+                ],
+                "outputs": [{"name": "IMAGE", "type": "IMAGE", "links": [9], "slot_index": 0}],
+            },
+            {
+                "id": 7,
+                "type": "SaveImage",
+                "mode": 0,
+                "inputs": [
+                    {"name": "images", "type": "IMAGE", "link": 9},
+                    {"name": "filename_prefix", "type": "STRING", "widget": {"name": "filename_prefix"}, "link": None},
+                ],
+                "widgets_values": ["test"],
+            },
+        ],
+        "links": [
+            [1, 1, 1, 2, 0, "CLIP"],
+            [2, 1, 1, 3, 0, "CLIP"],
+            [3, 1, 0, 5, 0, "MODEL"],
+            [4, 2, 0, 5, 1, "CONDITIONING"],
+            [5, 3, 0, 5, 2, "CONDITIONING"],
+            [6, 4, 0, 5, 3, "LATENT"],
+            [7, 5, 0, 6, 0, "LATENT"],
+            [8, 1, 2, 6, 1, "VAE"],
+            [9, 6, 0, 7, 0, "IMAGE"],
+        ],
+    }
+
+
+def _api_workflow():
+    return json.loads(
+        (EXAMPLES / "video_minimax_h3_r2v_2ref_API.json").read_text(encoding="utf-8")
+    )
+
+
+def test_convert_dry_run_reports_and_stores_nothing(client):
+    payload = {
+        "files": [
+            {
+                "source_path": "3-图像生成/Qwen-Image-2.1/test_t2i.json",
+                "workflow_json": _ui_graph(),
+            }
+        ],
+        "include_workflows": True,
+    }
+    resp = client.post("/api/workflows/convert", json=payload)
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["total"] == 1 and body["ok"] == 1 and body["importable"] == 1
+    entry = body["files"][0]
+    assert entry["name"] == "test_t2i"
+    assert entry["workflow_json"]["5"]["inputs"]["seed"] == 42
+    assert entry["suggested"]["usage"] == "text2image"
+    assert entry["suggested"]["category"] == "image"
+    assert entry["suggested"]["model_family"] == "Qwen"
+    assert entry["suggested"]["media_out"] == ["image"]
+    # dry-run must not write anything
+    assert client.get("/api/workflows").json() == []
+
+
+def test_convert_needs_confirmation_is_flagged_not_runnable(client):
+    graph = _ui_graph()
+    graph["nodes"].append(
+        {
+            "id": 10,
+            "type": "MysteryCustomNode",
+            "mode": 0,
+            "inputs": [],
+            "outputs": [{"name": "IMAGE", "type": "IMAGE", "links": [30], "slot_index": 0}],
+            "widgets_values": [1, 2, 3],
+        }
+    )
+    graph["nodes"].append(
+        {
+            "id": 11,
+            "type": "PreviewImage",
+            "mode": 0,
+            "inputs": [{"name": "images", "type": "IMAGE", "link": 30}],
+            "widgets_values": ["p"],
+        }
+    )
+    graph["links"].append([30, 10, 0, 11, 0, "IMAGE"])
+    resp = client.post(
+        "/api/workflows/convert",
+        json={"files": [{"source_path": "x/mystery.json", "workflow_json": graph}]},
+    )
+    body = resp.json()
+    assert body["needs_confirmation"] == 1
+    assert body["importable"] == 0
+    assert "需人工确认" in body["files"][0]["skip_reason"]
+
+
+def test_import_batch_then_rerun_is_idempotent(client):
+    files = [
+        {
+            "source_path": "3-图像生成/Qwen-Image-2.1/test_t2i.json",
+            "workflow_json": _ui_graph(),
+        }
+    ]
+    first = client.post("/api/workflows/import-batch", json={"files": files})
+    assert first.status_code == 200
+    body = first.json()
+    assert body["imported"] == 1 and body["skipped"] == 0
+    row_id = body["imported_ids"][0]
+
+    # second run with identical file: conflict skip, no duplicates
+    second = client.post("/api/workflows/import-batch", json={"files": files})
+    body2 = second.json()
+    assert body2["imported"] == 0 and body2["skipped"] == 1
+    assert body2["skipped_files"][0]["reason"] == "conflict"
+    listing = client.get("/api/workflows").json()
+    assert len(listing) == 1
+
+    detail = client.get(f"/api/workflows/{row_id}").json()
+    assert detail["usage"] == "text2image"
+    assert detail["is_enabled"] is True
+    assert detail["category"] == "image"
+    assert detail["model_family"] == "Qwen"
+    assert detail["media_io"] == {"in": [], "out": ["image"]}
+    assert detail["source"] == "ui_convert"
+    assert detail["source_path"] == "3-图像生成/Qwen-Image-2.1/test_t2i.json"
+
+
+def test_import_batch_skips_needs_confirmation_files(client):
+    graph = _ui_graph()
+    graph["nodes"].append(
+        {
+            "id": 10,
+            "type": "MysteryCustomNode",
+            "mode": 0,
+            "inputs": [],
+            "outputs": [{"name": "IMAGE", "type": "IMAGE", "links": [30], "slot_index": 0}],
+            "widgets_values": [1, 2, 3],
+        }
+    )
+    graph["nodes"].append(
+        {
+            "id": 11,
+            "type": "PreviewImage",
+            "mode": 0,
+            "inputs": [{"name": "images", "type": "IMAGE", "link": 30}],
+            "widgets_values": ["p"],
+        }
+    )
+    graph["links"].append([30, 10, 0, 11, 0, "IMAGE"])
+    resp = client.post(
+        "/api/workflows/import-batch",
+        json={"files": [{"name": "mystery", "workflow_json": graph}]},
+    )
+    body = resp.json()
+    assert body["imported"] == 0
+    assert "需人工确认" in body["skipped_files"][0]["reason"]
+    assert client.get("/api/workflows").json() == []
+
+
+def test_warn_sample_imports_with_human_confirmation(client):
+    """A warn file (muted branch → input dropped) still imports when the
+    wizard supplies the settled usage — the manual-correction path."""
+    graph = _ui_graph()
+    graph["nodes"][2]["mode"] = 2  # mute the negative prompt branch
+    dry = client.post(
+        "/api/workflows/convert",
+        json={"files": [{"name": "warned", "workflow_json": graph}]},
+    ).json()
+    assert dry["warn"] == 1 and dry["importable"] == 1
+    resp = client.post(
+        "/api/workflows/import-batch",
+        json={"files": [{"name": "warned", "workflow_json": graph, "usage": "text2image"}]},
+    )
+    body = resp.json()
+    assert body["imported"] == 1
+    detail = client.get(f"/api/workflows/{body['imported_ids'][0]}").json()
+    assert detail["is_enabled"] is True
+    assert "negative" not in detail["workflow_json"]["5"]["inputs"]
+
+
+@pytest.mark.skipif(not EXAMPLES.is_dir(), reason="example library missing")
+def test_import_scan_dry_run_over_repo_library(client):
+    resp = client.post(
+        "/api/workflows/import-scan",
+        json={"root": str(EXAMPLES), "dry_run": True},
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["total"] == 146
+    assert body["fail"] == 0
+    assert body["ok"] + body["warn"] == 146
+    assert body["needs_confirmation"] > 0  # the 3 uncensored files stay honest
+    assert body["importable"] < body["total"]
+    # a dry scan must leave the DB untouched
+    assert client.get("/api/workflows").json() == []
+
+
+def test_import_scan_commit_from_tmp_directory(client, tmp_path):
+    (tmp_path / "sub").mkdir()
+    (tmp_path / "sub" / "demo.json").write_text(json.dumps(_ui_graph()), encoding="utf-8")
+    (tmp_path / "plain_api.json").write_text(json.dumps(_api_workflow()), encoding="utf-8")
+    (tmp_path / "not_a_workflow.json").write_text(json.dumps({"hello": 1}), encoding="utf-8")
+    resp = client.post(
+        "/api/workflows/import-scan",
+        json={"root": str(tmp_path), "dry_run": False},
+    )
+    body = resp.json()
+    # only the one UI file went through this pipeline; API-format and junk skipped
+    assert body["imported"] == 1
+    listing = client.get("/api/workflows").json()
+    assert len(listing) == 1
+    assert listing[0]["source_path"] == "sub/demo.json"
+    # rerun: idempotent
+    again = client.post(
+        "/api/workflows/import-scan",
+        json={"root": str(tmp_path), "dry_run": False},
+    ).json()
+    assert again["imported"] == 0
+    assert len(client.get("/api/workflows").json()) == 1
+
+
+def test_meta_view_has_metadata_but_no_full_json(client):
+    client.post(
+        "/api/workflows/import-batch",
+        json={
+            "files": [
+                {
+                    "source_path": "3-图像生成/Qwen-Image-2.1/test_t2i.json",
+                    "workflow_json": _ui_graph(),
+                    "tags": ["极速"],
+                }
+            ]
+        },
+    )
+    meta = client.get("/api/workflows", params={"view": "meta"}).json()
+    assert len(meta) == 1
+    row = meta[0]
+    assert "workflow_json" not in row
+    assert "input_schema" not in row
+    assert "output_schema" not in row
+    assert row["category"] == "image"
+    assert row["model_family"] == "Qwen"
+    assert row["media_io"] == {"in": [], "out": ["image"]}
+    assert row["tags"] == ["极速"]
+    assert row["favorite"] is False
+    assert row["needs_mapping"] is False
+
+
+def test_legacy_list_shape_and_old_records_unchanged(client):
+    """Old imports (API format via POST /api/workflows) keep working; the
+    legacy list still carries workflow_json and the new columns default."""
+    created = client.post(
+        "/api/workflows",
+        json={"name": "旧导入", "workflow_json": _api_workflow()},
+    )
+    assert created.status_code == 200
+    legacy = client.get("/api/workflows").json()
+    assert len(legacy) == 1
+    row = legacy[0]
+    assert isinstance(row["workflow_json"], dict)
+    assert "usage" in row and "output_map" in row
+    assert row["favorite"] is False
+    assert row["tags"] == []
+    assert row["media_io"] == {"in": [], "out": []}
+    meta = client.get("/api/workflows", params={"view": "meta"}).json()
+    assert meta[0]["needs_mapping"] is False or meta[0]["needs_mapping"] is True
+    assert meta[0]["source"] is None
+
+
+def test_favorite_tags_patch_and_touch(client):
+    wid = client.post(
+        "/api/workflows/import-batch",
+        json={"files": [{"name": "fav", "workflow_json": _ui_graph()}]},
+    ).json()["imported_ids"][0]
+    resp = client.patch(f"/api/workflows/{wid}", json={"favorite": True, "tags": ["高质量"]})
+    assert resp.status_code == 200
+    assert resp.json()["favorite"] is True
+    assert resp.json()["tags"] == ["高质量"]
+    touch = client.post(f"/api/workflows/{wid}/touch")
+    assert touch.status_code == 200
+    meta = client.get("/api/workflows", params={"view": "meta"}).json()
+    assert meta[0]["last_used_at"] is not None
+    assert meta[0]["favorite"] is True
+
+
+def test_migration_adds_columns_on_old_database(tmp_path):
+    """A pre-S2 database (workflows table without the metadata columns)
+    migrates in place; existing rows survive with defaults."""
+    db_path = tmp_path / "old.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(
+        """
+        CREATE TABLE workflows (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT NOT NULL,
+            kind TEXT NOT NULL CHECK(kind IN ('image', 'video')),
+            workflow_json TEXT NOT NULL,
+            description TEXT,
+            prompt_profile TEXT NOT NULL DEFAULT 'prose',
+            is_enabled INTEGER NOT NULL DEFAULT 1,
+            usage TEXT,
+            usage_source TEXT,
+            output_map TEXT,
+            output_map_source TEXT
+        );
+        INSERT INTO workflows (name, kind, workflow_json)
+            VALUES ('legacy', 'video', '{"1": {"class_type": "SaveImage", "inputs": {}}}');
+        """
+    )
+    conn.commit()
+    conn.close()
+
+    import asyncio
+
+    asyncio.run(migrate_db(db_path))
+    conn = sqlite3.connect(db_path)
+    cols = {r[1] for r in conn.execute("PRAGMA table_info(workflows)").fetchall()}
+    assert {
+        "category",
+        "model_family",
+        "media_io",
+        "tags",
+        "favorite",
+        "last_used_at",
+        "source",
+        "source_path",
+    } <= cols
+    row = conn.execute("SELECT name, favorite, tags, category FROM workflows").fetchone()
+    assert row == ("legacy", 0, None, None)
+    conn.close()