Lock built-in pipeline wiring by default with Full control toggle.

Guided mode restricts loader outputs to their core ports, locks internal
Studio-to-sampler links, defaults image processing to Timeline scene image,
and blocks invalid wires at generation time. Full control restores manual rewiring.

movie_editor/backend/builder.py

@@ -23,6 +23,7 @@ import json
 from typing import Any, Optional
 
 from . import config
+from . import pipeline_wiring
 from .nodes import WIDGET_PRIMITIVES, connection_inputs, node_outputs, _combo_default
 
 # ── fixed core ────────────────────────────────────────────────────────────────
@@ -405,20 +406,20 @@ def build(object_info: dict, models_config: dict, params: dict, media: dict | No
                 graph[dnode]["inputs"][dinput] = [sid, oidx]
                 report["wired"].append(f"{s.get('node_class')}.{out_name} -> {dnode}.{dinput}")
 
-    # 4b. core input overrides: redirect a built-in core node input to a chosen source
-    # ("core:<coreId>:<idx>" or "out:<slotId>:<outName>"). Overrides the default wiring.
-    for cid, ovs in ((models_config or {}).get("core_overrides") or {}).items():
-        if cid not in graph:
-            continue
-        for inp, source in (ovs or {}).items():
-            if not source:
+    # 4b. core input overrides (full-control only — guided mode keeps the fixed core path).
+    if not pipeline_wiring.wiring_locked(models_config):
+        for cid, ovs in ((models_config or {}).get("core_overrides") or {}).items():
+            if cid not in graph:
                 continue
-            src = _resolve_source(source, slot_node_id, slot_def, object_info)
-            if src:
-                graph[cid]["inputs"][inp] = list(src)
-                report["wired"].append(f"{source} -> {cid}.{inp} (core override)")
-            else:
-                report["unsatisfied"].append(f"core override {cid}.{inp}: '{source}' could not be resolved.")
+            for inp, source in (ovs or {}).items():
+                if not source:
+                    continue
+                src = _resolve_source(source, slot_node_id, slot_def, object_info)
+                if src:
+                    graph[cid]["inputs"][inp] = list(src)
+                    report["wired"].append(f"{source} -> {cid}.{inp} (core override)")
+                else:
+                    report["unsatisfied"].append(f"core override {cid}.{inp}: '{source}' could not be resolved.")
 
     # 5. auto-wire remaining unbound typed inputs by unique producer.
     producers = _producers(graph, slots, slot_node_id, slot_def, object_info)
@@ -444,6 +445,10 @@ def build(object_info: dict, models_config: dict, params: dict, media: dict | No
                 active_slots.add(src.split(":", 2)[1])
     _autowire(graph, slots, slot_node_id, slot_def, object_info, producers, report, active_slots)
 
+    for msg in pipeline_wiring.validate_models_wiring(models_config):
+        report["blocking"].append(msg)
+        report["unsatisfied"].append(msg)
+
     # With the built-in pipeline disabled, the result must come from the global outputs (or a
     # save node the user wired). Warn (non-blocking) if nothing will surface to the editor.
     if disable_core and "global_out" not in graph:
@@ -462,6 +467,7 @@ def core_graph(object_info: dict, models_config: dict | None = None) -> list[dic
     object_info = object_info or {}
     overrides = (models_config or {}).get("core_overrides") or {}
     slots = (models_config or {}).get("slots", []) or []
+    locked = pipeline_wiring.wiring_locked(models_config)
 
     rev: dict[str, list] = {}
     for cid, links in CORE_LINKS.items():
@@ -513,13 +519,16 @@ def core_graph(object_info: dict, models_config: dict | None = None) -> list[dic
         for inp, (src, idx) in CORE_LINKS.get(cid, {}).items():
             t = _input_type(cid, inp) or "*"
             builtin = f"{src} · {_out_name(src, idx)}"
+            is_internal = True
             inputs.append({"name": inp, "type": t, "from": "internal", "detail": builtin,
-                           "value": ov.get(inp, ""), "options": _options(t, builtin, cid)})
+                           "value": ov.get(inp, ""), "options": _options(t, builtin, cid),
+                           "locked": locked})
         for inp, (t, req) in open_by_node.get(cid, {}).items():
             builtin = f"a {t} from your loaders"
             inputs.append({"name": inp, "type": t, "from": "loader", "required": req,
                            "detail": builtin + ("" if req else " (optional)"),
-                           "value": ov.get(inp, ""), "options": _options(t, "(auto-wire from loaders)", cid)})
+                           "value": ov.get(inp, ""), "options": _options(t, "(auto-wire from loaders)", cid),
+                           "locked": locked})
         outputs = []
         for i, o in enumerate(node_outputs(nd or {})):
             dests = [f"{d} · {di}" for (oi, d, di) in rev.get(cid, []) if oi == i]

movie_editor/backend/pipeline_wiring.py

@@ -0,0 +1,198 @@
+"""Built-in pipeline wiring rules: guided (locked) vs full-control (manual) mode.
+
+When the built-in FunPack core is enabled and ``full_control`` is false, loader outputs
+may only wire to their designated core ports. Internal core links (Studio -> Conditioning
+-> Sampler -> decode) are never user-overridable. Toggle ``full_control`` to restore the
+legacy free rewiring behaviour.
+"""
+from __future__ import annotations
+
+from typing import Any, Optional
+
+# port id (NodeClass.input) allowed for each model role + output type.
+# output_name: None = any output socket of that type on the slot node.
+ROLE_WIRE_TARGETS: dict[str, list[tuple[str, Optional[str], str]]] = {
+    "unet": [("MODEL", None, "FunPackStudio.model")],
+    "clip": [("CLIP", None, "FunPackStudio.clip")],
+    "video_vae": [("VAE", None, "FunPackLTXAVSceneChainSampler.vae")],
+    "audio_vae": [("VAE", None, "LTXVAudioVAEDecode.audio_vae")],
+    "audio_encoder": [("LATENT", None, "LTXVConcatAVLatent.audio_latent")],
+    "empty_latent": [("LATENT", None, "FunPackStudio.latent")],
+    "image_processing": [
+        ("IMAGE", None, "FunPackStudio.source_image"),
+        ("LATENT", None, "FunPackStudio.latent"),
+    ],
+}
+
+DEFAULT_WIRES_BY_ROLE: dict[str, dict[str, str]] = {
+    "unet": {"MODEL": "port:FunPackStudio.model"},
+    "clip": {"CLIP": "port:FunPackStudio.clip"},
+    "video_vae": {"VAE": "port:FunPackLTXAVSceneChainSampler.vae"},
+    "audio_vae": {"VAE": "port:LTXVAudioVAEDecode.audio_vae"},
+    "audio_encoder": {"LATENT": "port:LTXVConcatAVLatent.audio_latent"},
+    "empty_latent": {"LATENT": "port:FunPackStudio.latent"},
+    "image_processing": {"IMAGE": "port:FunPackStudio.source_image"},
+}
+
+DEFAULT_INPUT_SOURCES_BY_ROLE: dict[str, dict[str, str]] = {
+    "image_processing": {"image": "timeline"},
+}
+
+# Human labels for error messages / UI hints.
+PORT_LABELS: dict[str, str] = {
+    "FunPackStudio.model": "Studio · model",
+    "FunPackStudio.clip": "Studio · clip",
+    "FunPackStudio.source_image": "Studio · source_image (Img2Video anchor)",
+    "FunPackStudio.latent": "Studio · latent",
+    "FunPackLTXAVSceneChainSampler.vae": "Chain Sampler · vae",
+    "LTXVConcatAVLatent.audio_latent": "Concat AV Latent · audio_latent",
+    "LTXVAudioVAEDecode.audio_vae": "Audio VAE Decode · audio_vae",
+}
+
+
+def models_dict(models: Any) -> dict:
+    return models if isinstance(models, dict) else {}
+
+
+def uses_builtin_core(models: Any) -> bool:
+    return not bool(models_dict(models).get("disable_core"))
+
+
+def wiring_locked(models: Any) -> bool:
+    """Guided wiring: built-in core on and full_control off."""
+    m = models_dict(models)
+    return uses_builtin_core(m) and not bool(m.get("full_control"))
+
+
+def allowed_port_ids(role: str, out_type: str, out_name: Optional[str] = None) -> list[str]:
+    rules = ROLE_WIRE_TARGETS.get(role or "", [])
+    out = []
+    for t, name, port in rules:
+        if t != out_type:
+            continue
+        if name is not None and out_name is not None and name != out_name:
+            continue
+        out.append(port)
+    return out
+
+
+def port_label(port_id: str) -> str:
+    return PORT_LABELS.get(port_id, port_id.replace(".", " · "))
+
+
+def validate_port_wire(
+    *,
+    role: str,
+    out_type: str,
+    out_name: str,
+    target: str,
+    models: Any,
+) -> Optional[str]:
+    """Return an error string if the wire is forbidden in guided mode."""
+    if not wiring_locked(models):
+        return None
+    if not target or target in ("global:video", "global:audio"):
+        return None
+    if target.startswith("node:"):
+        return None  # slot-to-slot wiring stays free
+    if not target.startswith("port:"):
+        return f"Unknown wire target '{target}'."
+    port_id = target[5:]
+    allowed = allowed_port_ids(role, out_type, out_name)
+    if not allowed and role in ("custom", "", None):
+        # Legacy slots without a role: allow canonical built-in ports for this output type.
+        for rules in ROLE_WIRE_TARGETS.values():
+            for t, name, port in rules:
+                if t != out_type:
+                    continue
+                if name is not None and name != out_name:
+                    continue
+                allowed.append(port)
+        allowed = list(dict.fromkeys(allowed))
+    if not allowed:
+        return (
+            f"{out_name or out_type} from role '{role}' cannot wire into the built-in pipeline "
+            f"in guided mode — enable Full control in Models to wire it manually."
+        )
+    if port_id not in allowed:
+        labels = ", ".join(port_label(p) for p in allowed)
+        return (
+            f"{out_name} ({out_type}) may only wire to {labels} in guided mode "
+            f"(not {port_label(port_id)}). Enable Full control to wire freely."
+        )
+    return None
+
+
+def validate_models_wiring(models: Any) -> list[str]:
+    """Collect blocking wiring errors for all slots."""
+    m = models_dict(models)
+    if not wiring_locked(m):
+        return []
+    errors: list[str] = []
+    port_owners: dict[str, str] = {}
+    for slot in m.get("slots") or []:
+        role = slot.get("role") or "custom"
+        sid = slot.get("id") or "?"
+        label = slot.get("label") or slot.get("node_class") or sid
+        wires = slot.get("wires") or {}
+        for out_name, raw in wires.items():
+            targets = raw if isinstance(raw, list) else ([raw] if raw else [])
+            out_type = _infer_output_type(slot, out_name)
+            for t in targets:
+                if not t:
+                    continue
+                err = validate_port_wire(role=role, out_type=out_type, out_name=out_name,
+                                         target=t, models=m)
+                if err:
+                    errors.append(f"{label}: {err}")
+                    continue
+                if t.startswith("port:"):
+                    prev = port_owners.get(t[5:])
+                    if prev:
+                        errors.append(
+                            f"{port_label(t[5:])} is already wired from {prev} — "
+                            f"only one source per built-in input in guided mode.")
+                    else:
+                        port_owners[t[5:]] = label
+    return errors
+
+
+def default_slot_extras(role: str) -> dict[str, Any]:
+    """Default wires / input_sources for a newly added slot in guided mode."""
+    extras: dict[str, Any] = {}
+    wires = DEFAULT_WIRES_BY_ROLE.get(role)
+    if wires:
+        extras["wires"] = dict(wires)
+    srcs = DEFAULT_INPUT_SOURCES_BY_ROLE.get(role)
+    if srcs:
+        extras["input_sources"] = dict(srcs)
+    return extras
+
+
+def wiring_rules_payload() -> dict[str, Any]:
+    """Static rules for the Models UI."""
+    role_targets = {}
+    for role, rules in ROLE_WIRE_TARGETS.items():
+        role_targets[role] = [
+            {"type": t, "output_name": n, "port": p, "label": port_label(p)}
+            for t, n, p in rules
+        ]
+    return {
+        "role_targets": role_targets,
+        "default_wires": DEFAULT_WIRES_BY_ROLE,
+        "default_input_sources": DEFAULT_INPUT_SOURCES_BY_ROLE,
+        "port_labels": PORT_LABELS,
+    }
+
+
+def _infer_output_type(slot: dict, out_name: str) -> str:
+    """Best-effort type lookup; frontend validates with full spec."""
+    role = slot.get("role") or ""
+    for t, name, _ in ROLE_WIRE_TARGETS.get(role, []):
+        if name is None or name == out_name:
+            return t
+    # Common Comfy output names
+    upper = (out_name or "").upper()
+    if upper in ("MODEL", "CLIP", "VAE", "LATENT", "IMAGE", "AUDIO"):
+        return upper
+    return "LATENT" if "latent" in (out_name or "").lower() else upper or "*"

movie_editor/frontend/models.js

@@ -14,6 +14,7 @@
   let coreOpen = false;           // built-in pipeline panel expanded?
   let coreProducers = [];          // [{id,type,label}]
   let requirements = [];           // [{id,type,label,required,role_hint,hint}]
+  let wiringRules = {};            // guided wiring rules from /pipeline-ports
   let overlay = null;
   const expanded = new Set();     // slot ids currently expanded (collapsed by default)
   let linkMode = false;           // selecting inputs to bind into one shared control
@@ -29,6 +30,49 @@
   }
   function slotById(id) { return config.slots.find((s) => s.id === id); }
 
+  function pipelineLocked() {
+    return !config.disable_core && !config.full_control;
+  }
+
+  function defaultExtrasForRole(role, cand) {
+    if (!pipelineLocked() || !cand) return {};
+    const extras = {};
+    const wires = wiringRules.default_wires?.[role];
+    if (wires) {
+      extras.wires = {};
+      (cand.outputs || []).forEach((o) => {
+        const t = wires[o.name] || wires[o.type];
+        if (t) extras.wires[o.name] = t;
+      });
+    }
+    const srcs = wiringRules.default_input_sources?.[role];
+    if (srcs) extras.input_sources = { ...srcs };
+    return extras;
+  }
+
+  function allowedDestinations(slot, out) {
+    const all = destinations(slot, out.type);
+    if (!pipelineLocked()) return all;
+    const role = slot.role || "custom";
+    if (role === "custom") {
+      return all.filter((d) => !d.value.startsWith("port:"));
+    }
+    const rules = wiringRules.role_targets?.[role] || [];
+    const allowedPorts = rules
+      .filter((r) => r.type === out.type && (r.output_name == null || r.output_name === out.name))
+      .map((r) => "port:" + r.port);
+    return all.filter((d) => !d.value.startsWith("port:") || allowedPorts.includes(d.value));
+  }
+
+  function allowedSources(slot, ci) {
+    const all = sources(slot, ci.type);
+    if (!pipelineLocked()) return all;
+    if (slot.role === "image_processing" && ci.name === "image" && ci.type === "IMAGE") {
+      return all.filter((s) => !s.value || s.value === "timeline" || s.value.startsWith("out:"));
+    }
+    return all.filter((s) => !s.value || s.value.startsWith("out:") || s.value === "timeline");
+  }
+
   // ── linked inputs (one control drives several node inputs) ────────────────────
   function ensureLinks() { if (!config.links) config.links = []; return config.links; }
   function linkOf(slotId, input) { return (config.links || []).find((l) => (l.members || []).some((m) => m.slotId === slotId && m.input === input)); }
@@ -145,7 +189,7 @@
     if (outs.length && !outs.some((o) => wireTargets(wires[o.name]).some(Boolean))) {
       issues.push({ level: "warn", msg: "No outputs wired — this node feeds nothing." });
     }
-    const dests = (o) => destinations(slot, o.type);
+    const dests = (o) => allowedDestinations(slot, o);
     outs.forEach((o) => {
       wireTargets(wires[o.name]).filter(Boolean).forEach((t) => {
         if (!dests(o).some((d) => d.value === t))
@@ -300,11 +344,15 @@
         const row = el("div", "wire-row");
         row.append(el("span", "wire-out", `${ci.name} (${ci.type})`));
         row.append(el("span", "wire-arrow", "←"));
-        const srcs = sources(slot, ci.type);
+        const srcs = allowedSources(slot, ci);
         const sel = el("select", "wire-select");
         const cur = slot.input_sources[ci.name] || "";
         srcs.forEach((s) => { const o = el("option", null, s.label); o.value = s.value; if (s.value === cur) o.selected = true; sel.append(o); });
         if (cur && !srcs.some((s) => s.value === cur)) { const o = el("option", null, cur + " (missing)"); o.value = cur; o.selected = true; sel.append(o); }
+        if (pipelineLocked() && slot.role === "image_processing" && ci.name === "image" && !cur) {
+          slot.input_sources[ci.name] = "timeline";
+          sel.value = "timeline";
+        }
         sel.onchange = async () => { _setInputSource(slot, ci.name, sel.value); await persist(); render(); };
         row.append(sel);
         sbox.append(row);
@@ -416,12 +464,12 @@
 
     function renderRows() {
       clear(wrap);
-      const dests = destinations(slot, out.type);
+      const dests = allowedDestinations(slot, out);
       targets.forEach((t, i) => {
         const row = el("div", "wire-multi-row");
         const sel = el("select", "wire-select");
         dests.forEach((d) => { const o = el("option", null, d.label); o.value = d.value; if (d.value === t) o.selected = true; sel.append(o); });
-        if (t && !dests.some((d) => d.value === t)) { const o = el("option", null, t + " (missing)"); o.value = t; o.selected = true; sel.append(o); }
+        if (t && !dests.some((d) => d.value === t)) { const o = el("option", null, t + " (not allowed)"); o.value = t; o.selected = true; sel.append(o); }
         sel.onchange = async () => { _setWireTarget(slot, out.name, targets[i], sel.value); targets[i] = sel.value; await persist(); render(); };
         const rm = el("button", "btn ghost tiny wire-rm", "×");
         rm.title = "Remove this wire";
@@ -429,9 +477,12 @@
         row.append(sel, rm);
         wrap.append(row);
       });
-      const add = el("button", "btn ghost tiny wire-add", "+ Add");
-      add.onclick = () => { targets.push(""); renderRows(); };
-      wrap.append(add);
+      const portOpts = dests.filter((d) => d.value.startsWith("port:"));
+      if (!pipelineLocked() || portOpts.length > 1 || (portOpts.length === 1 && !targets.some((t) => t.startsWith("port:")))) {
+        const add = el("button", "btn ghost tiny wire-add", "+ Add");
+        add.onclick = () => { targets.push(""); renderRows(); };
+        wrap.append(add);
+      }
     }
 
     renderRows();
@@ -513,7 +564,13 @@
     addBtn.onclick = async () => {
       if (!curCand) return;
       const role = curRole === "__any__" ? "custom" : curRole;
-      config.slots.push({ id: uid(), role, node_class: curCand.class, inputs: preview._draft || defaultsFor(curCand), wires: {} });
+      const extras = defaultExtrasForRole(role, curCand);
+      config.slots.push({
+        id: uid(), role, node_class: curCand.class,
+        inputs: preview._draft || defaultsFor(curCand),
+        wires: extras.wires || {},
+        input_sources: extras.input_sources || {},
+      });
       await persist();
       typeSel.value = ""; nodeSel.innerHTML = ""; nodeSel.append(new Option("Node…", "")); nodeSel.disabled = true;
       search.style.display = "none"; search.value = "";
@@ -549,6 +606,10 @@
         "Built-in FunPack pipeline is off. Wire your workflow nodes below; FunPack loader requirements do not apply."));
       return sec;
     }
+    if (pipelineLocked()) {
+      sec.append(el("div", "req-hint guided-hint",
+        "Guided wiring is on — loader outputs wire only to their built-in ports. Image processing uses Timeline (scene image) by default. Enable Full control in the built-in pipeline panel to rewire everything manually."));
+    }
     sec.append(el("div", "req-title", "Pipeline requirements"));
 
     if (!requirements.length) {
@@ -737,11 +798,17 @@
         const row = el("div", "wire-row");
         row.append(el("span", "wire-out", `${inp.name} (${inp.type})`));
         row.append(el("span", "wire-arrow", "←"));
-        const sel = el("select", "wire-select");
-        (inp.options || []).forEach((o) => { const op = el("option", null, o.label); op.value = o.value; if (o.value === (inp.value || "")) op.selected = true; sel.append(op); });
-        if (inp.value && !(inp.options || []).some((o) => o.value === inp.value)) { const op = el("option", null, inp.value + " (missing)"); op.value = inp.value; op.selected = true; sel.append(op); }
-        sel.onchange = async () => { setCoreOverride(n.id, inp.name, sel.value); await persist(); coreNodes = (await API.coreGraph(window.Store?.get().project?.id).catch(() => ({}))).nodes || coreNodes; render(); };
-        row.append(sel);
+        if (inp.locked) {
+          const lbl = el("span", "wire-locked", inp.detail || "Fixed built-in link");
+          lbl.title = "Guided wiring — enable Full control to override this input";
+          row.append(lbl);
+        } else {
+          const sel = el("select", "wire-select");
+          (inp.options || []).forEach((o) => { const op = el("option", null, o.label); op.value = o.value; if (o.value === (inp.value || "")) op.selected = true; sel.append(op); });
+          if (inp.value && !(inp.options || []).some((o) => o.value === inp.value)) { const op = el("option", null, inp.value + " (missing)"); op.value = inp.value; op.selected = true; sel.append(op); }
+          sel.onchange = async () => { setCoreOverride(n.id, inp.name, sel.value); await persist(); coreNodes = (await API.coreGraph(window.Store?.get().project?.id).catch(() => ({}))).nodes || coreNodes; render(); };
+          row.append(sel);
+        }
         box.append(row);
       });
       card.append(box);
@@ -769,6 +836,20 @@
     toggle.onclick = () => { coreOpen = !coreOpen; render(); };
     head.append(toggle);
     head.append(el("span", "lib-sub", disabled ? "disabled" : `${coreNodes.length} fixed nodes`));
+    if (!disabled) {
+      const fc = el("button", "btn ghost tiny" + (config.full_control ? " active" : ""), config.full_control ? "Full control" : "Guided wiring");
+      fc.title = config.full_control
+        ? "Full control ON — every core socket can be rewired manually (advanced)."
+        : "Guided wiring ON — loaders wire only to their built-in ports. Click for full manual control.";
+      fc.onclick = async () => {
+        config.full_control = !config.full_control;
+        if (!config.full_control) config.core_overrides = {};
+        await persist();
+        try { coreNodes = (await API.coreGraph(window.Store?.get().project?.id)).nodes || coreNodes; } catch (_) {}
+        render();
+      };
+      head.append(fc);
+    }
     const dis = el("button", "btn ghost tiny" + (disabled ? " danger" : ""), disabled ? "↺ Enable built-in pipeline" : "⏻ Disable built-in pipeline");
     dis.title = disabled
       ? "Re-enable the built-in FunPack Studio → Chain Sampler → decode → combine pipeline."
@@ -781,7 +862,10 @@
       return sec;
     }
     if (coreOpen) {
-      sec.append(el("div", "links-hint", "The fixed FunPack nodes and their wiring. Each input defaults to its built-in source — pick another to re-wire it."));
+      const hint = pipelineLocked()
+        ? "Fixed FunPack path (Studio → Conditioning → Chain Sampler → decode). Loader outputs wire through the slots below — internal links are locked until you enable Full control."
+        : "The fixed FunPack nodes and their wiring. Each input defaults to its built-in source — pick another to re-wire it.";
+      sec.append(el("div", "links-hint", hint));
       coreNodes.forEach((n) => sec.append(coreCard(n)));
     }
     return sec;
@@ -831,7 +915,13 @@
   async function open() {
     if (overlay) return;  // already open — don't stack a second modal (orphans handlers)
     await ensureRoles();
-    try { const pp = await API.pipelinePorts(); ports = pp.ports || []; coreProducers = pp.core_producers || []; requirements = pp.requirements || []; } catch (_) { ports = []; coreProducers = []; requirements = []; }
+    try {
+      const pp = await API.pipelinePorts();
+      ports = pp.ports || [];
+      coreProducers = pp.core_producers || [];
+      requirements = pp.requirements || [];
+      wiringRules = pp.wiring || {};
+    } catch (_) { ports = []; coreProducers = []; requirements = []; wiringRules = {}; }
     try { config = await API.getModels(window.Store?.get().project?.id); } catch (_) { config = { slots: [] }; }
     try { coreNodes = (await API.coreGraph(window.Store?.get().project?.id)).nodes || []; } catch (_) { coreNodes = []; }
     await prewarmSpecs();

movie_editor/frontend/styles.css

@@ -1070,4 +1070,6 @@ label.field.linked input, label.field.linked select { opacity: .6; }
 .wf-nav { display: flex; gap: 8px; justify-content: flex-end; margin-top: 8px; }
 .wf-summary { margin: 0; padding-left: 18px; font-size: 13px; line-height: 1.6; }
 .wf-import-banner { font-size: 12px; color: var(--teal); padding: 8px 10px; background: rgba(80,200,180,.06); border: 1px solid rgba(80,200,180,.2); border-radius: 8px; margin-bottom: 12px; }
+.guided-hint { font-size: 11.5px; color: var(--muted); margin-bottom: 10px; line-height: 1.45; }
+.wire-locked { font-size: 11.5px; color: var(--faint); font-style: italic; padding: 4px 8px; background: var(--ink-2); border-radius: 6px; border: 1px solid var(--line); }
 

movie_editor/server.py

@@ -19,7 +19,7 @@ except Exception:  # pragma: no cover - only available inside ComfyUI
     web = None
     PromptServer = None
 
-from .backend import bridge, builder, config, media, nodes, pipeline_caps, projects, workflow_import
+from .backend import bridge, builder, config, media, nodes, pipeline_caps, pipeline_wiring, projects, workflow_import
 from .backend.nle_effects import zoompan_z_expr
 from .backend.timeline import (
     Project,
@@ -1483,6 +1483,7 @@ if web is not None and PromptServer is not None:
             "ports": nodes.pipeline_ports(oi),
             "core_producers": nodes.core_producers(),
             "requirements": nodes.pipeline_requirements(),
+            "wiring": pipeline_wiring.wiring_rules_payload(),
         })
 
     @routes.get(UI_PREFIX + "/api/image-targets")

movie_editor/tests/test_builder.py

@@ -85,7 +85,7 @@ def test_core_skeleton_links_and_params():
 
 
 def test_explicit_wires_and_autowire():
-    models = {"slots": [
+    models = {"full_control": True, "slots": [
         {"id": "u", "node_class": "UnetLoader", "inputs": {"unet_name": "m.safetensors"},
          "wires": {"MODEL": "port:FunPackStudio.model"}},
         {"id": "c", "node_class": "ClipLoader", "inputs": {}, "wires": {"CLIP": "port:FunPackStudio.clip"}},
@@ -152,7 +152,7 @@ def test_media_injects_loadimage_wired_to_target():
 
 def test_optional_open_ports_do_not_block():
     # loaders + vae + audio handled, no image source: source_image/latent optional -> ok
-    models = {"slots": [
+    models = {"full_control": True, "slots": [
         {"id": "u", "node_class": "UnetLoader", "inputs": {}, "wires": {"MODEL": "port:FunPackStudio.model"}},
         {"id": "c", "node_class": "ClipLoader", "inputs": {}, "wires": {"CLIP": "port:FunPackStudio.clip"}},
         {"id": "v", "node_class": "VaeLoader", "inputs": {}, "wires": {}},

movie_editor/tests/test_pipeline_wiring.py

@@ -0,0 +1,97 @@
+"""Tests for guided vs full-control pipeline wiring rules."""
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from movie_editor.backend import builder, pipeline_wiring  # noqa: E402
+from movie_editor.tests.test_builder import OI, PARAMS  # noqa: E402
+
+
+def test_wiring_locked_by_default():
+    assert pipeline_wiring.wiring_locked({"slots": []}) is True
+    assert pipeline_wiring.wiring_locked({"slots": [], "full_control": True}) is False
+    assert pipeline_wiring.wiring_locked({"slots": [], "disable_core": True}) is False
+
+
+def test_audio_latent_only_to_concat():
+    err = pipeline_wiring.validate_port_wire(
+        role="audio_encoder", out_type="LATENT", out_name="LATENT",
+        target="port:FunPackStudio.latent", models={"slots": []},
+    )
+    assert err is not None
+    assert "Concat AV Latent" in err
+
+    ok = pipeline_wiring.validate_port_wire(
+        role="audio_encoder", out_type="LATENT", out_name="LATENT",
+        target="port:LTXVConcatAVLatent.audio_latent", models={"slots": []},
+    )
+    assert ok is None
+
+
+def test_image_only_to_studio_source_image():
+    err = pipeline_wiring.validate_port_wire(
+        role="image_processing", out_type="IMAGE", out_name="IMAGE",
+        target="port:LTXVConcatAVLatent.audio_latent", models={"slots": []},
+    )
+    assert err is not None
+
+    ok = pipeline_wiring.validate_port_wire(
+        role="image_processing", out_type="IMAGE", out_name="IMAGE",
+        target="port:FunPackStudio.source_image", models={"slots": []},
+    )
+    assert ok is None
+
+
+def test_full_control_allows_any_port():
+    models = {"slots": [], "full_control": True}
+    ok = pipeline_wiring.validate_port_wire(
+        role="audio_encoder", out_type="LATENT", out_name="LATENT",
+        target="port:FunPackStudio.latent", models=models,
+    )
+    assert ok is None
+
+
+def test_duplicate_port_wire_blocked():
+    models = {
+        "slots": [
+            {"id": "a", "role": "unet", "node_class": "UnetLoader", "label": "U1",
+             "wires": {"MODEL": "port:FunPackStudio.model"}},
+            {"id": "b", "role": "unet", "node_class": "UnetLoader", "label": "U2",
+             "wires": {"MODEL": "port:FunPackStudio.model"}},
+        ],
+    }
+    errs = pipeline_wiring.validate_models_wiring(models)
+    assert any("already wired" in e for e in errs)
+
+
+def test_core_overrides_ignored_when_guided():
+    models = {
+        "full_control": False,
+        "core_overrides": {"sampler": {"vae": "out:bad:0"}},
+        "slots": [
+            {"id": "u", "node_class": "UnetLoader", "inputs": {}, "role": "unet",
+             "wires": {"MODEL": "port:FunPackStudio.model"}},
+            {"id": "c", "node_class": "ClipLoader", "inputs": {}, "role": "clip",
+             "wires": {"CLIP": "port:FunPackStudio.clip"}},
+            {"id": "v", "node_class": "VaeLoader", "inputs": {}, "role": "video_vae",
+             "wires": {"VAE": "port:FunPackLTXAVSceneChainSampler.vae"}},
+            {"id": "av", "node_class": "VaeLoader", "inputs": {}, "role": "audio_vae",
+             "wires": {"VAE": "port:LTXVAudioVAEDecode.audio_vae"}},
+            {"id": "ae", "node_class": "ImgProc", "inputs": {}, "role": "audio_encoder",
+             "wires": {"Latent": "port:LTXVConcatAVLatent.audio_latent"}},
+        ],
+    }
+    graph, _report = builder.build(OI, models, PARAMS)
+    assert graph["sampler"]["inputs"]["vae"] == ["slot_v", 0]
+
+
+def test_core_graph_marks_locked_inputs():
+    nodes = builder.core_graph(OI, {"slots": [], "full_control": False})
+    studio = next(n for n in nodes if n["id"] == "studio")
+    sampler_in = next(i for i in studio["inputs"] if i["name"] == "model")
+    assert sampler_in["locked"] is True
+    concat = next(n for n in nodes if n["id"] == "concat")
+    aud = next(i for i in concat["inputs"] if i["name"] == "audio_latent")
+    assert aud["locked"] is True