feat(sweep): attach to a running identical sweep, job_prompt route, configs in sweep meta

Identity = session + ordered job fingerprints + start_at_job + overwrite_existing; anything
else gets a clear 409. GET /config_tester/sweep/job_prompt returns one job's API prompt.
Sweep manifests now save their workflow configs in meta.configs_json.
pause/resume emit a progress event.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Jason Hoku
2026-11-04 00:52:56 -08:00
co-authored by Claude Fable 5.1
parent 525733d70b
commit 968b0dfdfc
6 changed files with 317 additions and 17 deletions
+12
View File
@@ -1508,6 +1508,18 @@ async def sweep_status(request):
return web.json_response({"error": "No such sweep"}, status=404)
return web.json_response(status)
@server.PromptServer.instance.routes.get("/config_tester/sweep/job_prompt")
async def sweep_job_prompt(request):
"""The API-format prompt of one sweep job (derived JSON, no file is served; ids only)."""
try:
from .workflow_sweep_runner import job_prompt
status, body = job_prompt(request.query.get("session_name") or "", request.query.get("sweep_id") or "",
request.query.get("job_index", ""))
return web.json_response(body, status=status)
except Exception as e:
print(f"[ConfigTester] Error building the sweep job prompt: {e}")
return web.json_response({"error": str(e)}, status=500)
async def _sweep_control(request, action):
try:
data = await request.json()
+19
View File
@@ -112,6 +112,25 @@ def _atomic_write_json(path, data):
pass
def update_meta(manifest_path, updates):
"""
Locked read-modify-write of ONLY keys of the manifest's "meta" dict (items written since the
caller last looked are preserved). Raises on a missing/unreadable manifest; nothing is written then.
"""
with _manifest_lock:
with open(manifest_path, "r") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("manifest is not a JSON object")
meta = data.get("meta")
if not isinstance(meta, dict):
meta = {}
meta.update(updates)
data["meta"] = meta
_atomic_write_json(manifest_path, data)
return meta
def update_manifest_comparisons(manifest_path, append=(), remove_keys=(), clear=False):
"""
Locked read-modify-write of ONLY the "comparisons" list.
+11 -3
View File
@@ -104,9 +104,17 @@ def test_run_tests_starts_sweep_and_returns_dashboard_without_waiting(sampler_no
assert captured["manifest"]["session_name"] == "gen_wf" and captured["manifest"]["items"] == []
sw = runner._active_sweep()
assert sw is not None and sw.origin_node == "77" and len(sw.jobs) == 2 # returned while jobs are pending
# a second Generator run while the sweep is active is refused with a readable error
with pytest.raises(RuntimeError, match="already active"):
sampler_node.SamplerGridTester().run_tests(cfg, unique_id="77")
# an IDENTICAL second Generator run attaches: same dashboard, one sweep, nothing re-queued
n_enq = len(bridge.enqueued)
out2 = sampler_node.SamplerGridTester().run_tests(cfg, unique_id="77")
assert out2 == ("<html>",) and captured["title"] == "gen_wf"
assert len(runner._sweeps) == 1 and runner._active_sweep() is sw and len(bridge.enqueued) >= n_enq
# a DIFFERENT sweep while one is active is refused with a readable error
other = wf_config()
other["workflow_bindings"][0]["values"] = [5, 6]
cfg2 = json.dumps({"configs": [other], "_session_settings": {"session_name": "gen_wf", "start_at_job": 0}})
with pytest.raises(RuntimeError, match="NOT started"):
sampler_node.SamplerGridTester().run_tests(cfg2, unique_id="77")
finally:
sw = runner._active_sweep()
if sw is not None:
+145
View File
@@ -16,6 +16,15 @@ import workflow_sweep as wfs # noqa: E402
import workflow_sweep_runner as runner # noqa: E402
def _real_manifest_utils():
# conftest registers a stub manifest_utils; load the real file for update_meta
import importlib.util
spec = importlib.util.spec_from_file_location("manifest_utils_real_sweep", os.path.join(ROOT, "manifest_utils.py"))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def load(name):
with open(os.path.join(ROOT, "tests", "fixtures", name), "r", encoding="utf-8") as f:
return json.load(f)
@@ -116,6 +125,7 @@ class FakeBridge:
def env(tmp_path, monkeypatch):
import folder_paths
monkeypatch.setattr(folder_paths, "get_output_directory", lambda: str(tmp_path), raising=False)
monkeypatch.setattr(runner, "manifest_utils", _real_manifest_utils(), raising=False)
runner.reset_registry()
bridge = FakeBridge()
yield types.SimpleNamespace(bridge=bridge, tmp=tmp_path)
@@ -626,3 +636,138 @@ def test_bridge_queue_helpers(monkeypatch):
assert b.history_entry("nope") is None
assert b.delete_queued("q1") is True and removed == [True]
assert b.interrupt("r1") is True and b.interrupt("q1") is False
# --------------------------------------------------------------------------- attach to the identical sweep
def _cfg_body(configs, session):
return {"configs_json": json.dumps({"configs": configs, "_session_settings": {"session_name": 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)
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
assert len(runner._sweeps) == 1 and len(env.bridge.enqueued) == n
runner.cancel(sw.id)
sw.thread.join(5)
def test_api_start_same_sweep_returns_attached_200(env):
sw = make(env)
assert runner.pause(sw.id)[0]
status, body = asyncio.run(runner.api_start(_cfg_body([entry()], "wf_test"), bridge=env.bridge, class_info=None))
assert status == 200 and body["attached"] is True and body["sweep_id"] == sw.id
assert body["state"] == "paused" and sw.state == "paused" # attach never changes the sweep state
assert len(runner._sweeps) == 1 and env.bridge.enqueued == []
def test_different_sweep_409_with_session_and_message(env):
sw = make(env)
# different values
status, body = asyncio.run(runner.api_start(_cfg_body([entry((7, 8))], "wf_test"), bridge=env.bridge, class_info=None))
assert status == 409 and body["sweep_id"] == sw.id and body["session_name"] == "wf_test"
assert "NOT started" in body["error"] and "already active" in body["error"]
# same workflow, different session
status, body = asyncio.run(runner.api_start(_cfg_body([entry()], "other_session"), bridge=env.bridge, class_info=None))
assert status == 409
# same workflow and session but a different start_at_job / overwrite_existing is NOT the same sweep
for extra in ({"start_at_job": 2}, {"overwrite_existing": True}):
with pytest.raises(runner.SweepBusy) as ei:
runner.prepare_sweep([entry()], {"session_name": "wf_test", **extra}, None, None, env.bridge, class_info=None)
assert not isinstance(ei.value, runner.SweepAlreadyRunning)
with pytest.raises(runner.SweepBusy) as ei:
runner.prepare_sweep([entry((7, 8))], {"session_name": "wf_test"}, None, None, env.bridge, class_info=None)
assert not isinstance(ei.value, runner.SweepAlreadyRunning) and "NOT started" in str(ei.value)
assert len(runner._sweeps) == 1
def test_random_extra_seeds_never_attach(env):
make(env)
raw = {"configs": [entry()], "_session_settings": {"session_name": "wf_test", "add_random_seeds_to_gens": 1}}
status, body = asyncio.run(runner.api_start({"configs_json": json.dumps(raw)}, bridge=env.bridge, class_info=None))
assert status == 409
def test_launch_race_same_sweep(env):
a = runner.prepare_sweep([entry()], {"session_name": "race"}, None, None, env.bridge, class_info=None)
b = runner.prepare_sweep([entry()], {"session_name": "race"}, None, None, env.bridge, class_info=None)
runner.launch(a, start_thread=False)
with pytest.raises(runner.SweepAlreadyRunning) as ei:
runner.launch(b, start_thread=False)
assert ei.value.sweep is a and b.id not in runner._sweeps
def test_pause_and_resume_emit_progress(env):
sw = make(env)
env.bridge.events.clear()
runner.pause(sw.id)
runner.resume(sw.id)
states = [d["state"] for d in env.bridge.progress_events()]
assert states == ["paused", "running"]
# --------------------------------------------------------------------------- job_prompt + sweep meta
def test_job_prompt_matches_apply_subs_and_has_no_collector(env):
sw = make(env, [entry((1, 2, 3))])
status, body = runner.job_prompt("wf_test", sw.id, 1)
job = sw.jobs[1]
tpl = sw.templates[job["conf"]["wf_template_hash"]]
assert status == 200 and body["prompt"] == wfs.apply_subs(tpl["prompt"], job["conf"]["wf_subs"])
assert "USCGSweepCollect" not in json.dumps(body["prompt"])
assert body["fingerprint"] == job["fingerprint"] and body["job_index"] == 1 and body["params"] is not None
# from the spec file, after the in-memory registry is gone
runner.reset_registry()
status2, body2 = runner.job_prompt("wf_test", sw.id, "1")
assert status2 == 200 and body2["prompt"] == body["prompt"] and body2["fingerprint"] == body["fingerprint"]
def test_job_prompt_validation_and_404(env):
sw = make(env)
for sid, idx in (("../x", 0), ("abc", 0), (sw.id, -1), (sw.id, "x"), (sw.id, "1.5"), ("", 0)):
assert runner.job_prompt("wf_test", sid, idx)[0] == 400, (sid, idx)
assert runner.job_prompt("wf_test", "deadbeef0000", 0)[0] == 404
assert runner.job_prompt("wf_test", sw.id, 99)[0] == 404
assert runner.job_prompt("some_other_session", sw.id, 0)[0] == 404 # wrong session never leaks another's job
def test_new_manifest_meta_carries_configs_json(env):
sw = make(env)
meta = wfs.read_manifest_items(wfs.session_paths("wf_test")["manifest"])["meta"]
parsed = json.loads(meta["configs_json"])
assert parsed["configs"][0]["model_type"] == "workflow" and parsed["_session_settings"]["session_name"] == "wf_test"
assert meta["workflow_sweep"] is True
def test_existing_sweep_manifest_meta_refreshed_but_standard_untouched(env):
paths = wfs.session_paths("wf_test")
os.makedirs(paths["images"], exist_ok=True)
old = {"session_name": "wf_test", "meta": {"workflow_sweep": True, "configs_json": ""}, "items": [{"id": 5, "file": "x"}]}
with open(paths["manifest"], "w") as f:
json.dump(old, f)
make(env)
data = wfs.read_manifest_items(paths["manifest"])
assert json.loads(data["meta"]["configs_json"])["configs"] and data["items"] == [{"id": 5, "file": "x"}]
runner.reset_registry()
std = {"session_name": "std_sess", "meta": {"configs_json": "{}"}, "items": []}
p2 = wfs.session_paths("std_sess")
os.makedirs(p2["images"], exist_ok=True)
with open(p2["manifest"], "w") as f:
json.dump(std, f)
before = open(p2["manifest"], "rb").read()
make(env, settings={"session_name": "std_sess"})
assert open(p2["manifest"], "rb").read() == before
def test_update_meta_preserves_items_written_since(tmp_path):
mu = _real_manifest_utils()
path = str(tmp_path / "manifest.json")
with open(path, "w") as f:
json.dump({"meta": {"a": 1}, "items": []}, f)
with open(path, "w") as f: # another writer adds an item after the caller last looked
json.dump({"meta": {"a": 1}, "items": [{"id": 9}]}, f)
mu.update_meta(path, {"configs_json": "z"})
data = json.load(open(path))
assert data["items"] == [{"id": 9}] and data["meta"] == {"a": 1, "configs_json": "z"}
@@ -609,6 +609,7 @@ export function renderWorkflowSweepSection(node, arrayElement, configArray, arra
}
let statusLine = null;
let attachedNote = "";
function paintStatus(extra) {
if (!statusLine) return;
const s = lastStatus;
@@ -621,7 +622,8 @@ export function renderWorkflowSweepSection(node, arrayElement, configArray, arra
+ (s.last_errors && s.last_errors.length ? `\nLast error (job ${s.last_errors[s.last_errors.length - 1].index + 1}): ${s.last_errors[s.last_errors.length - 1].error}` : "");
(s.warnings || []).forEach(w => { text += `\n⚠️ ${w}`; });
}
statusLine.textContent = text;
statusLine.textContent = attachedNote && s ? `${attachedNote}
${text}` : text;
}
async function fetchStatus() {
@@ -648,14 +650,16 @@ export function renderWorkflowSweepSection(node, arrayElement, configArray, arra
}
async function startSweep() {
attachedNote = "";
paintStatus("Starting...");
const { ok, status, data } = await post("/config_tester/sweep/start", { state: workflowOnlyState(node) });
if (!ok) {
paintStatus(status === 409 ? `Already running: ${data.error}` : `Could not start: ${data.error || status}`);
paintStatus(status === 409 ? `${data.error}` : `Could not start: ${data.error || status}`);
if (status === 409 && data.sweep_id) { sweepId = data.sweep_id; lastStatus = await fetchStatus(); rerender(); startPolling(); }
return;
}
sweepId = data.sweep_id;
attachedNote = data.attached ? `Already running: attached to sweep ${data.sweep_id}` : "";
lastStatus = await fetchStatus();
rerender();
startPolling();
+124 -12
View File
@@ -21,6 +21,7 @@ import inspect
import json
import os
import random
import re
import threading
import time
import traceback
@@ -28,8 +29,10 @@ import uuid
try:
from . import workflow_sweep as wfs
from . import manifest_utils
except ImportError: # imported as a top-level module (tests)
import workflow_sweep as wfs
import manifest_utils
WINDOW = 2 # max sweep prompts queued/running at once
@@ -44,10 +47,26 @@ TERMINAL_JOB = ("done", "error", "invalid", "skipped", "cancelled")
class SweepBusy(Exception):
"""Another workflow sweep is still active (route -> 409)."""
def __init__(self, sweep_id):
super().__init__(f"A workflow sweep is already active ({sweep_id}). Cancel it or wait for it to finish.")
"""Another (different) workflow sweep is still active (route -> 409)."""
def __init__(self, sweep_id, session=None, sweep=None):
if sweep is not None:
with sweep.lock:
detail = f"session '{sweep.session}', sweep {sweep_id}, {sweep.count('done')}/{sweep.to_run} done, {sweep.state}"
session = sweep.session
else:
detail = f"session '{session}', sweep {sweep_id}" if session else f"sweep {sweep_id}"
super().__init__(f"A workflow sweep is already active ({detail}). This run was NOT started. "
f"Pause or cancel it in the Builder's Workflow Sweep panel (or POST "
f"/config_tester/sweep/cancel {{sweep_id}}), or wait for it to finish.")
self.sweep_id = sweep_id
self.session = session
class SweepAlreadyRunning(SweepBusy):
"""The identical sweep is already active: callers attach to it instead of failing."""
def __init__(self, sweep):
super().__init__(sweep.id, sweep=sweep)
self.sweep = sweep
def summarize_errors(error, node_errors, prompt=None):
@@ -297,6 +316,15 @@ def _active_sweep():
return None
def _same_sweep(busy, session, fingerprints, start_at=0, overwrite=False):
"""True when `busy` is the sweep the caller is trying to start: same session, same ordered job
fingerprints, same start_at_job and overwrite_existing."""
return (busy.session == session
and [j["fingerprint"] for j in busy.jobs] == list(fingerprints)
and int(busy.settings.get("start_at_job", 0) or 0) == int(start_at or 0)
and bool(busy.settings.get("overwrite_existing", False)) == bool(overwrite))
def _find(sweep_id=None, session=None):
with _registry_lock:
if sweep_id:
@@ -331,9 +359,6 @@ def prepare_sweep(configs, session_settings=None, origin_node=None, client_id=No
Expand + fingerprint + skip finished jobs. Raises ValueError (readable) or SweepBusy.
Pure apart from reading the session manifest.
"""
busy = _active_sweep() if check_busy else None
if busy is not None:
raise SweepBusy(busy.id)
settings = dict(session_settings or {})
if not isinstance(configs, list) or not configs:
raise ValueError("There are no configs to run.")
@@ -352,6 +377,12 @@ def prepare_sweep(configs, session_settings=None, origin_node=None, client_id=No
start_at = max(0, int(settings.get("start_at_job", 0) or 0))
except (TypeError, ValueError):
start_at = 0
busy = _active_sweep() if check_busy else None
if busy is not None:
fps = [c["wf_fingerprint"] for c in plan["jobs"]]
if _same_sweep(busy, paths["session"], fps, start_at, overwrite):
raise SweepAlreadyRunning(busy)
raise SweepBusy(busy.id, sweep=busy)
by_fp, gen_base = _manifest_state(paths["manifest"])
jobs = []
@@ -376,6 +407,7 @@ def prepare_sweep(configs, session_settings=None, origin_node=None, client_id=No
list(plan["warnings"]), plan["axes"], bridge or get_bridge())
if sw.client_id is None:
sw.client_id = f"uscg-sweep-{sw.id}"
sw.source_configs = configs
return sw
@@ -404,11 +436,19 @@ async def validate_first(sw):
raise ValueError("The first job is not valid: " + summarize_errors(error, node_errors, prompt))
def _configs_json(sw):
"""Builder-loadable configs for the manifest meta (the workflow entries, so Load Session can restore them)."""
cfgs = getattr(sw, "source_configs", None)
if not cfgs:
return ""
return json.dumps({"configs": cfgs, "_session_settings": {"session_name": sw.session}})
def _new_meta(sw):
first = sw.jobs[0]["conf"] if sw.jobs else {}
return {"positive": first.get("positive", ""), "negative": first.get("negative", ""),
"model": first.get("model", ""), "seed": first.get("seed", 0) if isinstance(first.get("seed", 0), int) else 0,
"vae_batch_size": 1, "configs_json": "", "resolutions_json": "[]",
"vae_batch_size": 1, "configs_json": _configs_json(sw), "resolutions_json": "[]",
"image_format": sw.settings.get("image_format", "webp"), "workflow_sweep": True}
@@ -422,6 +462,12 @@ def ensure_manifest(sw):
json.dump(data, f, indent=4)
return data
data = wfs.read_manifest_items(paths["manifest"])
cj = _configs_json(sw)
if cj and (data.get("meta") or {}).get("workflow_sweep"):
try:
manifest_utils.update_meta(paths["manifest"], {"configs_json": cj})
except Exception as e:
print(f"[GridTester] ⚠️ Could not refresh the sweep session's saved configs: {e}")
return data
@@ -430,7 +476,10 @@ def launch(sw, start_thread=True):
with _registry_lock:
busy = _active_sweep()
if busy is not None and busy is not sw:
raise SweepBusy(busy.id)
if _same_sweep(busy, sw.session, [j["fingerprint"] for j in sw.jobs],
sw.settings.get("start_at_job", 0), sw.settings.get("overwrite_existing", False)):
raise SweepAlreadyRunning(busy)
raise SweepBusy(busy.id, sweep=busy)
_sweeps[sw.id] = sw
ensure_manifest(sw)
with sw.lock:
@@ -452,9 +501,14 @@ def start_sweep(configs, session_settings=None, origin_node=None, client_id=None
class_info=wfs.comfy_class_info):
"""Prepare + validate job 1 on the loop + launch. For a NON-loop thread (the Generator node).
Returns immediately after launch; never waits for any sweep prompt."""
sw = prepare_sweep(configs, session_settings, origin_node, client_id, bridge, class_info)
sw.bridge.run_on_loop(validate_first(sw))
return launch(sw)
try:
sw = prepare_sweep(configs, session_settings, origin_node, client_id, bridge, class_info)
sw.bridge.run_on_loop(validate_first(sw))
return launch(sw)
except SweepAlreadyRunning as e:
print(f"[GridTester] 🔁 Same workflow sweep already running ({e.sweep.id}, session "
f"'{e.sweep.session}'): attaching, nothing re-queued")
return e.sweep
# --------------------------------------------------------------------------- feeder thread
@@ -710,6 +764,7 @@ def pause(sweep_id):
return False, f"sweep is {sw.state}"
sw.state, sw.reason = "paused", "paused by user"
sw.wake.set()
_progress(sw)
return True, "paused"
@@ -726,6 +781,7 @@ def resume(sweep_id):
sw.consecutive_errors = 0
sw.state, sw.reason = "running", ""
sw.wake.set()
_progress(sw)
return True, "running"
@@ -784,6 +840,58 @@ def get_status(sweep_id=None, session=None):
"last_errors": list(sw.last_errors)}
_SWEEP_ID_RE = re.compile(r"^[0-9a-f]{6,32}$")
def job_prompt(session_name, sweep_id, job_index):
"""(http_status, body) for GET /config_tester/sweep/job_prompt: the API prompt of one sweep job
exactly as it ran (templates + subs), minus the USCG collector node. Takes ids, never paths: the
session name is sanitised by session_paths and the sweep id is regex-checked before it touches a path."""
sweep_id = str(sweep_id or "")
if not _SWEEP_ID_RE.match(sweep_id):
return 400, {"error": "sweep_id must be 6-32 lowercase hex characters."}
try:
idx = int(str(job_index).strip())
if idx < 0 or not re.match(r"^\d+$", str(job_index).strip()):
raise ValueError
except (TypeError, ValueError):
return 400, {"error": "job_index must be an integer >= 0."}
spec = None
sw = _find(sweep_id)
if sw is not None and sw.session == wfs.session_paths(session_name)["session"]:
with sw.lock:
if idx >= len(sw.jobs):
return 404, {"error": f"Sweep {sweep_id} has no job {idx}."}
job = sw.jobs[idx]
tpl = sw.templates.get(job["conf"].get("wf_template_hash"))
if tpl is None:
return 404, {"error": "The job's workflow template is not available."}
subs, params, fp = job["conf"].get("wf_subs") or [], job["conf"].get("wf_params"), job["fingerprint"]
template, output = tpl["prompt"], tpl["output"]
else:
path = os.path.join(wfs.session_paths(session_name)["sweeps"], sweep_id + ".json")
try:
with open(path, "r", encoding="utf-8") as f:
spec = json.load(f)
except (OSError, ValueError):
return 404, {"error": f"No sweep {sweep_id} was recorded for session '{session_name}'."}
jobs = spec.get("jobs") or []
if idx >= len(jobs):
return 404, {"error": f"Sweep {sweep_id} has no job {idx}."}
job = jobs[idx]
tpl = (spec.get("templates") or {}).get(job.get("template"))
if tpl is None:
return 404, {"error": "The job's workflow template is not in the sweep spec."}
subs, params, fp = job.get("subs") or [], job.get("params"), job.get("fingerprint")
template, output = tpl["prompt"], tpl.get("output")
try:
prompt = wfs.apply_subs(template, subs)
except ValueError as e:
return 404, {"error": str(e)}
return 200, {"sweep_id": sweep_id, "job_index": idx, "fingerprint": fp, "params": params,
"output": output, "prompt": prompt}
# --------------------------------------------------------------------------- progress + spec file
def _eta_str(seconds):
@@ -904,8 +1012,12 @@ async def api_start(body, bridge=None, class_info=wfs.comfy_class_info):
sw = 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:
sw = e.sweep
return 200, {**(get_status(sw.id) or {}), "attached": True, "sweep_id": sw.id, "total": len(sw.jobs),
"to_run": sw.to_run, "skipped_existing": sw.skipped_existing, "warnings": sw.warnings}
except SweepBusy as e:
return 409, {"error": str(e), "sweep_id": e.sweep_id}
return 409, {"error": str(e), "sweep_id": e.sweep_id, "session_name": e.session}
except ValueError as e:
return 400, {"error": str(e)}
return 200, {"sweep_id": sw.id, "total": len(sw.jobs), "to_run": sw.to_run,