feat(workflow): define-subgraph command emitting define_subgraph op

comfy_cli/command/workflow.py

@@ -1934,6 +1934,9 @@ def validate_cmd(
 
 from comfy_cli.command import workflow_edit as _wedit  # noqa: E402
 
+app.command("define-subgraph", help="Create a subgraph definition; emits one define_subgraph op.")(
+    _wedit.define_subgraph_cmd
+)
 app.command("add-node", help="Add a node to the graph; emits an add_node op.")(_wedit.add_node_cmd)
 app.command("connect", help="Wire an output slot to an input slot; emits a connect op.")(_wedit.connect_cmd)
 app.command("set-widget", help="Set a widget by name (`<id>.<widget>`); emits a set_widget op.")(_wedit.set_widget_cmd)

comfy_cli/command/workflow_edit.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import json
+import stat
 from pathlib import Path
 from typing import Annotated, Any
 
@@ -48,6 +49,7 @@
 HostOpt = Annotated[str | None, typer.Option(show_default=False)]
 PortOpt = Annotated[int | None, typer.Option(show_default=False)]
 WhereOpt = Annotated[str | None, typer.Option("--where", show_default=False, help="Catalog target: local | cloud.")]
+_MAX_DEFINITION_BYTES = 16 * 1024 * 1024
 
 
 def _emit_edit_error(renderer, e: ValueError, *, hint: str) -> None:
@@ -128,10 +130,69 @@ def _finish(renderer, p, workflow: dict, op: dict, base_version: int, stdout: bo
     renderer.emit(payload, command=command, changed=not stdout)
 
 
+def _emit_op(renderer, p: Path, op: dict, base_version: int, command: str) -> None:
+    """Emit an op without applying it or writing the source workflow."""
+    payload = {"workflow": str(p), "op": op, "base_version": base_version, "wrote": None}
+    if renderer.is_pretty():
+        rprint(f"[bold green]✓[/bold green] {op['op']} emitted for [dim]{p}[/dim]")
+    renderer.emit(payload, command=command, changed=False)
+
+
 def _graph_or_exit(input_path, host, port, renderer, where=None):
     return _get_graph(input_path, host, port, where=where)
 
 
+# ---------------------------------------------------------------------------
+# define-subgraph
+# ---------------------------------------------------------------------------
+
+
+def _read_subgraph_definition(path: Path):
+    """Read a bounded regular JSON file without blocking on devices or FIFOs."""
+    file_stat = path.stat()
+    if not stat.S_ISREG(file_stat.st_mode):
+        raise ValueError("subgraph definition must be a regular file")
+    if file_stat.st_size > _MAX_DEFINITION_BYTES:
+        raise ValueError(f"subgraph definition is too large (maximum {_MAX_DEFINITION_BYTES} bytes)")
+    try:
+        with path.open("rb") as definition_file:
+            raw = definition_file.read(_MAX_DEFINITION_BYTES + 1)
+        if len(raw) > _MAX_DEFINITION_BYTES:
+            raise ValueError(f"subgraph definition is too large (maximum {_MAX_DEFINITION_BYTES} bytes)")
+        return json.loads(raw.decode("utf-8"))
+    except (RecursionError, MemoryError) as error:
+        raise ValueError("subgraph definition is too deeply nested or too large") from error
+
+
+@tracking.track_command("workflow")
+def define_subgraph_cmd(
+    file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to update.")],
+    definition_file: Annotated[str, typer.Argument(help="Serializable subgraph definition JSON.")],
+    subgraph_id: Annotated[str | None, typer.Option("--id", show_default=False)] = None,
+    actor: ActorOpt = "cli",
+    base_version: BaseVersionOpt = 0,
+    stdout: StdoutOpt = False,
+):
+    """Create one subgraph definition and emit one ``define_subgraph`` op."""
+    renderer = get_renderer()
+    renderer.command = "workflow define-subgraph"
+    p, workflow = _load_workflow_or_fail(renderer, file)
+    try:
+        definition_path = Path(definition_file).expanduser()
+        definition = _read_subgraph_definition(definition_path)
+        _, op = workflow_ops.define_subgraph(
+            workflow,
+            definition,
+            subgraph_id=subgraph_id,
+            actor=actor,
+            base_version=base_version,
+        )
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError, RecursionError, MemoryError) as e:
+        _emit_edit_error(renderer, e, hint="provide a serializable subgraph definition JSON object")
+        raise typer.Exit(code=1) from e
+    _emit_op(renderer, p, op, base_version, "workflow define-subgraph")
+
+
 # ---------------------------------------------------------------------------
 # add-node
 # ---------------------------------------------------------------------------
@@ -754,6 +815,11 @@ def apply_cmd(
         raise typer.Exit(code=1) from e
 
     try:
+        deferred = [
+            spec.get("op") for spec in specs if isinstance(spec, dict) and spec.get("op") in workflow_ops.DEFERRED_OPS
+        ]
+        if deferred:
+            raise ValueError(f"local apply cannot persist deferred operation(s): {', '.join(deferred)}")
         workflow, ops, aliases = workflow_ops.apply_specs(
             workflow, graph, specs, actor=actor, base_version=base_version
         )
@@ -917,6 +983,13 @@ def foreach_cmd(
     out.mkdir(parents=True, exist_ok=True)
     written: list[str] = []
     try:
+        deferred = [
+            spec.get("op")
+            for spec in specs_template
+            if isinstance(spec, dict) and spec.get("op") in workflow_ops.DEFERRED_OPS
+        ]
+        if deferred:
+            raise ValueError(f"local foreach cannot persist deferred operation(s): {', '.join(deferred)}")
         for i, pset in enumerate(param_sets):
             if not isinstance(pset, dict):
                 raise workflow_ops.RecipeError(f"param-set #{i} must be a JSON object")

comfy_cli/workflow_ops.py

@@ -64,6 +64,7 @@
 # New ids live in [2**40, 2**53): always large (never collides with small
 # frontend counter ids), always inside JS Number.MAX_SAFE_INTEGER.
 _ID_FLOOR = 1 << 40
+_SUBGRAPH_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.I)
 
 
 def mint_id() -> int:
@@ -91,17 +92,25 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
 # ---------------------------------------------------------------------------
 
 #: Every op kind in the v1 vocabulary, including defined-but-deferred kinds.
-FROZEN_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node", "clear", "reset_doc")
-
-#: Kinds frozen in the contract whose replay is not implemented yet.
-#: ``apply_op`` must keep rejecting these. Empty since amendment v1.1:
-#: ``reset_doc`` was un-deferred by the bulk-writers ticket (V1-038).
-DEFERRED_OPS: tuple[str, ...] = ()
+FROZEN_OPS: tuple[str, ...] = (
+    "add_node",
+    "connect",
+    "set_widget",
+    "delete_node",
+    "clear",
+    "reset_doc",
+    "define_subgraph",
+)
+
+#: Kinds frozen in the contract whose replay is not implemented in the CLI.
+#: ``define_subgraph`` is emitted for cmp to validate and apply; the CLI must
+#: keep rejecting local replay to preserve that ownership boundary.
+DEFERRED_OPS: tuple[str, ...] = ("define_subgraph",)
 
 #: Kinds a batch (``apply_specs``) dispatches. ``clear`` and ``reset_doc`` are
 #: standalone-only: they rewrite the whole document, so they never ride inside
 #: an atomic batch.
-BATCHABLE_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node")
+BATCHABLE_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node", "define_subgraph")
 
 #: Per-kind rendering for :class:`NotBatchableError` — the registered error code
 #: and the standalone command that DOES do the job. One entry per frozen kind
@@ -1131,20 +1140,18 @@ class NotExpressibleError(ValueError):
 def _inexpressible_reason(workflow: dict) -> str | None:
     """Why ``workflow`` cannot be rebuilt from add_node/connect ops, or None.
 
-    The frozen vocabulary has four batchable kinds and none of them can create a
-    subgraph definition, a canvas group, or a reroute point — so a graph that
-    carries any of those is not reconstructible from ops, full stop. Enumerated
+    The frozen vocabulary cannot create a canvas group or a reroute point, so a
+    graph that carries either is not reconstructible from ops. Enumerated
     positively (a closed list of things we know we CAN'T do) rather than by
     trying and checking, so an unexpressible template fails before it has
     written anything.
     """
     if not isinstance(workflow, dict) or not isinstance(workflow.get("nodes"), list):
         return "not a frontend-format workflow (no `nodes` list) — only the save/UI format can be op-ified"
-    definitions = workflow.get("definitions")
-    if isinstance(definitions, dict) and definitions.get("subgraphs"):
-        return "the workflow contains a subgraph definition, which no frozen op kind can create"
     if workflow.get("groups"):
         return "the workflow contains canvas groups, which no frozen op kind can create"
+    if (workflow.get("definitions") or {}).get("subgraphs"):
+        return "the workflow contains subgraph definitions, which only cmp can project into ops"
     extra = workflow.get("extra")
     if isinstance(extra, dict) and (extra.get("reroutes") or extra.get("linkExtensions")):
         return "the workflow contains reroute points, which no frozen op kind can create"
@@ -1751,6 +1758,14 @@ def apply_specs(
                     workflow, op = delete_node(
                         workflow, graph, resolve_ref(spec["node"], aliases), actor=actor, base_version=base_version
                     )
+                elif kind == "define_subgraph":
+                    workflow, op = define_subgraph(
+                        workflow,
+                        spec["subgraph_definition"],
+                        subgraph_id=spec.get("subgraph_id"),
+                        actor=actor,
+                        base_version=base_version,
+                    )
                 elif kind in _NOT_BATCHABLE:
                     # In the frozen vocabulary but standalone-only — surfaced with
                     # its own registered code so the caller learns the standalone
@@ -1791,10 +1806,12 @@ def apply_specs(
 def apply_op(workflow: dict, op: dict, graph) -> dict:
     """Replay one op onto ``workflow`` in place and return it. Idempotent: an
     op whose ``op_id`` was already applied is a no-op."""
-    applied = workflow.setdefault("_applied_ops", [])
+    applied = workflow.get("_applied_ops", [])
     if op["op_id"] in applied:
         return workflow
     kind = op["op"]
+    if kind != "define_subgraph" and any(key in op for key in ("subgraph_id", "subgraph_definition", "definitions")):
+        raise ValueError(f"malformed_op: {kind} cannot carry a subgraph definition")
     # Snapshot the LWW bookkeeping so an exception escaping a handler cannot
     # leave a stamp committed WITHOUT its op_id recorded below. That pairing is
     # the poison state: a retry of the identical op loses to the failed
@@ -1828,6 +1845,45 @@ def apply_op(workflow: dict, op: dict, graph) -> dict:
     return workflow
 
 
+def define_subgraph(
+    workflow: dict,
+    definition: dict,
+    *,
+    subgraph_id: str | None = None,
+    actor: str = "cli",
+    base_version: int = 0,
+) -> tuple[dict, dict]:
+    """Emit a definition op; cmp owns semantic validation and application."""
+    if not isinstance(definition, dict):
+        raise ValueError("subgraph definition must be a JSON object")
+    definition = copy.deepcopy(definition)
+    if subgraph_id is not None:
+        definition_id = subgraph_id
+    elif "id" in definition:
+        definition_id = definition["id"]
+    else:
+        definition_id = str(uuid.uuid4())
+    if not isinstance(definition_id, str) or not definition_id:
+        raise ValueError("subgraph definition requires a non-empty string id")
+    if not _SUBGRAPH_UUID_RE.fullmatch(definition_id):
+        raise ValueError("subgraph definition id must be a valid UUID")
+    if "id" in definition and definition["id"] != definition_id:
+        raise ValueError("subgraph definition id must match --id")
+    definition["id"] = definition_id
+    try:
+        json.dumps(definition)
+    except (TypeError, ValueError, RecursionError) as error:
+        raise ValueError("subgraph definition must be JSON-serializable") from error
+    op = _new_op(
+        "define_subgraph",
+        actor,
+        base_version,
+        subgraph_id=definition_id,
+        subgraph_definition=definition,
+    )
+    return workflow, op
+
+
 def _apply_add_node(workflow: dict, op: dict) -> None:
     # Node identity is compared as a STRING everywhere in the apply path
     # (amendment v1.2): ids are legitimately either JSON type, and an exact
@@ -2258,6 +2314,8 @@ def _write_target(op: dict) -> tuple:
             # not share a target with its sibling (``model.reference_videos``).
             return ("input", str(op["to_node"]), "grow", _autogrow_base(str(grow["name"])))
         return ("input", str(op["to_node"]), op["to_slot"])
+    if kind == "define_subgraph":
+        return ("subgraph", str(op["subgraph_id"]))
     return (kind,)
 
 

docs/op-vocabulary-v1.md

@@ -17,7 +17,7 @@ citation must point at a commit on that branch.
 
 ## 1. Frozen op kinds
 
-Six kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with
+Seven kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with
 `ValueError("unknown op ...")` — it never ignores one.
 
 | Kind | Batchable | Standalone command | Summary |
@@ -28,6 +28,7 @@ Six kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with
 | `delete_node` | yes | `comfy workflow delete` | Remove one node and its incident links |
 | `clear` | no | `comfy workflow clear` | Remove every node, link, and group |
 | `reset_doc` | no | `comfy workflow reset-doc --confirm` | Reset the whole document to an empty baseline |
+| `define_subgraph` | yes | `comfy workflow define-subgraph` | Create one subgraph definition |
 
 Batchable = the kind is accepted by `apply_specs` (the `workflow apply` /
 `workflow foreach` batch surface). `clear` and `reset_doc` rewrite the whole
@@ -178,6 +179,32 @@ pre-reset `base_version` do not replay across it.
 * Never emitted implicitly: no `--emit-ops` surface and no bulk writer (§8.8)
   mints one. It exists only where a caller asked for it by name.
 
+### 1.7 `define_subgraph`
+
+Command: `comfy workflow define-subgraph <file> <definition-file> [--id <uuid>]`.
+The command performs envelope shape and JSON-serialization checks, assigns its
+id from `--id`, the definition's `id`, or a new UUID, and emits exactly one
+stamped op without modifying the local workflow:
+
+```json
+{
+  "op": "define_subgraph",
+  "op_id": "<uuid4 hex>",
+  "actor": "cli",
+  "base_version": 0,
+  "stamp": [0, "cli"],
+  "subgraph_id": "<uuid>",
+  "subgraph_definition": {"id": "<uuid>", "nodes": [], "links": []}
+}
+```
+
+Only this op may carry `subgraph_id` or definition payloads. The CLI does not
+semantically validate, apply, replay, project, or resolve conflicts for the
+definition. cmp owns those operations: reusing an id with different content
+returns `definition_conflict`, and references to unknown ids follow cmp's
+unknown-node failure path. The CLI surfaces server errors verbatim. Subsequent
+interior edits use the existing id-addressed op scopes.
+
 ## 2. Idempotency and identity
 
 * Every op carries `op_id`: uuid4 hex, minted by the **creator, before

tests/comfy_cli/command/test_workflow_edit.py

@@ -1385,6 +1385,22 @@ def test_batch_is_atomic_on_failure(self, patched_graph, tmp_path, capsys):
         assert env["error"]["code"] == "workflow_edit_invalid"
         assert path.read_text() == before, "failed batch must not write a partial graph"
 
+    def test_deferred_op_batch_is_not_persisted_locally(self, patched_graph, tmp_path, capsys):
+        path = self._empty(tmp_path)
+        before = path.read_text()
+        ops_path = tmp_path / "ops.json"
+        ops_path.write_text(
+            json.dumps(
+                [{"op": "define_subgraph", "subgraph_definition": {"id": "12345678-1234-4123-8123-123456789abc"}}]
+            )
+        )
+
+        env = _run(["apply", str(path), "--ops", str(ops_path)], capsys)
+
+        assert env["ok"] is False
+        assert "deferred operation" in env["error"]["message"]
+        assert path.read_text() == before
+
     def test_duplicate_alias_is_rejected(self, patched_graph, tmp_path, capsys):
         """A repeated `as` name would silently clobber the earlier node — reject it."""
         path = self._empty(tmp_path)
@@ -1604,6 +1620,27 @@ def test_foreach_bad_param_set_fails(self, patched_graph, tmp_path, capsys):
         assert env["ok"] is False
         assert "positive" in env["error"]["message"]
 
+    def test_foreach_rejects_deferred_ops_before_writing(self, patched_graph, tmp_path, capsys):
+        rp = tmp_path / "deferred.json"
+        rp.write_text(
+            json.dumps(
+                {
+                    "ops": [
+                        {"op": "define_subgraph", "subgraph_definition": {"id": "12345678-1234-4123-8123-123456789abc"}}
+                    ]
+                }
+            )
+        )
+        params = tmp_path / "sets.json"
+        params.write_text("[{}]")
+        out = tmp_path / "out"
+
+        env = _run(["foreach", str(rp), "--params", str(params), "--out-dir", str(out)], capsys)
+
+        assert env["ok"] is False
+        assert "deferred operation" in env["error"]["message"]
+        assert not list(out.glob("*.json"))
+
     def test_foreach_surfaces_partial_writes_on_mid_batch_failure(self, patched_graph, tmp_path, capsys):
         """foreach writes per param-set; a mid-batch failure leaves earlier files
         on disk, so the error must surface them (not leave the caller blind)."""