diff --git a/app/ui/boot.js b/app/ui/boot.js index 7a94fdbe..bee4357f 100644 --- a/app/ui/boot.js +++ b/app/ui/boot.js @@ -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") }); diff --git a/app/ui/shell/api.js b/app/ui/shell/api.js new file mode 100644 index 00000000..5f91462c --- /dev/null +++ b/app/ui/shell/api.js @@ -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)}`), +}; diff --git a/app/ui/shell/client.js b/app/ui/shell/client.js new file mode 100644 index 00000000..9edd661d --- /dev/null +++ b/app/ui/shell/client.js @@ -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; +} diff --git a/app/ui/shell/generate.js b/app/ui/shell/generate.js new file mode 100644 index 00000000..f76b58bf --- /dev/null +++ b/app/ui/shell/generate.js @@ -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(); + }, + }; +} diff --git a/app/ui/shell/pipeline.js b/app/ui/shell/pipeline.js new file mode 100644 index 00000000..fa8578b0 --- /dev/null +++ b/app/ui/shell/pipeline.js @@ -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 }; +} diff --git a/app/ui/shell/pipeline_link.js b/app/ui/shell/pipeline_link.js new file mode 100644 index 00000000..c3f2eff5 --- /dev/null +++ b/app/ui/shell/pipeline_link.js @@ -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 }; +} diff --git a/app/ui/shell/pipeline_state.js b/app/ui/shell/pipeline_state.js new file mode 100644 index 00000000..cfdad34b --- /dev/null +++ b/app/ui/shell/pipeline_state.js @@ -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, + }; +} diff --git a/app/ui/shell/project.js b/app/ui/shell/project.js index 1e220870..cf6ab5d3 100644 --- a/app/ui/shell/project.js +++ b/app/ui/shell/project.js @@ -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) || ""; }, diff --git a/app/ui/shell/reference_wiring.js b/app/ui/shell/reference_wiring.js new file mode 100644 index 00000000..e40be045 --- /dev/null +++ b/app/ui/shell/reference_wiring.js @@ -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 }; +} diff --git a/app/ui/shell/run.js b/app/ui/shell/run.js new file mode 100644 index 00000000..aa04ffe5 --- /dev/null +++ b/app/ui/shell/run.js @@ -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 }); + }, + }; +} diff --git a/app/ui/shell/session.js b/app/ui/shell/session.js new file mode 100644 index 00000000..832e46cb --- /dev/null +++ b/app/ui/shell/session.js @@ -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 + } +} diff --git a/app/ui/shell/tests/client.test.js b/app/ui/shell/tests/client.test.js new file mode 100644 index 00000000..4dd4c00f --- /dev/null +++ b/app/ui/shell/tests/client.test.js @@ -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); +}); diff --git a/app/ui/shell/tests/pipeline.test.js b/app/ui/shell/tests/pipeline.test.js new file mode 100644 index 00000000..dd04b51b --- /dev/null +++ b/app/ui/shell/tests/pipeline.test.js @@ -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 }); +}); diff --git a/app/ui/shell/tests/pipeline_state.test.js b/app/ui/shell/tests/pipeline_state.test.js new file mode 100644 index 00000000..6854c709 --- /dev/null +++ b/app/ui/shell/tests/pipeline_state.test.js @@ -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 +}); diff --git a/app/ui/shell/tests/reference_wiring.test.js b/app/ui/shell/tests/reference_wiring.test.js new file mode 100644 index 00000000..bfe905e0 --- /dev/null +++ b/app/ui/shell/tests/reference_wiring.test.js @@ -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); +}); diff --git a/app/ui/shell/tests/run.test.js b/app/ui/shell/tests/run.test.js new file mode 100644 index 00000000..89d85f61 --- /dev/null +++ b/app/ui/shell/tests/run.test.js @@ -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); +}); diff --git a/app/ui/shell/tests/session.test.js b/app/ui/shell/tests/session.test.js new file mode 100644 index 00000000..152108b6 --- /dev/null +++ b/app/ui/shell/tests/session.test.js @@ -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()); +}); diff --git a/app/ui/shell/tests/transport.test.js b/app/ui/shell/tests/transport.test.js new file mode 100644 index 00000000..375245af --- /dev/null +++ b/app/ui/shell/tests/transport.test.js @@ -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"); +}); diff --git a/app/ui/shell/transport.js b/app/ui/shell/transport.js new file mode 100644 index 00000000..364ace54 --- /dev/null +++ b/app/ui/shell/transport.js @@ -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"; + } +}