From f981c3509e84ba73b6df2df76a8cce34439a4080 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Tue, 25 Aug 2026 02:50:18 -0700 Subject: [PATCH 1/2] fix(jobs): emit one execution_cached event per node in `jobs watch` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `docs/json-output.md` promises the `comfy run` stream and the `comfy jobs watch` stream speak one dialect, but the two disagreed on `execution_cached`. `comfy run` fans ComfyUI's single cached-node message out into one event per node (`{"node": "", ...}`); `jobs watch` relayed the websocket frame's shape unchanged as one event carrying a `nodes` array. A consumer written against the documented run dialect — the only documented shape — counted one cached node per event and so undercounted every cached node beyond the first on a watch stream. `_watch_execution_cached` now emits one event per node, matching `on_cached` in `run/execution.py`. `title` / `class_type` are omitted: `jobs watch` has no workflow map to resolve them from, and both are optional in the run dialect's own emission. The `completed_nodes` accumulation and the pretty-mode line are unchanged. The `nodes` array stays legal on `execution_cached` in `run_event.json` so a stream captured from 1.16.0 still validates, but nothing emits it; the schema description now says so. `docs/json-output.md` documents the per-node shape for both streams and the fields `jobs watch` omits. The list shape was a passthrough of the websocket frame from the original `jobs watch` implementation, written before the one-dialect contract was documented; no history ties it to a deliberate design decision. --- CHANGELOG.md | 12 ++++ comfy_cli/command/jobs.py | 17 ++++-- comfy_cli/schemas/run_event.json | 4 +- docs/json-output.md | 27 ++++++++- tests/comfy_cli/command/test_run_json.py | 20 +++++- tests/comfy_cli/jobs/test_jobs.py | 77 +++++++++++++++++++++++- 6 files changed, 147 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fa5eda40b..7efecb337 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,18 @@ history. - `CONTRIBUTING.md` (renamed from `DEV_README.md`) and this changelog. +### Fixed + +- Stream-dialect conformance: `comfy jobs watch` now emits one + `execution_cached` event **per cached node** (`{"node": "", …}`), the same + shape `comfy run` has always emitted, instead of the single list-shaped + `{"nodes": [...]}` event that shipped in 1.16.0. `docs/json-output.md` + promises the run stream and the watch stream speak one dialect, so a consumer + written against the documented run dialect was undercounting every cached node + beyond the first on a watch stream. Consumers written to the documented shape + are unaffected or fixed by this; the `nodes` array remains accepted by the + published event schema so a stream captured from 1.16.0 still validates. + ## [1.16.0] - 2026-08-10 [Full notes](https://github.com/Comfy-Org/comfy-cli/releases/tag/v1.16.0) · 16 commits since v1.15.0. No breaking changes. diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 92ccbea6d..5ba05fe9e 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -1984,17 +1984,24 @@ def _watch_executing(state: _WatchState, data: dict[str, Any]) -> None: def _watch_execution_cached(state: _WatchState, data: dict[str, Any]) -> None: + # ONE event per cached node, matching what `comfy run` emits + # (run/execution.py `on_cached`). The two streams are documented as one + # dialect, so a consumer counting cached nodes per event must not have to + # special-case a list-shaped `nodes` here. `title`/`class_type` are omitted: + # watch has no workflow map to resolve them from, and both are optional in + # the run dialect's own emission. nodes = data.get("nodes") or [] for n in nodes: state.completed_nodes.add(str(n)) renderer = state.renderer if renderer.is_pretty(): renderer.console().print(f"[dim]✓[/dim] cached: {len(nodes)} node(s)") - renderer.event( - "execution_cached", - nodes=[str(n) for n in nodes], - prompt_id=state.prompt_id, - ) + for n in nodes: + renderer.event( + "execution_cached", + node=str(n), + prompt_id=state.prompt_id, + ) def _watch_progress(state: _WatchState, data: dict[str, Any]) -> None: diff --git a/comfy_cli/schemas/run_event.json b/comfy_cli/schemas/run_event.json index 5d2cd846e..d031d611b 100644 --- a/comfy_cli/schemas/run_event.json +++ b/comfy_cli/schemas/run_event.json @@ -29,7 +29,7 @@ ], "description": "`comfy run` emits converted?/prompt_preview/queued on both targets; its per-node types (executing, execution_cached, progress, executed, output, execution_error) are --where local only, because the cloud run path polls for a terminal record instead of streaming a session. `state` is emitted by `comfy jobs watch --where cloud` for each coarse status transition. This enum is ADVISORY and open-ended: it lists the types this comfy-cli version emits, and adding a type is an additive change that does NOT bump `event/1`. Agents must ignore types they do not recognise rather than treat them as a validation failure — fetch the current schema with `comfy --json discover` instead of pinning this list." }, - "node": {"type": ["string", "null"]}, + "node": {"type": ["string", "null"], "description": "The node this per-node event is about. `execution_cached` carries exactly one node id per event on BOTH the `comfy run` and the `comfy jobs watch` stream."}, "title": {"type": ["string", "null"]}, "class_type": {"type": ["string", "null"]}, "completed": {"type": ["integer", "null"]}, @@ -45,7 +45,7 @@ }, "nodes": { "type": ["array", "null"], - "description": "queued: manifest of every node in the submitted graph, as objects (node_id, class_type, title). Also carried by `comfy jobs watch`'s execution_cached event, there as a plain array of node-id strings. The per-type `allOf` below pins which of the two shapes is legal for each event, so a `queued` that regressed to bare id strings still fails validation." + "description": "queued: manifest of every node in the submitted graph, as objects (node_id, class_type, title). LEGACY on execution_cached: comfy-cli 1.16.0's `comfy jobs watch` emitted a single execution_cached carrying a plain array of node-id strings; it now emits one event per node with `node` instead, matching `comfy run`. The array form stays legal so a stream captured from 1.16.0 still validates, but nothing emits it. The per-type `allOf` below pins which of the two shapes is legal for each event, so a `queued` that regressed to bare id strings still fails validation." }, "base_url": {"type": ["string", "null"], "description": "queued, --where cloud only: the cloud endpoint the prompt was submitted to."}, "url": {"type": ["string", "null"]} diff --git a/docs/json-output.md b/docs/json-output.md index 24071c669..3ee66132d 100644 --- a/docs/json-output.md +++ b/docs/json-output.md @@ -40,6 +40,21 @@ it reconnects. is the union of what the watch observed and what `/history` records for the prompt, so it is populated even for a watch that attached after the job ended. +The per-node events `jobs watch` streams carry the **same shape** as the `comfy +run` ones — one event per node, keyed by `node`. ComfyUI's websocket delivers +`execution_cached` as a single message listing every cached node, and `jobs +watch` fans that out into one `execution_cached` event per node, exactly as +`comfy run` does, so a consumer written against the run dialect counts cached +nodes correctly on either stream. `jobs watch` has no workflow graph in hand, +so it omits the optional `title` / `class_type` fields that `comfy run` +attaches to `executing`, `execution_cached` and `executed`; it also does not +carry the `outputs` array on `executed`, reporting each artifact as its own +`output` event instead. *(Changed since 1.16.0, whose `jobs watch` emitted one +`execution_cached` carrying a `nodes` array — a run-dialect consumer counting +cached nodes per event undercounted every cached node beyond the first. The +`nodes` array form is still accepted by the published event schema so a stream +captured from 1.16.0 keeps validating, but nothing emits it.)* + ## Overview When `--json` is passed, `comfy run` switches into a strict @@ -306,10 +321,20 @@ closing the previous one. ### `execution_cached` One event per node whose outputs were retrieved from the execution cache -(from ComfyUI's `execution_cached` websocket message). Same fields as +(from ComfyUI's `execution_cached` websocket message, which lists every +cached node in one frame and is fanned out here). Same fields as `executing`. A cached output-bearing node (e.g., a cached `SaveImage`) may emit both `execution_cached` AND `executed`. +```json +{"schema": "event/1", "type": "execution_cached", "node": "1", "title": "Latent", "class_type": "EmptyLatentImage", "prompt_id": "9b1c…"} +``` + +`comfy jobs watch` emits the same per-node event without `title` / +`class_type` (see [`jobs watch` attaches as the submitting +session](#jobs-watch-attaches-as-the-submitting-session)). It never emits a +list-shaped `nodes` field on this event. + ### `progress` Per-step progress for samplers, video encoders, and any node that calls diff --git a/tests/comfy_cli/command/test_run_json.py b/tests/comfy_cli/command/test_run_json.py index 0d6db0e87..ce0c5592d 100644 --- a/tests/comfy_cli/command/test_run_json.py +++ b/tests/comfy_cli/command/test_run_json.py @@ -2007,9 +2007,27 @@ def test_queued_requires_object_node_records(self): self._validator().validate(regressed) def test_execution_cached_still_accepts_bare_node_ids(self): - """`comfy jobs watch --where cloud` emits plain id strings here.""" + """Legacy shape: comfy-cli 1.16.0's `comfy jobs watch` emitted one + `execution_cached` carrying a plain array of node ids. Nothing emits it + any more — watch now fans out one event per node, matching `comfy run` — + but a stream captured from 1.16.0 must keep validating.""" self._validator().validate({"schema": "event/1", "type": "execution_cached", "nodes": ["1", "2"]}) + def test_execution_cached_per_node_shape_validates(self): + """The shape BOTH streams emit today — one event, one `node`.""" + self._validator().validate( + { + "schema": "event/1", + "type": "execution_cached", + "node": "1", + "title": "Latent", + "class_type": "EmptyLatentImage", + "prompt_id": "p", + } + ) + # `comfy jobs watch` has no workflow map, so it omits title/class_type. + self._validator().validate({"schema": "event/1", "type": "execution_cached", "node": "1", "prompt_id": "p"}) + class TestMalformedRejectionPayloadStillYieldsAnEnvelope: """Every field of the cloud's `node_errors` is server-supplied and only diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index db66263f9..0c1d6369a 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -3181,7 +3181,50 @@ def test_watch_execution_cached_accumulates_completed_nodes(): st, r = _watch_state() jobs_mod._watch_execution_cached(st, {"nodes": [1, 2]}) assert st.completed_nodes == {"1", "2"} - assert r.events == [("execution_cached", {"nodes": ["1", "2"], "prompt_id": "pid"})] + assert r.events == [ + ("execution_cached", {"node": "1", "prompt_id": "pid"}), + ("execution_cached", {"node": "2", "prompt_id": "pid"}), + ] + + +def test_watch_execution_cached_emits_one_event_per_node(): + """`comfy run` emits one `execution_cached` per cached node, and + `docs/json-output.md` promises the two streams speak one dialect. A single + list-shaped event made a run-dialect consumer undercount every cached node + beyond the first.""" + st, r = _watch_state() + jobs_mod._watch_execution_cached(st, {"nodes": ["7", 8, "9"]}) + + assert len(r.events) == 3 + assert [name for name, _ in r.events] == ["execution_cached"] * 3 + assert [kw["node"] for _, kw in r.events] == ["7", "8", "9"] + # The list shape is gone — no event carries `nodes`. + assert all("nodes" not in kw for _, kw in r.events) + # Accumulation is unchanged: every cached node still lands in the envelope. + assert st.completed_nodes == {"7", "8", "9"} + + +def test_watch_execution_cached_events_validate_against_the_run_event_schema(): + """`comfy jobs watch` publishes the `run_event` schema (discovery.py), so + each emitted event must validate against it in the per-node shape.""" + import jsonschema + + st, r = _watch_state() + jobs_mod._watch_execution_cached(st, {"nodes": ["1", "2", "3"]}) + + schema_path = Path(jobs_mod.__file__).parent.parent / "schemas" / "run_event.json" + validator = jsonschema.Draft202012Validator(json.loads(schema_path.read_text())) + for name, kw in r.events: + validator.validate({"schema": "event/1", "type": name, **kw}) + + +def test_watch_execution_cached_empty_list_emits_nothing(): + st, r = _watch_state() + jobs_mod._watch_execution_cached(st, {"nodes": []}) + assert r.events == [] + assert st.completed_nodes == set() + jobs_mod._watch_execution_cached(st, {}) + assert r.events == [] def test_watch_progress_uses_throttled_event(): @@ -3562,6 +3605,38 @@ def test_watch_already_terminal_job_still_lists_completed_nodes(monkeypatch, cap jsonschema.Draft202012Validator(json.loads(schema_path.read_text())).validate(lines[-1]["data"]) +def test_watch_stream_fans_execution_cached_out_per_node(monkeypatch, capsys): + """End to end: a WS `execution_cached` naming 3 nodes must reach the NDJSON + stream as 3 run-dialect events, each validating against the published + `run_event` schema.""" + import jsonschema + + from comfy_cli import jobs_state + + jobs_state.write(jobs_state.new(prompt_id="pid-w", client_id="cid-sub", workflow="w", where="local")) + monkeypatch.setattr(jobs_mod, "_snapshot", lambda h, p, pid: {"prompt_id": pid, "status": "running", "outputs": []}) + monkeypatch.setattr(jobs_mod, "_history_completed_nodes", lambda h, p, pid: set()) + + messages = [ + {"type": "execution_cached", "data": {"prompt_id": "pid-w", "nodes": ["4", "5", "6"]}}, + {"type": "execution_success", "data": {"prompt_id": "pid-w"}}, + ] + result, _ws, lines = _run_local_watch(monkeypatch, capsys, messages=messages) + assert result.exit_code == 0, result.output + + cached = [ln for ln in lines if ln.get("type") == "execution_cached"] + assert [ln["node"] for ln in cached] == ["4", "5", "6"] + assert all("nodes" not in ln for ln in cached) + + schema_path = Path(jobs_mod.__file__).parent.parent / "schemas" / "run_event.json" + validator = jsonschema.Draft202012Validator(json.loads(schema_path.read_text())) + for ln in cached: + validator.validate(ln) + + # Every cached node still reaches the terminal envelope. + assert lines[-1]["data"]["completed_nodes"] == ["4", "5", "6"] + + def test_watch_client_id_flag_overrides_resolution(monkeypatch, capsys): monkeypatch.setattr(jobs_mod, "_snapshot", lambda h, p, pid: {"prompt_id": pid, "status": "running", "outputs": []}) monkeypatch.setattr(jobs_mod, "_resolve_watch_client_id", lambda h, p, pid: "resolved") From 684361d715d76dace9c2ff324166bb57ad29d606 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Wed, 26 Aug 2026 21:16:59 -0700 Subject: [PATCH 2/2] fix(jobs): apply the same cached-node guards to the /history path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review catch. `_history_completed_nodes` feeds the *same* `completed_nodes` set that `_watch_execution_cached` does — both end up in the terminal envelope — and it read `/history`'s `execution_cached` `nodes` and `execution_error`/`execution_interrupted` `executed` lists with a bare `str(n)`. So the null fix in 1f29cf8e only closed one of the two doors: a null id recorded in `/history` would still have seated a fabricated `"None"` node in the envelope, by the other path, from the same ComfyUI source that produces the websocket frame. Guarding the live path alone would have made that commit's own claim untrue. Non-list values get the same `isinstance` treatment, matching both the new watch guard and the isinstance checks this function already applies to every other server-supplied field it reads. Co-Authored-By: Claude Opus 5 --- comfy_cli/command/jobs.py | 8 ++++++-- tests/comfy_cli/jobs/test_jobs.py | 23 +++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 8f02e43ea..452650a49 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -1932,8 +1932,12 @@ def _history_completed_nodes(host: str, port: int, prompt_id: str) -> set[str]: listed = msg[1].get("executed") else: continue - for n in listed or []: - nodes.add(str(n)) + if not isinstance(listed, list): + continue + # Same null guard as the live `_watch_execution_cached` path: this set + # becomes the terminal envelope's completed nodes, and `str(None)` + # would seat a fabricated `"None"` node in it. + nodes.update(str(n) for n in listed if n is not None) outputs = body.get("outputs") if isinstance(outputs, dict): nodes.update(str(n) for n in outputs) diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index 848619745..8fe2b49a1 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -3488,6 +3488,29 @@ def fake_get(url, **kw): assert jobs_mod._history_completed_nodes("127.0.0.1", 8188, "pid-j") == set() +def test_history_completed_nodes_skips_nulls_and_non_list_node_fields(monkeypatch): + """`/history` feeds the same `completed_nodes` set the live stream does, so + it needs the same guards: a null id would seat a fabricated `"None"` node in + the terminal envelope, and a non-list `nodes` would be walked per character.""" + + def fake_get(url, **kw): + return { + "pid-n": { + "status": { + "messages": [ + ["execution_cached", {"nodes": [None, "1"]}], + ["execution_interrupted", {"executed": [None, "2"]}], + ["execution_cached", {"nodes": "34"}], + ], + }, + "outputs": {}, + } + } + + monkeypatch.setattr(jobs_mod, "_http_get_json", fake_get) + assert jobs_mod._history_completed_nodes("127.0.0.1", 8188, "pid-n") == {"1", "2"} + + class _ScriptedWS: """A `websocket.WebSocket` stand-in that replays a scripted message list.