fix(run): conform the cloud NDJSON stream to the documented event contract

comfy_cli/cmdline.py

@@ -1001,30 +1001,34 @@ def run(
                 preloaded=preloaded,
                 allow_spend=allow_spend,
             )
-            return
-
-        from comfy_cli.host_port import parse_host_port_arg, resolve_host_port
-
-        if host:
-            host, parsed_port = parse_host_port_arg(host)
-            if not port and parsed_port is not None:
-                port = parsed_port
-
-        host, port = resolve_host_port(host, port)
-
-        run_inner.execute(
-            workflow,
-            host,
-            port,
-            wait=wait,
-            verbose=verbose,
-            timeout=timeout,
-            notify=effective_notify,
-            api_key=api_key,
-            print_prompt=print_prompt,
-            preloaded=preloaded,
-            allow_spend=allow_spend,
-        )
+        else:
+            # Both targets must fall off the END of this try suite: Python skips
+            # a try's `else:` clause when the suite leaves via `return`, so an
+            # early return here would silently drop the `execution_success`
+            # tracking below (it only ever fired on the cloud branch by way of
+            # the `except typer.Exit` handler).
+            from comfy_cli.host_port import parse_host_port_arg, resolve_host_port
+
+            if host:
+                host, parsed_port = parse_host_port_arg(host)
+                if not port and parsed_port is not None:
+                    port = parsed_port
+
+            host, port = resolve_host_port(host, port)
+
+            run_inner.execute(
+                workflow,
+                host,
+                port,
+                wait=wait,
+                verbose=verbose,
+                timeout=timeout,
+                notify=effective_notify,
+                api_key=api_key,
+                print_prompt=print_prompt,
+                preloaded=preloaded,
+                allow_spend=allow_spend,
+            )
     except typer.Exit as e:
         if (e.exit_code or 0) == 0:
             tracking.track_event("execution_success", _track_props)

comfy_cli/command/run/__init__.py

@@ -29,6 +29,7 @@
 from comfy_cli.command.run.execution import ExecutionProgress as ExecutionProgress
 from comfy_cli.command.run.execution import WorkflowExecution as WorkflowExecution
 from comfy_cli.command.run.execution import _safe_close as _safe_close
+from comfy_cli.command.run.execution import workflow_manifest as workflow_manifest
 from comfy_cli.command.run.loader import _MAX_BODY_PREVIEW as _MAX_BODY_PREVIEW
 from comfy_cli.command.run.loader import WorkflowLoadError as WorkflowLoadError
 from comfy_cli.command.run.loader import _classify_api_workflow as _classify_api_workflow
@@ -779,6 +780,9 @@ def execute_cloud(
                 message="Workflow conversion produced no executable nodes",
             )
             raise typer.Exit(code=1)
+        # Same signal the local path emits at this point (docs/json-output.md
+        # "converted"): the input was a UI export and was lowered client-side.
+        renderer.event("converted", node_count=len(raw_workflow))
 
     kind, parsed_workflow = _classify_api_workflow(raw_workflow)
     if kind != "ok":
@@ -793,19 +797,27 @@ def execute_cloud(
     # its foreach item map to stash on the job state at submit time.
     compose_meta = pop_compose_meta(parsed_workflow)
 
+    # Stream mode: emit the workflow graph so agents have a complete audit
+    # trail of what the CLI is about to submit (no-op otherwise). Emitted
+    # unconditionally, like the local path — docs/json-output.md documents
+    # `prompt_preview` in every stream except the pre-flight-failure archetype,
+    # so gating it behind --print-prompt made the cloud stream undocumented.
+    renderer.event("prompt_preview", prompt=parsed_workflow)
+
     if print_prompt:
         # Documented dry-run: show the API-format graph that WOULD be sent and
-        # exit WITHOUT POSTing. Mirrors local execute()'s print_prompt branch.
+        # exit WITHOUT POSTing. Mirrors local execute()'s print_prompt branch,
+        # including returning (rather than raising `typer.Exit(0)`) so both
+        # targets end this stream the same way.
         if renderer.is_pretty():
             print(json.dumps(parsed_workflow, indent=2, ensure_ascii=False))
         else:
-            renderer.event("prompt_preview", prompt=parsed_workflow)
             renderer.emit(
                 {"workflow": workflow_name, "status": "preview", "prompt": parsed_workflow},
                 command="run",
                 where="cloud",
             )
-        raise typer.Exit(code=0)
+        return
 
     # Pre-submit validation via pure-Python CQL engine.
     # Cloud path uses cached/bundled object_info (no live server needed).
@@ -840,14 +852,15 @@ def execute_cloud(
     client_id = str(uuid.uuid4())
     start = time.time()
 
-    if wait:
-        if renderer.is_pretty():
-            pprint(f"[dim]▸[/dim] Executing [cyan]{workflow_name}[/cyan] on Comfy Cloud")
-            pprint(f"[dim]  base_url: {target.base_url}[/dim]")
-        else:
-            renderer.event("executing", workflow=workflow_name, base_url=target.base_url)
-    elif not renderer.is_pretty():
-        renderer.event("queued", workflow=workflow_name, base_url=target.base_url)
+    if wait and renderer.is_pretty():
+        pprint(f"[dim]▸[/dim] Executing [cyan]{workflow_name}[/cyan] on Comfy Cloud")
+        pprint(f"[dim]  base_url: {target.base_url}[/dim]")
+
+    # NOTE: no machine event is emitted here. This used to announce the submit
+    # with `queued` (async) / `executing` (--wait) carrying {workflow, base_url}
+    # — both wrong per docs/json-output.md: `queued` means "the server accepted
+    # the prompt" (so it cannot precede the POST), and `executing` is a per-node
+    # event. The contractual `queued` is emitted after submit, below.
 
     try:
         if not wait and renderer.is_pretty():
@@ -868,23 +881,54 @@ def execute_cloud(
         raise typer.Exit(code=1) from e
 
     if submit.node_errors:
-        # Parse per-node errors into readable hint lines
+        # Parse per-node errors into readable hint lines. Every field here is
+        # server-supplied and only documented by convention, so each level is
+        # shape-checked rather than duck-typed: an AttributeError escaping this
+        # loop would abort the process with a traceback and no envelope at all,
+        # breaking the "exactly one terminal envelope" guarantee this contract
+        # rests on. A record that isn't the documented dict still gets a line,
+        # since it is preserved in `details.node_errors` too.
         hint_lines = []
         for nid, record in submit.node_errors.items():
             if not isinstance(record, dict):
+                hint_lines.append(f"node {nid}: {record}")
                 continue
             ct = record.get("class_type", "unknown")
-            for err in record.get("errors") or []:
-                detail = err.get("details", "") or err.get("message", "")
+            errors = record.get("errors") or []
+            if not isinstance(errors, list):
+                errors = [errors]
+            for err in errors:
+                if isinstance(err, dict):
+                    detail = err.get("details", "") or err.get("message", "")
+                else:
+                    detail = err
                 hint_lines.append(f"node {nid} ({ct}): {detail}")
         renderer.error(
             code="prompt_rejected",
             message=f"Cloud server rejected {len(submit.node_errors)} node(s)",
             hint="\n".join(hint_lines) if hint_lines else "inspect node_errors in details",
-            details={"node_errors": submit.node_errors},
+            # Documented shape (docs/json-output.md#node_errors-shape): an array
+            # of self-contained records carrying `node_id`, not the server's
+            # id-keyed dict. Same transform the local path applies.
+            details={"node_errors": _node_errors_to_list(submit.node_errors)},
         )
         raise typer.Exit(code=1)
 
+    # The contractual `queued`: the server has the prompt. Same shape the local
+    # path emits from `WorkflowExecution.queue()` (docs/json-output.md#queued),
+    # plus `base_url` so a cloud consumer still learns the target.
+    # `validation_warnings` is always empty here — unlike local, the cloud
+    # treats any `node_errors` on an accepted submit as a hard `prompt_rejected`
+    # above, so a partially-valid graph never reaches this line.
+    renderer.event(
+        "queued",
+        prompt_id=submit.prompt_id,
+        client_id=client_id,
+        validation_warnings=[],
+        nodes=workflow_manifest(parsed_workflow),
+        base_url=target.base_url,
+    )
+
     if not wait:
         state = jobs_state.new(
             prompt_id=submit.prompt_id,
@@ -919,6 +963,11 @@ def execute_cloud(
                 "elapsed_seconds": None,
                 "base_url": target.base_url,
                 "state_file": str(state_file) if state_file else None,
+                # Same async-envelope field the local path carries: whether the
+                # detached watcher that keeps the state file fresh actually
+                # started. Consumers poll `comfy jobs status` themselves when
+                # it is false.
+                "watcher_spawned": watcher_spawned,
             },
             command="run",
             where="cloud",

comfy_cli/command/run/execution.py

@@ -56,6 +56,47 @@ def _safe_close(execution: WorkflowExecution) -> None:
         pass
 
 
+def node_title(workflow: dict, node_id) -> str:
+    """Display label: ``_meta.title`` if present, else ``class_type``, else the
+    node id. Defensive against unknown ids and non-dict nodes."""
+    node = workflow.get(node_id)
+    if node is None and not isinstance(node_id, str):
+        node = workflow.get(str(node_id))
+    if not isinstance(node, dict):
+        return str(node_id)
+    meta = node.get("_meta")
+    if isinstance(meta, dict):
+        title = meta.get("title")
+        if isinstance(title, str) and title:
+            return title
+    class_type = node.get("class_type")
+    return class_type if isinstance(class_type, str) and class_type else str(node_id)
+
+
+def workflow_manifest(workflow: dict) -> list[dict]:
+    """Build the `nodes` array for the `queued` event — one entry per node in
+    the submitted (post-conversion) workflow.
+
+    Module-level so the cloud submit path (``run.execute_cloud``) emits a
+    byte-identical manifest without owning a ``WorkflowExecution``: the two
+    pipelines are separate, the ``queued`` contract is not.
+    """
+    manifest: list[dict] = []
+    for node_id, node in workflow.items():
+        if not isinstance(node, dict):
+            continue
+        class_type = node.get("class_type", "")
+        class_type = class_type if isinstance(class_type, str) else ""
+        manifest.append(
+            {
+                "node_id": str(node_id),
+                "class_type": class_type,
+                "title": node_title(workflow, node_id),
+            }
+        )
+    return manifest
+
+
 class ExecutionProgress(Progress):
     def get_renderables(self):
         table_columns = (
@@ -147,20 +188,7 @@ def connect(self):
     def workflow_manifest(self) -> list[dict]:
         """Build the `nodes` array for the `queued` event — one entry per
         node in the submitted (post-conversion) workflow."""
-        manifest: list[dict] = []
-        for node_id, node in self.workflow.items():
-            if not isinstance(node, dict):
-                continue
-            class_type = node.get("class_type", "")
-            class_type = class_type if isinstance(class_type, str) else ""
-            manifest.append(
-                {
-                    "node_id": str(node_id),
-                    "class_type": class_type,
-                    "title": self.get_node_title(node_id),
-                }
-            )
-        return manifest
+        return workflow_manifest(self.workflow)
 
     def queue(self):
         data: dict = {"prompt": self.workflow, "client_id": self.client_id}
@@ -296,18 +324,7 @@ def update_overall_progress(self):
     def get_node_title(self, node_id):
         """Display label: ``_meta.title`` if present, else ``class_type``,
         else the node id. Defensive against unknown ids and non-dict nodes."""
-        node = self.workflow.get(node_id)
-        if node is None and not isinstance(node_id, str):
-            node = self.workflow.get(str(node_id))
-        if not isinstance(node, dict):
-            return str(node_id)
-        meta = node.get("_meta")
-        if isinstance(meta, dict):
-            title = meta.get("title")
-            if isinstance(title, str) and title:
-                return title
-        class_type = node.get("class_type")
-        return class_type if isinstance(class_type, str) and class_type else str(node_id)
+        return node_title(self.workflow, node_id)
 
     def _class_type(self, node_id):
         node = self.workflow.get(node_id)

comfy_cli/command/run/loader.py

@@ -22,10 +22,19 @@ def _node_errors_to_list(node_errors) -> list[dict]:
         return []
     result = []
     for node_id, record in node_errors.items():
-        if not isinstance(record, dict):
-            continue
-        entry = {"node_id": str(node_id)}
-        entry.update(record)
+        if isinstance(record, dict):
+            # Spread the server's record FIRST so the authoritative map key
+            # below wins if a (less-trusted, cloud-supplied) record carries a
+            # `node_id` of its own.
+            entry = dict(record)
+        else:
+            # A server reporting a bare value instead of the documented per-node
+            # dict (e.g. `{"1": "missing input"}`) must not vanish: an empty
+            # array under a "rejected N node(s)" message would strand the caller
+            # with no diagnostic at all. Wrap it so the outer shape stays
+            # uniform — one record per node, each carrying `node_id`.
+            entry = {"errors": list(record) if isinstance(record, list) else [record]}
+        entry["node_id"] = str(node_id)
         result.append(entry)
     return result
 

comfy_cli/schemas/run.json

@@ -7,27 +7,31 @@
   "additionalProperties": true,
   "properties": {
     "workflow": {"type": "string", "description": "Absolute path to the workflow file."},
-    "status": {"type": "string", "enum": ["queued", "completed", "cancelled"]},
+    "status": {"type": "string", "enum": ["queued", "completed", "cancelled", "preview"]},
     "prompt_id": {"type": ["string", "null"]},
     "client_id": {"type": ["string", "null"]},
+    "prompt": {
+      "type": "object",
+      "description": "--print-prompt only (both targets, status \"preview\"): the API-format workflow graph that WOULD be submitted, keyed by node id. Present only on the preview dry-run; a real submit omits it. Deliberately unconstrained beyond \"object\" — node shapes are the server's contract, not this CLI's, and over-tightening here is what made these schemas reject real output in the first place."
+    },
     "outputs": {
       "type": "array",
       "items": {"type": "string"},
       "description": "URLs or local paths to produced artifacts."
     },
     "outputs_by_node": {
       "type": "object",
-      "description": "Cloud --wait only: the same artifact URLs grouped by the node id that produced them.",
+      "description": "--wait only (both targets): the same artifact URLs grouped by the node id that produced them.",
       "additionalProperties": {"type": "array", "items": {"type": "string"}}
     },
     "outputs_by_item": {
       "type": "object",
-      "description": "Cloud --wait only: artifact URLs grouped by blueprint foreach item (via the compose item_map). Empty object when no item_map exists.",
+      "description": "--wait only (both targets): artifact URLs grouped by blueprint foreach item (via the compose item_map). Empty object when no item_map exists.",
       "additionalProperties": {"type": "array", "items": {"type": "string"}}
     },
     "warnings": {
       "type": "array",
-      "description": "Non-fatal diagnostics about the run (e.g. partial_execution when the cloud pruned an output branch). Empty when none.",
+      "description": "Cloud --wait only: non-fatal diagnostics about the run (e.g. partial_execution when the cloud pruned an output branch). Empty when none.",
       "items": {
         "type": "object",
         "required": ["code", "message"],
@@ -38,8 +42,24 @@
         }
       }
     },
+    "cached_node_ids": {
+      "type": "array",
+      "items": {"type": "string"},
+      "description": "Local --wait only: node ids the server reported as cached. Derived from the per-node event stream, which --where cloud does not have."
+    },
+    "executed_node_ids": {
+      "type": "array",
+      "items": {"type": "string"},
+      "description": "Local --wait only: node ids the executor ran. Derived from the per-node event stream, which --where cloud does not have."
+    },
     "elapsed_seconds": {"type": ["number", "null"]},
-    "host": {"type": "string"},
-    "port": {"type": "integer"}
+    "host": {"type": "string", "description": "--where local only: target server host."},
+    "port": {"type": "integer", "description": "--where local only: target server port."},
+    "base_url": {"type": "string", "description": "--where cloud only: target cloud endpoint."},
+    "state_file": {"type": ["string", "null"], "description": "Path of the job state file (poll with `comfy jobs status`)."},
+    "watcher_spawned": {
+      "type": "boolean",
+      "description": "Async (no --wait) only, both targets: whether the detached state-file watcher started. Poll `comfy jobs status` yourself when false."
+    }
   }
 }