From 92d5e68bfe1e60709f7262c72541e42e28610575 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Thu, 11 Jun 2026 22:19:35 +0300 Subject: [PATCH] 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 | 39 ++-- movie_editor/backend/pipeline_wiring.py | 198 +++++++++++++++++++++ movie_editor/frontend/models.js | 120 +++++++++++-- movie_editor/frontend/styles.css | 2 + movie_editor/server.py | 3 +- movie_editor/tests/test_builder.py | 4 +- movie_editor/tests/test_pipeline_wiring.py | 97 ++++++++++ 7 files changed, 430 insertions(+), 33 deletions(-) create mode 100644 movie_editor/backend/pipeline_wiring.py create mode 100644 movie_editor/tests/test_pipeline_wiring.py diff --git a/movie_editor/backend/builder.py b/movie_editor/backend/builder.py index a41c1894..49c24e49 100644 --- a/movie_editor/backend/builder.py +++ b/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::" or "out::"). 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] diff --git a/movie_editor/backend/pipeline_wiring.py b/movie_editor/backend/pipeline_wiring.py new file mode 100644 index 00000000..5aff2e21 --- /dev/null +++ b/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 "*" diff --git a/movie_editor/frontend/models.js b/movie_editor/frontend/models.js index 61c53d79..fe055eeb 100644 --- a/movie_editor/frontend/models.js +++ b/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(); diff --git a/movie_editor/frontend/styles.css b/movie_editor/frontend/styles.css index 0606bcce..d3fcd549 100644 --- a/movie_editor/frontend/styles.css +++ b/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); } diff --git a/movie_editor/server.py b/movie_editor/server.py index b9a4cd55..64779181 100644 --- a/movie_editor/server.py +++ b/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") diff --git a/movie_editor/tests/test_builder.py b/movie_editor/tests/test_builder.py index 3dadc36a..3ce4a549 100644 --- a/movie_editor/tests/test_builder.py +++ b/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": {}}, diff --git a/movie_editor/tests/test_pipeline_wiring.py b/movie_editor/tests/test_pipeline_wiring.py new file mode 100644 index 00000000..9a1582ad --- /dev/null +++ b/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