diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py index 850bd918b..602d8fc8f 100644 --- a/comfy_cli/command/workflow.py +++ b/comfy_cli/command/workflow.py @@ -1934,6 +1934,9 @@ def validate_cmd( from comfy_cli.command import workflow_edit as _wedit # noqa: E402 +app.command("define-subgraph", help="Create a subgraph definition; emits one define_subgraph op.")( + _wedit.define_subgraph_cmd +) app.command("add-node", help="Add a node to the graph; emits an add_node op.")(_wedit.add_node_cmd) app.command("connect", help="Wire an output slot to an input slot; emits a connect op.")(_wedit.connect_cmd) app.command("set-widget", help="Set a widget by name (`.`); 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..db16491f5 100644 --- a/comfy_cli/command/workflow_edit.py +++ b/comfy_cli/command/workflow_edit.py @@ -14,6 +14,7 @@ from __future__ import annotations import json +import stat from pathlib import Path from typing import Annotated, Any @@ -48,6 +49,7 @@ HostOpt = Annotated[str | None, typer.Option(show_default=False)] PortOpt = Annotated[int | None, typer.Option(show_default=False)] WhereOpt = Annotated[str | None, typer.Option("--where", show_default=False, help="Catalog target: local | cloud.")] +_MAX_DEFINITION_BYTES = 16 * 1024 * 1024 def _emit_edit_error(renderer, e: ValueError, *, hint: str) -> None: @@ -128,10 +130,69 @@ def _finish(renderer, p, workflow: dict, op: dict, base_version: int, stdout: bo renderer.emit(payload, command=command, changed=not stdout) +def _emit_op(renderer, p: Path, op: dict, base_version: int, command: str) -> None: + """Emit an op without applying it or writing the source workflow.""" + payload = {"workflow": str(p), "op": op, "base_version": base_version, "wrote": None} + if renderer.is_pretty(): + rprint(f"[bold green]✓[/bold green] {op['op']} emitted for [dim]{p}[/dim]") + renderer.emit(payload, command=command, changed=False) + + def _graph_or_exit(input_path, host, port, renderer, where=None): return _get_graph(input_path, host, port, where=where) +# --------------------------------------------------------------------------- +# define-subgraph +# --------------------------------------------------------------------------- + + +def _read_subgraph_definition(path: Path): + """Read a bounded regular JSON file without blocking on devices or FIFOs.""" + file_stat = path.stat() + if not stat.S_ISREG(file_stat.st_mode): + raise ValueError("subgraph definition must be a regular file") + if file_stat.st_size > _MAX_DEFINITION_BYTES: + raise ValueError(f"subgraph definition is too large (maximum {_MAX_DEFINITION_BYTES} bytes)") + try: + with path.open("rb") as definition_file: + raw = definition_file.read(_MAX_DEFINITION_BYTES + 1) + if len(raw) > _MAX_DEFINITION_BYTES: + raise ValueError(f"subgraph definition is too large (maximum {_MAX_DEFINITION_BYTES} bytes)") + return json.loads(raw.decode("utf-8")) + except (RecursionError, MemoryError) as error: + raise ValueError("subgraph definition is too deeply nested or too large") from error + + +@tracking.track_command("workflow") +def define_subgraph_cmd( + file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON to update.")], + definition_file: Annotated[str, typer.Argument(help="Serializable subgraph definition JSON.")], + subgraph_id: Annotated[str | None, typer.Option("--id", show_default=False)] = None, + actor: ActorOpt = "cli", + base_version: BaseVersionOpt = 0, + stdout: StdoutOpt = False, +): + """Create one subgraph definition and emit one ``define_subgraph`` op.""" + renderer = get_renderer() + renderer.command = "workflow define-subgraph" + p, workflow = _load_workflow_or_fail(renderer, file) + try: + definition_path = Path(definition_file).expanduser() + definition = _read_subgraph_definition(definition_path) + _, op = workflow_ops.define_subgraph( + workflow, + definition, + subgraph_id=subgraph_id, + actor=actor, + base_version=base_version, + ) + except (OSError, json.JSONDecodeError, UnicodeDecodeError, ValueError, RecursionError, MemoryError) as e: + _emit_edit_error(renderer, e, hint="provide a serializable subgraph definition JSON object") + raise typer.Exit(code=1) from e + _emit_op(renderer, p, op, base_version, "workflow define-subgraph") + + # --------------------------------------------------------------------------- # add-node # --------------------------------------------------------------------------- @@ -754,6 +815,11 @@ def apply_cmd( raise typer.Exit(code=1) from e try: + deferred = [ + spec.get("op") for spec in specs if isinstance(spec, dict) and spec.get("op") in workflow_ops.DEFERRED_OPS + ] + if deferred: + raise ValueError(f"local apply cannot persist deferred operation(s): {', '.join(deferred)}") workflow, ops, aliases = workflow_ops.apply_specs( workflow, graph, specs, actor=actor, base_version=base_version ) @@ -917,6 +983,13 @@ def foreach_cmd( out.mkdir(parents=True, exist_ok=True) written: list[str] = [] try: + deferred = [ + spec.get("op") + for spec in specs_template + if isinstance(spec, dict) and spec.get("op") in workflow_ops.DEFERRED_OPS + ] + if deferred: + raise ValueError(f"local foreach cannot persist deferred operation(s): {', '.join(deferred)}") for i, pset in enumerate(param_sets): if not isinstance(pset, dict): raise workflow_ops.RecipeError(f"param-set #{i} must be a JSON object") diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py index 9afa0a262..c06677f10 100644 --- a/comfy_cli/workflow_ops.py +++ b/comfy_cli/workflow_ops.py @@ -64,6 +64,7 @@ # New ids live in [2**40, 2**53): always large (never collides with small # frontend counter ids), always inside JS Number.MAX_SAFE_INTEGER. _ID_FLOOR = 1 << 40 +_SUBGRAPH_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.I) def mint_id() -> int: @@ -91,17 +92,25 @@ def _new_op(kind: str, actor: str, base_version: int, **fields: Any) -> dict[str # --------------------------------------------------------------------------- #: Every op kind in the v1 vocabulary, including defined-but-deferred kinds. -FROZEN_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node", "clear", "reset_doc") - -#: Kinds frozen in the contract whose replay is not implemented yet. -#: ``apply_op`` must keep rejecting these. Empty since amendment v1.1: -#: ``reset_doc`` was un-deferred by the bulk-writers ticket (V1-038). -DEFERRED_OPS: tuple[str, ...] = () +FROZEN_OPS: tuple[str, ...] = ( + "add_node", + "connect", + "set_widget", + "delete_node", + "clear", + "reset_doc", + "define_subgraph", +) + +#: Kinds frozen in the contract whose replay is not implemented in the CLI. +#: ``define_subgraph`` is emitted for cmp to validate and apply; the CLI must +#: keep rejecting local replay to preserve that ownership boundary. +DEFERRED_OPS: tuple[str, ...] = ("define_subgraph",) #: Kinds a batch (``apply_specs``) dispatches. ``clear`` and ``reset_doc`` are #: standalone-only: they rewrite the whole document, so they never ride inside #: an atomic batch. -BATCHABLE_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node") +BATCHABLE_OPS: tuple[str, ...] = ("add_node", "connect", "set_widget", "delete_node", "define_subgraph") #: Per-kind rendering for :class:`NotBatchableError` — the registered error code #: and the standalone command that DOES do the job. One entry per frozen kind @@ -1131,20 +1140,18 @@ class NotExpressibleError(ValueError): def _inexpressible_reason(workflow: dict) -> str | None: """Why ``workflow`` cannot be rebuilt from add_node/connect ops, or None. - The frozen vocabulary has four batchable kinds and none of them can create a - subgraph definition, a canvas group, or a reroute point — so a graph that - carries any of those is not reconstructible from ops, full stop. Enumerated + The frozen vocabulary cannot create a canvas group or a reroute point, so a + graph that carries either is not reconstructible from ops. Enumerated positively (a closed list of things we know we CAN'T do) rather than by trying and checking, so an unexpressible template fails before it has written anything. """ if not isinstance(workflow, dict) or not isinstance(workflow.get("nodes"), list): return "not a frontend-format workflow (no `nodes` list) — only the save/UI format can be op-ified" - definitions = workflow.get("definitions") - if isinstance(definitions, dict) and definitions.get("subgraphs"): - return "the workflow contains a subgraph definition, which no frozen op kind can create" if workflow.get("groups"): return "the workflow contains canvas groups, which no frozen op kind can create" + if (workflow.get("definitions") or {}).get("subgraphs"): + return "the workflow contains subgraph definitions, which only cmp can project into ops" extra = workflow.get("extra") if isinstance(extra, dict) and (extra.get("reroutes") or extra.get("linkExtensions")): return "the workflow contains reroute points, which no frozen op kind can create" @@ -1751,6 +1758,14 @@ def apply_specs( workflow, op = delete_node( workflow, graph, resolve_ref(spec["node"], aliases), actor=actor, base_version=base_version ) + elif kind == "define_subgraph": + workflow, op = define_subgraph( + workflow, + spec["subgraph_definition"], + subgraph_id=spec.get("subgraph_id"), + actor=actor, + base_version=base_version, + ) elif kind in _NOT_BATCHABLE: # In the frozen vocabulary but standalone-only — surfaced with # its own registered code so the caller learns the standalone @@ -1791,10 +1806,12 @@ def apply_specs( def apply_op(workflow: dict, op: dict, graph) -> dict: """Replay one op onto ``workflow`` in place and return it. Idempotent: an op whose ``op_id`` was already applied is a no-op.""" - applied = workflow.setdefault("_applied_ops", []) + applied = workflow.get("_applied_ops", []) if op["op_id"] in applied: return workflow kind = op["op"] + if kind != "define_subgraph" and any(key in op for key in ("subgraph_id", "subgraph_definition", "definitions")): + raise ValueError(f"malformed_op: {kind} cannot carry a subgraph definition") # Snapshot the LWW bookkeeping so an exception escaping a handler cannot # leave a stamp committed WITHOUT its op_id recorded below. That pairing is # the poison state: a retry of the identical op loses to the failed @@ -1828,6 +1845,45 @@ def apply_op(workflow: dict, op: dict, graph) -> dict: return workflow +def define_subgraph( + workflow: dict, + definition: dict, + *, + subgraph_id: str | None = None, + actor: str = "cli", + base_version: int = 0, +) -> tuple[dict, dict]: + """Emit a definition op; cmp owns semantic validation and application.""" + if not isinstance(definition, dict): + raise ValueError("subgraph definition must be a JSON object") + definition = copy.deepcopy(definition) + if subgraph_id is not None: + definition_id = subgraph_id + elif "id" in definition: + definition_id = definition["id"] + else: + definition_id = str(uuid.uuid4()) + if not isinstance(definition_id, str) or not definition_id: + raise ValueError("subgraph definition requires a non-empty string id") + if not _SUBGRAPH_UUID_RE.fullmatch(definition_id): + raise ValueError("subgraph definition id must be a valid UUID") + if "id" in definition and definition["id"] != definition_id: + raise ValueError("subgraph definition id must match --id") + definition["id"] = definition_id + try: + json.dumps(definition) + except (TypeError, ValueError, RecursionError) as error: + raise ValueError("subgraph definition must be JSON-serializable") from error + op = _new_op( + "define_subgraph", + actor, + base_version, + subgraph_id=definition_id, + subgraph_definition=definition, + ) + return workflow, op + + def _apply_add_node(workflow: dict, op: dict) -> None: # Node identity is compared as a STRING everywhere in the apply path # (amendment v1.2): ids are legitimately either JSON type, and an exact @@ -2258,6 +2314,8 @@ def _write_target(op: dict) -> tuple: # not share a target with its sibling (``model.reference_videos``). return ("input", str(op["to_node"]), "grow", _autogrow_base(str(grow["name"]))) return ("input", str(op["to_node"]), op["to_slot"]) + if kind == "define_subgraph": + return ("subgraph", str(op["subgraph_id"])) return (kind,) diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md index 69edcf4a0..8748297a4 100644 --- a/docs/op-vocabulary-v1.md +++ b/docs/op-vocabulary-v1.md @@ -17,7 +17,7 @@ citation must point at a commit on that branch. ## 1. Frozen op kinds -Six kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with +Seven kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with `ValueError("unknown op ...")` — it never ignores one. | Kind | Batchable | Standalone command | Summary | @@ -28,6 +28,7 @@ Six kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with | `delete_node` | yes | `comfy workflow delete` | Remove one node and its incident links | | `clear` | no | `comfy workflow clear` | Remove every node, link, and group | | `reset_doc` | no | `comfy workflow reset-doc --confirm` | Reset the whole document to an empty baseline | +| `define_subgraph` | yes | `comfy workflow define-subgraph` | Create one subgraph definition | Batchable = the kind is accepted by `apply_specs` (the `workflow apply` / `workflow foreach` batch surface). `clear` and `reset_doc` rewrite the whole @@ -178,6 +179,32 @@ pre-reset `base_version` do not replay across it. * Never emitted implicitly: no `--emit-ops` surface and no bulk writer (§8.8) mints one. It exists only where a caller asked for it by name. +### 1.7 `define_subgraph` + +Command: `comfy workflow define-subgraph [--id ]`. +The command performs envelope shape and JSON-serialization checks, assigns its +id from `--id`, the definition's `id`, or a new UUID, and emits exactly one +stamped op without modifying the local workflow: + +```json +{ + "op": "define_subgraph", + "op_id": "", + "actor": "cli", + "base_version": 0, + "stamp": [0, "cli"], + "subgraph_id": "", + "subgraph_definition": {"id": "", "nodes": [], "links": []} +} +``` + +Only this op may carry `subgraph_id` or definition payloads. The CLI does not +semantically validate, apply, replay, project, or resolve conflicts for the +definition. cmp owns those operations: reusing an id with different content +returns `definition_conflict`, and references to unknown ids follow cmp's +unknown-node failure path. The CLI surfaces server errors verbatim. Subsequent +interior edits use the existing id-addressed op scopes. + ## 2. Idempotency and identity * Every op carries `op_id`: uuid4 hex, minted by the **creator, before diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py index fddbb4aa4..f7fd33eb9 100644 --- a/tests/comfy_cli/command/test_workflow_edit.py +++ b/tests/comfy_cli/command/test_workflow_edit.py @@ -1385,6 +1385,22 @@ def test_batch_is_atomic_on_failure(self, patched_graph, tmp_path, capsys): assert env["error"]["code"] == "workflow_edit_invalid" assert path.read_text() == before, "failed batch must not write a partial graph" + def test_deferred_op_batch_is_not_persisted_locally(self, patched_graph, tmp_path, capsys): + path = self._empty(tmp_path) + before = path.read_text() + ops_path = tmp_path / "ops.json" + ops_path.write_text( + json.dumps( + [{"op": "define_subgraph", "subgraph_definition": {"id": "12345678-1234-4123-8123-123456789abc"}}] + ) + ) + + env = _run(["apply", str(path), "--ops", str(ops_path)], capsys) + + assert env["ok"] is False + assert "deferred operation" in env["error"]["message"] + assert path.read_text() == before + def test_duplicate_alias_is_rejected(self, patched_graph, tmp_path, capsys): """A repeated `as` name would silently clobber the earlier node — reject it.""" path = self._empty(tmp_path) @@ -1604,6 +1620,27 @@ def test_foreach_bad_param_set_fails(self, patched_graph, tmp_path, capsys): assert env["ok"] is False assert "positive" in env["error"]["message"] + def test_foreach_rejects_deferred_ops_before_writing(self, patched_graph, tmp_path, capsys): + rp = tmp_path / "deferred.json" + rp.write_text( + json.dumps( + { + "ops": [ + {"op": "define_subgraph", "subgraph_definition": {"id": "12345678-1234-4123-8123-123456789abc"}} + ] + } + ) + ) + params = tmp_path / "sets.json" + params.write_text("[{}]") + out = tmp_path / "out" + + env = _run(["foreach", str(rp), "--params", str(params), "--out-dir", str(out)], capsys) + + assert env["ok"] is False + assert "deferred operation" in env["error"]["message"] + assert not list(out.glob("*.json")) + def test_foreach_surfaces_partial_writes_on_mid_batch_failure(self, patched_graph, tmp_path, capsys): """foreach writes per param-set; a mid-batch failure leaves earlier files on disk, so the error must surface them (not leave the caller blind).""" diff --git a/tests/comfy_cli/test_define_subgraph_op.py b/tests/comfy_cli/test_define_subgraph_op.py new file mode 100644 index 000000000..5aa7b7679 --- /dev/null +++ b/tests/comfy_cli/test_define_subgraph_op.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import copy +import json +from pathlib import Path + +import pytest +import typer + +from comfy_cli import workflow_ops +from comfy_cli.command import workflow as workflow_cmd # noqa: F401 -- initializes the edit command cycle +from comfy_cli.command import workflow_edit + +SUBGRAPH_ID = "12345678-1234-4123-8123-123456789abc" + + +def _definition() -> dict: + return { + "id": SUBGRAPH_ID, + "name": "One", + "inputs": [], + "outputs": [], + "nodes": [{"id": 10, "type": "Inner"}], + "links": [], + } + + +def test_define_subgraph_emits_cmp_payload_without_mutating_workflow(): + workflow = {"nodes": [{"id": 1, "type": "Existing"}], "links": []} + before = copy.deepcopy(workflow) + definition = _definition() + + result, op = workflow_ops.define_subgraph(workflow, definition, actor="agent", base_version=4) + + assert result is workflow + assert workflow == before + assert op == { + "op": "define_subgraph", + "op_id": op["op_id"], + "actor": "agent", + "base_version": 4, + "stamp": [4, "agent"], + "subgraph_id": SUBGRAPH_ID, + "subgraph_definition": definition, + } + + +def test_define_subgraph_emits_semantically_odd_definition_for_cmp_validation(): + definition = {"id": SUBGRAPH_ID, "nodes": "future-cmp-shape", "links": {"also": "future"}} + + _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition) + + assert op["subgraph_definition"] == definition + + +def test_define_subgraph_rejects_local_replay(): + _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition()) + + with pytest.raises(ValueError, match="unknown op 'define_subgraph'"): + workflow_ops.apply_op({"nodes": [], "links": []}, op, None) + + +def test_define_subgraph_can_assign_an_explicit_new_id(): + definition = _definition() + definition.pop("id") + + _, op = workflow_ops.define_subgraph({"nodes": [], "links": []}, definition, subgraph_id=SUBGRAPH_ID) + + assert op["subgraph_id"] == SUBGRAPH_ID + assert op["subgraph_definition"]["id"] == SUBGRAPH_ID + + +@pytest.mark.parametrize( + "subgraph_id", ["12345678-1234-0123-8123-123456789abc", "12345678-1234-4123-7123-123456789abc"] +) +def test_define_subgraph_rejects_invalid_uuid_version_or_variant(subgraph_id): + with pytest.raises(ValueError, match="valid UUID"): + workflow_ops.define_subgraph({"nodes": [], "links": []}, _definition(), subgraph_id=subgraph_id) + + +def test_define_subgraph_command_emits_without_writing_workflow(tmp_path, monkeypatch): + workflow = tmp_path / "workflow.json" + original = {"nodes": [], "links": []} + workflow.write_text(json.dumps(original)) + definition_file = tmp_path / "definition.json" + definition_file.write_text(json.dumps(_definition())) + emitted = [] + + class Renderer: + command = "" + + def is_pretty(self): + return False + + def emit(self, payload, **kwargs): + emitted.append((payload, kwargs)) + + monkeypatch.setattr(workflow_edit, "get_renderer", Renderer) + + workflow_edit.define_subgraph_cmd(str(workflow), str(definition_file)) + + assert json.loads(workflow.read_text()) == original + assert emitted[0][0]["op"]["op"] == "define_subgraph" + assert emitted[0][0]["wrote"] is None + assert emitted[0][1]["changed"] is False + + +def test_define_subgraph_command_rejects_non_regular_definition_file(tmp_path, monkeypatch): + workflow = tmp_path / "workflow.json" + workflow.write_text(json.dumps({"nodes": [], "links": []})) + definition_dir = tmp_path / "definition" + definition_dir.mkdir() + errors = [] + + class Renderer: + command = "" + + def error(self, **kwargs): + errors.append(kwargs) + + monkeypatch.setattr(workflow_edit, "get_renderer", Renderer) + with pytest.raises(typer.Exit): + workflow_edit.define_subgraph_cmd(str(workflow), str(definition_dir)) + + assert "regular file" in errors[0]["message"] + + +def test_definition_reader_rejects_oversized_files(tmp_path): + path = tmp_path / "large.json" + path.write_bytes(b" " * (workflow_edit._MAX_DEFINITION_BYTES + 1)) + + with pytest.raises(ValueError, match="too large"): + workflow_edit._read_subgraph_definition(Path(path))