fix(jobs): emit one execution_cached event per node in jobs watch

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": "<id>", …}`), 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.

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:

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"}}
+            }
+          }
+        ]
       }
     }
   ]

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

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

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")