feat(workflows): manual output-node mapping for unknown custom outputs (HEL-638)

Unknown custom output nodes are never auto-typed by class-name semantics;
they stay pending until the user designates the real output node(s) and
media type. Picking a usage alone no longer enables a workflow, reanalyze/
migration/field edits never overwrite a user mapping, generation harvests
only the mapped nodes from ComfyUI history, and media-type mismatches or
missing node results fail for real instead of faking completion.

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

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

@@ -285,16 +285,46 @@ class ComfyUIClient:
         except Exception as exc:
             logger.warning("ComfyUI interrupt failed: %s", exc)
 
-    def extract_outputs(self, history: dict[str, Any]) -> list[dict[str, str]]:
+    # History items carry no explicit media type — infer from the file
+    # extension first, then the history key the node reported under.
+    _VIDEO_EXTS = frozenset({".mp4", ".webm", ".mkv", ".mov", ".gif", ".m4v", ".avi", ".flv"})
+    _IMAGE_EXTS = frozenset({".png", ".jpg", ".jpeg", ".webp", ".bmp", ".tif", ".tiff"})
+
+    @classmethod
+    def _infer_media_kind(cls, filename: str, history_key: str) -> str | None:
+        ext = Path(str(filename)).suffix.lower()
+        if ext in cls._VIDEO_EXTS:
+            return "video"
+        if ext in cls._IMAGE_EXTS:
+            return "image"
+        if history_key in ("gifs", "videos"):
+            return "video"
+        return None
+
+    def extract_outputs(
+        self, history: dict[str, Any], node_kinds: dict[str, str] | None = None
+    ) -> list[dict[str, str]]:
+        """Media items in a finished prompt's history.
+
+        ``node_kinds`` (HEL-638) scopes the harvest to the workflow's
+        confirmed output nodes — {nodeId: image|video} — so results from
+        nodes the user never designated are ignored, and each item carries
+        the node id plus its inferred media kind for mismatch checks.
+        """
         outputs: list[dict[str, str]] = []
-        for _node_id, node_out in (history.get("outputs") or {}).items():
+        for node_id, node_out in (history.get("outputs") or {}).items():
+            if node_kinds is not None and str(node_id) not in node_kinds:
+                continue
             for key in ("images", "gifs", "videos"):
                 for item in node_out.get(key) or []:
+                    filename = item.get("filename", "")
                     outputs.append(
                         {
-                            "filename": item.get("filename", ""),
+                            "filename": filename,
                             "subfolder": item.get("subfolder", ""),
                             "type": item.get("type", "output"),
+                            "node_id": str(node_id),
+                            "media_kind": self._infer_media_kind(filename, key) or "",
                         }
                     )
         return outputs

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

@@ -83,10 +83,11 @@ def workflow_warnings(workflow: dict[str, Any]) -> list[dict[str, str]]:
                 }
             )
 
-    # HEL-636: a workflow without any recognizable output node can never
+    # HEL-636/638: a workflow without any recognizable output node can never
     # return a result — block it with a readable reason instead of letting
-    # ComfyUI run blind. Unknown *possible* savers (needs_mapping) don't
-    # trigger this: the user confirms those on the import page.
+    # ComfyUI run blind. Unknown candidates (needs_mapping) don't trigger
+    # this: the user designates them on the import/connections page, and
+    # until then the row stays disabled.
     from calliope.comfyui.parser import detect_output_nodes
 
     outputs = detect_output_nodes(workflow)
@@ -110,7 +111,7 @@ def workflow_warnings(workflow: dict[str, Any]) -> list[dict[str, str]]:
                     "code": "unknown_output_node",
                     "node_id": names,
                     "message": f"存在无法识别的输出节点（{names}），"
-                    "导入后需人工确认用途；结果按 ComfyUI 历史的图片/视频输出提取",
+                    "需人工指定实际输出节点及其媒体类型（图片/视频）后才能生成",
                 }
             )
     return found

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

@@ -0,0 +1,148 @@
+"""Stored output-node mapping for imported workflows (HEL-638).
+
+Unknown custom output nodes are never auto-typed. The import confirm page
+(or the connections table) stores the user's explicit designation — which
+node ids are the real outputs and each one's media type — in
+``workflows.output_map`` with ``output_map_source='user'``. Everything that
+re-derives schemas (reanalyze, migration, plain field edits) must keep that
+decision; only a fresh explicit payload replaces it.
+
+Rows without unknown candidates carry an ``auto`` map derived from the
+trusted registry/tagged outputs so the generation path always knows which
+history nodes to harvest.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from calliope.comfyui.parser import parse_dynamic_outputs
+from calliope.comfyui.registry import (
+    IMAGE_OUTPUT_CLASSES,
+    NUMBER_CLASSES,
+    TEXT_AREA_CLASSES,
+    VIDEO_OUTPUT_CLASSES,
+)
+
+VALID_OUTPUT_KINDS = ("image", "video")
+
+
+def _wiring_class(class_type: str) -> bool:
+    """Parameter / wiring classes can never be designated as outputs."""
+    return (
+        not class_type
+        or class_type in TEXT_AREA_CLASSES
+        or class_type in NUMBER_CLASSES
+        or class_type.startswith("Empty")
+        or class_type.startswith("Primitive")
+    )
+
+
+def _entry_public(entry: dict[str, Any]) -> dict[str, Any]:
+    out = {
+        "nodeId": str(entry.get("nodeId") or ""),
+        "kind": entry.get("kind"),
+    }
+    if entry.get("label"):
+        out["label"] = entry["label"]
+    if entry.get("class_type"):
+        out["class_type"] = entry["class_type"]
+    return out
+
+
+def auto_output_map(workflow: dict[str, Any]) -> list[dict[str, Any]]:
+    """Trusted outputs only (tagged titles / registry classes)."""
+    if not isinstance(workflow, dict):
+        return []
+    return [
+        _entry_public(o)
+        for o in parse_dynamic_outputs(workflow)
+        if o.get("kind") in VALID_OUTPUT_KINDS
+    ]
+
+
+def needs_user_output_map(workflow: dict[str, Any]) -> bool:
+    """True when the graph has unknown output candidates (HEL-638)."""
+    if not isinstance(workflow, dict):
+        return False
+    return any(
+        o.get("evidence") == "unknown_node" for o in parse_dynamic_outputs(workflow)
+    )
+
+
+def validate_output_map(
+    workflow: dict[str, Any], entries: Any
+) -> list[str]:
+    """Chinese problems with a user-supplied output map; empty = valid."""
+    if not isinstance(entries, list) or not entries:
+        return ["输出节点映射不能为空：请至少指定一个输出节点及其媒体类型"]
+    seen: set[str] = set()
+    problems: list[str] = []
+    for entry in entries:
+        if not isinstance(entry, dict):
+            problems.append("输出节点映射格式有误（应为对象数组）")
+            continue
+        nid = str(entry.get("nodeId") or "")
+        node = workflow.get(nid) if isinstance(workflow, dict) else None
+        if not nid or not isinstance(node, dict):
+            problems.append(f"输出节点 {nid or '?'} 不存在于该工作流中")
+            continue
+        ct = str(node.get("class_type") or "").strip()
+        if not ct:
+            problems.append(f"节点 {nid} 缺少节点类型，不能作为输出节点")
+        elif _wiring_class(ct) and ct not in (IMAGE_OUTPUT_CLASSES | VIDEO_OUTPUT_CLASSES):
+            problems.append(f"节点 {nid}（{ct}）是参数/连线节点，不能作为输出节点")
+        if entry.get("kind") not in VALID_OUTPUT_KINDS:
+            problems.append(f"输出节点 {nid} 的媒体类型必须是图片或视频")
+        if nid in seen:
+            problems.append(f"输出节点 {nid} 重复指定")
+        seen.add(nid)
+    return problems
+
+
+def load_output_map(raw: str | None) -> list[dict[str, Any]]:
+    if not raw:
+        return []
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return []
+    return data if isinstance(data, list) else []
+
+
+def resolve_output_map(
+    output_map_raw: str | None, source: str | None, workflow: dict[str, Any]
+) -> tuple[list[dict[str, Any]], str | None]:
+    """The map generation harvests, plus whose decision it is.
+
+    A user map wins as long as its node ids still exist in the stored JSON
+    (defensive pruning — PATCH cannot edit the JSON, but keep it safe);
+    otherwise the trusted auto map applies. Empty list back means "not
+    confirmed yet".
+    """
+    if source == "user":
+        entries = [
+            e for e in load_output_map(output_map_raw) if str(e.get("nodeId") or "") in
+            (workflow if isinstance(workflow, dict) else {})
+        ]
+        if entries:
+            return entries, "user"
+        return [], None
+    auto = auto_output_map(workflow)
+    return (auto, "auto") if auto else ([], None)
+
+
+def mapping_problem(
+    workflow: dict[str, Any], output_map_raw: str | None, source: str | None
+) -> str | None:
+    """Chinese reason this workflow's outputs are not confirmed, else None."""
+    if not needs_user_output_map(workflow):
+        return None
+    entries, resolved = resolve_output_map(output_map_raw, source, workflow)
+    if resolved == "user" and entries:
+        return None
+    return (
+        "该工作流含无法识别的自定义输出节点，需在「ComfyUI 连接」页指定"
+        "实际输出节点及其媒体类型（图片/视频）后才能启用或生成"
+    )

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

@@ -12,6 +12,8 @@ from calliope.comfyui.registry import (
     IMAGE_CLASSES,
     IMAGE_OUTPUT_CLASSES,
     IMAGE_URL_CLASSES,
+    NUMBER_CLASSES,
+    TEXT_AREA_CLASSES,
     VIDEO_CLASSES,
     VIDEO_OUTPUT_CLASSES,
     ComfyOutputKind,
@@ -707,6 +709,16 @@ def parse_dynamic_outputs(workflow: dict[str, Any]) -> list[dict[str, Any]]:
 # Nodes whose output another node consumes are wiring, not results. A node
 # nothing consumes is a sink; media sinks are the workflow's outputs.
 _INPUT_ONLY_CLASSES = IMAGE_CLASSES | IMAGE_URL_CLASSES | AUDIO_CLASSES | VIDEO_CLASSES
+# Latent/source shapers (EmptyLatentImage…) are wiring even when nothing
+# consumes them — they never carry results out of the graph.
+_SOURCE_ONLY_CLASSES = _INPUT_ONLY_CLASSES | TEXT_AREA_CLASSES | NUMBER_CLASSES
+
+# Classes that are known-internal tails: standard decode steps and cache
+# cleaners consume media but never emit history outputs, so they can never
+# be output candidates (HEL-638: a graph ending in these has no output).
+_NON_OUTPUT_CLASSES = frozenset(
+    {"VAEDecode", "VAEDecodeTiled", "VAEDecodeAudio", "easy cleanGpuUsed"}
+)
 
 # Widget names save-style custom nodes carry, and link input names that mark
 # "consumes rendered media" — used to flag unknown sinks that need mapping.
@@ -743,6 +755,17 @@ def _unknown_sink_looks_like_save(node: dict[str, Any], workflow: dict[str, Any]
     return False
 
 
+def _node_has_save_widget(node: dict[str, Any]) -> bool:
+    """A concrete save widget (filename/prefix/path) — the only signal that
+    counts for unknown nodes that something else consumes; media links from
+    decode steps are meaningless there (every transform has them)."""
+    inputs = node.get("inputs") or {}
+    return any(
+        name in _SAVE_WIDGET_NAMES and isinstance(val, (str, int, float, bool))
+        for name, val in inputs.items()
+    )
+
+
 def _video_semantics(workflow: dict[str, Any]) -> bool:
     """Auxiliary evidence the graph renders video (frame counts, video nodes)."""
     for node in workflow.values():
@@ -757,52 +780,78 @@ def _video_semantics(workflow: dict[str, Any]) -> bool:
     return False
 
 
+def _suggest_output_kind(class_type: str, videoish: bool) -> str | None:
+    """Best-guess media type for an unknown candidate — UI default only."""
+    lower = class_type.lower()
+    if "video" in lower or "animatediff" in lower:
+        return "video"
+    if "image" in lower or "save" in lower:
+        return "image"
+    return "video" if videoish else None
+
+
+def _unknown_output_entry(
+    nid: str, node: dict[str, Any], workflow: dict[str, Any], videoish: bool
+) -> dict[str, Any]:
+    ct = str(node.get("class_type") or "")
+    return {
+        "nodeId": nid,
+        "label": _title_of(node) or ct,
+        "role": None,
+        # HEL-638: unknown custom nodes are never auto-typed — the user
+        # picks the media type; `suggested_kind` is only a form default.
+        "kind": None,
+        "class_type": ct,
+        "evidence": "unknown_node",
+        "suggested_kind": _suggest_output_kind(ct, videoish),
+    }
+
+
 def detect_output_nodes(workflow: dict[str, Any]) -> list[dict[str, Any]]:
     """Infer output nodes of an untagged workflow from its sinks.
 
     Exact registry classes (SaveImage, VHS_VideoCombine…) count whether or
     not something downstream consumes them (a SaveVideo feeding a
-    cache-cleanup tail is still the output). Fuzzy class-name matches only
-    count at a sink. Unknown sink classes that look like savers become
-    ``other`` + ``unknown_node`` evidence — the import page then asks the
-    user to map them.
+    cache-cleanup tail is still the output). Every OTHER class at an output
+    position is an *unknown candidate*: kind stays null, the import page
+    asks the user to designate the real output node(s) and their media
+    type, and the workflow stays pending until then (HEL-638). Sinks with
+    no save features only become candidates when the graph has no trusted
+    output at all — next to a real saver they are utility tails.
     """
     if not isinstance(workflow, dict):
         return []
     consumers = _consumers(workflow)
     videoish = _video_semantics(workflow)
-    results: list[dict[str, Any]] = []
+    registry_outputs: list[dict[str, Any]] = []
+    candidates: list[dict[str, Any]] = []
+    seen: set[str] = set()
+
+    def _is_wiring(ct: str) -> bool:
+        return (
+            not ct
+            or ct in _SOURCE_ONLY_CLASSES
+            or ct in _NON_OUTPUT_CLASSES
+            or ct.startswith("Empty")
+        )
+
+    # 1) Trusted registry outputs at sink positions.
     for nid, node in _sinks(workflow, consumers):
         ct = str(node.get("class_type") or "")
-        if not ct or ct in _INPUT_ONLY_CLASSES:
-            continue
-        title = _title_of(node)
         if ct in IMAGE_OUTPUT_CLASSES or ct in VIDEO_OUTPUT_CLASSES:
-            kind: ComfyOutputKind = "video" if ct in VIDEO_OUTPUT_CLASSES else "image"
-            evidence = "registry"
-        else:
-            kind = class_to_output_kind(ct)
-            if kind in ("image", "video"):
-                evidence = "class_name_match"
-            elif _unknown_sink_looks_like_save(node, workflow):
-                # A custom saver we cannot type: with video evidence in the
-                # graph it most likely saves video, but the user decides.
-                kind = "video" if videoish else "other"
-                evidence = "unknown_node"
-            else:
-                continue  # utility/primitive sink — not an output
-        results.append(
-            {
-                "nodeId": nid,
-                "label": title or ct,
-                "role": None,
-                "kind": kind,
-                "class_type": ct,
-                "evidence": evidence,
-            }
-        )
-    # Registry output classes consumed by other nodes are outputs too.
-    seen = {r["nodeId"] for r in results}
+            registry_outputs.append(
+                {
+                    "nodeId": nid,
+                    "label": _title_of(node) or ct,
+                    "role": None,
+                    "kind": "video" if ct in VIDEO_OUTPUT_CLASSES else "image",
+                    "class_type": ct,
+                    "evidence": "registry",
+                }
+            )
+            seen.add(nid)
+
+    # 2) Registry output classes consumed by other nodes are outputs too.
     for nid, node in sorted(
         ((str(k), v) for k, v in workflow.items() if isinstance(v, dict)),
         key=lambda kv: _node_sort_key(kv[0]),
@@ -810,7 +859,7 @@ def detect_output_nodes(workflow: dict[str, Any]) -> list[dict[str, Any]]:
         ct = str(node.get("class_type") or "")
         if nid in seen or ct not in (IMAGE_OUTPUT_CLASSES | VIDEO_OUTPUT_CLASSES):
             continue
-        results.append(
+        registry_outputs.append(
             {
                 "nodeId": nid,
                 "label": _title_of(node) or ct,
@@ -820,6 +869,37 @@ def detect_output_nodes(workflow: dict[str, Any]) -> list[dict[str, Any]]:
                 "evidence": "registry",
             }
         )
+        seen.add(nid)
+
+    # 3) Unknown sink candidates. Save-looking sinks always need mapping;
+    # feature-less sinks only when nothing trusted exists (utility tails
+    # next to a real saver stay invisible).
+    has_trusted = bool(registry_outputs)
+    for nid, node in _sinks(workflow, consumers):
+        ct = str(node.get("class_type") or "")
+        if nid in seen or _is_wiring(ct) or ct in (IMAGE_OUTPUT_CLASSES | VIDEO_OUTPUT_CLASSES):
+            continue
+        if _unknown_sink_looks_like_save(node, workflow) or not has_trusted:
+            candidates.append(_unknown_output_entry(nid, node, workflow, videoish))
+            seen.add(nid)
+
+    # 4) Unknown save-looking nodes that something consumes (a custom saver
+    # feeding a cleanup tail is still an output candidate). Only concrete
+    # save widgets count here — decode-fed media links are ubiquitous among
+    # mid-graph transforms (CreateVideo & co.) and never mark an output.
+    for nid, node in sorted(
+        ((str(k), v) for k, v in workflow.items() if isinstance(v, dict)),
+        key=lambda kv: _node_sort_key(kv[0]),
+    ):
+        ct = str(node.get("class_type") or "")
+        if nid in seen or _is_wiring(ct) or ct in (IMAGE_OUTPUT_CLASSES | VIDEO_OUTPUT_CLASSES):
+            continue
+        if _node_has_save_widget(node):
+            candidates.append(_unknown_output_entry(nid, node, workflow, videoish))
+            seen.add(nid)
+
+    results = registry_outputs + candidates
+    results.sort(key=lambda r: _node_sort_key(r["nodeId"]))
     return results
 
 
@@ -899,7 +979,7 @@ def classify_usage(workflow: dict[str, Any]) -> dict[str, Any]:
         names = "、".join(
             f"节点 {o['nodeId']}（{o.get('class_type') or '?'}）" for o in unknown_outputs[:4]
         )
-        summary.append(f"存在无法识别的输出节点：{names}，需人工确认")
+        summary.append(f"存在无法识别的输出节点：{names}，需人工指定输出节点及媒体类型")
     for s in image_slots:
         if s["role"] != "mask":
             summary.append(f"图片输入：节点 {s['nodeId']}（{s['label']}）")
@@ -914,6 +994,18 @@ def classify_usage(workflow: dict[str, Any]) -> dict[str, Any]:
             "未在图中找到输出节点（SaveImage / 视频合成等），"
             "请确认导出的是 API 格式且包含保存节点"
         )
+    elif unknown_outputs:
+        # HEL-638: any unknown custom output node keeps the workflow
+        # pending — no name-based guessing, and picking a usage alone must
+        # not bypass the output-node designation.
+        names = "、".join(
+            f"节点 {o['nodeId']}（{o.get('class_type') or '?'}）" for o in unknown_outputs[:4]
+        )
+        ambiguous = True
+        problem = (
+            f"存在无法识别的输出节点（{names}），"
+            "需人工指定实际输出节点及其媒体类型（图片/视频）后才能保存启用"
+        )
     elif len(out_kinds) > 1:
         ambiguous = True
         problem = "同一工作流同时输出图片和视频，需人工选择主用途"
@@ -929,10 +1021,6 @@ def classify_usage(workflow: dict[str, Any]) -> dict[str, Any]:
         else:
             ambiguous = True
             problem = "视频工作流中的单张图片输入无法判定是首帧还是参考图，需人工确认"
-    else:
-        # only unknown outputs — the user must map them before it is usable
-        ambiguous = True
-        problem = "输出节点的媒体类型无法识别，需人工确认用途后才能生成"
 
     return {
         "usage": usage,

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

@@ -30,7 +30,8 @@ VIDEO_OUTPUT_CLASSES = frozenset(
     {"VHS_VideoCombine", "SaveVideo", "VideoOutput", "AnimateDiffCombine"}
 )
 IMAGE_OUTPUT_CLASSES = frozenset(
-    {"SaveImage", "PreviewImage", "SaveImageWebsocket", "ETN_SendImageWebSocket"}
+    {"SaveImage", "PreviewImage", "SaveImageWebsocket", "ETN_SendImageWebSocket",
+     "SaveImageAdvanced"}
 )
 
 

calliope-backend/src/calliope/db.py

@@ -121,7 +121,11 @@ CREATE TABLE IF NOT EXISTS workflows (
     description TEXT,
     prompt_profile TEXT NOT NULL DEFAULT 'prose',
     is_enabled INTEGER NOT NULL DEFAULT 1,
-    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    usage TEXT,
+    usage_source TEXT,
+    output_map TEXT,
+    output_map_source TEXT
 );
 
 CREATE TABLE IF NOT EXISTS jobs (
@@ -370,6 +374,42 @@ def _backfill_workflow_usage(conn: sqlite3.Connection) -> None:
             )
 
 
+def _backfill_workflow_output_map(conn: sqlite3.Connection) -> None:
+    """HEL-638: settle the output-node mapping for existing rows.
+
+    Rows the user already mapped (output_map_source='user') are never
+    touched. Determinate graphs get the trusted ``auto`` map. Rows whose
+    graph now contains unknown custom output candidates get no auto map and
+    are parked disabled — the usage decision survives, but until the user
+    designates the output nodes the workflow must not generate.
+    """
+    from calliope.comfyui.output_map import auto_output_map, needs_user_output_map
+
+    rows = conn.execute(
+        "SELECT id, workflow_json FROM workflows"
+        " WHERE output_map_source IS NULL OR output_map_source != 'user'"
+    ).fetchall()
+    for row in rows:
+        try:
+            workflow = json.loads(row[1])
+        except (json.JSONDecodeError, TypeError):
+            workflow = None
+        if not isinstance(workflow, dict):
+            continue
+        if needs_user_output_map(workflow):
+            conn.execute(
+                "UPDATE workflows SET output_map = NULL, output_map_source = NULL,"
+                " is_enabled = 0 WHERE id = ?",
+                (row[0],),
+            )
+        else:
+            conn.execute(
+                "UPDATE workflows SET output_map = ?, output_map_source = 'auto'"
+                " WHERE id = ?",
+                (json.dumps(auto_output_map(workflow)), row[0]),
+            )
+
+
 async def migrate_db(db_path: Path) -> None:
     db_path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(db_path)
@@ -421,9 +461,17 @@ async def migrate_db(db_path: Path) -> None:
         conn.execute("ALTER TABLE workflows ADD COLUMN usage TEXT")
     if "usage_source" not in cols:
         conn.execute("ALTER TABLE workflows ADD COLUMN usage_source TEXT")
+    # HEL-638: explicit output-node designation for unknown custom output
+    # nodes ({nodeId, kind} JSON + 'user'|'auto'). Idempotent backfill
+    # below parks unmapped unknown-output rows (disabled) like usage does.
+    if "output_map" not in cols:
+        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")
     # Idempotent backfill: pending rows are re-judged on every boot (parser
     # upgrades can settle them); confirmed rows are never touched.
     _backfill_workflow_usage(conn)
+    _backfill_workflow_output_map(conn)
     scene_cols = {r[1] for r in conn.execute("PRAGMA table_info(scenes)").fetchall()}
     if "location_id" not in scene_cols:
         conn.execute("ALTER TABLE scenes ADD COLUMN location_id INTEGER")

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

@@ -222,6 +222,11 @@ class WorkflowCreate(BaseModel):
     usage_source: Literal["auto", "user"] | None = None
     # Schema corrected on the confirm page (manual role overrides included).
     input_schema: list[dict[str, Any]] | None = None
+    # HEL-638: explicit output-node designation for unknown custom output
+    # nodes — [{nodeId, kind: image|video}]. Required before a workflow
+    # with unknown outputs can be enabled or generated.
+    output_map: list[dict[str, Any]] | None = None
+    output_map_source: Literal["auto", "user"] | None = None
 
 
 class WorkflowUpdate(BaseModel):
@@ -234,6 +239,10 @@ class WorkflowUpdate(BaseModel):
     usage: WorkflowUsageKind | None = None
     usage_source: Literal["auto", "user"] | None = None
     input_schema: list[dict[str, Any]] | None = None
+    # HEL-638: set/replace the user output mapping; an explicit empty list
+    # resets the workflow to pending confirmation.
+    output_map: list[dict[str, Any]] | None = None
+    output_map_source: Literal["auto", "user"] | None = None
 
 
 class WorkflowAnalyze(BaseModel):

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

@@ -196,15 +196,25 @@ class QueueWorker:
                 )
 
             workflow_id = job.get("workflow_id") or payload.get("workflow_id")
-            workflow = self._load_workflow(workflow_id)
-            if not workflow:
+            workflow_row = self._load_workflow_row(workflow_id)
+            if not workflow_row:
                 raise RuntimeError(
                     "找不到该任务使用的工作流（可能已被删除），请重新选择工作流后再生成"
                 )
+            workflow = workflow_row["workflow_json"]
             broken = blocking_problem(workflow)
             if broken:
                 raise RuntimeError(broken)
 
+            # HEL-638: harvest only the confirmed output nodes. A workflow
+            # with unknown custom outputs and no user mapping must not run.
+            node_kinds = self._output_node_kinds(workflow_row)
+            if node_kinds is None:
+                raise RuntimeError(
+                    "该工作流含无法识别的自定义输出节点，尚未指定输出节点及媒体类型，"
+                    "请到「ComfyUI 连接」页确认后再生成"
+                )
+
             input_values = payload.get("input_values") or {}
             if payload.get("continue_source") and kind == "video":
                 input_values = await self._resolve_continue_source(
@@ -240,13 +250,31 @@ class QueueWorker:
                     f"ComfyUI 执行出错（任务 ID {prompt_id}）：{_comfy_messages(messages)}"
                 )
 
-            outputs_meta = client.extract_outputs(history)
+            outputs_meta = client.extract_outputs(history, node_kinds=node_kinds)
             saved = [m for m in outputs_meta if m.get("type") == "output"]
             outputs_meta = saved or outputs_meta
             if not outputs_meta:
+                checked = "、".join(sorted(node_kinds))
                 raise RuntimeError(
                     f"ComfyUI 显示任务已结束（任务 ID {prompt_id}），但没有返回任何图片或视频，"
-                    "不算生成完成。请确认工作流带有保存/预览节点，并到 ComfyUI 历史中核对该任务。"
+                    "不算生成完成。已按输出映射检查节点：" f"{checked}；"
+                    "请到 ComfyUI 历史中核对该任务，并确认所指定的输出节点正确。"
+                )
+            mismatch = next(
+                (
+                    m
+                    for m in outputs_meta
+                    if m.get("media_kind") and m["media_kind"] != node_kinds.get(m["node_id"])
+                ),
+                None,
+            )
+            if mismatch:
+                want = node_kinds.get(mismatch["node_id"])
+                raise RuntimeError(
+                    f"输出节点 {mismatch['node_id']} 的媒体类型与映射不符"
+                    f"（映射为{'视频' if want == 'video' else '图片'}，"
+                    f"实际返回 {mismatch['filename']}），任务不算生成完成。"
+                    "请到「ComfyUI 连接」页修正该节点的媒体类型。"
                 )
             queue_manager.set_phase(job["id"], "saving")
             dest_dir = config.settings.assets_dir / str(project_id) / kind
@@ -476,20 +504,45 @@ class QueueWorker:
         finally:
             conn.close()
 
-    def _load_workflow(self, workflow_id: int | None) -> dict[str, Any] | None:
+    def _load_workflow_row(self, workflow_id: int | None) -> dict[str, Any] | None:
         if not workflow_id:
             return None
         conn = get_db(config.settings.db_path)
         try:
             row = conn.execute(
-                "SELECT workflow_json FROM workflows WHERE id = ?", (workflow_id,)
+                "SELECT workflow_json, output_map, output_map_source FROM workflows"
+                " WHERE id = ?",
+                (workflow_id,),
             ).fetchone()
             if not row:
                 return None
-            return json.loads(row["workflow_json"])
+            return {
+                "workflow_json": json.loads(row["workflow_json"]),
+                "output_map": row["output_map"],
+                "output_map_source": row["output_map_source"],
+            }
         finally:
             conn.close()
 
+    @staticmethod
+    def _output_node_kinds(workflow_row: dict[str, Any]) -> dict[str, str] | None:
+        """Confirmed {nodeId: image|video} for this workflow, or None when
+        the outputs are not confirmed (unknown nodes, no user mapping)."""
+        from calliope.comfyui.output_map import mapping_problem, resolve_output_map
+
+        workflow = workflow_row["workflow_json"]
+        problem = mapping_problem(
+            workflow, workflow_row.get("output_map"), workflow_row.get("output_map_source")
+        )
+        if problem:
+            return None
+        entries, _source = resolve_output_map(
+            workflow_row.get("output_map"),
+            workflow_row.get("output_map_source"),
+            workflow,
+        )
+        return {str(e["nodeId"]): str(e["kind"]) for e in entries} or None
+
     def _job_label(self, job: dict[str, Any]) -> str:
         payload = json.loads(job.get("payload_json") or "{}")
         kind = job.get("kind") or "job"

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

@@ -14,6 +14,7 @@ from pydantic import BaseModel, Field
 from calliope import config
 from calliope.agent.prompts import item_reference_prompt
 from calliope.comfyui.lint import blocking_problem
+from calliope.comfyui.output_map import mapping_problem
 from calliope.comfyui.parser import effective_input_schema
 from calliope.comfyui.validate import validate_input_values
 from calliope.db import get_db, row_to_dict
@@ -335,7 +336,8 @@ async def generate(payload: PlaygroundGenerate) -> dict[str, Any]:
     try:
         pid = ensure_playground_project(conn)
         wf = conn.execute(
-            "SELECT id, kind, is_enabled, workflow_json, input_schema FROM workflows WHERE id = ?",
+            "SELECT id, kind, is_enabled, workflow_json, input_schema, output_map,"
+            " output_map_source FROM workflows WHERE id = ?",
             (payload.workflow_id,),
         ).fetchone()
         if not wf:
@@ -348,6 +350,11 @@ async def generate(payload: PlaygroundGenerate) -> dict[str, Any]:
             )
         kind = wf["kind"]
         workflow_json = json.loads(wf["workflow_json"])
+        # HEL-638: a workflow with unknown custom output nodes must have a
+        # confirmed output mapping before anything is queued.
+        map_problem = mapping_problem(workflow_json, wf["output_map"], wf["output_map_source"])
+        if map_problem:
+            raise HTTPException(status_code=400, detail=map_problem)
         schema = effective_input_schema(
             json.loads(wf["input_schema"]) if wf["input_schema"] else [], workflow_json
         )

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

@@ -6,6 +6,13 @@ from typing import Any
 from fastapi import APIRouter, HTTPException
 
 from calliope.comfyui.lint import workflow_warnings
+from calliope.comfyui.output_map import (
+    auto_output_map,
+    load_output_map,
+    mapping_problem,
+    needs_user_output_map,
+    validate_output_map,
+)
 from calliope.comfyui.parser import (
     classify_usage,
     effective_input_schema,
@@ -26,9 +33,17 @@ def _serialize_workflow(row: Any) -> dict[str, Any]:
     data["workflow_json"] = json.loads(data["workflow_json"])
     stored = json.loads(data["input_schema"]) if data.get("input_schema") else []
     data["input_schema"] = effective_input_schema(stored, data["workflow_json"])
-    data["output_schema"] = json.loads(data["output_schema"]) if data.get("output_schema") else []
+    # Outputs are pure graph derivation (no manual overrides) — always fresh,
+    # so parser upgrades reach old rows without a re-import.
+    data["output_schema"] = parse_dynamic_outputs(data["workflow_json"])
     data["is_enabled"] = bool(data.get("is_enabled"))
     data["warnings"] = workflow_warnings(data["workflow_json"])
+    raw_map = data.get("output_map")
+    data["output_map"] = load_output_map(raw_map)
+    # Pending flag: unknown output candidates the user has not mapped yet.
+    data["needs_mapping"] = (
+        mapping_problem(data["workflow_json"], raw_map, data.get("output_map_source")) is not None
+    )
     return data
 
 
@@ -63,9 +78,9 @@ async def reanalyze_workflow(workflow_id: int) -> dict[str, Any]:
     """Recompute usage and input/output schemas from the stored workflow JSON.
 
     Workflows imported before a parser upgrade keep stale (often empty)
-    schemas — this refreshes them without re-importing. A usage the user
-    set by hand is never overwritten; only ``auto`` (and pending NULL)
-    rows get the new judgment.
+    schemas — this refreshes them without re-importing. A usage or output
+    mapping the user set by hand is never overwritten; only ``auto`` (and
+    pending NULL) rows get the new judgment.
     """
     conn = get_db(settings.db_path)
     try:
@@ -83,10 +98,31 @@ async def reanalyze_workflow(workflow_id: int) -> dict[str, Any]:
         stored = json.loads(data["input_schema"]) if data["input_schema"] else []
         schema = effective_input_schema(stored, workflow)
         outputs = parse_dynamic_outputs(workflow)
+        if data["output_map_source"] == "user":
+            output_map, output_map_source = data["output_map"], "user"
+        elif needs_user_output_map(workflow):
+            # Unknown output candidates: no auto map, and the row must not
+            # generate until the user designates the output nodes.
+            output_map, output_map_source = None, None
+        else:
+            output_map, output_map_source = json.dumps(auto_output_map(workflow)), "auto"
+        is_enabled = data["is_enabled"]
+        if output_map_source is None or usage is None:
+            is_enabled = 0
         conn.execute(
             "UPDATE workflows SET input_schema = ?, output_schema = ?,"
-            " usage = ?, usage_source = ? WHERE id = ?",
-            (json.dumps(schema), json.dumps(outputs), usage, usage_source, workflow_id),
+            " usage = ?, usage_source = ?, output_map = ?, output_map_source = ?,"
+            " is_enabled = ? WHERE id = ?",
+            (
+                json.dumps(schema),
+                json.dumps(outputs),
+                usage,
+                usage_source,
+                output_map,
+                output_map_source,
+                is_enabled,
+                workflow_id,
+            ),
         )
         conn.commit()
         fresh = conn.execute(
@@ -158,6 +194,20 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
         }
         kind = "video" if out_kinds == {"video"} else "image"
     profile = payload.prompt_profile or detect_prompt_profile(payload.workflow_json)
+    # HEL-638: an explicit user output map wins; otherwise the trusted auto
+    # map. Either way a graph with unknown output candidates is unusable
+    # until the user designates the output nodes.
+    if payload.output_map:
+        problems = validate_output_map(payload.workflow_json, payload.output_map)
+        if problems:
+            raise HTTPException(status_code=422, detail="；".join(problems))
+        output_map = json.dumps(payload.output_map)
+        output_map_source = payload.output_map_source or "user"
+    else:
+        auto_map = auto_output_map(payload.workflow_json)
+        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(payload.workflow_json) or bool(payload.output_map)
     conn = get_db(settings.db_path)
     try:
         # Byte-identical re-imports are the main way the library fills up with
@@ -174,16 +224,18 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
                     f" “{existing['name']}”)."
                 ),
             )
-        # Unconfirmed workflows (ambiguous / unknown outputs) are stored but
-        # disabled: they must not reach the generate pages until the user
-        # confirms a usage on the connections page (PATCH sets it back on).
-        is_enabled = 1 if usage else 0
+        # Unconfirmed workflows (ambiguous usage / unknown outputs without a
+        # user mapping) are stored but disabled: they must not reach the
+        # generate pages until the user confirms usage AND, for unknown
+        # output nodes, designates the outputs on the connections page.
+        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)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                                   endpoint_id, is_enabled, usage, usage_source,
+                                   output_map, output_map_source)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 payload.name,
@@ -197,6 +249,8 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
                 is_enabled,
                 usage,
                 usage_source,
+                output_map,
+                output_map_source,
             ),
         )
         conn.commit()
@@ -228,20 +282,54 @@ 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
+        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
+        # resets the row to pending (disabled).
+        if "output_map" in data:
+            if data["output_map"]:
+                problems = validate_output_map(workflow_json, data["output_map"])
+                if problems:
+                    raise HTTPException(status_code=422, detail="；".join(problems))
+                data["output_map"] = json.dumps(data["output_map"])
+                data.setdefault("output_map_source", "user")
+            else:
+                data["output_map"] = None
+                data["output_map_source"] = None
+                data["is_enabled"] = 0
         if data.get("usage"):
             # Confirming / correcting the usage is an explicit user decision:
-            # it re-enables the workflow and keeps kind (output media) in sync.
+            # kind (output media) follows the usage. Whether it re-enables is
+            # decided by the gate below — with unknown output nodes the user
+            # must also designate the output nodes (HEL-638).
             data.setdefault("usage_source", "user")
             data["kind"] = usage_to_kind(data["usage"]) or existing["kind"]
-            if "is_enabled" not in data:
-                data["is_enabled"] = 1
         if "input_schema" in data and data["input_schema"] is not None:
             if not all(isinstance(e, dict) for e in data["input_schema"]):
                 raise HTTPException(status_code=422, detail="input_schema 必须是参数对象数组")
             data["input_schema"] = json.dumps(data["input_schema"])
-        # endpoint_id: explicit null clears the binding; other Nones are skips.
-        explicit_null = {"endpoint_id"} & {k for k, v in data.items() if v is None}
+        # endpoint_id: explicit null clears the binding; other Nones are skips
+        # (an output_map reset keeps its explicit NULLs).
+        explicit_null = {"endpoint_id", "output_map", "output_map_source"} & {
+            k for k, v in data.items() if v is None
+        }
         data = {k: v for k, v in data.items() if v is not None or k in explicit_null}
+
+        # Enable gate: usage must be settled and, when the graph carries
+        # unknown custom output nodes, a user output mapping must exist.
+        usage_ok = bool(data.get("usage") or existing["usage"])
+        map_raw = data.get("output_map", existing["output_map"])
+        map_source = data.get("output_map_source", existing["output_map_source"])
+        map_problem = mapping_problem(workflow_json, map_raw, map_source)
+        if data.get("is_enabled"):
+            if not usage_ok:
+                raise HTTPException(status_code=400, detail="请先选择工作流用途，再启用该工作流")
+            if map_problem:
+                raise HTTPException(status_code=400, detail=map_problem)
+        elif (data.get("usage") or data.get("output_map")) and "is_enabled" not in data:
+            # A usage-confirm or output-map-confirm path: enabled only when
+            # both decisions are settled.
+            data["is_enabled"] = 1 if (usage_ok and not map_problem) else 0
         if data:
             fields = ", ".join(f"{k} = :{k}" for k in data)
             data["id"] = workflow_id

calliope-backend/tests/conftest.py

@@ -10,6 +10,10 @@ from fastapi.testclient import TestClient
 from calliope.db import migrate_db
 from calliope.main import create_app
 
+# Shared fake-ComfyUI HTTP server (defined in test_studio_generation,
+# reused by the generation / status / unknown-output-mapping suites).
+from tests.test_studio_generation import fake_comfy  # noqa: F401,E402
+
 
 @pytest.fixture
 def client(monkeypatch):

calliope-backend/tests/test_studio_generation.py

@@ -25,9 +25,19 @@ PNG_1PX = bytes.fromhex(
 
 
 class _FakeComfyHandler(BaseHTTPRequestHandler):
-    """Minimal ComfyUI: /system_stats, /prompt, /history/{id}, /view, /upload/image."""
+    """Minimal ComfyUI: /system_stats, /prompt, /history/{id}, /view, /upload/image.
+
+    History outputs are keyed by the prompt's real output node id (what
+    ComfyUI does); a known output class that is absent falls back to
+    ``server.forced_output_node`` (default "9"). Scripted tests can pin
+    ``server.forced_output_node`` or drop entries via ``server.drop_outputs``.
+    """
 
     server_version = "FakeComfy/1.0"
+    KNOWN_OUTPUT_CLASSES = {
+        "SaveImage", "PreviewImage", "SaveImageAdvanced", "SaveImageWebsocket",
+        "SaveVideo", "VideoOutput", "VHS_VideoCombine", "AnimateDiffCombine",
+    }
 
     def do_GET(self):
         self.server.requests.append(self.path)
@@ -64,11 +74,13 @@ class _FakeComfyHandler(BaseHTTPRequestHandler):
             prompt_id = uuid.uuid4().hex
             outputs_key = self.server.outputs_key
             out_entry = {
-                outputs_key: [{"filename": self.server.out_name, "subfolder": "", "type": "output"}]
+                outputs_key: [
+                    {"filename": self.server.out_name, "subfolder": "", "type": "output"}
+                ]
             }
             self.server.history[prompt_id] = {
                 "status": {"status_str": "success", "completed": True},
-                "outputs": {"9": out_entry},
+                "outputs": {self._output_node(body): out_entry},
                 "prompt": [],
             }
             self.server.prompts.append(json.loads(body.decode("utf-8") or "{}"))
@@ -80,6 +92,19 @@ class _FakeComfyHandler(BaseHTTPRequestHandler):
             return
         self._json(404, {"error": "not found"})
 
+    def _output_node(self, body: bytes) -> str:
+        forced = getattr(self.server, "forced_output_node", None)
+        if forced:
+            return forced
+        try:
+            prompt = json.loads(body.decode("utf-8") or "{}").get("prompt") or {}
+        except json.JSONDecodeError:
+            prompt = {}
+        for nid, node in prompt.items():
+            if isinstance(node, dict) and node.get("class_type") in self.KNOWN_OUTPUT_CLASSES:
+                return str(nid)
+        return "9"
+
     def _json(self, status: int, payload: dict) -> None:
         data = json.dumps(payload).encode("utf-8")
         self.send_response(status)
@@ -109,6 +134,7 @@ def fake_comfy():
     server.fail_prompt = False
     server.out_name = "out-1.png"
     server.outputs_key = "images"
+    server.forced_output_node = None
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     try:

calliope-backend/tests/test_unknown_output_mapping.py

@@ -0,0 +1,371 @@
+"""HEL-638: unknown custom output nodes close the loop — no name-based
+auto-typing, feature-less sinks become mappable candidates, a usage choice
+alone never bypasses the output designation, and generation harvests (and
+checks) exactly the nodes the user mapped.
+
+Two brand-new class names drive the cases:
+
+* ``NebulaGlaxVideoVault`` — screams "video" in its name and carries a
+  filename_prefix; before HEL-638 the fuzzy name match auto-typed it video
+  and enabled the workflow.
+* ``QuarkMuffinDisplay`` — no save widget, no media link name; before
+  HEL-638 such a graph only reported ``no_output`` with no way to map.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from calliope.config import settings
+from calliope.db import get_db, migrate_db
+from tests.test_studio_generation import _real_run, _wait_done
+
+FIXTURES = Path(__file__).parent / "fixtures" / "comfy_native"
+
+
+def _vault_workflow() -> dict:
+    return {
+        "1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
+        "7": {
+            "class_type": "GlaxSamplerNode",
+            "inputs": {"prompt": ["1", 0], "length": 121},
+        },
+        "12": {
+            "class_type": "NebulaGlaxVideoVault",
+            "inputs": {"filename_prefix": "anim", "frames": ["7", 0]},
+        },
+    }
+
+
+def _muffin_workflow() -> dict:
+    return {
+        "1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "p"}},
+        "7": {
+            "class_type": "GlaxSamplerNode",
+            "inputs": {"prompt": ["1", 0], "length": 121},
+        },
+        "15": {"class_type": "QuarkMuffinDisplay", "inputs": {"payload": ["7", 0]}},
+    }
+
+
+def _analyze(client, wf: dict) -> dict:
+    r = client.post("/api/workflows/analyze", json={"workflow_json": wf})
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+# ---------------------------------------------------------------------------
+# 1. Unknown outputs stay pending — names never decide (gap 1)
+# ---------------------------------------------------------------------------
+
+
+def test_video_semantic_unknown_node_is_never_auto_typed(client):
+    body = _analyze(client, _vault_workflow())
+    entry = next(o for o in body["usage_evidence"]["outputs"] if o["nodeId"] == "12")
+    assert entry["evidence"] == "unknown_node"
+    assert entry["kind"] is None  # the user picks the media type
+    assert entry["suggested_kind"] == "video"  # only a form default
+    assert body["needs_mapping"] is True
+    assert body["usage"] is None and body["usage_ambiguous"] is True
+    assert "指定实际输出节点" in body["usage_problem"]
+    codes = {(w["level"], w["code"]) for w in body["warnings"]}
+    assert ("warning", "unknown_output_node") in codes
+    assert ("error", "no_output") not in codes
+
+
+def test_choosing_a_usage_alone_does_not_bypass_output_mapping(client):
+    """create/PATCH with a usage but no output map keeps the row disabled."""
+    r = client.post(
+        "/api/workflows",
+        json={"name": "未知·视频语义", "workflow_json": _vault_workflow(),
+              "usage": "text2video", "usage_source": "user"},
+    )
+    assert r.status_code == 200, r.text
+    wf = r.json()
+    assert wf["usage"] == "text2video" and wf["usage_source"] == "user"
+    assert wf["is_enabled"] is False  # usage set, outputs still unconfirmed
+    assert wf["needs_mapping"] is True and not wf["output_map"]
+
+    gen = client.post(
+        "/api/playground/generate",
+        json={"workflow_id": wf["id"], "input_values": {"1": "x"}},
+    )
+    assert gen.status_code == 400
+
+    # forcing the switch on is refused with the mapping reason
+    r = client.patch(f"/api/workflows/{wf['id']}", json={"is_enabled": True})
+    assert r.status_code == 400
+    assert "指定" in r.json()["detail"] and "输出节点" in r.json()["detail"]
+
+
+# ---------------------------------------------------------------------------
+# 2. Feature-less unknown sinks become mappable candidates (gap 2)
+# ---------------------------------------------------------------------------
+
+
+def test_featureless_unknown_node_is_a_candidate_not_a_hard_error(client):
+    body = _analyze(client, _muffin_workflow())
+    entry = next(o for o in body["usage_evidence"]["outputs"] if o["nodeId"] == "15")
+    assert entry["evidence"] == "unknown_node" and entry["kind"] is None
+    assert body["needs_mapping"] is True
+    codes = {(w["level"], w["code"]) for w in body["warnings"]}
+    assert ("warning", "unknown_output_node") in codes
+    assert ("error", "no_output") not in codes
+
+
+def test_truly_outputless_graph_still_reports_the_chinese_error(client):
+    wf = {
+        "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "p"}},
+        "2": {"class_type": "VAEDecode", "inputs": {"samples": ["1", 0]}},
+    }
+    body = _analyze(client, wf)
+    assert body["usage"] is None and body["needs_mapping"] is False
+    assert "未在图中找到输出节点" in body["usage_problem"]
+
+
+# ---------------------------------------------------------------------------
+# 3. Manual mapping → enable → scoped generation (fake ComfyUI round trip)
+# ---------------------------------------------------------------------------
+
+
+def test_manual_mapping_enables_and_extracts_from_the_mapped_node(client, fake_comfy):
+    fake_comfy.outputs_key = "gifs"
+    fake_comfy.out_name = "clip.mp4"
+    fake_comfy.forced_output_node = "12"  # the mapped node reports results
+    with _real_run(client, fake_comfy):
+        r = client.post(
+            "/api/workflows",
+            json={
+                "name": "映射·视频库",
+                "workflow_json": _vault_workflow(),
+                "usage": "text2video",
+                "usage_source": "user",
+                "output_map": [{"nodeId": "12", "kind": "video"}],
+                "output_map_source": "user",
+            },
+        )
+        assert r.status_code == 200, r.text
+        wf = r.json()
+        assert wf["is_enabled"] is True
+        assert wf["output_map"] == [{"nodeId": "12", "kind": "video"}]
+        assert wf["output_map_source"] == "user"
+
+        r = client.post(
+            "/api/playground/generate",
+            json={"workflow_id": wf["id"], "input_values": {"1": "星空延时"},
+                  "source": "studio-video"},
+        )
+        assert r.status_code == 200, r.text
+        job = _wait_done(client, r.json()["job"]["id"])
+        assert job["status"] == "done", job.get("error")
+        assert job["output_paths"][0].endswith(".mp4")
+
+
+def test_mapped_node_returning_nothing_fails_for_real(client, fake_comfy):
+    """Results that only land on an unmapped node are NOT a completion."""
+    fake_comfy.forced_output_node = "7"  # something else reports results
+    with _real_run(client, fake_comfy):
+        r = client.post(
+            "/api/workflows",
+            json={
+                "name": "映射·空手而归",
+                "workflow_json": _muffin_workflow(),
+                "usage": "text2image",
+                "usage_source": "user",
+                "output_map": [{"nodeId": "15", "kind": "image"}],
+            },
+        )
+        assert r.status_code == 200 and r.json()["is_enabled"] is True
+        r = client.post(
+            "/api/playground/generate",
+            json={"workflow_id": r.json()["id"], "input_values": {"1": "x"}},
+        )
+        assert r.status_code == 200, r.text
+        job = _wait_done(client, r.json()["job"]["id"])
+        assert job["status"] == "failed"
+        assert "没有返回任何图片或视频" in job["error"]
+        assert "15" in job["error"]  # names the node it waited on
+        assert job["output_paths"] == []
+
+
+def test_media_type_mismatch_fails_for_real(client, fake_comfy):
+    """A video mapping that only yields a png is a real failure."""
+    fake_comfy.outputs_key = "images"
+    fake_comfy.out_name = "still.png"
+    fake_comfy.forced_output_node = "12"
+    with _real_run(client, fake_comfy):
+        r = client.post(
+            "/api/workflows",
+            json={
+                "name": "映射·类型不符",
+                "workflow_json": _vault_workflow(),
+                "usage": "text2video",
+                "usage_source": "user",
+                "output_map": [{"nodeId": "12", "kind": "video"}],
+            },
+        )
+        wid = r.json()["id"]
+        r = client.post(
+            "/api/playground/generate",
+            json={"workflow_id": wid, "input_values": {"1": "x"}},
+        )
+        job = _wait_done(client, r.json()["job"]["id"])
+        assert job["status"] == "failed"
+        assert "媒体类型与映射不符" in job["error"]
+        assert job["output_paths"] == []
+
+
+# ---------------------------------------------------------------------------
+# 4. Bad / cancelled mappings
+# ---------------------------------------------------------------------------
+
+
+def test_invalid_mappings_are_rejected_in_chinese(client):
+    r = client.post(
+        "/api/workflows",
+        json={"name": "坏映射", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "999", "kind": "video"}]},
+    )
+    assert r.status_code == 422 and "不存在" in r.json()["detail"]
+
+    r = client.post(
+        "/api/workflows",
+        json={"name": "坏映射2", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "12", "kind": "audio"}]},
+    )
+    assert r.status_code == 422 and "媒体类型" in r.json()["detail"]
+
+    # a prompt/parameter node can never be an output
+    r = client.post(
+        "/api/workflows",
+        json={"name": "坏映射3", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "1", "kind": "video"}]},
+    )
+    assert r.status_code == 422 and "不能作为输出节点" in r.json()["detail"]
+
+    wid = client.post(
+        "/api/workflows",
+        json={"name": "先好后坏", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "12", "kind": "video"}]},
+    ).json()["id"]
+    r = client.patch(
+        f"/api/workflows/{wid}", json={"output_map": [{"nodeId": "42", "kind": "image"}]}
+    )
+    assert r.status_code == 422 and "不存在" in r.json()["detail"]
+
+
+def test_clearing_the_mapping_resets_to_pending(client):
+    wid = client.post(
+        "/api/workflows",
+        json={"name": "映射后清空", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "12", "kind": "video"}]},
+    ).json()["id"]
+    wf = client.get(f"/api/workflows/{wid}").json()
+    assert wf["is_enabled"] is True
+
+    r = client.patch(f"/api/workflows/{wid}", json={"output_map": []})
+    assert r.status_code == 200
+    wf = r.json()
+    assert wf["is_enabled"] is False and not wf["output_map"]
+    assert wf["usage"] == "text2video"  # the usage decision survives
+    gen = client.post(
+        "/api/playground/generate",
+        json={"workflow_id": wid, "input_values": {"1": "x"}},
+    )
+    assert gen.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# 5. Saved mappings survive reanalyze / field edits / migration
+# ---------------------------------------------------------------------------
+
+
+def test_saved_mapping_survives_reanalyze_and_field_edits(client):
+    r = client.post(
+        "/api/workflows",
+        json={"name": "重分析·保映射", "workflow_json": _vault_workflow(),
+              "usage": "text2video",
+              "output_map": [{"nodeId": "12", "kind": "video"}]},
+    )
+    wid = r.json()["id"]
+
+    r = client.patch(f"/api/workflows/{wid}", json={"name": "改名不影响映射",
+                                                    "description": "编辑其他字段"})
+    wf = r.json()
+    assert wf["output_map"] == [{"nodeId": "12", "kind": "video"}]
+    assert wf["output_map_source"] == "user" and wf["is_enabled"] is True
+
+    r = client.post(f"/api/workflows/{wid}/reanalyze")
+    assert r.status_code == 200
+    wf = r.json()
+    assert wf["output_map"] == [{"nodeId": "12", "kind": "video"}]
+    assert wf["output_map_source"] == "user"
+    assert wf["usage"] == "text2video" and wf["usage_source"] == "user"
+    assert wf["is_enabled"] is True
+
+
+def test_sql_forced_enabled_row_without_mapping_is_blocked(client):
+    """Backend-direct generation on an unconfirmed row is refused."""
+    wid = client.post(
+        "/api/workflows",
+        json={"name": "强开", "workflow_json": _vault_workflow(),
+              "usage": "text2video"},
+    ).json()["id"]
+    conn = get_db(settings.db_path)
+    try:
+        conn.execute("UPDATE workflows SET is_enabled = 1 WHERE id = ?", (wid,))
+        conn.commit()
+    finally:
+        conn.close()
+    gen = client.post(
+        "/api/playground/generate",
+        json={"workflow_id": wid, "input_values": {"1": "x"}},
+    )
+    assert gen.status_code == 400
+    assert "输出节点" in gen.json()["detail"]
+
+
+def test_migration_parks_unknown_output_rows_and_keeps_trusted_ones(client):
+    import asyncio
+
+    conn = get_db(settings.db_path)
+    try:
+        # Pre-HEL-638 shape: user confirmed a usage on an unknown-output
+        # graph and the old PATCH re-enabled it.
+        conn.execute(
+            "INSERT INTO workflows (name, kind, workflow_json, input_schema,"
+            " output_schema, is_enabled, usage, usage_source)"
+            " VALUES (?, 'video', ?, '[]', '[]', 1, 'text2video', 'user')",
+            ("存量·未知输出", json.dumps(_vault_workflow())),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+    asyncio.run(migrate_db(settings.db_path))  # idempotent re-run
+
+    wfs = {w["name"]: w for w in client.get("/api/workflows").json()}
+    parked = wfs["存量·未知输出"]
+    assert parked["is_enabled"] is False
+    assert parked["usage"] == "text2video"  # decision preserved
+    assert not parked["output_map"] and parked["needs_mapping"] is True
+
+    # A trusted five-type import keeps working untouched.
+    r = client.post(
+        "/api/workflows",
+        json={"name": "存量·可信", "kind": "image",
+              "workflow_json": json.loads(
+                  (FIXTURES / "qwen_image_t2i_API.json").read_text(encoding="utf-8")
+              )},
+    )
+    assert r.status_code == 200
+    trusted = next(w for w in client.get("/api/workflows").json() if w["id"] == r.json()["id"])
+    assert trusted["is_enabled"] is True
+    assert trusted["output_map_source"] == "auto"
+    assert any(e["nodeId"] == "60" and e["kind"] == "image" for e in trusted["output_map"])

calliope-backend/tests/test_video_continue.py

@@ -268,12 +268,14 @@ def _scene_ids(pid: int) -> list[int]:
 
 
 async def _fake_history(_client, _prompt_id, *_args):
-    """A successful ComfyUI history with one video output named out.mp4.
-    Extra positional args (job_id) are accepted for the cancel-check wiring."""
+    """A successful ComfyUI history with one video output named out.mp4,
+    reported under the workflow's real output node (30 = the tagged
+    VHS_VideoCombine). Extra positional args (job_id) are accepted for the
+    cancel-check wiring."""
     return {
         "status": {"status_str": "success", "completed": True},
         "outputs": {
-            "99": {"videos": [{"filename": "out.mp4", "subfolder": "", "type": "output"}]}
+            "30": {"videos": [{"filename": "out.mp4", "subfolder": "", "type": "output"}]}
         },
     }
 

calliope-web/src/lib/api.ts

@@ -137,6 +137,7 @@ import type {
   WorkflowAnalyzeResult,
   ComfyDynamicInput,
   ComfyDynamicOutput,
+  WorkflowOutputMapEntry,
 } from "./comfy/types";
 export type {
   Job,
@@ -148,6 +149,7 @@ export type {
   WorkflowAnalyzeResult,
   ComfyDynamicInput,
   ComfyDynamicOutput,
+  WorkflowOutputMapEntry,
 };
 
 async function api<T>(path: string, init?: RequestInit): Promise<T> {
@@ -470,6 +472,8 @@ export const workflows = {
     usage?: WorkflowUsage;
     usage_source?: "auto" | "user";
     input_schema?: ComfyDynamicInput[];
+    output_map?: WorkflowOutputMapEntry[];
+    output_map_source?: "auto" | "user";
     description?: string;
     prompt_profile?: string;
     endpoint_id?: string | null;
@@ -490,6 +494,8 @@ export const workflows = {
       usage: WorkflowUsage;
       usage_source: "auto" | "user";
       input_schema: ComfyDynamicInput[];
+      output_map: WorkflowOutputMapEntry[];
+      output_map_source: "auto" | "user";
     }>,
   ) =>
     api<Workflow>(`/api/workflows/${id}`, {

calliope-web/src/lib/comfy/types.ts

@@ -45,10 +45,21 @@ export interface ComfyDynamicOutput {
 	label: string;
 	/** Role from title tag, e.g. (Output:image). */
 	role?: string | null;
-	kind: ComfyOutputKind;
+	/** Null while an unknown custom node awaits the user's media-type pick. */
+	kind: ComfyOutputKind | null;
 	/** Untagged detection: node class + which rule typed it. */
 	class_type?: string;
 	evidence?: 'registry' | 'class_name_match' | 'unknown_node';
+	/** Form default for unknown candidates — never auto-applied. */
+	suggested_kind?: 'image' | 'video' | null;
+}
+
+/** One entry of a workflow's confirmed output mapping (HEL-638). */
+export interface WorkflowOutputMapEntry {
+	nodeId: string;
+	kind: 'image' | 'video';
+	label?: string;
+	class_type?: string;
 }
 
 /** The five workflow usages the studio distinguishes (HEL-636). */
@@ -114,6 +125,11 @@ export interface Workflow {
 	/** Bound ComfyUI connection id (settings.comfyui_endpoints); null = active. */
 	endpoint_id?: string | null;
 	warnings?: WorkflowWarning[];
+	/** HEL-638: confirmed output mapping ({nodeId, kind}); null/empty = pending. */
+	output_map?: WorkflowOutputMapEntry[] | null;
+	output_map_source?: 'auto' | 'user' | null;
+	/** True when the graph has unknown output nodes needing designation. */
+	needs_mapping?: boolean;
 }
 
 export interface Job {

calliope-web/src/lib/components/studio/StudioGenerate.svelte

@@ -63,13 +63,20 @@
 
 	const endpoints = $derived($settingsQuery.data?.comfyui_endpoints ?? []);
 	const activeId = $derived($settingsQuery.data?.comfyui_active_id ?? endpoints[0]?.id ?? '');
-	// HEL-636: the page serves its usages (image → t2i/i2i, video →
-	// t2v/i2v/ref2v); pending workflows (usage = null) never appear.
-	const kindWorkflows = $derived(
-		($workflowsQuery.data ?? []).filter(
-			(w) => w.is_enabled && (w.usage ?? null) !== null && USAGE_BY_KIND[kind].includes(w.usage!),
-		),
-	);
+ 	// HEL-636: the page serves its usages (image → t2i/i2i, video →
+ 	// t2v/i2v/ref2v); pending workflows (usage = null) never appear.
+ 	// HEL-638: workflows with unconfirmed unknown output nodes are also
+ 	// hidden — enabled rows carry a settled output mapping, this is a
+ 	// second guard for stale rows.
+ 	const kindWorkflows = $derived(
+ 		($workflowsQuery.data ?? []).filter(
+ 			(w) =>
+ 				w.is_enabled &&
+ 				(w.usage ?? null) !== null &&
+ 				!w.needs_mapping &&
+ 				USAGE_BY_KIND[kind].includes(w.usage!),
+ 		),
+ 	);
 
 	const allJobs = $derived.by(() => {
 		void $jobsQuery.data;

calliope-web/src/lib/i18n/de.ts

@@ -1435,6 +1435,17 @@ export const de: Dict = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': 'Ausgabeknoten',
+	'conn.wfOutputNeedPick': 'Unbekannte benutzerdefinierte Ausgabeknoten — bitte zuordnen',
+	'conn.wfOutputHint': 'Markieren Sie die Knoten, die tatsächlich Ergebnisse liefern, und wählen Sie je den Medientyp; die Generierung sammelt Bilder/Videos nur aus diesen Knoten.',
+	'conn.wfOutputPick': 'Mindestens einen Ausgabeknoten markieren',
+	'conn.wfOutputKindAria': 'Medientyp des Ausgabeknotens',
+	'conn.wfOutputPendingBadge': 'Ausgabe offen',
+	'conn.wfOutputNeedMap': 'Dieser Workflow enthält nicht erkannte Ausgabeknoten; ordnen Sie sie samt Medientyp zu, um ihn zu aktivieren',
+	'conn.wfOutputApply': 'Ausgabezuordnung speichern',
+	'conn.wfMapCancel': 'Abbrechen',
+	'conn.wfOutputApplied': 'Ausgabezuordnung gespeichert: Knoten {nodes}',
+	'conn.wfOutputCleared': 'Ausgabezuordnung gelöscht — der Workflow ist wieder ausstehend',
 	'conn.missingMaterial': 'Ein dediziertes Qwen Image 2.1-Workflow-JSON fehlt noch im Repo: aus ComfyUI exportieren und hier importieren — der generische Importweg steht bereit.',
 
 	// --- Eigenständige Bild-/Videogenerierung (a3/a4, HEL-612) ---

calliope-web/src/lib/i18n/en.ts

@@ -1435,6 +1435,17 @@ export const en = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': 'Output nodes',
+	'conn.wfOutputNeedPick': 'Unrecognized custom output nodes — designate them',
+	'conn.wfOutputHint': 'Check the nodes that actually produce results and pick each one’s media type; generation harvests images/videos only from these nodes.',
+	'conn.wfOutputPick': 'Check at least one output node',
+	'conn.wfOutputKindAria': 'Output node media type',
+	'conn.wfOutputPendingBadge': 'Outputs unset',
+	'conn.wfOutputNeedMap': 'This workflow has unrecognized output nodes; designate them and their media type to enable it',
+	'conn.wfOutputApply': 'Save output mapping',
+	'conn.wfMapCancel': 'Cancel',
+	'conn.wfOutputApplied': 'Output mapping saved: node {nodes}',
+	'conn.wfOutputCleared': 'Output mapping cleared — the workflow is pending again',
 	'conn.missingMaterial': 'A dedicated Qwen Image 2.1 workflow JSON is not in the repo yet: export it from ComfyUI and import it here — the generic import path is ready.',
 	// --- Standalone image / video generation (a3/a4, HEL-612) ---
 	'studio.image.title': 'Image',

calliope-web/src/lib/i18n/es.ts

@@ -1435,6 +1435,17 @@ export const es: Dict = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': 'Nodos de salida',
+	'conn.wfOutputNeedPick': 'Nodos de salida personalizados no reconocidos: asígnalos',
+	'conn.wfOutputHint': 'Marca los nodos que realmente producen resultados y elige el tipo de medio de cada uno; la generación solo extrae imágenes/vídeos de estos nodos.',
+	'conn.wfOutputPick': 'Marca al menos un nodo de salida',
+	'conn.wfOutputKindAria': 'Tipo de medio del nodo de salida',
+	'conn.wfOutputPendingBadge': 'Salida pendiente',
+	'conn.wfOutputNeedMap': 'Este flujo tiene nodos de salida no reconocidos; asígnalos y su tipo de medio para activarlo',
+	'conn.wfOutputApply': 'Guardar mapeo de salida',
+	'conn.wfMapCancel': 'Cancelar',
+	'conn.wfOutputApplied': 'Mapeo de salida guardado: nodo {nodes}',
+	'conn.wfOutputCleared': 'Mapeo de salida borrado: el flujo vuelve a estar pendiente',
 	'conn.missingMaterial': 'Aún falta en el repo un JSON dedicado de Qwen Image 2.1: expórtalo desde ComfyUI e impórtalo aquí; la vía genérica ya está lista.',
 
 	// --- Generación independiente de imagen / vídeo (a3/a4, HEL-612) ---

calliope-web/src/lib/i18n/fr.ts

@@ -1435,6 +1435,17 @@ export const fr: Dict = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': 'Nœuds de sortie',
+	'conn.wfOutputNeedPick': 'Nœuds de sortie personnalisés non reconnus — à désigner',
+	'conn.wfOutputHint': 'Cochez les nœuds qui produisent réellement les résultats et choisissez leur type de média ; la génération ne collecte les images/vidéos que sur ces nœuds.',
+	'conn.wfOutputPick': 'Cochez au moins un nœud de sortie',
+	'conn.wfOutputKindAria': 'Type de média du nœud de sortie',
+	'conn.wfOutputPendingBadge': 'Sortie à définir',
+	'conn.wfOutputNeedMap': 'Ce workflow contient des nœuds de sortie non reconnus ; désignez-les avec leur type de média pour l’activer',
+	'conn.wfOutputApply': 'Enregistrer le mapping de sortie',
+	'conn.wfMapCancel': 'Annuler',
+	'conn.wfOutputApplied': 'Mapping de sortie enregistré : nœud {nodes}',
+	'conn.wfOutputCleared': 'Mapping de sortie effacé — le workflow est de nouveau en attente',
 	'conn.missingMaterial': 'Un JSON de workflow Qwen Image 2.1 dédié manque encore au dépôt : exportez-le depuis ComfyUI et importez-le ici — le chemin d\'import générique est prêt.',
 
 	// --- Génération indépendante image / vidéo (a3/a4, HEL-612) ---

calliope-web/src/lib/i18n/ja.ts

@@ -1435,6 +1435,17 @@ export const ja: Dict = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': '出力ノード',
+	'conn.wfOutputNeedPick': '認識できないカスタム出力ノードがあります。手動で指定してください',
+	'conn.wfOutputHint': '実際に結果を出すノードにチェックし、それぞれのメディア種別を選んでください。生成時にこれらのノードからのみ画像/動画を取り出します。',
+	'conn.wfOutputPick': '出力ノードを1つ以上チェックしてください',
+	'conn.wfOutputKindAria': '出力ノードのメディア種別',
+	'conn.wfOutputPendingBadge': '出力未指定',
+	'conn.wfOutputNeedMap': '認識できない出力ノードを含みます。ノードとメディア種別を指定すると有効化できます',
+	'conn.wfOutputApply': '出力マッピングを保存',
+	'conn.wfMapCancel': 'キャンセル',
+	'conn.wfOutputApplied': '出力マッピングを保存しました：ノード {nodes}',
+	'conn.wfOutputCleared': '出力マッピングを解除しました。ワークフローは保留に戻ります',
 	'conn.missingMaterial': 'Qwen Image 2.1 専用のワークフロー JSON はまだリポジトリにありません。ComfyUI から書き出してここで取り込んでください。汎用の取り込み経路は用意済みです。',
 
 	// --- 独立画像/動画生成（a3/a4、HEL-612）---

calliope-web/src/lib/i18n/ko.ts

@@ -1435,6 +1435,17 @@ export const ko: Dict = {
 	'conn.wfSourceAuto': 'Usage auto-detected',
 	'conn.wfDisabledBadge': 'Disabled',
 	'conn.wfConfirmAria': 'Confirm workflow usage',
+	'conn.wfOutputLabel': '출력 노드',
+	'conn.wfOutputNeedPick': '인식할 수 없는 사용자 지정 출력 노드가 있습니다. 직접 지정하세요',
+	'conn.wfOutputHint': '실제 결과를 내는 노드를 체크하고 각각의 미디어 유형을 선택하세요. 생성 시 이 노드에서만 이미지/동영상을 수집합니다.',
+	'conn.wfOutputPick': '출력 노드를 하나 이상 체크하세요',
+	'conn.wfOutputKindAria': '출력 노드 미디어 유형',
+	'conn.wfOutputPendingBadge': '출력 미지정',
+	'conn.wfOutputNeedMap': '인식할 수 없는 출력 노드가 포함되어 있습니다. 노드와 미디어 유형을 지정하면 활성화됩니다',
+	'conn.wfOutputApply': '출력 매핑 저장',
+	'conn.wfMapCancel': '취소',
+	'conn.wfOutputApplied': '출력 매핑 저장됨: 노드 {nodes}',
+	'conn.wfOutputCleared': '출력 매핑이 삭제되었습니다. 워크플로가 다시 대기 상태가 됩니다',
 	'conn.missingMaterial': 'Qwen Image 2.1 전용 워크플로 JSON이 아직 저장소에 없습니다. ComfyUI에서 내보내 여기서 가져오세요. 범용 가져오기 경로는 준비되어 있습니다.',
 
 	// --- 독립 이미지/비디오 생성 (a3/a4, HEL-612) ---

calliope-web/src/lib/i18n/zh.ts

@@ -1435,6 +1435,17 @@ export const zh: Dict = {
 	'conn.wfSourceAuto': '用途为自动判定',
 	'conn.wfDisabledBadge': '已停用',
 	'conn.wfConfirmAria': '确认工作流用途',
+	'conn.wfOutputLabel': '输出节点',
+	'conn.wfOutputNeedPick': '存在无法识别的自定义输出节点，需人工指定',
+	'conn.wfOutputHint': '勾选实际产生结果的节点，并为每个节点选择媒体类型；生成时只从这些节点提取图片/视频。',
+	'conn.wfOutputPick': '请至少勾选一个输出节点',
+	'conn.wfOutputKindAria': '输出节点媒体类型',
+	'conn.wfOutputPendingBadge': '待指定输出',
+	'conn.wfOutputNeedMap': '该工作流含无法识别的输出节点，指定输出节点及媒体类型后才能启用',
+	'conn.wfOutputApply': '保存输出映射',
+	'conn.wfMapCancel': '取消',
+	'conn.wfOutputApplied': '已保存输出映射：节点 {nodes}',
+	'conn.wfOutputCleared': '已清除输出映射，工作流回到待确认状态',
  	'conn.missingMaterial': 'Qwen Image 2.1 的专用工作流 JSON 仓库中暂缺：请从 ComfyUI 导出后在此导入，通用导入通道已就绪。',
 	// --- 独立生图 / 生视频（a3/a4，HEL-612）---
 	'studio.image.title': '生图',

calliope-web/src/routes/connections/+page.svelte

@@ -7,7 +7,7 @@
 	import ConfirmDialog from '$lib/components/ui/ConfirmDialog.svelte';
 	import Icon from '$lib/components/ui/Icon.svelte';
 	import EmptyState from '$lib/components/ui/EmptyState.svelte';
-	import { apiErrorMessage, comfy, settings, workflows, type ComfyEndpoint, type Workflow, type WorkflowUsage, type WorkflowAnalyzeResult } from '$lib/api';
+	import { apiErrorMessage, comfy, settings, workflows, type ComfyEndpoint, type Workflow, type WorkflowOutputMapEntry, type WorkflowUsage, type WorkflowAnalyzeResult } from '$lib/api';
 	import { WORKFLOW_USAGES } from '$lib/comfy/types';
 	import { toast } from '$lib/toast';
 	import { t } from '$lib/i18n.svelte';
@@ -337,6 +337,9 @@
 	let wfAnalysis = $state<WorkflowAnalyzeResult | null>(null);
 	let wfUsage = $state<WorkflowUsage | null>(null);
 	let wfRoleOverrides = $state<Record<string, string>>({});
+	// HEL-638: explicit output-node designation for unknown custom nodes.
+	let wfOutputPicked = $state<Record<string, boolean>>({});
+	let wfOutputKind = $state<Record<string, 'image' | 'video'>>({});
 	let wfAnalyzing = $state(false);
 	let wfAnalyzeError = $state('');
 
@@ -374,6 +377,8 @@
 		wfAnalysis = null;
 		wfUsage = null;
 		wfRoleOverrides = {};
+		wfOutputPicked = {};
+		wfOutputKind = {};
 		wfAnalyzeError = '';
 	}
 
@@ -390,6 +395,19 @@
 			wfAnalysis = res;
 			wfUsage = res.usage ?? null;
 			wfRoleOverrides = {};
+			// Unknown output candidates: preselect only when there is exactly
+			// one — with several the user must decide (HEL-638).
+			const candidates = (res.usage_evidence.outputs ?? []).filter(
+				(o) => o.evidence === 'unknown_node',
+			);
+			const picked: Record<string, boolean> = {};
+			const kinds: Record<string, 'image' | 'video'> = {};
+			for (const c of candidates) {
+				kinds[c.nodeId] = c.suggested_kind ?? 'image';
+			}
+			if (candidates.length === 1) picked[candidates[0].nodeId] = true;
+			wfOutputPicked = picked;
+			wfOutputKind = kinds;
 			wfStep = 'confirm';
 		} catch (err) {
 			wfAnalyzeError = apiErrorMessage(err);
@@ -402,6 +420,22 @@
 	const wfFatal = $derived(
 		wfStep === 'confirm' && wfAnalysis !== null && (wfAnalysis.outputs?.length ?? 0) === 0,
 	);
+	/** Unknown custom output nodes the user must designate (HEL-638). */
+	const wfOutputCandidates = $derived(
+		wfStep === 'confirm' && wfAnalysis !== null
+			? (wfAnalysis.usage_evidence.outputs ?? []).filter((o) => o.evidence === 'unknown_node')
+			: [],
+	);
+	const wfOutputRows = $derived(
+		wfOutputCandidates.map((o) => ({
+			...o,
+			picked: wfOutputPicked[o.nodeId] ?? false,
+			kind: wfOutputKind[o.nodeId] ?? o.suggested_kind ?? 'image',
+		})),
+	);
+	const wfOutputPickedCount = $derived(
+		Object.values(wfOutputPicked).filter(Boolean).length,
+	);
 	const wfMustPick = $derived(
 		wfStep === 'confirm' &&
 			wfAnalysis !== null &&
@@ -409,7 +443,12 @@
 			(wfAnalysis.usage_ambiguous || wfAnalysis.needs_mapping) &&
 			wfUsage === null,
 	);
-	const wfSaveable = $derived(wfStep === 'confirm' && wfUsage !== null && !wfFatal);
+	const wfSaveable = $derived(
+		wfStep === 'confirm' &&
+			wfUsage !== null &&
+			!wfFatal &&
+			(wfOutputCandidates.length === 0 || wfOutputPickedCount > 0),
+	);
 	/** Media slots the confirm page lets the user re-role. */
 	const wfSlotRows = $derived.by(() => {
 		if (wfStep !== 'confirm' || !wfAnalysis) return [];
@@ -430,18 +469,29 @@
 		mutationFn: async () => {
 			if (!wfParsed.valid || !wfParsed.json || !wfAnalysis || !wfUsage)
 				throw new Error(t('conn.wfPickUsage'));
+			if (wfOutputCandidates.length > 0 && wfOutputPickedCount === 0)
+				throw new Error(t('conn.wfOutputPick'));
 			const corrected = Object.keys(wfRoleOverrides).length > 0;
 			const schema = wfAnalysis.inputs.map((e) =>
 				wfRoleOverrides[e.nodeId]
 					? { ...e, role: wfRoleOverrides[e.nodeId], manual: true }
 					: e,
 			);
+			const outputMap = wfOutputRows
+				.filter((r) => r.picked)
+				.map((r) => ({
+					nodeId: r.nodeId,
+					kind: r.kind,
+					label: r.label,
+					...(r.class_type ? { class_type: r.class_type } : {}),
+				}));
 			return workflows.create({
 				name: wfImportName.trim(),
 				workflow_json: wfParsed.json,
 				usage: wfUsage,
 				usage_source: wfUsage !== wfAnalysis.usage || corrected ? 'user' : 'auto',
 				input_schema: schema,
+				...(outputMap.length ? { output_map: outputMap, output_map_source: 'user' } : {}),
 				...(wfImportEndpoint ? { endpoint_id: wfImportEndpoint } : {}),
 			});
 		},
@@ -488,6 +538,80 @@
 		},
 	});
 
+	// ------------------------------------------------------------------
+	// HEL-638: per-row output-node designation for pending workflows with
+	// unknown custom output nodes (imported earlier or parked by migration).
+	// ------------------------------------------------------------------
+	let wfMapEditId = $state<number | null>(null);
+	let wfMapPicked = $state<Record<string, boolean>>({});
+	let wfMapKind = $state<Record<string, 'image' | 'video'>>({});
+
+	function rowOutputCandidates(wf: Workflow) {
+		return (wf.output_schema ?? []).filter((o) => o.evidence === 'unknown_node');
+	}
+
+	function rowNeedsOutputMap(wf: Workflow): boolean {
+		return Boolean(wf.needs_mapping) && !(wf.output_map ?? []).length;
+	}
+
+	function toggleWfMapEdit(wf: Workflow) {
+		if (wfMapEditId === wf.id) {
+			wfMapEditId = null;
+			return;
+		}
+		const picked: Record<string, boolean> = {};
+		const kinds: Record<string, 'image' | 'video'> = {};
+		const mapped = new Map((wf.output_map ?? []).map((e) => [e.nodeId, e.kind]));
+		for (const c of rowOutputCandidates(wf)) {
+			const known = mapped.get(c.nodeId);
+			if (known) picked[c.nodeId] = true;
+			kinds[c.nodeId] = known ?? c.suggested_kind ?? 'image';
+		}
+		wfMapPicked = picked;
+		wfMapKind = kinds;
+		wfMapEditId = wf.id;
+	}
+
+	const wfMapMutation = createMutation({
+		mutationFn: ({ id, outputMap }: { id: number; outputMap: WorkflowOutputMapEntry[] }) =>
+			workflows.update(
+				id,
+				outputMap.length
+					? { output_map: outputMap, output_map_source: 'user' }
+					: { output_map: [] },
+			),
+		onSuccess: (wf) => {
+			client.invalidateQueries({ queryKey: ['workflows'] });
+			wfMapEditId = null;
+			const nodes = (wf.output_map ?? []).map((e) => e.nodeId).join('、');
+			toast.success(
+				wf.output_map?.length
+					? t('conn.wfOutputApplied', { nodes })
+					: t('conn.wfOutputCleared'),
+			);
+		},
+		onError: (err) => {
+			toast.error(
+				t('conn.wfImportFailed', {
+					detail: err instanceof Error ? err.message : String(err),
+				}),
+			);
+		},
+	});
+
+	function submitWfMap(wf: Workflow) {
+		if ($wfMapMutation.isPending) return;
+		const outputMap = rowOutputCandidates(wf)
+			.filter((c) => wfMapPicked[c.nodeId])
+			.map((c) => ({
+				nodeId: c.nodeId,
+				kind: wfMapKind[c.nodeId] ?? c.suggested_kind ?? 'image',
+				label: c.label,
+				...(c.class_type ? { class_type: c.class_type } : {}),
+			}));
+		$wfMapMutation.mutate({ id: wf.id, outputMap });
+	}
+
 	const wfBindMutation = createMutation({
 		mutationFn: ({ id, endpointId }: { id: number; endpointId: string | null }) =>
 			workflows.update(id, { endpoint_id: endpointId }),
@@ -856,10 +980,56 @@
 											</button>
 										{/each}
 									</div>
-									{#if wfMustPick}<span class="f-error">{t('conn.wfPickUsage')}</span>{/if}
-								</div>
+								{#if wfMustPick}<span class="f-error">{t('conn.wfPickUsage')}</span>{/if}
+							</div>
 
-								{#if wfSlotRows.length}
+							{#if wfOutputRows.length}
+								<div class="f-field">
+									<span class="f-label">
+										{t('conn.wfOutputLabel')}<em>*</em>
+										<span class="f-hint-inline">{t('conn.wfOutputNeedPick')}</span>
+									</span>
+									<div class="wf-slot-rows">
+										{#each wfOutputRows as row (row.nodeId)}
+											<div class="wf-slot-row" class:off={!row.picked}>
+												<label class="wf-out-check">
+													<input
+														type="checkbox"
+														checked={row.picked}
+														onchange={(e) =>
+															(wfOutputPicked = {
+																...wfOutputPicked,
+																[row.nodeId]: e.currentTarget.checked,
+															})}
+													/>
+													<span class="wf-slot-id mono">{row.nodeId}</span>
+													<span class="wf-slot-name" title={row.class_type}>{row.label}</span>
+												</label>
+												<select
+													class="a-select"
+													aria-label={t('conn.wfOutputKindAria')}
+													disabled={!row.picked}
+													value={row.kind}
+													onchange={(e) =>
+														(wfOutputKind = {
+															...wfOutputKind,
+															[row.nodeId]: e.currentTarget.value as 'image' | 'video',
+														})}
+												>
+													<option value="image">{t('conn.wfKindImage')}</option>
+													<option value="video">{t('conn.wfKindVideo')}</option>
+												</select>
+											</div>
+										{/each}
+									</div>
+									<span class="f-hint">{t('conn.wfOutputHint')}</span>
+									{#if wfOutputPickedCount === 0}
+										<span class="f-error">{t('conn.wfOutputPick')}</span>
+									{/if}
+								</div>
+							{/if}
+
+							{#if wfSlotRows.length}
 									<div class="f-field">
 										<span class="f-label">{t('conn.wfSlotsLabel')}</span>
 										<div class="wf-slot-rows">
@@ -916,7 +1086,11 @@
 										{t('conn.wfBack')}
 									</Button>
 									<span class="need-test" class:passed={wfSaveable && wfImportName.trim() !== ''}>
-										{wfSaveable ? usageLabel(wfUsage) : t('conn.wfUsageNeedPick')}
+										{wfUsage
+											? wfOutputCandidates.length && wfOutputPickedCount === 0
+												? t('conn.wfOutputPick')
+												: usageLabel(wfUsage)
+											: t('conn.wfUsageNeedPick')}
 									</span>
 									<Button
 										variant="primary"
@@ -948,71 +1122,139 @@
 								<span class="wf-pending-note">· {t('conn.wfPendingCount', { n: pending })}</span>
 							{/if}
 						</p>
-						<div class="wf-table">
-							{#each wfs as wf (wf.id)}
-								<div class="wf-row" class:pending={!wf.usage}>
-									<span class="wf-kind {wf.kind}">
-										<Icon name={wf.kind === 'image' ? 'image' : 'film'} size={14} />
-									</span>
-									<span class="wf-name" title={wf.description ?? wf.name}>{wf.name}</span>
-									{#if wf.usage}
-										<span
-											class="tag tag-usage"
-											title={wf.usage_source === 'user' ? t('conn.wfSourceUser') : t('conn.wfSourceAuto')}
-										>
-											{usageLabel(wf.usage)}
-										</span>
-									{:else}
-										<select
-											class="a-select wf-confirm-select"
-											aria-label={t('conn.wfConfirmAria')}
-											disabled={$wfConfirmMutation.isPending}
-											onchange={(e) => {
-												const usage = e.currentTarget.value as WorkflowUsage;
-												if (usage) $wfConfirmMutation.mutate({ id: wf.id, usage });
-												e.currentTarget.value = '';
-											}}
-										>
-											<option value="">{t('usage.pending')}…</option>
-											{#each WORKFLOW_USAGES as u (u)}
-												<option value={u}>{usageLabel(u)}</option>
-											{/each}
-										</select>
-									{/if}
-									{#if !wf.is_enabled && wf.usage}
-										<span class="tag tag-disabled">{t('conn.wfDisabledBadge')}</span>
-									{/if}
-									{#if wfProblems(wf).length}
-										<span class="tag tag-error" title={wfProblems(wf).join('\n')}>
-											{t('conn.wfBrokenBadge')}
-										</span>
-									{/if}
-									<span class="wf-params">{t('conn.wfParams', { n: wf.input_schema?.length ?? 0 })}</span>
-									<select
-										class="a-select"
-										value={wf.endpoint_id ?? ''}
-										onchange={(e) =>
-											$wfBindMutation.mutate({
-												id: wf.id,
-												endpointId: e.currentTarget.value || null,
-											})}
+					<div class="wf-table">
+						{#each wfs as wf (wf.id)}
+							<div class="wf-row" class:pending={!wf.usage}>
+								<span class="wf-kind {wf.kind}">
+									<Icon name={wf.kind === 'image' ? 'image' : 'film'} size={14} />
+								</span>
+								<span class="wf-name" title={wf.description ?? wf.name}>{wf.name}</span>
+								{#if wf.usage}
+									<span
+										class="tag tag-usage"
+										title={wf.usage_source === 'user' ? t('conn.wfSourceUser') : t('conn.wfSourceAuto')}
 									>
-										<option value="">{t('conn.wfConnectionActive')}</option>
-										{#each endpoints as ep (ep.id)}
-											<option value={ep.id}>{ep.name}</option>
+										{usageLabel(wf.usage)}
+									</span>
+								{:else}
+									<select
+										class="a-select wf-confirm-select"
+										aria-label={t('conn.wfConfirmAria')}
+										disabled={$wfConfirmMutation.isPending}
+										onchange={(e) => {
+											const usage = e.currentTarget.value as WorkflowUsage;
+											if (usage) $wfConfirmMutation.mutate({ id: wf.id, usage });
+											e.currentTarget.value = '';
+										}}
+									>
+										<option value="">{t('usage.pending')}…</option>
+										{#each WORKFLOW_USAGES as u (u)}
+											<option value={u}>{usageLabel(u)}</option>
 										{/each}
 									</select>
+								{/if}
+								{#if rowNeedsOutputMap(wf)}
 									<button
 										type="button"
-										class="link-danger"
-										title={t('conn.delete')}
-										onclick={() => $wfDeleteMutation.mutate(wf.id)}
+										class="tag tag-mapbtn"
+										class:open={wfMapEditId === wf.id}
+										title={t('conn.wfOutputNeedMap')}
+										onclick={() => toggleWfMapEdit(wf)}
 									>
-										<Icon name="trash" size={14} />
+										<Icon name="alert" size={12} />
+										{t('conn.wfOutputPendingBadge')}
 									</button>
+								{/if}
+								{#if !wf.is_enabled && wf.usage && !rowNeedsOutputMap(wf)}
+									<span class="tag tag-disabled">{t('conn.wfDisabledBadge')}</span>
+								{/if}
+								{#if wfProblems(wf).length}
+									<span class="tag tag-error" title={wfProblems(wf).join('\n')}>
+										{t('conn.wfBrokenBadge')}
+									</span>
+								{/if}
+								<span class="wf-params">{t('conn.wfParams', { n: wf.input_schema?.length ?? 0 })}</span>
+								<select
+									class="a-select"
+									value={wf.endpoint_id ?? ''}
+									onchange={(e) =>
+										$wfBindMutation.mutate({
+											id: wf.id,
+											endpointId: e.currentTarget.value || null,
+										})}
+								>
+									<option value="">{t('conn.wfConnectionActive')}</option>
+									{#each endpoints as ep (ep.id)}
+										<option value={ep.id}>{ep.name}</option>
+									{/each}
+								</select>
+								<button
+									type="button"
+									class="link-danger"
+									title={t('conn.delete')}
+									onclick={() => $wfDeleteMutation.mutate(wf.id)}
+								>
+									<Icon name="trash" size={14} />
+								</button>
+							</div>
+							{#if wfMapEditId === wf.id}
+								<div class="wf-map-editor">
+									<p class="wf-map-hint">{t('conn.wfOutputHint')}</p>
+									<div class="wf-slot-rows">
+										{#each rowOutputCandidates(wf) as c (c.nodeId)}
+											<div class="wf-slot-row" class:off={!wfMapPicked[c.nodeId]}>
+												<label class="wf-out-check">
+													<input
+														type="checkbox"
+														checked={wfMapPicked[c.nodeId] ?? false}
+														onchange={(e) =>
+															(wfMapPicked = {
+																...wfMapPicked,
+																[c.nodeId]: e.currentTarget.checked,
+															})}
+													/>
+													<span class="wf-slot-id mono">{c.nodeId}</span>
+													<span class="wf-slot-name" title={c.class_type}>{c.label}</span>
+												</label>
+												<select
+													class="a-select"
+													aria-label={t('conn.wfOutputKindAria')}
+													disabled={!wfMapPicked[c.nodeId]}
+													value={wfMapKind[c.nodeId] ?? c.suggested_kind ?? 'image'}
+													onchange={(e) =>
+														(wfMapKind = {
+															...wfMapKind,
+															[c.nodeId]: e.currentTarget.value as 'image' | 'video',
+														})}
+												>
+													<option value="image">{t('conn.wfKindImage')}</option>
+													<option value="video">{t('conn.wfKindVideo')}</option>
+												</select>
+											</div>
+										{/each}
+									</div>
+									<div class="form-actions">
+										<Button
+											variant="secondary"
+											size="sm"
+											onclick={() => (wfMapEditId = null)}
+										>
+											{t('conn.wfMapCancel')}
+										</Button>
+										<Button
+											variant="primary"
+											size="sm"
+											disabled={!Object.values(wfMapPicked).some(Boolean)}
+											loading={$wfMapMutation.isPending}
+											onclick={() => submitWfMap(wf)}
+										>
+											{t('conn.wfOutputApply')}
+										</Button>
+									</div>
 								</div>
-							{/each}
-						</div>
+							{/if}
+						{/each}
+					</div>
 					{/if}
 					</section>
 				</section>
@@ -1606,6 +1848,52 @@
 	.wf-slot-row .a-select {
 		width: 130px;
 	}
+	.wf-slot-row.off {
+		opacity: 0.55;
+	}
+	.wf-out-check {
+		display: flex;
+		align-items: center;
+		gap: 8px;
+		flex: 1;
+		min-width: 0;
+		cursor: pointer;
+	}
+	.wf-out-check input[type='checkbox'] {
+		accent-color: var(--accent);
+	}
+	.tag-mapbtn {
+		display: inline-flex;
+		align-items: center;
+		gap: 4px;
+		cursor: pointer;
+		color: var(--error);
+		background: var(--error-dim);
+		border: 1px solid transparent;
+	}
+	.tag-mapbtn.open {
+		color: var(--accent);
+		background: var(--accent-dim);
+		border-color: var(--accent);
+	}
+	.wf-map-editor {
+		display: flex;
+		flex-direction: column;
+		gap: 8px;
+		padding: 10px 12px;
+		margin: 2px 0 6px;
+		border-radius: 7px;
+		border: 1px dashed var(--border);
+		background: var(--surface);
+	}
+	.wf-map-hint {
+		margin: 0;
+		font-size: 11.5px;
+		color: var(--text-3);
+	}
+	.wf-map-editor .form-actions {
+		justify-content: flex-end;
+	}
 	.wf-params {
 		margin: 0 0 10px;
 		font-size: 11.5px;