diff --git a/CHANGELOG.md b/CHANGELOG.md index b4970ebb3..f8553db99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -119,6 +119,24 @@ history. the builder actually serves; it still required the pre-rename `distribution` key, so a valid `comfy build from-snapshot --json` payload failed validation. +- 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. **Breaking for a `jobs watch --json` + consumer that reads `ev["nodes"]` on this event**: the published `run_event` + schema did describe that array as watch's shape, and watch no longer emits + it — read `ev["node"]`, one event per node, the same field the run stream has + always required. `event/1` is deliberately not bumped: the list shape existed + for a single release and was itself the deviation from the one dialect the + schema and `docs/json-output.md` document, and no other event type changes — + bumping the CLI-wide event contract would force every consumer of every event + to revalidate over a one-event, one-release regression. The `nodes` array + remains accepted by the published schema so a stream captured from 1.16.0 + still validates, but nothing emits it. + ## [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..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) @@ -1984,17 +1988,33 @@ def _watch_executing(state: _WatchState, data: dict[str, Any]) -> None: def _watch_execution_cached(state: _WatchState, data: dict[str, Any]) -> None: - nodes = data.get("nodes") or [] - for n in nodes: - state.completed_nodes.add(str(n)) + # 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. + raw = data.get("nodes") + if not isinstance(raw, list): + # Server-supplied. A bare string would otherwise iterate per character + # and fan out one bogus event each, and a non-iterable would raise + # TypeError out of the handler; sibling handlers guard the same way + # (see `_watch_progress_state`'s isinstance check). + return + # `comfy run`'s on_cached skips null entries (run/execution.py). Do the + # same: stringifying one would emit a phantom `node: "None"` event and + # record a fabricated id in the terminal envelope's completed nodes. + nodes = [str(n) for n in raw if n is not None] + state.completed_nodes.update(nodes) 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 node_id in nodes: + renderer.event( + "execution_cached", + node=node_id, + 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..589c4b954 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. Every `execution_cached` emitted by this version carries exactly one node id here, on BOTH the `comfy run` and the `comfy jobs watch` stream; the `allOf` branch below requires it whenever the legacy `nodes` array is absent. A stream captured from comfy-cli 1.16.0 may instead carry the list-shaped `nodes` (see its description)."}, "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"]} @@ -62,9 +62,21 @@ { "if": {"required": ["type"], "properties": {"type": {"const": "execution_cached"}}}, "then": { - "properties": { - "nodes": {"type": ["array", "null"], "items": {"type": "string"}} - } + "description": "Exactly one of the two shapes, never both and never neither: the per-node `node` string every emitter produces today, or the legacy `nodes` array of id strings that only comfy-cli 1.16.0's `comfy jobs watch` ever emitted.", + "oneOf": [ + { + "required": ["node"], + "not": {"required": ["nodes"]}, + "properties": {"node": {"type": "string"}} + }, + { + "required": ["nodes"], + "not": {"required": ["node"]}, + "properties": { + "nodes": {"type": ["array", "null"], "items": {"type": "string"}} + } + } + ] } } ] diff --git a/docs/json-output.md b/docs/json-output.md index 24071c669..d49f01c6c 100644 --- a/docs/json-output.md +++ b/docs/json-output.md @@ -40,6 +40,23 @@ 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. If you +wrote a watch consumer against 1.16.0 and read `ev["nodes"]` on this event, +read `ev["node"]` instead, once per event. 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 +323,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..c9bc269af 100644 --- a/tests/comfy_cli/command/test_run_json.py +++ b/tests/comfy_cli/command/test_run_json.py @@ -2007,9 +2007,55 @@ 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"}) + + def test_execution_cached_rejects_an_event_carrying_neither_shape(self): + """`node` and the legacy `nodes` are the only two payloads. An event + with neither names no node at all, so a consumer counting cached nodes + silently reads zero — the schema must catch that, not pass it.""" + import jsonschema + + with pytest.raises(jsonschema.ValidationError): + self._validator().validate({"schema": "event/1", "type": "execution_cached", "prompt_id": "p"}) + + def test_execution_cached_rejects_an_event_carrying_both_shapes(self): + """Both fields at once is ambiguous: a consumer reading `node` and one + reading `nodes` would disagree about how many nodes were cached.""" + import jsonschema + + with pytest.raises(jsonschema.ValidationError): + self._validator().validate( + {"schema": "event/1", "type": "execution_cached", "node": "1", "nodes": ["1", "2"]} + ) + + def test_execution_cached_rejects_a_null_node(self): + """Neither emitter can produce one: `comfy run`'s on_cached skips null + entries and `jobs watch` filters them, so `node: null` is a regression + that would surface a phantom cached node.""" + import jsonschema + + with pytest.raises(jsonschema.ValidationError): + self._validator().validate({"schema": "event/1", "type": "execution_cached", "node": None}) + 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..8fe2b49a1 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -3181,7 +3181,73 @@ 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_execution_cached_skips_null_nodes(): + """`comfy run`'s on_cached does `if n is None: continue`, and the two + streams speak one dialect. Stringifying a null would emit a phantom + `node: "None"` event and put a fabricated id in the terminal envelope.""" + st, r = _watch_state() + jobs_mod._watch_execution_cached(st, {"nodes": [None, "1"]}) + + assert [kw["node"] for _, kw in r.events] == ["1"] + assert st.completed_nodes == {"1"} + + +def test_watch_execution_cached_ignores_a_non_list_nodes_field(): + """`nodes` is server-supplied. A bare string would otherwise fan out one + bogus event per character, and a non-iterable would raise TypeError out of + the handler and tear down the watch.""" + st, r = _watch_state() + for malformed in ("12", 5, {"1": "x"}): + jobs_mod._watch_execution_cached(st, {"nodes": malformed}) + + assert r.events == [] + assert st.completed_nodes == set() def test_watch_progress_uses_throttled_event(): @@ -3422,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. @@ -3562,6 +3651,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")