From b8ca805d859840982bb124213822422309aa4bb8 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 13:21:28 -0700
Subject: [PATCH 1/9] test(workflow): add failing define_subgraph contract
 tests

---
 tests/comfy_cli/test_define_subgraph_op.py | 89 ++++++++++++++++++++++
 1 file changed, 89 insertions(+)
 create mode 100644 tests/comfy_cli/test_define_subgraph_op.py

diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
new file mode 100644
index 000000000..1608b1231
--- /dev/null
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from comfy_cli import workflow_ops
+
+
+SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
+
+
+def _definition(value: int = 1) -> dict:
+    return {
+        "id": SUBGRAPH_ID,
+        "name": "One",
+        "inputs": [],
+        "outputs": [],
+        "nodes": [{"id": 10, "type": "Inner", "widgets_values": [value]}],
+        "links": [],
+    }
+
+
+def test_define_subgraph_emits_cmp_payload_and_inserts_definition():
+    workflow = {"nodes": [], "links": []}
+    definition = _definition()
+    snapshot = copy.deepcopy(definition)
+
+    result, op = workflow_ops.define_subgraph(workflow, definition, actor="agent", base_version=4)
+
+    assert definition == snapshot
+    assert op == {
+        "op": "define_subgraph",
+        "op_id": op["op_id"],
+        "actor": "agent",
+        "base_version": 4,
+        "stamp": [4, "agent"],
+        "subgraph_id": SUBGRAPH_ID,
+        "subgraph_definition": definition,
+    }
+    assert result["definitions"]["subgraphs"] == [definition]
+
+
+@pytest.mark.parametrize(
+    ("definition", "match"),
+    [
+        ([], "JSON object"),
+        ({"nodes": [], "links": []}, "non-empty string id"),
+        ({"id": SUBGRAPH_ID, "nodes": {}, "links": []}, "nodes and links must be arrays"),
+    ],
+)
+def test_define_subgraph_rejects_malformed_input_before_mutation(definition, match):
+    workflow = {"nodes": [], "links": []}
+    before = copy.deepcopy(workflow)
+
+    with pytest.raises(ValueError, match=match):
+        workflow_ops.define_subgraph(workflow, definition)
+
+    assert workflow == before
+
+
+def test_define_subgraph_can_assign_an_explicit_new_id():
+    definition = _definition()
+    definition.pop("id")
+
+    _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition, subgraph_id=SUBGRAPH_ID)
+
+    assert op["subgraph_id"] == SUBGRAPH_ID
+    assert op["subgraph_definition"]["id"] == SUBGRAPH_ID
+
+
+def test_define_subgraph_rejects_existing_id_and_different_definition():
+    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition()]}}
+
+    with pytest.raises(ValueError, match="already exists"):
+        workflow_ops.define_subgraph(workflow, _definition(2))
+
+
+def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_rejected():
+    workflow = {"nodes": [], "links": []}
+    _, op = workflow_ops.define_subgraph(workflow, _definition())
+    before = copy.deepcopy(workflow)
+
+    workflow_ops.apply_op(workflow, op, None)
+    assert workflow == before
+
+    conflicting = {**op, "op_id": "f" * 32, "subgraph_definition": _definition(2)}
+    with pytest.raises(ValueError, match="already exists with different content"):
+        workflow_ops.apply_op(workflow, conflicting, None)

From f4232bd897dffc60d0633124324cac66e41aaf15 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 13:23:29 -0700
Subject: [PATCH 2/9] feat(workflow): emit define_subgraph operations

---
 comfy_cli/command/workflow.py              |  3 +
 comfy_cli/command/workflow_edit.py         | 34 ++++++++
 comfy_cli/workflow_ops.py                  | 90 +++++++++++++++++++++-
 docs/op-vocabulary-v1.md                   | 26 ++++++-
 tests/comfy_cli/test_define_subgraph_op.py |  3 +-
 5 files changed, 151 insertions(+), 5 deletions(-)

diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py
index 850bd918b..602d8fc8f 100644
--- a/comfy_cli/command/workflow.py
+++ b/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)
diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 39e039740..a4ba30345 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -132,6 +132,40 @@ def _graph_or_exit(input_path, host, port, renderer, where=None):
     return _get_graph(input_path, host, port, where=where)
 
 
+# ---------------------------------------------------------------------------
+# define-subgraph
+# ---------------------------------------------------------------------------
+
+
+@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 = json.loads(definition_path.read_text(encoding="utf-8"))
+        workflow, op = workflow_ops.define_subgraph(
+            workflow,
+            definition,
+            subgraph_id=subgraph_id,
+            actor=actor,
+            base_version=base_version,
+        )
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
+        _emit_edit_error(renderer, e, hint="provide a serializable subgraph definition JSON object")
+        raise typer.Exit(code=1) from e
+    _finish(renderer, p, workflow, op, base_version, stdout, "workflow define-subgraph")
+
+
 # ---------------------------------------------------------------------------
 # add-node
 # ---------------------------------------------------------------------------
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 9afa0a262..f87ca6031 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -91,7 +91,15 @@ 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")
+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 yet.
 #: ``apply_op`` must keep rejecting these. Empty since amendment v1.1:
@@ -101,7 +109,7 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
 #: 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
@@ -1751,6 +1759,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
@@ -1795,6 +1811,8 @@ def apply_op(workflow: dict, op: dict, graph) -> dict:
     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
@@ -1813,6 +1831,8 @@ def apply_op(workflow: dict, op: dict, graph) -> dict:
             _apply_clear(workflow, op)
         elif kind == "reset_doc":
             _apply_reset_doc(workflow, op)
+        elif kind == "define_subgraph":
+            _apply_define_subgraph(workflow, op)
         else:
             raise ValueError(f"unknown op {kind!r}")
     except BaseException:
@@ -1828,6 +1848,72 @@ 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]:
+    """Create one subgraph definition and emit the matching cmp op."""
+    if not isinstance(definition, dict):
+        raise ValueError("subgraph definition must be a JSON object")
+    definition = copy.deepcopy(definition)
+    definition_id = subgraph_id or definition.get("id") or str(uuid.uuid4())
+    if not isinstance(definition_id, str) or not definition_id:
+        raise ValueError("subgraph definition requires a non-empty string id")
+    if "id" in definition and definition["id"] != definition_id:
+        raise ValueError("subgraph definition id must match --id")
+    definition["id"] = definition_id
+    if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
+        raise ValueError("subgraph definition nodes and links must be arrays")
+    existing = _subgraph_definition(workflow, definition_id)
+    if existing is not None:
+        raise ValueError(f"subgraph definition {definition_id!r} already exists; define-subgraph only creates new ids")
+    op = _new_op(
+        "define_subgraph",
+        actor,
+        base_version,
+        subgraph_id=definition_id,
+        subgraph_definition=definition,
+    )
+    apply_op(workflow, op, None)
+    return workflow, op
+
+
+def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
+    definitions = workflow.get("definitions")
+    if definitions is None:
+        return None
+    if not isinstance(definitions, dict) or not isinstance(definitions.get("subgraphs", []), list):
+        raise ValueError("malformed workflow: definitions.subgraphs must be an array")
+    for definition in definitions.get("subgraphs", []):
+        if isinstance(definition, dict) and str(definition.get("id")) == subgraph_id:
+            return definition
+    return None
+
+
+def _apply_define_subgraph(workflow: dict, op: dict) -> None:
+    subgraph_id = op.get("subgraph_id")
+    definition = op.get("subgraph_definition")
+    if (
+        not isinstance(subgraph_id, str)
+        or not subgraph_id
+        or not isinstance(definition, dict)
+        or definition.get("id") != subgraph_id
+        or not isinstance(definition.get("nodes"), list)
+        or not isinstance(definition.get("links"), list)
+    ):
+        raise ValueError("malformed_op: subgraph_id must match a definition with nodes and links arrays")
+    existing = _subgraph_definition(workflow, subgraph_id)
+    if existing is not None:
+        if existing == definition:
+            return
+        raise ValueError(f"malformed_op: definition {subgraph_id!r} already exists with different content")
+    workflow.setdefault("definitions", {}).setdefault("subgraphs", []).append(copy.deepcopy(definition))
+
+
 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
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 69edcf4a0..00bb081cd 100644
--- a/docs/op-vocabulary-v1.md
+++ b/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,29 @@ 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 validates a serializable definition, assigns its id from `--id`,
+the definition's `id`, or a new UUID, and emits exactly one stamped op:
+
+```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. Creation fails
+before writing when the id already exists. Exact op replay is a no-op; a
+different definition under an existing id is rejected as `malformed_op`.
+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
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index 1608b1231..09251ce24 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -6,7 +6,6 @@
 
 from comfy_cli import workflow_ops
 
-
 SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
 
 
@@ -45,7 +44,7 @@ def test_define_subgraph_emits_cmp_payload_and_inserts_definition():
     ("definition", "match"),
     [
         ([], "JSON object"),
-        ({"nodes": [], "links": []}, "non-empty string id"),
+        ({"id": 7, "nodes": [], "links": []}, "non-empty string id"),
         ({"id": SUBGRAPH_ID, "nodes": {}, "links": []}, "nodes and links must be arrays"),
     ],
 )

From 3d67e35b7970055c1778153804811d1a17e50473 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 13:46:11 -0700
Subject: [PATCH 3/9] fix(workflow): validate subgraph definition UUIDs

---
 comfy_cli/workflow_ops.py                  | 31 ++++++++++++++++++++--
 tests/comfy_cli/test_define_subgraph_op.py | 22 +++++++++++++++
 2 files changed, 51 insertions(+), 2 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index f87ca6031..250277eb3 100644
--- a/comfy_cli/workflow_ops.py
+++ b/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
+_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:
@@ -1863,11 +1864,12 @@ def define_subgraph(
     definition_id = subgraph_id or definition.get("id") or 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 _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
-    if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
-        raise ValueError("subgraph definition nodes and links must be arrays")
+    _validate_subgraph_definition(definition)
     existing = _subgraph_definition(workflow, definition_id)
     if existing is not None:
         raise ValueError(f"subgraph definition {definition_id!r} already exists; define-subgraph only creates new ids")
@@ -1882,6 +1884,31 @@ def define_subgraph(
     return workflow, op
 
 
+def _validate_subgraph_definition(definition: dict, path: str = "subgraph definition") -> None:
+    """Validate definition ids and containers while preserving its serialized shape."""
+    definition_id = definition.get("id")
+    if not isinstance(definition_id, str) or not _UUID_RE.fullmatch(definition_id):
+        raise ValueError(f"{path} id must be a valid UUID")
+    if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
+        raise ValueError(f"{path} nodes and links must be arrays")
+    nested = definition.get("definitions")
+    if nested is None:
+        return
+    if not isinstance(nested, dict) or not isinstance(nested.get("subgraphs"), list):
+        raise ValueError(f"{path}.definitions.subgraphs must be an array")
+    seen: set[str] = set()
+    for index, child in enumerate(nested["subgraphs"]):
+        child_path = f"{path}.definitions.subgraphs[{index}]"
+        if not isinstance(child, dict):
+            raise ValueError(f"{child_path} must be a JSON object")
+        child_id = child.get("id")
+        if isinstance(child_id, str) and child_id in seen:
+            raise ValueError(f"{child_path} duplicates subgraph definition id {child_id!r}")
+        if isinstance(child_id, str):
+            seen.add(child_id)
+        _validate_subgraph_definition(child, child_path)
+
+
 def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
     definitions = workflow.get("definitions")
     if definitions is None:
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index 09251ce24..7f3955ef5 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -7,6 +7,7 @@
 from comfy_cli import workflow_ops
 
 SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
+NESTED_ID = "abcdefab-cdef-4abc-8def-abcdefabcdef"
 
 
 def _definition(value: int = 1) -> dict:
@@ -45,7 +46,17 @@ def test_define_subgraph_emits_cmp_payload_and_inserts_definition():
     [
         ([], "JSON object"),
         ({"id": 7, "nodes": [], "links": []}, "non-empty string id"),
+        ({"id": "not-a-uuid", "nodes": [], "links": []}, "valid UUID"),
         ({"id": SUBGRAPH_ID, "nodes": {}, "links": []}, "nodes and links must be arrays"),
+        (
+            {
+                "id": SUBGRAPH_ID,
+                "nodes": [],
+                "links": [],
+                "definitions": {"subgraphs": [{"id": "not-a-uuid", "nodes": [], "links": []}]},
+            },
+            "definitions.subgraphs\\[0\\].*valid UUID",
+        ),
     ],
 )
 def test_define_subgraph_rejects_malformed_input_before_mutation(definition, match):
@@ -75,6 +86,17 @@ def test_define_subgraph_rejects_existing_id_and_different_definition():
         workflow_ops.define_subgraph(workflow, _definition(2))
 
 
+def test_define_subgraph_preserves_nested_definitions_inside_single_parent_op():
+    nested = {"id": NESTED_ID, "nodes": [], "links": []}
+    definition = {**_definition(), "definitions": {"subgraphs": [nested]}}
+
+    result, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition)
+
+    assert op["subgraph_definition"]["definitions"] == {"subgraphs": [nested]}
+    assert result["definitions"]["subgraphs"] == [definition]
+    assert op["op"] == "define_subgraph"
+
+
 def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_rejected():
     workflow = {"nodes": [], "links": []}
     _, op = workflow_ops.define_subgraph(workflow, _definition())

From 5ea8866fccc95cfc0b6460ecea7f274387da2079 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 14:41:39 -0700
Subject: [PATCH 4/9] Fix recursive subgraph definition validation

---
 comfy_cli/workflow_ops.py                  | 27 +++++-----
 tests/comfy_cli/test_define_subgraph_op.py | 61 +++++++++++++++++++++-
 2 files changed, 75 insertions(+), 13 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 250277eb3..002fe1165 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1808,7 +1808,7 @@ 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"]
@@ -1884,11 +1884,18 @@ def define_subgraph(
     return workflow, op
 
 
-def _validate_subgraph_definition(definition: dict, path: str = "subgraph definition") -> None:
+def _validate_subgraph_definition(
+    definition: dict, path: str = "subgraph definition", seen: set[str] | None = None
+) -> None:
     """Validate definition ids and containers while preserving its serialized shape."""
+    if seen is None:
+        seen = set()
     definition_id = definition.get("id")
     if not isinstance(definition_id, str) or not _UUID_RE.fullmatch(definition_id):
         raise ValueError(f"{path} id must be a valid UUID")
+    if definition_id in seen:
+        raise ValueError(f"{path} duplicates subgraph definition id {definition_id!r}")
+    seen.add(definition_id)
     if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
         raise ValueError(f"{path} nodes and links must be arrays")
     nested = definition.get("definitions")
@@ -1896,17 +1903,11 @@ def _validate_subgraph_definition(definition: dict, path: str = "subgraph defini
         return
     if not isinstance(nested, dict) or not isinstance(nested.get("subgraphs"), list):
         raise ValueError(f"{path}.definitions.subgraphs must be an array")
-    seen: set[str] = set()
     for index, child in enumerate(nested["subgraphs"]):
         child_path = f"{path}.definitions.subgraphs[{index}]"
         if not isinstance(child, dict):
             raise ValueError(f"{child_path} must be a JSON object")
-        child_id = child.get("id")
-        if isinstance(child_id, str) and child_id in seen:
-            raise ValueError(f"{child_path} duplicates subgraph definition id {child_id!r}")
-        if isinstance(child_id, str):
-            seen.add(child_id)
-        _validate_subgraph_definition(child, child_path)
+        _validate_subgraph_definition(child, child_path, seen)
 
 
 def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
@@ -1929,15 +1930,17 @@ def _apply_define_subgraph(workflow: dict, op: dict) -> None:
         or not subgraph_id
         or not isinstance(definition, dict)
         or definition.get("id") != subgraph_id
-        or not isinstance(definition.get("nodes"), list)
-        or not isinstance(definition.get("links"), list)
     ):
         raise ValueError("malformed_op: subgraph_id must match a definition with nodes and links arrays")
+    try:
+        _validate_subgraph_definition(definition)
+    except ValueError as error:
+        raise ValueError(f"malformed_op: {error}") from error
     existing = _subgraph_definition(workflow, subgraph_id)
     if existing is not None:
         if existing == definition:
             return
-        raise ValueError(f"malformed_op: definition {subgraph_id!r} already exists with different content")
+        raise ValueError(f"definition_conflict: definition {subgraph_id!r} already exists with different content")
     workflow.setdefault("definitions", {}).setdefault("subgraphs", []).append(copy.deepcopy(definition))
 
 
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index 7f3955ef5..a8f8bb5bd 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -8,6 +8,7 @@
 
 SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
 NESTED_ID = "abcdefab-cdef-4abc-8def-abcdefabcdef"
+OTHER_NESTED_ID = "fedcbafe-dcba-4fed-8cba-fedcbafedcba"
 
 
 def _definition(value: int = 1) -> dict:
@@ -97,6 +98,63 @@ def test_define_subgraph_preserves_nested_definitions_inside_single_parent_op():
     assert op["op"] == "define_subgraph"
 
 
+@pytest.mark.parametrize(
+    "definition",
+    [
+        {
+            **_definition(),
+            "definitions": {"subgraphs": [{"id": SUBGRAPH_ID, "nodes": [], "links": []}]},
+        },
+        {
+            **_definition(),
+            "definitions": {
+                "subgraphs": [
+                    {
+                        "id": NESTED_ID,
+                        "nodes": [],
+                        "links": [],
+                        "definitions": {"subgraphs": [{"id": OTHER_NESTED_ID, "nodes": [], "links": []}]},
+                    },
+                    {"id": OTHER_NESTED_ID, "nodes": [], "links": []},
+                ]
+            },
+        },
+    ],
+    ids=["ancestor", "across-branches"],
+)
+def test_define_subgraph_rejects_duplicate_ids_across_entire_definition_tree(definition):
+    workflow = {"nodes": [], "links": []}
+    before = copy.deepcopy(workflow)
+
+    with pytest.raises(ValueError, match="duplicates subgraph definition id"):
+        workflow_ops.define_subgraph(workflow, definition)
+
+    assert workflow == before
+
+
+def test_apply_define_subgraph_rejects_malformed_nested_definition_atomically():
+    workflow = {"nodes": [], "links": []}
+    before = copy.deepcopy(workflow)
+    definition = {
+        **_definition(),
+        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": {}, "links": []}]},
+    }
+    op = {
+        "op": "define_subgraph",
+        "op_id": "a" * 32,
+        "actor": "peer",
+        "base_version": 0,
+        "stamp": [0, "peer"],
+        "subgraph_id": SUBGRAPH_ID,
+        "subgraph_definition": definition,
+    }
+
+    with pytest.raises(ValueError, match="malformed_op:.*definitions.subgraphs\\[0\\].*nodes and links"):
+        workflow_ops.apply_op(workflow, op, None)
+
+    assert workflow == before
+
+
 def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_rejected():
     workflow = {"nodes": [], "links": []}
     _, op = workflow_ops.define_subgraph(workflow, _definition())
@@ -106,5 +164,6 @@ def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_reject
     assert workflow == before
 
     conflicting = {**op, "op_id": "f" * 32, "subgraph_definition": _definition(2)}
-    with pytest.raises(ValueError, match="already exists with different content"):
+    with pytest.raises(ValueError, match="definition_conflict:.*already exists with different content"):
         workflow_ops.apply_op(workflow, conflicting, None)
+    assert workflow == before

From 83a004fa43f5cafebd850d4ec80ca0486846a165 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 14:43:00 -0700
Subject: [PATCH 5/9] Reject nested definition id collisions

---
 comfy_cli/workflow_ops.py                  | 27 ++++++++++++++++++++--
 tests/comfy_cli/test_define_subgraph_op.py | 20 ++++++++++++++++
 2 files changed, 45 insertions(+), 2 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 002fe1165..71c86225d 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1873,6 +1873,7 @@ def define_subgraph(
     existing = _subgraph_definition(workflow, definition_id)
     if existing is not None:
         raise ValueError(f"subgraph definition {definition_id!r} already exists; define-subgraph only creates new ids")
+    _validate_subgraph_definition(definition, seen=_subgraph_definition_ids(workflow))
     op = _new_op(
         "define_subgraph",
         actor,
@@ -1917,11 +1918,33 @@ def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
     if not isinstance(definitions, dict) or not isinstance(definitions.get("subgraphs", []), list):
         raise ValueError("malformed workflow: definitions.subgraphs must be an array")
     for definition in definitions.get("subgraphs", []):
-        if isinstance(definition, dict) and str(definition.get("id")) == subgraph_id:
-            return definition
+        if isinstance(definition, dict):
+            if str(definition.get("id")) == subgraph_id:
+                return definition
+            nested = _subgraph_definition(definition, subgraph_id)
+            if nested is not None:
+                return nested
     return None
 
 
+def _subgraph_definition_ids(workflow: dict) -> set[str]:
+    """Collect all definition ids recursively from a workflow or definition."""
+    definitions = workflow.get("definitions")
+    if definitions is None:
+        return set()
+    if not isinstance(definitions, dict) or not isinstance(definitions.get("subgraphs", []), list):
+        raise ValueError("malformed workflow: definitions.subgraphs must be an array")
+    ids: set[str] = set()
+    for definition in definitions.get("subgraphs", []):
+        if not isinstance(definition, dict):
+            continue
+        definition_id = definition.get("id")
+        if isinstance(definition_id, str):
+            ids.add(definition_id)
+        ids.update(_subgraph_definition_ids(definition))
+    return ids
+
+
 def _apply_define_subgraph(workflow: dict, op: dict) -> None:
     subgraph_id = op.get("subgraph_id")
     definition = op.get("subgraph_definition")
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index a8f8bb5bd..9dcf61886 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -132,6 +132,26 @@ def test_define_subgraph_rejects_duplicate_ids_across_entire_definition_tree(def
     assert workflow == before
 
 
+def test_define_subgraph_rejects_id_already_nested_in_workflow():
+    existing = {
+        "id": OTHER_NESTED_ID,
+        "nodes": [],
+        "links": [],
+        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": [], "links": []}]},
+    }
+    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [existing]}}
+    before = copy.deepcopy(workflow)
+    definition = {
+        **_definition(),
+        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": [], "links": []}]},
+    }
+
+    with pytest.raises(ValueError, match="duplicates subgraph definition id"):
+        workflow_ops.define_subgraph(workflow, definition)
+
+    assert workflow == before
+
+
 def test_apply_define_subgraph_rejects_malformed_nested_definition_atomically():
     workflow = {"nodes": [], "links": []}
     before = copy.deepcopy(workflow)

From b7da23b75addc74404fed63d054120af3e7b9bc7 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 15:16:09 -0700
Subject: [PATCH 6/9] fix: harden define-subgraph replay and validation

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282753
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282758
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282777
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282782
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282787
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282792
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282798
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282802
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993282806
---
 comfy_cli/command/workflow_edit.py         |  23 ++++-
 comfy_cli/workflow_ops.py                  | 100 ++++++++++++++++--
 tests/comfy_cli/test_define_subgraph_op.py | 114 ++++++++++++++++++++-
 3 files changed, 220 insertions(+), 17 deletions(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index a4ba30345..15c264dfc 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/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:
@@ -137,6 +139,23 @@ def _graph_or_exit(input_path, host, port, renderer, where=None):
 # ---------------------------------------------------------------------------
 
 
+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.")],
@@ -152,7 +171,7 @@ def define_subgraph_cmd(
     p, workflow = _load_workflow_or_fail(renderer, file)
     try:
         definition_path = Path(definition_file).expanduser()
-        definition = json.loads(definition_path.read_text(encoding="utf-8"))
+        definition = _read_subgraph_definition(definition_path)
         workflow, op = workflow_ops.define_subgraph(
             workflow,
             definition,
@@ -160,7 +179,7 @@ def define_subgraph_cmd(
             actor=actor,
             base_version=base_version,
         )
-    except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
+    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
     _finish(renderer, p, workflow, op, base_version, stdout, "workflow define-subgraph")
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 71c86225d..8febf76f1 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1140,18 +1140,14 @@ 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"
     extra = workflow.get("extra")
@@ -1227,6 +1223,39 @@ def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int =
         raise NotExpressibleError(reason)
 
     ops: list[dict] = []
+    old_definitions = {
+        str(definition.get("id")): definition
+        for definition in ((old.get("definitions") or {}).get("subgraphs") or [])
+        if isinstance(definition, dict)
+    }
+    new_definitions = {
+        str(definition.get("id")): definition
+        for definition in ((new.get("definitions") or {}).get("subgraphs") or [])
+        if isinstance(definition, dict)
+    }
+    if any(identifier not in new_definitions for identifier in old_definitions):
+        raise NotExpressibleError("the replacement removes a subgraph definition, which no frozen op kind can delete")
+    for identifier, definition in new_definitions.items():
+        existing = old_definitions.get(identifier)
+        if existing is not None and existing != definition:
+            raise NotExpressibleError(
+                "the replacement changes an existing subgraph definition; emit id-addressed edits instead"
+            )
+        if existing is None:
+            try:
+                _validate_subgraph_definition(definition)
+                _validate_subgraph_cycles(definition)
+            except ValueError as error:
+                raise NotExpressibleError(f"the workflow contains a malformed subgraph definition: {error}") from error
+            ops.append(
+                _new_op(
+                    "define_subgraph",
+                    actor,
+                    base_version,
+                    subgraph_id=identifier,
+                    subgraph_definition=copy.deepcopy(definition),
+                )
+            )
     old_links = [link for link in (old.get("links") or []) if isinstance(link, list) and len(link) >= 5]
     for node in old.get("nodes") or []:
         if not isinstance(node, dict) or node.get("id") is None:
@@ -1861,7 +1890,12 @@ def define_subgraph(
     if not isinstance(definition, dict):
         raise ValueError("subgraph definition must be a JSON object")
     definition = copy.deepcopy(definition)
-    definition_id = subgraph_id or definition.get("id") or str(uuid.uuid4())
+    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 _UUID_RE.fullmatch(definition_id):
@@ -1870,6 +1904,7 @@ def define_subgraph(
         raise ValueError("subgraph definition id must match --id")
     definition["id"] = definition_id
     _validate_subgraph_definition(definition)
+    _validate_subgraph_cycles(definition)
     existing = _subgraph_definition(workflow, definition_id)
     if existing is not None:
         raise ValueError(f"subgraph definition {definition_id!r} already exists; define-subgraph only creates new ids")
@@ -1899,6 +1934,16 @@ def _validate_subgraph_definition(
     seen.add(definition_id)
     if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
         raise ValueError(f"{path} nodes and links must be arrays")
+    for index, node in enumerate(definition["nodes"]):
+        node_path = f"{path}.nodes[{index}]"
+        if not isinstance(node, dict) or node.get("id") is None or not isinstance(node.get("type"), str):
+            raise ValueError(f"{node_path} must be an object with id and string type")
+        for field in ("inputs", "outputs"):
+            if field in node and not isinstance(node[field], list):
+                raise ValueError(f"{node_path}.{field} must be an array")
+    for index, link in enumerate(definition["links"]):
+        if not isinstance(link, list) or len(link) < 5:
+            raise ValueError(f"{path}.links[{index}] must be a link tuple with at least 5 items")
     nested = definition.get("definitions")
     if nested is None:
         return
@@ -1911,6 +1956,36 @@ def _validate_subgraph_definition(
         _validate_subgraph_definition(child, child_path, seen)
 
 
+def _validate_subgraph_cycles(root: dict) -> None:
+    """Reject recursive definition references that expansion cannot terminate."""
+    definitions: dict[str, dict] = {}
+
+    def collect(definition: dict) -> None:
+        definitions[definition["id"]] = definition
+        for child in (definition.get("definitions") or {}).get("subgraphs") or []:
+            collect(child)
+
+    collect(root)
+    visiting: set[str] = set()
+    visited: set[str] = set()
+
+    def visit(identifier: str) -> None:
+        if identifier in visiting:
+            raise ValueError(f"cyclic subgraph reference involving {identifier!r}")
+        if identifier in visited:
+            return
+        visiting.add(identifier)
+        for node in definitions[identifier]["nodes"]:
+            target = node.get("type")
+            if target in definitions:
+                visit(target)
+        visiting.remove(identifier)
+        visited.add(identifier)
+
+    for identifier in definitions:
+        visit(identifier)
+
+
 def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
     definitions = workflow.get("definitions")
     if definitions is None:
@@ -1957,14 +2032,15 @@ def _apply_define_subgraph(workflow: dict, op: dict) -> None:
         raise ValueError("malformed_op: subgraph_id must match a definition with nodes and links arrays")
     try:
         _validate_subgraph_definition(definition)
+        _validate_subgraph_cycles(definition)
     except ValueError as error:
         raise ValueError(f"malformed_op: {error}") from error
     existing = _subgraph_definition(workflow, subgraph_id)
     if existing is not None:
-        if existing == definition:
-            return
-        raise ValueError(f"definition_conflict: definition {subgraph_id!r} already exists with different content")
-    workflow.setdefault("definitions", {}).setdefault("subgraphs", []).append(copy.deepcopy(definition))
+        return
+    if workflow.get("definitions") is None:
+        workflow["definitions"] = {}
+    workflow["definitions"].setdefault("subgraphs", []).append(copy.deepcopy(definition))
 
 
 def _apply_add_node(workflow: dict, op: dict) -> None:
@@ -2397,6 +2473,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,)
 
 
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index 9dcf61886..ad07919e7 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -1,10 +1,15 @@
 from __future__ import annotations
 
 import copy
+import json
+from pathlib import Path
 
 import pytest
+import typer
 
 from comfy_cli import workflow_ops
+from comfy_cli.command import workflow as workflow_cmd  # noqa: F401 -- initializes the edit command cycle
+from comfy_cli.command import workflow_edit
 
 SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
 NESTED_ID = "abcdefab-cdef-4abc-8def-abcdefabcdef"
@@ -175,7 +180,7 @@ def test_apply_define_subgraph_rejects_malformed_nested_definition_atomically():
     assert workflow == before
 
 
-def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_rejected():
+def test_apply_define_subgraph_replays_are_idempotent():
     workflow = {"nodes": [], "links": []}
     _, op = workflow_ops.define_subgraph(workflow, _definition())
     before = copy.deepcopy(workflow)
@@ -184,6 +189,107 @@ def test_apply_define_subgraph_exact_replay_is_idempotent_and_conflict_is_reject
     assert workflow == before
 
     conflicting = {**op, "op_id": "f" * 32, "subgraph_definition": _definition(2)}
-    with pytest.raises(ValueError, match="definition_conflict:.*already exists with different content"):
-        workflow_ops.apply_op(workflow, conflicting, None)
-    assert workflow == before
+    workflow_ops.apply_op(workflow, conflicting, None)
+    assert workflow["definitions"] == before["definitions"]
+
+
+def test_define_subgraph_write_targets_are_scoped_by_definition_id():
+    first = {"op": "define_subgraph", "subgraph_id": SUBGRAPH_ID}
+    other = {"op": "define_subgraph", "subgraph_id": NESTED_ID}
+
+    assert workflow_ops.detect_conflict(first, other) is False
+    assert workflow_ops.detect_conflict(first, dict(first)) is True
+
+
+def test_apply_define_subgraph_accepts_null_definitions_container():
+    workflow = {"nodes": [], "links": [], "definitions": None}
+    _, op = workflow_ops.define_subgraph(workflow, _definition())
+
+    assert workflow["definitions"]["subgraphs"] == [op["subgraph_definition"]]
+
+
+@pytest.mark.parametrize(
+    ("definition", "match"),
+    [
+        ({**_definition(), "nodes": [{"id": 1, "type": "Inner", "inputs": 1}]}, "inputs must be an array"),
+        ({**_definition(), "links": [[1, 2]]}, "link.*tuple"),
+        ({**_definition(), "nodes": [{"id": 1, "type": SUBGRAPH_ID}]}, "cyclic subgraph reference"),
+        (
+            {
+                **_definition(),
+                "nodes": [{"id": 1, "type": NESTED_ID}],
+                "definitions": {
+                    "subgraphs": [{"id": NESTED_ID, "nodes": [{"id": 2, "type": SUBGRAPH_ID}], "links": []}]
+                },
+            },
+            "cyclic subgraph reference",
+        ),
+    ],
+)
+def test_define_subgraph_rejects_malformed_or_recursive_interior(definition, match):
+    with pytest.raises(ValueError, match=match):
+        workflow_ops.define_subgraph({"nodes": [], "links": []}, definition)
+
+
+def test_define_subgraph_preserves_explicit_falsy_ids_for_validation():
+    with pytest.raises(ValueError, match="non-empty string id"):
+        workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition(), subgraph_id="")
+    with pytest.raises(ValueError, match="non-empty string id"):
+        workflow_ops.define_subgraph({"nodes": [], "links": []}, {"id": 0, "nodes": [], "links": []})
+
+
+def test_apply_define_subgraph_different_redelivery_is_a_noop():
+    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition(2)]}}
+    before = copy.deepcopy(workflow)
+    _, original_op = workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition())
+
+    workflow_ops.apply_op(workflow, original_op, None)
+
+    assert workflow["definitions"] == before["definitions"]
+    assert original_op["op_id"] in workflow["_applied_ops"]
+
+
+def test_apply_define_subgraph_fresh_op_id_exercises_definition_idempotency():
+    workflow = {"nodes": [], "links": []}
+    _, op = workflow_ops.define_subgraph(workflow, _definition())
+    workflow_ops.strip_internal(workflow)
+
+    workflow_ops.apply_op(workflow, {**op, "op_id": "e" * 32}, None)
+
+    assert workflow["definitions"]["subgraphs"] == [_definition()]
+
+
+def test_replace_ops_emits_definition_before_nodes():
+    new = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition()]}}
+
+    ops = workflow_ops.replace_ops({"nodes": [], "links": []}, new)
+
+    assert [op["op"] for op in ops] == ["define_subgraph"]
+
+
+def test_define_subgraph_command_rejects_non_regular_definition_file(tmp_path, monkeypatch):
+    workflow = tmp_path / "workflow.json"
+    workflow.write_text(json.dumps({"nodes": [], "links": []}))
+    definition_dir = tmp_path / "definition"
+    definition_dir.mkdir()
+    errors = []
+
+    class Renderer:
+        command = ""
+
+        def error(self, **kwargs):
+            errors.append(kwargs)
+
+    monkeypatch.setattr(workflow_edit, "get_renderer", Renderer)
+    with pytest.raises(typer.Exit):
+        workflow_edit.define_subgraph_cmd(str(workflow), str(definition_dir))
+
+    assert "regular file" in errors[0]["message"]
+
+
+def test_definition_reader_rejects_oversized_files(tmp_path):
+    path = tmp_path / "large.json"
+    path.write_bytes(b" " * (workflow_edit._MAX_DEFINITION_BYTES + 1))
+
+    with pytest.raises(ValueError, match="too large"):
+        workflow_edit._read_subgraph_definition(Path(path))

From ac378c63050eedcaa196dabfbb58aeed590c395a Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 15:56:26 -0700
Subject: [PATCH 7/9] refactor(define_subgraph): make CLI emit-only per cmp
 ownership contract

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993917406
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993917409
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993917413
https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r3993917416
---
 comfy_cli/command/workflow_edit.py         |  12 +-
 comfy_cli/workflow_ops.py                  | 181 +--------------
 docs/op-vocabulary-v1.md                   |  15 +-
 tests/comfy_cli/test_define_subgraph_op.py | 248 ++++-----------------
 4 files changed, 69 insertions(+), 387 deletions(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 15c264dfc..8d0348d92 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -130,6 +130,14 @@ 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)
 
@@ -172,7 +180,7 @@ def define_subgraph_cmd(
     try:
         definition_path = Path(definition_file).expanduser()
         definition = _read_subgraph_definition(definition_path)
-        workflow, op = workflow_ops.define_subgraph(
+        _, op = workflow_ops.define_subgraph(
             workflow,
             definition,
             subgraph_id=subgraph_id,
@@ -182,7 +190,7 @@ def define_subgraph_cmd(
     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
-    _finish(renderer, p, workflow, op, base_version, stdout, "workflow define-subgraph")
+    _emit_op(renderer, p, op, base_version, "workflow define-subgraph")
 
 
 # ---------------------------------------------------------------------------
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 8febf76f1..fb8731969 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -102,10 +102,10 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
     "define_subgraph",
 )
 
-#: 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, ...] = ()
+#: 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
@@ -1150,6 +1150,8 @@ def _inexpressible_reason(workflow: dict) -> str | None:
         return "not a frontend-format workflow (no `nodes` list) — only the save/UI format can be op-ified"
     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"
@@ -1223,39 +1225,6 @@ def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int =
         raise NotExpressibleError(reason)
 
     ops: list[dict] = []
-    old_definitions = {
-        str(definition.get("id")): definition
-        for definition in ((old.get("definitions") or {}).get("subgraphs") or [])
-        if isinstance(definition, dict)
-    }
-    new_definitions = {
-        str(definition.get("id")): definition
-        for definition in ((new.get("definitions") or {}).get("subgraphs") or [])
-        if isinstance(definition, dict)
-    }
-    if any(identifier not in new_definitions for identifier in old_definitions):
-        raise NotExpressibleError("the replacement removes a subgraph definition, which no frozen op kind can delete")
-    for identifier, definition in new_definitions.items():
-        existing = old_definitions.get(identifier)
-        if existing is not None and existing != definition:
-            raise NotExpressibleError(
-                "the replacement changes an existing subgraph definition; emit id-addressed edits instead"
-            )
-        if existing is None:
-            try:
-                _validate_subgraph_definition(definition)
-                _validate_subgraph_cycles(definition)
-            except ValueError as error:
-                raise NotExpressibleError(f"the workflow contains a malformed subgraph definition: {error}") from error
-            ops.append(
-                _new_op(
-                    "define_subgraph",
-                    actor,
-                    base_version,
-                    subgraph_id=identifier,
-                    subgraph_definition=copy.deepcopy(definition),
-                )
-            )
     old_links = [link for link in (old.get("links") or []) if isinstance(link, list) and len(link) >= 5]
     for node in old.get("nodes") or []:
         if not isinstance(node, dict) or node.get("id") is None:
@@ -1861,8 +1830,6 @@ def apply_op(workflow: dict, op: dict, graph) -> dict:
             _apply_clear(workflow, op)
         elif kind == "reset_doc":
             _apply_reset_doc(workflow, op)
-        elif kind == "define_subgraph":
-            _apply_define_subgraph(workflow, op)
         else:
             raise ValueError(f"unknown op {kind!r}")
     except BaseException:
@@ -1886,7 +1853,7 @@ def define_subgraph(
     actor: str = "cli",
     base_version: int = 0,
 ) -> tuple[dict, dict]:
-    """Create one subgraph definition and emit the matching cmp op."""
+    """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)
@@ -1903,12 +1870,10 @@ def define_subgraph(
     if "id" in definition and definition["id"] != definition_id:
         raise ValueError("subgraph definition id must match --id")
     definition["id"] = definition_id
-    _validate_subgraph_definition(definition)
-    _validate_subgraph_cycles(definition)
-    existing = _subgraph_definition(workflow, definition_id)
-    if existing is not None:
-        raise ValueError(f"subgraph definition {definition_id!r} already exists; define-subgraph only creates new ids")
-    _validate_subgraph_definition(definition, seen=_subgraph_definition_ids(workflow))
+    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,
@@ -1916,133 +1881,9 @@ def define_subgraph(
         subgraph_id=definition_id,
         subgraph_definition=definition,
     )
-    apply_op(workflow, op, None)
     return workflow, op
 
 
-def _validate_subgraph_definition(
-    definition: dict, path: str = "subgraph definition", seen: set[str] | None = None
-) -> None:
-    """Validate definition ids and containers while preserving its serialized shape."""
-    if seen is None:
-        seen = set()
-    definition_id = definition.get("id")
-    if not isinstance(definition_id, str) or not _UUID_RE.fullmatch(definition_id):
-        raise ValueError(f"{path} id must be a valid UUID")
-    if definition_id in seen:
-        raise ValueError(f"{path} duplicates subgraph definition id {definition_id!r}")
-    seen.add(definition_id)
-    if not isinstance(definition.get("nodes"), list) or not isinstance(definition.get("links"), list):
-        raise ValueError(f"{path} nodes and links must be arrays")
-    for index, node in enumerate(definition["nodes"]):
-        node_path = f"{path}.nodes[{index}]"
-        if not isinstance(node, dict) or node.get("id") is None or not isinstance(node.get("type"), str):
-            raise ValueError(f"{node_path} must be an object with id and string type")
-        for field in ("inputs", "outputs"):
-            if field in node and not isinstance(node[field], list):
-                raise ValueError(f"{node_path}.{field} must be an array")
-    for index, link in enumerate(definition["links"]):
-        if not isinstance(link, list) or len(link) < 5:
-            raise ValueError(f"{path}.links[{index}] must be a link tuple with at least 5 items")
-    nested = definition.get("definitions")
-    if nested is None:
-        return
-    if not isinstance(nested, dict) or not isinstance(nested.get("subgraphs"), list):
-        raise ValueError(f"{path}.definitions.subgraphs must be an array")
-    for index, child in enumerate(nested["subgraphs"]):
-        child_path = f"{path}.definitions.subgraphs[{index}]"
-        if not isinstance(child, dict):
-            raise ValueError(f"{child_path} must be a JSON object")
-        _validate_subgraph_definition(child, child_path, seen)
-
-
-def _validate_subgraph_cycles(root: dict) -> None:
-    """Reject recursive definition references that expansion cannot terminate."""
-    definitions: dict[str, dict] = {}
-
-    def collect(definition: dict) -> None:
-        definitions[definition["id"]] = definition
-        for child in (definition.get("definitions") or {}).get("subgraphs") or []:
-            collect(child)
-
-    collect(root)
-    visiting: set[str] = set()
-    visited: set[str] = set()
-
-    def visit(identifier: str) -> None:
-        if identifier in visiting:
-            raise ValueError(f"cyclic subgraph reference involving {identifier!r}")
-        if identifier in visited:
-            return
-        visiting.add(identifier)
-        for node in definitions[identifier]["nodes"]:
-            target = node.get("type")
-            if target in definitions:
-                visit(target)
-        visiting.remove(identifier)
-        visited.add(identifier)
-
-    for identifier in definitions:
-        visit(identifier)
-
-
-def _subgraph_definition(workflow: dict, subgraph_id: str) -> dict | None:
-    definitions = workflow.get("definitions")
-    if definitions is None:
-        return None
-    if not isinstance(definitions, dict) or not isinstance(definitions.get("subgraphs", []), list):
-        raise ValueError("malformed workflow: definitions.subgraphs must be an array")
-    for definition in definitions.get("subgraphs", []):
-        if isinstance(definition, dict):
-            if str(definition.get("id")) == subgraph_id:
-                return definition
-            nested = _subgraph_definition(definition, subgraph_id)
-            if nested is not None:
-                return nested
-    return None
-
-
-def _subgraph_definition_ids(workflow: dict) -> set[str]:
-    """Collect all definition ids recursively from a workflow or definition."""
-    definitions = workflow.get("definitions")
-    if definitions is None:
-        return set()
-    if not isinstance(definitions, dict) or not isinstance(definitions.get("subgraphs", []), list):
-        raise ValueError("malformed workflow: definitions.subgraphs must be an array")
-    ids: set[str] = set()
-    for definition in definitions.get("subgraphs", []):
-        if not isinstance(definition, dict):
-            continue
-        definition_id = definition.get("id")
-        if isinstance(definition_id, str):
-            ids.add(definition_id)
-        ids.update(_subgraph_definition_ids(definition))
-    return ids
-
-
-def _apply_define_subgraph(workflow: dict, op: dict) -> None:
-    subgraph_id = op.get("subgraph_id")
-    definition = op.get("subgraph_definition")
-    if (
-        not isinstance(subgraph_id, str)
-        or not subgraph_id
-        or not isinstance(definition, dict)
-        or definition.get("id") != subgraph_id
-    ):
-        raise ValueError("malformed_op: subgraph_id must match a definition with nodes and links arrays")
-    try:
-        _validate_subgraph_definition(definition)
-        _validate_subgraph_cycles(definition)
-    except ValueError as error:
-        raise ValueError(f"malformed_op: {error}") from error
-    existing = _subgraph_definition(workflow, subgraph_id)
-    if existing is not None:
-        return
-    if workflow.get("definitions") is None:
-        workflow["definitions"] = {}
-    workflow["definitions"].setdefault("subgraphs", []).append(copy.deepcopy(definition))
-
-
 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
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 00bb081cd..8748297a4 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -182,8 +182,9 @@ pre-reset `base_version` do not replay across it.
 ### 1.7 `define_subgraph`
 
 Command: `comfy workflow define-subgraph <file> <definition-file> [--id <uuid>]`.
-The command validates a serializable definition, assigns its id from `--id`,
-the definition's `id`, or a new UUID, and emits exactly one stamped op:
+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
 {
@@ -197,10 +198,12 @@ the definition's `id`, or a new UUID, and emits exactly one stamped op:
 }
 ```
 
-Only this op may carry `subgraph_id` or definition payloads. Creation fails
-before writing when the id already exists. Exact op replay is a no-op; a
-different definition under an existing id is rejected as `malformed_op`.
-Subsequent interior edits use the existing id-addressed op scopes.
+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
 
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index ad07919e7..f7f5747a0 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -12,29 +12,28 @@
 from comfy_cli.command import workflow_edit
 
 SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc"
-NESTED_ID = "abcdefab-cdef-4abc-8def-abcdefabcdef"
-OTHER_NESTED_ID = "fedcbafe-dcba-4fed-8cba-fedcbafedcba"
 
 
-def _definition(value: int = 1) -> dict:
+def _definition() -> dict:
     return {
         "id": SUBGRAPH_ID,
         "name": "One",
         "inputs": [],
         "outputs": [],
-        "nodes": [{"id": 10, "type": "Inner", "widgets_values": [value]}],
+        "nodes": [{"id": 10, "type": "Inner"}],
         "links": [],
     }
 
 
-def test_define_subgraph_emits_cmp_payload_and_inserts_definition():
-    workflow = {"nodes": [], "links": []}
+def test_define_subgraph_emits_cmp_payload_without_mutating_workflow():
+    workflow = {"nodes": [{"id": 1, "type": "Existing"}], "links": []}
+    before = copy.deepcopy(workflow)
     definition = _definition()
-    snapshot = copy.deepcopy(definition)
 
     result, op = workflow_ops.define_subgraph(workflow, definition, actor="agent", base_version=4)
 
-    assert definition == snapshot
+    assert result is workflow
+    assert workflow == before
     assert op == {
         "op": "define_subgraph",
         "op_id": op["op_id"],
@@ -44,35 +43,21 @@ def test_define_subgraph_emits_cmp_payload_and_inserts_definition():
         "subgraph_id": SUBGRAPH_ID,
         "subgraph_definition": definition,
     }
-    assert result["definitions"]["subgraphs"] == [definition]
-
-
-@pytest.mark.parametrize(
-    ("definition", "match"),
-    [
-        ([], "JSON object"),
-        ({"id": 7, "nodes": [], "links": []}, "non-empty string id"),
-        ({"id": "not-a-uuid", "nodes": [], "links": []}, "valid UUID"),
-        ({"id": SUBGRAPH_ID, "nodes": {}, "links": []}, "nodes and links must be arrays"),
-        (
-            {
-                "id": SUBGRAPH_ID,
-                "nodes": [],
-                "links": [],
-                "definitions": {"subgraphs": [{"id": "not-a-uuid", "nodes": [], "links": []}]},
-            },
-            "definitions.subgraphs\\[0\\].*valid UUID",
-        ),
-    ],
-)
-def test_define_subgraph_rejects_malformed_input_before_mutation(definition, match):
-    workflow = {"nodes": [], "links": []}
-    before = copy.deepcopy(workflow)
 
-    with pytest.raises(ValueError, match=match):
-        workflow_ops.define_subgraph(workflow, definition)
 
-    assert workflow == before
+def test_define_subgraph_emits_semantically_odd_definition_for_cmp_validation():
+    definition = {"id": SUBGRAPH_ID, "nodes": "future-cmp-shape", "links": {"also": "future"}}
+
+    _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition)
+
+    assert op["subgraph_definition"] == definition
+
+
+def test_define_subgraph_rejects_local_replay():
+    _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition())
+
+    with pytest.raises(ValueError, match="unknown op 'define_subgraph'"):
+        workflow_ops.apply_op({"nodes": [], "links": []}, op, None)
 
 
 def test_define_subgraph_can_assign_an_explicit_new_id():
@@ -85,186 +70,31 @@ def test_define_subgraph_can_assign_an_explicit_new_id():
     assert op["subgraph_definition"]["id"] == SUBGRAPH_ID
 
 
-def test_define_subgraph_rejects_existing_id_and_different_definition():
-    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition()]}}
-
-    with pytest.raises(ValueError, match="already exists"):
-        workflow_ops.define_subgraph(workflow, _definition(2))
-
-
-def test_define_subgraph_preserves_nested_definitions_inside_single_parent_op():
-    nested = {"id": NESTED_ID, "nodes": [], "links": []}
-    definition = {**_definition(), "definitions": {"subgraphs": [nested]}}
-
-    result, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition)
-
-    assert op["subgraph_definition"]["definitions"] == {"subgraphs": [nested]}
-    assert result["definitions"]["subgraphs"] == [definition]
-    assert op["op"] == "define_subgraph"
-
-
-@pytest.mark.parametrize(
-    "definition",
-    [
-        {
-            **_definition(),
-            "definitions": {"subgraphs": [{"id": SUBGRAPH_ID, "nodes": [], "links": []}]},
-        },
-        {
-            **_definition(),
-            "definitions": {
-                "subgraphs": [
-                    {
-                        "id": NESTED_ID,
-                        "nodes": [],
-                        "links": [],
-                        "definitions": {"subgraphs": [{"id": OTHER_NESTED_ID, "nodes": [], "links": []}]},
-                    },
-                    {"id": OTHER_NESTED_ID, "nodes": [], "links": []},
-                ]
-            },
-        },
-    ],
-    ids=["ancestor", "across-branches"],
-)
-def test_define_subgraph_rejects_duplicate_ids_across_entire_definition_tree(definition):
-    workflow = {"nodes": [], "links": []}
-    before = copy.deepcopy(workflow)
-
-    with pytest.raises(ValueError, match="duplicates subgraph definition id"):
-        workflow_ops.define_subgraph(workflow, definition)
-
-    assert workflow == before
-
-
-def test_define_subgraph_rejects_id_already_nested_in_workflow():
-    existing = {
-        "id": OTHER_NESTED_ID,
-        "nodes": [],
-        "links": [],
-        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": [], "links": []}]},
-    }
-    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [existing]}}
-    before = copy.deepcopy(workflow)
-    definition = {
-        **_definition(),
-        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": [], "links": []}]},
-    }
-
-    with pytest.raises(ValueError, match="duplicates subgraph definition id"):
-        workflow_ops.define_subgraph(workflow, definition)
-
-    assert workflow == before
-
-
-def test_apply_define_subgraph_rejects_malformed_nested_definition_atomically():
-    workflow = {"nodes": [], "links": []}
-    before = copy.deepcopy(workflow)
-    definition = {
-        **_definition(),
-        "definitions": {"subgraphs": [{"id": NESTED_ID, "nodes": {}, "links": []}]},
-    }
-    op = {
-        "op": "define_subgraph",
-        "op_id": "a" * 32,
-        "actor": "peer",
-        "base_version": 0,
-        "stamp": [0, "peer"],
-        "subgraph_id": SUBGRAPH_ID,
-        "subgraph_definition": definition,
-    }
-
-    with pytest.raises(ValueError, match="malformed_op:.*definitions.subgraphs\\[0\\].*nodes and links"):
-        workflow_ops.apply_op(workflow, op, None)
-
-    assert workflow == before
-
-
-def test_apply_define_subgraph_replays_are_idempotent():
-    workflow = {"nodes": [], "links": []}
-    _, op = workflow_ops.define_subgraph(workflow, _definition())
-    before = copy.deepcopy(workflow)
-
-    workflow_ops.apply_op(workflow, op, None)
-    assert workflow == before
-
-    conflicting = {**op, "op_id": "f" * 32, "subgraph_definition": _definition(2)}
-    workflow_ops.apply_op(workflow, conflicting, None)
-    assert workflow["definitions"] == before["definitions"]
-
-
-def test_define_subgraph_write_targets_are_scoped_by_definition_id():
-    first = {"op": "define_subgraph", "subgraph_id": SUBGRAPH_ID}
-    other = {"op": "define_subgraph", "subgraph_id": NESTED_ID}
-
-    assert workflow_ops.detect_conflict(first, other) is False
-    assert workflow_ops.detect_conflict(first, dict(first)) is True
-
-
-def test_apply_define_subgraph_accepts_null_definitions_container():
-    workflow = {"nodes": [], "links": [], "definitions": None}
-    _, op = workflow_ops.define_subgraph(workflow, _definition())
-
-    assert workflow["definitions"]["subgraphs"] == [op["subgraph_definition"]]
-
-
-@pytest.mark.parametrize(
-    ("definition", "match"),
-    [
-        ({**_definition(), "nodes": [{"id": 1, "type": "Inner", "inputs": 1}]}, "inputs must be an array"),
-        ({**_definition(), "links": [[1, 2]]}, "link.*tuple"),
-        ({**_definition(), "nodes": [{"id": 1, "type": SUBGRAPH_ID}]}, "cyclic subgraph reference"),
-        (
-            {
-                **_definition(),
-                "nodes": [{"id": 1, "type": NESTED_ID}],
-                "definitions": {
-                    "subgraphs": [{"id": NESTED_ID, "nodes": [{"id": 2, "type": SUBGRAPH_ID}], "links": []}]
-                },
-            },
-            "cyclic subgraph reference",
-        ),
-    ],
-)
-def test_define_subgraph_rejects_malformed_or_recursive_interior(definition, match):
-    with pytest.raises(ValueError, match=match):
-        workflow_ops.define_subgraph({"nodes": [], "links": []}, definition)
-
-
-def test_define_subgraph_preserves_explicit_falsy_ids_for_validation():
-    with pytest.raises(ValueError, match="non-empty string id"):
-        workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition(), subgraph_id="")
-    with pytest.raises(ValueError, match="non-empty string id"):
-        workflow_ops.define_subgraph({"nodes": [], "links": []}, {"id": 0, "nodes": [], "links": []})
-
-
-def test_apply_define_subgraph_different_redelivery_is_a_noop():
-    workflow = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition(2)]}}
-    before = copy.deepcopy(workflow)
-    _, original_op = workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition())
-
-    workflow_ops.apply_op(workflow, original_op, None)
-
-    assert workflow["definitions"] == before["definitions"]
-    assert original_op["op_id"] in workflow["_applied_ops"]
-
-
-def test_apply_define_subgraph_fresh_op_id_exercises_definition_idempotency():
-    workflow = {"nodes": [], "links": []}
-    _, op = workflow_ops.define_subgraph(workflow, _definition())
-    workflow_ops.strip_internal(workflow)
+def test_define_subgraph_command_emits_without_writing_workflow(tmp_path, monkeypatch):
+    workflow = tmp_path / "workflow.json"
+    original = {"nodes": [], "links": []}
+    workflow.write_text(json.dumps(original))
+    definition_file = tmp_path / "definition.json"
+    definition_file.write_text(json.dumps(_definition()))
+    emitted = []
 
-    workflow_ops.apply_op(workflow, {**op, "op_id": "e" * 32}, None)
+    class Renderer:
+        command = ""
 
-    assert workflow["definitions"]["subgraphs"] == [_definition()]
+        def is_pretty(self):
+            return False
 
+        def emit(self, payload, **kwargs):
+            emitted.append((payload, kwargs))
 
-def test_replace_ops_emits_definition_before_nodes():
-    new = {"nodes": [], "links": [], "definitions": {"subgraphs": [_definition()]}}
+    monkeypatch.setattr(workflow_edit, "get_renderer", Renderer)
 
-    ops = workflow_ops.replace_ops({"nodes": [], "links": []}, new)
+    workflow_edit.define_subgraph_cmd(str(workflow), str(definition_file))
 
-    assert [op["op"] for op in ops] == ["define_subgraph"]
+    assert json.loads(workflow.read_text()) == original
+    assert emitted[0][0]["op"]["op"] == "define_subgraph"
+    assert emitted[0][0]["wrote"] is None
+    assert emitted[0][1]["changed"] is False
 
 
 def test_define_subgraph_command_rejects_non_regular_definition_file(tmp_path, monkeypatch):

From fad25f13ecb25423a5c1720d606545ed7d37c45c Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Thu, 17 Sep 2026 19:03:43 -0700
Subject: [PATCH 8/9] fix(workflow): preserve strict subgraph UUID validation

Addresses review feedback:

https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r4033155284
---
 comfy_cli/workflow_ops.py                  | 4 ++--
 tests/comfy_cli/test_define_subgraph_op.py | 8 ++++++++
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index fb8731969..c06677f10 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -64,7 +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
-_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)
+_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:
@@ -1865,7 +1865,7 @@ def define_subgraph(
         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 _UUID_RE.fullmatch(definition_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")
diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py
index f7f5747a0..5aa7b7679 100644
--- a/tests/comfy_cli/test_define_subgraph_op.py
+++ b/tests/comfy_cli/test_define_subgraph_op.py
@@ -70,6 +70,14 @@ def test_define_subgraph_can_assign_an_explicit_new_id():
     assert op["subgraph_definition"]["id"] == SUBGRAPH_ID
 
 
+@pytest.mark.parametrize(
+    "subgraph_id", ["12345678-1234-0123-8123-123456789abc", "12345678-1234-4123-7123-123456789abc"]
+)
+def test_define_subgraph_rejects_invalid_uuid_version_or_variant(subgraph_id):
+    with pytest.raises(ValueError, match="valid UUID"):
+        workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition(), subgraph_id=subgraph_id)
+
+
 def test_define_subgraph_command_emits_without_writing_workflow(tmp_path, monkeypatch):
     workflow = tmp_path / "workflow.json"
     original = {"nodes": [], "links": []}

From a8e11c8404ab90040b48b4f6570a6e6764509a87 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Thu, 17 Sep 2026 19:04:48 -0700
Subject: [PATCH 9/9] fix(workflow): reject deferred ops on local write paths

Addresses review feedback:

https://github.com/Comfy-Org/comfy-cli/pull/865#discussion_r4033155289
---
 comfy_cli/command/workflow_edit.py            | 12 ++++++
 tests/comfy_cli/command/test_workflow_edit.py | 37 +++++++++++++++++++
 2 files changed, 49 insertions(+)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 8d0348d92..db16491f5 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -815,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
         )
@@ -978,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")
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index fddbb4aa4..f7fd33eb9 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/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)."""