diff --git a/comfy_cli/cmdline.py b/comfy_cli/cmdline.py
index 0eccad95..49ddf67b 100644
--- a/comfy_cli/cmdline.py
+++ b/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)
diff --git a/comfy_cli/command/run/__init__.py b/comfy_cli/command/run/__init__.py
index ad299286..ce0dff94 100644
--- a/comfy_cli/command/run/__init__.py
+++ b/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
@@ -384,6 +385,12 @@ def execute(
             wait_state.status = "running"
             _write_state(wait_state)
 
+            # Only now, with the durable record on disk, announce the queue.
+            # The event-order contract (docs/json-output.md) is
+            # `prompt_preview → queued → node events`, so this must also
+            # precede `watch_execution`'s WebSocket stream.
+            _emit_queued(renderer, execution)
+
             # `watch_execution` reports a terminal server event by rendering
             # the error and raising `typer.Exit` (1 for `execution_error`, 130
             # for `execution_interrupted`) — the ordinary failure path, not an
@@ -459,6 +466,10 @@ def execute(
             state_file = jobs_state.write(state)
             watcher_spawned = _spawn_watcher(execution.prompt_id, where="local", host=host, port=port, notify=notify)
 
+            # Emitted after the state file and the watcher spawn attempt, so a
+            # consumer reading this line can already `comfy jobs status` it.
+            _emit_queued(renderer, execution)
+
             if renderer.is_pretty():
                 from comfy_cli.output.glyphs import status_glyph
 
@@ -526,6 +537,12 @@ def execute(
         )
         raise typer.Exit(code=1)
     except (WebSocketException, ConnectionError, OSError) as e:
+        if isinstance(e, BrokenPipeError):
+            # A closed stdout is the consumer leaving, not the server dying —
+            # let click/__main__ turn it into the documented silent exit 0 and
+            # leave the just-written state record untouched. Same principle as
+            # `completed_payload` being emitted outside this try.
+            raise
         # If we closed the WebSocket ourselves in response to Ctrl-C, the recv
         # loop exits with a WebSocketException that *looks* like the server
         # vanished. Check the cancellation token first so we emit the right
@@ -593,6 +610,24 @@ def execute(
         renderer.emit(completed_payload, command="run", where="local")
 
 
+def _emit_queued(renderer, execution) -> None:
+    """Emit the contractual local `queued` event (docs/json-output.md#queued).
+
+    Lives in the caller rather than ``WorkflowExecution.queue()`` so it can be
+    emitted *after* the job's state file is persisted — a consumer that sees
+    this line can immediately `comfy jobs status <prompt_id>` / see the job in
+    `jobs ls`, and a `BrokenPipeError` here can no longer abort setup before
+    the durable record exists. The field set is unchanged.
+    """
+    renderer.event(
+        "queued",
+        prompt_id=execution.prompt_id,
+        client_id=execution.client_id,
+        validation_warnings=execution.validation_warnings,
+        nodes=execution.workflow_manifest(),
+    )
+
+
 def _write_state(state):
     """Best-effort ``jobs_state.write``. Returns the path, or None if the
     write was skipped, the state dir was unwritable, or the prompt_id was
@@ -779,6 +814,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 +831,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 +886,16 @@ 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 *and* after the
+    # job state file is written, below.
 
     try:
         if not wait and renderer.is_pretty():
@@ -868,23 +916,62 @@ 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 (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.
+    #
+    # Deliberately NOT emitted at this point: both branches below emit it only
+    # once the job's state file (and, on the async branch, the watcher) exist.
+    # A `BrokenPipeError` here — `comfy run --where cloud --json-stream … |
+    # head -n1`, where the NDJSON renderer flushes every line — used to abort
+    # setup before `jobs_state.write` ran while `__main__` still exited 0,
+    # leaving a billable cloud job with no journal entry, state file or watcher.
+    def _emit_queued_cloud() -> None:
+        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,
@@ -898,6 +985,9 @@ def execute_cloud(
         _journal_run(workflow_name, submit.prompt_id, "cloud")
         watcher_spawned = _spawn_watcher(submit.prompt_id, where="cloud", notify=notify)
 
+        # Durable record + watcher exist: safe to announce.
+        _emit_queued_cloud()
+
         if renderer.is_pretty():
             from comfy_cli.output.glyphs import status_glyph
 
@@ -919,6 +1009,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",
@@ -942,6 +1037,12 @@ def execute_cloud(
     state_file = jobs_state.write(state)
     _journal_run(workflow_name, submit.prompt_id, "cloud")
 
+    # Announced once the record is durable, and *outside* the polling try below
+    # — none of its handlers should ever see (and rewrite state for) a
+    # BrokenPipeError raised by this emit. It propagates to click/`__main__`,
+    # which exits 0 silently, with the job already recorded.
+    _emit_queued_cloud()
+
     try:
 
         def _probe():
diff --git a/comfy_cli/command/run/execution.py b/comfy_cli/command/run/execution.py
index 13e00257..7f4c62ec 100644
--- a/comfy_cli/command/run/execution.py
+++ b/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 = (
@@ -122,6 +163,12 @@ def __init__(
         # compute nodes that never fire a server-side `executed` event.
         self.cached_node_ids: list[str] = []
         self.executed_node_ids: list[str] = []
+        # Per-node issues the server reported alongside a successful (HTTP 200)
+        # queue, stashed by ``queue()`` for the caller's `queued` event. The
+        # event itself is emitted by ``comfy_cli.command.run.execute`` — only
+        # *after* the job's state file is persisted — so a consumer sees it and
+        # can immediately rely on `comfy jobs status` / `jobs ls` finding it.
+        self.validation_warnings: list[dict] = []
         # Classified verdict of a terminal server-side `execution_error`,
         # stashed by `on_error` before it raises `typer.Exit`. The `--wait`
         # caller only sees the exit, so this is how the real cause reaches
@@ -147,20 +194,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}
@@ -261,16 +295,11 @@ def queue(self):
 
         # 200 may still carry node_errors if some output chains failed
         # validation but others passed — surface as warnings, not a failure.
+        # Stored rather than emitted here: the contractual `queued` event is
+        # emitted by the caller once the job state file exists (see
+        # ``validation_warnings`` above and ``run._emit_queued``).
         node_errors = body.get("node_errors") if isinstance(body, dict) else None
-        validation_warnings = _node_errors_to_list(node_errors)
-
-        self.renderer.event(
-            "queued",
-            prompt_id=prompt_id,
-            client_id=self.client_id,
-            validation_warnings=validation_warnings,
-            nodes=self.workflow_manifest(),
-        )
+        self.validation_warnings = _node_errors_to_list(node_errors)
 
     def watch_execution(self):
         if self.ws is None:
@@ -296,18 +325,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)
diff --git a/comfy_cli/command/run/loader.py b/comfy_cli/command/run/loader.py
index 03bf093a..48b1a4d6 100644
--- a/comfy_cli/command/run/loader.py
+++ b/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
 
diff --git a/comfy_cli/schemas/run.json b/comfy_cli/schemas/run.json
index 3a41afaa..a1d3d163 100644
--- a/comfy_cli/schemas/run.json
+++ b/comfy_cli/schemas/run.json
@@ -7,9 +7,13 @@
   "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"},
@@ -17,17 +21,17 @@
     },
     "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."
+    }
   }
 }
diff --git a/comfy_cli/schemas/run_event.json b/comfy_cli/schemas/run_event.json
index 511ec81b..5d2cd846 100644
--- a/comfy_cli/schemas/run_event.json
+++ b/comfy_cli/schemas/run_event.json
@@ -14,6 +14,8 @@
     "type": {
       "type": "string",
       "enum": [
+        "converted",
+        "prompt_preview",
         "queued",
         "executing",
         "execution_cached",
@@ -21,8 +23,11 @@
         "executed",
         "execution_error",
         "output",
+        "state",
+        "login_url",
         "cancelled"
-      ]
+      ],
+      "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"]},
     "title": {"type": ["string", "null"]},
@@ -30,6 +35,37 @@
     "completed": {"type": ["integer", "null"]},
     "total": {"type": ["integer", "null"]},
     "prompt_id": {"type": ["string", "null"]},
+    "client_id": {"type": ["string", "null"]},
+    "node_count": {"type": ["integer", "null"], "description": "converted: node count of the client-side converted graph."},
+    "prompt": {"type": ["object", "null"], "description": "prompt_preview: the API-format graph about to be submitted."},
+    "validation_warnings": {
+      "type": ["array", "null"],
+      "items": {"type": "object"},
+      "description": "queued: per-node issues the server reported alongside a successful queue. Always empty on --where cloud, which rejects such a submit outright."
+    },
+    "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."
+    },
+    "base_url": {"type": ["string", "null"], "description": "queued, --where cloud only: the cloud endpoint the prompt was submitted to."},
     "url": {"type": ["string", "null"]}
-  }
+  },
+  "allOf": [
+    {
+      "if": {"required": ["type"], "properties": {"type": {"const": "queued"}}},
+      "then": {
+        "properties": {
+          "nodes": {"type": ["array", "null"], "items": {"type": "object"}}
+        }
+      }
+    },
+    {
+      "if": {"required": ["type"], "properties": {"type": {"const": "execution_cached"}}},
+      "then": {
+        "properties": {
+          "nodes": {"type": ["array", "null"], "items": {"type": "string"}}
+        }
+      }
+    }
+  ]
 }
diff --git a/docs/json-output.md b/docs/json-output.md
index 5e512ce5..68f7f282 100644
--- a/docs/json-output.md
+++ b/docs/json-output.md
@@ -89,14 +89,99 @@ The stream always ends with exactly one line of `type: "envelope"`:
 | `--no-wait` queued (default) | `[converted]? + prompt_preview + queued + envelope(ok, data.status="queued")` |
 | `--print-prompt`          | `[converted]? + prompt_preview + envelope(ok, data.status="preview")`        |
 | Failure mid-execution     | `[converted]? + prompt_preview + queued + [node events]* + envelope(error)`  |
-| Failure during submission | `[converted]? + prompt_preview + envelope(error)`                            |
-| Failure pre-flight        | `envelope(error)`                                                            |
+| Failure at validation, consent, or submission | `[converted]? + prompt_preview + envelope(error)`         |
+| Failure before the graph is parsed | `[converted]? + envelope(error)`                                    |
 
 Where `[node events]*` is zero or more interleaved `execution_cached`,
 `executing`, `progress`, `executed`, and `output` events. `[X]?` means X
 may or may not appear. An error envelope can replace any non-terminal
 line, ending the stream early.
 
+The last two rows split on **whether the CLI has a parsed graph in hand
+yet**, because `prompt_preview` is emitted as soon as it does — before any
+check that could still refuse the run:
+
+- **Before the graph is parsed** — the workflow file is missing, unreadable,
+  or not JSON; a UI→API conversion failed (`conversion_error`,
+  `conversion_crash`, `cql_no_graph`); the graph is empty
+  (`workflow_empty`) or not in API format (`workflow_not_api_format`); no
+  local server is running (`server_not_running`). These emit a bare error
+  envelope. `converted` can precede it in exactly one case: conversion
+  succeeded but its output still failed API-format classification.
+- **After it is parsed** — the CQL pre-flight (`workflow_unknown_nodes` and
+  friends), the `spend_consent_required` consent gate, cloud authentication
+  (`cloud_unauthorized`), and the submit call itself all run *after*
+  `prompt_preview`. A run refused by any of them therefore emits
+  `prompt_preview` first and *then* the error envelope — the previewed graph
+  is what the CLI *would* have submitted, not a promise that it did.
+
+This ordering is identical on both targets. Note that `prompt_preview`
+carries the full workflow graph, so `--json-stream` output should be treated
+as sensitive if custom nodes embed local paths or credential-like widget
+values in it. Events are emitted in `--json-stream` mode only — neither
+pretty nor plain `--json` mode ever writes a `prompt_preview` line.
+
+These archetypes hold for **both** `--where local` and `--where cloud`, with
+one exception: `comfy run --where cloud` produces no `[node events]*` — use
+`comfy jobs watch --where cloud` for in-flight cloud progress (see
+[Per-target differences](#per-target-differences)). Everything else — the
+`converted` / `prompt_preview` / `queued` prefix, the single terminal
+envelope, and the exit-code mapping — is identical on both targets.
+
+## Per-target differences
+
+`comfy run` has two execution targets — `--where local` (the default: an HTTP
+submit plus a WebSocket session against a ComfyUI server you run) and
+`--where cloud` (an HTTPS submit plus polling against Comfy Cloud). They emit
+the **same event dialect** and the same envelope framing, but the targets are
+not the same machine and a few things genuinely cannot match. The complete
+list of differences:
+
+| Aspect | `--where local` | `--where cloud` |
+| ------ | --------------- | --------------- |
+| Per-node events (`executing`, `execution_cached`, `progress`, `executed`, `output`, `execution_error`) | Emitted, streamed live from the server WebSocket | Not emitted by `comfy run`. The cloud API is polled for a terminal record, so a `--wait` run goes straight from `queued` to the final envelope. For in-flight cloud progress, watch the job instead: `comfy --json-stream jobs watch <prompt_id> --where cloud` emits a coarse `state` event per status transition plus an `output` event per artifact |
+| `queued.validation_warnings` | May be non-empty: the server can accept a prompt (HTTP 200) while reporting per-node issues | Always `[]` — the cloud rejects any submit carrying `node_errors` outright, as a `prompt_rejected` error envelope |
+| `queued.base_url` | Absent | Present — the cloud endpoint the prompt was submitted to |
+| Envelope `where` | `"local"` | `"cloud"` |
+| Envelope target fields | `data.host` (str), `data.port` (int) | `data.base_url` (str) |
+| Envelope `data.cached_node_ids` / `data.executed_node_ids` | Present on `--wait` | Absent — they are derived from the per-node event stream, which the cloud has none of |
+| Envelope `data.warnings` | Absent | Present on `--wait` success: an array of non-fatal warning objects (currently only `partial_execution`, see below). `[]` when there are none |
+| `prompt_rejected` `details` | `status` (400) and `node_errors` | `node_errors` only — the cloud reports rejected nodes on an otherwise-2xx submit, so there is no 4xx status to report. The `node_errors` value has the same [array-of-records shape](#node_errors-shape) on both targets |
+| Error codes | The local set below | The workflow/pre-flight codes and the node-failure codes (`execution_error`, `transient_auth`, `prompt_rejected`, `cancelled`, `spend_consent_required`) plus the cloud-only codes below. The local-server- and WebSocket-specific codes cannot occur: `server_not_running`, `object_info_unavailable`, `connection_error`, `ws_timeout`, `ws_disconnected`, `invalid_response`, `client_error`, `server_error`, `partner_node_requires_credential` (the cloud injects the caller's credential itself) |
+
+Everything not in that table is the same on both targets, including
+`converted`, `prompt_preview`, the `queued` field set, `data.status`,
+`data.prompt_id` / `client_id` / `outputs` / `outputs_by_node` /
+`outputs_by_item` / `state_file` / `watcher_spawned` / `elapsed_seconds`, and
+the `--print-prompt` and `--no-wait` stream shapes.
+
+### Cloud-only error codes
+
+| `code`               | Triggered when                                                     | `details`                       | Exit |
+| -------------------- | ------------------------------------------------------------------ | -------------------------------- | ---- |
+| `cloud_unauthorized` | No usable cloud session, or the session was rejected — run `comfy cloud login` | —                     | 1 |
+| `cloud_http_error`   | The cloud API returned a non-2xx response on submit or while polling | `status` (int), `body` (str) on submit; `status`, `prompt_id` while polling | 1 |
+| `cloud_timeout`      | The cloud job produced no progress for `--timeout` seconds          | `prompt_id` (str)               | 1 |
+| `cql_no_graph`       | A UI-format workflow needs the cloud `object_info` snapshot to be lowered to API format, and it could not be loaded — run `comfy nodes refresh --where cloud` | — | 1 |
+
+All of these are registered in `comfy_cli/error_codes.py` and listed by
+`comfy --json discover`, exactly like the local codes.
+
+### `partial_execution` warning
+
+The cloud prunes workflow branches that fail server-side validation and still
+reports the job as completed. On a `--wait` success the CLI diffs the output
+nodes it submitted against the ones that returned outputs and, when some are
+missing, appends a warning object to `data.warnings` rather than passing the
+run off as a clean success:
+
+```json
+{"code": "partial_execution", "message": "submitted 2 output node(s) but the cloud returned outputs for only 1; 1 branch(es) were pruned server-side (likely failed validation) and produced nothing", "submitted_output_nodes": 2, "returned_output_nodes": 1}
+```
+
+The envelope is still `ok: true` (exit `0`) — the job did run. Agents that
+need all-or-nothing semantics should check `data.warnings` is empty.
+
 ## Event reference
 
 | `type`             | When                                                 |
@@ -152,7 +237,9 @@ final envelope and exits 0 without queuing.
 
 ### `queued`
 
-Emitted after `POST /prompt` returns 200.
+Emitted after the submit request returns success — `POST /prompt` returning
+200 locally, the equivalent cloud submit under `--where cloud` — **and** after
+the job's state file has been persisted.
 
 ```json
 {
@@ -174,6 +261,15 @@ Emitted after `POST /prompt` returns 200.
 | `client_id`           | str           | Client-generated UUID (sent with `/prompt`)            |
 | `validation_warnings` | array of dict | Per-node validation issues ComfyUI reported alongside a successful queue (some output chains validated, others didn't). Same record shape as `prompt_rejected`'s `details.node_errors` (see [shape](#node_errors-shape)). Empty (`[]`) in the common case. |
 | `nodes`               | array of dict | Manifest of every node in the submitted (post-conversion) workflow: `node_id` (str), `class_type` (str), `title` (str). Lets piped consumers render a per-node UI without the workflow file. |
+| `base_url`            | str           | **Cloud only.** The cloud endpoint the prompt was submitted to. Absent on `--where local`. |
+
+`queued` is emitted **after** the submit call returns successfully **and after
+the job's state file has been persisted** — on the async (`--no-wait`) paths,
+after the background watcher spawn attempt too. On both targets it therefore
+means more than "the server has the prompt": the durable record exists, so a
+consumer may run `comfy jobs status <prompt_id>` — or expect the job in
+`comfy jobs ls` — the moment it reads this line. A run whose submit fails emits
+its error envelope with no `queued` line at all.
 
 ### `executing`
 
@@ -293,21 +389,30 @@ On `--wait` success, `data` carries:
 | `prompt_id`         | str             | Server-assigned prompt UUID                            |
 | `client_id`         | str             | Client-generated UUID                                  |
 | `outputs`           | array of str    | URL (or local path) per file-like output, deduplicated |
-| `cached_node_ids`   | array of str    | Node IDs the server reported as cached                 |
-| `executed_node_ids` | array of str    | Node IDs the executor *ran* — the union of every node that appeared in an `executing` or `executed` event, including intermediate compute nodes |
+| `outputs_by_node`   | dict            | The same outputs grouped by the node id that produced them |
+| `outputs_by_item`   | dict            | The same outputs grouped by `compose` foreach item; `{}` when the workflow carried no item map |
+| `cached_node_ids`   | array of str    | **Local only.** Node IDs the server reported as cached  |
+| `executed_node_ids` | array of str    | **Local only.** Node IDs the executor *ran* — the union of every node that appeared in an `executing` or `executed` event, including intermediate compute nodes |
+| `warnings`          | array of dict   | **Cloud only.** Non-fatal warnings about the completed run — see [`partial_execution`](#partial_execution-warning). `[]` when there are none |
 | `elapsed_seconds`   | float \| null   | Wall-clock duration (null when not waiting)            |
-| `host` / `port`     | str / int       | Target server                                          |
+| `host` / `port`     | str / int       | **Local only.** Target server                          |
+| `base_url`          | str             | **Cloud only.** Target cloud endpoint                  |
 | `state_file`        | str \| null     | Path of the job state file (poll with `comfy jobs status`) |
 
 `cached_node_ids` and `executed_node_ids` may overlap: a cached
 output-bearing node emits both `execution_cached` and `executed`. Agents
 wanting "ran fresh, not from cache" should compute
-`set(executed_node_ids) - set(cached_node_ids)`.
+`set(executed_node_ids) - set(cached_node_ids)`. Both are derived from the
+per-node event stream and so are local-only — see
+[Per-target differences](#per-target-differences).
 
 Without `--wait` (the default), the stream ends at the `queued` envelope
 (`data.status: "queued"`, `data.watcher_spawned: bool`) and a detached
 watcher keeps the state file updated; follow up with
-`comfy jobs watch <prompt_id>` or `comfy jobs status <prompt_id>`.
+`comfy jobs watch <prompt_id>` or `comfy jobs status <prompt_id>`. This is
+the same on both targets, `watcher_spawned` included (add `--where cloud` to
+the follow-up commands for a cloud job). The async envelope carries no
+`outputs_by_node` / `outputs_by_item` / `warnings` — nothing has run yet.
 
 ## `comfy validate --json` envelope
 
@@ -357,7 +462,9 @@ Every failure envelope carries:
 
 Codes raised by `comfy run` against a local server, with their `details`
 payloads. All of them are registered in `comfy_cli/error_codes.py` (the
-registry test enforces this) and surfaced by `comfy discover`.
+registry test enforces this) and surfaced by `comfy discover`. For
+`--where cloud`, see [Cloud-only error codes](#cloud-only-error-codes) — it
+lists what the cloud adds and which of the codes below cannot occur there.
 
 | `code`                    | Triggered when                                                                  | `details`                                          | Exit |
 | ------------------------- | ------------------------------------------------------------------------------- | -------------------------------------------------- | ---- |
@@ -382,6 +489,7 @@ registry test enforces this) and surfaced by `comfy discover`.
 | `ws_disconnected`         | WebSocket connection dropped mid-execution                                      | —                                                  | 1 |
 | `cancelled`               | Run was interrupted — client `SIGINT` (Ctrl-C) or the server's `execution_interrupted` (e.g. `/interrupt`) | —                       | 130 |
 | `execution_error`         | A node raised during execution (server emitted `execution_error`)               | `node_id` (str), `class_type` (str), `title` (str), `exception_type` (str), `traceback` (str) | 1 |
+| `transient_auth`          | The `execution_error` cause was an API node's server-side session token expiring mid-execution — transient, so resubmitting the same workflow succeeds. Local credentials are fine; `comfy cloud login` does not help | Same fields as `execution_error` | 1 |
 
 ### `exception_type` field
 
@@ -431,6 +539,14 @@ and may evolve with ComfyUI versions — agents should ignore unknown
 fields. The CLI guarantees only that the outer value is an array of
 dicts, each carrying a `node_id` (str).
 
+If a server reports a bare value instead of the per-node dict above (e.g.
+`{"1": "missing input"}`), the CLI does not drop the record — it wraps the
+value as `{"node_id": "1", "errors": ["missing input"]}` so the count in the
+message always matches the array and the diagnostic survives. `errors` items
+are objects in the normal case; treat a non-object item as an opaque message.
+`node_id` is always taken from the payload's own map key, so a server-supplied
+`node_id` field inside a record cannot override it.
+
 ## Output object
 
 Entries of `executed.outputs`:
@@ -476,6 +592,19 @@ Stderr may contain a Python traceback in these cases.
 
 Exit code: `0`.
 
+### Successful cloud run (`--where cloud --wait`)
+
+Same prefix as the local stream; no per-node events, because the cloud path
+polls for a terminal record instead of streaming a WebSocket session.
+
+```json
+{"schema":"event/1","type":"prompt_preview","prompt":{"1":{"class_type":"GeminiNanoBanana2","inputs":{"prompt":"a banana"}},"2":{"class_type":"SaveImage","inputs":{"filename_prefix":"banana_test","images":["1",0]}}}}
+{"schema":"event/1","type":"queued","prompt_id":"9b1c…","client_id":"fe2a…","validation_warnings":[],"nodes":[{"node_id":"1","class_type":"GeminiNanoBanana2","title":"GeminiNanoBanana2"},{"node_id":"2","class_type":"SaveImage","title":"SaveImage"}],"base_url":"https://api.comfy.org"}
+{"schema":"envelope/1","type":"envelope","ok":true,"command":"run","version":"1.6.1","where":"cloud","data":{"workflow":"/path/wf.json","status":"completed","prompt_id":"9b1c…","client_id":"fe2a…","outputs":["https://…/banana_test_00001_.png"],"outputs_by_node":{"2":["https://…/banana_test_00001_.png"]},"outputs_by_item":{},"warnings":[],"elapsed_seconds":21.7,"base_url":"https://api.comfy.org","state_file":"…"},"error":null}
+```
+
+Exit code: `0`.
+
 ### Failure: workflow file missing
 
 ```json
diff --git a/tests/comfy_cli/command/test_run_cli.py b/tests/comfy_cli/command/test_run_cli.py
index af36c86c..f5fb9197 100644
--- a/tests/comfy_cli/command/test_run_cli.py
+++ b/tests/comfy_cli/command/test_run_cli.py
@@ -236,9 +236,11 @@ def test_execute_keeps_workflows_with_no_cleanup(self, tmp_path, monkeypatch):
 
 
 def test_cloud_print_prompt_does_not_submit(monkeypatch, tmp_path):
-    """--print-prompt on the cloud route prints the graph and never submits."""
-    import typer
+    """--print-prompt on the cloud route prints the graph and never submits.
 
+    It also *returns* rather than raising `typer.Exit(0)` — same termination as
+    the local path's dry run (both exit 0 through the CLI either way).
+    """
     import comfy_cli.comfy_client as cc
     from comfy_cli.command.run import execute_cloud
 
@@ -257,9 +259,7 @@ def submit_prompt(self, *a, **k):
     # should return BEFORE Client is even constructed.
     monkeypatch.setattr(cc, "Client", FakeClient)
 
-    with pytest.raises(typer.Exit) as exc:
-        execute_cloud(str(wf), wait=True, print_prompt=True, timeout=5)
-    assert exc.value.exit_code == 0
+    assert execute_cloud(str(wf), wait=True, print_prompt=True, timeout=5) is None
 
 
 if __name__ == "__main__":
diff --git a/tests/comfy_cli/command/test_run_json.py b/tests/comfy_cli/command/test_run_json.py
index 65af669e..89ae833f 100644
--- a/tests/comfy_cli/command/test_run_json.py
+++ b/tests/comfy_cli/command/test_run_json.py
@@ -22,6 +22,7 @@
 import io
 import json
 import os
+import sys
 import tempfile
 import urllib.error
 from unittest.mock import MagicMock, patch
@@ -30,6 +31,7 @@
 import typer
 from websocket import WebSocketException, WebSocketTimeoutException
 
+import comfy_cli
 from comfy_cli.command.run import (
     WorkflowExecution,
     _classify_api_workflow,
@@ -1684,3 +1686,590 @@ def test_cloud_route_load_failure_emits_envelope(tmp_path, capsys):
     lines = [ln for ln in out.splitlines() if ln.strip()]
     env = json.loads(lines[-1])
     assert env["ok"] is False and env["error"]["code"] == "workflow_not_found"
+
+
+# --------------------------------------------------------------------------
+# Cloud (`--where cloud`) NDJSON contract
+#
+# The cloud pipeline is a separate ~400-line function from the local one, and
+# its event stream had drifted from docs/json-output.md: no `converted`, a
+# `prompt_preview` only under `--print-prompt`, and a `queued` emitted BEFORE
+# the submit carrying `{workflow, base_url}` instead of the documented
+# `{prompt_id, client_id, validation_warnings, nodes}`. These tests pin the
+# conformed stream so the two targets can't silently drift apart again.
+# --------------------------------------------------------------------------
+
+
+class _FakeTarget:
+    base_url = "https://api.comfy.org"
+    is_cloud = True
+    auth_token = "tok"
+    api_key = None
+
+
+def _install_cloud_stubs(monkeypatch, *, client_cls, object_info=None, watcher_spawned=True):
+    """Patch out everything execute_cloud reaches for besides the renderer."""
+    import comfy_cli.comfy_client as cc
+    import comfy_cli.cql.engine as cql_engine
+    import comfy_cli.jobs_state as jobs_state_mod
+    import comfy_cli.target as target_mod
+    from comfy_cli.command import run as run_pkg
+
+    monkeypatch.setattr(cc, "Client", client_cls)
+    monkeypatch.setattr(target_mod, "resolve_target", lambda **_kw: _FakeTarget())
+    # Empty object_info keeps CQL preflight + partner detection fail-open, so
+    # these tests exercise the event contract and nothing else.
+    monkeypatch.setattr(cql_engine, "_load_from_target", lambda **_kw: object_info or {})
+    monkeypatch.setattr(jobs_state_mod, "write", lambda state: "/tmp/state.json")
+    monkeypatch.setattr(run_pkg, "_spawn_watcher", lambda *a, **k: watcher_spawned)
+
+
+def _cloud_capture(capsys, workflow_path, **kwargs):
+    """Run execute_cloud() and return (parsed stdout lines, exit_code)."""
+    from comfy_cli.command.run import execute_cloud
+
+    exit_code = 0
+    try:
+        execute_cloud(workflow_path, **kwargs)
+    except typer.Exit as e:
+        exit_code = e.exit_code or 0
+    out, _err = capsys.readouterr()
+    return _parse_lines(out), exit_code
+
+
+class _FakeSubmit:
+    def __init__(self, prompt_id="cloud-pid", node_errors=None):
+        self.prompt_id = prompt_id
+        self.number = 1
+        self.node_errors = node_errors or {}
+
+
+def _fake_client(*, submit=None, submit_exc=None, record=None):
+    class FakeClient:
+        submitted = []
+
+        def __init__(self, *a, **k):
+            pass
+
+        def submit_prompt(self, workflow, client_id, **k):
+            FakeClient.submitted.append((workflow, client_id))
+            if submit_exc is not None:
+                raise submit_exc
+            return submit if submit is not None else _FakeSubmit()
+
+        def get_job_status(self, prompt_id):
+            return {"status": "running"}
+
+        def wait_for_completion(self, prompt_id, **k):
+            return record if record is not None else {"status": {"status_str": "success"}, "outputs": {}}
+
+        def extract_outputs(self, rec):
+            return []
+
+    FakeClient.submitted = []
+    return FakeClient
+
+
+class TestCloudStreamContract:
+    def test_api_workflow_async_stream_matches_documented_archetype(
+        self, monkeypatch, workflow_file, simple_workflow, capsys
+    ):
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+
+        assert exit_code == 0
+        assert [ln["type"] for ln in lines] == ["prompt_preview", "queued", "envelope"]
+        assert lines[0]["prompt"] == simple_workflow
+
+        queued = lines[1]
+        assert queued["schema"] == "event/1"
+        assert queued["prompt_id"] == "cloud-pid"
+        assert isinstance(queued["client_id"], str) and queued["client_id"]
+        assert queued["validation_warnings"] == []
+        assert queued["nodes"] == [
+            {"node_id": "1", "class_type": "EmptyLatentImage", "title": "Latent"},
+            {"node_id": "2", "class_type": "SaveImage", "title": "Save"},
+        ]
+        assert queued["base_url"] == "https://api.comfy.org"
+
+        env = _envelope(lines)
+        assert env["ok"] is True and env["where"] == "cloud"
+        assert env["data"]["status"] == "queued"
+        # Parity with the local async envelope (docs/json-output.md).
+        assert env["data"]["watcher_spawned"] is True
+
+    def test_async_envelope_reports_a_watcher_that_did_not_start(self, monkeypatch, workflow_file, capsys):
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client(), watcher_spawned=False)
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+        assert exit_code == 0
+        assert _envelope(lines)["data"]["watcher_spawned"] is False
+
+    def test_ui_workflow_emits_converted_before_prompt_preview(self, monkeypatch, tmp_path, capsys):
+        from comfy_cli.command import run as run_pkg
+
+        ui = tmp_path / "ui.json"
+        ui.write_text(json.dumps({"nodes": [{"id": 1, "type": "SaveImage"}], "links": []}))
+        converted = {"1": {"class_type": "SaveImage", "inputs": {}}}
+        monkeypatch.setattr(run_pkg, "convert_ui_to_api", lambda *_a, **_k: converted)
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+
+        lines, exit_code = _cloud_capture(capsys, str(ui), wait=False, timeout=5)
+
+        assert exit_code == 0
+        assert [ln["type"] for ln in lines] == ["converted", "prompt_preview", "queued", "envelope"]
+        assert lines[0]["node_count"] == 1
+        assert lines[1]["prompt"] == converted
+
+    def test_print_prompt_stream_is_preview_only(self, monkeypatch, workflow_file, simple_workflow, capsys):
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=True, print_prompt=True, timeout=5)
+
+        assert exit_code == 0
+        assert [ln["type"] for ln in lines] == ["prompt_preview", "envelope"]
+        # Emitted exactly once — the unconditional emission must not double up
+        # with the dry-run branch.
+        assert lines[0]["prompt"] == simple_workflow
+        env = _envelope(lines)
+        assert env["ok"] is True and env["data"]["status"] == "preview"
+
+    def test_no_queued_when_the_submit_fails(self, monkeypatch, workflow_file, capsys):
+        """`queued` means the server accepted the prompt — so a failed submit
+        must not produce one (it did before: the event preceded the POST)."""
+        from comfy_cli.comfy_client import HTTPError
+
+        _install_cloud_stubs(
+            monkeypatch,
+            client_cls=_fake_client(submit_exc=HTTPError(500, "boom", "boom")),
+        )
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+
+        assert exit_code == 1
+        assert [ln["type"] for ln in lines] == ["prompt_preview", "envelope"]
+        assert _envelope(lines)["error"]["code"] == "cloud_http_error"
+
+    def test_no_queued_when_the_server_rejects_nodes(self, monkeypatch, workflow_file, capsys):
+        rejected = _FakeSubmit(node_errors={"1": {"class_type": "X", "errors": [{"message": "bad"}]}})
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client(submit=rejected))
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+
+        assert exit_code == 1
+        assert "queued" not in [ln["type"] for ln in lines]
+        err = _envelope(lines)["error"]
+        assert err["code"] == "prompt_rejected"
+        # Documented node_errors shape: an array of self-contained records
+        # carrying `node_id`, not the server's id-keyed dict.
+        assert err["details"]["node_errors"] == [{"node_id": "1", "class_type": "X", "errors": [{"message": "bad"}]}]
+
+    def test_wait_stream_emits_queued_not_a_bogus_executing_event(self, monkeypatch, workflow_file, capsys):
+        """`--wait` used to announce the submit as `executing` — a per-node
+        event type — with `{workflow, base_url}` and no `node` field."""
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=True, timeout=5)
+
+        assert exit_code == 0
+        assert [ln["type"] for ln in lines] == ["prompt_preview", "queued", "envelope"]
+        assert _envelope(lines)["data"]["status"] == "completed"
+
+    def test_queued_node_manifest_matches_the_local_path(self, simple_workflow):
+        """Both targets build `queued.nodes` from the same helper, so a
+        divergence in the manifest shape can't reappear on one side only."""
+        from comfy_cli.command.run import workflow_manifest
+
+        ex = WorkflowExecution(
+            workflow=simple_workflow,
+            host="127.0.0.1",
+            port=8188,
+            verbose=False,
+            progress=None,
+            local_paths=None,
+            timeout=5,
+        )
+        assert ex.workflow_manifest() == workflow_manifest(simple_workflow)
+
+    @pytest.mark.parametrize("wait", [False, True])
+    def test_cloud_stream_validates_against_the_published_schemas(self, monkeypatch, workflow_file, capsys, wait):
+        """`comfy --json discover` publishes run_event.json / run.json as the
+        machine-readable half of this contract. A consumer validating against
+        them must not reject a real cloud stream — which is exactly how the
+        undocumented pre-submit `queued` and the missing `prompt_preview`
+        would have surfaced."""
+        import jsonschema
+
+        schemas = os.path.join(os.path.dirname(comfy_cli.__file__), "schemas")
+        with open(os.path.join(schemas, "run_event.json")) as f:
+            event_schema = json.load(f)
+        with open(os.path.join(schemas, "run.json")) as f:
+            data_schema = json.load(f)
+
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=wait, timeout=5)
+
+        assert exit_code == 0
+        for event in _events(lines):
+            jsonschema.Draft202012Validator(event_schema).validate(event)
+        jsonschema.Draft202012Validator(data_schema).validate(_envelope(lines)["data"])
+
+    def test_local_stream_validates_against_the_published_schemas(self, monkeypatch, workflow_file, capsys):
+        """Same gate for the local async stream — `converted` /
+        `prompt_preview` were absent from the published event enum too, so a
+        strict consumer rejected valid output on BOTH targets."""
+        import jsonschema
+
+        schemas = os.path.join(os.path.dirname(comfy_cli.__file__), "schemas")
+        with open(os.path.join(schemas, "run_event.json")) as f:
+            event_schema = json.load(f)
+        with open(os.path.join(schemas, "run.json")) as f:
+            data_schema = json.load(f)
+
+        from comfy_cli.command import run as run_pkg
+
+        monkeypatch.setattr(run_pkg, "_fetch_object_info", lambda *a, **k: {})
+        monkeypatch.setattr(run_pkg, "_spawn_watcher", lambda *a, **k: True)
+        monkeypatch.setattr(run_pkg.jobs_state, "write", lambda state: "/tmp/state.json")
+
+        with (
+            patch("comfy_cli.command.run.check_comfy_server_running", return_value=True),
+            patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open,
+        ):
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps(
+                {"prompt_id": "local-pid"}
+            ).encode()
+            lines, exit_code = _run_execute_capture(workflow_file, capsys, wait=False)
+
+        assert exit_code == 0
+        # A real submit, so the local `queued` event is in this stream.
+        assert [ln["type"] for ln in lines] == ["prompt_preview", "queued", "envelope"]
+        for event in _events(lines):
+            jsonschema.Draft202012Validator(event_schema).validate(event)
+        jsonschema.Draft202012Validator(data_schema).validate(_envelope(lines)["data"])
+
+
+class TestNodeErrorsRecordShape:
+    """`_node_errors_to_list` is the single transform behind both
+    `prompt_rejected.details.node_errors` and `queued.validation_warnings`, and
+    since this PR it also carries less-trusted cloud payloads."""
+
+    def test_non_dict_records_survive_instead_of_vanishing(self):
+        """A server reporting a bare value (`{"1": "missing input"}`) must not
+        yield an empty array under a "rejected 1 node(s)" message — that leaves
+        the user with a hint pointing at nothing."""
+        from comfy_cli.command.run.loader import _node_errors_to_list
+
+        assert _node_errors_to_list({"1": "missing input"}) == [{"node_id": "1", "errors": ["missing input"]}]
+        assert _node_errors_to_list({"2": ["a", "b"]}) == [{"node_id": "2", "errors": ["a", "b"]}]
+        # The count in the message always matches the array length.
+        payload = {"1": "bare", "2": {"errors": [{"message": "structured"}]}}
+        assert len(_node_errors_to_list(payload)) == len(payload)
+
+    def test_map_key_wins_over_a_server_supplied_node_id(self):
+        """A record carrying its own `node_id` must not shadow the authoritative
+        map key — agents correlate this id against the per-node events."""
+        from comfy_cli.command.run.loader import _node_errors_to_list
+
+        out = _node_errors_to_list({"7": {"node_id": "999", "errors": [{"message": "x"}]}})
+        assert out == [{"node_id": "7", "errors": [{"message": "x"}]}]
+
+    def test_well_formed_records_are_unchanged(self):
+        from comfy_cli.command.run.loader import _node_errors_to_list
+
+        assert _node_errors_to_list({"1": {"class_type": "X", "errors": [{"message": "bad"}]}}) == [
+            {"node_id": "1", "class_type": "X", "errors": [{"message": "bad"}]}
+        ]
+        assert _node_errors_to_list({}) == []
+        assert _node_errors_to_list(None) == []
+
+
+class TestEventSchemaDiscriminatesNodesShape:
+    """`nodes` serves two events with different item shapes. Typing it as
+    `["object", "string"]` would let a `queued` that regressed to bare node-id
+    strings validate — precisely the drift these schema tests exist to catch."""
+
+    @staticmethod
+    def _validator():
+        import jsonschema
+
+        schemas = os.path.join(os.path.dirname(comfy_cli.__file__), "schemas")
+        with open(os.path.join(schemas, "run_event.json")) as f:
+            return jsonschema.Draft202012Validator(json.load(f))
+
+    def test_queued_requires_object_node_records(self):
+        import jsonschema
+
+        good = {
+            "schema": "event/1",
+            "type": "queued",
+            "nodes": [{"node_id": "1", "class_type": "X", "title": "X"}],
+        }
+        self._validator().validate(good)
+
+        regressed = {"schema": "event/1", "type": "queued", "nodes": ["1", "2"]}
+        with pytest.raises(jsonschema.ValidationError):
+            self._validator().validate(regressed)
+
+    def test_execution_cached_still_accepts_bare_node_ids(self):
+        """`comfy jobs watch --where cloud` emits plain id strings here."""
+        self._validator().validate({"schema": "event/1", "type": "execution_cached", "nodes": ["1", "2"]})
+
+
+class TestMalformedRejectionPayloadStillYieldsAnEnvelope:
+    """Every field of the cloud's `node_errors` is server-supplied and only
+    documented by convention. A shape the CLI did not expect must still produce
+    exactly one terminal envelope — a traceback would emit none at all."""
+
+    def test_bare_string_record_keeps_the_diagnostic(self, monkeypatch, workflow_file, capsys):
+        rejected = _FakeSubmit(node_errors={"1": "missing input"})
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client(submit=rejected))
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+
+        assert exit_code == 1
+        err = _envelope(lines)["error"]
+        assert err["code"] == "prompt_rejected"
+        # Not an empty array under a "rejected 1 node(s)" message.
+        assert err["details"]["node_errors"] == [{"node_id": "1", "errors": ["missing input"]}]
+        assert "missing input" in err["hint"]
+
+    def test_non_dict_error_items_do_not_crash_the_hint_builder(self, monkeypatch, workflow_file, capsys):
+        rejected = _FakeSubmit(node_errors={"1": {"class_type": "X", "errors": ["bad"]}})
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client(submit=rejected))
+        lines, exit_code = _cloud_capture(capsys, workflow_file, wait=False, timeout=5)
+
+        assert exit_code == 1
+        assert _envelope(lines)["error"]["code"] == "prompt_rejected"
+        assert "node 1 (X): bad" in _envelope(lines)["error"]["hint"]
+
+
+# --------------------------------------------------------------------------
+# `queued` is emitted only after the job's durable state exists (BE-6072)
+#
+# The event used to precede `jobs_state.write` on every path. That made
+# state-file-backed follow-ups (`jobs ls`, `jobs wait`, compose's item_map)
+# raceable against a just-queued job, and — worse — let a `BrokenPipeError` at
+# the emit (`comfy run … --json-stream | head -n1`) abort setup before the
+# record was ever written, while `__main__`'s broken-pipe handling still exited
+# 0: a billable cloud job with no journal entry, no state file and no watcher.
+# These tests pin the ordering and the propagation on all four paths.
+# --------------------------------------------------------------------------
+
+
+class _TapStream:
+    """Stand-in for the renderer's machine stream.
+
+    `on_line` sees every NDJSON line before it is written and may raise to
+    simulate a closed stdout. Writes delegate to ``sys.stdout`` at call time so
+    `capsys` still captures the stream.
+    """
+
+    def __init__(self, on_line):
+        self.on_line = on_line
+
+    def write(self, s: str) -> int:
+        self.on_line(s)
+        return sys.stdout.write(s)
+
+    def flush(self) -> None:
+        sys.stdout.flush()
+
+
+def _tap_queued(renderer, callback):
+    """Invoke `callback` just before each `queued` line reaches the stream."""
+
+    def on_line(s: str) -> None:
+        try:
+            payload = json.loads(s)
+        except (json.JSONDecodeError, ValueError):
+            return
+        if isinstance(payload, dict) and payload.get("type") == "queued":
+            callback()
+
+    renderer.machine_stream = _TapStream(on_line)
+
+
+def _raise_broken_pipe() -> None:
+    """What writing an NDJSON line to a closed stdout (`… | head -n1`) does."""
+    raise BrokenPipeError(32, "Broken pipe")
+
+
+def _recording_write(writes: list):
+    """A ``jobs_state.write`` stand-in that records each persisted prompt_id."""
+
+    def fake_write(state):
+        writes.append(state.prompt_id)
+        return "/tmp/s.json"
+
+    return fake_write
+
+
+def _recording_spawn(spawned: list):
+    """A ``_spawn_watcher`` stand-in that records each spawn attempt."""
+
+    def fake_spawn(prompt_id, **_kw):
+        spawned.append(prompt_id)
+        return True
+
+    return fake_spawn
+
+
+def _log_state_writes(monkeypatch, log):
+    """Record every ``jobs_state.write`` (which ``_write_state`` delegates to)
+    in `log`, keeping the real return contract (a path)."""
+    from comfy_cli import jobs_state as jobs_state_mod
+
+    def fake_write(state):
+        log.append(("state_write",))
+        return "/tmp/state.json"
+
+    monkeypatch.setattr(jobs_state_mod, "write", fake_write)
+
+
+def _local_ws_that_finishes(MockWs):
+    ws_instance = MagicMock()
+    MockWs.return_value = ws_instance
+    ws_instance.recv.side_effect = [
+        json.dumps({"type": "executing", "data": {"prompt_id": "p", "node": None}}),
+    ]
+    return ws_instance
+
+
+class TestQueuedFollowsStatePersistence:
+    """Ordering spies: `state_write` must precede `queued_emit` on all four
+    target × wait-mode combinations."""
+
+    def test_local_async(self, monkeypatch, workflow_file, capsys, ndjson_renderer):
+        log: list[tuple] = []
+        _log_state_writes(monkeypatch, log)
+        _tap_queued(ndjson_renderer, lambda: log.append(("queued_emit",)))
+        with (
+            patch("comfy_cli.command.run.check_comfy_server_running", return_value=True),
+            patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open,
+            patch("comfy_cli.command.run._spawn_watcher", return_value=True),
+        ):
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps({"prompt_id": "p"}).encode()
+            _lines, exit_code = _run_execute_capture(workflow_file, capsys, wait=False)
+        assert exit_code == 0
+        assert ("queued_emit",) in log
+        assert log.index(("state_write",)) < log.index(("queued_emit",))
+
+    def test_local_wait(self, monkeypatch, workflow_file, capsys, ndjson_renderer):
+        log: list[tuple] = []
+        _log_state_writes(monkeypatch, log)
+        _tap_queued(ndjson_renderer, lambda: log.append(("queued_emit",)))
+        with (
+            patch("comfy_cli.command.run.check_comfy_server_running", return_value=True),
+            patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open,
+            patch("comfy_cli.command.run.WebSocket") as MockWs,
+        ):
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps({"prompt_id": "p"}).encode()
+            _local_ws_that_finishes(MockWs)
+            _lines, exit_code = _run_execute_capture(workflow_file, capsys, wait=True)
+        assert exit_code == 0
+        assert ("queued_emit",) in log
+        assert log.index(("state_write",)) < log.index(("queued_emit",))
+
+    @pytest.mark.parametrize("wait", [False, True])
+    def test_cloud(self, monkeypatch, workflow_file, capsys, ndjson_renderer, wait):
+        log: list[tuple] = []
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        # After the stubs, so this write spy wins over their no-op one.
+        _log_state_writes(monkeypatch, log)
+        _tap_queued(ndjson_renderer, lambda: log.append(("queued_emit",)))
+        _lines, exit_code = _cloud_capture(capsys, workflow_file, wait=wait, timeout=5)
+        assert exit_code == 0
+        assert ("queued_emit",) in log
+        assert log.index(("state_write",)) < log.index(("queued_emit",))
+
+
+class TestBrokenPipeAtTheQueuedEmit:
+    """A closed stdout at the `queued` line is the consumer leaving. It must
+    propagate as itself — `__main__` turns that into the documented silent exit
+    0 — and only ever after the durable record (and the watcher) exist."""
+
+    def test_local_async_persists_before_it_raises(self, monkeypatch, workflow_file, capsys, ndjson_renderer):
+        writes: list[str] = []
+        spawned: list[str] = []
+        from comfy_cli import jobs_state as jobs_state_mod
+
+        monkeypatch.setattr(jobs_state_mod, "write", _recording_write(writes))
+        _tap_queued(ndjson_renderer, _raise_broken_pipe)
+
+        with (
+            patch("comfy_cli.command.run.check_comfy_server_running", return_value=True),
+            patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open,
+            patch("comfy_cli.command.run._spawn_watcher", side_effect=_recording_spawn(spawned)),
+        ):
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps({"prompt_id": "p"}).encode()
+            with pytest.raises(BrokenPipeError):
+                execute(workflow_file, host="127.0.0.1", port=8188, wait=False, verbose=False, timeout=30)
+        capsys.readouterr()
+        assert writes == ["p"]
+        assert spawned == ["p"]
+
+    def test_cloud_async_persists_before_it_raises(self, monkeypatch, workflow_file, capsys, ndjson_renderer):
+        from comfy_cli.command.run import execute_cloud
+
+        writes: list[str] = []
+        spawned: list[str] = []
+        _install_cloud_stubs(monkeypatch, client_cls=_fake_client())
+        from comfy_cli import jobs_state as jobs_state_mod
+        from comfy_cli.command import run as run_pkg
+
+        monkeypatch.setattr(jobs_state_mod, "write", _recording_write(writes))
+        monkeypatch.setattr(run_pkg, "_spawn_watcher", _recording_spawn(spawned))
+        _tap_queued(ndjson_renderer, _raise_broken_pipe)
+
+        with pytest.raises(BrokenPipeError):
+            execute_cloud(workflow_file, wait=False, timeout=5)
+        capsys.readouterr()
+        assert writes == ["cloud-pid"]
+        assert spawned == ["cloud-pid"]
+
+    def test_local_wait_leaves_the_running_record_intact(self, monkeypatch, workflow_file, capsys, ndjson_renderer):
+        """The relocated emit sits inside the `except (WebSocketException,
+        ConnectionError, OSError)` block's try. Without the handler's
+        BrokenPipeError guard a closed stdout would be misreported as
+        `ws_disconnected` *and* rewrite this healthy record to error/server_died."""
+        from comfy_cli import jobs_state as jobs_state_mod
+
+        _tap_queued(ndjson_renderer, _raise_broken_pipe)
+        with (
+            patch("comfy_cli.command.run.check_comfy_server_running", return_value=True),
+            patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open,
+            patch("comfy_cli.command.run.WebSocket") as MockWs,
+        ):
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps({"prompt_id": "p"}).encode()
+            _local_ws_that_finishes(MockWs)
+            with pytest.raises(BrokenPipeError):
+                execute(workflow_file, host="127.0.0.1", port=8188, wait=True, verbose=False, timeout=30)
+        capsys.readouterr()
+        record = jobs_state_mod.read("p")
+        assert record is not None, "the submit-time state file must exist"
+        assert record.status == "running"
+        assert record.error is None
+
+
+class TestQueueStashesWarningsWithoutEmitting:
+    """`WorkflowExecution.queue()` no longer emits — it stashes the warnings so
+    the caller can emit the identical event after persisting state."""
+
+    def test_queue_emits_nothing_and_stashes_the_warnings(self, simple_workflow, capsys):
+        ex = _make_workflow_execution(simple_workflow)
+        body = json.dumps(
+            {
+                "prompt_id": "p",
+                "node_errors": {"3": {"errors": [{"type": "x", "message": "skipped"}], "class_type": "X"}},
+            }
+        ).encode()
+        with patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open:
+            mock_open.return_value.__enter__.return_value.read.return_value = body
+            ex.queue()
+        out, _err = capsys.readouterr()
+        assert "queued" not in out, f"queue() must not emit the event itself: {out!r}"
+        assert ex.prompt_id == "p"
+        assert ex.validation_warnings[0]["node_id"] == "3"
+
+    def test_no_node_errors_leaves_the_warnings_empty(self, simple_workflow, capsys):
+        ex = _make_workflow_execution(simple_workflow)
+        with patch("comfy_cli.http._AUTHED_OPENER.open") as mock_open:
+            mock_open.return_value.__enter__.return_value.read.return_value = json.dumps({"prompt_id": "p"}).encode()
+            ex.queue()
+        capsys.readouterr()
+        assert ex.validation_warnings == []
diff --git a/tests/comfy_cli/test_run_execution_lifecycle.py b/tests/comfy_cli/test_run_execution_lifecycle.py
index fb5779cf..d93324c9 100644
--- a/tests/comfy_cli/test_run_execution_lifecycle.py
+++ b/tests/comfy_cli/test_run_execution_lifecycle.py
@@ -113,6 +113,70 @@ def test_api_key_is_redacted_in_lifecycle_properties(self, runner, tracked_run,
             assert "sk-supersecret" not in str(props)
 
 
+class TestRunCloudTarget:
+    """``--where cloud`` must be analytics-equivalent to the local target.
+
+    The cloud branch used to `return` straight out of the `try` suite, and
+    Python skips a try's ``else:`` clause on `return` — so ``execution_success``
+    never fired for a cloud run (only `--print-prompt` got it by accident, via
+    the ``except typer.Exit`` handler, back when it raised ``Exit(0)``).
+    """
+
+    def test_cloud_submit_emits_execution_start_then_success(self, runner, tracked_run, monkeypatch):
+        from comfy_cli.cmdline import app
+
+        monkeypatch.setattr("comfy_cli.where.cloud_preflight_or_exit", lambda *a, **kw: None)
+        with patch("comfy_cli.cmdline.run_inner.execute_cloud") as mock_cloud:
+            mock_cloud.return_value = None
+            result = runner.invoke(app, ["run", "--workflow", "wf.json", "--where", "cloud"])
+
+        assert result.exit_code == 0, f"stdout={result.output!r} exc={result.exception!r}"
+        assert mock_cloud.called
+        assert _event_names(tracked_run) == ["execution_start", "execution_success"]
+
+    def test_cloud_print_prompt_emits_execution_success(self, runner, tracked_run, monkeypatch):
+        # execute_cloud()'s --print-prompt branch returns rather than raising
+        # Exit(0); the success event must still land.
+        from comfy_cli.cmdline import app
+
+        monkeypatch.setattr("comfy_cli.where.cloud_preflight_or_exit", lambda *a, **kw: None)
+        with patch("comfy_cli.cmdline.run_inner.execute_cloud") as mock_cloud:
+            mock_cloud.return_value = None
+            result = runner.invoke(app, ["run", "--workflow", "wf.json", "--where", "cloud", "--print-prompt"])
+
+        assert result.exit_code == 0, f"stdout={result.output!r} exc={result.exception!r}"
+        assert mock_cloud.call_args.kwargs["print_prompt"] is True
+        assert _event_names(tracked_run) == ["execution_start", "execution_success"]
+
+    def test_cloud_failure_still_emits_execution_error_only(self, runner, tracked_run, monkeypatch):
+        from comfy_cli.cmdline import app
+
+        monkeypatch.setattr("comfy_cli.where.cloud_preflight_or_exit", lambda *a, **kw: None)
+        with patch("comfy_cli.cmdline.run_inner.execute_cloud") as mock_cloud:
+            mock_cloud.side_effect = typer.Exit(code=1)
+            result = runner.invoke(app, ["run", "--workflow", "wf.json", "--where", "cloud"])
+
+        assert result.exit_code == 1
+        names = _event_names(tracked_run)
+        assert "execution_error" in names
+        assert "execution_success" not in names
+
+    def test_cloud_target_does_not_resolve_host_port(self, runner, tracked_run, monkeypatch):
+        # host/port aren't applicable to the HTTPS+Bearer cloud path; the
+        # if/else must keep that resolution off the cloud branch.
+        from comfy_cli.cmdline import app
+
+        monkeypatch.setattr("comfy_cli.where.cloud_preflight_or_exit", lambda *a, **kw: None)
+        with (
+            patch("comfy_cli.cmdline.run_inner.execute_cloud"),
+            patch("comfy_cli.host_port.resolve_host_port") as mock_resolve,
+        ):
+            result = runner.invoke(app, ["run", "--workflow", "wf.json", "--where", "cloud"])
+
+        assert result.exit_code == 0, f"stdout={result.output!r} exc={result.exception!r}"
+        assert not mock_resolve.called
+
+
 class TestRunFailurePath:
     def test_typer_exit_1_emits_execution_error_with_exit_code(self, runner, tracked_run):
         from comfy_cli.cmdline import app