From 861def3a6254636fc5e40c25ce9344a712f7839a Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Thu, 23 Jul 2026 15:33:49 -0700 Subject: [PATCH 1/2] fix(jobs): emit prompt_not_found when local cancel targets an unknown id (BE-4218) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `comfy jobs cancel ` on local exited 0 with an ok envelope for an id that never existed: ComfyUI's `POST /queue {"delete": [id]}` returns 200 regardless, and the interrupt is skipped when the id isn't running, so both sub-results read as success. An agent could not tell "cancelled" from "already finished" from "typo'd the id". Probe existence before mutating anything: one `GET /queue` (shared with the interrupt gate) covers running + pending, `GET /history/` covers terminal prompts, and the local state store covers prompts we submitted that the server has since forgotten. Known ids in any state — including terminal — keep the documented idempotent ok and now carry a `found` field; an id that is nowhere gets a `prompt_not_found` error envelope + exit 1, matching what the cloud path already does with a 404. Probe failures never produce prompt_not_found: absence of evidence is not evidence of absence. --- comfy_cli/command/jobs.py | 74 ++++++++++++---- tests/comfy_cli/jobs/test_jobs.py | 137 +++++++++++++++++++++++++++++- 2 files changed, 193 insertions(+), 18 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 9d6db398..a79bf6e1 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -954,7 +954,7 @@ def fetch(pid: str) -> dict | None: @app.command( "cancel", - help="Cancel a job. Idempotent — calling on an already-terminal prompt returns ok.", + help="Cancel a job. Idempotent for known jobs; unknown ids error with prompt_not_found.", ) @tracking.track_command("jobs") def cancel_cmd( @@ -978,13 +978,61 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: interrupting any in-flight execution. ComfyUI splits these into two endpoints; we hit both so the call works regardless of phase. - Returns 200 (ok) regardless of whether the prompt was actually - queued/running — mirrors cloud's idempotent behavior. + Idempotent for prompts we can prove exist — running, pending, in the + server's history, or in the local state store — including already-terminal + ones, which return ok. An id that is nowhere is a `prompt_not_found` error + (exit 1), matching what the cloud path does with a 404: ``POST /queue + {"delete": [id]}`` 200s for unknown ids, so without the probe below a + typo'd id is indistinguishable from a real cancel. """ renderer = get_renderer() base = f"http://{host}:{port}" + from comfy_cli import jobs_state + + # 1. Existence probe, BEFORE mutating anything — the queue delete in step 2 + # would erase the only evidence that a pending prompt ever existed. This + # single /queue fetch also feeds the interrupt gate in step 3, so the + # check costs no extra round-trip. + try: + queue = _http_get_json(f"{base}/queue") + queue_reachable = True + except RuntimeError: + queue = {} + queue_reachable = False + if not isinstance(queue, dict): + queue = {} + running_ids = {str(_safe_queue_entry(entry)[0]) for entry in (queue.get("queue_running") or [])} + pending_ids = {str(_safe_queue_entry(entry)[0]) for entry in (queue.get("queue_pending") or [])} + + is_running = prompt_id in running_ids + existing = jobs_state.read(prompt_id) + # A state file means WE submitted it, so it existed even if the server has + # since forgotten it (restart, history trimmed). + found = is_running or prompt_id in pending_ids or existing is not None + probes_reachable = queue_reachable + if not found and queue_reachable: + # /history/ is `{}` for an unknown id, and a dict keyed by the id + # for a known one (running, completed, errored, or cancelled). Quote the + # id into the path so a hostile value can't escape the segment — same + # defense in depth as the cloud path. + try: + history = _http_get_json(f"{base}/history/{urllib.parse.quote(prompt_id, safe='')}") + found = isinstance(history, dict) and bool(history) + except RuntimeError: + # Unreachable is not "absent" — absence of evidence isn't evidence + # of absence, so fall through to the idempotent path instead. + probes_reachable = False + + if not found and probes_reachable: + renderer.error( + code="prompt_not_found", + message=f"no local job with id {prompt_id!r}", + hint="check `comfy jobs ls`", + details={"prompt_id": prompt_id, "host": host, "port": port}, + ) + raise typer.Exit(code=1) - # 1. Remove from the pending queue (no-op if not pending). + # 2. Remove from the pending queue (no-op if not pending). queue_body = json.dumps({"delete": [prompt_id]}).encode("utf-8") queue_req = urllib.request.Request( f"{base}/queue", data=queue_body, method="POST", headers={"Content-Type": "application/json"} @@ -998,21 +1046,14 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: # Don't fail the whole command — try the interrupt next. queue_ok = False - # 2. Interrupt only if THIS prompt is the one currently executing. + # 3. Interrupt only if THIS prompt is the one currently executing. # /interrupt takes NO prompt_id — it kills whatever is running — so # blindly posting it after a pending-job delete would also abort an # unrelated running job ("cancel B" silently cancelling A). Gate on - # /queue's queue_running list; the queue delete above already covers - # pending jobs. + # /queue's queue_running list (read in step 1); the queue delete above + # already covers pending jobs. interrupt_ok = True - try: - queue = _http_get_json(f"{base}/queue") - queue_reachable = True - except RuntimeError: - queue = {} - queue_reachable = False - running_ids = {str(_safe_queue_entry(entry)[0]) for entry in (queue.get("queue_running") or [])} - if prompt_id in running_ids: + if is_running: interrupt_req = urllib.request.Request(f"{base}/interrupt", method="POST") try: with urllib.request.urlopen(interrupt_req, timeout=10) as resp: @@ -1034,12 +1075,11 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: "where": "local", "host": host, "port": port, + "found": found, "queue_delete_ok": queue_ok, "interrupt_ok": interrupt_ok, } - from comfy_cli import jobs_state - existing = jobs_state.read(prompt_id) if existing is not None: existing.status = "cancelled" jobs_state.write(existing) diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index 5ed15898..f8977b80 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -433,7 +433,10 @@ def test_jobs_cancel_local_pending_job_does_not_interrupt(monkeypatch: pytest.Mo monkeypatch, { # prompt-pending is queued; a *different* prompt is running. - "GET /queue": {"queue_running": [[0, "prompt-running", {}, {}, {}]], "queue_pending": []}, + "GET /queue": { + "queue_running": [[0, "prompt-running", {}, {}, {}]], + "queue_pending": [[1, "prompt-pending", {}, {}, {}]], + }, "POST /queue": b"{}", "/interrupt": b"{}", }, @@ -466,6 +469,138 @@ def test_jobs_cancel_local_both_fail_returns_error(monkeypatch: pytest.MonkeyPat assert result.exit_code == 1, result.output +# --------------------------------------------------------------------------- +# `jobs cancel` local — existence probe (prompt_not_found for ids nowhere) +# --------------------------------------------------------------------------- + + +def _cancel_local_json(prompt_id: str): + """Run ``comfy --json jobs cancel --where local`` in-process and + return (result, envelope) — the envelope being the last stdout line.""" + from typer.testing import CliRunner + + from comfy_cli.cmdline import app + + result = CliRunner().invoke(app, ["--json", "jobs", "cancel", prompt_id, "--where", "local"]) + lines = [ln for ln in result.stdout.splitlines() if ln.strip()] + assert lines, f"no stdout envelope: {result.output!r}" + return result, json.loads(lines[-1]) + + +def test_jobs_cancel_local_unknown_id_emits_prompt_not_found(monkeypatch: pytest.MonkeyPatch): + """An id the server and the state store have never seen is an error, not a + silent idempotent ok — otherwise a typo'd id is indistinguishable from a + real cancel (`POST /queue {"delete": [id]}` 200s for unknown ids).""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen( + monkeypatch, + { + "GET /queue": {"queue_running": [], "queue_pending": []}, + # ComfyUI returns {} from /history/ for an unknown id. + "GET /history/": {}, + "POST /queue": b"{}", + "/interrupt": b"{}", + }, + ) + + result, env = _cancel_local_json("not-a-real-id") + assert result.exit_code == 1, result.output + assert env["ok"] is False + assert env["error"]["code"] == "prompt_not_found" + assert env["error"]["details"]["prompt_id"] == "not-a-real-id" + + # The probe must run BEFORE any mutation — nothing was deleted or + # interrupted on the way to deciding the id doesn't exist. + assert not any(c["method"] == "POST" for c in calls), calls + + +def test_jobs_cancel_local_known_only_in_history_is_idempotent_ok(monkeypatch: pytest.MonkeyPatch): + """A completed (terminal) prompt is still a KNOWN prompt — the documented + idempotent contract keeps returning ok for it.""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen( + monkeypatch, + { + "GET /queue": {"queue_running": [], "queue_pending": []}, + "GET /history/abc-1": {"abc-1": _HISTORY_FIXTURE["abc-1"]}, + "POST /queue": b"{}", + "/interrupt": b"{}", + }, + ) + + result, env = _cancel_local_json("abc-1") + assert result.exit_code == 0, result.output + assert env["ok"] is True + assert env["data"]["found"] is True + # Terminal, not running → no /interrupt (that would kill an unrelated job). + assert not any("/interrupt" in c["url"] for c in calls), calls + + +def test_jobs_cancel_local_known_only_in_state_file_is_ok(monkeypatch: pytest.MonkeyPatch): + """A state file means WE submitted the prompt, so it existed — even if the + server has since forgotten it (restart, trimmed history). No history probe + is needed; the routes below would AssertionError if one were made.""" + from comfy_cli import jobs_state + + jobs_state.write(jobs_state.new(prompt_id="pid-local", client_id="c", workflow="w", where="local")) + + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen( + monkeypatch, + { + "GET /queue": {"queue_running": [], "queue_pending": []}, + "POST /queue": b"{}", + }, + ) + + result, env = _cancel_local_json("pid-local") + assert result.exit_code == 0, result.output + assert env["ok"] is True and env["data"]["found"] is True + assert not any("/history" in c["url"] for c in calls), calls + + +def test_jobs_cancel_local_unreachable_queue_is_not_prompt_not_found(monkeypatch: pytest.MonkeyPatch): + """Absence of evidence isn't evidence of absence: when the existence probe + can't reach the server, keep the old idempotent behavior rather than + claiming the id doesn't exist.""" + import urllib.error + + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + _capture_urlopen( + monkeypatch, + { + "GET /queue": urllib.error.URLError("connection refused"), + "POST /queue": b"{}", + }, + ) + + result, env = _cancel_local_json("ghost-id") + assert result.exit_code == 0, result.output + assert env["ok"] is True + assert env["data"]["found"] is False + + +def test_jobs_cancel_local_unreachable_history_is_not_prompt_not_found(monkeypatch: pytest.MonkeyPatch): + """Same guard one probe further in: /queue answered but /history failed, so + existence is unproven — don't report prompt_not_found.""" + import urllib.error + + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + _capture_urlopen( + monkeypatch, + { + "GET /queue": {"queue_running": [], "queue_pending": []}, + "GET /history/": urllib.error.URLError("connection reset"), + "POST /queue": b"{}", + }, + ) + + result, env = _cancel_local_json("ghost-id") + assert result.exit_code == 0, result.output + assert env["ok"] is True + assert env["data"]["found"] is False + + def test_jobs_cancel_cloud_posts_to_jobs_cancel_endpoint(monkeypatch: pytest.MonkeyPatch): """Cloud cancel POSTs to /api/jobs//cancel with the auth header.""" from typer.testing import CliRunner From 3fa916641662ebcc27a290b2be07b65883dd256e Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Thu, 23 Jul 2026 15:55:12 -0700 Subject: [PATCH 2/2] fix(jobs): re-read the running set before /interrupt; keep terminal state (BE-4218) Follow-up to the prompt_not_found existence probe, addressing the cursor-review panel on #580: - The probe reused its pre-delete /queue snapshot to gate /interrupt, widening the window the gate exists to close: a prompt that goes pending->running during the delete round-trip was neither deleted nor interrupted, and a prompt that lost the running slot could get an unrelated job killed in its place. Re-read /queue immediately before interrupting; fall back to the snapshot only when that read fails (server down, where /interrupt would no-op anyway). - The final cancel_failed gate judged reachability from the same stale pre-delete probe, so a server dying mid-cancel reported a clean cancel. It now uses the post-delete read. - Cancelling an already-terminal job no longer rewrites its recorded status to 'cancelled' (jobs ls showed completed runs as cancelled), and the state file is re-read right before the write so a concurrent `jobs watch` update isn't clobbered. - An empty/whitespace id is rejected up front: quoted into the probe it became `GET /history/`, the list-all endpoint, whose non-empty body read as "found". - _http_get_json wraps JSONDecodeError (a ValueError) into RuntimeError, so a non-JSON 200 from a proxy is a probe failure, not a traceback. Six regression tests, each verified to fail against the previous commit. Co-Authored-By: Claude Opus 4.8 --- comfy_cli/command/jobs.py | 68 +++++++++++--- tests/comfy_cli/jobs/test_jobs.py | 146 ++++++++++++++++++++++++++++++ 2 files changed, 199 insertions(+), 15 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index a79bf6e1..8e196328 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -87,6 +87,11 @@ def _http_get_json(url: str, *, timeout: float = 10.0) -> Any: return json.loads(resp.read()) except urllib.error.URLError as e: raise RuntimeError(f"failed to GET {url}: {e}") from e + except ValueError as e: + # json.JSONDecodeError is a ValueError — a non-JSON 200 (captive + # portal, proxy error page) must look like any other GET failure to + # callers, not crash them with an uncaught decode error. + raise RuntimeError(f"failed to parse JSON from {url}: {e}") from e # --------------------------------------------------------------------------- @@ -989,28 +994,39 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: base = f"http://{host}:{port}" from comfy_cli import jobs_state + # 0. An empty/whitespace id can never name a real prompt, and it would turn + # the /history/ probe below into `GET /history/` — the list-ALL + # endpoint, whose non-empty body would read as "found". Reject it up + # front, before any probe or mutation. + if not prompt_id.strip(): + renderer.error( + code="prompt_not_found", + message="prompt id must be a non-empty string", + hint="check `comfy jobs ls`", + details={"prompt_id": prompt_id, "host": host, "port": port}, + ) + raise typer.Exit(code=1) + # 1. Existence probe, BEFORE mutating anything — the queue delete in step 2 - # would erase the only evidence that a pending prompt ever existed. This - # single /queue fetch also feeds the interrupt gate in step 3, so the - # check costs no extra round-trip. + # would erase the only evidence that a pending prompt ever existed. try: queue = _http_get_json(f"{base}/queue") - queue_reachable = True + queue_reachable_pre = True except RuntimeError: queue = {} - queue_reachable = False + queue_reachable_pre = False if not isinstance(queue, dict): queue = {} running_ids = {str(_safe_queue_entry(entry)[0]) for entry in (queue.get("queue_running") or [])} pending_ids = {str(_safe_queue_entry(entry)[0]) for entry in (queue.get("queue_pending") or [])} - is_running = prompt_id in running_ids - existing = jobs_state.read(prompt_id) # A state file means WE submitted it, so it existed even if the server has - # since forgotten it (restart, history trimmed). - found = is_running or prompt_id in pending_ids or existing is not None - probes_reachable = queue_reachable - if not found and queue_reachable: + # since forgotten it (restart, history trimmed). Read for existence only — + # the status write at the end re-reads, so a concurrent `jobs watch` update + # in the meantime isn't clobbered. + found = prompt_id in running_ids or prompt_id in pending_ids or jobs_state.read(prompt_id) is not None + probes_reachable = queue_reachable_pre + if not found and queue_reachable_pre: # /history/ is `{}` for an unknown id, and a dict keyed by the id # for a known one (running, completed, errored, or cancelled). Quote the # id into the path so a hostile value can't escape the segment — same @@ -1049,9 +1065,24 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: # 3. Interrupt only if THIS prompt is the one currently executing. # /interrupt takes NO prompt_id — it kills whatever is running — so # blindly posting it after a pending-job delete would also abort an - # unrelated running job ("cancel B" silently cancelling A). Gate on - # /queue's queue_running list (read in step 1); the queue delete above - # already covers pending jobs. + # unrelated running job ("cancel B" silently cancelling A). Gate on a + # FRESH read of /queue's queue_running: the step-1 snapshot predates the + # delete round-trip, and in that window our prompt can go pending→running + # (missing it = silent cancel failure) or a different prompt can take + # over the running slot (interrupting it = cancelling A instead of B). + try: + queue_now = _http_get_json(f"{base}/queue") + queue_reachable = True + if not isinstance(queue_now, dict): + queue_now = {} + is_running = prompt_id in {str(_safe_queue_entry(entry)[0]) for entry in (queue_now.get("queue_running") or [])} + except RuntimeError: + # Can't confirm; fall back to the step-1 snapshot. The server is very + # likely down, in which case /interrupt fails harmlessly too — better + # a best-effort interrupt than a silently skipped cancel. + queue_reachable = False + is_running = prompt_id in running_ids + interrupt_ok = True if is_running: interrupt_req = urllib.request.Request(f"{base}/interrupt", method="POST") @@ -1080,7 +1111,14 @@ def _local_cancel(prompt_id: str, host: str, port: int) -> None: "interrupt_ok": interrupt_ok, } - if existing is not None: + # Re-read right before the write: the step-1 read predates three network + # round-trips, and a concurrent `jobs watch` may have recorded newer + # status/outputs in the meantime (the per-file lock stops torn writes, not + # stale overwrites). Already-terminal jobs keep their recorded outcome — + # cancelling a finished job is an idempotent ok, not a re-labelling of a + # 'completed' run as 'cancelled'. + existing = jobs_state.read(prompt_id) + if existing is not None and not existing.is_terminal: existing.status = "cancelled" jobs_state.write(existing) diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index f8977b80..02a6f94c 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -329,6 +329,21 @@ def test_orphaned_flag_visible_in_help(): # --------------------------------------------------------------------------- +class _Seq: + """Route payload that changes per call: element i answers the i-th matching + request, and the last element repeats. Lets a test model a queue whose + contents change between two GET /queue reads.""" + + def __init__(self, *payloads): + self._payloads = list(payloads) + self._i = 0 + + def next(self): + payload = self._payloads[min(self._i, len(self._payloads) - 1)] + self._i += 1 + return payload + + def _capture_urlopen(monkeypatch: pytest.MonkeyPatch, routes: dict): """Capture calls to urlopen and return a list of (url, method, headers) per call.""" calls: list[dict] = [] @@ -358,6 +373,8 @@ def _fake(req, timeout=None): if " " in needle: want_method, sub = needle.split(" ", 1) if sub in url and (want_method is None or want_method == method): + if isinstance(payload, _Seq): + payload = payload.next() if isinstance(payload, Exception): raise payload return _Resp(payload if isinstance(payload, bytes) else json.dumps(payload).encode()) @@ -601,6 +618,135 @@ def test_jobs_cancel_local_unreachable_history_is_not_prompt_not_found(monkeypat assert env["data"]["found"] is False +def test_jobs_cancel_local_empty_id_is_prompt_not_found(monkeypatch: pytest.MonkeyPatch): + """An empty/whitespace id must be rejected before any probe: quoting it into + the history probe would produce `GET /history/` — the list-ALL endpoint — + whose non-empty body would read as 'found' and wave a garbage id through.""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen(monkeypatch, {}) # any HTTP call at all is a failure + + result, env = _cancel_local_json(" ") + assert result.exit_code == 1, result.output + assert env["error"]["code"] == "prompt_not_found" + assert calls == [], f"empty id must not touch the server: {calls}" + + +def test_jobs_cancel_local_non_json_body_is_not_a_traceback(monkeypatch: pytest.MonkeyPatch): + """A 200 with a non-JSON body (proxy error page, captive portal) must be + treated like any other probe failure, not crash with a JSONDecodeError.""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + _capture_urlopen( + monkeypatch, + { + "GET /queue": b"gateway timeout", + "POST /queue": b"{}", + }, + ) + + result, env = _cancel_local_json("ghost-id") + assert result.exit_code == 0, result.output + assert env["ok"] is True and env["data"]["found"] is False + + +def test_jobs_cancel_local_rereads_running_set_before_interrupt(monkeypatch: pytest.MonkeyPatch): + """A job that goes pending→running while the queue delete is in flight must + still be interrupted. Gating on the pre-delete snapshot would skip the + interrupt and report a successful cancel of a job that keeps running.""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen( + monkeypatch, + { + "GET /queue": _Seq( + # Before the delete: ours is pending, another job is running. + {"queue_running": [[0, "other", {}, {}, {}]], "queue_pending": [[1, "mine", {}, {}, {}]]}, + # After the delete: 'other' finished and ours started. + {"queue_running": [[0, "mine", {}, {}, {}]], "queue_pending": []}, + ), + "POST /queue": b"{}", + "/interrupt": b"{}", + }, + ) + + result, env = _cancel_local_json("mine") + assert result.exit_code == 0, result.output + assert env["ok"] is True + assert any("/interrupt" in c["url"] for c in calls), f"must interrupt the now-running job: {calls}" + + +def test_jobs_cancel_local_does_not_interrupt_a_job_that_took_over(monkeypatch: pytest.MonkeyPatch): + """The mirror case: our job finished during the delete round-trip and a + DIFFERENT job now holds the running slot. /interrupt takes no prompt_id, so + firing it off the stale snapshot would cancel that unrelated job.""" + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + calls = _capture_urlopen( + monkeypatch, + { + "GET /queue": _Seq( + {"queue_running": [[0, "mine", {}, {}, {}]], "queue_pending": []}, + {"queue_running": [[0, "other", {}, {}, {}]], "queue_pending": []}, + ), + "POST /queue": b"{}", + "/interrupt": b"{}", + }, + ) + + result, _env = _cancel_local_json("mine") + assert result.exit_code == 0, result.output + assert not any("/interrupt" in c["url"] for c in calls), f"must not kill an unrelated job: {calls}" + + +def test_jobs_cancel_local_server_dying_mid_cancel_is_not_a_success(monkeypatch: pytest.MonkeyPatch): + """Reachability for the final gate must be judged AFTER the delete. If the + server dies between the existence probe and the delete, the delete fails and + the command must surface cancel_failed rather than report a clean cancel.""" + import urllib.error + + from comfy_cli import jobs_state + + jobs_state.write(jobs_state.new(prompt_id="pid-dying", client_id="c", workflow="w", where="local")) + + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + _capture_urlopen( + monkeypatch, + { + "GET /queue": _Seq( + {"queue_running": [], "queue_pending": []}, + urllib.error.URLError("connection refused"), + ), + "POST /queue": urllib.error.URLError("connection refused"), + }, + ) + + result, env = _cancel_local_json("pid-dying") + assert result.exit_code == 1, result.output + assert env["error"]["code"] == "cancel_failed" + + +def test_jobs_cancel_local_keeps_terminal_status(monkeypatch: pytest.MonkeyPatch): + """Cancelling an already-completed job is an idempotent ok — it must NOT + rewrite the recorded outcome, or `jobs ls` reports a completed run as + cancelled.""" + from comfy_cli import jobs_state + + st = jobs_state.new(prompt_id="pid-done", client_id="c", workflow="w", where="local") + st.status = "completed" + jobs_state.write(st) + + monkeypatch.setattr(jobs_mod, "_server_or_error", lambda h, p, **kw: True) + _capture_urlopen( + monkeypatch, + { + "GET /queue": {"queue_running": [], "queue_pending": []}, + "POST /queue": b"{}", + }, + ) + + result, env = _cancel_local_json("pid-done") + assert result.exit_code == 0, result.output + assert env["ok"] is True and env["data"]["found"] is True + assert jobs_state.read("pid-done").status == "completed" + + def test_jobs_cancel_cloud_posts_to_jobs_cancel_endpoint(monkeypatch: pytest.MonkeyPatch): """Cloud cancel POSTs to /api/jobs//cancel with the auth header.""" from typer.testing import CliRunner