From 968b0dfdfceea71e05844e7224c865c130733528 Mon Sep 17 00:00:00 2001 From: Jason Hoku Date: Wed, 4 Nov 2026 00:52:56 -0800 Subject: [PATCH] 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 --- __init__.py | 12 ++ manifest_utils.py | 19 +++ tests/test_workflow_sweep_generator.py | 14 +- tests/test_workflow_sweep_runner.py | 145 ++++++++++++++++++ .../conf-builder-workflow-sweep.js | 8 +- workflow_sweep_runner.py | 136 ++++++++++++++-- 6 files changed, 317 insertions(+), 17 deletions(-) diff --git a/__init__.py b/__init__.py index faa3bfd..8d6aded 100644 --- a/__init__.py +++ b/__init__.py @@ -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() diff --git a/manifest_utils.py b/manifest_utils.py index 9c91efb..7d7aab1 100644 --- a/manifest_utils.py +++ b/manifest_utils.py @@ -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. diff --git a/tests/test_workflow_sweep_generator.py b/tests/test_workflow_sweep_generator.py index c13f45b..834aa8b 100644 --- a/tests/test_workflow_sweep_generator.py +++ b/tests/test_workflow_sweep_generator.py @@ -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 == ("",) 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: diff --git a/tests/test_workflow_sweep_runner.py b/tests/test_workflow_sweep_runner.py index 2fecf68..75d5fba 100644 --- a/tests/test_workflow_sweep_runner.py +++ b/tests/test_workflow_sweep_runner.py @@ -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"} diff --git a/web/conf_builder/conf-builder-workflow-sweep.js b/web/conf_builder/conf-builder-workflow-sweep.js index c74fb0c..8527a92 100644 --- a/web/conf_builder/conf-builder-workflow-sweep.js +++ b/web/conf_builder/conf-builder-workflow-sweep.js @@ -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(); diff --git a/workflow_sweep_runner.py b/workflow_sweep_runner.py index 0c60f63..b57f097 100644 --- a/workflow_sweep_runner.py +++ b/workflow_sweep_runner.py @@ -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,