feat: app/ui services — pipeline state (ES module), generate bridge, pipeline follows the project; run/session/client ported as is

This commit is contained in:
digital-garbage
2026-10-03 19:05:01 +03:00
parent bd8ad114fc
commit b44f031c6f
19 changed files with 3690 additions and 4 deletions
+10 -4
View File
@@ -3,6 +3,10 @@ import { build } from "./shell/frame.js";
import { loadFeatures } from "./shell/features.js";
import { createSettings } from "./shell/settings.js";
import { createProject } from "./shell/project.js";
import { createPipelineState } from "./shell/pipeline_state.js";
import { createGenerate } from "./shell/generate.js";
import { linkPipeline } from "./shell/pipeline_link.js";
import { api } from "./shell/api.js";
import { hostFor } from "./shell/mounts.js";
import paths from "./modules.js";
@@ -10,10 +14,12 @@ import paths from "./modules.js";
// different project was opened, "change" for an edit.
const heard = new Set();
const say = (what) => heard.forEach((fn) => { try { fn(what); } catch (err) { console.error(err); } });
const app = {
project: createProject({ onChange: () => say("change"), onOpen: () => say("open"), onError: (err) => console.warn(err) }),
on: (fn) => { heard.add(fn); return () => heard.delete(fn); },
};
const on = (fn) => { heard.add(fn); return () => heard.delete(fn); };
const project = createProject({ onChange: () => say("change"), onOpen: () => say("open"), onError: (err) => console.warn(err) });
const pipeline = createPipelineState(api);
const generate = createGenerate({ pipeline });
linkPipeline({ project, pipeline, onOpen: (fn) => on((what) => { if (what === "open") fn(); }) });
const app = { project, pipeline, generate, api, on };
const frame = build(document.getElementById("app"));
const settings = createSettings({ menubar: hostFor("menubar.menus") });
+26
View File
@@ -0,0 +1,26 @@
// The server's routes, one line each. An error carries the sentence the server gave, not a bare status.
const BASE = "/funpack";
function why(res, payload) {
const p = payload && typeof payload === "object" ? payload : {};
const detail = Array.isArray(p.detail) ? p.detail.map((d) => (d && d.msg) || JSON.stringify(d)).join("; ") : p.detail;
const list = (p.problems && p.problems.length ? p.problems : p.refused) || [];
return [detail, p.why, p.error, list.join("; ")].find((s) => typeof s === "string" && s.trim()) || `HTTP ${res.status}`;
}
export async function call(method, path, body) {
const res = await fetch(BASE + path, {
method, headers: body === undefined ? {} : { "Content-Type": "application/json" },
body: body === undefined ? undefined : JSON.stringify(body),
});
const payload = await res.json().catch(() => null);
if (!res.ok) throw new Error(why(res, payload));
return payload;
}
export const api = {
pipeline: () => call("GET", "/api/pipeline"),
editPipeline: (body) => call("POST", "/api/pipeline", body),
modules: (traits) => call("GET", `/api/modules${traits ? `?traits=${encodeURIComponent(traits)}` : ""}`),
probeFamily: (file) => call("GET", `/api/probe?file=${encodeURIComponent(file)}`),
};
+132
View File
@@ -0,0 +1,132 @@
// This tab's identity to ComfyUI, and its socket.
//
// The id is REMEMBERED, because ComfyUI addresses progress and results to the
// client that queued the run: a reload with a new id hears nothing about the
// generation already in flight -- the run continues, the GPU time is spent, and
// the app sits there saying Ready.
//
// Remembered per TAB, not per browser. ComfyUI keys its socket table by this
// id and evicts the old entry when a new socket arrives with the same one
// (`self.sockets.pop(sid, None)` on connect), so two tabs sharing an id means
// the second tab silently takes the first tab's updates -- the first freezes
// mid-progress while the GPU carries on. Worse, the socket's `finally` pops
// that id again on close, so closing the dead tab disconnects the live one.
// sessionStorage is exactly the right lifetime: it survives a reload of this
// tab, which is what reattach needs, and does not reach another tab.
//
// The one gap: duplicating a tab copies its sessionStorage, so a duplicate
// starts out sharing the id. That is the same collision, from a deliberate act,
// and there is nothing readable on the page that would tell the two apart.
const KEY = "funpack.client-id";
export function clientId() {
try {
const kept = window.sessionStorage.getItem(KEY);
if (kept) return kept;
} catch { /* private mode: a fresh id per load, which still works */ }
const fresh = (globalThis.crypto && globalThis.crypto.randomUUID)
? globalThis.crypto.randomUUID()
: `funpack-${Date.now()}-${Math.random().toString(36).slice(2)}`;
try { window.sessionStorage.setItem(KEY, fresh); } catch { /* not fatal */ }
return fresh;
}
/** ComfyUI's socket, on whatever origin served this page. */
export function connect(id) {
const scheme = window.location.protocol === "https:" ? "wss:" : "ws:";
return new WebSocket(`${scheme}//${window.location.host}/ws?clientId=${encodeURIComponent(id)}`);
}
/**
* What this browser already has in the queue: {promptId, running}, or null.
*
* Asked of ComfyUI's own queue rather than remembered locally: a run this page
* queued may have finished, been cancelled, or been dequeued while the page was
* gone, and a remembered id would reattach the UI to a run that no longer
* exists.
*
* BOTH halves. `/queue` answers with `queue_running` and `queue_pending`, and a
* job sits in the second one from the moment /prompt returns until the worker
* picks it up -- always briefly, and for as long as it takes whenever something
* is running ahead of it. Reading only the running half meant a reload in that
* window found nothing, handed Generate back, and let the user queue the same
* job twice: the first copy then ran unnamed, its messages failing the id check,
* burning time while the UI narrated the second.
*
* Items in both are [number, prompt_id, prompt, extra_data, outputs], and
* extra_data carries the client_id ComfyUI was given when the run was queued.
*/
export async function queuedFor(id, { fetch: doFetch = globalThis.fetch, base = "" } = {}) {
let response;
try {
response = await doFetch(`${base}/queue`);
} catch {
return null; // no queue reachable: nothing to reattach to
}
if (!response.ok) return null;
const body = await response.json().catch(() => ({}));
// Running first: if this browser somehow has both, the one under way is the
// one worth showing. Which half it came from is returned as well, because a
// job waiting its turn is not a job in progress, and telling the user it is
// working when nothing is happening yet is a small lie the UI does not need
// to tell.
for (const [half, running] of [[body.queue_running, true], [body.queue_pending, false]]) {
for (const item of half || []) {
if (!Array.isArray(item) || item.length < 4) continue;
const [, promptId, , extra] = item;
if (extra && extra.client_id === id && promptId) {
// Where the run belongs, if it said so when queued -- carried through
// so a reload mid-generation can still land the result on its scene
// instead of just knowing a run is going. `sceneIds` (plural) is
// separate: a run covering more than one scene (no chain sampler to
// split it into several /prompt calls) still needs the FIRST id for
// the single-scalar attribution every existing caller reads, but
// reattach needs the whole list or every scene past the first
// silently gets no result recorded on reload.
return { promptId, running, sceneId: extra.funpack_scene_id || null,
sceneIds: Array.isArray(extra.funpack_scene_ids) ? extra.funpack_scene_ids : [],
projectId: extra.funpack_project_id || null };
}
}
}
return null;
}
/**
* Which of these runs was this browser's, among ones that have already
* finished.
*
* Only ids seen on this page's own socket are ever asked about, so nothing here
* can resurrect a result from an hour ago: a run that ended while the page was
* loading is the case this exists for, and it is indistinguishable from a lost
* result without it. A history entry's `prompt` is
* [number, prompt_id, prompt, extra_data, outputs].
*/
export async function finishedFor(id, candidates,
{ fetch: doFetch = globalThis.fetch, base = "", onFound } = {}) {
for (const promptId of [...(candidates || [])].reverse()) {
let response;
try {
response = await doFetch(`${base}/history/${encodeURIComponent(promptId)}`);
} catch {
return null; // no history reachable
}
if (!response.ok) continue;
const body = await response.json().catch(() => ({}));
const entry = body && body[promptId];
const prompt = entry && entry.prompt;
if (Array.isArray(prompt) && prompt.length > 3 && prompt[3]
&& prompt[3].client_id === id) {
// `onFound` hands back the extra_data this run was queued with, without
// changing what this function returns everywhere else it is called --
// where the run belongs is the reattach path's business, not this one's.
if (onFound) onFound(prompt[3]);
return promptId;
}
}
return null;
}
+74
View File
@@ -0,0 +1,74 @@
// Queues runs on ComfyUI and reports where they are. A thin shell over run.js / session.js / pipeline.js:
// one real queue per run, each taken to a terminal state before the next starts. Nothing here knows what a
// scene is -- the caller says which scenes a run covers and what inputs it sends.
import { createRun, viewUrl, DONE, FAILED, CANCELLED } from "./run.js";
import { clientId, connect, queuedFor, finishedFor } from "./client.js";
import { wire, waitForTerminal } from "./session.js";
import { check } from "./pipeline.js";
import { wireReferences } from "./reference_wiring.js";
export function createGenerate({ pipeline }) {
const id = clientId();
const run = createRun({ clientId: id, connect });
const listeners = { say: [], warn: [], hold: [], release: [], adopt: [] };
let lastAdopt = null; // replayed to a late listener: it carries the only copy of which scenes a found run belongs to
const fire = (kind, arg) => {
if (kind === "adopt") lastAdopt = arg;
(listeners[kind] || []).forEach((fn) => { try { fn(arg); } catch { /* one listener must not stop the rest */ } });
};
// The hooks session.js calls on its page.transport. Nothing here draws.
const transport = {
generate: { setDisabled() {} }, draw() {},
say: (msg) => fire("say", msg), warn: (msg) => fire("warn", msg),
hold: (msg) => fire("hold", msg), release: (state) => fire("release", state),
};
// Which scenes/project the next run belongs to, read when it actually queues (and sent as extra_data so a
// reload mid-run can find it again). `inputs` is {slotId: {input: value}}.
let ranFor = null, ranForList = [], ranForProject = null, pendingInputs = null;
const session = wire({
run, page: { transport }, check, id, queuedFor, finishedFor,
slots: () => pipeline.slots(),
inputs: async () => pendingInputs || {},
extra: () => (ranForProject && ranFor
? { funpack_scene_id: ranFor, funpack_scene_ids: ranForList, funpack_project_id: ranForProject } : null),
onAdopt: (sceneId, projectId, sceneIds) => {
ranFor = sceneId; ranForProject = projectId;
ranForList = (sceneIds && sceneIds.length) ? sceneIds : (sceneId ? [sceneId] : []);
fire("adopt", { sceneId, projectId, sceneIds: ranForList });
},
});
/** The slot+role pair a pipeline role points at, or null when nothing offers it. */
const slotForRole = (at) => {
for (const slot of pipeline.slots() || []) {
for (const role of slot.roles || []) if (role.at === at) return { slot, role };
}
return null;
};
return {
DONE, FAILED, CANCELLED, run, slotForRole, wireReferences,
viewUrl: (image) => viewUrl(image),
state: () => run.state,
subscribe: (fn) => run.subscribe(fn),
on(kind, fn) {
(listeners[kind] ||= []).push(fn);
if (kind === "adopt" && lastAdopt) { try { fn(lastAdopt); } catch { /* as above */ } }
return () => { listeners[kind] = (listeners[kind] || []).filter((f) => f !== fn); };
},
cancel: () => run.cancel(),
waitForTerminal: () => waitForTerminal(run),
ready: session.ready,
/** Queue one run; resolves true once it is on ComfyUI's queue. */
async generate({ sceneId, sceneIds, projectId, inputs } = {}) {
ranFor = sceneId || null;
ranForList = (sceneIds && sceneIds.length) ? sceneIds : (sceneId ? [sceneId] : []);
ranForProject = projectId || null;
pendingInputs = inputs || null;
return session.generate();
},
};
}
+123
View File
@@ -0,0 +1,123 @@
// What the server would queue, asked for rather than assembled here.
//
// The graph is built by core from the pipeline's slots, so the app never
// composes one: two builders would be two answers to "what runs", and the one
// the user sees would be whichever was written last. The app asks, and gets
// either a prompt or the reasons there is not one.
const ENDPOINT = "/funpack/api/pipeline";
const NODES = "/funpack/api/nodes";
/**
* check({slots, values}) -> { slots, refused, incomplete, notes, queueable, prompt }
*
* `refused` and `incomplete` are different things and stay apart: refused means
* the edit did not happen, incomplete means it did and the pipeline still is
* not ready -- an unset file picker on a fresh install is the normal case, not
* a failure. `notes` is a third thing again: the pipeline is fine and something
* about it is worth saying, such as settings that will not be applied.
*/
export async function check({ fetch: doFetch = globalThis.fetch, slots, values, inputs,
action = "check", slot, node,
input, from_slot: fromSlot, from_output: fromOutput } = {}) {
// `slots` and an action travel together on purpose: a remove is "take this
// one out of THIS pipeline", and sending the action without the pipeline the
// user is looking at would apply it to the server's defaults instead.
const body = { action };
if (slots) body.slots = slots;
// Sent with the pipeline, every time. The server holds no copy of what the
// panels say: two stores of "what the user picked" is two answers to what a
// run used, and the one believed would be whichever was written last.
if (values) body.values = values;
// Values for named inputs, from the controls the pipeline asked the main
// window to show. Only sent when there are some: an empty object is a
// request to override nothing, and saying it every time is noise on the wire.
if (inputs && Object.keys(inputs).length) body.inputs = inputs;
if (slot !== undefined) body.slot = slot;
if (node !== undefined) body.node = node;
// wire's own fields, forwarded the same way -- an unlisted key here is one
// that reaches the server as though it was never sent, which is exactly how
// this dropped `input` silently until a live click on Unwire caught it.
if (input !== undefined) body.input = input;
if (fromSlot !== undefined) body.from_slot = fromSlot;
if (fromOutput !== undefined) body.from_output = fromOutput;
const response = await doFetch(ENDPOINT, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
const payload = await response.json().catch(() => ({}));
if (!response.ok) {
// A 400 here is a malformed REQUEST, which is the app's fault and not the
// user's -- so it is reported as a refusal rather than dressed up as an
// unfinished pipeline.
return {
slots: payload.slots || [],
refused: payload.problems || payload.refused || [`the pipeline could not be read (${response.status})`],
incomplete: [],
notes: [],
queueable: false,
prompt: null,
};
}
return {
slots: payload.slots || [],
refused: payload.refused || [],
incomplete: payload.incomplete || [],
notes: payload.notes || [],
queueable: Boolean(payload.queueable),
prompt: payload.prompt || null,
};
}
/** The pipeline as the server holds it, before anything has been edited. */
export async function load({ fetch: doFetch = globalThis.fetch } = {}) {
const response = await doFetch(ENDPOINT);
if (!response.ok) throw new Error(`the pipeline could not be read (${response.status})`);
const payload = await response.json();
return {
slots: payload.slots || [],
incomplete: payload.incomplete || [],
queueable: Boolean(payload.queueable),
};
}
/**
* presets() -> [{ id, title, module, slots }]
*
* Alternate starting points a module offers beside the one true default --
* loaded into the same editable pipeline `load()` returns, not a different
* kind of thing. An empty list is a real answer (nothing offers one), not
* a failure to fetch.
*/
export async function presets({ fetch: doFetch = globalThis.fetch } = {}) {
const response = await doFetch(`${ENDPOINT}/presets`);
if (!response.ok) throw new Error(`presets could not be read (${response.status})`);
const payload = await response.json();
return payload.presets || [];
}
/**
* describe(classes) -> { [className]: description | null }
*
* A null is an answer, not a gap: the slot points at a node this install does
* not have. The caller asked about it, so it is in the reply saying so.
*/
export async function describe(classes, { fetch: doFetch = globalThis.fetch } = {}) {
const wanted = [...new Set(classes)].filter(Boolean);
if (!wanted.length) return {};
const response = await doFetch(`${NODES}?classes=${encodeURIComponent(wanted.join(","))}`);
if (!response.ok) throw new Error(`those nodes could not be described (${response.status})`);
return (await response.json()).nodes || {};
}
/** Installed nodes matching a query, and how many there were before the cut. */
export async function search(query, { fetch: doFetch = globalThis.fetch, limit = 40 } = {}) {
const response = await doFetch(
`${NODES}/search?q=${encodeURIComponent(query || "")}&limit=${limit}`);
if (!response.ok) throw new Error(`nodes could not be searched (${response.status})`);
const payload = await response.json();
return { nodes: payload.nodes || [], total: payload.total || 0 };
}
+32
View File
@@ -0,0 +1,32 @@
// The pipeline travels with the project: opening a project puts ITS pipeline over the session's, and
// every landed pipeline edit becomes the project's copy (project.models). One writer -- the pipeline state.
export function linkPipeline({ project, pipeline, onOpen, say = () => {} }) {
let adopted = true; // false while the open project's pipeline is not the live one: its copy is not rewritten
let retries = 0;
async function adopt() {
const saved = project.project && project.project.models || {};
let ok = false;
try {
// Bounded: a hung request must not hold the project behind it.
ok = await Promise.race([pipeline.adopt(saved.slots || [], saved.removed, saved.unwired), new Promise((r) => setTimeout(() => r(false), 20000))]);
} catch { ok = false; }
adopted = Boolean(ok && pipeline.slots());
if (adopted) {
retries = 0;
project.setField("models", { ...saved, slots: JSON.parse(JSON.stringify(pipeline.slots())), removed: pipeline.removedIds(), unwired: pipeline.unwiredMap() }, { quiet: true });
} else if (retries++ < 3) {
setTimeout(adopt, 5000);
} else {
say("This project's saved pipeline could not be loaded; the default is in use.");
}
}
pipeline.subscribe((slots) => {
if (!project.project || !adopted) return;
const had = project.project.models || {};
project.setField("models", { ...had, slots: JSON.parse(JSON.stringify(slots || [])), removed: pipeline.removedIds(), unwired: pipeline.unwiredMap() }, { quiet: true });
});
onOpen(adopt);
return { adopt };
}
+535
View File
@@ -0,0 +1,535 @@
// The pipeline's slots, held ONCE per page session and shared by every panel
// that edits them (Engine Settings' module-values, Models & Pipeline's
// node-input edits). Two independent copies of "the current slots", each
// re-fetched on its own panel's mount, is how one panel's save silently
// reverts the other's: GET /api/pipeline is stateless (core/routes.py builds
// the DEFAULT graph fresh on every call, remembering nothing), so a panel
// that fetches on reopen throws away whatever ANY panel already placed --
// its own edits included, per the bug this file was extracted to fix once
// for good rather than have each new panel rediscover it.
export function createPipelineState(API) {
// Loader nodes core knows how to point /api/probe at, and which of their
// inputs names the file. Mirrors the same map api.js's probeFamily() is
// meant to be fed from -- nothing else in this app currently resolves
// "the file the pipeline's model loader points at" from raw slots.
const MODEL_FILE_INPUT = {
FunPackCheckpointLoader: "ckpt_name",
FunPackDiffusionModelLoader: "model_name",
};
function currentModelFile(pipelineSlots) {
const slot = (pipelineSlots || []).find((s) => MODEL_FILE_INPUT[s.node] && s.inputs);
return slot ? slot.inputs[MODEL_FILE_INPUT[slot.node]] || null : null;
}
// Which traits the pipeline's chosen model has, or null when that is not
// knowable (no loader slot, no file chosen, or the probe found nothing).
// null must reach API.modules() as "do not filter" rather than as an
// empty list -- /api/modules?traits= with an empty traits set hides every
// module that requires ANY trait at all, for every model this cheap,
// load-free probe has no opinion about yet (this cheap path only knows
// MiniMax H3's own traits today -- see modules/models/minimax_h3).
async function probeModelTraits(pipelineSlots) {
const filename = currentModelFile(pipelineSlots);
if (!filename) return null;
try {
const data = await API.probeFamily(filename);
return data.detected ? (data.traits || []) : null;
} catch (_) {
return null;
}
}
let modulesById = {};
let offered = null; // the server's default pipeline, as first fetched: what a project's saved copy is laid over
let slots = null; // null = never loaded this session; loaded exactly once
// The model file `modulesById` was last filtered for. A save() whose slots
// still name the same file has nothing new to learn -- re-probing on every
// unrelated edit (a prompt, a slider) would mean one extra round trip per
// keystroke-driven save for no reason.
let lastProbedFile; // undefined = never probed (a pipeline with no model file is null, and still needs its first probe)
async function refreshManifest() {
const file = currentModelFile(slots);
if (file === lastProbedFile) return;
const traits = await probeModelTraits(slots);
const manifest = await API.modules(traits ? traits.join(",") : undefined);
modulesById = {};
(manifest.modules || []).forEach((m) => { modulesById[m.id] = m; });
controlState = manifest.control || controlState;
lastProbedFile = file; // only once it worked: a failed probe is tried again
}
let incomplete = [];
let refused = [];
let queueable = false;
let loading = false;
let loadError = null;
let unwired = {}; // default links the person cut: {slotId: [input, ...]}
let removed = new Set(); // default slots the person took out of this project's pipeline
let adoptGate = null; // a promise while a project's pipeline is being put in
let deferredRemoved, deferredUnwired;
let deferred = null; // a project's saved pipeline waiting for the first successful load
let epoch = 0; // bumped whenever a project's pipeline replaces the live one
let saving = false;
let pending = false;
let pendingBody = null;
let saveNotes = [];
// What a settings_sink already holds, read back out of the live slots --
// place() (core/graph.py) writes a whole values blob as one opaque JSON
// string into the sink's input, never merging. No route tells the client
// WHICH node/input is the real sink (core does not name an implementation,
// by design), so this scans every string input for one that decodes to a
// plain object -- and only accepts a decoded key that names a module this
// session already knows is installed, so an unrelated node's incidentally
// JSON-shaped string can't inject a bogus entry that round-trips forever.
function valuesAlreadyPlaced() {
const known = new Set([...Object.keys(modulesById), "_off"]); // _off: the modules this project turned off
const merged = {};
(slots || []).forEach((slot) => {
Object.values(slot.inputs || {}).forEach((v) => {
if (typeof v !== "string") return;
let parsed;
try { parsed = JSON.parse(v); } catch (_) { return; }
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return;
Object.entries(parsed).forEach(([moduleId, own]) => {
if (!known.has(moduleId)) return;
if (own && typeof own === "object") merged[moduleId] = { ...(merged[moduleId] || {}), ...own };
});
});
});
return merged;
}
// Two panels can mount close enough together that both call this before
// the first fetch settles (`slots` only becomes non-null at the end of a
// load, so a bare `if (slots !== null) return` doesn't cover the window
// while one is already in flight). Every caller awaits the SAME in-flight
// promise instead, so exactly one fetch ever goes out per session, however
// many panels ask.
let loadPromise = null;
async function ensureLoaded() {
if (slots !== null) return; // session-wide: every consumer shares this one load
if (loadPromise) return loadPromise;
loading = true; loadError = null;
loadPromise = (async () => {
try {
// Sequential, not Promise.all: which modules are even compatible
// depends on which model the pipeline's loader slot names, so the
// manifest fetch has to know that before it can ask -- refreshManifest()
// reads `slots`, so it has to be set first.
const pipe = await API.pipeline();
slots = pipe.slots || [];
offered = JSON.parse(JSON.stringify(slots));
incomplete = pipe.incomplete || [];
refused = pipe.refused || [];
queueable = !!pipe.queueable;
await refreshManifest();
} catch (e) {
loadError = e && e.message ? e.message : String(e);
}
loading = false;
loadPromise = null;
if (slots !== null && deferred) { const d = deferred, r = deferredRemoved, u = deferredUnwired; deferred = null; deferredRemoved = undefined; deferredUnwired = undefined; await adopt(d, r, u); }
})();
return loadPromise;
}
// `save()`'s body is one level deeper than a shallow spread can merge
// correctly: `inputs`/`values` are both {outerId: {innerKey: value}}, and
// two queued saves that both touch `inputs` (two different slots, or even
// two fields on the SAME slot from models.js's per-call-fresh objects) had
// the second's `inputs` object silently REPLACE the first's rather than
// combine with it -- the earlier edit vanished with no error. Merges one
// level deeper than Object.assign/spread does, for exactly the two keys
// that are ever shaped this way.
function mergeBodies(a, b) {
const out = { ...(a || {}) };
Object.entries(b || {}).forEach(([key, val]) => {
const prior = out[key];
const bothPlainObjects = val && typeof val === "object" && !Array.isArray(val)
&& prior && typeof prior === "object" && !Array.isArray(prior);
if (!bothPlainObjects) { out[key] = val; return; }
const merged = { ...prior };
Object.entries(val).forEach(([innerId, inner]) => {
const priorInner = merged[innerId];
merged[innerId] = (inner && typeof inner === "object" && !Array.isArray(inner)
&& priorInner && typeof priorInner === "object" && !Array.isArray(priorInner))
? { ...priorInner, ...inner }
: inner;
});
out[key] = merged;
});
return out;
}
// One shared save queue for every caller. `body` is layered onto the
// CURRENT `slots` (the one shared array every panel mutates through its
// own edits) each time a request actually goes out -- so a Models &
// Pipeline node-input edit and an Engine Settings module-value edit, made
// moments apart, both survive: neither's save can send a snapshot that
// predates the other's. See engine_settings.js's original fix for why the
// retry-after-in-flight shape exists (a same-panel double-edit dropped
// one edit before this).
async function save(body) {
if (adoptGate) await adoptGate; // an edit made while a project's pipeline goes in lands on THAT pipeline
pendingBody = mergeBodies(pendingBody, body);
if (saving) { pending = true; return; }
saving = true;
do {
pending = false;
applyGroups();
const toSend = { slots, ...pendingBody };
pendingBody = null;
const mine = epoch;
try {
const res = await API.editPipeline(toSend);
// A different pipeline was put in while this was in flight (another project opened):
// this answer is about the old one and must not overwrite the new.
if (mine !== epoch) continue;
if (res && res.slots) slots = res.slots;
applyGroups();
incomplete = (res && res.incomplete) || [];
refused = (res && res.refused) || [];
queueable = !!(res && res.queueable);
saveNotes = (res && res.notes) || [];
_changed();
// A no-op for the common case (same file as last time) -- see
// refreshManifest()'s own guard. Only a Models & Pipeline edit that
// actually changes the loader's file does a second round trip here.
// Its own try/catch: a failure here is not "could not save" -- the
// edit above already landed -- so it must not overwrite saveNotes
// with a message about the wrong failure.
try {
await refreshManifest();
} catch (e) {
console.warn(`[FunPack] could not refresh modules for the new model: ${e && e.message ? e.message : e}`);
}
} catch (e) {
saveNotes = [`Could not save: ${e && e.message ? e.message : e}`];
}
} while (pending);
saving = false;
}
// Edits already sent toward the server but not yet confirmed by a landed
// `slots` response. valuesAlreadyPlaced() only reflects an edit AFTER its
// save's response arrives -- without this overlay, a second setModuleValue
// call fired before the first one's round-trip resolves would compute its
// own "whole tree" snapshot from data that does not include the first
// edit yet, and silently revert it when its own write lands. Cleared per
// key once valuesAlreadyPlaced() actually agrees with it (see
// _reconcilePending), not wholesale on every resolve -- clearing the
// whole thing on any one save's completion would erase a DIFFERENT edit
// that is still in flight.
let pendingValues = {};
function _reconcilePending() {
const already = valuesAlreadyPlaced();
Object.keys(pendingValues).forEach((moduleId) => {
const pend = pendingValues[moduleId];
const real = already[moduleId] || {};
Object.keys(pend).forEach((name) => {
if (real[name] === pend[name] || (typeof pend[name] === "object" && JSON.stringify(real[name]) === JSON.stringify(pend[name]))) delete pend[name];
});
if (!Object.keys(pend).length) delete pendingValues[moduleId];
});
}
// The full module-settings tree as it stands right now: every installed
// module's own defaults, then whatever the graph already holds wins (same
// recovery valuesAlreadyPlaced() exists for), then any not-yet-confirmed
// edit wins over that. Always computed fresh rather than cached, so two
// editors of the same tree (Engine Settings, the sampler quick-access bar)
// can never hold a stale copy of each other's CONFIRMED state between
// them -- the pending overlay above is the one deliberate exception,
// needed so an in-flight edit from either editor isn't lost by the other.
function currentValues() {
const merged = {};
Object.values(modulesById).forEach((m) => {
const own = {};
Object.entries(m.settings || {}).forEach(([name, spec]) => { own[name] = spec.default; });
if (Object.keys(own).length) merged[m.id] = own;
});
const already = valuesAlreadyPlaced();
Object.entries(already).forEach(([moduleId, own]) => {
merged[moduleId] = { ...(merged[moduleId] || {}), ...own };
});
Object.entries(pendingValues).forEach(([moduleId, own]) => {
merged[moduleId] = { ...(merged[moduleId] || {}), ...own };
});
return merged;
}
// Which modules this project turned off (a reserved entry of the same settings tree), and which
// the server has quarantined after a fault. Neither shows in Engine settings or the quick bar.
let controlState = {};
function isOff(id) {
const off = (currentValues()._off || {}).modules;
return Array.isArray(off) && off.includes(id);
}
function setOff(id, off) {
const cur = new Set(((currentValues()._off || {}).modules) || []);
if (off) cur.add(id); else cur.delete(id);
return setModuleValue("_off", "modules", [...cur]);
}
async function refreshControl() {
try { controlState = (await API.modules()).control || {}; } catch (_) { /* keeps the last answer */ }
return controlState;
}
function isQuarantined(id) { return !!(controlState[id] && controlState[id].quarantine); }
function activeModules() {
return Object.values(modulesById).filter((m) => !isOff(m.id) && !isQuarantined(m.id));
}
// Patch ONE field and save the WHOLE tree. place() (core/graph.py) writes
// values as a single opaque JSON blob with no merge on the server side --
// sending only the module being edited would silently erase every other
// module's settings from that blob. Every write goes through here so that
// rule lives in one place instead of being re-derived per caller.
// pendingValues is updated SYNCHRONOUSLY, before the read below, so a
// second setModuleValue call made while this one is still in flight sees
// this edit too (see currentValues()) -- that is what actually closes the
// race, not save()'s own body-merge queue, which only merges DELTAS and
// has nothing to merge against once a caller is sending a full snapshot.
//
// The returned promise is NOT "this specific edit is server-confirmed" --
// save()'s queue has exactly one caller actually await the network (the
// one that finds `saving` false); every other concurrent caller's save()
// call returns as soon as it queues, before its own write is sent, let
// alone answered. Harmless today: both real callers (engine_settings.js,
// sampler_quickbar.js) only do `.then(render)`, and render() reads
// currentValues(), which already carries this edit optimistically via
// pendingValues regardless of where the real network call is. A future
// caller that needs "my write actually landed" (a per-edit success
// toast, reading saveNotes()/incomplete() right after resolve) cannot
// get that from this promise as written -- it would need save() itself
// reworked to resolve each queued caller at ITS OWN write's landing, not
// at whichever write happens to be in flight when the queue drains.
function setModuleValue(moduleId, name, value) {
pendingValues[moduleId] = { ...(pendingValues[moduleId] || {}), [name]: value };
const values = currentValues();
return save({ values }).then(_reconcilePending);
}
// A change to the pipeline's SHAPE (add / replace / remove / wire / unwire), sent on its own: two
// of these folded into one request would lose the first. Waits for any value edit in flight and
// for a project's pipeline going in. -> {refused: [...]}; empty when it happened.
async function edit(body) {
await ensureLoaded();
if (slots === null) return { refused: ["The pipeline has not loaded: is ComfyUI reachable?"] };
if (adoptGate) await adoptGate;
for (let waited = 0; saving && waited < 10000; waited += 20) await new Promise((r) => setTimeout(r, 20));
if (saving) return { refused: ["A save is still in progress: try again in a moment."] };
const mine = epoch;
saving = true;
let refusedNow = [];
try {
applyGroups();
const res = await API.editPipeline({ slots, ...body });
if (mine !== epoch) return { refused: ["The project was changed while this was being sent."] };
refusedNow = (res && res.refused) || [];
if (res && res.slots && !refusedNow.length) { slots = res.slots; applyGroups(); }
incomplete = (res && res.incomplete) || [];
refused = refusedNow;
queueable = !!(res && res.queueable);
saveNotes = (res && res.notes) || [];
if (!refusedNow.length) {
if (body.action === "remove" && offered.some((s) => s.id === body.slot)) removed.add(body.slot); // the person's own additions just go
if (body.action === "add") (res.slots || []).forEach((s) => removed.delete(s.id));
if (body.action === "unwire" && offered.some((s) => s.id === body.slot)) {
unwired[body.slot] = [...new Set([...(unwired[body.slot] || []), body.input])];
}
if (body.action === "wire" && unwired[body.slot]) {
unwired[body.slot] = unwired[body.slot].filter((k) => k !== body.input);
if (!unwired[body.slot].length) delete unwired[body.slot];
}
if (body.action === "remove") { delete unwired[body.slot]; delete groupEdits[body.slot]; }
// A default slot that was swapped for the same node, or taken out and put back, starts empty:
// none of the default's links may come back over it when the project is opened again.
const fresh = body.action === "replace" ? [body.slot] : body.action === "add" ? (res.slots || []).map((s) => s.id) : [];
fresh.forEach((id) => {
const def = offered.find((s) => s.id === id);
if (!def) return;
const links = Object.keys(def.inputs || {}).filter((k) => Array.isArray(def.inputs[k]));
if (links.length) unwired[id] = [...new Set([...(unwired[id] || []), ...links])];
});
try { await refreshManifest(); } catch (_) { /* the next save retries it */ }
_changed();
}
} catch (e) {
refusedNow = [e && e.message ? e.message : String(e)];
} finally {
saving = false;
}
if (pending) save({}); // a value edit queued behind this one
return { refused: refusedNow };
}
// The pipeline as it stands, to come back to: Models & Pipeline's Cancel puts this back.
function snapshot() {
if (slots === null) return null;
return { slots: JSON.parse(JSON.stringify(slots)), removed: [...removed], unwired: JSON.parse(JSON.stringify(unwired)) };
}
// -> {refused: [...]}; empty when the snapshot is back. Announced as a change, so the project follows.
async function restore(snap) {
if (!snap || !Array.isArray(snap.slots)) return { refused: ["Nothing to go back to."] };
if (adoptGate) await adoptGate;
for (let waited = 0; saving && waited < 10000; waited += 20) await new Promise((r) => setTimeout(r, 20));
if (saving) return { refused: ["A save is still in progress: try again in a moment."] };
const mine = epoch;
saving = true;
let refusedNow = [];
try {
const res = await API.editPipeline({ slots: JSON.parse(JSON.stringify(snap.slots)) });
if (mine !== epoch) return { refused: ["The project was changed while this was being sent."] };
refusedNow = (res && res.refused) || [];
if (res && res.slots && !refusedNow.length) {
slots = res.slots;
removed = new Set(snap.removed || []);
unwired = JSON.parse(JSON.stringify(snap.unwired || {}));
groupEdits = {}; pendingValues = {}; pendingBody = null; pending = false;
incomplete = res.incomplete || []; refused = []; queueable = !!res.queueable;
saveNotes = res.notes || [];
try { await refreshManifest(); } catch (_) { /* the next save retries it */ }
_changed();
}
} catch (e) {
refusedNow = [e && e.message ? e.message : String(e)];
} finally {
saving = false;
}
return { refused: refusedNow };
}
// A slot's group is the person's own to name (any node in any group).
// Kept until it has landed: a save already in flight answers with the slots as they were when
// it left, and would otherwise put the old group back.
let groupEdits = {};
function applyGroups() {
(slots || []).forEach((s) => {
if (!(s.id in groupEdits)) return;
if (groupEdits[s.id]) s.group = groupEdits[s.id]; else delete s.group;
});
}
function setGroup(slotId, group) {
if (!(slots || []).some((s) => s.id === slotId)) return Promise.resolve();
groupEdits[slotId] = String(group || "").trim();
applyGroups();
const mine = groupEdits[slotId];
return save({}).then(() => { if (groupEdits[slotId] === mine) delete groupEdits[slotId]; });
}
// Whoever keeps the pipeline with the project hears every landed edit.
const listeners = new Set();
function _changed() { listeners.forEach((fn) => { try { fn(slots); } catch (e) { console.error(e); } }); }
// Put a project's saved pipeline over the one the server offers: the person's inputs (model
// files, node values, the module-settings blob), group and bypass win on every slot the
// default still has; slots they added are kept. The server's own defaults fill in whatever
// the saved copy predates, so an update's new settings are not lost to an old project.
// Not announced as a change: it IS the project's own copy.
// True when the project's pipeline is in (or it had none to put in); false when it could not be.
async function adopt(saved, removedIds, unwiredMap) {
let open;
const gate = new Promise((r) => { open = r; });
adoptGate = gate; // before anything awaits: an edit right behind this waits too
try { await ensureLoaded(); return await _adopt(saved, removedIds, unwiredMap); } finally { if (adoptGate === gate) adoptGate = null; open(); }
}
async function _adopt(saved, removedIds, unwiredMap) {
// Not loaded (ComfyUI unreachable): remember what the project holds and put it in the moment a
// load succeeds, so the default is never what the next edit is built on.
const hadSomething = (Array.isArray(saved) && saved.length) || (Array.isArray(removedIds) && removedIds.length);
if (slots === null) { if (hadSomething) { deferred = saved; deferredRemoved = removedIds; deferredUnwired = unwiredMap; } return !hadSomething; }
if (!hadSomething) return true;
if (!Array.isArray(saved)) saved = [];
// A saved slot is only what the server would accept: a project file outlives the code that
// wrote it (v4 files name the node differently), and one bad slot refuses every later edit.
const sound = (s) => s && typeof s.id === "string" && typeof s.node === "string"
&& s.inputs && typeof s.inputs === "object" && !Array.isArray(s.inputs)
&& (s.group === undefined || (typeof s.group === "string" && s.group.trim()))
&& (s.roles === undefined || Array.isArray(s.roles));
const byId = new Map(saved.filter(sound).map((s) => [s.id, s]));
const have = new Set(offered.map((s) => s.id));
// Slots the person took out stay out; ones whose node they swapped keep their swap.
const gone = new Set(Array.isArray(removedIds) ? removedIds.filter((x) => typeof x === "string") : []);
const lay = (extras) => offered.filter((def) => !gone.has(def.id)).map((def) => {
const mine = byId.get(def.id);
if (!mine) return JSON.parse(JSON.stringify(def));
if (mine.node !== def.node) return JSON.parse(JSON.stringify(mine));
// A default LINK the saved copy lacks was unwired by the person; a default VALUE it lacks is
// one an update added, and fills in.
const inputs = { ...(def.inputs || {}) };
const cut = (unwiredMap || {})[def.id];
Object.keys(inputs).forEach((k) => { if (Array.isArray(cut) && cut.includes(k) && !(k in mine.inputs)) delete inputs[k]; });
const out = { ...def, inputs: { ...inputs, ...mine.inputs } };
if (typeof mine.group === "string" && mine.group) out.group = mine.group;
if (typeof mine.bypassed === "boolean") out.bypassed = mine.bypassed;
return out;
}).concat(extras ? saved.filter((s) => sound(s) && !have.has(s.id)) : []);
// Replaces whatever pipeline was live: nothing queued for the old one may land on the new.
const mine = ++epoch;
pendingValues = {}; // the new project's values win, not an edit made to the old one
removed = new Set(gone);
unwired = {};
Object.entries(unwiredMap && typeof unwiredMap === "object" ? unwiredMap : {}).forEach(([id, list]) => {
if (Array.isArray(list)) unwired[id] = list.filter((x) => typeof x === "string");
});
groupEdits = {};
pendingBody = null; pending = false;
for (let waited = 0; saving && waited < 10000; waited += 20) await new Promise((r) => setTimeout(r, 20));
for (const extras of [true, false]) {
try {
const res = await API.editPipeline({ slots: lay(extras) });
if (mine !== epoch) return true; // a later project was opened meanwhile: not this one's to report
if (res && res.slots && !(res.refused || []).length) {
slots = res.slots;
incomplete = res.incomplete || []; refused = []; queueable = !!res.queueable;
saveNotes = res.notes || [];
// The modules offered depend on the model this pipeline names, which may differ from the
// previous project's: refresh, or the next settings edit drops this project's modules.
try { await refreshManifest(); } catch (_) { /* the next save retries it */ }
return true;
}
} catch (_) { /* this layering was refused; try the next */ }
}
// Nothing saved was accepted (a failed request, or a file the server refuses): the default
// runs, said, and the caller must NOT save it over the project's own copy.
slots = JSON.parse(JSON.stringify(offered));
saveNotes = ["This project's saved pipeline could not be loaded, so the default is in use and edits are not being saved."];
return false;
}
// Every slot's own values as they are NOW, as per-run overrides: a Generate is several runs, and the
// settings are global to it -- an edit made while it runs reaches the next Generate, never half of
// this one. Inputs another node feeds (arrays) are left to that node.
function frozenInputs() {
const out = {};
(slots || []).forEach((s) => {
const own = {};
Object.entries(s.inputs || {}).forEach(([k, v]) => { if (v !== undefined && !Array.isArray(v)) own[k] = JSON.parse(JSON.stringify(v)); });
out[s.id] = own;
});
return out;
}
return {
frozenInputs,
ensureLoaded, save, edit, restore, snapshot, setGroup, adopt, subscribe: (fn) => { listeners.add(fn); return () => listeners.delete(fn); }, valuesAlreadyPlaced, currentValues, setModuleValue,
modulesById: () => modulesById,
activeModules, isOff, setOff, refreshControl, control: () => controlState,
removedIds: () => [...removed],
unwiredMap: () => JSON.parse(JSON.stringify(unwired)),
slots: () => slots,
incomplete: () => incomplete,
refused: () => refused,
queueable: () => queueable,
loading: () => loading,
loadError: () => loadError,
saving: () => saving,
saveNotes: () => saveNotes,
};
}
+10
View File
@@ -295,6 +295,16 @@ export function createProject({ onChange, onError, onOpen } = {}) {
changed();
},
/** Any project-level field. `quiet` is for what the app maintains by itself (the pipeline's copy):
* saved, but not an edit a person made, so not an undo step. */
setField(key, value, { quiet = false } = {}) {
if (!project) return;
if (!quiet) remember();
project[key] = value;
scheduleSave();
if (!quiet) changed();
},
/** The negative prompt, whichever scene is current -- a project-level
* choice, the same as video, not a per-scene one. */
get negative() { return (project && project.negative) || ""; },
+49
View File
@@ -0,0 +1,49 @@
// Turns a scene's picked reference-media ids into slot overrides for whatever
// pipeline is current -- pulled out of boot.js so it can be tested without
// booting the whole app (boot.js runs start() on import).
/**
* wireReferences(mediaIds, slots, reservedSlotIds) -> { overrides, unwired }
*
* `overrides` is keyed by slot id, merge these into the raw input overrides
* sent with a run. `unwired` counts references that did not make it into the
* run -- the pipeline has fewer "assets.reference_N" roles than the scene has
* references (e.g. added under a bigger preset, then the preset was swapped
* for a smaller one), two references' roles resolve to the same destination,
* or a reference's slot is one the caller already assigned to something else
* (`reservedSlotIds` -- e.g. a slot that also carries the scene's
* `assets.source_image` role). Any of these silently overwriting or dropping
* a reference would look like it was used; the caller is the one place that
* knows this happened and must say so.
*/
export function wireReferences(mediaIds, slots, reservedSlotIds) {
const pool = slots || [];
const overrides = {};
const claimed = new Set(reservedSlotIds || []);
let unwired = 0;
mediaIds.forEach((mediaId, i) => {
const at = `assets.reference_${i + 1}`;
const slot = pool.find((s) => (s.roles || []).some((r) => r.at === at));
if (!slot) { unwired += 1; return; }
// The wire's target is the role's own data (`wireTo`), never a slot id
// this file would otherwise have to know: the preset declares where its
// own reference lands, the same way it declares everything else.
const role = slot.roles.find((r) => r.at === at);
// A destination -- the slot's own media_id input, or wherever `wireTo`
// points -- can hold one value. Two references claiming the SAME slot
// (a preset with two reference roles on one slot) or the SAME wireTo
// target (two different slots both wired into one downstream input) are
// the same failure either way: the second silently overwrites the first
// unless it is counted unwired instead.
const dest = role && role.wireTo ? `${role.wireTo.slot}.${role.wireTo.input}` : slot.id;
if (claimed.has(slot.id) || claimed.has(dest)) { unwired += 1; return; }
claimed.add(slot.id);
claimed.add(dest);
overrides[slot.id] = { ...overrides[slot.id], media_id: mediaId };
if (role && role.wireTo) {
overrides[role.wireTo.slot] = { ...overrides[role.wireTo.slot],
[role.wireTo.input]: [slot.id, 0] };
}
});
return { overrides, unwired };
}
+380
View File
@@ -0,0 +1,380 @@
// A generation, from Generate to a picture on screen.
//
// ComfyUI's own queue and its own socket, not a second one alongside them.
// FunPack has no queue of its own and must not grow one: two things tracking
// what is running is two things that can disagree, and the one the user
// believes is whichever spoke last. So this speaks to /prompt, /interrupt and
// /ws exactly as any ComfyUI client does, and holds only what the UI needs to
// draw -- which is a view of the server's state and never a copy of it.
//
// Everything the outside world provides is injected: `fetch`, `connect`, the
// client id. A run can then be driven end to end in a test without a server,
// and the test exercises this file rather than a mock of it.
export const IDLE = "idle";
export const QUEUED = "queued";
export const RUNNING = "running";
export const DONE = "done";
export const FAILED = "failed";
export const CANCELLED = "cancelled";
/** The socket, which is a different question from what a run is doing. */
export const OFFLINE = "offline"; // not opened yet
export const LIVE = "live";
export const LOST = "lost"; // dropped, and trying again
/** How long to wait before the next attempt: 1s, 2s, 4s ... capped at 30. */
export const backoff = (attempt) => Math.min(30_000, 1000 * 2 ** Math.max(0, attempt));
/** Where ComfyUI serves a produced file from. */
export function viewUrl(image, base = "") {
const q = new URLSearchParams({
filename: image.filename || "",
subfolder: image.subfolder || "",
type: image.type || "output",
});
return `${base}/view?${q}`;
}
export function createRun({
fetch: doFetch = globalThis.fetch,
connect,
clientId,
base = "",
} = {}) {
const listeners = new Set();
let state = {
phase: IDLE, promptId: null, progress: null, images: [], audio: [],
error: null, node: null,
// Whether this page can still hear ComfyUI at all. Part of the run state
// rather than a second thing to subscribe to: one state object is what
// stops two parts of the UI disagreeing about what is happening.
connection: OFFLINE,
};
const emit = (next) => {
state = { ...state, ...next };
for (const fn of listeners) {
try { fn(state); } catch { /* one listener must not stop the others */ }
}
};
// Only messages carrying OUR prompt id are acted on. ComfyUI broadcasts some
// of them (an interrupt is broadcast to every client), so a second tab
// generating would otherwise drive this one's progress bar and, worse, mark
// this run finished when it was somebody else's that ended.
//
// Strict about the id being KNOWN, which it is not for the length of the
// /prompt round trip. Treating "no id yet" as "everything is mine" adopted a
// stranger's result during that window -- this run went to Done, showing
// somebody else's picture, before it had started.
const mine = (data) => Boolean(state.promptId) && Boolean(data)
&& data.prompt_id === state.promptId;
// Messages that arrived before the id did. Ours cannot be told apart from
// anyone else's yet, so they are kept and sorted out once the id is known --
// dropping them would lose a run that finished inside the round trip.
// Bounded, because on a busy shared server this fills with other people's
// traffic and nothing here needs the older end of it.
const PENDING = 200;
let pending = [];
function settle(promptId) {
emit({ promptId });
const held = pending;
pending = [];
for (const message of held) handle(message);
// A cancel asked for while the id was unknown was never sent, because a
// null prompt_id is a GLOBAL interrupt at ComfyUI: it stops whatever any
// client is running. Now that the run has a name, it can be stopped by it.
if (cancelWanted) sendCancel();
}
function handle(message) {
const { type, data } = message || {};
if (!type) return;
// Not yet named: hold it rather than guess whose it is.
//
// Whenever the id is unknown, not only while queueing. A page that reloads
// during a generation opens its socket first and learns which run is its
// own a moment later, and everything arriving in between was being dropped
// -- including the whole finish of a run that ended inside that moment,
// which then left the app at Ready with the result lost.
if (!state.promptId) {
pending.push(message);
if (pending.length > PENDING) pending.shift();
return;
}
switch (type) {
case "execution_start":
if (mine(data)) emit({ phase: RUNNING, progress: null });
break;
case "progress_state": {
if (!mine(data)) break;
// One entry per active node; the run's progress is the node that is
// actually working. Summing them would report a percentage of a total
// nobody can know before the graph has run.
const nodes = Object.values(data.nodes || {});
const active = nodes.find((n) => n.state === "running") || nodes[nodes.length - 1];
if (active && active.max) {
emit({ phase: RUNNING, node: active.display_node_id ?? active.node_id,
progress: { value: active.value || 0, max: active.max } });
}
break;
}
case "executed": {
if (!mine(data)) break;
const output = data.output || {};
// Appended, not replaced: a graph may save more than once, and the
// second save is not a correction of the first.
if (output.images) emit({ images: [...state.images, ...output.images] });
if (output.audio) emit({ audio: [...state.audio, ...output.audio] });
break;
}
case "execution_success":
if (mine(data)) emit({ phase: DONE, progress: null, node: null });
break;
case "execution_error":
if (!mine(data)) break;
emit({
phase: FAILED, progress: null,
// ComfyUI's own words. Rewriting them would lose the one thing that
// says which node and why.
error: {
node: data.node_type || data.node_id,
message: data.exception_message || "the run failed",
traceback: data.traceback || null,
},
});
break;
case "execution_interrupted":
// Broadcast, so the prompt id is the only way to know it was ours.
if (data && data.prompt_id === state.promptId) {
emit({ phase: CANCELLED, progress: null, node: null });
}
break;
default:
break; // not ours to interpret
}
}
// A cancel asked for before the run had a name.
let cancelWanted = false;
async function sendCancel() {
cancelWanted = false;
const id = state.promptId;
// ComfyUI's own state-agnostic cancel: it interrupts a job that is running
// and dequeues one that is only pending, atomically, and answers
// {"cancelled": false} for an id that has already finished.
//
// /interrupt alone was not enough and looked like it was. It only consults
// the RUNNING half of the queue, so cancelling a job waiting behind another
// one returned 200, logged "not currently running, skipping interrupt", and
// let the job run in full when its turn came -- a cancel that did nothing
// and said nothing, which is worse than a cancel that fails.
let response;
try {
response = await doFetch(`${base}/api/jobs/${encodeURIComponent(id)}/cancel`, {
method: "POST",
headers: { "Content-Type": "application/json" },
});
} catch {
response = null;
}
if (response && response.ok) {
// A job that was only PENDING is dequeued and never runs, so no
// execution_interrupted is ever sent -- nothing would arrive to move the
// UI off "Queued", and the run would look stuck forever after a cancel
// that worked. A RUNNING job is left to the socket, which says when it
// actually stopped rather than when it was asked to.
const body = await response.json().catch(() => ({}));
if (body && body.cancelled && state.phase === QUEUED) {
emit({ phase: CANCELLED, progress: null, node: null });
}
return response;
}
// An older ComfyUI without that route. Interrupting is all it can do, and
// it is right for the case people hit most: a run that is under way.
return doFetch(`${base}/interrupt`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ prompt_id: id }),
});
}
let socket = null;
let attempt = 0;
let waiting = null;
// A CLOSED socket is still an object. "Already listening" therefore cannot be
// "there is a socket": every drop -- a laptop sleeping, a tunnel to a rented
// box hiccuping, ComfyUI restarting -- was permanent for that tab, and
// silent. Progress and results simply stopped arriving while the GPU carried
// on, and only a manual reload brought them back.
//
// readyState is undefined on a stand-in socket, and a stand-in is a live one.
const alive = (s) => Boolean(s) && (s.readyState === undefined || s.readyState <= 1);
function listen() {
if (typeof connect !== "function") return null;
if (alive(socket)) return socket;
if (waiting) { clearTimeout(waiting); waiting = null; }
socket = connect(clientId);
if (!socket) return null;
socket.addEventListener("open", () => { attempt = 0; emit({ connection: LIVE }); });
socket.addEventListener("message", (event) => {
// Binary frames are previews, which nothing here consumes yet. A parse
// failure must not take the socket down with it.
if (typeof event.data !== "string") return;
try { handle(JSON.parse(event.data)); } catch { /* not for us */ }
});
// close AND error: a refused connection fires error without ever closing on
// some browsers, and a dropped one closes without an error on others.
const dropped = (which) => () => {
if (which !== socket) return; // an older socket letting go
emit({ connection: LOST });
retry();
};
socket.addEventListener("close", dropped(socket));
socket.addEventListener("error", dropped(socket));
return socket;
}
function retry() {
if (waiting) return;
const wait = backoff(attempt);
attempt += 1;
waiting = setTimeout(() => { waiting = null; listen(); }, wait);
// Not a reason to keep a process alive: under a test runner a pending timer
// is the difference between a suite that finishes and one that hangs.
if (waiting && typeof waiting.unref === "function") waiting.unref();
}
return {
get state() { return { ...state }; },
/** Prompt ids seen on the socket while this page had no run of its own.
* Whoever is reattaching can ask ComfyUI which of them belongs here. */
seen: () => [...new Set(pending.map((m) => m && m.data && m.data.prompt_id)
.filter(Boolean))],
subscribe(fn) {
listeners.add(fn);
// Guarded like every later delivery. Handing a subscriber the current
// state is a delivery too, and doing it bare meant one bad listener took
// down whoever was wiring the UI up -- at startup, before anything is on
// screen to say so.
try { fn(state); } catch { /* the same rule as emit */ }
return () => listeners.delete(fn);
},
handle, // for tests and for a socket owned elsewhere
listen,
/** Stop trying. Only a page being torn down wants this. */
stopListening() {
if (waiting) { clearTimeout(waiting); waiting = null; }
if (socket && typeof socket.close === "function") socket.close();
socket = null;
},
/**
* Queue a prompt. Resolves with the prompt id, or throws with the reason.
*
* `extra`, when given, rides in `extra_data` alongside `client_id` --
* ComfyUI stores and echoes it back on `/queue` and `/history`, which is
* the only way a reload can learn what a run in flight belongs to.
*/
async start(prompt, { extra } = {}) {
listen();
pending = [];
cancelWanted = false;
emit({ phase: QUEUED, promptId: null, progress: null, images: [], audio: [],
error: null, node: null });
let response;
try {
response = await doFetch(`${base}/prompt`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(extra ? { prompt, client_id: clientId, extra_data: extra }
: { prompt, client_id: clientId }),
});
} catch (err) {
// The queue could not be reached at all -- no ComfyUI, or the dev
// server, which serves the app and has no queue behind it.
emit({ phase: FAILED, error: { message: `the queue could not be reached: ${err.message}` } });
throw err;
}
const body = await response.json().catch(() => ({}));
if (!response.ok) {
// ComfyUI refuses a bad graph here, BEFORE anything loads a model, and
// says which node and why. That is the message worth showing.
const detail = body.error && (body.error.message || body.error.type);
emit({ phase: FAILED, error: {
message: detail || `the queue refused this graph (${response.status})`,
nodes: body.node_errors || null,
} });
throw new Error(detail || `queue refused: ${response.status}`);
}
settle(body.prompt_id || null);
return body.prompt_id;
},
/** Ask for the run to stop. */
async cancel() {
if (state.phase !== QUEUED && state.phase !== RUNNING) return false;
if (!state.promptId) {
// Nothing to name yet. Remembered rather than sent: ComfyUI reads a
// missing prompt_id as "interrupt whatever is running", which on a
// shared box is someone else's generation, and on this one is not
// necessarily the run the user is looking at.
cancelWanted = true;
return true;
}
await sendCancel();
// Not marked cancelled here: the server says when it actually stopped,
// and claiming it early is how a UI ends up showing "cancelled" over a
// run that is still burning GPU time.
return true;
},
/**
* Take over a run already in flight -- after a reload, the generation the
* previous page load queued. The id comes from ComfyUI's own queue, so this
* adopts a real run rather than assuming one.
*/
adopt(promptId, { running = true } = {}) {
if (!promptId) return false;
listen();
// A job waiting its turn is queued, not running. Saying "working" over a
// run that has not started is the same small lie as saying "cancelled"
// over one that has not stopped.
emit({ phase: running ? RUNNING : QUEUED, images: [], audio: [],
error: null, node: null });
settle(promptId);
return true;
},
images: () => state.images.map((image) => viewUrl(image, base)),
reset() {
pending = [];
cancelWanted = false;
emit({ phase: IDLE, promptId: null, progress: null, images: [],
audio: [], error: null, node: null });
},
};
}
+237
View File
@@ -0,0 +1,237 @@
// The two things the page does around a run: start one, and take back over one
// that was already going.
//
// Kept out of boot.js because boot.js runs on import and needs a document, so
// nothing in it could be driven by a test -- and both of the faults that lived
// here were ordering faults in exactly that untested wiring: a button still
// live during a round trip, and a socket listening before anyone knew which run
// was ours.
import { DONE, FAILED, CANCELLED } from "./run.js";
const TERMINAL = new Set([DONE, FAILED, CANCELLED]);
/**
* Resolves with the phase a run ends at -- done, failed, or cancelled.
*
* Subscribe FIRST, before whatever starts the run: subscribing after risks
* missing the very transition being waited for. subscribe() delivers the
* CURRENT state immediately, which is not a new transition and is not what
* this is waiting for -- if a page reload adopts a run that already finished,
* that first delivery would resolve this before the caller's own "start"
* step has even run. Only a state delivered AFTER this call counts.
*
* The returned promise carries its own `.cancel()`: subscribing before the
* caller knows whether a run will actually start is the whole point (see
* above), but a start attempt that gets refused before it ever reaches
* run.start() -- an incomplete pipeline, a queue that says no -- never
* transitions the run at all, so nothing would ever resolve this and the
* subscription would sit on `run` forever. Call `.cancel()` when the attempt
* this was waiting on turns out never to have started.
*/
export function waitForTerminal(run) {
let unsubscribe = () => {};
const promise = new Promise((resolve) => {
let first = true;
unsubscribe = run.subscribe((state) => {
if (first) { first = false; return; }
if (TERMINAL.has(state.phase)) { unsubscribe(); resolve(state.phase); }
});
});
promise.cancel = () => unsubscribe();
return promise;
}
/**
* onGenerate: ask what to queue, then queue it, once at a time.
*
* The button only goes dead when the RUN reaches "queued", and reaching it takes
* a round trip to ask what to queue. For the whole of that round trip the button
* was live, so two clicks each posted their own /prompt: two jobs burning GPU
* time, with the UI following one of them.
*/
export function createGenerator({ run, transport, check, ready, slots, values, inputs, extra }) {
let starting = false;
return async function onGenerate() {
// Whatever the page is still working out about runs already in progress
// finishes first. On a reload during a generation the button is live and
// the phase still reads "idle" for the length of the queue lookup, so a
// click in that window queued a SECOND job and orphaned the first -- the
// adopt that would have found it then refused, because the phase was no
// longer idle, and the original run's messages were discarded from then on
// as belonging to nobody.
if (ready) await ready;
if (starting || run.state.phase === "queued" || run.state.phase === "running") return false;
starting = true;
transport.generate.setDisabled(true);
try {
let plan;
try {
// The pipeline the user is looking at, or -- before anything has been
// edited -- none, which asks the server for its own defaults. Sending
// an empty array instead would be a request to run nothing at all.
const edited = slots ? slots() : null;
plan = await check({
...(edited ? { slots: edited } : {}),
// What the boxes on the main window hold, addressed by slot. Sent
// apart from `slots` on purpose: the pipeline window owns the
// structure and these own one value inside it, so neither has to
// carry the other's. Awaited: boot.js's own `inputs` expands the
// prompt box's text (anchor, shortcuts, $variables, postfix) with a
// server round-trip before a run reads it, and a plain synchronous
// return value awaits to itself unchanged.
...(inputs ? { inputs: await inputs() } : {}),
// What every panel in the app holds. Sent even when empty: "nothing
// is set" is an answer, and the server tells the difference between
// that and a client too old to send any.
values: values ? values() : {},
});
} catch (err) {
transport.say(`The pipeline could not be read: ${err.message}`);
return false;
}
// True about the pipeline whatever happens next, so it is said before the
// run is either started or refused -- and it stays up while the run goes.
transport.warn((plan.notes || [])[0] || null);
if (!plan.queueable || !plan.prompt) {
// The first reason, whichever kind it is. A refusal and an unfinished
// pipeline both stop the run, and both belong next to Generate.
const stopping = [...(plan.refused || []), ...(plan.incomplete || [])];
transport.say(stopping[0] || "This pipeline is not ready to run.");
return false;
}
try {
// Where this run belongs, so a reload mid-generation can still find it
// -- omitted rather than sent empty, since ComfyUI stores whatever
// extra_data it is given verbatim.
const meta = extra ? extra() : null;
await run.start(plan.prompt, meta ? { extra: meta } : {});
return true;
} catch {
return false; // run.state carries the reason
}
} finally {
starting = false;
// draw() owns the button from here; this only undoes the click-time
// disable for the paths that never reached a run at all.
transport.draw(run.state);
}
};
}
/**
* Take back over whatever this tab had running before the page reloaded.
*
* Two questions, because a run can be in two places: still in the queue, or
* already finished. The second is not an edge case -- the socket opens before
* the queue can answer, and a run ending in that moment leaves the queue before
* it is asked about, so without this the result is lost and the app says Ready.
* Only ids this page actually saw on its own socket are asked about, so nothing
* older than this page load can be resurrected.
*/
/**
* Everything the page does on load, in the order it has to happen.
*
* Kept here rather than in boot.js because boot.js runs on import and needs a
* document, so nothing in it can be driven by a test -- and every fault on this
* path so far has been an ORDERING fault in exactly that untested wiring. The
* ordering is the thing under test, so the ordering lives where tests can reach
* it.
*/
export function wire({ run, page, check, id, queuedFor, finishedFor, slots, values, inputs, extra, onAdopt }) {
// One subscription draws everything a run affects. It lives here rather than
// beside it in boot.js because subscribe() delivers the CURRENT state at
// once, and doing that after the button was deliberately disabled handed it
// straight back -- live-looking, and doing nothing when pressed.
run.subscribe((state) => {
page.transport.draw(state);
// Whether ComfyUI is still there rides on the same state object, so it is
// drawn by the same subscription: two things watching the connection is two
// things that can disagree about whether the app is talking to anything.
if (page.connection) page.connection.draw(state);
// Results go to the BIN, and the bin decides what is on screen. Pointing
// the viewer at the newest image from here instead put two things in charge
// of it: every progress message re-showed the newest result, so clicking an
// older one in the bin held for as long as it took the next message to
// arrive.
if (page.bin) page.bin.absorb(state.images);
});
// The socket opens at load, not at Generate. A generation queued before a
// reload keeps running on the server, and a page that only listens once it
// starts one of its own hears nothing about it.
run.listen();
// Not awaited by the caller, but the button waits on it. Until this settles
// nobody can know whether this page already has a run, and starting another
// one on the strength of "phase is idle" is starting one on a guess.
page.transport.hold("Looking for a run already in progress");
const ready = reattach(run, id, { queuedFor, finishedFor, onAdopt }).then((adopted) => {
page.transport.release(run.state);
return adopted;
});
const generate = createGenerator({ run, transport: page.transport, check, ready, slots, values, inputs, extra });
return { ready, generate };
}
/**
* Resolves to `{promptId, sceneId, projectId}` for a run this page adopted, or
* null. `sceneId`/`projectId` are whatever the run was queued with (see
* `run.start`'s `extra`) -- null if it predates that, or was queued some other
* way.
*
* `onAdopt(sceneId, projectId)`, when given, fires SYNCHRONOUSLY, before
* `run.adopt()` -- not after this function returns. A run that already
* finished while the page was reloading can go straight to DONE the moment it
* is adopted (whatever the socket buffered gets replayed inside `adopt()`
* itself), and a listener reading "which scene is this for" only after
* `reattach` resolves would read it one tick too late, after that DONE
* already passed the caller's own run.subscribe with nothing to attach it to.
*/
export async function reattach(run, id, { queuedFor, finishedFor, onAdopt } = {}) {
try {
const queued = await queuedFor(id);
if (queued) {
if (run.state.phase === "idle") {
const sceneId = queued.sceneId || null;
const projectId = queued.projectId || null;
// A caller that queued more than one scene in a single run (no chain
// sampler to split them, so the whole selection goes as one /prompt)
// has nowhere else to recover that list from on reattach -- `sceneId`
// alone would only ever be the first of them. Optional third
// argument: every existing caller takes just the first two and never
// sees this.
const sceneIds = (queued.sceneIds && queued.sceneIds.length) ? queued.sceneIds
: (sceneId ? [sceneId] : []);
if (onAdopt) onAdopt(sceneId, projectId, sceneIds);
run.adopt(queued.promptId, { running: queued.running });
return { promptId: queued.promptId, sceneId, projectId };
}
return null;
}
const seen = run.seen();
if (!seen.length || run.state.phase !== "idle") return null;
let meta = null;
const finished = await finishedFor(id, seen, { onFound: (extra) => { meta = extra; } });
if (finished && run.state.phase === "idle") {
const sceneId = (meta && meta.funpack_scene_id) || null;
const projectId = (meta && meta.funpack_project_id) || null;
const sceneIds = (meta && Array.isArray(meta.funpack_scene_ids) && meta.funpack_scene_ids.length)
? meta.funpack_scene_ids : (sceneId ? [sceneId] : []);
if (onAdopt) onAdopt(sceneId, projectId, sceneIds);
run.adopt(finished);
return { promptId: finished, sceneId, projectId };
}
return null;
} catch {
return null; // nothing to reattach to
}
}
+199
View File
@@ -0,0 +1,199 @@
// Reattaching to a run that outlived the page.
import test from "node:test";
import assert from "node:assert/strict";
import { queuedFor, finishedFor } from "../client.js";
const answering = (body, ok = true) => async () => ({
ok, status: ok ? 200 : 500, json: async () => body,
});
test("the run this browser queued is found by its client id", async () => {
const found = await queuedFor("me", {
fetch: answering({
queue_running: [
[1, "someone-elses", {}, { client_id: "them" }, []],
[2, "mine", {}, { client_id: "me" }, []],
],
queue_pending: [],
}),
});
assert.deepEqual(found, { promptId: "mine", running: true, sceneId: null, sceneIds: [], projectId: null });
});
test("another browser's run is not adopted", async () => {
const promptId = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "theirs", {}, { client_id: "them" }, []]],
queue_pending: [[2, "also-theirs", {}, { client_id: "them" }, []]],
}),
});
assert.equal(promptId, null);
});
test("nothing running means nothing to reattach to", async () => {
assert.equal(await queuedFor("me", {
fetch: answering({ queue_running: [], queue_pending: [] }),
}), null);
});
test("a queue that cannot be reached is not an error the app has to show", async () => {
// The dev server has no queue behind it, which is the ordinary case while
// working on the UI.
const promptId = await queuedFor("me", {
fetch: async () => { throw new TypeError("Failed to fetch"); },
});
assert.equal(promptId, null);
});
test("a malformed queue entry is skipped rather than crashing the load", async () => {
const found = await queuedFor("me", {
fetch: answering({
queue_running: [null, "nonsense", [1], [2, "mine", {}, { client_id: "me" }, []]],
queue_pending: [undefined, {}, [3, null, {}, { client_id: "me" }, []]],
}),
});
assert.deepEqual(found, { promptId: "mine", running: true, sceneId: null, sceneIds: [], projectId: null });
});
test("where a run belongs travels with it, when it said so at queue time", async () => {
const found = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "mine", {}, { client_id: "me", funpack_scene_id: "s1",
funpack_project_id: "p1" }, []]],
queue_pending: [],
}),
});
assert.deepEqual(found, { promptId: "mine", running: true, sceneId: "s1", sceneIds: [], projectId: "p1" });
});
test("a run covering more than one scene carries the whole list, not just the first", async () => {
// No chain sampler to split a multi-scene selection into several /prompt
// calls, so one run can cover several scenes at once -- reattach needs
// every one of them, or a reload only ever recovers the first (see this
// field's own comment in client.js for why sceneId alone is not enough).
const found = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "mine", {}, { client_id: "me", funpack_scene_id: "s1",
funpack_scene_ids: ["s1", "s2", "s3"],
funpack_project_id: "p1" }, []]],
queue_pending: [],
}),
});
assert.deepEqual(found, {
promptId: "mine", running: true, sceneId: "s1", sceneIds: ["s1", "s2", "s3"], projectId: "p1",
});
});
test("a run queued before this existed has nowhere it said it belongs, and says so with null", async () => {
const found = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "mine", {}, { client_id: "me" }, []]],
queue_pending: [],
}),
});
assert.equal(found.sceneId, null);
assert.equal(found.projectId, null);
});
test("a finished run is recognised as this browser's from its history entry", async () => {
const asked = [];
const promptId = await finishedFor("me", ["theirs", "mine"], {
fetch: async (url) => {
asked.push(url);
const id = url.split("/").pop();
const owner = id === "mine" ? "me" : "them";
return { ok: true, status: 200, json: async () => ({
[id]: { prompt: [1, id, {}, { client_id: owner }, []], outputs: {}, status: {} },
}) };
},
});
assert.equal(promptId, "mine");
// Newest first: the run that just ended is the one being looked for.
assert.equal(asked[0], "/history/mine");
});
test("onFound hands back the extra_data of a match, without changing what this returns", async () => {
const found = [];
const promptId = await finishedFor("me", ["mine"], {
fetch: async () => ({ ok: true, status: 200, json: async () => ({
mine: { prompt: [1, "mine", {}, { client_id: "me", funpack_scene_id: "s1",
funpack_project_id: "p1" }, []] },
}) }),
onFound: (extra) => found.push(extra),
});
assert.equal(promptId, "mine");
assert.deepEqual(found, [{ client_id: "me", funpack_scene_id: "s1", funpack_project_id: "p1" }]);
});
test("onFound is never called when nothing matches", async () => {
let called = false;
await finishedFor("me", ["theirs"], {
fetch: async () => ({ ok: true, status: 200, json: async () => ({
theirs: { prompt: [1, "theirs", {}, { client_id: "them" }, []] },
}) }),
onFound: () => { called = true; },
});
assert.equal(called, false);
});
test("a finished run belonging to someone else is not adopted", async () => {
const promptId = await finishedFor("me", ["theirs"], {
fetch: async (url) => ({ ok: true, status: 200, json: async () => ({
theirs: { prompt: [1, "theirs", {}, { client_id: "them" }, []] },
}) }),
});
assert.equal(promptId, null);
});
test("a history entry with nothing in it is skipped, not trusted", async () => {
for (const body of [{}, { mine: {} }, { mine: { prompt: [] } }, { mine: { prompt: [1, "mine", {}] } }]) {
const promptId = await finishedFor("me", ["mine"], {
fetch: async () => ({ ok: true, status: 200, json: async () => body }),
});
assert.equal(promptId, null, JSON.stringify(body));
}
});
test("no history to ask is not an error", async () => {
const promptId = await finishedFor("me", ["mine"], {
fetch: async () => { throw new TypeError("Failed to fetch"); },
});
assert.equal(promptId, null);
});
test("a run still waiting its turn is found too", async () => {
// /queue answers with both halves, and a job sits in the pending one from the
// moment /prompt returns until the worker picks it up -- always briefly, and
// for as long as it takes whenever something is running ahead of it. A reload
// in that window found nothing and let the same job be queued twice.
const promptId = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "theirs", {}, { client_id: "them" }, []]],
queue_pending: [[2, "mine", {}, { client_id: "me" }, []]],
}),
});
assert.deepEqual(promptId, { promptId: "mine", running: false, sceneId: null, sceneIds: [], projectId: null },
"a job waiting its turn was reported as one already under way");
});
test("a run under way is preferred over one still waiting", async () => {
const promptId = await queuedFor("me", {
fetch: answering({
queue_running: [[1, "under-way", {}, { client_id: "me" }, []]],
queue_pending: [[2, "waiting", {}, { client_id: "me" }, []]],
}),
});
assert.deepEqual(promptId, { promptId: "under-way", running: true, sceneId: null, sceneIds: [], projectId: null });
});
test("somebody else's pending run is not adopted either", async () => {
const promptId = await queuedFor("me", {
fetch: answering({
queue_running: [],
queue_pending: [[1, "theirs", {}, { client_id: "them" }, []]],
}),
});
assert.equal(promptId, null);
});
+99
View File
@@ -0,0 +1,99 @@
// The pipeline client: what actually reaches /api/pipeline on the wire.
//
// pipeline_window.test.js injects a FAKE check(), so it can never catch a
// field this module forgets to forward -- which is exactly what happened:
// check()'s destructured signature had no `input`/`from_slot`/`from_output`,
// so a real wire()/unwire() call silently dropped them and the server
// answered "which input to unwire is named by a string, not a NoneType" to a
// button click that looked correct in every jsdom test. Caught live, in a
// browser, not by any suite -- this is the suite that would have caught it.
import test from "node:test";
import assert from "node:assert/strict";
import { check, load, describe, search } from "../pipeline.js";
function fetchRecording(response = { slots: [], refused: [], incomplete: [], queueable: true }) {
const calls = [];
const fetch = async (url, opts) => {
calls.push({ url, opts, body: opts && opts.body ? JSON.parse(opts.body) : undefined });
return { ok: true, json: async () => response };
};
return { fetch, calls };
}
test("wire's own fields all reach the request body", async () => {
const { fetch, calls } = fetchRecording();
await check({
fetch, slots: [{ id: "a", node: "X", inputs: {} }],
action: "wire", slot: "sampler", input: "model", from_slot: "loader", from_output: 0,
});
assert.deepEqual(calls[0].body, {
action: "wire", slots: [{ id: "a", node: "X", inputs: {} }],
slot: "sampler", input: "model", from_slot: "loader", from_output: 0,
});
});
test("unwire's fields reach the request body, including a zero from_output on wire", async () => {
const { fetch, calls } = fetchRecording();
await check({ fetch, action: "unwire", slot: "sampler", input: "model" });
assert.deepEqual(calls[0].body, { action: "unwire", slot: "sampler", input: "model" });
const second = fetchRecording();
await check({
fetch: second.fetch, action: "wire", slot: "a", input: "x",
from_slot: "b", from_output: 0,
});
// 0 is a legal output index and must not be dropped the way `undefined` is.
assert.equal(second.calls[0].body.from_output, 0);
});
test("fields that were not passed are not sent as null or undefined", async () => {
const { fetch, calls } = fetchRecording();
await check({ fetch, action: "check" });
assert.equal("input" in calls[0].body, false);
assert.equal("from_slot" in calls[0].body, false);
assert.equal("from_output" in calls[0].body, false);
assert.equal("slot" in calls[0].body, false);
assert.equal("node" in calls[0].body, false);
});
test("a refused check comes back with the server's reasons, not a thrown error", async () => {
const fetch = async () => ({
ok: false, status: 400,
json: async () => ({ problems: ["which input to unwire is named by a string, not a NoneType"] }),
});
const result = await check({ fetch, action: "unwire", slot: "a" });
assert.equal(result.queueable, false);
assert.deepEqual(result.refused, ["which input to unwire is named by a string, not a NoneType"]);
});
test("load reads the pipeline as it stands", async () => {
const fetch = async () => ({
ok: true, json: async () => ({ slots: [{ id: "a" }], incomplete: ["x needs y"], queueable: false }),
});
const result = await load({ fetch });
assert.deepEqual(result, { slots: [{ id: "a" }], incomplete: ["x needs y"], queueable: false });
});
test("load throws with the status when the server refuses to answer at all", async () => {
const fetch = async () => ({ ok: false, status: 500 });
await assert.rejects(() => load({ fetch }), /500/);
});
test("describe asks nothing for an empty list, and answers by class name", async () => {
let asked = false;
const fetch = async () => { asked = true; return { ok: true, json: async () => ({ nodes: {} }) }; };
assert.deepEqual(await describe([], { fetch }), {});
assert.equal(asked, false);
const real = async (url) => ({
ok: true, json: async () => ({ nodes: { A: { node: "A" }, B: null } }),
});
assert.deepEqual(await describe(["A", "B"], { fetch: real }), { A: { node: "A" }, B: null });
});
test("search answers with nodes and a total even when the server sends neither", async () => {
const fetch = async () => ({ ok: true, json: async () => ({}) });
assert.deepEqual(await search("x", { fetch }), { nodes: [], total: 0 });
});
+273
View File
@@ -0,0 +1,273 @@
import test from "node:test";
import assert from "node:assert";
import { createPipelineState } from "../pipeline_state.js";
let lastApi;
function load(posts, opts = {}) {
const defaults = [
{ id: "model", node: "Loader", group: "Loaders", inputs: { file: "", dtype: "bf16" } },
{ id: "gen", node: "Gen", group: "Sampling", inputs: { steps: 8, model: ["model", 0] } },
];
const API = {
pipeline: async () => ({ slots: JSON.parse(JSON.stringify(defaults)), incomplete: [], refused: [], queueable: true }),
editPipeline: async (body) => {
posts.push(body);
if (opts.delay) await new Promise((r) => setTimeout(r, opts.delay));
const bad = (body.slots || []).findIndex((s) => typeof s.node !== "string");
if (bad >= 0) { const e = new Error(`slot ${bad} has no node`); throw e; } for (const [id, edits] of Object.entries(body.inputs || {})) { const sl = body.slots.find((x) => x.id === id); if (sl) sl.inputs = { ...sl.inputs, ...edits }; }
return { slots: body.slots, incomplete: [], refused: [], queueable: true }; },
modules: async () => ({ modules: [] }),
probeFamily: async () => ({}),
};
lastApi = API;
return createPipelineState(API);
}
test("a project's saved pipeline lays its values over the server's default, keeping what the default added", async () => {
const posts = [];
const PS = load(posts);
await PS.adopt([
{ id: "model", node: "Loader", group: "Mine", inputs: { file: "h3.safetensors" } },
{ id: "gen", node: "OtherNode", inputs: { steps: 99 } }, // a swapped node: the person's swap stays
{ id: "extra", node: "Added", inputs: { x: 1 } }, // the person's own
]);
const by = Object.fromEntries(PS.slots().map((s) => [s.id, s]));
assert.deepStrictEqual(by.model.inputs, { file: "h3.safetensors", dtype: "bf16" });
assert.strictEqual(by.model.group, "Mine");
assert.deepStrictEqual([by.gen.node, by.gen.inputs.steps], ["OtherNode", 99]);
assert.ok(by.extra);
});
test("adopting is the project's own copy, so it is not announced as an edit; a real edit is", async () => {
const posts = [];
const PS = load(posts);
let heard = 0;
PS.subscribe(() => { heard++; });
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "a" } }]);
assert.strictEqual(heard, 0);
await PS.save({ inputs: { model: { file: "b" } } });
assert.strictEqual(heard, 1);
});
test("nothing saved leaves the session's pipeline as it is", async () => {
const PS = load([]);
await PS.ensureLoaded();
const before = JSON.stringify(PS.slots());
await PS.adopt([]);
assert.strictEqual(JSON.stringify(PS.slots()), before);
});
test("a v4-shaped or hostile saved slot cannot brick the pipeline", async () => {
const PS = load([]);
await PS.adopt([
{ id: "model", node: "Loader", inputs: { file: "h3.safetensors" } },
{ id: "abc123", node_class: "OldV4Node", inputs: {} },
{ id: "bad", node: "X", inputs: [1, 2] },
]);
assert.deepStrictEqual(PS.slots().map((s) => s.id), ["model", "gen"]);
assert.strictEqual(PS.slots()[0].inputs.file, "h3.safetensors");
await PS.save({ inputs: { model: { file: "new" } } });
assert.strictEqual(PS.slots()[0].inputs.file, "new");
});
test("an answer about the previous project's pipeline does not overwrite the one just opened", async () => {
const PS = load([], { delay: 30 });
await PS.ensureLoaded();
const inflight = PS.save({ inputs: { model: { file: "A-file" } } });
await new Promise((r) => setTimeout(r, 5));
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "B-file" } }]);
await inflight;
assert.strictEqual(PS.slots()[0].inputs.file, "B-file");
});
test("nothing loaded (ComfyUI unreachable) leaves slots null, never an empty pipeline", async () => {
const PS = load([]);
lastApi.pipeline = async () => { throw new Error("down"); };
const P2 = createPipelineState(lastApi);
await P2.adopt([{ id: "model", node: "Loader", inputs: {} }]);
assert.strictEqual(P2.slots(), null);
});
test("a project's pipeline waits out an unreachable server and goes in on the first successful load", async () => {
const PS = load([]);
const good = lastApi.pipeline;
let up = false;
lastApi.pipeline = async () => { if (!up) throw new Error("down"); return good(); };
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "h3.safetensors" } }]);
assert.strictEqual(PS.slots(), null);
up = true;
await PS.ensureLoaded();
assert.strictEqual(PS.slots().find((s) => s.id === "model").inputs.file, "h3.safetensors");
});
test("adopt says so when the project's pipeline could not go in", async () => {
const PS = load([]);
await PS.ensureLoaded();
lastApi.editPipeline = async () => { throw new Error("tunnel"); };
assert.strictEqual(await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "x" } }]), false);
assert.strictEqual(await PS.adopt([]), true);
});
test("an edit made while a project's pipeline goes in lands on that pipeline", async () => {
const PS = load([], { delay: 30 });
await PS.ensureLoaded();
const adopting = PS.adopt([{ id: "model", node: "Loader", inputs: { file: "B-file", dtype: "fp8" } }]);
const edit = PS.save({ inputs: { model: { dtype: "int8" } } });
await Promise.all([adopting, edit]);
const model = PS.slots().find((s) => s.id === "model");
assert.deepStrictEqual([model.inputs.file, model.inputs.dtype], ["B-file", "int8"]);
});
test("a failed module-list fetch is retried on the next save, not remembered as done", async () => {
const posts = [];
const PS = load(posts);
let calls = 0;
lastApi.modules = async () => { if (++calls === 1) throw new Error("down"); return { modules: [{ id: "m" }] }; };
await PS.ensureLoaded();
assert.deepStrictEqual(Object.keys(PS.modulesById()), []);
await PS.save({ inputs: { model: { file: "z.safetensors" } } });
assert.deepStrictEqual(Object.keys(PS.modulesById()), ["m"]);
});
test("a slot the person removed stays removed when the project is opened again", async () => {
const PS = load([]);
await PS.adopt([{ id: "model", node: "Loader", inputs: {} }], ["gen"]);
assert.deepStrictEqual(PS.slots().map((s) => s.id), ["model"]);
assert.deepStrictEqual(PS.removedIds(), ["gen"]);
});
test("a structural edit is sent on its own and its refusal is handed back", async () => {
const posts = [];
const PS = load(posts);
await PS.ensureLoaded();
lastApi.editPipeline = async (body) => {
posts.push(body);
return { slots: body.slots, refused: ["nope"], incomplete: [], queueable: false };
};
const res = await PS.edit({ action: "remove", slot: "gen" });
assert.deepStrictEqual(res.refused, ["nope"]);
assert.strictEqual(posts.at(-1).action, "remove");
assert.deepStrictEqual(PS.removedIds(), []); // refused: nothing is remembered as removed
});
test("a default link the person unwired stays unwired when the project is opened again", async () => {
const PS = load([]);
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "x" } },
{ id: "gen", node: "Gen", inputs: { steps: 8 } }], [], { gen: ["model"] });
assert.strictEqual("model" in PS.slots().find((s) => s.id === "gen").inputs, false);
assert.deepStrictEqual(PS.unwiredMap(), { gen: ["model"] });
});
test("a default link an older project never had (an update added it) is kept", async () => {
const PS = load([]);
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "x" } },
{ id: "gen", node: "Gen", inputs: { steps: 8 } }]); // nothing recorded as unwired
assert.strictEqual("model" in PS.slots().find((s) => s.id === "gen").inputs, true);
});
test("a group moved while a save is in flight is still moved when it lands", async () => {
const PS = load([], { delay: 30 });
await PS.ensureLoaded();
const inflight = PS.save({ inputs: { model: { file: "z" } } });
await new Promise((r) => setTimeout(r, 5));
await PS.setGroup("gen", "Mine");
await inflight;
assert.strictEqual(PS.slots().find((s) => s.id === "gen").group, "Mine");
});
test("a default slot swapped for the same node, or removed and added back, does not get its links back on reopen", async () => {
const PS = load([]);
await PS.ensureLoaded();
const API = lastApi;
API.editPipeline = async (body) => ({ refused: [], incomplete: [], queueable: true,
slots: body.slots.map((s) => (s.id === "gen" ? { ...s, inputs: {} } : s)) });
await PS.edit({ action: "replace", slot: "gen", node: "Gen" });
assert.deepStrictEqual(PS.unwiredMap(), { gen: ["model"] });
const saved = JSON.parse(JSON.stringify(PS.slots()));
const again = load([]);
await again.adopt(saved, [], PS.unwiredMap());
assert.strictEqual("model" in again.slots().find((s) => s.id === "gen").inputs, false);
});
test("a project with every slot removed does not get the default back", async () => {
const posts = [];
const PS = load(posts);
await PS.adopt([], ["model", "gen"]);
assert.ok(posts.length > 0);
assert.deepStrictEqual(posts.at(-1).slots, []);
});
test("a group change queued behind a save in flight is not re-stamped over a later group", async () => {
const PS = load([], { delay: 20 });
await PS.ensureLoaded();
const inflight = PS.save({ inputs: { model: { file: "z" } } });
const first = PS.setGroup("gen", "Mine");
await Promise.all([inflight, first]);
await PS.setGroup("gen", "Other");
assert.strictEqual(PS.slots().find((s) => s.id === "gen").group, "Other");
});
test("restore puts back the pipeline as snapshotted: slots, removed, unwired, and tells the project", async () => {
const PS = load([]);
await PS.ensureLoaded();
const snap = PS.snapshot();
await PS.save({ inputs: { model: { file: "changed" } } });
lastApi.editPipeline = async (body) => ({ refused: [], incomplete: [], queueable: true, slots: body.slots });
await PS.edit({ action: "remove", slot: "gen" });
assert.deepStrictEqual(PS.removedIds(), ["gen"]);
let heard = 0;
PS.subscribe(() => { heard++; });
const res = await PS.restore(snap);
assert.deepStrictEqual(res.refused, []);
assert.deepStrictEqual(PS.slots(), snap.slots);
assert.deepStrictEqual(PS.removedIds(), []);
assert.strictEqual(heard, 1);
});
test("a refused restore leaves the pipeline as it is", async () => {
const PS = load([]);
await PS.ensureLoaded();
const snap = PS.snapshot();
await PS.save({ inputs: { model: { file: "changed" } } });
lastApi.editPipeline = async () => ({ refused: ["no"], slots: [] });
const res = await PS.restore(snap);
assert.deepStrictEqual(res.refused, ["no"]);
assert.strictEqual(PS.slots().find((s) => s.id === "model").inputs.file, "changed");
});
test("a module switched off is remembered with the settings, listed as off, and can be switched back on", async () => {
const posts = [];
const PS = load(posts);
await PS.ensureLoaded();
lastApi.modules = async () => ({ modules: [{ id: "sharpen", settings: {} }, { id: "alg", settings: {} }], control: {} });
await PS.refreshControl();
await PS.setOff("sharpen", true);
assert.strictEqual(PS.isOff("sharpen"), true);
assert.strictEqual(PS.isOff("alg"), false);
assert.deepStrictEqual(posts.at(-1).values._off, { modules: ["sharpen"] });
await PS.setOff("alg", true);
await PS.setOff("sharpen", false);
assert.deepStrictEqual(posts.at(-1).values._off, { modules: ["alg"] });
});
test("a project opened after another does not inherit the first one's pending switch", async () => {
const PS = load([]);
await PS.ensureLoaded();
await PS.setOff("sharpen", true);
await PS.adopt([{ id: "model", node: "Loader", inputs: { file: "x" } }]);
assert.strictEqual(PS.isOff("sharpen"), false);
});
test("the settings of a Generate are taken once: frozenInputs is a copy, wired inputs are left to their node", async () => {
const PS = load([]);
await PS.ensureLoaded();
const frozen = PS.frozenInputs();
assert.deepStrictEqual(frozen.model, { file: "", dtype: "bf16" });
assert.deepStrictEqual(frozen.gen, { steps: 8 }); // `model: ["model", 0]` is fed by a node
await PS.save({ inputs: { gen: { steps: 4 } } });
assert.strictEqual(PS.slots().find((s) => s.id === "gen").inputs.steps, 4);
assert.strictEqual(frozen.gen.steps, 8); // the snapshot does not follow the edit
frozen.model.dtype = "fp16";
assert.strictEqual(PS.slots().find((s) => s.id === "model").inputs.dtype, "bf16"); // and is not the live object
});
@@ -0,0 +1,89 @@
import test from "node:test";
import assert from "node:assert/strict";
import { wireReferences } from "../reference_wiring.js";
function refSlot(n) {
return { id: `ref_source_${n}`, roles: [{ at: `assets.reference_${n}`,
input: "media_id", wireTo: { slot: "r2v", input: `ref_images.ref_image_${n - 1}` } }] };
}
test("each picked reference is wired to its slot and to r2v", () => {
const slots = [refSlot(1), refSlot(2)];
const { overrides, unwired } = wireReferences(["aaa111111111", "bbb222222222"], slots);
assert.equal(unwired, 0);
assert.equal(overrides.ref_source_1.media_id, "aaa111111111");
assert.deepEqual(overrides.r2v["ref_images.ref_image_0"], ["ref_source_1", 0]);
assert.deepEqual(overrides.r2v["ref_images.ref_image_1"], ["ref_source_2", 0]);
});
test("a reference beyond the pipeline's slots is counted, not silently dropped", () => {
// This is the bug an adversarial review caught: a scene can carry more
// references than the CURRENT pipeline offers slots for (e.g. added under
// a bigger preset, then the preset was swapped for a smaller one). The old
// code in boot.js just returned from the forEach iteration and moved on --
// no override, no count, no way for the caller to tell the user anything
// was left out.
const slots = [refSlot(1)]; // only one slot offered
const { overrides, unwired } = wireReferences(
["aaa111111111", "bbb222222222", "ccc333333333"], slots);
assert.equal(unwired, 2);
assert.equal(overrides.ref_source_1.media_id, "aaa111111111");
assert.equal(Object.keys(overrides).length, 2); // ref_source_1 + r2v, nothing else
});
test("no references and no slots wires nothing and reports no shortfall", () => {
const { overrides, unwired } = wireReferences([], []);
assert.deepEqual(overrides, {});
assert.equal(unwired, 0);
});
test("slots not yet loaded (null, boot.js's own initial state) is treated as no slots, not a crash", () => {
const { overrides, unwired } = wireReferences(["aaa111111111"], null);
assert.deepEqual(overrides, {});
assert.equal(unwired, 1);
});
test("a slot whose roles claim two reference numbers does not let the second silently overwrite the first", () => {
// Not producible by the shipped H3 preset (one role per slot), but nothing
// stops a future preset's roles array from doing this -- the function's own
// contract ("no reference silently dropped") has to hold for the input
// shape, not just the one preset that happens to exist today.
const combo = { id: "combo_slot", roles: [
{ at: "assets.reference_1", wireTo: { slot: "r2v", input: "ref_images.ref_image_0" } },
{ at: "assets.reference_2", wireTo: { slot: "r2v", input: "ref_images.ref_image_1" } },
] };
const { overrides, unwired } = wireReferences(["AAA", "BBB"], [combo]);
assert.equal(overrides.combo_slot.media_id, "AAA");
assert.equal(unwired, 1);
});
test("two different slots wired to the same downstream input do not let the second overwrite the first", () => {
// Same failure as the combo-slot case above, reached a different way: two
// separate reference slots whose `wireTo` both point at one destination
// (e.g. a copy-pasted preset). The destination can hold one value.
const slots = [
{ id: "ref_source_1", roles: [{ at: "assets.reference_1",
wireTo: { slot: "r2v", input: "ref_images.ref_image_0" } }] },
{ id: "ref_source_2", roles: [{ at: "assets.reference_2",
wireTo: { slot: "r2v", input: "ref_images.ref_image_0" } }] },
];
const { overrides, unwired } = wireReferences(["AAA", "BBB"], slots);
assert.deepEqual(overrides.r2v["ref_images.ref_image_0"], ["ref_source_1", 0]);
assert.equal(overrides.ref_source_1.media_id, "AAA");
assert.equal(overrides.ref_source_2, undefined);
assert.equal(unwired, 1);
});
test("a slot the caller already assigned (e.g. the source-image slot) is not silently overwritten by a reference", () => {
// boot.js sets the source-image slot's media_id BEFORE calling
// wireReferences() -- if a preset ever reused that same slot id for a
// reference role too, the reference would clobber the source image with
// no toast, since wireReferences() has no visibility into that earlier
// write on its own. The caller passes it in as reserved instead.
const slots = [refSlot(1)];
const { overrides, unwired } = wireReferences(
["AAA"], slots, ["ref_source_1"]);
assert.equal(overrides.ref_source_1, undefined);
assert.equal(unwired, 1);
});
+472
View File
@@ -0,0 +1,472 @@
// The run lifecycle, driven without a server.
//
// Everything the outside provides is injected, so these tests exercise the real
// file rather than a mock of it: a fake fetch returns what ComfyUI returns, and
// messages are the ones ComfyUI actually sends (checked against execution.py
// and comfy_execution/progress.py, not from memory).
import test from "node:test";
import assert from "node:assert/strict";
import { createRun, viewUrl, IDLE, QUEUED, RUNNING, DONE, FAILED, CANCELLED }
from "../run.js";
const ok = (body) => async () => ({ ok: true, status: 200, json: async () => body });
const refuse = (status, body) => async () => ({ ok: false, status, json: async () => body });
const message = (type, data) => ({ data: JSON.stringify({ type, data }) });
function runner(fetchImpl, { clientId = "c1" } = {}) {
const sent = [];
const record = async (url, init) => {
sent.push({ url, body: init && init.body ? JSON.parse(init.body) : null });
return fetchImpl(url, init);
};
return { run: createRun({ fetch: record, clientId }), sent };
}
test("queueing a prompt reports it and keeps the id the server gave", async () => {
const { run, sent } = runner(ok({ prompt_id: "p-1", number: 3 }));
const id = await run.start({ "1": { class_type: "X", inputs: {} } });
assert.equal(id, "p-1");
assert.equal(run.state.phase, QUEUED);
assert.equal(run.state.promptId, "p-1");
assert.equal(sent[0].url, "/prompt");
assert.equal(sent[0].body.client_id, "c1", "the socket would never hear about this run");
});
test("extra rides in extra_data, which ComfyUI echoes back on /queue and /history", async () => {
// The only way a reload can learn which scene/project a run in flight
// belongs to -- nothing else carries it through the queue.
const { run, sent } = runner(ok({ prompt_id: "p-2", number: 4 }));
await run.start({ "1": {} }, { extra: { funpack_scene_id: "s1", funpack_project_id: "pr1" } });
assert.deepEqual(sent[0].body.extra_data, { funpack_scene_id: "s1", funpack_project_id: "pr1" });
assert.equal(sent[0].body.client_id, "c1", "client_id must still be sent alongside extra_data");
});
test("no extra given is no extra_data sent, not an empty one", async () => {
const { run, sent } = runner(ok({ prompt_id: "p-3", number: 5 }));
await run.start({ "1": {} });
assert.equal("extra_data" in sent[0].body, false);
});
test("a graph the queue refuses fails with ComfyUI's own reason", async () => {
const { run } = runner(refuse(400, {
error: { type: "prompt_outputs_failed_validation", message: "Prompt outputs failed validation" },
node_errors: { "4": { errors: [{ message: "Value not in list" }] } },
}));
await assert.rejects(() => run.start({}), /failed validation/);
assert.equal(run.state.phase, FAILED);
assert.match(run.state.error.message, /failed validation/);
assert.ok(run.state.error.nodes["4"], "the node that was wrong was dropped");
});
test("a queue that cannot be reached says so rather than looking idle", async () => {
// The dev server serves the app and has no queue behind it, so this is the
// ordinary case while working on the UI, not an exotic one.
const { run } = runner(async () => { throw new TypeError("Failed to fetch"); });
await assert.rejects(() => run.start({}), /Failed to fetch/);
assert.equal(run.state.phase, FAILED);
assert.match(run.state.error.message, /could not be reached/);
});
test("progress follows the node that is working, not the sum of them", async () => {
const { run } = runner(ok({ prompt_id: "p-1" }));
await run.start({});
run.handle(JSON.parse(message("execution_start", { prompt_id: "p-1" }).data));
assert.equal(run.state.phase, RUNNING);
run.handle(JSON.parse(message("progress_state", {
prompt_id: "p-1",
nodes: {
"3": { state: "finished", value: 7, max: 7, node_id: "3", display_node_id: "3" },
"5": { state: "running", value: 2, max: 20, node_id: "5", display_node_id: "5" },
},
}).data));
assert.deepEqual(run.state.progress, { value: 2, max: 20 });
assert.equal(run.state.node, "5");
});
test("images are collected from every save, not replaced by the last", async () => {
const { run } = runner(ok({ prompt_id: "p-1" }));
await run.start({});
for (const name of ["a.png", "b.png"]) {
run.handle(JSON.parse(message("executed", {
prompt_id: "p-1",
output: { images: [{ filename: name, subfolder: "", type: "output" }] },
}).data));
}
run.handle(JSON.parse(message("execution_success", { prompt_id: "p-1" }).data));
assert.equal(run.state.phase, DONE);
assert.equal(run.state.images.length, 2);
assert.deepEqual(run.images(), [
"/view?filename=a.png&subfolder=&type=output",
"/view?filename=b.png&subfolder=&type=output",
]);
});
test("another client's run does not drive this one", async () => {
// execution_interrupted is BROADCAST, so a second tab cancelling its own run
// would otherwise mark this one cancelled while it is still generating.
const { run } = runner(ok({ prompt_id: "mine" }));
await run.start({});
run.handle(JSON.parse(message("execution_start", { prompt_id: "mine" }).data));
run.handle(JSON.parse(message("execution_interrupted", { prompt_id: "theirs" }).data));
assert.equal(run.state.phase, RUNNING, "somebody else's cancel stopped this run");
run.handle(JSON.parse(message("execution_success", { prompt_id: "theirs" }).data));
assert.equal(run.state.phase, RUNNING, "somebody else's result finished this run");
run.handle(JSON.parse(message("execution_success", { prompt_id: "mine" }).data));
assert.equal(run.state.phase, DONE);
});
test("a failure carries the node and the message ComfyUI gave", async () => {
const { run } = runner(ok({ prompt_id: "p-1" }));
await run.start({});
run.handle(JSON.parse(message("execution_error", {
prompt_id: "p-1", node_id: "9", node_type: "KSampler",
exception_message: "Given groups=1, expected input to have 4 channels",
traceback: ["line one", "line two"],
}).data));
assert.equal(run.state.phase, FAILED);
assert.equal(run.state.error.node, "KSampler");
assert.match(run.state.error.message, /4 channels/);
assert.ok(run.state.error.traceback, "the traceback was thrown away");
});
test("cancelling asks the server and waits for it to say it stopped", async () => {
const { run, sent } = runner(ok({ prompt_id: "p-1" }));
await run.start({});
run.handle(JSON.parse(message("execution_start", { prompt_id: "p-1" }).data));
await run.cancel();
// The state-agnostic route, not /interrupt: a job waiting behind another one
// is not in the running half of the queue, so /interrupt answers 200 and does
// nothing at all.
assert.equal(sent[1].url, "/api/jobs/p-1/cancel");
// Claiming it early is how a UI shows "cancelled" over a run still burning
// GPU time. The server says when it actually stopped.
assert.equal(run.state.phase, RUNNING);
run.handle(JSON.parse(message("execution_interrupted", { prompt_id: "p-1" }).data));
assert.equal(run.state.phase, CANCELLED);
});
test("cancelling when nothing is running asks the server nothing", async () => {
const { run, sent } = runner(ok({}));
assert.equal(await run.cancel(), false);
assert.equal(sent.length, 0, "an idle cancel interrupted whatever else was queued");
});
test("subscribers hear the current state immediately and on every change", async () => {
const { run } = runner(ok({ prompt_id: "p-1" }));
const seen = [];
const stop = run.subscribe((s) => seen.push(s.phase));
assert.deepEqual(seen, [IDLE], "a subscriber had to wait for a change to learn anything");
await run.start({});
assert.ok(seen.includes(QUEUED));
stop();
const before = seen.length;
run.handle(JSON.parse(message("execution_success", { prompt_id: "p-1" }).data));
assert.equal(seen.length, before, "unsubscribing did not stop the updates");
});
test("a listener that throws does not stop the others", async () => {
const { run } = runner(ok({ prompt_id: "p-1" }));
const seen = [];
run.subscribe(() => { throw new Error("bad listener"); });
run.subscribe((s) => seen.push(s.phase));
await run.start({});
assert.ok(seen.includes(QUEUED));
});
test("a binary frame and a malformed one leave the socket alone", async () => {
const events = {};
const socket = { addEventListener: (type, fn) => { events[type] = fn; } };
const run = createRun({ fetch: ok({}), clientId: "c1", connect: () => socket });
run.listen();
events.message({ data: new Uint8Array([1, 2, 3]) }); // a preview image
events.message({ data: "{not json" });
assert.equal(run.state.phase, IDLE);
});
test("viewUrl escapes what goes into it", () => {
const url = viewUrl({ filename: "a b&c.png", subfolder: "s/1", type: "temp" });
assert.match(url, /filename=a\+b%26c\.png/);
assert.match(url, /subfolder=s%2F1/);
assert.match(url, /type=temp/);
});
// --- while /prompt is still in flight ---------------------------------------
//
// Every test above awaits start() before sending a message, so none of them
// ever enters the window between "queued" and "the server said what this run is
// called". That window is a real network round trip, the socket is already
// live, and everything ComfyUI broadcasts arrives during it.
function held() {
let release;
const gate = new Promise((resolve) => { release = resolve; });
const sent = [];
const fetchImpl = async (url, init) => {
sent.push({ url, body: init && init.body ? JSON.parse(init.body) : null });
if (url.endsWith("/prompt")) {
await gate;
return { ok: true, status: 200, json: async () => ({ prompt_id: "mine" }) };
}
return { ok: true, status: 200, json: async () => ({}) };
};
return { run: createRun({ fetch: fetchImpl, clientId: "c1" }), sent, release };
}
test("a stranger's result during the round trip is not adopted as this run's", async () => {
const { run, release } = held();
const started = run.start({});
// Another client finishes while this request is still on the wire.
run.handle({ type: "executed", data: { prompt_id: "theirs", output: { images: [{ filename: "not-mine.png" }] } } });
run.handle({ type: "execution_success", data: { prompt_id: "theirs" } });
assert.equal(run.state.phase, QUEUED, "somebody else's run finished this one");
assert.deepEqual(run.state.images, [], "somebody else's picture was taken as this run's result");
release();
await started;
assert.equal(run.state.phase, QUEUED);
assert.deepEqual(run.state.images, []);
});
test("this run's own messages during the round trip are not lost", async () => {
// The other half. A short run can finish inside the round trip, and dropping
// what arrives early would leave the app waiting for a result it already had.
const { run, release } = held();
const started = run.start({});
run.handle({ type: "execution_start", data: { prompt_id: "mine" } });
run.handle({ type: "executed", data: { prompt_id: "mine", output: { images: [{ filename: "mine.png" }] } } });
run.handle({ type: "execution_success", data: { prompt_id: "mine" } });
release();
await started;
assert.equal(run.state.phase, DONE);
assert.deepEqual(run.state.images.map((i) => i.filename), ["mine.png"]);
});
test("cancelling before the run has a name never interrupts anything else", async () => {
// ComfyUI reads a missing prompt_id as "interrupt whatever is running". On a
// shared box that is another person's generation; the clicking user's own
// request may not even be queued yet.
const { run, sent, release } = held();
const started = run.start({});
assert.equal(await run.cancel(), true, "the cancel was refused rather than remembered");
assert.deepEqual(sent.filter((s) => s.url.includes("cancel") || s.url.endsWith("/interrupt")), [],
"a cancel went out with no prompt id, which ComfyUI reads as a global interrupt");
release();
await started;
const cancels = sent.filter((s) => s.url.includes("/cancel"));
assert.equal(cancels.length, 1, "the cancel the user asked for never happened");
assert.equal(cancels[0].url, "/api/jobs/mine/cancel");
});
test("a cancel is never sent with a null prompt id, whatever the phase", async () => {
const { run } = runner(ok({ prompt_id: null })); // a server answering oddly
await run.start({});
const before = run.state.promptId;
assert.equal(before, null);
await run.cancel();
// Remembered, not sent: there is still nothing to name.
assert.equal(run.state.phase, QUEUED);
});
test("held messages do not grow without bound", async () => {
const { run, release } = held();
const started = run.start({});
for (let i = 0; i < 500; i += 1) {
run.handle({ type: "progress_state", data: { prompt_id: "theirs", nodes: {} } });
}
release();
await started;
assert.equal(run.state.phase, QUEUED, "a flood of other traffic changed this run");
});
// --- reattaching after a reload ---------------------------------------------
test("adopting a run already in flight picks up its progress and result", async () => {
const { run } = runner(ok({}));
assert.equal(run.adopt("was-running"), true);
assert.equal(run.state.phase, RUNNING);
assert.equal(run.state.promptId, "was-running");
run.handle(JSON.parse(message("progress_state", {
prompt_id: "was-running",
nodes: { "5": { state: "running", value: 4, max: 20, node_id: "5" } },
}).data));
assert.deepEqual(run.state.progress, { value: 4, max: 20 });
run.handle(JSON.parse(message("executed", {
prompt_id: "was-running", output: { images: [{ filename: "late.png" }] },
}).data));
run.handle(JSON.parse(message("execution_success", { prompt_id: "was-running" }).data));
assert.equal(run.state.phase, DONE);
assert.deepEqual(run.state.images.map((i) => i.filename), ["late.png"]);
});
test("adopting nothing does nothing", async () => {
const { run } = runner(ok({}));
assert.equal(run.adopt(null), false);
assert.equal(run.state.phase, IDLE);
});
test("an adopted run can be cancelled by its own id", async () => {
const { run, sent } = runner(ok({}));
run.adopt("was-running");
await run.cancel();
assert.equal(sent[0].url, "/api/jobs/was-running/cancel");
});
// --- a run that finishes while the page is loading ---------------------------
//
// The socket opens before the queue can be asked which run is ours, so a run
// that ends in that moment has already left `queue_running` by the time anyone
// looks. Nothing about it is recoverable unless what arrived on the socket in
// the meantime was kept.
test("messages arriving before a page has any run of its own are kept", () => {
const { run } = runner(ok({}));
assert.equal(run.state.phase, IDLE);
run.handle(JSON.parse(message("executed", {
prompt_id: "prev-run",
output: { images: [{ filename: "finished-during-reload.png" }] },
}).data));
run.handle(JSON.parse(message("execution_success", { prompt_id: "prev-run" }).data));
// Nothing was acted on -- this page does not yet know the run is its own.
assert.equal(run.state.phase, IDLE);
assert.deepEqual(run.state.images, []);
// But the id was noticed, which is what makes it findable.
assert.deepEqual(run.seen(), ["prev-run"]);
});
test("adopting a run that already finished still produces its result", () => {
const { run } = runner(ok({}));
run.handle(JSON.parse(message("executed", {
prompt_id: "prev-run",
output: { images: [{ filename: "finished-during-reload.png" }] },
}).data));
run.handle(JSON.parse(message("execution_success", { prompt_id: "prev-run" }).data));
run.adopt("prev-run");
assert.equal(run.state.phase, DONE, "the result was dropped rather than replayed");
assert.deepEqual(run.state.images.map((i) => i.filename), ["finished-during-reload.png"]);
});
test("what was seen while idle does not become somebody else's run", () => {
const { run } = runner(ok({}));
run.handle(JSON.parse(message("execution_success", { prompt_id: "theirs" }).data));
run.adopt("mine");
assert.equal(run.state.phase, RUNNING, "a stranger's finish was replayed onto this run");
});
test("a run waiting behind another one can still be cancelled", async () => {
// The case /interrupt cannot serve. ComfyUI only consults the RUNNING half of
// its queue there, so a job that is merely pending gets a 200, a log line
// saying it was skipped, and a full generation when its turn arrives.
const { run, sent } = runner(ok({ prompt_id: "waiting" }));
await run.start({});
assert.equal(run.state.phase, QUEUED, "this test needs a run that has not started");
await run.cancel();
assert.equal(sent[1].url, "/api/jobs/waiting/cancel",
"a queued run was cancelled with a call that only stops a running one");
});
test("an older ComfyUI without the cancel route still interrupts", async () => {
const sent = [];
const fetchImpl = async (url, init) => {
sent.push({ url, body: init && init.body ? JSON.parse(init.body) : null });
if (url.includes("/api/jobs/")) return { ok: false, status: 404, json: async () => ({}) };
return { ok: true, status: 200, json: async () => ({ prompt_id: "p-1" }) };
};
const run = createRun({ fetch: fetchImpl, clientId: "c1" });
await run.start({});
run.handle({ type: "execution_start", data: { prompt_id: "p-1" } });
await run.cancel();
const urls = sent.map((s) => s.url);
assert.ok(urls.includes("/api/jobs/p-1/cancel"));
assert.ok(urls.includes("/interrupt"), "nothing was tried after the route was absent");
assert.equal(sent[sent.length - 1].body.prompt_id, "p-1",
"the fallback interrupted globally rather than this run");
});
test("a cancelled pending run does not sit at Queued forever", async () => {
// Dequeuing a job that never ran sends nothing on the socket -- there is no
// execution to interrupt and so no execution_interrupted. Without reading the
// answer, a cancel that worked leaves the UI saying Queued indefinitely.
const sent = [];
const fetchImpl = async (url, init) => {
sent.push(url);
if (url.includes("/cancel")) return { ok: true, status: 200, json: async () => ({ cancelled: true }) };
return { ok: true, status: 200, json: async () => ({ prompt_id: "waiting" }) };
};
const run = createRun({ fetch: fetchImpl, clientId: "c1" });
await run.start({});
await run.cancel();
assert.equal(run.state.phase, CANCELLED);
});
test("a running run is still left to the socket to declare stopped", async () => {
const fetchImpl = async (url) => ({
ok: true, status: 200,
json: async () => (url.includes("/cancel") ? { cancelled: true } : { prompt_id: "p-1" }),
});
const run = createRun({ fetch: fetchImpl, clientId: "c1" });
await run.start({});
run.handle({ type: "execution_start", data: { prompt_id: "p-1" } });
await run.cancel();
// Claiming it early is how a UI shows "cancelled" over a run still burning
// GPU time: the interrupt is a request, and the model finishes its step.
assert.equal(run.state.phase, RUNNING);
run.handle({ type: "execution_interrupted", data: { prompt_id: "p-1" } });
assert.equal(run.state.phase, CANCELLED);
});
test("a cancel the server says did nothing does not claim it did", async () => {
const fetchImpl = async (url) => ({
ok: true, status: 200,
json: async () => (url.includes("/cancel") ? { cancelled: false } : { prompt_id: "gone" }),
});
const run = createRun({ fetch: fetchImpl, clientId: "c1" });
await run.start({});
await run.cancel();
assert.equal(run.state.phase, QUEUED, "the run was declared cancelled on the server's word that it was not");
});
test("a run adopted while it is still waiting its turn says Queued, not Working", () => {
const { run } = runner(ok({}));
run.adopt("waiting", { running: false });
assert.equal(run.state.phase, QUEUED);
assert.equal(run.state.promptId, "waiting");
// And it becomes running when the server says it did, not before.
run.handle(JSON.parse(message("execution_start", { prompt_id: "waiting" }).data));
assert.equal(run.state.phase, RUNNING);
});
+609
View File
@@ -0,0 +1,609 @@
// Starting a run, and taking one back over.
//
// This wiring used to live in boot.js, which runs on import and needs a
// document, so none of it could be driven -- and both faults found here were
// ordering faults in exactly that untested wiring.
import test from "node:test";
import assert from "node:assert/strict";
import { setupDom, teardownDom } from "../../composer/tests/_dom.js";
let createGenerator, reattach, wire, waitForTerminal, createTransport;
test.before(async () => {
setupDom();
await import("../../composer/composer.js");
({ createGenerator, reattach, wire, waitForTerminal } = await import("../session.js"));
({ createTransport } = await import("../transport.js"));
});
test.after(() => teardownDom());
// The REAL transport, not a stand-in.
//
// A hand-rolled one recorded setText and setDisabled into arrays and had a
// draw() that did nothing -- so it could not show the fault it was standing in
// for: the real draw() writes the status from the run state, and a refusal set
// just before it was overwritten with "Ready" a moment after it appeared. The
// test passed and the button did nothing on screen.
function transport() {
// The real transport, mounted the way the shell mounts it: its controls go in
// a zone head and what it is saying goes at the far end of the same head.
// There is no bar -- an editor's controls belong to the region they act on.
const t = createTransport({});
const head = document.createElement("header");
for (const handle of [...t.actions, ...t.status, t.warning]) head.appendChild(handle.node);
document.body.replaceChildren(head);
return {
...t,
head,
get text() { return t.statusText.node.textContent; },
get disabled() { return t.generate.node.disabled; },
};
}
/** A run whose start() can be held open, like a real POST.
*
* Open by DEFAULT, and held only when a test asks. A gate that starts shut
* means any regression letting an extra start() through hangs the suite
* forever instead of failing it, and a hang says far less than an assertion. */
function fakeRun(phase = "idle") {
const started = [];
let release = () => {};
let gate = Promise.resolve();
return {
hold() {
gate = new Promise((r) => { release = () => r(); });
},
started,
release: () => release(),
state: { phase, promptId: null, progress: null, images: [], audio: [], error: null, node: null },
listened: 0,
listen() { this.listened += 1; },
// The real run delivers the CURRENT state on subscribe, at once. Leaving
// that off is how a fake stays tidier than the thing it stands in for: the
// synchronous first delivery is exactly what handed a deliberately disabled
// button straight back, and a fake without it could not show that.
subscribers: new Set(),
subscribe(fn) {
this.subscribers.add(fn);
try { fn(this.state); } catch { /* as the real one does */ }
return () => this.subscribers.delete(fn);
},
announce() { for (const fn of this.subscribers) fn(this.state); },
seen: () => [],
adopt(id) {
this.state = { ...this.state, phase: "running", promptId: id, images: [] };
this.announce();
return true;
},
async start(prompt) {
started.push(prompt);
this.state = { ...this.state, phase: "queued" };
await gate;
return "p-1";
},
};
}
const plan = (over = {}) => async () => ({
slots: [], refused: [], incomplete: [], queueable: true, prompt: { "1": {} }, ...over,
});
test("two clicks during the round trip queue one run, not two", async () => {
// The button only goes dead when the RUN is queued, and getting there takes a
// round trip to ask what to queue. That round trip is the whole window.
const run = fakeRun();
run.hold();
const t = transport();
let asked = 0;
const check = async () => { asked += 1; return (await plan()()); };
const onGenerate = createGenerator({ run, transport: t, check });
const first = onGenerate();
const second = onGenerate();
run.release();
const [a, b] = await Promise.all([first, second]);
assert.equal(run.started.length, 1, "two /prompt posts went out");
assert.equal(asked, 1, "the pipeline was assembled twice for one click");
assert.equal(a, true);
assert.equal(b, false, "the second click reported that it had started a run");
});
test("the button goes dead at the click, not at the queue", async () => {
const run = fakeRun();
run.hold();
const t = transport();
const onGenerate = createGenerator({ run, transport: t, check: plan() });
const running = onGenerate();
assert.equal(t.disabled, true, "the button stayed live during the round trip");
run.release();
await running;
});
test("a run already in flight is not started again", async () => {
for (const phase of ["queued", "running"]) {
const run = fakeRun(phase);
const onGenerate = createGenerator({ run, transport: transport(), check: plan() });
assert.equal(await onGenerate(), false);
assert.equal(run.started.length, 0, `a ${phase} run was started a second time`);
}
});
test("a pipeline that is not ready says why, beside Generate", async () => {
const run = fakeRun();
const t = transport();
const onGenerate = createGenerator({
run, transport: t,
check: plan({ queueable: false, prompt: null, incomplete: ["model: nothing is chosen"] }),
});
assert.equal(await onGenerate(), false);
// Read off the element, so a later redraw wiping it is visible here.
assert.equal(t.text, "model: nothing is chosen");
assert.equal(run.started.length, 0);
assert.equal(t.disabled, false, "a refused attempt left Generate dead");
});
test("a refusal is shown even when nothing is incomplete", async () => {
const t = transport();
const onGenerate = createGenerator({
run: fakeRun(), transport: t,
check: plan({ queueable: false, prompt: null, refused: ["a slot cannot feed itself"] }),
});
await onGenerate();
assert.equal(t.text, "a slot cannot feed itself");
});
test("a pipeline that cannot be read at all is reported rather than swallowed", async () => {
const t = transport();
const onGenerate = createGenerator({
run: fakeRun(), transport: t,
check: async () => { throw new TypeError("Failed to fetch"); },
});
assert.equal(await onGenerate(), false);
assert.match(t.text, /could not be read/);
});
test("a failed attempt leaves Generate usable again", async () => {
const run = fakeRun();
const t = transport();
const onGenerate = createGenerator({
run, transport: t, check: plan({ queueable: false, prompt: null }),
});
await onGenerate();
assert.equal(t.disabled, false, "nothing gave the button back after the attempt ended");
// And a second click is allowed, which it would not be if the guard stuck.
assert.equal(await onGenerate(), false);
assert.equal(t.disabled, false);
});
// --- reattach ---------------------------------------------------------------
test("a run still in the queue is taken back over", async () => {
const run = fakeRun();
const got = await reattach(run, "me", {
queuedFor: async () => ({ promptId: "still-going", running: true, sceneId: null, projectId: null }),
finishedFor: async () => { throw new Error("history should not have been asked"); },
});
assert.equal(got.promptId, "still-going");
assert.equal(run.state.phase, "running");
});
test("a run that finished during the load is found in history", async () => {
const run = fakeRun();
run.seen = () => ["ended-just-now"];
const got = await reattach(run, "me", {
queuedFor: async () => null,
finishedFor: async (_id, seen) => (seen.includes("ended-just-now") ? "ended-just-now" : null),
});
assert.equal(got.promptId, "ended-just-now");
});
test("which scene/project a run belongs to travels with it, for both halves", async () => {
const run = fakeRun();
const queued = await reattach(run, "me", {
queuedFor: async () => ({ promptId: "q1", running: true, sceneId: "s1", projectId: "p1" }),
finishedFor: async () => { throw new Error("should not be asked"); },
});
assert.deepEqual(queued, { promptId: "q1", sceneId: "s1", projectId: "p1" });
const run2 = fakeRun();
run2.seen = () => ["f1"];
const finished = await reattach(run2, "me", {
queuedFor: async () => null,
finishedFor: async (_id, _seen, { onFound } = {}) => {
if (onFound) onFound({ client_id: "me", funpack_scene_id: "s2", funpack_project_id: "p2" });
return "f1";
},
});
assert.deepEqual(finished, { promptId: "f1", sceneId: "s2", projectId: "p2" });
});
test("onAdopt fires BEFORE the run is adopted, not after reattach resolves", async () => {
// An already-finished run can go straight to DONE inside adopt() itself
// (whatever the socket buffered gets replayed there) -- a listener that
// learns which scene this is for only once reattach's own promise settles
// would learn it one tick after that DONE already reached run.subscribe with
// no scene to attach it to. This is the ordering the fix actually depends on.
const order = [];
const run = fakeRun();
const realAdopt = run.adopt.bind(run);
run.adopt = (id, opts) => { order.push("adopt"); return realAdopt(id, opts); };
run.subscribe((state) => { if (state.phase === "running") order.push("subscriber-saw-it"); });
await reattach(run, "me", {
queuedFor: async () => ({ promptId: "q1", running: true, sceneId: "s1", projectId: "p1" }),
finishedFor: async () => { throw new Error("should not be asked"); },
onAdopt: (sceneId, projectId) => { order.push(`onAdopt:${sceneId}:${projectId}`); },
});
assert.deepEqual(order, ["onAdopt:s1:p1", "adopt", "subscriber-saw-it"]);
});
test("onAdopt's third argument carries every scene a multi-scene run covers", async () => {
// No chain sampler to split a multi-scene selection into several /prompt
// calls, so one run can cover several scenes at once. sceneId alone (the
// first) is what every existing caller still reads; sceneIds is additive,
// for a caller that needs the whole list back on reattach.
const run = fakeRun();
let got = null;
await reattach(run, "me", {
queuedFor: async () => ({ promptId: "q1", running: true, sceneId: "s1",
sceneIds: ["s1", "s2", "s3"], projectId: "p1" }),
finishedFor: async () => { throw new Error("should not be asked"); },
onAdopt: (sceneId, projectId, sceneIds) => { got = { sceneId, projectId, sceneIds }; },
});
assert.deepEqual(got, { sceneId: "s1", projectId: "p1", sceneIds: ["s1", "s2", "s3"] });
// The finished-run path recovers it from the raw extra_data finishedFor
// hands back through onFound, the same way it already reads funpack_scene_id.
const run2 = fakeRun();
run2.seen = () => ["f1"];
let got2 = null;
await reattach(run2, "me", {
queuedFor: async () => null,
finishedFor: async (_id, _seen, { onFound } = {}) => {
if (onFound) onFound({ client_id: "me", funpack_scene_id: "s2",
funpack_scene_ids: ["s2", "s4"], funpack_project_id: "p2" });
return "f1";
},
onAdopt: (sceneId, projectId, sceneIds) => { got2 = { sceneId, projectId, sceneIds }; },
});
assert.deepEqual(got2, { sceneId: "s2", projectId: "p2", sceneIds: ["s2", "s4"] });
});
test("onAdopt's third argument falls back to just the first scene when nothing said more", async () => {
const run = fakeRun();
let got = null;
await reattach(run, "me", {
queuedFor: async () => ({ promptId: "q1", running: true, sceneId: "s1", projectId: "p1" }),
finishedFor: async () => { throw new Error("should not be asked"); },
onAdopt: (sceneId, projectId, sceneIds) => { got = sceneIds; },
});
assert.deepEqual(got, ["s1"]);
});
test("history is not asked about a run this page never saw", async () => {
const run = fakeRun();
let asked = false;
const got = await reattach(run, "me", {
queuedFor: async () => null,
finishedFor: async () => { asked = true; return "something-old"; },
});
assert.equal(got, null);
assert.equal(asked, false, "a result from before this page load could be resurrected");
});
test("a page already running its own generation is not reattached to another", async () => {
const run = fakeRun("running");
const got = await reattach(run, "me", {
queuedFor: async () => ({ promptId: "some-other", running: true }),
finishedFor: async () => null,
});
assert.equal(got, null);
});
test("no queue and no history is silence, not an error", async () => {
const run = fakeRun();
const got = await reattach(run, "me", {
queuedFor: async () => { throw new TypeError("Failed to fetch"); },
finishedFor: async () => null,
});
assert.equal(got, null);
assert.equal(run.state.phase, "idle");
});
test("a reason for not starting is not wiped by the next redraw", () => {
// The redraw happens on the way out of every attempt, and an attempt that was
// refused leaves the run exactly where it was: idle. Drawing an idle run says
// "Ready", which is how the reason appeared and vanished in the same frame.
const t = transport();
t.say("model: nothing is chosen");
t.draw({ phase: "idle", progress: null, images: [], error: null, node: null });
assert.equal(t.text, "model: nothing is chosen");
});
test("a run that actually starts outranks the last refusal", () => {
const t = transport();
t.say("model: nothing is chosen");
t.draw({ phase: "queued", progress: null, images: [], error: null, node: null });
assert.equal(t.text, "Queued");
t.draw({ phase: "idle", progress: null, images: [], error: null, node: null });
assert.equal(t.text, "Ready", "the stale reason came back after a real run");
});
// --- the load ordering, which is the thing that keeps breaking ---------------
//
// Every fault on this path has been an ordering fault, and every one lived in
// boot.js, where nothing could reach it. These drive the real order.
function wired({ running = null, finished = null, phase = "idle" } = {}) {
const t = transport();
const run = fakeRun(phase);
let releaseQueue;
const queueAnswered = new Promise((r) => { releaseQueue = r; });
const session = wire({
run,
page: { transport: t },
check: plan(),
id: "me",
queuedFor: async () => { await queueAnswered; return running ? { promptId: running, running: true } : null; },
finishedFor: async () => finished,
});
return { t, run, session, releaseQueue };
}
test("Generate waits for the page to work out whether a run is already going", async () => {
// The reload-during-a-generation case. The button is on screen and the phase
// still reads idle for the whole length of the queue lookup, so a click in
// that window used to queue a second job and orphan the first.
const { t, run, session, releaseQueue } = wired({ running: "already-going" });
assert.equal(t.disabled, true, "Generate was live before anything was known");
const clicked = session.generate();
releaseQueue();
await session.ready;
assert.equal(await clicked, false, "a second run was queued over the one already going");
assert.equal(run.started.length, 0);
assert.equal(run.state.promptId, "already-going", "the run in progress was orphaned");
});
test("with nothing already running, Generate comes back and works", async () => {
const { t, run, session, releaseQueue } = wired({ running: null });
releaseQueue();
await session.ready;
assert.equal(t.disabled, false, "Generate never came back");
assert.equal(t.text, "Ready", `the loading note stuck: ${t.text}`);
const started = session.generate();
run.release();
assert.equal(await started, true);
assert.equal(run.started.length, 1);
});
test("the page says what it is doing while it works it out", async () => {
const { t, releaseQueue, session } = wired({ running: null });
assert.match(t.text, /Looking for a run/);
releaseQueue();
await session.ready;
});
test("a run found in history is adopted before Generate is offered again", async () => {
const { t, run, session, releaseQueue } = wired({ running: null, finished: "ended-just-now" });
run.seen = () => ["ended-just-now"];
releaseQueue();
await session.ready;
assert.equal(run.state.promptId, "ended-just-now");
// The button now follows the RUN, not the lookup: the loading note is gone
// and what is on screen is whatever that run is doing.
assert.doesNotMatch(t.text, /Looking for a run/);
});
test("a queue that cannot be reached still gives the button back", async () => {
// The dev server, and any ComfyUI that is down. Waiting forever on a lookup
// that will never answer is the same as no button at all.
const t = transport();
const run = fakeRun();
const session = wire({
run, page: { transport: t }, check: plan(), id: "me",
queuedFor: async () => { throw new TypeError("Failed to fetch"); },
finishedFor: async () => null,
});
await session.ready;
assert.equal(t.disabled, false);
assert.equal(t.text, "Ready");
});
test("a state delivered while the page is still looking does not hand Generate back", () => {
// subscribe() delivers the current state at once, and every later change
// redraws. Each of those draws decides the button from the run's phase alone,
// so an idle run -- which is what a page has while it is still finding out
// whether it has a run at all -- re-enabled a button that had just been
// deliberately disabled. Live-looking, and doing nothing when pressed.
const t = transport();
const run = fakeRun();
wire({
run, page: { transport: t }, check: plan(), id: "me",
queuedFor: () => new Promise(() => {}), // never answers
finishedFor: async () => null,
});
assert.equal(t.disabled, true);
run.announce(); // any redraw at all
assert.equal(t.disabled, true, "a redraw handed the button back mid-lookup");
t.draw(run.state);
assert.equal(t.disabled, true, "drawing the idle run handed the button back");
});
test("results of a run go to the bin, and only to the bin", async () => {
// The subscription draws the transport AND hands results on. It hands them to
// the BIN, which owns what is on screen: pointing the viewer at the newest
// image from here as well put two things in charge of one view, and every
// progress message then walked a user's choice of an older result back to the
// newest one.
const taken = [];
const shown = [];
const t = transport();
const run = fakeRun();
wire({
run,
page: {
transport: t,
bin: { absorb: (images) => { taken.push(images); return []; } },
viewer: { setSource: (src, kind) => shown.push([src, kind]) },
},
check: plan(), id: "me",
queuedFor: async () => null, finishedFor: async () => null,
});
const images = [{ filename: "a.png", subfolder: "", type: "output" }];
run.state = { ...run.state, phase: "done", images };
run.announce();
assert.deepEqual(taken[taken.length - 1], images);
assert.deepEqual(shown, [], "the viewer was driven from here as well as from the bin");
});
test("a run found waiting in the queue is adopted as waiting", async () => {
const t = transport();
const run = fakeRun();
const adopted = [];
run.adopt = (id, opts) => { adopted.push([id, opts]); return true; };
const session = wire({
run, page: { transport: t }, check: plan(), id: "me",
queuedFor: async () => ({ promptId: "waiting", running: false }),
finishedFor: async () => null,
});
await session.ready;
assert.deepEqual(adopted, [["waiting", { running: false }]]);
});
test("what the panels hold is sent with every run", async () => {
// Until this was wired, every module panel in the app was decoration: the
// values lived in the browser and nothing ever sent them, so a modifier
// switched on was a modifier the run never heard about.
const run = fakeRun();
const asked = [];
const check = async (body) => { asked.push(body); return plan()(); };
const onGenerate = createGenerator({
run, transport: transport(), check,
values: () => ({ sampling_alg: { enabled: true } }),
});
await onGenerate();
assert.deepEqual(asked[0].values, { sampling_alg: { enabled: true } });
});
test("with no panels holding anything, an empty set is still sent", async () => {
// "Nothing is set" is an answer. Sending nothing at all is a client that
// cannot say the difference, and the server would have to guess.
const run = fakeRun();
const asked = [];
const check = async (body) => { asked.push(body); return plan()(); };
await createGenerator({ run, transport: transport(), check })();
assert.deepEqual(asked[0].values, {});
});
test("a setting that will not be applied is said, and does not stop the run", async () => {
const run = fakeRun();
const t = transport();
const note = "nothing in this pipeline accepts module settings";
const started = await createGenerator({
run, transport: t, check: plan({ notes: [note] }),
})();
assert.equal(started, true, "a note stopped a run that was otherwise fine");
assert.match(t.warning.node.textContent, /accepts module settings/);
assert.equal(t.warning.node.hasAttribute("hidden"), false);
});
test("a note stays up while the run it is about is going", async () => {
// The status line is redrawn by every message a run sends, so a warning put
// there was gone the moment the run said "Queued" -- which is roughly when
// the user looks at it.
const run = fakeRun();
const t = transport();
await createGenerator({ run, transport: t, check: plan({ notes: ["will not be applied"] }) })();
t.draw({ phase: "running", images: [], progress: { value: 1, max: 20 } });
t.draw({ phase: "done", images: [] });
assert.match(t.warning.node.textContent, /will not be applied/);
assert.equal(t.warning.node.hasAttribute("hidden"), false);
});
test("a pipeline with nothing to say takes the note back down", async () => {
const run = fakeRun();
const t = transport();
await createGenerator({ run, transport: t, check: plan({ notes: ["will not be applied"] }) })();
assert.equal(t.warning.node.hasAttribute("hidden"), false);
run.state = { ...run.state, phase: "idle" };
await createGenerator({ run, transport: t, check: plan() })();
assert.equal(t.warning.node.hasAttribute("hidden"), true, "a stale warning stayed up");
assert.equal(t.warning.node.textContent.trim(), "");
});
test("what is typed on the main window is sent with the run", async () => {
const run = fakeRun();
const asked = [];
const check = async (body) => { asked.push(body); return plan()(); };
await createGenerator({
run, transport: transport(), check,
inputs: () => ({ positive: { text: "a cat on a roof" } }),
})();
assert.deepEqual(asked[0].inputs, { positive: { text: "a cat on a roof" } });
});
// --- waitForTerminal ---------------------------------------------------------
test("waitForTerminal resolves with the phase a run actually ends at", async () => {
const run = fakeRun();
const waiting = waitForTerminal(run);
run.state = { ...run.state, phase: "running" };
run.announce();
run.state = { ...run.state, phase: "done" };
run.announce();
assert.equal(await waiting, "done");
});
test("the snapshot delivered at subscribe time is not mistaken for a transition", async () => {
// A page reload adopting a run that already finished delivers ITS finished
// state the moment subscribe() is called -- if that counted, this would
// resolve before the caller's own "start" step had even run.
const run = fakeRun("done");
let resolved = false;
waitForTerminal(run).then(() => { resolved = true; });
await Promise.resolve(); await Promise.resolve();
assert.equal(resolved, false, "resolved from the snapshot, not a real transition");
run.announce(); // the same "done" state, delivered again -- now a real one
await Promise.resolve(); await Promise.resolve();
assert.equal(resolved, true);
});
test("cancel() unsubscribes a waiter whose run never actually started", () => {
// A generation refused before it ever reaches run.start() (an incomplete
// pipeline, a queue that says no) never transitions the run -- nothing
// would ever resolve this promise, and the subscription would sit on the
// run forever without a way out.
const run = fakeRun();
assert.equal(run.subscribers.size, 0);
const waiting = waitForTerminal(run);
assert.equal(run.subscribers.size, 1);
waiting.cancel();
assert.equal(run.subscribers.size, 0, "cancel() did not unsubscribe");
});
test("cancel() after the promise already resolved is harmless", async () => {
const run = fakeRun();
const waiting = waitForTerminal(run);
run.state = { ...run.state, phase: "done" };
run.announce();
assert.equal(await waiting, "done");
assert.doesNotThrow(() => waiting.cancel());
});
+153
View File
@@ -0,0 +1,153 @@
// The transport, driven by run states.
//
// The reason this is one component reading one state object: a button here, a
// progress bar there and a status somewhere else each get their own idea of
// what is happening, and the user believes whichever spoke last.
import test from "node:test";
import assert from "node:assert/strict";
import { setupDom, teardownDom } from "../../composer/tests/_dom.js";
import { IDLE, QUEUED, RUNNING, DONE, FAILED, CANCELLED } from "../run.js";
let createTransport, describe;
test.before(async () => {
setupDom();
await import("../../composer/composer.js");
({ createTransport, describe } = await import("../transport.js"));
});
test.after(() => teardownDom());
const state = (over = {}) => ({
phase: IDLE, promptId: null, progress: null, images: [], audio: [],
error: null, node: null, ...over,
});
const built = (handlers) => {
// Mounted the way the shell mounts it: the controls go in a zone head, and
// there is no bar of its own to append.
const t = createTransport(handlers);
const head = document.createElement("header");
for (const handle of [...t.actions, ...t.status, t.warning]) head.appendChild(handle.node);
document.body.replaceChildren(head);
return t;
};
test("Generate is disabled while a run is in flight and comes back after", () => {
const t = built({});
t.draw(state({ phase: RUNNING }));
assert.equal(t.generate.node.disabled, true);
t.draw(state({ phase: DONE, images: [{ filename: "a.png" }] }));
assert.equal(t.generate.node.disabled, false, "a finished run left Generate dead");
});
test("Cancel is only offered while there is something to cancel", () => {
const t = built({});
assert.equal(t.cancel.node.hasAttribute("hidden"), true, "idle offered a cancel");
t.draw(state({ phase: QUEUED }));
assert.equal(t.cancel.node.hasAttribute("hidden"), false);
t.draw(state({ phase: CANCELLED }));
assert.equal(t.cancel.node.hasAttribute("hidden"), true);
});
test("the progress bar appears only when there is a real measure", () => {
const t = built({});
// Running with nothing to measure yet: a bar at zero reads as "stuck", which
// is worse than no bar.
t.draw(state({ phase: RUNNING }));
assert.equal(t.progress.node.hasAttribute("hidden"), true);
t.draw(state({ phase: RUNNING, progress: { value: 3, max: 20 } }));
assert.equal(t.progress.node.hasAttribute("hidden"), false);
assert.equal(t.progress.node.getAttribute("aria-valuenow"), "3");
assert.equal(t.progress.node.getAttribute("aria-valuemax"), "20");
});
test("the buttons report to the handlers they were given", () => {
let generated = 0, cancelled = 0;
const t = built({ onGenerate: () => { generated += 1; }, onCancel: () => { cancelled += 1; } });
t.generate.node.click();
t.cancel.node.click();
assert.equal(generated, 1);
assert.equal(cancelled, 1);
});
test("a run that saved nothing does not read the same as one that saved something", () => {
// ComfyUI reports execution_success for a graph with no save node too. "Done"
// over an empty preview is the kind of quiet wrong this project keeps finding.
assert.match(describe(state({ phase: DONE, images: [] })), /nothing was saved/);
assert.match(describe(state({ phase: DONE, images: [{}] })), /1 result/);
assert.match(describe(state({ phase: DONE, images: [{}, {}] })), /2 results/);
});
test("a failure says which node and what it said", () => {
const text = describe(state({
phase: FAILED,
error: { node: "KSampler", message: "expected 4 channels" },
}));
assert.match(text, /KSampler/);
assert.match(text, /4 channels/);
});
test("every phase says something, including ones added later", () => {
for (const phase of [IDLE, QUEUED, RUNNING, DONE, FAILED, CANCELLED, "something-new"]) {
const text = describe(state({ phase }));
assert.ok(text && text.trim(), `${phase} draws an empty status`);
}
});
// --- a refusal from the queue, which names the nodes ------------------------
test("a graph the queue refused says which node and why", () => {
// run.js captured ComfyUI's node_errors from the start; describe() only ever
// read the singular `node`, which is set by the WebSocket crash path and
// never by this one. So every value refusal showed ComfyUI's top-level
// string -- "Prompt outputs failed validation" -- over a pipeline that can
// hold a dozen loaders, with nothing saying which to fix.
const said = describe({
phase: FAILED, images: [],
error: {
message: "Prompt outputs failed validation",
nodes: {
vae: {
class_type: "FunPackVAELoader",
errors: [{ type: "value_not_in_list", message: "Value not in list",
details: "vae_name: 'gone.safetensors' not in []" }],
},
},
},
});
assert.match(said, /FunPackVAELoader/);
assert.match(said, /vae/);
assert.match(said, /Value not in list/);
assert.match(said, /gone\.safetensors/);
});
test("a crash during execution still names its one node", () => {
const said = describe({
phase: FAILED, images: [],
error: { node: "sampler", message: "shape mismatch" },
});
assert.equal(said, "sampler: shape mismatch");
});
test("a refusal with no node errors falls back to what it did say", () => {
const said = describe({
phase: FAILED, images: [],
error: { message: "the queue could not be reached: fetch failed", nodes: {} },
});
assert.match(said, /could not be reached/);
});
test("a node error carrying nothing readable does not swallow the message", () => {
// An entry with an empty errors array would otherwise return "" and read as
// a run that failed for no reason at all.
const said = describe({
phase: FAILED, images: [],
error: { message: "refused", nodes: { "4": { class_type: "X", errors: [] } } },
});
assert.equal(said, "refused");
});
+188
View File
@@ -0,0 +1,188 @@
// Generate, and everything that happens after it.
//
// One object owns the whole lifecycle, because the alternative is a button
// here and a progress bar there and a status somewhere else, each with its own
// idea of what is happening. This reads ONE state object and draws it -- so
// "queued" cannot appear next to a finished result, whatever order the messages
// arrive in.
//
// It owns the CONTROLS, not a row: they are handed out and the shell puts them
// in the head of the zone the run fills. A bar across the bottom of the window
// acts on "whatever is in front" and belongs to no region -- which is what made
// this read as a dashboard rather than an editor. v4 puts Generate in the head
// of the zone it produces, and so does this.
//
// Warnings live here too, beside Generate, rather than in the panel that would
// have caused them: a setting that will do nothing has to be said where the run
// is started, not where it was switched on.
import { composer } from "../composer/composer.js";
import { IDLE, QUEUED, RUNNING, DONE, FAILED, CANCELLED } from "./run.js";
const WORKING = new Set([QUEUED, RUNNING]);
export function createTransport({ onGenerate, onCancel } = {}) {
const generate = composer.button.md({
label: "Generate", tone: "primary",
onClick: () => { if (onGenerate) onGenerate(); },
});
const cancel = composer.button.sm({
label: "Cancel", tone: "ghost",
onClick: () => { if (onCancel) onCancel(); },
});
const progress = composer.progress.bar({ value: 0, max: 100, label: "Generating" });
const status = composer.text.sm({ text: "Ready" });
// Something true about the NEXT run, said before it starts and left up while
// it does: a setting that will not be applied is not a status, and putting it
// in the status line meant it was gone the moment the run said "Queued". It
// is here rather than in the panel that would have caused it, because that is
// the panel a person has already stopped looking at.
const warning = composer.banner.warn({ text: "" });
warning.node.setAttribute("hidden", "");
// Something to say that is not about a run in progress: why the last attempt
// did not become one. It has to survive the next draw(), because a refused
// attempt leaves the run exactly where it was -- idle -- and drawing an idle
// run says "Ready", which wiped the reason a moment after it appeared.
let note = null;
// Held for a reason that is not a run: the page is still working out whether
// it already has one. Kept as a flag rather than left to whoever disabled the
// button, because draw() runs on every state change and each of those would
// otherwise hand the button back -- which is exactly what a subscription
// delivering the current state did, one line after the button was disabled.
let held = false;
// Hidden by attribute rather than removed: the row's shape then does not jump
// as a run starts and finishes, and there is one node to find in a test.
const hide = (handle, yes) => {
if (yes) handle.node.setAttribute("hidden", "");
else handle.node.removeAttribute("hidden");
};
hide(progress, true);
hide(cancel, true);
function draw(state) {
// Anything other than idle means a run answered for itself, and what it
// says outranks a note about an attempt that never started.
if (state.phase !== IDLE) note = null;
const working = WORKING.has(state.phase);
generate.setDisabled(working || held);
generate.setBusy(working);
hide(cancel, !working);
const measured = state.progress && state.progress.max;
hide(progress, !(working && measured));
if (measured) {
progress.node.setAttribute("aria-valuemax", String(state.progress.max));
progress.setValue(state.progress.value);
}
status.setText(note || describe(state));
}
/** Say why nothing was started, or -- with null -- stop saying it. Cleared by
* the next run that says otherwise. */
function say(text, state) {
note = text || null;
status.setText(note || describe(state || { phase: IDLE, images: [] }));
}
/** Stop offering Generate for a reason of the page's own, and say why. */
function hold(reason) {
held = Boolean(reason);
if (reason) say(reason);
generate.setDisabled(true);
}
/** Offer it again. The run's own state decides from here. */
function release(state) {
held = false;
say(null, state);
draw(state);
}
/** Say something about the pipeline itself, or -- with null -- stop saying it.
* Unlike say(), this survives a run: it is still true while one is going. */
function warn(text) {
warning.setText(text || "");
warning.node.toggleAttribute("hidden", !text);
}
return {
// What the shell puts in a zone head: the controls that start and stop a
// run, and -- at the far end of the same head -- what it is doing.
actions: [cancel, generate],
// The progress bar goes with the words rather than with the buttons: it is
// the same statement in another form.
status: [progress, status],
warning,
draw, say, warn, hold, release, generate, cancel, progress, statusText: status,
};
}
/**
* Why a run failed, in one line, from whichever of the two shapes it arrived in.
*
* A crash during execution names one node -- `node`. A graph the queue refused
* names as many as were wrong -- `nodes`, an object keyed by node id, each with
* its own reasons. Only the first was ever read, so the whole of the second was
* captured and thrown away: the bar showed ComfyUI's top-level string, which is
* "Prompt outputs failed validation" and says nothing about which of a dozen
* loaders to go and fix.
*/
export function failure(error) {
if (!error) return "the run failed";
const first = firstNodeError(error.nodes);
if (first) return first;
const where = error.node ? `${error.node}: ` : "";
return `${where}${error.message || "the run failed"}`;
}
function firstNodeError(nodes) {
if (!nodes || typeof nodes !== "object") return null;
for (const [id, entry] of Object.entries(nodes)) {
const reasons = entry && Array.isArray(entry.errors) ? entry.errors : [];
for (const reason of reasons) {
const what = reason && (reason.message || reason.type);
if (!what) continue;
// The class type as well as the id: a node id is a number in ComfyUI's
// own graphs and a slot name in ours, and neither on its own is much to
// go on when the message is "Value not in list".
const which = entry.class_type ? `${entry.class_type} (${id})` : id;
const detail = reason.details ? ` — ${reason.details}` : "";
return `${which}: ${what}${detail}`;
}
}
return null;
}
/** One line saying where the run is. The only place these words are decided. */
export function describe(state) {
switch (state.phase) {
case QUEUED:
return "Queued";
case RUNNING: {
const where = state.node ? `node ${state.node}` : "working";
if (state.progress && state.progress.max) {
return `${where} — ${state.progress.value} of ${state.progress.max}`;
}
return where.charAt(0).toUpperCase() + where.slice(1);
}
case DONE:
// The count, because a graph that saved nothing finished successfully and
// produced no picture, and those must not read the same.
return state.images.length
? `Done — ${state.images.length} result${state.images.length === 1 ? "" : "s"}`
: "Done, but nothing was saved";
case FAILED:
return failure(state.error);
case CANCELLED:
return "Cancelled";
default:
return "Ready";
}
}