From e35030f0a48d70dcba20422123e5c45a0e9239c3 Mon Sep 17 00:00:00 2001 From: Jason Hoku Date: Wed, 4 Nov 2026 15:25:46 -0800 Subject: [PATCH 1/2] fix: ultrareview slice D (waves 2+3) findings - lazy shell: SAVE never persists __uscg_lazy__ (saveState guard + save_manifest pop); the manifest fetch route heals files that already carry it - dashboard node: a Builder session switch does a full reload (session_name from the payload, not the stale widget) instead of re-fetching the old session - dashboard node: window message handlers only accept a dashboard iframe as source - scanned ids are namespaced by session (same relpath in two sessions no longer collides on merge); rescan merge donates each old id once and only via unambiguous basenames - deferred upscale resume is keyed on base_fp == config_fp (fixed-seed grids upscale every cell) - legacy distributed companion ignores the preview pass: master keeps the upscale phase - connected latent: recompute w/h before gen_w/gen_h so 16x models record real dimensions - Start on an identical paused sweep resumes it - outputFromKey keeps subgraph node ids ("57:35") whole Co-Authored-By: Claude Fable 5.1 --- __init__.py | 4 +- directory_scanner.py | 16 ++- generation_orchestrator.py | 40 ++++-- manifest_utils.py | 23 +++- resources/logic_utils.js | 9 +- run_api.py | 2 + tests/js/review_d_c.test.js | 49 +++++++ tests/test_review_d_a.py | 74 ++++++++++ tests/test_review_d_b.py | 130 ++++++++++++++++++ tests/test_review_d_c.py | 32 +++++ .../conf-builder-workflow-sweep.js | 6 +- web/dashboard.js | 19 ++- workflow_sweep_runner.py | 5 + 13 files changed, 385 insertions(+), 24 deletions(-) create mode 100644 tests/js/review_d_c.test.js create mode 100644 tests/test_review_d_a.py create mode 100644 tests/test_review_d_b.py create mode 100644 tests/test_review_d_c.py diff --git a/__init__.py b/__init__.py index 6a081f9..e647d92 100644 --- a/__init__.py +++ b/__init__.py @@ -509,6 +509,8 @@ async def save_manifest(request): data = await request.json() session_name = data.get("session_name") manifest_data = data.get("manifest") # This is from dashboard (may be stale) + if isinstance(manifest_data, dict): + manifest_data.pop("__uscg_lazy__", None) # never persist the lazy-shell flag # --- sanitize --- if session_name: @@ -1331,7 +1333,7 @@ async def scan_directory_route(request): # Run the scan in a thread pool to avoid blocking the event loop loop = asyncio.get_event_loop() items, stats = await loop.run_in_executor( - None, scan_directory_for_images, directory_path, 5000, view_subfolder + None, scan_directory_for_images, directory_path, 5000, view_subfolder, session_name ) from_manifest = stats.get("from_manifest", 0) diff --git a/directory_scanner.py b/directory_scanner.py index b65afe0..cfb7f12 100644 --- a/directory_scanner.py +++ b/directory_scanner.py @@ -11,7 +11,7 @@ import random from PIL import Image from .metadata_packer import extract_metadata_from_image -from .manifest_utils import stable_item_id, scan_relpath, assign_scan_ids +from .manifest_utils import stable_item_id, scan_relpath, assign_scan_ids, scan_key # ============================================================================= @@ -400,7 +400,7 @@ def _probe_mp4_dimensions(file_path): return None -def scan_directory_for_images(directory_path, max_items=5000, view_subfolder=""): +def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="", session_name=""): """ Scan a directory for image files and build manifest-compatible items. @@ -409,6 +409,8 @@ def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="") max_items: Maximum number of images to process view_subfolder: Subfolder path for ComfyUI's /view endpoint (e.g. "benchmarks/session/external_images") + session_name: Optional session name; namespaces the stable scan ids so the same + relpath in two sessions does not collide (blank = legacy ids) Returns: tuple: (items_list, stats_dict) @@ -477,7 +479,7 @@ def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="") item["gen_index"] = gen_idx # Ensure it has an ID if "id" not in item: - item["id"] = stable_item_id("scan:" + item["source_relpath"]) + item["id"] = stable_item_id(scan_key(item["source_relpath"], session_name)) needs_id.append(item) # For videos: ensure media_type is set and probe real dimensions # if the cached entry has missing or stale (default 1280x720) values. @@ -502,7 +504,7 @@ def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="") # No manifest entry โ€” fall through to normal image processing try: - item = _process_single_image(file_path, directory_path, view_subfolder, gen_index=gen_idx) + item = _process_single_image(file_path, directory_path, view_subfolder, gen_index=gen_idx, session_name=session_name) if item: items.append(item) needs_id.append(item) @@ -513,7 +515,7 @@ def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="") # Same files -> same ids on every scan; a hash clash is resolved deterministically. _scan_assigned = {id(i) for i in needs_id} - assign_scan_ids(needs_id, reserved={i["id"] for i in items if id(i) not in _scan_assigned and "id" in i}) + assign_scan_ids(needs_id, reserved={i["id"] for i in items if id(i) not in _scan_assigned and "id" in i}, namespace=session_name) # Count items that had metadata stats["with_metadata"] = sum( @@ -526,7 +528,7 @@ def scan_directory_for_images(directory_path, max_items=5000, view_subfolder="") return items, stats -def _process_single_image(file_path, directory_path, view_subfolder="", gen_index=0): +def _process_single_image(file_path, directory_path, view_subfolder="", gen_index=0, session_name=""): """ Process a single image or video file into a manifest-compatible item dict. """ @@ -536,7 +538,7 @@ def _process_single_image(file_path, directory_path, view_subfolder="", gen_inde # Stable id: derived from the path relative to the scanned folder, so a re-scan keeps it relpath = scan_relpath(file_path, directory_path) - ts = stable_item_id("scan:" + relpath) + ts = stable_item_id(scan_key(relpath, session_name)) # Build URL using ComfyUI's built-in /view endpoint from urllib.parse import quote diff --git a/generation_orchestrator.py b/generation_orchestrator.py index 5fe2d8f..afd39ef 100644 --- a/generation_orchestrator.py +++ b/generation_orchestrator.py @@ -462,11 +462,23 @@ def run_deferred_upscales( # Check if already upscaled (resume support) already_upscaled = False + _job_fp = job.get("config_fp") for item in existing_data.get("items", []): - if (item.get("upscaled") and - item.get("seed") == current_seed and + if not item.get("upscaled"): + continue + if _job_fp: + # keyed on the base cell's identity (sampler/cfg/... included): a fixed-seed grid has + # many cells sharing seed + prompt, and each one needs its own upscale + if item.get("base_fp") == _job_fp: + already_upscaled = True + break + continue + # legacy manifests / queue entries without a fingerprint: seed + prompt, and the same base size + if (item.get("seed") == current_seed and item.get("positive") == actual_positive_prompt and - item.get("width") != pipe_w): + item.get("width") != pipe_w and + item.get("base_width", pipe_w) == pipe_w and + item.get("base_height", pipe_h) == pipe_h): already_upscaled = True break if already_upscaled: @@ -1180,6 +1192,7 @@ def run_generation_loop( print(f"[GridTester] โ„น๏ธ distribution_config is None/empty, running normal generation") # ==== PREVIEW PASS: no upscaling at all (one change point covers inline, deferred, HiRes pre-encode) ==== + session_settings_full = session_settings # pre-override: a legacy distributed companion ignores the preview pass if preview_on: session_settings = {**(session_settings or {}), "upscaling": {"enabled": False}} print(f"[GridTester] ๐Ÿ”Ž Preview pass: {pv_steps} steps, x{pv_scale} scale" @@ -1204,7 +1217,8 @@ def run_generation_loop( session_settings=session_settings, job_order=job_order, preview_on=preview_on, pv_steps=pv_steps, pv_scale=pv_scale, pv_cfg=pv_cfg, meta_run=_meta_run, job_plan=job_plan, - skip_rejected_previews=skip_rejected_previews, only_favorited_previews=only_favorited_previews + skip_rejected_previews=skip_rejected_previews, only_favorited_previews=only_favorited_previews, + upscale_session_settings=session_settings_full ) # The schedule above used `expanded`; the main loop walks job_plan, so the smart cache needs @@ -3382,7 +3396,8 @@ def _run_distributed_generation( optional_positive, optional_negative, optional_latent, model_cache, session_settings=None, job_order="as_written", preview_on=False, pv_steps=8, pv_scale=0.5, pv_cfg=None, - meta_run=None, job_plan=None, skip_rejected_previews=True, only_favorited_previews=False + meta_run=None, job_plan=None, skip_rejected_previews=True, only_favorited_previews=False, + upscale_session_settings=None ): """ Run generation in distributed mode. @@ -3447,6 +3462,14 @@ def _run_distributed_generation( only_favorited=only_favorited_previews ) + # A legacy companion (no populate_from_plan) ignores the preview pass, so its workers render FULL + # images: the master must still upscale them, with the settings from before the preview override. + if (preview_on and upscale_session_settings is not None + and not (job_plan is not None and hasattr(manager, "populate_from_plan"))): + session_settings = upscale_session_settings + print(f"[GridTester] โš ๏ธ Legacy distributed companion ignores the preview pass: workers render full " + f"images, so the master keeps the upscale phase enabled") + # Make manager accessible to API endpoints distribution.set_active_manager(manager) _send_distribution_status(manager) @@ -3706,15 +3729,16 @@ def _run_distributed_generation( # --- Create Latent --- latent_channels = get_latent_channels(loaded_model, optional_latent) latent_ratio = get_latent_downscale_ratio(loaded_model) + if optional_latent is not None: + # pixel size of a connected latent = cells x this model's ratio + w = int(optional_latent["samples"].shape[3]) * latent_ratio + h = int(optional_latent["samples"].shape[2]) * latent_ratio gen_w, gen_h = w, h _pv_scale = (ux.get("render") or {}).get("preview_scale") if _pv_scale and optional_latent is None: from .config_utils import preview_dims gen_w, gen_h = preview_dims(w, h, _pv_scale, latent_ratio) if optional_latent is not None: - # pixel size of a connected latent = cells x this model's ratio - w = int(optional_latent["samples"].shape[3]) * latent_ratio - h = int(optional_latent["samples"].shape[2]) * latent_ratio latent_in = {"samples": optional_latent["samples"].clone()} else: latent_in = make_empty_latent(latent_channels, gen_w, gen_h, latent_ratio) diff --git a/manifest_utils.py b/manifest_utils.py index 3d344e0..a4f93fa 100644 --- a/manifest_utils.py +++ b/manifest_utils.py @@ -388,7 +388,12 @@ def scan_relpath(file_path, directory_path): return os.path.relpath(file_path, directory_path).replace(os.sep, "/") -def assign_scan_ids(items, reserved=()): +def scan_key(relpath, namespace=""): + """Hash key for a scanned item: `scan:`, or `scan::` when a session namespace is given.""" + return f"scan:{namespace}:{relpath}" if namespace else "scan:" + relpath + + +def assign_scan_ids(items, reserved=(), namespace=""): """ Give every item a stable id derived from its `source_relpath`, in place. Items are processed sorted by relpath (not mtime, which changes), `reserved` ids are never reused, and a clash is @@ -405,7 +410,7 @@ def assign_scan_ids(items, reserved=()): for it in sorted(todo, key=lambda x: x["source_relpath"]): salt = 0 while True: - cid = stable_item_id("scan:" + it["source_relpath"], salt) + cid = stable_item_id(scan_key(it["source_relpath"], namespace), salt) if cid not in used: break salt += 1 @@ -425,17 +430,25 @@ def merge_rescan(old, new): return new old_items = [i for i in (old.get("items") or []) if isinstance(i, dict)] by_rel, by_src = {}, {} + src_count = {} for oi in old_items: if oi.get("source_relpath"): by_rel.setdefault(oi["source_relpath"], oi) if oi.get("source_file"): + src_count[oi["source_file"]] = src_count.get(oi["source_file"], 0) + 1 by_src.setdefault(oi["source_file"], oi) + # legacy basename fallback only for basenames that are unambiguous among the old items + by_src = {k: v for k, v in by_src.items() if src_count[k] == 1} + consumed = set() reused, pending = set(), [] for it in new.get("items", []): oi = by_rel.get(it.get("source_relpath")) or by_src.get(it.get("source_file")) + if oi is not None and id(oi) in consumed: + oi = None # an old item donates its id/user fields to ONE new item only if oi is None: pending.append(it) continue + consumed.add(id(oi)) if "id" in oi: it["id"] = oi["id"] reused.add(oi["id"]) @@ -449,7 +462,11 @@ def merge_rescan(old, new): # new files: ids stay stable unless they clash with a reused legacy id fresh = [i for i in pending if isinstance(i.get("id"), int) and i["id"] >= SCAN_ID_BASE and i.get("source_relpath")] keep = {i["id"] for i in pending if i not in fresh and "id" in i} - assign_scan_ids(fresh, reserved=reused | keep) + # keep whichever key family the scanner used (session-namespaced, or legacy un-namespaced) + ns = new.get("session_name") or "" + if ns and not any(i["id"] == stable_item_id(scan_key(i["source_relpath"], ns)) for i in fresh): + ns = "" + assign_scan_ids(fresh, reserved=reused | keep, namespace=ns) if isinstance(old.get("comparisons"), list): new["comparisons"] = old["comparisons"] old_meta = old.get("meta") diff --git a/resources/logic_utils.js b/resources/logic_utils.js index 4016b6b..e639d5e 100644 --- a/resources/logic_utils.js +++ b/resources/logic_utils.js @@ -38,13 +38,20 @@ function setupKeyReloadFullscreen() { // Persist session to disk async function saveState() { + // The lazy shell is empty on purpose and carries __uscg_lazy__: saving it would persist the flag. + if (fullManifest && fullManifest.__uscg_lazy__) { + console.log('[Save] skipped: manifest still loading (lazy shell)'); + return; + } + const toSave = Object.assign({}, fullManifest); + delete toSave.__uscg_lazy__; try { await fetch('/config_tester/save_manifest', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ session_name: document.getElementById('session-input')?.value || "default", - manifest: fullManifest + manifest: toSave }) }); } catch (e) { console.error("Save failed", e); } diff --git a/run_api.py b/run_api.py index b657a26..02d1a35 100644 --- a/run_api.py +++ b/run_api.py @@ -65,6 +65,8 @@ def load_manifest_for_fetch(benchmarks_base, raw_session, if_none_match=None): return 503, json.dumps({"error": "manifest is being written, retry"}).encode(), {"Retry-After": "1"} if (st2.st_mtime_ns, st2.st_size) != (st1.st_mtime_ns, st1.st_size): return 503, json.dumps({"error": "manifest changed while reading, retry"}).encode(), {"Retry-After": "1"} + if isinstance(data, dict): + data.pop("__uscg_lazy__", None) # heal files corrupted by a SAVE on the lazy shell body = json.dumps(data, separators=(",", ":"), ensure_ascii=False).encode("utf-8") return 200, body, headers diff --git a/tests/js/review_d_c.test.js b/tests/js/review_d_c.test.js new file mode 100644 index 0000000..c5416bb --- /dev/null +++ b/tests/js/review_d_c.test.js @@ -0,0 +1,49 @@ +// Review-d (C): outputFromKey must survive subgraph node ids ("57:35"); saveState must not persist the lazy shell. +const test = require('node:test'); +const assert = require('node:assert'); +const fs = require('fs'); +const path = require('path'); + +function grab(src, name) { + const m = new RegExp('(?:export )?(?:async )?function ' + name + '[(]').exec(src); + assert.ok(m, name + ' not found'); + let i = src.indexOf('{', m.index), depth = 0, j = i; + for (; j < src.length; j++) { + if (src[j] === '{') depth++; + else if (src[j] === '}') { depth--; if (depth === 0) break; } + } + return src.slice(m.index, j + 1).replace(/^export /, ''); +} + +const sweep = fs.readFileSync(path.join(__dirname, '../../web/conf_builder/conf-builder-workflow-sweep.js'), 'utf8'); +const { outputKey, outputFromKey } = new Function(grab(sweep, 'outputKey') + '\n' + grab(sweep, 'outputFromKey') + + '\nreturn { outputKey, outputFromKey };')(); + +test('subgraph node id round-trips', () => { + const img = { node_id: '57:35', output_index: 1 }; + assert.deepStrictEqual(outputFromKey(outputKey(img)), img); + const vid = { node_id: '57:35', output_index: 0, media: 'video', frames_from: '12:3' }; + assert.deepStrictEqual(outputFromKey(outputKey(vid)), vid); + assert.deepStrictEqual(outputFromKey(outputKey({ node_id: '8', output_index: 0 })), { node_id: '8', output_index: 0 }); +}); + +const utils = fs.readFileSync(path.join(__dirname, '../../resources/logic_utils.js'), 'utf8'); +function runSave(manifest) { + const calls = []; + const f = new Function('fullManifest', 'fetch', 'document', 'console', + grab(utils, 'saveState') + '\nreturn saveState();'); + return f(manifest, async (u, o) => { calls.push(JSON.parse(o.body)); }, + { getElementById: () => ({ value: 's1' }) }, { log() {}, error() {} }).then(() => calls); +} + +test('saveState does nothing on the lazy shell', async () => { + assert.deepStrictEqual(await runSave({ items: [], __uscg_lazy__: true, session_name: 's1' }), []); +}); + +test('saveState sends a copy without the lazy flag', async () => { + const m = { items: [{ id: 1 }], meta: {} }; + const calls = await runSave(m); + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].manifest.items.length, 1); + assert.ok(!('__uscg_lazy__' in calls[0].manifest)); +}); diff --git a/tests/test_review_d_a.py b/tests/test_review_d_a.py new file mode 100644 index 0000000..e8bad9a --- /dev/null +++ b/tests/test_review_d_a.py @@ -0,0 +1,74 @@ +"""Review-d findings 3 + 6: session-namespaced scan ids, merge_rescan basename fallback.""" +import importlib.util +import os + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_PKG = os.path.basename(_ROOT) + + +def _load(name, modname): + spec = importlib.util.spec_from_file_location(modname, os.path.join(_ROOT, name)) + m = importlib.util.module_from_spec(spec) + m.__package__ = _PKG + spec.loader.exec_module(m) + return m + + +mu = _load("manifest_utils.py", "manifest_utils_review_d_a") + + +def _items(*rels): + return [{"source_relpath": r, "source_file": r.split("/")[-1]} for r in rels] + + +def test_ids_differ_across_sessions_for_same_relpath(): + a = mu.assign_scan_ids(_items("ComfyUI_00001_.png"), namespace="sess-a")[0]["id"] + b = mu.assign_scan_ids(_items("ComfyUI_00001_.png"), namespace="sess-b")[0]["id"] + assert a != b + + +def test_same_session_same_relpath_is_stable(): + a = mu.assign_scan_ids(_items("x/ComfyUI_00001_.png"), namespace="sess-a")[0]["id"] + b = mu.assign_scan_ids(_items("x/ComfyUI_00001_.png"), namespace="sess-a")[0]["id"] + assert a == b + assert a == mu.stable_item_id("scan:sess-a:x/ComfyUI_00001_.png") + + +def test_no_namespace_keeps_legacy_key(): + got = mu.assign_scan_ids(_items("c.png"))[0]["id"] + assert got == mu.stable_item_id("scan:c.png") + + +def test_merge_rescan_same_basename_no_duplicate_ids(): + # legacy old items: no source_relpath, only source_file; one old item, two new same-basename files + old = {"items": [{"id": 111, "source_file": "a.png", "favorited": True}]} + new = {"session_name": "s", "items": [ + {"id": mu.stable_item_id("scan:s:a.png"), "source_relpath": "a.png", "source_file": "a.png"}, + {"id": mu.stable_item_id("scan:s:images/a.png"), "source_relpath": "images/a.png", "source_file": "a.png"}, + ]} + mu.merge_rescan(old, new) + ids = [i["id"] for i in new["items"]] + assert len(set(ids)) == 2 + assert ids.count(111) == 1 + assert sum(1 for i in new["items"] if i.get("favorited")) == 1 + + +def test_non_unique_basenames_do_not_match_via_source_file(): + old = {"items": [ + {"id": 1, "source_file": "a.png", "favorited": True}, + {"id": 2, "source_file": "a.png"}, + ]} + new = {"session_name": "s", "items": [ + {"id": mu.stable_item_id("scan:s:a.png"), "source_relpath": "a.png", "source_file": "a.png"}, + ]} + mu.merge_rescan(old, new) + assert new["items"][0]["id"] not in (1, 2) + assert not new["items"][0].get("favorited") + + +def test_merge_rescan_fresh_ids_use_session_namespace(): + old = {"items": []} + new = {"session_name": "s", "items": _items("a.png")} + new["items"][0]["id"] = mu.stable_item_id("scan:s:a.png") + mu.merge_rescan(old, new) + assert new["items"][0]["id"] == mu.stable_item_id("scan:s:a.png") diff --git a/tests/test_review_d_b.py b/tests/test_review_d_b.py new file mode 100644 index 0000000..f795ccf --- /dev/null +++ b/tests/test_review_d_b.py @@ -0,0 +1,130 @@ +"""Review-d fixes: deferred-upscale resume keyed on base_fp, legacy-companion preview upscales, +latent dimension ordering, and a paused identical sweep being resumed by Start.""" +import importlib.util +import os +import re +import sys +import types + +import pytest + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, _ROOT) + +import generation_orchestrator as go # noqa: E402 +import workflow_sweep_runner as runner # noqa: E402 + +_spec = importlib.util.spec_from_file_location("_mu_real_rdb", os.path.join(_ROOT, "manifest_utils.py")) +_mu = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mu) + + +# ---- deferred upscale resume check --------------------------------------------------------------------------- + +def _job(fp, tmp_path, w=512, h=512, seed=7): + (tmp_path / "base.webp").write_bytes(b"not an image") # makes it past the file check, fails at decode + base = {"id": 1, "file": "/view?filename=base.webp&type=output", "seed": seed, + "width": w, "height": h, "positive": "p", "filename": "base.webp"} + j = {"config": {}, "width": w, "height": h, "seed": seed, "batch_idx": 0, + "actual_positive_prompt": "p", "actual_negative_prompt": "n", "gen_index": 0, "base_item": base} + if fp: + j["config_fp"] = fp + return j + + +def _run(monkeypatch, capsys, tmp_path, job, items): + monkeypatch.setattr(go, "item_filename", _mu.item_filename) + ig = [m for n, m in sys.modules.items() if n.endswith("image_generation") and hasattr(m, "generate_image")][0] + monkeypatch.setattr(ig, "upscale_image", lambda *a, **kw: None, raising=False) + monkeypatch.setattr(sys.modules["comfy.model_management"], "processing_interrupted", + lambda: False, raising=False) + go.run_deferred_upscales([job], {"pipelines": [{}]}, {}, None, None, None, {"positive": {}, "negative": {}}, + {"images": str(tmp_path)}, {"items": items}, "s", "u", None) + out = capsys.readouterr().out + return "already upscaled" in out + + +def _up(base_fp=None, seed=7, w=1024, **kw): + it = {"upscaled": True, "seed": seed, "positive": "p", "width": w, "height": w} + if base_fp: + it["base_fp"] = base_fp + it.update(kw) + return it + + +def test_resume_keyed_on_base_fp_not_seed(monkeypatch, capsys, tmp_path): + # two cells share seed + prompt; only cell A has been upscaled + items = [_up(base_fp="fpA")] + assert _run(monkeypatch, capsys, tmp_path, _job("fpA", tmp_path), items) is True + assert _run(monkeypatch, capsys, tmp_path, _job("fpB", tmp_path), items) is False + # both upscaled -> both skipped + items = [_up(base_fp="fpA"), _up(base_fp="fpB")] + assert _run(monkeypatch, capsys, tmp_path, _job("fpA", tmp_path), items) is True + assert _run(monkeypatch, capsys, tmp_path, _job("fpB", tmp_path), items) is True + + +def test_fingerprinted_job_ignores_legacy_upscaled_items(monkeypatch, capsys, tmp_path): + assert _run(monkeypatch, capsys, tmp_path, _job("fpA", tmp_path), [_up()]) is False + + +def test_legacy_fallback_seed_prompt_and_base_dims(monkeypatch, capsys, tmp_path): + j = _job(None, tmp_path) + assert _run(monkeypatch, capsys, tmp_path, j, [_up()]) is True + assert _run(monkeypatch, capsys, tmp_path, j, [_up(seed=8)]) is False + # base dims recorded on the upscaled item must match the job's + assert _run(monkeypatch, capsys, tmp_path, j, [_up(base_width=512, base_height=512)]) is True + assert _run(monkeypatch, capsys, tmp_path, j, [_up(base_width=640, base_height=512)]) is False + + +# ---- source guards (heavy paths cannot run under the stubs) --------------------------------------------------- + +def _text(name): + with open(os.path.join(_ROOT, name), encoding="utf-8") as f: + return f.read() + + +def test_preview_override_keeps_full_settings_for_legacy_companion(): + src = _text("generation_orchestrator.py") + assert src.index("session_settings_full = session_settings") < src.index('"upscaling": {"enabled": False}}') + assert "upscale_session_settings=session_settings_full" in src + m = re.search(r"\ndef _run_distributed_generation\(.*?(?=\ndef |\Z)", src, re.S).group(0) + legacy = m.index("session_settings = upscale_session_settings") + assert 'hasattr(manager, "populate_from_plan")' in m[:legacy] + assert legacy < m.index("Running upscales on master") + + +def test_latent_dims_recomputed_before_gen_dims(): + # the local loop needs a model + sampler, so check the statement order in the source + src = _text("generation_orchestrator.py") + i = src.index('optional_latent["samples"].shape[3]) * latent_ratio') + j = src.index("gen_w, gen_h = w, h", i - 400) + assert i < j + assert src.count("gen_w, gen_h = w, h\n _pv_scale") == 1 + + +# ---- sweep: Start on an identical paused sweep resumes it ------------------------------------------------------ + +def test_start_sweep_resumes_identical_paused_sweep(): + sys.path.insert(0, os.path.join(_ROOT, "tests")) + import test_workflow_sweep_runner as t + mp = pytest.MonkeyPatch() + import folder_paths + import tempfile + tmp = tempfile.mkdtemp() + mp.setattr(folder_paths, "get_output_directory", lambda: tmp, raising=False) + mp.setattr(runner, "manifest_utils", t._real_manifest_utils(), raising=False) + runner.reset_registry() + try: + bridge = t.FakeBridge() + env = types.SimpleNamespace(bridge=bridge) + sw = t.make(env) + assert runner.pause(sw.id)[0] and sw.state == "paused" + again = runner.start_sweep([t.entry()], {"session_name": "wf_test"}, "n2", None, bridge, class_info=None) + assert again is sw and sw.state == "running" + # a running one is still just attached + n = len(bridge.enqueued) + same = runner.start_sweep([t.entry()], {"session_name": "wf_test"}, "n3", None, bridge, class_info=None) + assert same is sw and sw.state == "running" and len(bridge.enqueued) == n + finally: + runner.reset_registry() + mp.undo() diff --git a/tests/test_review_d_c.py b/tests/test_review_d_c.py new file mode 100644 index 0000000..c221810 --- /dev/null +++ b/tests/test_review_d_c.py @@ -0,0 +1,32 @@ +"""Review-d (C): the lazy-shell flag must never persist nor be served.""" +import importlib.util +import json +import os + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_PKG = os.path.basename(_ROOT) + + +def _load_run_api(): + spec = importlib.util.spec_from_file_location(f"{_PKG}.run_api_real_rdc", os.path.join(_ROOT, "run_api.py")) + mod = importlib.util.module_from_spec(spec) + mod.__package__ = _PKG + spec.loader.exec_module(mod) + return mod + + +def test_fetch_heals_flagged_manifest(tmp_path): + base = tmp_path / "benchmarks" + (base / "s1").mkdir(parents=True) + with open(base / "s1" / "manifest.json", "w", encoding="utf-8") as f: + json.dump({"session_name": "s1", "meta": {}, "items": [{"id": 1}], "__uscg_lazy__": True}, f) + status, body, _ = _load_run_api().load_manifest_for_fetch(str(base), "s1") + assert status == 200 + data = json.loads(body) + assert "__uscg_lazy__" not in data and data["items"] == [{"id": 1}] + + +def test_save_manifest_route_drops_flag(): + src = open(os.path.join(_ROOT, "__init__.py"), encoding="utf-8").read() + seg = src.split("async def save_manifest(request)")[1].split("@server.PromptServer")[0] + assert 'manifest_data.pop("__uscg_lazy__", None)' in seg diff --git a/web/conf_builder/conf-builder-workflow-sweep.js b/web/conf_builder/conf-builder-workflow-sweep.js index 148ecd0..4945946 100644 --- a/web/conf_builder/conf-builder-workflow-sweep.js +++ b/web/conf_builder/conf-builder-workflow-sweep.js @@ -107,7 +107,11 @@ export function outputKey(o) { /** Inverse of outputKey -> the stored `wf.output` shape. Image outputs carry only node_id/output_index. */ export function outputFromKey(key) { - const [n, i, media, frames] = String(key).split(":"); + // Subgraph node ids contain ":" ("57:35"), so the id is everything before the last three fields + // (index, media, frames_from); frames_from may itself be a subgraph id. + const m = /^(.*?):(\d+):(image|video):(.*)$/.exec(String(key)); + const parts = String(key).split(":"); + const [n, i, media, frames] = m ? [m[1], m[2], m[3], m[4]] : [parts.slice(0, -3).join(":") || parts[0], ...parts.slice(-3)]; if (n === undefined || n === "") return null; const out = { node_id: n, output_index: parseInt(i, 10) || 0 }; if (media === "video") { diff --git a/web/dashboard.js b/web/dashboard.js index e817ccc..91ae2cd 100644 --- a/web/dashboard.js +++ b/web/dashboard.js @@ -149,7 +149,14 @@ app.registerExtension({ // 3. FULLSCREEN + CTRL+S LISTENER (SINGLETON - only register once) if (!window.__ultimateGridFullscreenListenerInstalled) { + // Only the iframes of dashboard nodes may drive these handlers. + const isDashboardSource = (event) => { + const src = event.source; + if (!src) return false; + return (app.graph?._nodes || []).some(n => n.iframe?.contentWindow === src); + }; window.addEventListener("message", (event) => { + if (!isDashboardSource(event)) return; // Forward Ctrl+S from dashboard iframe to ComfyUI's save workflow if (event.data?.type === 'forward_ctrl_s') { document.dispatchEvent(new KeyboardEvent('keydown', { @@ -419,8 +426,14 @@ app.registerExtension({ const currentSession = getSessionName(); // Optimization: If session already loaded, try incremental update - if (node.loaded_session === currentSession) { - const newHtml = message.html[0]; + // The widget can be stale when the connected Builder switched the session, so + // trust the session_name inside the HTML payload when it has one. + const newHtml = message.html[0]; + const sm = /\/\*__JSON_START__\*\/\s*let fullManifest = (\{[\s\S]*?\});\s*\/\*__JSON_END__\*\//.exec(newHtml); + let payloadSession = null; + if (sm && sm[1]) { try { payloadSession = JSON.parse(sm[1]).session_name || null; } catch (e) { /* ignore */ } } + const targetSession = payloadSession ? String(payloadSession) : currentSession; + if (node.loaded_session === targetSession) { // Try to extract just the manifest data const pattern = /\/\*__JSON_START__\*\/\s*let fullManifest = (\{[\s\S]*?\});\s*\/\*__JSON_END__\*\//; @@ -460,7 +473,7 @@ app.registerExtension({ // Full reload (when incremental fails or first load) node.iframe.srcdoc = message.html[0]; - node.loaded_session = currentSession; + node.loaded_session = targetSession; console.log('[UltimateGrid] ๐Ÿ”„ Full reload'); } diff --git a/workflow_sweep_runner.py b/workflow_sweep_runner.py index 3dbb38e..97b321f 100644 --- a/workflow_sweep_runner.py +++ b/workflow_sweep_runner.py @@ -507,6 +507,11 @@ def start_sweep(configs, session_settings=None, origin_node=None, client_id=None sw.bridge.run_on_loop(validate_first(sw)) return launch(sw) except SweepAlreadyRunning as e: + if e.sweep.state == "paused": + ok, msg = resume(e.sweep.id) + print(f"[GridTester] โ–ถ๏ธ Same workflow sweep was paused ({e.sweep.id}, session " + f"'{e.sweep.session}'): resumed ({msg})") + return e.sweep print(f"[GridTester] ๐Ÿ” Same workflow sweep already running ({e.sweep.id}, session " f"'{e.sweep.session}'): attaching, nothing re-queued") return e.sweep From f88129cb373ffcedf0c126a5e3614d9c30e3bda6 Mon Sep 17 00:00:00 2001 From: Jason Hoku Date: Wed, 4 Nov 2026 16:44:19 -0800 Subject: [PATCH 2/2] fix: ultrareview slice C (workflow sweep) findings - append_items never replaces an unreadable manifest with a fresh one (raises, file untouched) and writes through the retrying atomic helper; the sweep spec write retries PermissionError too - breaker only pauses while jobs remain; a paused sweep with nothing in flight or pending completes instead of blocking new sweeps with SweepBusy - sampler IS_CHANGED returns NaN when configs_json is linked (absent from kwargs), so a re-queued sweep is never served from ComfyUI's cache - Start At Job # = N skips N-1 jobs like the orchestrator - dry_run / api_start expand the sweep in an executor, off the event loop - Builder sweep section re-reads configArray.workflow on every render (load/remove replaced it) - favorites export and GENERATE NEW skip workflow-sweep items (no template) - Pairwise Review excludes all wf_* per-job keys - test: same-sweep attach waits for the feeder's first enqueue (pre-existing race) Co-Authored-By: Claude Fable 5.1 --- manifest_utils.py | 11 +- resources/logic_ratings.js | 4 +- resources/logic_utils.js | 19 ++- sampler_node.py | 4 +- tests/js/ratings.test.js | 4 +- tests/js/review_c_js.test.js | 33 +++++ tests/test_review_c_py.py | 113 ++++++++++++++++++ tests/test_workflow_sweep_collect.py | 7 +- tests/test_workflow_sweep_runner.py | 6 +- .../conf-builder-workflow-sweep.js | 4 +- workflow_sweep_runner.py | 28 ++++- 11 files changed, 212 insertions(+), 21 deletions(-) create mode 100644 tests/js/review_c_js.test.js create mode 100644 tests/test_review_c_py.py diff --git a/manifest_utils.py b/manifest_utils.py index a4f93fa..a1cfd7e 100644 --- a/manifest_utils.py +++ b/manifest_utils.py @@ -301,11 +301,15 @@ def append_items(manifest_path, items, remove_ids=(), session_name=None, default with _manifest_lock: data = None if os.path.exists(manifest_path): + # An unreadable existing manifest (half-written by a dashboard save, say) must NOT be replaced by a + # fresh one holding only the new item: that wipes every earlier item/favourite/note. Fail the job. try: with open(manifest_path, "r") as f: data = json.load(f) except Exception as e: - print(f"[GridTester] โš ๏ธ Could not read manifest before append: {e}") + raise RuntimeError(f"could not read the manifest {manifest_path} before appending (left untouched): {e}") from e + if not isinstance(data, dict): + raise RuntimeError(f"the manifest {manifest_path} is not a JSON object (left untouched)") if not isinstance(data, dict): data = {"session_name": session_name or "default", "items": []} data.setdefault("items", []) @@ -334,10 +338,7 @@ def append_items(manifest_path, items, remove_ids=(), session_name=None, default kept.append(it) data["items"] = kept os.makedirs(os.path.dirname(os.path.abspath(manifest_path)), exist_ok=True) - tmp = manifest_path + ".tmp" - with open(tmp, "w") as f: - json.dump(data, f, indent=4) - os.replace(tmp, manifest_path) + _atomic_write_json(manifest_path, data) return data diff --git a/resources/logic_ratings.js b/resources/logic_ratings.js index 71ee5a2..ddf4885 100644 --- a/resources/logic_ratings.js +++ b/resources/logic_ratings.js @@ -24,10 +24,10 @@ var CMP_EXCLUDED_KEYS = new Set([ // per-image outcomes / bookkeeping, not settings 'config_fp', 'base_fp', 'metrics', 'preview_of', 'source_relpath', 'dist_worker', 'florence2_bbox', 'florence2_detection_count', 'florence2_no_detection', - // workflow-sweep per-job bookkeeping (wf_params are the settings and stay in) + // workflow-sweep per-job bookkeeping (all wf_* keys are also dropped by CMP_EXCLUDED_RE: they differ for every pair) 'wf_fingerprint', 'wf_sweep_id', 'wf_job_index', 'poster' ]); -var CMP_EXCLUDED_RE = /^upscale_|^hires_prompt_|_hash$|^worker|^_/; +var CMP_EXCLUDED_RE = /^upscale_|^hires_prompt_|^wf_|_hash$|^worker|^_/; // Phone swipe-to-vote: which side a finger gesture picks (same mapping as ArrowLeft = A, ArrowRight = B). var CMP_SWIPE = { minPx: 60, frac: 0.15, ratio: 1.5, maxMs: 800 }; diff --git a/resources/logic_utils.js b/resources/logic_utils.js index e639d5e..df711f4 100644 --- a/resources/logic_utils.js +++ b/resources/logic_utils.js @@ -455,8 +455,11 @@ function exportFavoritesAsConfigJSON() { // single values for sampler, scheduler, model, etc. (NOT arrays). var flatConfigs = []; var seen = {}; + var skippedWorkflow = 0; for (var fi = 0; fi < favorites.length; fi++) { var d = favorites[fi]; + // Workflow-sweep items have no template: a config built from them would be routed to start_sweep and fail + if (d.model_type === 'workflow') { skippedWorkflow++; continue; } // Build the flat config from the manifest item's actual fields var flatConfig = { @@ -512,11 +515,20 @@ function exportFavoritesAsConfigJSON() { flatConfigs.push(flatConfig); } + if (flatConfigs.length === 0 && skippedWorkflow) { + alert(skippedWorkflow + ' workflow-sweep item' + (skippedWorkflow !== 1 ? 's' : '') + ' skipped: they have no template. Nothing to export.'); + return; + } + // Wrap in the format the sampler node expects: {"configs": [...]} var configOutput = { configs: flatConfigs }; + var wfNote = skippedWorkflow + ? '\n\n' + skippedWorkflow + ' workflow-sweep item' + (skippedWorkflow !== 1 ? 's' : '') + ' skipped: they have no template.' + : ''; + // Check for GGUF/diffusion models that need a VAE connected var needsVaeWarning = flatConfigs.some(function(c) { return c.model_type && c.model_type !== 'checkpoint' && (!c.vae || c.vae === 'Default' || c.vae === 'None'); @@ -528,7 +540,7 @@ function exportFavoritesAsConfigJSON() { // Copy to clipboard var jsonStr = JSON.stringify(configOutput, null, 2); navigator.clipboard.writeText(jsonStr).then(function() { - alert('Config JSON copied to clipboard! (' + favorites.length + ' favorites \u2192 ' + flatConfigs.length + ' config' + (flatConfigs.length !== 1 ? 's' : '') + ')\n\nPaste into the Sampler Grid node\'s configs_json input.' + vaeNote); + alert('Config JSON copied to clipboard! (' + favorites.length + ' favorites \u2192 ' + flatConfigs.length + ' config' + (flatConfigs.length !== 1 ? 's' : '') + ')\n\nPaste into the Sampler Grid node\'s configs_json input.' + vaeNote + wfNote); }).catch(function() { // Fallback: show in a prompt dialog prompt('Copy this Config JSON:', jsonStr); @@ -817,6 +829,11 @@ async function triggerGen(btn) { // Build a complete config from the modal fields + the original item's metadata. const d = activeData ? activeData.find(x => x.id === window.currentModalId) : null; + if (d && d.model_type === 'workflow') { + alert('Workflow-sweep items cannot be regenerated here: they have no template.'); + return; + } + const config = { sampler: document.getElementById('f-smp').value, scheduler: document.getElementById('f-sch').value, diff --git a/sampler_node.py b/sampler_node.py index d17ba72..a604245 100644 --- a/sampler_node.py +++ b/sampler_node.py @@ -113,7 +113,9 @@ class SamplerGridTester: # Workflow Sweep configs: a re-queue must restart/resume the sweep (finished jobs are skipped by # fingerprint), never hit ComfyUI's node cache - if _workflow_configs(kwargs.get("configs_json")): + # ComfyUI calls IS_CHANGED with outputs=None, so a LINKED configs_json (wired from the Builder, the + # normal setup) is absent from kwargs: treat "absent" as "linked" and never cache. + if "configs_json" not in kwargs or _workflow_configs(kwargs.get("configs_json")): return float("NaN") # Check if any optional input is connected (not None and present in kwargs) diff --git a/tests/js/ratings.test.js b/tests/js/ratings.test.js index 4de84df..bc6760d 100644 --- a/tests/js/ratings.test.js +++ b/tests/js/ratings.test.js @@ -305,7 +305,7 @@ test('cmpSwipeSide decisions', () => { assert.strictEqual(R.cmpSwipeSide(-250, 0, 200, 1600), 'a'); }); -test('sweep per-job bookkeeping is excluded from the compare ctx (wf_params stay in)', () => { +test('sweep per-job bookkeeping is excluded from the compare ctx (all wf_* keys via CMP_EXCLUDED_RE)', () => { for (const k of ['wf_fingerprint', 'wf_sweep_id', 'wf_job_index', 'poster']) assert.ok(R.CMP_EXCLUDED_KEYS.has(k), k); - assert.ok(!R.CMP_EXCLUDED_KEYS.has('wf_params')); + assert.ok(R.CMP_EXCLUDED_RE.test('wf_params')); }); diff --git a/tests/js/review_c_js.test.js b/tests/js/review_c_js.test.js new file mode 100644 index 0000000..d357592 --- /dev/null +++ b/tests/js/review_c_js.test.js @@ -0,0 +1,33 @@ +'use strict'; +const test = require('node:test'); +const assert = require('node:assert'); +const fs = require('fs'); +const path = require('path'); +const R = require('../../resources/logic_ratings.js'); + +const base = { file: '/view?x', model: 'm', lora: 'None', sampler: 'euler', scheduler: 'simple', cfg: 7, steps: 20, + denoise: 1, seed: 1, width: 1024, height: 1024, config_positive: 'a cat', negative: '' }; + +test('CMP_EXCLUDED_RE drops every wf_* key', () => { + for (const k of ['wf_subs', 'wf_params', 'wf_source', 'wf_output', 'wf_output_media', 'wf_frames_from']) { + assert.ok(R.CMP_EXCLUDED_RE.test(k), k); + } + assert.ok(!R.CMP_EXCLUDED_RE.test('sampler')); +}); + +test('sweep items differing only in wf_* keys have equal compare signatures', () => { + const a = Object.assign({ id: 1, model_type: 'workflow' }, base, { wf_subs: { x: 1 }, wf_params: { 'K.cfg': 4 }, wf_source: 'a.json', wf_output: '3:0' }); + const b = Object.assign({ id: 2, model_type: 'workflow' }, base, { wf_subs: { x: 2 }, wf_params: { 'K.cfg': 5 }, wf_source: 'b.json', wf_output: '9:0' }); + assert.strictEqual(JSON.stringify(R.itemSig(a)), JSON.stringify(R.itemSig(b))); +}); + +test('web workflow-sweep render re-reads configArray.workflow (no stale const wf)', () => { + const src = fs.readFileSync(path.join(__dirname, '../../web/conf_builder/conf-builder-workflow-sweep.js'), 'utf8'); + assert.match(src, /let wf = ensureWorkflowState\(configArray\)/); + assert.match(src, /function render\(\) \{\s*wf = ensureWorkflowState\(configArray\)/); +}); + +test('logic_utils export/generate skip model_type workflow (source check; functions need a DOM)', () => { + const src = fs.readFileSync(path.join(__dirname, '../../resources/logic_utils.js'), 'utf8'); + assert.strictEqual((src.match(/model_type === 'workflow'/g) || []).length, 2); +}); diff --git a/tests/test_review_c_py.py b/tests/test_review_c_py.py new file mode 100644 index 0000000..3d056bc --- /dev/null +++ b/tests/test_review_c_py.py @@ -0,0 +1,113 @@ +"""Review fixes (python side): manifest append safety, breaker completion, IS_CHANGED for linked inputs, Start At Job #.""" +import asyncio +import json +import os +import sys + +import pytest + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) +sys.path.insert(0, os.path.join(ROOT, "tests")) + +import workflow_sweep_runner as runner # noqa: E402 +from test_workflow_sweep_runner import ( # noqa: E402,F401 + env, entry, make, drive, _real_manifest_utils, _cfg_body) +from test_workflow_sweep_generator import sampler_node # noqa: E402,F401 + + +def test_append_items_corrupt_manifest_raises_and_is_untouched(tmp_path): + mu = _real_manifest_utils() + path = tmp_path / "manifest.json" + path.write_text('{"session_name": "s", "items": [{"id": 1}', encoding="utf-8") # half-written + before = path.read_bytes() + with pytest.raises(RuntimeError) as ei: + mu.append_items(str(path), [{"id": 2}], session_name="s") + assert str(path) in str(ei.value) + assert path.read_bytes() == before + + +def test_append_items_creates_missing_and_keeps_existing(tmp_path): + mu = _real_manifest_utils() + path = tmp_path / "sub" / "manifest.json" + data = mu.append_items(str(path), [{"id": 1}], session_name="s") + assert [i["id"] for i in data["items"]] == [1] + mu.append_items(str(path), [{"id": 2}], session_name="s") + assert [i["id"] for i in json.loads(path.read_text())["items"]] == [2, 1] + assert not (tmp_path / "sub" / "manifest.json.tmp").exists() + + +def _fail(sw, bridge): + drive(sw, bridge) + for pid in list(bridge.queue): + bridge.start(pid) + for pid in list(bridge.running): + bridge.finish(pid, "error") + drive(sw, bridge) + + +def test_breaker_does_not_pause_when_no_jobs_remain(env): + sw = make(env, [entry((1, 2, 3))]) + for _ in range(10): + if not runner.tick(sw): + break + for pid in list(env.bridge.queue): + env.bridge.start(pid) + for pid in list(env.bridge.running): + env.bridge.finish(pid, "error") + assert sw.state == "completed" and "3 job(s) failed" in sw.reason + + +def test_paused_sweep_with_nothing_left_finishes(env): + sw = make(env, [entry((1, 2, 3, 4))]) + sw.state, sw.reason = "paused", "3 jobs in a row failed; last: x" + for j in sw.jobs: + j["status"] = "error" + assert runner.tick(sw) is False + assert sw.state == "completed" and "4 job(s) failed" in sw.reason + # and it no longer blocks a new sweep + runner.prepare_sweep([entry((9,))], {"session_name": "wf_test"}, None, None, env.bridge, class_info=None) + + +def test_paused_sweep_with_pending_jobs_stays_paused(env): + sw = make(env, [entry((1, 2, 3, 4))]) + sw.state = "paused" + for j in sw.jobs[:3]: + j["status"] = "error" + assert runner.tick(sw) is True and sw.state == "paused" + + +def test_is_changed_nan_when_configs_json_is_linked(sampler_node): + cls = sampler_node.SamplerGridTester + assert cls.IS_CHANGED() != cls.IS_CHANGED() # linked input: absent from kwargs + assert cls.IS_CHANGED(steps=20) != cls.IS_CHANGED(steps=20) + + +def test_start_at_job_n_skips_n_minus_1(env): + sw = make(env, [entry((1, 2, 3, 4))], {"start_at_job": 3}) + assert [j["status"] for j in sw.jobs] == ["skipped", "skipped", "pending", "pending"] + runner.reset_registry() + sw = make(env, [entry((1, 2, 3, 4))], {"start_at_job": 1}) + assert [j["status"] for j in sw.jobs] == ["pending"] * 4 + + +def test_write_spec_retries_permission_error(env, monkeypatch): + sw = make(env) + real, calls = os.replace, {"n": 0} + + def flaky(a, b): + calls["n"] += 1 + if calls["n"] < 3: + raise PermissionError("open in dashboard") + return real(a, b) + monkeypatch.setattr(runner.os, "replace", flaky) + runner._write_spec(sw) + assert calls["n"] >= 3 + + +def test_dry_run_and_api_start_still_work_via_executor(env): + body = _cfg_body([entry()], "wf_test") + out = asyncio.run(runner.dry_run(body, bridge=env.bridge, class_info=None)) + assert out["total"] == 6 + status, res = asyncio.run(runner.api_start(body, bridge=env.bridge, class_info=None)) + assert status == 200 and res["total"] == 6 diff --git a/tests/test_workflow_sweep_collect.py b/tests/test_workflow_sweep_collect.py index 7cdbbed..1c5e358 100644 --- a/tests/test_workflow_sweep_collect.py +++ b/tests/test_workflow_sweep_collect.py @@ -122,12 +122,13 @@ def test_append_items_remove_ids_deletes_item_and_file(manifest_utils, tmp_path) assert outside.exists() # never deletes outside /images -def test_append_items_tolerates_corrupt_manifest(manifest_utils, tmp_path): +def test_append_items_refuses_corrupt_manifest(manifest_utils, tmp_path): mp = tmp_path / "s" / "manifest.json" mp.parent.mkdir() mp.write_text("{not json") - d = manifest_utils.append_items(str(mp), [{"id": 1}], session_name="s") - assert d["items"] == [{"id": 1}] + with pytest.raises(RuntimeError): # never replace an unreadable manifest with a fresh one (wipes the session) + manifest_utils.append_items(str(mp), [{"id": 1}], session_name="s") + assert mp.read_text() == "{not json" def test_session_paths_sanitises_name(monkeypatch, tmp_path): diff --git a/tests/test_workflow_sweep_runner.py b/tests/test_workflow_sweep_runner.py index 1bb51f8..6e37fcb 100644 --- a/tests/test_workflow_sweep_runner.py +++ b/tests/test_workflow_sweep_runner.py @@ -379,7 +379,7 @@ def test_overwrite_existing_reruns_and_carries_replace_ids(env): def test_start_at_job_skips_first_n(env): sw = make(env, [entry((1, 2, 3, 4))], {"start_at_job": 2}) - assert [j["status"] for j in sw.jobs] == ["skipped", "skipped", "pending", "pending"] + assert [j["status"] for j in sw.jobs] == ["skipped", "pending", "pending", "pending"] # job #2 is the first to run (N-1 skipped) assert sw.skipped_existing == 0 @@ -646,6 +646,10 @@ def _cfg_body(configs, session): def test_same_sweep_attaches_and_queues_nothing(env): sw = runner.start_sweep([entry()], {"session_name": "wf_test"}, "n1", None, env.bridge, class_info=None) + for _ in range(200): # the feeder thread enqueues job 1 asynchronously: wait for it (race-free count) + if env.bridge.enqueued: + break + time.sleep(0.01) n = len(env.bridge.enqueued) again = runner.start_sweep([entry()], {"session_name": "wf_test"}, "n2", None, env.bridge, class_info=None) assert again is sw diff --git a/web/conf_builder/conf-builder-workflow-sweep.js b/web/conf_builder/conf-builder-workflow-sweep.js index 4945946..76b617a 100644 --- a/web/conf_builder/conf-builder-workflow-sweep.js +++ b/web/conf_builder/conf-builder-workflow-sweep.js @@ -264,7 +264,8 @@ export function createWorkflowArrayElement(node, configArray, arrayIdx, accentCo // --------------------------------------------------------------------------- export function renderWorkflowSweepSection(node, arrayElement, configArray, arrayIdx) { - const wf = ensureWorkflowState(configArray); + // `let`: loading/removing a workflow REPLACES configArray.workflow, so render() re-reads it every time + let wf = ensureWorkflowState(configArray); const section = el("div", { className: "cb-section full-width" }, "margin-top: 8px;"); section.appendChild(createSectionHeader("๐Ÿงช", "Workflow Sweep", "configs")); const body = el("div", {}, "padding: 8px 4px; display: flex; flex-direction: column; gap: 10px;"); @@ -779,6 +780,7 @@ ${text}` : text; // ---- render ------------------------------------------------------- function render() { + wf = ensureWorkflowState(configArray); body.textContent = ""; footerCount = null; statusLine = null; diff --git a/workflow_sweep_runner.py b/workflow_sweep_runner.py index 97b321f..2430781 100644 --- a/workflow_sweep_runner.py +++ b/workflow_sweep_runner.py @@ -17,6 +17,7 @@ All loop-bound ComfyUI access goes through `ComfyBridge`, so the runner is testa """ import asyncio +import functools import inspect import json import os @@ -391,7 +392,7 @@ def prepare_sweep(configs, session_settings=None, origin_node=None, client_id=No "prompt_id": None, "error": None, "gen_index": gen_base + idx, "replace_ids": [], "enqueued_at": None, "started_at": None, "finished_at": None, "duration": None, "collected": False, "missing_since": None, "success_seen_at": None} - if idx < start_at: + if idx + 1 < start_at: job["status"], job["skip_reason"] = "skipped", "start_at_job" elif conf["wf_fingerprint"] in by_fp: if overwrite: @@ -546,7 +547,8 @@ def _fail_job(sw, job, status, error, params=None): sw.consecutive_errors += 1 sw.terminal_since_spec += 1 print(f"[GridTester] โŒ Sweep job {job['index'] + 1}/{len(sw.jobs)} {status}: {error}") - if sw.consecutive_errors >= BREAKER and sw.state == "running": + others_left = any(j is not job and j["status"] in ("pending", "enqueued", "running") for j in sw.jobs) + if sw.consecutive_errors >= BREAKER and sw.state == "running" and others_left: sw.state = "paused" sw.reason = f"{BREAKER} jobs in a row failed; last: {error}" @@ -735,6 +737,12 @@ def tick(sw): if pulled: with sw.lock: job.update(status="pending", prompt_id=None, enqueued_at=None, missing_since=None) + with sw.lock: + nothing_left = (sw.state == "paused" and not sw.in_flight() + and not any(j["status"] == "pending" for j in sw.jobs)) + if nothing_left: # the breaker (or a user pause) left nothing to resume: complete instead of blocking new sweeps + _finish(sw) # reason = "N job(s) failed" when there were failures + return False return True while True: @@ -957,7 +965,14 @@ def _write_spec(sw): path = os.path.join(paths["sweeps"], f"{sw.id}.json") with open(path + ".tmp", "w") as f: json.dump(spec, f, indent=1) - os.replace(path + ".tmp", path) + for attempt in range(5): + try: + os.replace(path + ".tmp", path) + break + except PermissionError: # Windows: a reader (the dashboard) has the file open + if attempt == 4: + raise + time.sleep(0.04) except Exception as e: print(f"[GridTester] โš ๏ธ Could not write the sweep spec file: {e}") @@ -992,7 +1007,8 @@ async def dry_run(body, bridge=None, class_info=wfs.comfy_class_info): limit = max(1, min(20, int(body.get("limit", 3)))) except (TypeError, ValueError): limit = 3 - sw = prepare_sweep(configs, settings, None, None, bridge, class_info, check_busy=False) + sw = await asyncio.get_running_loop().run_in_executor( + None, functools.partial(prepare_sweep, configs, settings, None, None, bridge, class_info, check_busy=False)) jobs_out = [] for job in sw.jobs[:limit]: prompt, partial = build_job_request(sw, job) @@ -1017,7 +1033,9 @@ async def api_start(body, bridge=None, class_info=wfs.comfy_class_info): """POST /config_tester/sweep/start. Returns (http_status, json_body). Runs on the event loop.""" try: configs, settings = configs_from_body(body) - sw = prepare_sweep(configs, settings, body.get("node_id"), body.get("client_id"), bridge, class_info) + sw = await asyncio.get_running_loop().run_in_executor( + None, functools.partial(prepare_sweep, configs, settings, body.get("node_id"), body.get("client_id"), + bridge, class_info)) await validate_first(sw) launch(sw) except SweepAlreadyRunning as e: