refactor(workflow): re-home API-workflow validate under comfy workflow validate (BE-3000)

comfy_cli/cmdline.py

@@ -902,7 +902,9 @@ def run(
 
 
 @app.command(
-    help="Validate an API-format workflow without submitting. Checks class_types, input shapes, enum values, and edge wiring."
+    hidden=True,
+    help="[DEPRECATED — use 'comfy workflow validate'] Validate an API-format workflow without submitting. "
+    "Checks class_types, input shapes, enum values, and edge wiring.",
 )
 @tracking.track_command()
 def validate(
@@ -927,81 +929,14 @@ def validate(
         typer.Option("--input", show_default=False, help="Path to a saved object_info JSON (offline mode)."),
     ] = None,
 ):
-    from pathlib import Path
-
-    from comfy_cli.cql.engine import Graph, LoadError
-
-    renderer = get_renderer()
-
-    # Load workflow
-    wf_path = Path(workflow).expanduser()
-    if not wf_path.is_file():
-        renderer.error(code="workflow_not_found", message=f"Workflow file not found: {workflow}", hint="check the path")
-        raise typer.Exit(code=1)
-    try:
-        wf_data = json.loads(wf_path.read_text(encoding="utf-8"))
-    except json.JSONDecodeError as e:
-        renderer.error(code="workflow_invalid_json", message=f"Invalid JSON: {e}", hint="re-export from ComfyUI")
-        raise typer.Exit(code=1) from e
-    if not isinstance(wf_data, dict):
-        renderer.error(
-            code="workflow_not_api_format", message="Workflow must be a JSON object", hint="use File > Export (API)"
-        )
-        raise typer.Exit(code=1)
-
-    # Load graph
-    mode = "local"
-    if where:
-        mode = where
-    else:
-        config = ConfigManager()
-        try:
-            decision = where_module.resolve(flag=None, config_value=config.get(where_module.CONFIG_KEY_WHERE_DEFAULT))
-            mode = decision.target.value
-        except Exception:
-            pass
-
-    try:
-        graph = Graph.load(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",
-            message=str(e),
-            hint=e.details.get("hint", "pass --input <object_info.json>, or start the server"),
-            details=e.details,
-        )
-        raise typer.Exit(code=1) from e
-
-    result = graph.validate_workflow(wf_data)
-
-    payload = {
-        "workflow": str(wf_path),
-        "valid": result["valid"],
-        "error_count": len(result["errors"]),
-        "warning_count": len(result["warnings"]),
-        "errors": result["errors"],
-        "warnings": result["warnings"],
-    }
-
-    if renderer.is_pretty():
-        if result["valid"]:
-            rprint(f"[bold green]✓[/bold green] workflow is valid ({len(wf_data)} nodes)")
-            for w in result["warnings"]:
-                rprint(f"  [yellow]⚠[/yellow] {w.get('message', '')}")
-        else:
-            rprint(f"[bold red]✗[/bold red] {len(result['errors'])} error(s)")
-            for e in result["errors"]:
-                msg = e.get("message", "")
-                suggestions = e.get("suggestions", [])
-                if suggestions:
-                    msg += f" (did you mean: {', '.join(suggestions[:3])}?)"
-                rprint(f"  [red]•[/red] node {e.get('node_id', '?')}: {msg}")
-            for w in result["warnings"]:
-                rprint(f"  [yellow]⚠[/yellow] {w.get('message', '')}")
-    renderer.emit(payload, command="validate", ok=result["valid"])
-
-    if not result["valid"]:
-        raise typer.Exit(code=1)
+    # Deprecated alias for `comfy workflow validate` (its canonical home). Kept
+    # functional for backward compatibility; warns and delegates to the shared
+    # implementation. The warning routes to stderr in JSON modes, so structured
+    # output stays clean.
+    from comfy_cli.command.workflow import validate_api_workflow
+
+    get_renderer().warn("`comfy validate` is deprecated — use `comfy workflow validate` instead.")
+    validate_api_workflow(workflow, where=where, host=host, port=port, input_path=input_path, command="validate")
 
 
 @app.command(help="Upload files to the ComfyUI server's input directory.")

comfy_cli/command/custom_nodes/command.py

@@ -1013,7 +1013,7 @@ def validate_node_for_publishing():
     return config
 
 
-@app.command("validate", help="Run validation checks for publishing")
+@app.command("validate", help="Validate this custom node for registry publishing")
 @tracking.track_command("publish")
 def validate():
     """

comfy_cli/command/workflow.py

@@ -1146,6 +1146,162 @@ def delete_cmd(
     renderer.emit(payload, command="workflow delete", where="cloud", changed=True)
 
 
+# ---------------------------------------------------------------------------
+# validate — API-format workflow validation
+# ---------------------------------------------------------------------------
+# The canonical home for API-format workflow validation. The top-level
+# `comfy validate` is kept as a hidden deprecated alias that delegates to the
+# shared implementation below (see cmdline.py).
+
+
+def validate_api_workflow(
+    workflow: str,
+    *,
+    where: str | None = None,
+    host: str | None = None,
+    port: int | None = None,
+    input_path: str | None = None,
+    command: str = "workflow validate",
+) -> None:
+    """Validate an API-format workflow without submitting it.
+
+    Shared implementation behind ``comfy workflow validate`` (canonical) and the
+    deprecated top-level ``comfy validate`` alias. Checks class_types, input
+    shapes, enum values, and edge wiring against object_info loaded from the run
+    target. ``command`` labels the emitted envelope so each entry point reports
+    its own path.
+    """
+    from comfy_cli import where as where_module
+    from comfy_cli.cql.engine import Graph, LoadError
+
+    renderer = get_renderer()
+
+    # Load workflow
+    wf_path = Path(workflow).expanduser()
+    if not wf_path.is_file():
+        renderer.error(code="workflow_not_found", message=f"Workflow file not found: {workflow}", hint="check the path")
+        raise typer.Exit(code=1)
+    try:
+        wf_data = json.loads(wf_path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as e:
+        renderer.error(code="workflow_invalid_json", message=f"Invalid JSON: {e}", hint="re-export from ComfyUI")
+        raise typer.Exit(code=1) from e
+    except (OSError, UnicodeDecodeError) as e:
+        # e.g. a non-UTF-8 file, permission denied, or a TOCTOU race if the file
+        # vanished after the is_file() check above. Report structurally instead
+        # of crashing with a raw traceback.
+        renderer.error(
+            code="workflow_read_error",
+            message=f"Unable to read workflow file: {e}",
+            hint="check file permissions and encoding",
+        )
+        raise typer.Exit(code=1) from e
+    if not isinstance(wf_data, dict):
+        renderer.error(
+            code="workflow_not_api_format", message="Workflow must be a JSON object", hint="use File > Export (API)"
+        )
+        raise typer.Exit(code=1)
+
+    # Load graph. Resolve routing up front so a typo (e.g. `clod`) fails here
+    # with an envelope: forwarded verbatim, it would surface as a bare
+    # ValueError from Graph.load's target resolution (online path only —
+    # Graph.load short-circuits to --input when supplied).
+    mode = "local"
+    try:
+        mode = where_module.resolve_default(flag=where).target.value
+    except ValueError as e:
+        if where:
+            renderer.error(
+                code="where_invalid",
+                message=str(e),
+                hint="use `--where local` or `--where cloud`",
+            )
+            raise typer.Exit(code=1) from e
+        # A bad env/project/config value with no explicit flag never breaks the
+        # command — drop to the local default, as before.
+    except Exception:
+        pass
+
+    try:
+        graph = Graph.load(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",
+            message=str(e),
+            hint=e.details.get("hint", "pass --input <object_info.json>, or start the server"),
+            details=e.details,
+        )
+        raise typer.Exit(code=1) from e
+
+    result = graph.validate_workflow(wf_data)
+
+    payload = {
+        "workflow": str(wf_path),
+        "valid": result["valid"],
+        "error_count": len(result["errors"]),
+        "warning_count": len(result["warnings"]),
+        "errors": result["errors"],
+        "warnings": result["warnings"],
+    }
+
+    if renderer.is_pretty():
+        # Workflow-supplied strings (node ids, messages, enum/input values echoed
+        # in messages) flow into rprint, which parses Rich markup. Escape them so a
+        # crafted file can't inject markup to spoof/hide output (e.g. fake a green ✓).
+        from rich.markup import escape
+
+        if result["valid"]:
+            rprint(f"[bold green]✓[/bold green] workflow is valid ({len(wf_data)} nodes)")
+            for w in result["warnings"]:
+                rprint(f"  [yellow]⚠[/yellow] {escape(str(w.get('message', '')))}")
+        else:
+            rprint(f"[bold red]✗[/bold red] {len(result['errors'])} error(s)")
+            for e in result["errors"]:
+                msg = str(e.get("message", ""))
+                suggestions = e.get("suggestions", [])
+                if suggestions:
+                    msg += f" (did you mean: {', '.join(str(s) for s in suggestions[:3])}?)"
+                rprint(f"  [red]•[/red] node {escape(str(e.get('node_id', '?')))}: {escape(msg)}")
+            for w in result["warnings"]:
+                rprint(f"  [yellow]⚠[/yellow] {escape(str(w.get('message', '')))}")
+    renderer.emit(payload, command=command, ok=result["valid"])
+
+    if not result["valid"]:
+        raise typer.Exit(code=1)
+
+
+@app.command(
+    "validate",
+    help="Validate an API-format workflow without submitting. Checks class_types, input shapes, enum values, and edge wiring.",
+)
+@tracking.track_command("workflow")
+def validate_cmd(
+    workflow: Annotated[
+        str,
+        typer.Option(help="Path to the API-format workflow JSON file."),
+    ],
+    where: Annotated[
+        str | None,
+        typer.Option("--where", show_default=False, help="Routing target for object_info: 'local' or 'cloud'."),
+    ] = None,
+    host: Annotated[
+        str | None,
+        typer.Option(show_default=False, help="ComfyUI host (default 127.0.0.1)."),
+    ] = None,
+    port: Annotated[
+        int | None,
+        typer.Option(show_default=False, help="ComfyUI port (default 8188)."),
+    ] = None,
+    input_path: Annotated[
+        str | None,
+        typer.Option("--input", show_default=False, help="Path to a saved object_info JSON (offline mode)."),
+    ] = None,
+):
+    validate_api_workflow(
+        workflow, where=where, host=host, port=port, input_path=input_path, command="workflow validate"
+    )
+
+
 # ---------------------------------------------------------------------------
 # compose / fragment — fragment-based workflow composition
 # ---------------------------------------------------------------------------

comfy_cli/command/workflow_fragments.py

@@ -453,7 +453,7 @@ def fragment_show_cmd(
     renderer.emit(payload, command="workflow fragment show")
 
 
-@fragment_app.command("validate", help="Validate that a fragment file is well-formed.")
+@fragment_app.command("validate", help="Validate that a workflow-fragment file is well-formed")
 @tracking.track_command("workflow")
 def fragment_validate_cmd(
     fragment: Annotated[str, typer.Argument(help="Fragment name (looked up in --lib) or path to .json.")],

comfy_cli/discovery.py

@@ -38,6 +38,9 @@
     "comfy jobs wait": "jobs_wait",
     # help / validation
     "comfy help": "help",
+    # `comfy workflow validate` is the canonical home; `comfy validate` is the
+    # hidden deprecated alias — both emit the same workflow-validation payload.
+    "comfy workflow validate": "workflow",
     "comfy validate": "workflow",
     # nodes introspection
     "comfy nodes ls": "nodes",

comfy_cli/skills/comfy/SKILL.md

@@ -678,7 +678,7 @@ Before `comfy run`, verify the workflow will succeed:
 comfy run --workflow wf.json --print-prompt
 
 # Full pre-flight: checks class_types, input shapes, enum values, edge wiring
-comfy --json validate --workflow api.json
+comfy --json workflow validate --workflow api.json
 
 # Spot-check a single node class exists on the target
 comfy --json nodes show <ClassName>
@@ -863,7 +863,7 @@ Hard-won lessons per domain. Not a tutorial — a reference card.
 - Assembly: GetVideoComponents → BatchImagesNode → CreateVideo → SaveVideo (ImageBatch is deprecated)
 - Autogrow inputs (type COMFY_AUTOGROW_*, e.g. BatchImagesNode `images`): wire ONE slot key per
   connection — `"images.image0": [..], "images.image1": [..]` — never a single `images` link.
-  `nodes show` prints the `wire_as` form; `comfy validate` rejects the bare form before submit.
+  `nodes show` prints the `wire_as` form; `comfy workflow validate` rejects the bare form before submit.
 - I2V pattern: LoadImage → I2VNode → SaveVideo (check `nodes show` for the I2V node)
 - **Model enums mix t2v and i2v variants** — a node's `model` choices may include
   image-to-video-only models (e.g. `grok-imagine-video-1.5`) that fail at runtime