From 700e47e2cc1ab60eb8235360c8427270f0c1a0ad Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 00:47:46 -0700
Subject: [PATCH 01/13] feat(workflow): add insert-workflow command emitting
 stamped insert_workflow op

---
 comfy_cli/command/workflow.py              |   3 +
 comfy_cli/command/workflow_edit.py         |  24 +++++
 comfy_cli/error_codes.py                   |   6 ++
 comfy_cli/workflow_ops.py                  | 115 ++++++++++++++++++++-
 docs/op-vocabulary-v1.md                   |  29 +++++-
 tests/comfy_cli/test_insert_workflow_op.py |  57 ++++++++++
 6 files changed, 232 insertions(+), 2 deletions(-)
 create mode 100644 tests/comfy_cli/test_insert_workflow_op.py

diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py
index 850bd918b..ad5cf94b4 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("insert-workflow", help="Insert a complete workflow; emits one insert_workflow op.")(
+    _wedit.insert_workflow_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..2f80c40bd 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -137,6 +137,30 @@ def _graph_or_exit(input_path, host, port, renderer, where=None):
 # ---------------------------------------------------------------------------
 
 
+@tracking.track_command("workflow")
+def insert_workflow_cmd(
+    file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to update.")],
+    template: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to insert.")],
+    actor: ActorOpt = "cli",
+    base_version: BaseVersionOpt = 0,
+    stdout: StdoutOpt = False,
+):
+    """Insert a workflow template and emit one atomic ``insert_workflow`` op."""
+    renderer = get_renderer()
+    renderer.command = "workflow insert-workflow"
+    p, workflow = _load_workflow_or_fail(renderer, file)
+    try:
+        template_path = Path(template).expanduser()
+        inserted = json.loads(template_path.read_text(encoding="utf-8"))
+        if not isinstance(inserted, dict):
+            raise ValueError("template must be a JSON object")
+        workflow, op = workflow_ops.insert_workflow(workflow, inserted, actor=actor, base_version=base_version)
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
+        _emit_edit_error(renderer, e, hint="provide a frontend-format workflow template JSON file")
+        raise typer.Exit(code=1) from e
+    _finish(renderer, p, workflow, op, base_version, stdout, "workflow insert-workflow")
+
+
 @tracking.track_command("workflow")
 def add_node_cmd(
     file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON.")],
diff --git a/comfy_cli/error_codes.py b/comfy_cli/error_codes.py
index d5de7ac85..c7b81db3c 100644
--- a/comfy_cli/error_codes.py
+++ b/comfy_cli/error_codes.py
@@ -600,6 +600,12 @@ class ErrorCode:
         "(docs/op-vocabulary-v1.md: batchable = no) and the batch was rejected atomically — nothing was applied.",
         "run the standalone `comfy workflow reset-doc <file> --confirm` first, then apply the remaining ops as a batch",
     ),
+    ErrorCode(
+        "workflow_insert_workflow_not_batchable",
+        "A batch contained an `insert_workflow` op. A complete workflow insertion is one standalone atomic op, "
+        "so nesting it in the spec batch protocol is rejected and nothing is applied.",
+        "run `comfy workflow insert-workflow <file> <template>` instead",
+    ),
     ErrorCode(
         "workflow_reset_doc_unconfirmed",
         "`comfy workflow reset-doc` was called without `--confirm`. The command fails closed: it erases every "
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 9afa0a262..1dd4b0afe 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",
+    "insert_workflow",
+)
 
 #: Kinds frozen in the contract whose replay is not implemented yet.
 #: ``apply_op`` must keep rejecting these. Empty since amendment v1.1:
@@ -117,6 +125,11 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
         "command": "comfy workflow reset-doc <file> --confirm",
         "does": "resets the whole document to the empty baseline and erases its replay history",
     },
+    "insert_workflow": {
+        "code": "workflow_insert_workflow_not_batchable",
+        "command": "comfy workflow insert-workflow <file> <template>",
+        "does": "inserts a complete workflow in one transaction",
+    },
 }
 
 
@@ -1295,6 +1308,57 @@ def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int =
     return ops
 
 
+def remap_workflow_ids(workflow: dict, *, node_id_start: int, link_id_start: int) -> dict:
+    """Port of cmp ``remapWorkflowIds``: remap only the top-level namespace.
+
+    Definition interiors are independent graphs and deliberately remain
+    untouched. The input is never mutated.
+    """
+    out = copy.deepcopy(workflow)
+    nodes = out.get("nodes") or []
+    links = out.get("links") or []
+    node_ids = {node.get("id"): node_id_start + index for index, node in enumerate(nodes)}
+    link_ids = {link[0]: link_id_start + index for index, link in enumerate(links) if isinstance(link, list)}
+
+    for node in nodes:
+        original_id = node.get("id")
+        node["id"] = node_ids[original_id]
+        if isinstance(node.get("inputs"), list):
+            for slot in node["inputs"]:
+                if isinstance(slot, dict) and "link" in slot and slot["link"] in link_ids:
+                    slot["link"] = link_ids[slot["link"]]
+        if isinstance(node.get("outputs"), list):
+            for slot in node["outputs"]:
+                if isinstance(slot, dict) and isinstance(slot.get("links"), list):
+                    slot["links"] = [link_ids.get(link_id, link_id) for link_id in slot["links"]]
+
+    for link in links:
+        if not isinstance(link, list):
+            continue
+        link[0] = link_ids.get(link[0], link[0])
+        link[1] = node_ids.get(link[1], link[1])
+        link[3] = node_ids.get(link[3], link[3])
+    return out
+
+
+def insert_workflow(
+    workflow: dict,
+    template: dict,
+    *,
+    actor: str = "cli",
+    base_version: int = 0,
+) -> tuple[dict, dict]:
+    """Remap and insert a complete workflow with one stamped op."""
+    numeric_nodes = [n.get("id") for n in workflow.get("nodes", []) if isinstance(n, dict)]
+    numeric_links = [link[0] for link in workflow.get("links", []) if isinstance(link, list) and link]
+    node_start = max([workflow.get("last_node_id", 0), *[x for x in numeric_nodes if isinstance(x, int)]]) + 1
+    link_start = max([workflow.get("last_link_id", 0), *[x for x in numeric_links if isinstance(x, int)]]) + 1
+    remapped = remap_workflow_ids(template, node_id_start=node_start, link_id_start=link_start)
+    op = _new_op("insert_workflow", actor, base_version, workflow=remapped)
+    apply_op(workflow, op, None)
+    return workflow, op
+
+
 def delete_node(
     workflow: dict,
     graph,
@@ -1813,6 +1877,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 == "insert_workflow":
+            _apply_insert_workflow(workflow, op)
         else:
             raise ValueError(f"unknown op {kind!r}")
     except BaseException:
@@ -1842,6 +1908,53 @@ def _apply_add_node(workflow: dict, op: dict) -> None:
         workflow["last_node_id"] = max(workflow.get("last_node_id") or 0, op["node_id"])
 
 
+def _apply_insert_workflow(workflow: dict, op: dict) -> None:
+    inserted = op.get("workflow")
+    if not isinstance(inserted, dict):
+        raise ValueError("malformed_op: insert_workflow workflow must be an object")
+    nodes, links = inserted.get("nodes"), inserted.get("links", [])
+    if not isinstance(nodes, list) or not isinstance(links, list):
+        raise ValueError("malformed_op: insert_workflow nodes and links must be arrays")
+    live_node_ids = {str(n.get("id")) for n in workflow.get("nodes", []) if isinstance(n, dict)}
+    live_link_ids = {str(link[0]) for link in workflow.get("links", []) if isinstance(link, list) and link}
+    seen_nodes: set[str] = set()
+    for node in nodes:
+        if (
+            not isinstance(node, dict)
+            or node.get("id") is None
+            or not isinstance(node.get("type"), str)
+            or not node["type"]
+        ):
+            raise ValueError("invalid_node_payload: insert_workflow every node requires id and type")
+        key = str(node["id"])
+        if key in live_node_ids or key in seen_nodes:
+            raise ValueError(f"node_id_collision: insert_workflow node id {key!r} collides")
+        seen_nodes.add(key)
+    seen_links: set[str] = set()
+    for link in links:
+        if not isinstance(link, list) or not link or link[0] is None:
+            raise ValueError("malformed_op: insert_workflow every link must be a tuple with an id")
+        key = str(link[0])
+        if key in live_link_ids or key in seen_links:
+            raise ValueError(f"link_id_collision: insert_workflow link id {key!r} collides")
+        seen_links.add(key)
+    definitions = inserted.get("definitions")
+    if definitions is not None and not isinstance(definitions, dict):
+        raise ValueError("malformed_op: insert_workflow definitions must be an object")
+    subgraphs = definitions.get("subgraphs", []) if definitions else []
+    if not isinstance(subgraphs, list):
+        raise ValueError("malformed_op: insert_workflow definitions.subgraphs must be an array")
+
+    workflow.setdefault("nodes", []).extend(copy.deepcopy(nodes))
+    workflow.setdefault("links", []).extend(copy.deepcopy(links))
+    if subgraphs:
+        workflow.setdefault("definitions", {}).setdefault("subgraphs", []).extend(copy.deepcopy(subgraphs))
+    workflow["last_node_id"] = max(
+        workflow.get("last_node_id", 0), *[n["id"] for n in nodes if isinstance(n["id"], int)]
+    )
+    workflow["last_link_id"] = max(workflow.get("last_link_id", 0), *[x[0] for x in links if isinstance(x[0], int)])
+
+
 def _apply_set_widget(workflow: dict, op: dict, graph) -> None:
     # Last-writer-wins: a lower-stamped concurrent write to this target is
     # dropped, so the surviving value is the same in any apply order.
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 69edcf4a0..8d8c960b0 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 |
+| `insert_workflow` | no | `comfy workflow insert-workflow` | Merge a complete workflow template in one transaction |
 
 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 `insert_workflow` — standalone only
+
+Command: `comfy workflow insert-workflow <file> <template>`. The CLI remaps the
+template's top-level node and link ids away from ids in the live document, then
+emits exactly one stamped op:
+
+```json
+{
+  "op": "insert_workflow",
+  "op_id": "<uuid4 hex>",
+  "actor": "cli",
+  "base_version": 0,
+  "stamp": [0, "cli"],
+  "workflow": {"nodes": [], "links": [], "definitions": {"subgraphs": []}}
+}
+```
+
+The `workflow` payload is authoritative and carries top-level `nodes`, `links`,
+and optional `definitions.subgraphs`. Remapping rewrites top-level node ids,
+link tuple ids/endpoints, node input `link` fields, and output `links` arrays.
+Definition interiors are independent graph namespaces and remain unchanged.
+The applier validates all payloads before mutation and rejects with
+`node_id_collision`, `link_id_collision`, `malformed_op`,
+`invalid_node_payload`, or `catalog_required`. The kind is not batchable and a
+spec batch rejects it as `workflow_insert_workflow_not_batchable`.
+
 ## 2. Idempotency and identity
 
 * Every op carries `op_id`: uuid4 hex, minted by the **creator, before
diff --git a/tests/comfy_cli/test_insert_workflow_op.py b/tests/comfy_cli/test_insert_workflow_op.py
new file mode 100644
index 000000000..c3e144dd1
--- /dev/null
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from comfy_cli import workflow_ops
+from comfy_cli.cql.engine import Graph
+
+
+def _template() -> dict:
+    return {
+        "nodes": [
+            {"id": 100, "type": "Src", "outputs": [{"name": "o", "links": [200]}]},
+            {"id": 101, "type": "def-2", "inputs": [{"name": "a", "link": 200}]},
+        ],
+        "links": [[200, 100, 0, 101, 0, "X"]],
+        "definitions": {"subgraphs": [{"id": "def-2", "nodes": [{"id": 5, "type": "Inner"}], "links": []}]},
+    }
+
+
+def test_remap_workflow_ids_matches_cmp_fixture_and_does_not_mutate_input():
+    template = _template()
+    snapshot = copy.deepcopy(template)
+
+    out = workflow_ops.remap_workflow_ids(template, node_id_start=8, link_id_start=8)
+
+    assert template == snapshot
+    assert [node["id"] for node in out["nodes"]] == [8, 9]
+    assert out["links"] == [[8, 8, 0, 9, 0, "X"]]
+    assert out["nodes"][0]["outputs"][0]["links"] == [8]
+    assert out["nodes"][1]["inputs"] == [{"name": "a", "link": 8}]
+    assert out["definitions"]["subgraphs"][0]["nodes"][0]["id"] == 5
+
+
+def test_insert_workflow_emits_one_op_with_nested_definitions_preserved():
+    live = {"last_node_id": 7, "last_link_id": 7, "nodes": [], "links": []}
+
+    result, op = workflow_ops.insert_workflow(live, _template(), actor="agent", base_version=4)
+
+    assert op["op"] == "insert_workflow"
+    assert op["actor"] == "agent"
+    assert op["stamp"] == [4, "agent"]
+    assert [node["id"] for node in op["workflow"]["nodes"]] == [8, 9]
+    assert op["workflow"]["definitions"] == _template()["definitions"]
+    assert result["definitions"] == _template()["definitions"]
+    assert result["_applied_ops"] == [op["op_id"]]
+
+
+def test_insert_workflow_is_not_batchable():
+    with pytest.raises(workflow_ops.NotBatchableError) as exc:
+        workflow_ops.apply_specs(
+            {"nodes": [], "links": []},
+            Graph.from_object_info({}),
+            [{"op": "insert_workflow", "workflow": _template()}],
+        )
+    assert exc.value.code == "workflow_insert_workflow_not_batchable"

From fd09a3f9629172ba7b5596823af9098889953436 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 14:53:25 -0700
Subject: [PATCH 02/13] fix(workflow): validate insert payloads atomically

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3993617352
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3993617363
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3993617370
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3993617375
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3993617380
---
 comfy_cli/workflow_ops.py                  | 71 ++++++++++++++++-----
 docs/op-vocabulary-v1.md                   |  7 ++-
 tests/comfy_cli/test_insert_workflow_op.py | 72 ++++++++++++++++++++++
 3 files changed, 131 insertions(+), 19 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 1dd4b0afe..8b073caec 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1341,6 +1341,40 @@ def remap_workflow_ids(workflow: dict, *, node_id_start: int, link_id_start: int
     return out
 
 
+def _validate_insert_workflow_payload(inserted: Any, live_node_ids: set[str]) -> tuple[list, list, list]:
+    """Validate the complete insert payload before remapping or applying it."""
+    if not isinstance(inserted, dict):
+        raise ValueError("malformed_op: insert_workflow workflow must be an object")
+    nodes, links = inserted.get("nodes"), inserted.get("links", [])
+    if not isinstance(nodes, list) or not isinstance(links, list):
+        raise ValueError("malformed_op: insert_workflow nodes and links must be arrays")
+    node_ids: set[str] = set()
+    for node in nodes:
+        if (
+            not isinstance(node, dict)
+            or node.get("id") is None
+            or not isinstance(node.get("type"), str)
+            or not node["type"]
+        ):
+            raise ValueError("invalid_node_payload: insert_workflow every node requires id and type")
+        node_ids.add(str(node["id"]))
+    valid_endpoints = live_node_ids | node_ids
+    for link in links:
+        if not isinstance(link, list) or len(link) < 5 or link[0] is None:
+            raise ValueError("malformed_op: insert_workflow every link must have id and endpoint fields")
+        if str(link[1]) not in valid_endpoints or str(link[3]) not in valid_endpoints:
+            raise ValueError("dangling_link_endpoint: insert_workflow link references a missing node")
+    definitions = inserted.get("definitions")
+    if definitions is not None and not isinstance(definitions, dict):
+        raise ValueError("malformed_op: insert_workflow definitions must be an object")
+    subgraphs = definitions.get("subgraphs", []) if definitions else []
+    if not isinstance(subgraphs, list):
+        raise ValueError("malformed_op: insert_workflow definitions.subgraphs must be an array")
+    if any(not isinstance(definition, dict) or definition.get("id") is None for definition in subgraphs):
+        raise ValueError("malformed_op: insert_workflow every subgraph definition requires an id")
+    return nodes, links, subgraphs
+
+
 def insert_workflow(
     workflow: dict,
     template: dict,
@@ -1349,6 +1383,8 @@ def insert_workflow(
     base_version: int = 0,
 ) -> tuple[dict, dict]:
     """Remap and insert a complete workflow with one stamped op."""
+    live_node_ids = {str(n.get("id")) for n in workflow.get("nodes", []) if isinstance(n, dict)}
+    _validate_insert_workflow_payload(template, live_node_ids)
     numeric_nodes = [n.get("id") for n in workflow.get("nodes", []) if isinstance(n, dict)]
     numeric_links = [link[0] for link in workflow.get("links", []) if isinstance(link, list) and link]
     node_start = max([workflow.get("last_node_id", 0), *[x for x in numeric_nodes if isinstance(x, int)]]) + 1
@@ -1859,6 +1895,8 @@ def apply_op(workflow: dict, op: dict, graph) -> dict:
     if op["op_id"] in applied:
         return workflow
     kind = op["op"]
+    if kind != "insert_workflow" and "definitions" in op:
+        raise ValueError(f"malformed_op: {kind} does not accept definitions")
     # 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
@@ -1910,12 +1948,8 @@ def _apply_add_node(workflow: dict, op: dict) -> None:
 
 def _apply_insert_workflow(workflow: dict, op: dict) -> None:
     inserted = op.get("workflow")
-    if not isinstance(inserted, dict):
-        raise ValueError("malformed_op: insert_workflow workflow must be an object")
-    nodes, links = inserted.get("nodes"), inserted.get("links", [])
-    if not isinstance(nodes, list) or not isinstance(links, list):
-        raise ValueError("malformed_op: insert_workflow nodes and links must be arrays")
     live_node_ids = {str(n.get("id")) for n in workflow.get("nodes", []) if isinstance(n, dict)}
+    nodes, links, subgraphs = _validate_insert_workflow_payload(inserted, live_node_ids)
     live_link_ids = {str(link[0]) for link in workflow.get("links", []) if isinstance(link, list) and link}
     seen_nodes: set[str] = set()
     for node in nodes:
@@ -1932,27 +1966,30 @@ def _apply_insert_workflow(workflow: dict, op: dict) -> None:
         seen_nodes.add(key)
     seen_links: set[str] = set()
     for link in links:
-        if not isinstance(link, list) or not link or link[0] is None:
-            raise ValueError("malformed_op: insert_workflow every link must be a tuple with an id")
         key = str(link[0])
         if key in live_link_ids or key in seen_links:
             raise ValueError(f"link_id_collision: insert_workflow link id {key!r} collides")
         seen_links.add(key)
-    definitions = inserted.get("definitions")
-    if definitions is not None and not isinstance(definitions, dict):
-        raise ValueError("malformed_op: insert_workflow definitions must be an object")
-    subgraphs = definitions.get("subgraphs", []) if definitions else []
-    if not isinstance(subgraphs, list):
-        raise ValueError("malformed_op: insert_workflow definitions.subgraphs must be an array")
+    existing_subgraphs = workflow.get("definitions", {}).get("subgraphs", [])
+    existing_by_id = {str(definition.get("id")): definition for definition in existing_subgraphs}
+    new_subgraphs = []
+    for definition in subgraphs:
+        existing = existing_by_id.get(str(definition["id"]))
+        if existing is not None:
+            if existing != definition:
+                raise ValueError(f"definition_conflict: subgraph definition id {definition['id']!r} differs")
+            continue
+        existing_by_id[str(definition["id"])] = definition
+        new_subgraphs.append(definition)
 
     workflow.setdefault("nodes", []).extend(copy.deepcopy(nodes))
     workflow.setdefault("links", []).extend(copy.deepcopy(links))
-    if subgraphs:
-        workflow.setdefault("definitions", {}).setdefault("subgraphs", []).extend(copy.deepcopy(subgraphs))
+    if new_subgraphs:
+        workflow.setdefault("definitions", {}).setdefault("subgraphs", []).extend(copy.deepcopy(new_subgraphs))
     workflow["last_node_id"] = max(
-        workflow.get("last_node_id", 0), *[n["id"] for n in nodes if isinstance(n["id"], int)]
+        [workflow.get("last_node_id", 0), *[n["id"] for n in nodes if isinstance(n["id"], int)]]
     )
-    workflow["last_link_id"] = max(workflow.get("last_link_id", 0), *[x[0] for x in links if isinstance(x[0], int)])
+    workflow["last_link_id"] = max([workflow.get("last_link_id", 0), *[x[0] for x in links if isinstance(x[0], int)]])
 
 
 def _apply_set_widget(workflow: dict, op: dict, graph) -> None:
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 8d8c960b0..a05ca8b00 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -202,8 +202,11 @@ link tuple ids/endpoints, node input `link` fields, and output `links` arrays.
 Definition interiors are independent graph namespaces and remain unchanged.
 The applier validates all payloads before mutation and rejects with
 `node_id_collision`, `link_id_collision`, `malformed_op`,
-`invalid_node_payload`, or `catalog_required`. The kind is not batchable and a
-spec batch rejects it as `workflow_insert_workflow_not_batchable`.
+`invalid_node_payload`, `dangling_link_endpoint`, `definition_conflict`, or
+`catalog_required`. Only `define_subgraph` and `insert_workflow` ops may carry
+definitions; edit ops reject a `definitions` field as `malformed_op`. The kind
+is not batchable and a spec batch rejects it as
+`workflow_insert_workflow_not_batchable`.
 
 ## 2. Idempotency and identity
 
diff --git a/tests/comfy_cli/test_insert_workflow_op.py b/tests/comfy_cli/test_insert_workflow_op.py
index c3e144dd1..fe7b14de6 100644
--- a/tests/comfy_cli/test_insert_workflow_op.py
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -55,3 +55,75 @@ def test_insert_workflow_is_not_batchable():
             [{"op": "insert_workflow", "workflow": _template()}],
         )
     assert exc.value.code == "workflow_insert_workflow_not_batchable"
+
+
+def test_insert_workflow_accepts_empty_collections_without_partial_failure():
+    live = {"nodes": [], "links": [], "last_node_id": 0, "last_link_id": 0}
+
+    result, op = workflow_ops.insert_workflow(live, {"nodes": [], "links": [], "definitions": {"subgraphs": []}})
+
+    assert result["nodes"] == []
+    assert result["links"] == []
+    assert result["_applied_ops"] == [op["op_id"]]
+
+
+def test_insert_workflow_rejects_dangling_link_before_mutation():
+    live = {"nodes": [{"id": 1, "type": "Live"}], "links": []}
+    before = copy.deepcopy(live)
+    template = {"nodes": [{"id": 10, "type": "New"}], "links": [[20, 10, 0, 999, 0, "X"]]}
+
+    with pytest.raises(ValueError, match="dangling_link_endpoint"):
+        workflow_ops.insert_workflow(live, template)
+
+    assert live == before
+
+
+@pytest.mark.parametrize(
+    "template",
+    [
+        {"nodes": ["not-an-object"], "links": []},
+        {"nodes": [{"type": "MissingId"}], "links": []},
+        {"nodes": [{"id": 1, "type": "A"}], "links": [[2, 1]]},
+    ],
+)
+def test_insert_workflow_rejects_malformed_members_before_mutation(template):
+    live = {"nodes": [], "links": []}
+    before = copy.deepcopy(live)
+
+    with pytest.raises(ValueError, match="malformed_op|invalid_node_payload"):
+        workflow_ops.insert_workflow(live, template)
+
+    assert live == before
+
+
+def test_insert_workflow_definition_collision_is_idempotent_or_conflict():
+    definition = {"id": "def-1", "nodes": [], "links": []}
+    live = {"nodes": [], "links": [], "definitions": {"subgraphs": [copy.deepcopy(definition)]}}
+
+    workflow_ops.insert_workflow(
+        live, {"nodes": [], "links": [], "definitions": {"subgraphs": [copy.deepcopy(definition)]}}
+    )
+    assert live["definitions"]["subgraphs"] == [definition]
+
+    before = copy.deepcopy(live)
+    conflicting = {"id": "def-1", "nodes": [{"id": 1, "type": "Changed"}], "links": []}
+    with pytest.raises(ValueError, match="definition_conflict"):
+        workflow_ops.insert_workflow(live, {"nodes": [], "links": [], "definitions": {"subgraphs": [conflicting]}})
+    assert live == before
+
+
+def test_non_definition_op_rejects_definitions_field():
+    live = {"nodes": [], "links": []}
+    op = workflow_ops._new_op(
+        "add_node",
+        "test",
+        0,
+        node_id=1,
+        node={"id": 1, "type": "A"},
+        definitions={"subgraphs": []},
+    )
+
+    with pytest.raises(ValueError, match="malformed_op.*definitions"):
+        workflow_ops.apply_op(live, op, None)
+
+    assert live == {"nodes": [], "links": [], "_applied_ops": []}

From 10e556e8206bcca2ab10e0d9b978858d5dff1985 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 15:33:15 -0700
Subject: [PATCH 03/13] fix(workflow): keep insert workflow emit-only

---
 comfy_cli/command/workflow_edit.py         |  12 +-
 comfy_cli/workflow_ops.py                  | 136 +--------------------
 docs/op-vocabulary-v1.md                   |  19 ++-
 tests/comfy_cli/test_insert_workflow_op.py |  95 ++++----------
 4 files changed, 45 insertions(+), 217 deletions(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 2f80c40bd..6dbef63f5 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -128,6 +128,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)
 
@@ -154,11 +162,11 @@ def insert_workflow_cmd(
         inserted = json.loads(template_path.read_text(encoding="utf-8"))
         if not isinstance(inserted, dict):
             raise ValueError("template must be a JSON object")
-        workflow, op = workflow_ops.insert_workflow(workflow, inserted, actor=actor, base_version=base_version)
+        _, op = workflow_ops.insert_workflow(workflow, inserted, actor=actor, base_version=base_version)
     except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
         _emit_edit_error(renderer, e, hint="provide a frontend-format workflow template JSON file")
         raise typer.Exit(code=1) from e
-    _finish(renderer, p, workflow, op, base_version, stdout, "workflow insert-workflow")
+    _emit_op(renderer, p, op, base_version, "workflow insert-workflow")
 
 
 @tracking.track_command("workflow")
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 8b073caec..88d3f71af 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -101,10 +101,10 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
     "insert_workflow",
 )
 
-#: 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.
+#: ``insert_workflow`` is emitted for cmp to validate and apply; the CLI must
+#: keep rejecting local replay to preserve that ownership boundary.
+DEFERRED_OPS: tuple[str, ...] = ("insert_workflow",)
 
 #: Kinds a batch (``apply_specs``) dispatches. ``clear`` and ``reset_doc`` are
 #: standalone-only: they rewrite the whole document, so they never ride inside
@@ -1308,73 +1308,6 @@ def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int =
     return ops
 
 
-def remap_workflow_ids(workflow: dict, *, node_id_start: int, link_id_start: int) -> dict:
-    """Port of cmp ``remapWorkflowIds``: remap only the top-level namespace.
-
-    Definition interiors are independent graphs and deliberately remain
-    untouched. The input is never mutated.
-    """
-    out = copy.deepcopy(workflow)
-    nodes = out.get("nodes") or []
-    links = out.get("links") or []
-    node_ids = {node.get("id"): node_id_start + index for index, node in enumerate(nodes)}
-    link_ids = {link[0]: link_id_start + index for index, link in enumerate(links) if isinstance(link, list)}
-
-    for node in nodes:
-        original_id = node.get("id")
-        node["id"] = node_ids[original_id]
-        if isinstance(node.get("inputs"), list):
-            for slot in node["inputs"]:
-                if isinstance(slot, dict) and "link" in slot and slot["link"] in link_ids:
-                    slot["link"] = link_ids[slot["link"]]
-        if isinstance(node.get("outputs"), list):
-            for slot in node["outputs"]:
-                if isinstance(slot, dict) and isinstance(slot.get("links"), list):
-                    slot["links"] = [link_ids.get(link_id, link_id) for link_id in slot["links"]]
-
-    for link in links:
-        if not isinstance(link, list):
-            continue
-        link[0] = link_ids.get(link[0], link[0])
-        link[1] = node_ids.get(link[1], link[1])
-        link[3] = node_ids.get(link[3], link[3])
-    return out
-
-
-def _validate_insert_workflow_payload(inserted: Any, live_node_ids: set[str]) -> tuple[list, list, list]:
-    """Validate the complete insert payload before remapping or applying it."""
-    if not isinstance(inserted, dict):
-        raise ValueError("malformed_op: insert_workflow workflow must be an object")
-    nodes, links = inserted.get("nodes"), inserted.get("links", [])
-    if not isinstance(nodes, list) or not isinstance(links, list):
-        raise ValueError("malformed_op: insert_workflow nodes and links must be arrays")
-    node_ids: set[str] = set()
-    for node in nodes:
-        if (
-            not isinstance(node, dict)
-            or node.get("id") is None
-            or not isinstance(node.get("type"), str)
-            or not node["type"]
-        ):
-            raise ValueError("invalid_node_payload: insert_workflow every node requires id and type")
-        node_ids.add(str(node["id"]))
-    valid_endpoints = live_node_ids | node_ids
-    for link in links:
-        if not isinstance(link, list) or len(link) < 5 or link[0] is None:
-            raise ValueError("malformed_op: insert_workflow every link must have id and endpoint fields")
-        if str(link[1]) not in valid_endpoints or str(link[3]) not in valid_endpoints:
-            raise ValueError("dangling_link_endpoint: insert_workflow link references a missing node")
-    definitions = inserted.get("definitions")
-    if definitions is not None and not isinstance(definitions, dict):
-        raise ValueError("malformed_op: insert_workflow definitions must be an object")
-    subgraphs = definitions.get("subgraphs", []) if definitions else []
-    if not isinstance(subgraphs, list):
-        raise ValueError("malformed_op: insert_workflow definitions.subgraphs must be an array")
-    if any(not isinstance(definition, dict) or definition.get("id") is None for definition in subgraphs):
-        raise ValueError("malformed_op: insert_workflow every subgraph definition requires an id")
-    return nodes, links, subgraphs
-
-
 def insert_workflow(
     workflow: dict,
     template: dict,
@@ -1382,17 +1315,8 @@ def insert_workflow(
     actor: str = "cli",
     base_version: int = 0,
 ) -> tuple[dict, dict]:
-    """Remap and insert a complete workflow with one stamped op."""
-    live_node_ids = {str(n.get("id")) for n in workflow.get("nodes", []) if isinstance(n, dict)}
-    _validate_insert_workflow_payload(template, live_node_ids)
-    numeric_nodes = [n.get("id") for n in workflow.get("nodes", []) if isinstance(n, dict)]
-    numeric_links = [link[0] for link in workflow.get("links", []) if isinstance(link, list) and link]
-    node_start = max([workflow.get("last_node_id", 0), *[x for x in numeric_nodes if isinstance(x, int)]]) + 1
-    link_start = max([workflow.get("last_link_id", 0), *[x for x in numeric_links if isinstance(x, int)]]) + 1
-    remapped = remap_workflow_ids(template, node_id_start=node_start, link_id_start=link_start)
-    op = _new_op("insert_workflow", actor, base_version, workflow=remapped)
-    apply_op(workflow, op, None)
-    return workflow, op
+    """Emit an insert op; cmp owns validation, ID remapping, and application."""
+    return workflow, _new_op("insert_workflow", actor, base_version, workflow=copy.deepcopy(template))
 
 
 def delete_node(
@@ -1915,8 +1839,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 == "insert_workflow":
-            _apply_insert_workflow(workflow, op)
         else:
             raise ValueError(f"unknown op {kind!r}")
     except BaseException:
@@ -1946,52 +1868,6 @@ def _apply_add_node(workflow: dict, op: dict) -> None:
         workflow["last_node_id"] = max(workflow.get("last_node_id") or 0, op["node_id"])
 
 
-def _apply_insert_workflow(workflow: dict, op: dict) -> None:
-    inserted = op.get("workflow")
-    live_node_ids = {str(n.get("id")) for n in workflow.get("nodes", []) if isinstance(n, dict)}
-    nodes, links, subgraphs = _validate_insert_workflow_payload(inserted, live_node_ids)
-    live_link_ids = {str(link[0]) for link in workflow.get("links", []) if isinstance(link, list) and link}
-    seen_nodes: set[str] = set()
-    for node in nodes:
-        if (
-            not isinstance(node, dict)
-            or node.get("id") is None
-            or not isinstance(node.get("type"), str)
-            or not node["type"]
-        ):
-            raise ValueError("invalid_node_payload: insert_workflow every node requires id and type")
-        key = str(node["id"])
-        if key in live_node_ids or key in seen_nodes:
-            raise ValueError(f"node_id_collision: insert_workflow node id {key!r} collides")
-        seen_nodes.add(key)
-    seen_links: set[str] = set()
-    for link in links:
-        key = str(link[0])
-        if key in live_link_ids or key in seen_links:
-            raise ValueError(f"link_id_collision: insert_workflow link id {key!r} collides")
-        seen_links.add(key)
-    existing_subgraphs = workflow.get("definitions", {}).get("subgraphs", [])
-    existing_by_id = {str(definition.get("id")): definition for definition in existing_subgraphs}
-    new_subgraphs = []
-    for definition in subgraphs:
-        existing = existing_by_id.get(str(definition["id"]))
-        if existing is not None:
-            if existing != definition:
-                raise ValueError(f"definition_conflict: subgraph definition id {definition['id']!r} differs")
-            continue
-        existing_by_id[str(definition["id"])] = definition
-        new_subgraphs.append(definition)
-
-    workflow.setdefault("nodes", []).extend(copy.deepcopy(nodes))
-    workflow.setdefault("links", []).extend(copy.deepcopy(links))
-    if new_subgraphs:
-        workflow.setdefault("definitions", {}).setdefault("subgraphs", []).extend(copy.deepcopy(new_subgraphs))
-    workflow["last_node_id"] = max(
-        [workflow.get("last_node_id", 0), *[n["id"] for n in nodes if isinstance(n["id"], int)]]
-    )
-    workflow["last_link_id"] = max([workflow.get("last_link_id", 0), *[x[0] for x in links if isinstance(x[0], int)]])
-
-
 def _apply_set_widget(workflow: dict, op: dict, graph) -> None:
     # Last-writer-wins: a lower-stamped concurrent write to this target is
     # dropped, so the surviving value is the same in any apply order.
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index a05ca8b00..4bee2d1e0 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -181,9 +181,8 @@ pre-reset `base_version` do not replay across it.
 
 ### 1.7 `insert_workflow` — standalone only
 
-Command: `comfy workflow insert-workflow <file> <template>`. The CLI remaps the
-template's top-level node and link ids away from ids in the live document, then
-emits exactly one stamped op:
+Command: `comfy workflow insert-workflow <file> <template>`. The CLI preserves
+the template payload verbatim and emits exactly one stamped op:
 
 ```json
 {
@@ -197,15 +196,11 @@ emits exactly one stamped op:
 ```
 
 The `workflow` payload is authoritative and carries top-level `nodes`, `links`,
-and optional `definitions.subgraphs`. Remapping rewrites top-level node ids,
-link tuple ids/endpoints, node input `link` fields, and output `links` arrays.
-Definition interiors are independent graph namespaces and remain unchanged.
-The applier validates all payloads before mutation and rejects with
-`node_id_collision`, `link_id_collision`, `malformed_op`,
-`invalid_node_payload`, `dangling_link_endpoint`, `definition_conflict`, or
-`catalog_required`. Only `define_subgraph` and `insert_workflow` ops may carry
-definitions; edit ops reject a `definitions` field as `malformed_op`. The kind
-is not batchable and a spec batch rejects it as
+and optional `definitions.subgraphs`. The CLI does not validate graph semantics,
+remap IDs, apply the op, or write a mutated workflow document; cmp owns those
+operations and returns their errors. Only `define_subgraph` and
+`insert_workflow` ops may carry definitions; edit ops reject a `definitions`
+field as `malformed_op`. The kind is not batchable and a spec batch rejects it as
 `workflow_insert_workflow_not_batchable`.
 
 ## 2. Idempotency and identity
diff --git a/tests/comfy_cli/test_insert_workflow_op.py b/tests/comfy_cli/test_insert_workflow_op.py
index fe7b14de6..7f0d77918 100644
--- a/tests/comfy_cli/test_insert_workflow_op.py
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -19,97 +19,46 @@ def _template() -> dict:
     }
 
 
-def test_remap_workflow_ids_matches_cmp_fixture_and_does_not_mutate_input():
-    template = _template()
-    snapshot = copy.deepcopy(template)
-
-    out = workflow_ops.remap_workflow_ids(template, node_id_start=8, link_id_start=8)
-
-    assert template == snapshot
-    assert [node["id"] for node in out["nodes"]] == [8, 9]
-    assert out["links"] == [[8, 8, 0, 9, 0, "X"]]
-    assert out["nodes"][0]["outputs"][0]["links"] == [8]
-    assert out["nodes"][1]["inputs"] == [{"name": "a", "link": 8}]
-    assert out["definitions"]["subgraphs"][0]["nodes"][0]["id"] == 5
-
-
-def test_insert_workflow_emits_one_op_with_nested_definitions_preserved():
+def test_insert_workflow_emits_input_payload_without_remapping_ids():
     live = {"last_node_id": 7, "last_link_id": 7, "nodes": [], "links": []}
+    template = _template()
 
-    result, op = workflow_ops.insert_workflow(live, _template(), actor="agent", base_version=4)
+    result, op = workflow_ops.insert_workflow(live, template, actor="agent", base_version=4)
 
     assert op["op"] == "insert_workflow"
     assert op["actor"] == "agent"
     assert op["stamp"] == [4, "agent"]
-    assert [node["id"] for node in op["workflow"]["nodes"]] == [8, 9]
-    assert op["workflow"]["definitions"] == _template()["definitions"]
-    assert result["definitions"] == _template()["definitions"]
-    assert result["_applied_ops"] == [op["op_id"]]
-
-
-def test_insert_workflow_is_not_batchable():
-    with pytest.raises(workflow_ops.NotBatchableError) as exc:
-        workflow_ops.apply_specs(
-            {"nodes": [], "links": []},
-            Graph.from_object_info({}),
-            [{"op": "insert_workflow", "workflow": _template()}],
-        )
-    assert exc.value.code == "workflow_insert_workflow_not_batchable"
-
-
-def test_insert_workflow_accepts_empty_collections_without_partial_failure():
-    live = {"nodes": [], "links": [], "last_node_id": 0, "last_link_id": 0}
-
-    result, op = workflow_ops.insert_workflow(live, {"nodes": [], "links": [], "definitions": {"subgraphs": []}})
+    assert op["workflow"] == template
+    assert result is live
 
-    assert result["nodes"] == []
-    assert result["links"] == []
-    assert result["_applied_ops"] == [op["op_id"]]
 
+def test_insert_workflow_emits_semantically_invalid_json_for_server_validation():
+    live = {"nodes": [], "links": []}
+    template = {"nodes": [{"id": 10}], "links": [[20, 10, 0, 999, 0]], "definitions": {"subgraphs": [{}]}}
 
-def test_insert_workflow_rejects_dangling_link_before_mutation():
-    live = {"nodes": [{"id": 1, "type": "Live"}], "links": []}
-    before = copy.deepcopy(live)
-    template = {"nodes": [{"id": 10, "type": "New"}], "links": [[20, 10, 0, 999, 0, "X"]]}
+    _, op = workflow_ops.insert_workflow(live, template)
 
-    with pytest.raises(ValueError, match="dangling_link_endpoint"):
-        workflow_ops.insert_workflow(live, template)
+    assert op["workflow"] == template
 
-    assert live == before
 
-
-@pytest.mark.parametrize(
-    "template",
-    [
-        {"nodes": ["not-an-object"], "links": []},
-        {"nodes": [{"type": "MissingId"}], "links": []},
-        {"nodes": [{"id": 1, "type": "A"}], "links": [[2, 1]]},
-    ],
-)
-def test_insert_workflow_rejects_malformed_members_before_mutation(template):
-    live = {"nodes": [], "links": []}
+def test_insert_workflow_does_not_mutate_local_workflow():
+    live = {"nodes": [{"id": 100, "type": "Existing"}], "links": [], "last_node_id": 100}
     before = copy.deepcopy(live)
 
-    with pytest.raises(ValueError, match="malformed_op|invalid_node_payload"):
-        workflow_ops.insert_workflow(live, template)
+    result, _ = workflow_ops.insert_workflow(live, _template())
 
+    assert result is live
     assert live == before
 
 
-def test_insert_workflow_definition_collision_is_idempotent_or_conflict():
-    definition = {"id": "def-1", "nodes": [], "links": []}
-    live = {"nodes": [], "links": [], "definitions": {"subgraphs": [copy.deepcopy(definition)]}}
-
-    workflow_ops.insert_workflow(
-        live, {"nodes": [], "links": [], "definitions": {"subgraphs": [copy.deepcopy(definition)]}}
-    )
-    assert live["definitions"]["subgraphs"] == [definition]
-
-    before = copy.deepcopy(live)
-    conflicting = {"id": "def-1", "nodes": [{"id": 1, "type": "Changed"}], "links": []}
-    with pytest.raises(ValueError, match="definition_conflict"):
-        workflow_ops.insert_workflow(live, {"nodes": [], "links": [], "definitions": {"subgraphs": [conflicting]}})
-    assert live == before
+def test_insert_workflow_is_not_batchable():
+    with pytest.raises(workflow_ops.NotBatchableError) as exc:
+        workflow_ops.apply_specs(
+            {"nodes": [], "links": []},
+            Graph.from_object_info({}),
+            [{"op": "insert_workflow", "workflow": _template()}],
+        )
+    assert exc.value.code == "workflow_insert_workflow_not_batchable"
 
 
 def test_non_definition_op_rejects_definitions_field():

From 6f811bfa86123bc2249d57a762190e78c8e046ad Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 16:06:20 -0700
Subject: [PATCH 04/13] test(workflow): verify insert op transport

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994020326
---
 tests/comfy_cli/command/test_workflow_edit.py | 28 +++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index fddbb4aa4..24ee15c93 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -16,6 +16,7 @@
 import json
 from pathlib import Path
 from typing import Any
+from unittest.mock import Mock
 
 import pytest
 from typer.testing import CliRunner
@@ -529,6 +530,33 @@ def _run(args: list[str], capsys) -> dict[str, Any]:
 # ---------------------------------------------------------------------------
 
 
+class TestInsertWorkflow:
+    def test_transmits_op_once_through_renderer(self, monkeypatch, tmp_path):
+        workflow = _write(tmp_path, {"nodes": [], "links": [], "groups": []})
+        template = _write(tmp_path, {"nodes": [], "links": [], "groups": []}, "template.json")
+        renderer = _force_json_renderer()
+        transport = Mock()
+        monkeypatch.setattr(renderer, "emit", transport)
+
+        result = CliRunner().invoke(
+            workflow_cmd.app,
+            ["insert-workflow", str(workflow), str(template), "--actor", "agent", "--base-version", "4"],
+            standalone_mode=False,
+        )
+
+        assert result.exit_code == 0
+        transport.assert_called_once()
+        payload = transport.call_args.args[0]
+        assert transport.call_args.kwargs == {"command": "workflow insert-workflow", "changed": False}
+        assert payload["workflow"] == str(workflow)
+        assert payload["base_version"] == 4
+        assert payload["wrote"] is None
+        assert payload["op"]["op"] == "insert_workflow"
+        assert payload["op"]["actor"] == "agent"
+        assert payload["op"]["stamp"] == [4, "agent"]
+        assert payload["op"]["workflow"] == {"nodes": [], "links": [], "groups": []}
+
+
 class TestAddNode:
     def test_adds_node_and_emits_op(self, patched_graph, tmp_path, capsys):
         path = _write(tmp_path, _base_workflow())

From f20eae5b16b07aff26c73346df0af78c4425ade7 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 16:06:24 -0700
Subject: [PATCH 05/13] fix(workflow): require insert payload collections

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994020330
---
 comfy_cli/workflow_ops.py                  | 11 +++++-
 docs/op-vocabulary-v1.md                   | 10 +++---
 tests/comfy_cli/test_insert_workflow_op.py | 42 +++++++++++++++++++++-
 3 files changed, 56 insertions(+), 7 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 88d3f71af..d9807cf87 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1315,7 +1315,16 @@ def insert_workflow(
     actor: str = "cli",
     base_version: int = 0,
 ) -> tuple[dict, dict]:
-    """Emit an insert op; cmp owns validation, ID remapping, and application."""
+    """Structurally validate and emit an insert op without applying it."""
+    if not isinstance(template, dict):
+        raise ValueError("insert_workflow workflow must be a JSON object")
+    for field in ("nodes", "links", "groups"):
+        if field not in template:
+            raise ValueError(f"insert_workflow missing required field: {field}")
+        if not isinstance(template[field], list):
+            raise ValueError(f"insert_workflow field {field} must be an array")
+    if "definitions" in template and not isinstance(template["definitions"], dict):
+        raise ValueError("insert_workflow field definitions must be an object")
     return workflow, _new_op("insert_workflow", actor, base_version, workflow=copy.deepcopy(template))
 
 
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 4bee2d1e0..4aa2d7410 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -191,14 +191,14 @@ the template payload verbatim and emits exactly one stamped op:
   "actor": "cli",
   "base_version": 0,
   "stamp": [0, "cli"],
-  "workflow": {"nodes": [], "links": [], "definitions": {"subgraphs": []}}
+  "workflow": {"nodes": [], "links": [], "groups": [], "definitions": {"subgraphs": []}}
 }
 ```
 
-The `workflow` payload is authoritative and carries top-level `nodes`, `links`,
-and optional `definitions.subgraphs`. The CLI does not validate graph semantics,
-remap IDs, apply the op, or write a mutated workflow document; cmp owns those
-operations and returns their errors. Only `define_subgraph` and
+The `workflow` payload is authoritative and requires top-level `nodes`, `links`,
+and `groups` arrays; `definitions` is optional. The CLI checks only this outer
+shape. It does not validate graph semantics, remap IDs, apply the op, or write a
+mutated workflow document; cmp owns those operations. Only `define_subgraph` and
 `insert_workflow` ops may carry definitions; edit ops reject a `definitions`
 field as `malformed_op`. The kind is not batchable and a spec batch rejects it as
 `workflow_insert_workflow_not_batchable`.
diff --git a/tests/comfy_cli/test_insert_workflow_op.py b/tests/comfy_cli/test_insert_workflow_op.py
index 7f0d77918..8a8a55acc 100644
--- a/tests/comfy_cli/test_insert_workflow_op.py
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -15,6 +15,7 @@ def _template() -> dict:
             {"id": 101, "type": "def-2", "inputs": [{"name": "a", "link": 200}]},
         ],
         "links": [[200, 100, 0, 101, 0, "X"]],
+        "groups": [],
         "definitions": {"subgraphs": [{"id": "def-2", "nodes": [{"id": 5, "type": "Inner"}], "links": []}]},
     }
 
@@ -34,13 +35,52 @@ def test_insert_workflow_emits_input_payload_without_remapping_ids():
 
 def test_insert_workflow_emits_semantically_invalid_json_for_server_validation():
     live = {"nodes": [], "links": []}
-    template = {"nodes": [{"id": 10}], "links": [[20, 10, 0, 999, 0]], "definitions": {"subgraphs": [{}]}}
+    template = {
+        "nodes": [{"id": 10}],
+        "links": [[20, 10, 0, 999, 0]],
+        "groups": [],
+        "definitions": {"subgraphs": [{}]},
+    }
 
     _, op = workflow_ops.insert_workflow(live, template)
 
     assert op["workflow"] == template
 
 
+@pytest.mark.parametrize("missing", ["nodes", "links", "groups"])
+def test_insert_workflow_rejects_each_missing_required_collection(missing):
+    template = _template()
+    del template[missing]
+
+    with pytest.raises(ValueError, match=rf"insert_workflow.*missing required field.*{missing}"):
+        workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
+
+
+def test_insert_workflow_accepts_required_collections_without_definitions():
+    template = {"nodes": [], "links": [], "groups": []}
+
+    _, op = workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
+
+    assert op["workflow"] == template
+
+
+@pytest.mark.parametrize("field", ["nodes", "links", "groups"])
+def test_insert_workflow_rejects_non_array_required_collection(field):
+    template = _template()
+    template[field] = {}
+
+    with pytest.raises(ValueError, match=rf"insert_workflow field {field} must be an array"):
+        workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
+
+
+def test_insert_workflow_rejects_non_object_definitions():
+    template = _template()
+    template["definitions"] = []
+
+    with pytest.raises(ValueError, match="insert_workflow field definitions must be an object"):
+        workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
+
+
 def test_insert_workflow_does_not_mutate_local_workflow():
     live = {"nodes": [{"id": 100, "type": "Existing"}], "links": [], "last_node_id": 100}
     before = copy.deepcopy(live)

From c867b3f2237c95eba82e275fcb2985141e174dc6 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 16:45:31 -0700
Subject: [PATCH 06/13] fix(workflow): surface insert dispatch rejections

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994204133
---
 comfy_cli/command/workflow_edit.py            |  9 ++++++++-
 tests/comfy_cli/command/test_workflow_edit.py | 15 +++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 6dbef63f5..8ba8b5823 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -166,7 +166,14 @@ def insert_workflow_cmd(
     except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
         _emit_edit_error(renderer, e, hint="provide a frontend-format workflow template JSON file")
         raise typer.Exit(code=1) from e
-    _emit_op(renderer, p, op, base_version, "workflow insert-workflow")
+    try:
+        _emit_op(renderer, p, op, base_version, "workflow insert-workflow")
+    except ValueError as e:
+        # The outbound edit transport may return cmp's rejection. Preserve its
+        # message exactly: it is already the most actionable explanation and
+        # callers must not have to unwrap a second client-side paraphrase.
+        renderer.error(code="workflow_edit_invalid", message=str(e))
+        raise typer.Exit(code=1) from e
 
 
 @tracking.track_command("workflow")
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index 24ee15c93..af7fc10b8 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -19,6 +19,7 @@
 from unittest.mock import Mock
 
 import pytest
+import typer
 from typer.testing import CliRunner
 
 from comfy_cli import workflow_ops
@@ -556,6 +557,20 @@ def test_transmits_op_once_through_renderer(self, monkeypatch, tmp_path):
         assert payload["op"]["stamp"] == [4, "agent"]
         assert payload["op"]["workflow"] == {"nodes": [], "links": [], "groups": []}
 
+    def test_doc_host_rejection_is_surfaced_verbatim(self, monkeypatch, tmp_path, capsys):
+        workflow = _write(tmp_path, {"nodes": [], "links": []})
+        template = _write(tmp_path, {"nodes": []}, "template.json")
+        renderer = _force_json_renderer()
+        transport = Mock(side_effect=ValueError("cmp rejected insert: node_id_collision 100"))
+        monkeypatch.setattr(renderer, "emit", transport)
+
+        with pytest.raises(typer.Exit) as exc:
+            workflow_edit.insert_workflow_cmd(str(workflow), str(template))
+
+        assert exc.value.exit_code == 1
+        transport.assert_called_once()
+        assert "cmp rejected insert: node_id_collision 100" in capsys.readouterr().out
+
 
 class TestAddNode:
     def test_adds_node_and_emits_op(self, patched_graph, tmp_path, capsys):

From 32566b67c08506b1b2af3a0aac2cd4d6e2883db9 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 16:45:46 -0700
Subject: [PATCH 07/13] docs(workflow): record insert ID ownership

Addresses review feedback with the vetoable contract decision:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994204147
---
 docs/op-vocabulary-v1.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 4aa2d7410..4a2d40e65 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -198,7 +198,8 @@ the template payload verbatim and emits exactly one stamped op:
 The `workflow` payload is authoritative and requires top-level `nodes`, `links`,
 and `groups` arrays; `definitions` is optional. The CLI checks only this outer
 shape. It does not validate graph semantics, remap IDs, apply the op, or write a
-mutated workflow document; cmp owns those operations. Only `define_subgraph` and
+mutated workflow document. Per the contract decision recorded in the TDD
+(vetoable), cmp owns deterministic ID remapping from the op envelope ID. Only `define_subgraph` and
 `insert_workflow` ops may carry definitions; edit ops reject a `definitions`
 field as `malformed_op`. The kind is not batchable and a spec batch rejects it as
 `workflow_insert_workflow_not_batchable`.

From df7bf3fcc00855a57751bc6b3e995627aa8f0761 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 16:46:00 -0700
Subject: [PATCH 08/13] fix(workflow): make insert collections optional

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994204140
---
 comfy_cli/workflow_ops.py                  | 13 +++++++++----
 docs/op-vocabulary-v1.md                   |  9 +++++----
 tests/comfy_cli/test_insert_workflow_op.py | 20 +++++++++++++++-----
 3 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index d9807cf87..2e70edb5f 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -1315,13 +1315,18 @@ def insert_workflow(
     actor: str = "cli",
     base_version: int = 0,
 ) -> tuple[dict, dict]:
-    """Structurally validate and emit an insert op without applying it."""
+    """Structurally validate and emit an insert op without applying it.
+
+    The CLI deliberately preserves all source IDs. Per the vetoable contract
+    decision recorded in the TDD, cmp owns deterministic ID remapping from the
+    op envelope ID when it applies this payload.
+    """
     if not isinstance(template, dict):
         raise ValueError("insert_workflow workflow must be a JSON object")
+    if "nodes" not in template:
+        raise ValueError("insert_workflow missing required field: nodes")
     for field in ("nodes", "links", "groups"):
-        if field not in template:
-            raise ValueError(f"insert_workflow missing required field: {field}")
-        if not isinstance(template[field], list):
+        if field in template and not isinstance(template[field], list):
             raise ValueError(f"insert_workflow field {field} must be an array")
     if "definitions" in template and not isinstance(template["definitions"], dict):
         raise ValueError("insert_workflow field definitions must be an object")
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 4a2d40e65..0bf59b9d8 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -195,10 +195,11 @@ the template payload verbatim and emits exactly one stamped op:
 }
 ```
 
-The `workflow` payload is authoritative and requires top-level `nodes`, `links`,
-and `groups` arrays; `definitions` is optional. The CLI checks only this outer
-shape. It does not validate graph semantics, remap IDs, apply the op, or write a
-mutated workflow document. Per the contract decision recorded in the TDD
+The `workflow` payload is authoritative and requires a top-level `nodes` array;
+`links`, `groups`, and `definitions` are optional. When present, `links` and
+`groups` must be arrays and `definitions` must be an object. The CLI checks only
+this outer shape. It does not validate graph semantics, remap IDs, apply the op,
+or write a mutated workflow document. Per the contract decision recorded in the TDD
 (vetoable), cmp owns deterministic ID remapping from the op envelope ID. Only `define_subgraph` and
 `insert_workflow` ops may carry definitions; edit ops reject a `definitions`
 field as `malformed_op`. The kind is not batchable and a spec batch rejects it as
diff --git a/tests/comfy_cli/test_insert_workflow_op.py b/tests/comfy_cli/test_insert_workflow_op.py
index 8a8a55acc..ac2a85d46 100644
--- a/tests/comfy_cli/test_insert_workflow_op.py
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -47,15 +47,25 @@ def test_insert_workflow_emits_semantically_invalid_json_for_server_validation()
     assert op["workflow"] == template
 
 
-@pytest.mark.parametrize("missing", ["nodes", "links", "groups"])
-def test_insert_workflow_rejects_each_missing_required_collection(missing):
+def test_insert_workflow_rejects_missing_nodes():
     template = _template()
-    del template[missing]
+    del template["nodes"]
 
-    with pytest.raises(ValueError, match=rf"insert_workflow.*missing required field.*{missing}"):
+    with pytest.raises(ValueError, match=r"insert_workflow.*missing required field.*nodes"):
         workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
 
 
+def test_insert_workflow_accepts_nodes_only_and_preserves_omitted_optional_collections():
+    template = {"nodes": [{"id": 100, "type": "Src"}]}
+
+    _, op = workflow_ops.insert_workflow({"nodes": [], "links": []}, template)
+
+    assert op["workflow"] == template
+    assert "links" not in op["workflow"]
+    assert "groups" not in op["workflow"]
+    assert op["workflow"]["nodes"][0]["id"] == 100
+
+
 def test_insert_workflow_accepts_required_collections_without_definitions():
     template = {"nodes": [], "links": [], "groups": []}
 
@@ -65,7 +75,7 @@ def test_insert_workflow_accepts_required_collections_without_definitions():
 
 
 @pytest.mark.parametrize("field", ["nodes", "links", "groups"])
-def test_insert_workflow_rejects_non_array_required_collection(field):
+def test_insert_workflow_rejects_non_array_collection(field):
     template = _template()
     template[field] = {}
 

From ef08532cb45097452285fbb31cff71910c4740b0 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 17:32:35 -0700
Subject: [PATCH 09/13] fix: align insert-workflow emit-only interface

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994366813
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994367124
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994367463
---
 comfy_cli/command/workflow_edit.py            | 10 +------
 comfy_cli/discovery.py                        |  1 +
 comfy_cli/skills/comfy/SKILL.md               |  7 +++--
 tests/comfy_cli/command/test_workflow_edit.py | 28 ++++++++++++++-----
 4 files changed, 27 insertions(+), 19 deletions(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 8ba8b5823..9c92c542e 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -151,7 +151,6 @@ def insert_workflow_cmd(
     template: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to insert.")],
     actor: ActorOpt = "cli",
     base_version: BaseVersionOpt = 0,
-    stdout: StdoutOpt = False,
 ):
     """Insert a workflow template and emit one atomic ``insert_workflow`` op."""
     renderer = get_renderer()
@@ -166,14 +165,7 @@ def insert_workflow_cmd(
     except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError) as e:
         _emit_edit_error(renderer, e, hint="provide a frontend-format workflow template JSON file")
         raise typer.Exit(code=1) from e
-    try:
-        _emit_op(renderer, p, op, base_version, "workflow insert-workflow")
-    except ValueError as e:
-        # The outbound edit transport may return cmp's rejection. Preserve its
-        # message exactly: it is already the most actionable explanation and
-        # callers must not have to unwrap a second client-side paraphrase.
-        renderer.error(code="workflow_edit_invalid", message=str(e))
-        raise typer.Exit(code=1) from e
+    _emit_op(renderer, p, op, base_version, "workflow insert-workflow")
 
 
 @tracking.track_command("workflow")
diff --git a/comfy_cli/discovery.py b/comfy_cli/discovery.py
index d7af98050..03c0aa472 100644
--- a/comfy_cli/discovery.py
+++ b/comfy_cli/discovery.py
@@ -98,6 +98,7 @@
     "comfy workflow notes": "workflow",
     "comfy workflow print": "workflow",
     # structured edit primitives + recipes (CRDT op-based authoring)
+    "comfy workflow insert-workflow": "workflow",
     "comfy workflow add-node": "workflow",
     "comfy workflow connect": "workflow",
     "comfy workflow set-widget": "workflow",
diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md
index 4517faa3d..de2980a60 100644
--- a/comfy_cli/skills/comfy/SKILL.md
+++ b/comfy_cli/skills/comfy/SKILL.md
@@ -352,7 +352,7 @@ of a live graph, use the structured-edit primitives below — never raw `jq`/`se
 (This rule exists because that exact jq-on-`id==128` hand-edit is the anti-pattern
 `decompose` — and these primitives — were built to kill.)
 
-## Structured graph edits — `add-node` / `connect` / `set-widget` / `delete-node`
+## Structured graph edits — `insert-workflow` / `add-node` / `connect` / `set-widget` / `delete-node`
 
 The **sanctioned** way to mutate a graph's *structure* from code — the
 alternative to `jq`/`sed` on `nodes`/`links`/`widgets_values`. Each edit is
@@ -368,8 +368,8 @@ numbers come back as soft `warnings`) and emits a replayable **operation** in
   nodes; the in-app agent's path; any edit that must merge with a concurrent
   human editor) → the primitives here.
 
-> **Live co-editing / CRDT:** only the structured-edit primitives (`add-node`/
-> `connect`/`set-widget`/`delete-node`/`apply`) emit a mergeable **op** in
+> **Live co-editing / CRDT:** only the structured-edit primitives (`insert-workflow`/
+> `add-node`/`connect`/`set-widget`/`delete-node`/`apply`) emit a mergeable **op** in
 > `data.op`/`data.ops` (`op_id` + `actor` + `base_version` + `stamp`). Fragments +
 > `compose` produce a **whole-document** graph — fine for authoring a *fresh*
 > draft (the base), but it does **not** emit ops and will clobber a concurrent
@@ -385,6 +385,7 @@ CAT="--where cloud"
 # Start from an existing graph, or an empty one:
 echo '{"nodes":[],"links":[],"last_node_id":0,"last_link_id":0}' > wf.json
 
+comfy --json workflow insert-workflow wf.json template.json                # emits one atomic insert_workflow op
 comfy --json workflow add-node    wf.json KSampler --at 400,200 $CAT  # → data.op.node_id (minted)
 comfy --json workflow connect     wf.json 7.LATENT 3.samples $CAT     # source out-slot → target in-slot
 comfy --json workflow set-widget  wf.json 3.steps 35 $CAT             # widget by NAME; op carries {old,value}
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index af7fc10b8..15c3dba3e 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -535,6 +535,7 @@ class TestInsertWorkflow:
     def test_transmits_op_once_through_renderer(self, monkeypatch, tmp_path):
         workflow = _write(tmp_path, {"nodes": [], "links": [], "groups": []})
         template = _write(tmp_path, {"nodes": [], "links": [], "groups": []}, "template.json")
+        original = workflow.read_text(encoding="utf-8")
         renderer = _force_json_renderer()
         transport = Mock()
         monkeypatch.setattr(renderer, "emit", transport)
@@ -556,20 +557,33 @@ def test_transmits_op_once_through_renderer(self, monkeypatch, tmp_path):
         assert payload["op"]["actor"] == "agent"
         assert payload["op"]["stamp"] == [4, "agent"]
         assert payload["op"]["workflow"] == {"nodes": [], "links": [], "groups": []}
+        assert workflow.read_text(encoding="utf-8") == original
 
-    def test_doc_host_rejection_is_surfaced_verbatim(self, monkeypatch, tmp_path, capsys):
+    def test_stdout_is_not_offered_for_emit_only_command(self):
+        result = CliRunner().invoke(workflow_cmd.app, ["insert-workflow", "--help"])
+
+        assert result.exit_code == 0
+        assert "--stdout" not in result.output
+        assert "--in-place" not in result.output
+
+    def test_command_is_discoverable_in_schema_and_bundled_skill(self):
+        from comfy_cli.discovery import COMMAND_SCHEMAS
+
+        skill = Path(workflow_edit.__file__).parent.parent / "skills" / "comfy" / "SKILL.md"
+
+        assert COMMAND_SCHEMAS["comfy workflow insert-workflow"] == "workflow"
+        assert "insert-workflow" in skill.read_text(encoding="utf-8")
+
+    def test_local_validation_error_is_reported(self, tmp_path, capsys):
         workflow = _write(tmp_path, {"nodes": [], "links": []})
-        template = _write(tmp_path, {"nodes": []}, "template.json")
-        renderer = _force_json_renderer()
-        transport = Mock(side_effect=ValueError("cmp rejected insert: node_id_collision 100"))
-        monkeypatch.setattr(renderer, "emit", transport)
+        template = _write(tmp_path, {"links": []}, "template.json")
+        _force_json_renderer()
 
         with pytest.raises(typer.Exit) as exc:
             workflow_edit.insert_workflow_cmd(str(workflow), str(template))
 
         assert exc.value.exit_code == 1
-        transport.assert_called_once()
-        assert "cmp rejected insert: node_id_collision 100" in capsys.readouterr().out
+        assert "missing required field: nodes" in capsys.readouterr().out
 
 
 class TestAddNode:

From e769c27e9dbf123077ff00e779397e6d88d9eda6 Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Fri, 11 Sep 2026 18:40:42 -0700
Subject: [PATCH 10/13] fix(insert-workflow): read workflow from stdin for '-'
 and clarify emit-only validation

Addresses https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994578049
Addresses https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r3994578054
---
 comfy_cli/command/workflow_edit.py            | 11 ++++++++---
 comfy_cli/skills/comfy/SKILL.md               | 13 ++++++++-----
 tests/comfy_cli/command/test_workflow_edit.py | 17 +++++++++++++++++
 3 files changed, 33 insertions(+), 8 deletions(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 9c92c542e..0adcda0f2 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -148,7 +148,7 @@ def _graph_or_exit(input_path, host, port, renderer, where=None):
 @tracking.track_command("workflow")
 def insert_workflow_cmd(
     file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to update.")],
-    template: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to insert.")],
+    template: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to insert, or '-' for stdin.")],
     actor: ActorOpt = "cli",
     base_version: BaseVersionOpt = 0,
 ):
@@ -157,8 +157,13 @@ def insert_workflow_cmd(
     renderer.command = "workflow insert-workflow"
     p, workflow = _load_workflow_or_fail(renderer, file)
     try:
-        template_path = Path(template).expanduser()
-        inserted = json.loads(template_path.read_text(encoding="utf-8"))
+        if template == "-":
+            import sys
+
+            raw = sys.stdin.read()
+        else:
+            raw = Path(template).expanduser().read_text(encoding="utf-8")
+        inserted = json.loads(raw)
         if not isinstance(inserted, dict):
             raise ValueError("template must be a JSON object")
         _, op = workflow_ops.insert_workflow(workflow, inserted, actor=actor, base_version=base_version)
diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md
index de2980a60..ec62aa5e3 100644
--- a/comfy_cli/skills/comfy/SKILL.md
+++ b/comfy_cli/skills/comfy/SKILL.md
@@ -355,11 +355,14 @@ of a live graph, use the structured-edit primitives below — never raw `jq`/`se
 ## Structured graph edits — `insert-workflow` / `add-node` / `connect` / `set-widget` / `delete-node`
 
 The **sanctioned** way to mutate a graph's *structure* from code — the
-alternative to `jq`/`sed` on `nodes`/`links`/`widgets_values`. Each edit is
-validated against `object_info` (node class, widget name, widget value **shape**,
-and connection **type** are hard-checked; unknown COMBO values / out-of-range
-numbers come back as soft `warnings`) and emits a replayable **operation** in
-`data.op`.
+alternative to `jq`/`sed` on `nodes`/`links`/`widgets_values`. `insert-workflow`
+is emit-only: the CLI checks JSON shape and forwards the workflow without
+semantic validation or id remapping. The cmp applier on the server validates node
+types, links, and definition ids, remaps ids, and returns errors that the CLI
+surfaces verbatim. The other edits are validated against `object_info` (node
+class, widget name, widget value **shape**, and connection **type** are
+hard-checked; unknown COMBO values / out-of-range numbers come back as soft
+`warnings`) and emit a replayable **operation** in `data.op`.
 
 **When to use which editing path:**
 - **Reusable / human-authored workflow** → fragments + blueprint (above). *Default.*
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index 15c3dba3e..a7e271e19 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -532,6 +532,23 @@ def _run(args: list[str], capsys) -> dict[str, Any]:
 
 
 class TestInsertWorkflow:
+    def test_reads_template_from_stdin_when_path_is_dash(self, monkeypatch, tmp_path):
+        workflow = _write(tmp_path, {"nodes": [], "links": [], "groups": []})
+        template = {"nodes": [{"id": 12, "type": "ServerValidatedNode"}], "links": [], "groups": []}
+        renderer = _force_json_renderer()
+        transport = Mock()
+        monkeypatch.setattr(renderer, "emit", transport)
+
+        result = CliRunner().invoke(
+            workflow_cmd.app,
+            ["insert-workflow", str(workflow), "-"],
+            input=json.dumps(template),
+            standalone_mode=False,
+        )
+
+        assert result.exit_code == 0
+        assert transport.call_args.args[0]["op"]["workflow"] == template
+
     def test_transmits_op_once_through_renderer(self, monkeypatch, tmp_path):
         workflow = _write(tmp_path, {"nodes": [], "links": [], "groups": []})
         template = _write(tmp_path, {"nodes": [], "links": [], "groups": []}, "template.json")

From 04c9b3dd26c4a139127cc0832884c286aec6e18e Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Thu, 17 Sep 2026 11:57:09 -0700
Subject: [PATCH 11/13] fix: clarify insert-workflow source is unchanged

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r4033103298
---
 comfy_cli/command/workflow_edit.py            | 2 +-
 tests/comfy_cli/command/test_workflow_edit.py | 2 ++
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 0adcda0f2..307aa74f2 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -147,7 +147,7 @@ def _graph_or_exit(input_path, host, port, renderer, where=None):
 
 @tracking.track_command("workflow")
 def insert_workflow_cmd(
-    file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to update.")],
+    file: Annotated[str, typer.Argument(help="Source frontend-format workflow JSON; emit-only, file is not modified.")],
     template: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to insert, or '-' for stdin.")],
     actor: ActorOpt = "cli",
     base_version: BaseVersionOpt = 0,
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index a7e271e19..5cf3dfafe 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -580,6 +580,8 @@ def test_stdout_is_not_offered_for_emit_only_command(self):
         result = CliRunner().invoke(workflow_cmd.app, ["insert-workflow", "--help"])
 
         assert result.exit_code == 0
+        assert "emit-only" in result.output
+        assert "file is not modified" in result.output
         assert "--stdout" not in result.output
         assert "--in-place" not in result.output
 

From 3c6361eb9d0d6434495a2d534a6d07af7a7d9bed Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Thu, 17 Sep 2026 11:57:32 -0700
Subject: [PATCH 12/13] docs: describe insert-workflow stdin template

Addresses review feedback:
https://github.com/Comfy-Org/comfy-cli/pull/863#discussion_r4033103311
---
 comfy_cli/skills/comfy/SKILL.md | 1 +
 docs/op-vocabulary-v1.md        | 5 +++--
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md
index ec62aa5e3..473427d66 100644
--- a/comfy_cli/skills/comfy/SKILL.md
+++ b/comfy_cli/skills/comfy/SKILL.md
@@ -389,6 +389,7 @@ CAT="--where cloud"
 echo '{"nodes":[],"links":[],"last_node_id":0,"last_link_id":0}' > wf.json
 
 comfy --json workflow insert-workflow wf.json template.json                # emits one atomic insert_workflow op
+cat template.json | comfy --json workflow insert-workflow wf.json -        # '-' reads the template from stdin
 comfy --json workflow add-node    wf.json KSampler --at 400,200 $CAT  # → data.op.node_id (minted)
 comfy --json workflow connect     wf.json 7.LATENT 3.samples $CAT     # source out-slot → target in-slot
 comfy --json workflow set-widget  wf.json 3.steps 35 $CAT             # widget by NAME; op carries {old,value}
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 0bf59b9d8..9ef7f59ca 100644
--- a/docs/op-vocabulary-v1.md
+++ b/docs/op-vocabulary-v1.md
@@ -181,8 +181,9 @@ pre-reset `base_version` do not replay across it.
 
 ### 1.7 `insert_workflow` — standalone only
 
-Command: `comfy workflow insert-workflow <file> <template>`. The CLI preserves
-the template payload verbatim and emits exactly one stamped op:
+Command: `comfy workflow insert-workflow <file> <template>`, where `<template>`
+may be `-` to read the template payload from stdin. The CLI preserves the
+template payload verbatim and emits exactly one stamped op:
 
 ```json
 {

From b643aac6042a06357c12b7daafb70cee03639c9c Mon Sep 17 00:00:00 2001
From: bymyself <cbyrne@comfy.org>
Date: Thu, 17 Sep 2026 13:10:45 -0700
Subject: [PATCH 13/13] fix(workflow): summarize non-batchable apply errors

---
 comfy_cli/command/workflow_edit.py       |  8 +++++++-
 comfy_cli/workflow_ops.py                |  3 +++
 tests/comfy_cli/command/test_ack_flag.py | 26 ++++++++++++++++++++++++
 3 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py
index 307aa74f2..7279ccaff 100644
--- a/comfy_cli/command/workflow_edit.py
+++ b/comfy_cli/command/workflow_edit.py
@@ -796,7 +796,13 @@ def apply_cmd(
     except workflow_ops.NotBatchableError as e:
         # A standalone-only op (clear) inside the batch: its own registered code,
         # with the hint naming the standalone command to run instead.
-        renderer.error(code=e.code, message=f"batch failed: {e}", hint=e.hint)
+        details = None
+        if ack == "summary":
+            details = {
+                "failed": {"index": e.spec_index, "op": e.spec_op, "code": e.code},
+                "applied_count": e.applied_count,
+            }
+        renderer.error(code=e.code, message=f"batch failed: {e}", hint=e.hint, details=details)
         raise typer.Exit(code=1) from e
     except workflow_ops.DeprecatedNodeType as e:
         renderer.error(
diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 2e70edb5f..43b78417c 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -154,6 +154,9 @@ def __init__(self, index: int, kind: str = "clear"):
         command = entry["command"]
         self.code = entry["code"]
         self.kind = kind
+        self.spec_index = index
+        self.spec_op = kind
+        self.applied_count = 0
         self.hint = f"run the standalone `{command}` first, then apply the remaining ops as a batch"
         super().__init__(
             f"spec #{index}: `{kind}` {entry['does']} and is standalone-only (op-vocabulary-v1: "
diff --git a/tests/comfy_cli/command/test_ack_flag.py b/tests/comfy_cli/command/test_ack_flag.py
index 86ca78f31..f2c282297 100644
--- a/tests/comfy_cli/command/test_ack_flag.py
+++ b/tests/comfy_cli/command/test_ack_flag.py
@@ -186,6 +186,32 @@ def run_once(extra: list[str]) -> str:
 
 
 class TestAckSummaryPartialFailure:
+    def test_ack_summary_not_batchable_reports_index_and_code(self, patched_graph, tmp_path, capsys):
+        path = _empty(tmp_path)
+        ops_path = _ops_file(tmp_path, [{"op": "insert_workflow", "workflow": {}}])
+
+        env = _run(["apply", str(path), "--ops", str(ops_path), "--ack", "summary"], capsys)
+
+        assert env["ok"] is False
+        assert env["error"]["code"] == "workflow_insert_workflow_not_batchable"
+        assert env["error"]["details"] == {
+            "failed": {
+                "index": 0,
+                "op": "insert_workflow",
+                "code": "workflow_insert_workflow_not_batchable",
+            },
+            "applied_count": 0,
+        }
+
+    def test_full_mode_not_batchable_failure_envelope_unchanged(self, patched_graph, tmp_path, capsys):
+        path = _empty(tmp_path)
+        ops_path = _ops_file(tmp_path, [{"op": "insert_workflow", "workflow": {}}])
+
+        env = _run(["apply", str(path), "--ops", str(ops_path)], capsys)
+
+        assert env["error"]["code"] == "workflow_insert_workflow_not_batchable"
+        assert env["error"]["details"] is None
+
     def test_ack_summary_partial_failure_reports_index_and_code(self, patched_graph, tmp_path, capsys):
         """Op 2 of 3 (0-based index 1) fails → same code/atomicity as full
         mode, plus a structured receipt: failed.{index,op,code} + applied_count.