feat(workflow): agent-first workflow editing — CRDT edit primitives, recipes, offline catalog, cloud run association

comfy_cli/cmdline.py

@@ -5,7 +5,6 @@
 import webbrowser
 from typing import Annotated
 
-import questionary
 import typer
 from rich.console import Console
 
@@ -787,11 +786,35 @@ def run(
             ),
         ),
     ] = False,
+    workflow_id: Annotated[
+        str | None,
+        typer.Option(
+            "--workflow-id",
+            show_default=False,
+            help="Cloud workflow entity id to associate this run with (enables draft auto-save on run).",
+        ),
+    ] = None,
+    no_watch: Annotated[
+        bool,
+        typer.Option(
+            "--no-watch",
+            show_default=False,
+            help=(
+                "Suppress the detached background watcher subprocess for non-blocking "
+                "runs (equivalent to setting COMFY_NO_WATCH=1). Agentic callers with "
+                "their own job-wait loop don't need a second process polling in the "
+                "background; it just holds onto credentials after the parent exits."
+            ),
+        ),
+    ] = False,
 ):
     # Snapshot kwargs before the body mutates api_key/host/port — analytics should record what user actually supplied.
     _track_props = tracking.filter_command_kwargs(dict(locals()))
     tracking.track_event("execution_start", _track_props, mixpanel_name="run")
 
+    if no_watch:
+        os.environ["COMFY_NO_WATCH"] = "1"
+
     try:
         if api_key:
             api_key = api_key.strip() or None
@@ -812,6 +835,11 @@ def run(
             renderer.error(code="where_invalid", message=str(e), hint="use --where local or --where cloud")
             raise typer.Exit(code=1)
 
+        # Record the RESOLVED routing target so submission analytics can tell a
+        # cloud run from a local one even when --where was defaulted (the raw
+        # `where` kwarg is None then). Rides on the execution_success/_error events.
+        _track_props["target"] = "cloud" if decision.target is where_module.WhereTarget.CLOUD else "local"
+
         # Default for --notify: on when a human is at the terminal, off for
         # agents (they shouldn't get surprise side-channel processes they didn't
         # ask for). The user can override either way with --notify/--no-notify.
@@ -850,38 +878,42 @@ def run(
         if decision.target is where_module.WhereTarget.CLOUD:
             where_module.cloud_preflight_or_exit()
             # Cloud path uses HTTPS + Bearer auth; host/port aren't applicable.
+            # NOTE: do NOT `return` here — falling through to the try's `else`
+            # is what fires `execution_success`. An early return skipped it, so
+            # successful cloud submissions emitted `execution_start` but never
+            # `execution_success` (local runs were unaffected).
             run_inner.execute_cloud(
                 workflow,
                 wait=wait,
                 verbose=verbose,
                 timeout=timeout,
                 notify=effective_notify,
                 print_prompt=print_prompt,
+                workflow_id=workflow_id,
+                preloaded=preloaded,
+            )
+        else:
+            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,
             )
-            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,
-        )
     except typer.Exit as e:
         if (e.exit_code or 0) == 0:
             tracking.track_event("execution_success", _track_props)
@@ -902,13 +934,15 @@ def run(
 
 
 @app.command(
-    help="Validate an API-format workflow without submitting. Checks class_types, input shapes, enum values, and edge wiring."
+    help="Validate a workflow without submitting. Accepts API-format or a frontend/canvas "
+    "graph (auto-converted to API first). Checks class_types, required inputs, input shapes, "
+    "enum values, and edge wiring."
 )
 @tracking.track_command()
 def validate(
     workflow: Annotated[
         str,
-        typer.Option(help="Path to the API-format workflow JSON file."),
+        typer.Option(help="Path to the workflow JSON file (API format or a frontend/canvas graph)."),
     ],
     where: Annotated[
         str | None,
@@ -930,6 +964,8 @@ def validate(
     from pathlib import Path
 
     from comfy_cli.cql.engine import Graph, LoadError
+    from comfy_cli.cql.loader import resilient_load_object_info
+    from comfy_cli.workflow_to_api import WorkflowConversionError, convert_ui_to_api, is_api_format
 
     renderer = get_renderer()
 
@@ -961,8 +997,16 @@ def validate(
         except Exception:
             pass
 
+    # Resolve object_info ONCE through the shared loader so validate honors the
+    # same catalog every other command does — an explicit --input dump, the
+    # COMFY_OBJECT_INFO_FILE offline catalog, or the cache-first live fetch — and
+    # so the graph we validate against is built from the SAME catalog used to
+    # lower a canvas workflow below (previously the graph came from Graph.load,
+    # which ignored COMFY_OBJECT_INFO_FILE, while lowering honored it).
     try:
-        graph = Graph.load(mode=mode, input_path=input_path, host=host or "127.0.0.1", port=port or 8188)
+        object_info = resilient_load_object_info(
+            mode=mode, input_path=input_path, host=host or "127.0.0.1", port=port or 8188
+        )
     except LoadError as e:
         renderer.error(
             code="cql_no_graph",
@@ -971,9 +1015,46 @@ def validate(
             details=e.details,
         )
         raise typer.Exit(code=1) from e
+    graph = Graph.from_object_info(object_info)
+    graph._try_default_annotations()
+
+    # `validate_workflow` only inspects the API/prompt shape
+    # ({id: {class_type, inputs}}) — it iterates node inputs and checks wiring,
+    # required inputs, enums, and shapes. A frontend/canvas graph
+    # ({nodes: [...], links: [...]}) never gets its nodes examined: every
+    # top-level key is treated as a non-node and the result comes back
+    # valid:true even when the wiring is structurally broken. So a canvas
+    # workflow MUST be lowered to API format FIRST, using the SAME converter
+    # (and the SAME object_info resolution) the `run` path uses, so validate
+    # inspects exactly what the server would execute.
+    lowered = False
+    if not is_api_format(wf_data):
+        try:
+            wf_data = convert_ui_to_api(wf_data, object_info)
+        except WorkflowConversionError as e:
+            renderer.error(
+                code="conversion_error",
+                message=str(e),
+                hint="check that every node's required inputs are connected",
+            )
+            raise typer.Exit(code=1) from e
+        lowered = True
 
     result = graph.validate_workflow(wf_data)
 
+    # When the caller handed us a CANVAS graph, they have never seen the
+    # flattened ids the lowering mints for subgraph interiors (`57:3`) — their
+    # edit surface (slots / set-widget) speaks `57/3`. Key every issue by the
+    # editable address so a validate error can be acted on directly; keep the
+    # raw API id alongside for anyone correlating with server node_errors. An
+    # already-API input skips this: its ids address the document as given.
+    if lowered:
+        for issue in (*result["errors"], *result["warnings"]):
+            nid = str(issue.get("node_id", ""))
+            if ":" in nid:
+                issue["api_node_id"] = nid
+                issue["node_id"] = nid.replace(":", "/")
+
     payload = {
         "workflow": str(wf_path),
         "valid": result["valid"],
@@ -1510,6 +1591,10 @@ def feedback(
             else str(usability_satisfaction_score),
         },
     )
+    # Imported lazily: questionary pulls in prompt_toolkit (~50ms) and is only
+    # needed on this interactive feedback path.
+    import questionary
+
     if (
         sent
         and questionary.confirm("Do you want to provide additional feature-specific feedback on our GitHub page?").ask()

comfy_cli/comfy_client.py

@@ -324,10 +324,15 @@ def submit_prompt(
         *,
         timeout: float | None = None,
         extra_data: dict | None = None,
+        workflow_id: str | None = None,
     ) -> SubmitResult:
         """POST {prefix}/prompt — submit a workflow for execution.
 
         Caller may pass ``extra_data`` (merged into the request, not overwritten).
+        For cloud submissions, ``workflow_id`` (the cloud workflow entity id) is
+        forwarded as a top-level ``workflow_id`` field so the server can associate
+        the job with an existing workflow and auto-promote a draft on run. Omitted
+        from the body entirely when unset.
         For cloud submissions, the user's OAuth token is injected as
         ``auth_token_comfy_org`` so partner-API nodes (BFL Flux Pro, Gemini
         Nano Banana, etc.) can call out to comfy.org — matching what the web
@@ -352,6 +357,10 @@ def payload() -> dict[str, Any]:
                     merged_extra.setdefault("api_key_comfy_org", self.target.api_key)
             if merged_extra:
                 request_payload["extra_data"] = merged_extra
+            # Cloud workflow entity id: associate this job with an existing
+            # workflow (auto-promotes a draft on run). Only sent when provided.
+            if workflow_id:
+                request_payload["workflow_id"] = workflow_id
             return request_payload
 
         resp = self._request("POST", ("prompt",), body_factory=payload, timeout=timeout)

comfy_cli/command/assets_library.py

@@ -0,0 +1,113 @@
+"""``comfy assets library`` — browse and borrow assets from Comfy Cloud's
+asset library.
+
+Mirrors the cloud-saved-workflow subcommands in ``workflow.py`` (``list``,
+``get``, ...): thin Typer commands over ``cloud_http``'s shared helpers,
+emitting a JSON envelope via the renderer. Cloud-only — there is no local
+``/api/assets`` surface.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated, Any
+
+import typer
+
+from comfy_cli import tracking
+from comfy_cli.command.cloud_http import (
+    cloud_target_or_local_error,
+    handle_cloud_http_error,
+    http_request,
+)
+from comfy_cli.output.renderer import get_renderer
+
+app = typer.Typer(help="Browse your Comfy Cloud asset library (list, borrow).")
+
+
+@app.command("ls", help="List your assets on Comfy Cloud.")
+@tracking.track_command("assets")
+def ls_cmd(
+    name: Annotated[
+        str | None,
+        typer.Option("--name", show_default=False, help="Case-insensitive substring match on asset name."),
+    ] = None,
+    tags: Annotated[
+        str | None,
+        typer.Option(
+            "--tags", show_default=False, help="Comma-separated tags; assets must have ALL of them (e.g. input,output)."
+        ),
+    ] = None,
+    limit: Annotated[int, typer.Option("--limit", help="Cap rows returned (max 500).")] = 20,
+    where: Annotated[str | None, typer.Option("--where", show_default=False)] = None,
+):
+    import urllib.error
+    import urllib.parse
+
+    renderer = get_renderer()
+    target = cloud_target_or_local_error(where, renderer)
+
+    params: list[tuple[str, Any]] = [("limit", min(max(limit, 1), 500))]
+    if name:
+        params.append(("name_contains", name))
+    for t in tags.split(",") if tags else []:
+        t = t.strip()
+        if t:
+            params.append(("include_tags", t))
+    url = target.url("assets") + "?" + urllib.parse.urlencode(params)
+
+    try:
+        _, body = http_request(url, target)
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError) as e:
+        raise handle_cloud_http_error(renderer, e, operation="list") from e
+
+    rows = (body or {}).get("assets") or []
+    payload = {
+        "count": len(rows),
+        "assets": [
+            {
+                "id": r.get("id"),
+                "name": r.get("name"),
+                "hash": r.get("hash"),
+                "mime_type": r.get("mime_type"),
+                "size": r.get("size"),
+                "tags": r.get("tags"),
+                "preview_url": r.get("preview_url"),
+                "job_id": r.get("job_id"),
+                "created_at": r.get("created_at"),
+            }
+            for r in rows
+            if isinstance(r, dict)
+        ],
+    }
+    renderer.emit(payload, command="assets library ls", where="cloud")
+
+
+@app.command("ensure", help="Ensure you own an asset by content hash (borrows public/shared bytes, no re-upload).")
+@tracking.track_command("assets")
+def ensure_cmd(
+    hash: Annotated[str, typer.Option("--hash", help="Asset content hash (as returned by `assets library ls`).")],
+    tags: Annotated[
+        str,
+        typer.Option("--tags", help="Comma-separated tags to attach (>=1 required by the API)."),
+    ] = "input",
+    where: Annotated[str | None, typer.Option("--where", show_default=False)] = None,
+):
+    import urllib.error
+
+    renderer = get_renderer()
+    target = cloud_target_or_local_error(where, renderer)
+
+    tag_list = [t.strip() for t in tags.split(",") if t.strip()] or ["input"]
+    url = target.url("assets/from-hash")
+    try:
+        status, body = http_request(url, target, method="POST", body={"hash": hash, "tags": tag_list})
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError) as e:
+        raise handle_cloud_http_error(renderer, e, operation="ensure") from e
+
+    b = body or {}
+    payload = {
+        "id": b.get("id"),
+        "hash": b.get("hash", hash),
+        "created_new": status == 201,
+    }
+    renderer.emit(payload, command="assets library ensure", where="cloud")