feat(generate): --emit-ops writes a frontend-format workflow plus a stamped op batch (BE-11131)

comfy_cli/command/generate/app.py

@@ -284,7 +284,19 @@ def _generate_entry(
 
 def _separate_meta_flags(extra_args: list[str]) -> tuple[list[str], dict[str, str | bool]]:
     """Pull run-level flags out of the user's argv tail."""
-    meta_names = {"download", "async", "json", "timeout", "api-key", "emit-workflow", "output-prefix", "yes"}
+    meta_names = {
+        "download",
+        "async",
+        "json",
+        "timeout",
+        "api-key",
+        "emit-workflow",
+        "emit-ops",
+        "actor",
+        "base-version",
+        "output-prefix",
+        "yes",
+    }
     meta: dict[str, str | bool] = {}
     remaining: list[str] = []
     i = 0
@@ -296,7 +308,7 @@ def _separate_meta_flags(extra_args: list[str]) -> tuple[list[str], dict[str, st
             if "=" in body:
                 body, raw = body.split("=", 1)
             if body in meta_names:
-                if body in {"async", "json", "yes"}:
+                if body in {"async", "json", "yes", "emit-ops"}:
                     meta[body] = True if raw is None else raw.lower() not in {"false", "0", "no"}
                     i += 1
                     continue
@@ -564,14 +576,64 @@ def _track_error(error_kind: str, exc: BaseException) -> None:
                 hint=f"Run `comfy generate schema {name}` for the full parameter list.",
             )
 
+        emit_ops_mode = bool(meta.get("emit-ops", False))
+        if emit_ops_mode and not emit_path:
+            _bail(
+                _track_error,
+                schema.SchemaError("--emit-ops requires --emit-workflow <path>"),
+                code="generate_bad_args",
+                message="--emit-ops requires --emit-workflow <path>: the op batch describes the workflow written there",
+                kind="schema",
+                hint="add --emit-workflow workflow.json",
+            )
         if emit_path:
             # Emit a runnable workflow that drives the partner *node* and return
             # — no proxy call, no API key required. The artifact is the result.
             name = gen_props["model_alias"] or ep.id
             prefix = meta.get("output-prefix") if isinstance(meta.get("output-prefix"), str) else "generate"
             renderer = get_renderer()
+            ops: list | None = None
             try:
-                workflow = emit.write_workflow(name, values, Path(emit_path).expanduser(), output_prefix=prefix)
+                if emit_ops_mode:
+                    # FRONTEND-format file + a stamped replace_ops batch, so the
+                    # written graph is canvas-editable and a shared-document
+                    # consumer folds it in as attributed ops instead of a
+                    # wholesale replacement — same contract as
+                    # `templates fetch --emit-ops`. The graph loads through the
+                    # same resilient path every workflow edit verb uses
+                    # (COMFY_OBJECT_INFO_FILE honored, cache fallback).
+                    from comfy_cli.command.workflow import _get_graph
+
+                    actor = meta.get("actor") if isinstance(meta.get("actor"), str) else "cli"
+                    try:
+                        base_version = int(meta.get("base-version", 0))
+                    except (TypeError, ValueError) as e:
+                        _bail(
+                            _track_error,
+                            e,
+                            code="generate_bad_args",
+                            message=f"--base-version must be an integer, got {meta.get('base-version')!r}",
+                            kind="schema",
+                        )
+                    try:
+                        graph = _get_graph(None, None, None)
+                    except typer.Exit as e:
+                        # _get_graph already rendered cql_no_graph; only the
+                        # generate:error record is owed here, or generate:start
+                        # is left without its terminal event.
+                        _track_error("emit", e)
+                        raise
+                    workflow, ops = emit.write_frontend_workflow(
+                        name,
+                        values,
+                        Path(emit_path).expanduser(),
+                        graph,
+                        actor=actor,
+                        base_version=base_version,
+                        output_prefix=prefix,
+                    )
+                else:
+                    workflow = emit.write_workflow(name, values, Path(emit_path).expanduser(), output_prefix=prefix)
             except emit.UnsupportedModelError as e:
                 # Its own code: the remedy is "pick another model", which is
                 # not what the umbrella `emit_workflow_failed` hint says, and
@@ -600,14 +662,20 @@ def _track_error(error_kind: str, exc: BaseException) -> None:
                     hint=hint,
                 )
                 raise typer.Exit(code=1) from e
-            tracking.track_event("generate:emit", {**gen_props, "node_count": len(workflow)})
+            node_count = len(workflow["nodes"]) if emit_ops_mode else len(workflow)
+            tracking.track_event("generate:emit", {**gen_props, "node_count": node_count})
             if renderer.is_pretty():
                 rprint(f"[bold green]Wrote workflow:[/bold green] {emit_path}")
                 rprint(f"  run it: comfy run --workflow {emit_path}")
-            renderer.emit(
-                {"out": str(Path(emit_path).expanduser()), "model": name, "nodes": len(workflow)},
-                command="generate emit-workflow",
-            )
+            payload = {
+                "out": str(Path(emit_path).expanduser()),
+                "model": name,
+                "nodes": node_count,
+                "format": "frontend" if emit_ops_mode else "api",
+            }
+            if ops is not None:
+                payload["ops"] = ops
+            renderer.emit(payload, command="generate emit-workflow")
             return
 
         # Spend gate — a proxy call spends Comfy credits, so consent comes
@@ -1191,6 +1259,10 @@ def _print_top_help() -> None:
         '  comfy generate flux-2 --prompt "a fox" --emit-workflow flux.json   '
         "[dim]# write a runnable workflow instead of calling the proxy[/dim]"
     )
+    rprint(
+        '  comfy generate flux-2 --prompt "a fox" --emit-workflow flux.json --emit-ops [--actor ID] [--base-version N]'
+    )
+    rprint("      [dim]# frontend-format (canvas-editable) file plus a stamped op batch in the envelope[/dim]")
     rprint("")
     rprint("[bold]Actions:[/bold]")
     rprint("  comfy generate list                    Browse available models")

comfy_cli/command/generate/emit.py

@@ -365,3 +365,139 @@ def write_workflow(
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(json.dumps(workflow, indent=2) + "\n", encoding="utf-8")
     return workflow
+
+
+# ---------------------------------------------------------------------------
+# --emit-ops: the same graph, expressed as the frozen op vocabulary
+# ---------------------------------------------------------------------------
+#
+# ``--emit-workflow`` writes API format, which the canvas and every edit tool
+# refuse (workflow_not_frontend_format) and which the CRDT write path cannot
+# attribute. Rather than converting API→frontend after the fact — a second
+# implementation of widget order and layout — the emitter mints the SAME graph
+# as add_node/set_widget/connect specs and lets ``workflow_ops.apply_specs``
+# materialize the frontend workflow: the exact machinery every hand edit
+# already uses, so widget ordering, autogrow growth and position assignment
+# have one answer. The API graph from :func:`build_workflow` stays the single
+# source of the model→node mapping; this is a mechanical re-expression of it.
+
+
+def _is_link_ref(value: Any, node_ids: set[str]) -> bool:
+    """An API input value of the shape ``[node_id, output_index]``."""
+    return (
+        isinstance(value, list)
+        and len(value) == 2
+        and str(value[0]) in node_ids
+        and isinstance(value[1], int)
+        and not isinstance(value[1], bool)
+    )
+
+
+def ops_from_api_workflow(api_wf: dict[str, Any], graph: Any) -> list[dict[str, Any]]:
+    """Re-express an API-format graph as batch specs for ``apply_specs``.
+
+    Shape: every ``add_node`` first (each with a batch-local alias), then every
+    ``set_widget`` (non-dotted keys before dotted ones, so a dynamic-combo
+    selection lands before the sub-widgets it exposes), then every ``connect``
+    — an order in which every referenced endpoint already exists.
+
+    ``graph`` is accepted for parity with the applier's signature and future
+    schema-aware canonicalization; the current mapping is purely structural.
+    """
+    del graph  # structural mapping today; see docstring
+    node_ids = {str(k) for k in api_wf}
+
+    def alias(nid: Any) -> str:
+        return f"gen{nid}"
+
+    adds: list[dict[str, Any]] = []
+    widgets: list[dict[str, Any]] = []
+    connects: list[dict[str, Any]] = []
+    for nid in sorted(api_wf, key=str):
+        node = api_wf[nid]
+        # allow_deprecated: the model→node mapping is curated (and pinned by
+        # test_emit's endpoint invariant), so a class the catalog has since
+        # flagged deprecated is still the intended target — the gate exists to
+        # stop a GUESSED class, not a mapped one.
+        adds.append({"op": "add_node", "class_type": node["class_type"], "as": alias(nid), "allow_deprecated": True})
+        inputs = node.get("inputs") or {}
+        keys = sorted(inputs, key=lambda k: (k.count("."), list(inputs).index(k)))
+        for key in keys:
+            value = inputs[key]
+            if _is_link_ref(value, node_ids):
+                connects.append(
+                    {
+                        "op": "connect",
+                        "from": f"${alias(value[0])}.{value[1]}",
+                        "to": f"${alias(nid)}.{key}",
+                    }
+                )
+            else:
+                widgets.append({"op": "set_widget", "node": f"${alias(nid)}", "widget": key, "value": value})
+    return adds + widgets + connects
+
+
+_EMPTY_FRONTEND: dict[str, Any] = {
+    "nodes": [],
+    "links": [],
+    "version": 0.4,
+    "last_node_id": 0,
+    "last_link_id": 0,
+}
+
+
+def write_frontend_workflow(
+    model: str,
+    values: dict[str, Any],
+    path: Path,
+    graph: Any,
+    *,
+    actor: str = "cli",
+    base_version: int = 0,
+    output_prefix: str = "generate",
+) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """Build the workflow for ``model`` as a FRONTEND-format graph and write it
+    to ``path``; return ``(workflow, ops)`` where ``ops`` is the stamped
+    ``replace_ops`` batch that turns whatever ``path`` previously held into the
+    new graph (empty previous ⇒ no delete half), ready for the envelope exactly
+    like ``templates fetch --emit-ops``.
+
+    Raises ``EmitError``/``UnsupportedModelError`` like :func:`write_workflow`;
+    an applier failure surfaces as ``EmitError`` (the request itself was
+    expressible — a failure here is a schema/catalog mismatch worth reporting).
+    """
+    from comfy_cli import workflow_ops
+
+    api = build_workflow(model, values, output_prefix=output_prefix)
+    specs = ops_from_api_workflow(api, graph)
+    try:
+        workflow, _ops, _aliases = workflow_ops.apply_specs(  # noqa: F841 — wf is the product; batch below is replace-shaped
+            json.loads(json.dumps(_EMPTY_FRONTEND)), graph, specs, actor=actor, base_version=base_version
+        )
+    except (ValueError, KeyError) as e:
+        raise EmitError(f"could not materialize the {model!r} workflow as canvas ops: {e}") from e
+
+    previous: dict[str, Any] = {}
+    try:
+        loaded = json.loads(path.read_text(encoding="utf-8"))
+        if isinstance(loaded, dict) and isinstance(loaded.get("nodes"), list):
+            previous = loaded
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        previous = {}
+
+    # The document leaves this process here — through `strip_internal`, like
+    # every other write path. `apply_specs` leaves `_applied_ops` and
+    # `_widget_stamps` on the workflow; written to disk, the emit run's stamps
+    # seed the LWW register and outrank the next `set-widget` (at a higher
+    # `--base-version` deterministically, at the defaults on a coin flip), and
+    # the edit reports `ok: true` while changing nothing. Strip BEFORE
+    # `replace_ops` so the returned workflow, the file and the batch describe
+    # one document.
+    workflow_ops.strip_internal(workflow)
+    try:
+        ops = workflow_ops.replace_ops(previous, workflow, actor=actor, base_version=base_version)
+    except workflow_ops.NotExpressibleError as e:  # can't-happen for our own built graph; fail loudly if it does
+        raise EmitError(f"the {model!r} workflow cannot be expressed as ops: {e}") from e
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(workflow, indent=2) + "\n", encoding="utf-8")
+    return workflow, ops

comfy_cli/workflow_to_api.py

@@ -1237,13 +1237,27 @@ def consume(name: str, spec: Any, depth: int = 0, next_spec: Any = None) -> None
         ):
             vidx += 1
 
-    # Flatten required+optional first so each input knows its successor's schema.
+    # Flatten required+optional first so each input knows its successor's
+    # schema. Within each section, honor ``input_order`` the way the cql
+    # engine's ``_ordered_names`` does (listed names first, leftovers in dict
+    # order): the input DICT's own order is only trustworthy on a catalog that
+    # was never re-serialized, and pairing widgets positionally from a sorted
+    # dict silently swaps neighboring values (observed: GeminiImageNode's
+    # prompt/model traded places on an alphabetized fixture).
+    input_order = schema.get("input_order") if isinstance(schema, dict) else None
+    if not isinstance(input_order, dict):
+        input_order = {}
     ordered: list[tuple[str, Any]] = []
     for section in ("required", "optional"):
         section_def = input_def.get(section) or {}
         if not isinstance(section_def, dict):
             continue
-        ordered.extend(section_def.items())
+        section_order = input_order.get(section)
+        names = list(section_def.keys())
+        if isinstance(section_order, list):
+            listed = [n for n in section_order if n in section_def]
+            names = listed + [n for n in names if n not in listed]
+        ordered.extend((n, section_def[n]) for n in names)
     for i, (input_name, input_spec) in enumerate(ordered):
         consume(input_name, input_spec, 0, next_widget_spec(ordered, i + 1))
     return pairs

tests/comfy_cli/command/generate/test_app_lifecycle.py

@@ -209,6 +209,14 @@ def test_download_flag_sets_has_download(self, runner, captured_events, api_key,
 
 
 class TestGenerateExecutionErrorPaths:
+    def test_emit_ops_without_emit_workflow_emits_generate_error_with_kind_schema(self, runner, captured_events):
+        r = runner.invoke(cli_app, ["generate", "nano-banana", "--prompt", "x", "--emit-ops"])
+        assert r.exit_code == 1
+
+        err_props = _props(captured_events, "generate:error")
+        assert len(err_props) == 1, "generate:start needs its terminal generate:error"
+        assert err_props[0]["error_kind"] == "schema"
+
     def test_api_error_emits_generate_error_with_kind_api(self, runner, captured_events, api_key, monkeypatch):
         resp = httpx.Response(401, json={"message": "Invalid token"})
         monkeypatch.setattr(gen_app.client.httpx, "post", lambda *a, **kw: resp)