diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py
index 850bd918..ad5cf94b 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 39e03974..7279ccaf 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)
 
@@ -137,6 +145,34 @@ 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="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,
+):
+    """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:
+        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)
+    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")
+
+
 @tracking.track_command("workflow")
 def add_node_cmd(
     file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON.")],
@@ -760,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/discovery.py b/comfy_cli/discovery.py
index d7af9805..03c0aa47 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/error_codes.py b/comfy_cli/error_codes.py
index d5de7ac8..c7b81db3 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/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md
index 4517faa3..473427d6 100644
--- a/comfy_cli/skills/comfy/SKILL.md
+++ b/comfy_cli/skills/comfy/SKILL.md
@@ -352,14 +352,17 @@ 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
-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.*
@@ -368,8 +371,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 +388,8 @@ 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
+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/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py
index 9afa0a26..43b78417 100644
--- a/comfy_cli/workflow_ops.py
+++ b/comfy_cli/workflow_ops.py
@@ -91,12 +91,20 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str
 # ---------------------------------------------------------------------------
 
 #: Every op kind in the v1 vocabulary, including defined-but-deferred kinds.
-FROZEN_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node", "clear", "reset_doc")
-
-#: Kinds frozen in the contract whose replay is not implemented yet.
-#: ``apply_op`` must keep rejecting these. Empty since amendment v1.1:
-#: ``reset_doc`` was un-deferred by the bulk-writers ticket (V1-038).
-DEFERRED_OPS: tuple[str, ...] = ()
+FROZEN_OPS: tuple[str, ...] = (
+    "add_node",
+    "connect",
+    "set_widget",
+    "delete_node",
+    "clear",
+    "reset_doc",
+    "insert_workflow",
+)
+
+#: 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
@@ -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",
+    },
 }
 
 
@@ -141,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: "
@@ -1295,6 +1311,31 @@ def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int =
     return ops
 
 
+def insert_workflow(
+    workflow: dict,
+    template: dict,
+    *,
+    actor: str = "cli",
+    base_version: int = 0,
+) -> tuple[dict, dict]:
+    """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 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")
+    return workflow, _new_op("insert_workflow", actor, base_version, workflow=copy.deepcopy(template))
+
+
 def delete_node(
     workflow: dict,
     graph,
@@ -1795,6 +1836,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
diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md
index 69edcf4a..9ef7f59c 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,33 @@ 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>`, 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
+{
+  "op": "insert_workflow",
+  "op_id": "<uuid4 hex>",
+  "actor": "cli",
+  "base_version": 0,
+  "stamp": [0, "cli"],
+  "workflow": {"nodes": [], "links": [], "groups": [], "definitions": {"subgraphs": []}}
+}
+```
+
+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
+`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/command/test_ack_flag.py b/tests/comfy_cli/command/test_ack_flag.py
index 86ca78f3..f2c28229 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.
diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py
index fddbb4aa..5cf3dfaf 100644
--- a/tests/comfy_cli/command/test_workflow_edit.py
+++ b/tests/comfy_cli/command/test_workflow_edit.py
@@ -16,8 +16,10 @@
 import json
 from pathlib import Path
 from typing import Any
+from unittest.mock import Mock
 
 import pytest
+import typer
 from typer.testing import CliRunner
 
 from comfy_cli import workflow_ops
@@ -529,6 +531,80 @@ 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")
+        original = workflow.read_text(encoding="utf-8")
+        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": []}
+        assert workflow.read_text(encoding="utf-8") == original
+
+    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
+
+    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, {"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
+        assert "missing required field: nodes" in capsys.readouterr().out
+
+
 class TestAddNode:
     def test_adds_node_and_emits_op(self, patched_graph, tmp_path, capsys):
         path = _write(tmp_path, _base_workflow())
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 00000000..ac2a85d4
--- /dev/null
+++ b/tests/comfy_cli/test_insert_workflow_op.py
@@ -0,0 +1,128 @@
+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"]],
+        "groups": [],
+        "definitions": {"subgraphs": [{"id": "def-2", "nodes": [{"id": 5, "type": "Inner"}], "links": []}]},
+    }
+
+
+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)
+
+    assert op["op"] == "insert_workflow"
+    assert op["actor"] == "agent"
+    assert op["stamp"] == [4, "agent"]
+    assert op["workflow"] == template
+    assert result is live
+
+
+def test_insert_workflow_emits_semantically_invalid_json_for_server_validation():
+    live = {"nodes": [], "links": []}
+    template = {
+        "nodes": [{"id": 10}],
+        "links": [[20, 10, 0, 999, 0]],
+        "groups": [],
+        "definitions": {"subgraphs": [{}]},
+    }
+
+    _, op = workflow_ops.insert_workflow(live, template)
+
+    assert op["workflow"] == template
+
+
+def test_insert_workflow_rejects_missing_nodes():
+    template = _template()
+    del template["nodes"]
+
+    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": []}
+
+    _, 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_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)
+
+    result, _ = workflow_ops.insert_workflow(live, _template())
+
+    assert result is live
+    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():
+    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": []}