diff --git a/comfy_cli/command/templates.py b/comfy_cli/command/templates.py index 7e87ffcdc..15db37dbd 100644 --- a/comfy_cli/command/templates.py +++ b/comfy_cli/command/templates.py @@ -28,7 +28,7 @@ import typer -from comfy_cli import tracking +from comfy_cli import tracking, workflow_ops from comfy_cli.file_utils import atomic_write_bytes from comfy_cli.http import ResponseTooLarge, plain_urlopen, read_capped from comfy_cli.output import get_renderer, rprint @@ -633,6 +633,22 @@ def fetch_cmd( bool, typer.Option("--refresh", help="Re-fetch the gallery index from GitHub before resolving."), ] = False, + emit_ops: Annotated[ + bool, + typer.Option( + "--emit-ops", + help=( + "Also emit `ops`: the stamped op batch that turns the file being replaced INTO this " + "template (delete_node + add_node + connect, frozen vocabulary). Replays through a merge " + "consumer AND is a legal `comfy workflow apply --ops` batch. Omitted, with `ops_skipped` " + "saying why, for templates the vocabulary cannot express (subgraphs, groups)." + ), + ), + ] = False, + actor: Annotated[str, typer.Option("--actor", help="Op author id for --emit-ops (CRDT stamping).")] = "cli", + base_version: Annotated[ + int, typer.Option("--base-version", help="Draft version the emitted ops are stamped against.") + ] = 0, ): renderer = get_renderer() @@ -698,6 +714,18 @@ def fetch_cmd( ) raise typer.Exit(code=1) from e + # The graph this fetch is REPLACING, read before the write clobbers it — + # `--emit-ops` needs it to emit the delete_node half of the batch. Only read + # when asked: an unparseable file at the target is not an error for a plain + # fetch (it is about to be overwritten), so it must not become one here. + previous: dict[str, Any] = {} + if emit_ops and out: + try: + loaded = json.loads(Path(out).expanduser().read_text(encoding="utf-8")) + previous = loaded if isinstance(loaded, dict) else {} + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + previous = {} + if out: out_path = Path(out).expanduser() out_path.parent.mkdir(parents=True, exist_ok=True) @@ -727,6 +755,17 @@ def fetch_cmd( # can get the workflow — emit() owns stdout in JSON mode, so without this # the fetch would produce nothing but metadata. payload["workflow"] = wf + if emit_ops: + # A bulk writer that emits ops stops being a whole-document replacement: + # the consumer folds the batch into the document it already has, so the + # replaced canvas keeps ONE identity and an attributed history instead of + # being re-seeded (op-vocabulary-v1 §8.6). Failure is NOT fatal — the + # fetch itself succeeded and the file is written; the consumer falls back + # to whatever it did before ops existed, and `ops_skipped` says why. + try: + payload["ops"] = workflow_ops.replace_ops(previous, wf, actor=actor, base_version=base_version) + except workflow_ops.NotExpressibleError as e: + payload["ops_skipped"] = str(e) if renderer.is_pretty() and out: rprint(f"[green]✓[/green] wrote {len(body):,} bytes ({payload['node_count']} nodes) to {target_repr}") renderer.emit(payload, command="templates fetch") diff --git a/comfy_cli/command/workflow.py b/comfy_cli/command/workflow.py index 54f9a4076..499ae2dcc 100644 --- a/comfy_cli/command/workflow.py +++ b/comfy_cli/command/workflow.py @@ -1483,6 +1483,10 @@ def delete_cmd( app.command("set-widget", help="Set a widget by name (`.`); emits a set_widget op.")(_wedit.set_widget_cmd) app.command("delete-node", help="Delete a node and its links; emits a delete_node op.")(_wedit.delete_cmd) app.command("clear", help="Remove every node, link, and group; emits one clear op.")(_wedit.clear_cmd) +app.command( + "reset-doc", + help="Reset the document to the empty baseline — nodes, ids AND replay history. Requires --confirm.", +)(_wedit.reset_doc_cmd) app.command("ls-nodes", help="List nodes (id/type/title) in a workflow file.")(_wedit.ls_nodes_cmd) app.command("apply", help="Apply a recipe / batch of edits in one pass; supports node aliases + --param.")( _wedit.apply_cmd diff --git a/comfy_cli/command/workflow_edit.py b/comfy_cli/command/workflow_edit.py index f84cc603d..b2657b62d 100644 --- a/comfy_cli/command/workflow_edit.py +++ b/comfy_cli/command/workflow_edit.py @@ -282,6 +282,54 @@ def clear_cmd( _finish(renderer, p, workflow, op, base_version, stdout, "workflow clear") +# --------------------------------------------------------------------------- +# reset-doc — the guarded document reset (op-vocabulary-v1 §1.6) +# --------------------------------------------------------------------------- + + +@tracking.track_command("workflow") +def reset_doc_cmd( + file: Annotated[str, typer.Argument(help="Frontend-format workflow JSON.")], + confirm: Annotated[ + bool, + typer.Option( + "--confirm", + help="REQUIRED. Without it the command fails closed and writes nothing.", + ), + ] = False, + actor: ActorOpt = "cli", + base_version: BaseVersionOpt = 0, + stdout: StdoutOpt = False, + where: WhereOpt = None, # accepted for caller uniformity; reset needs no catalog +): + """Reset the document to the empty baseline — nodes, links, groups, ids AND + the applied-op history. + + Guarded, unlike every other edit command, because it is the only one whose + effect no later op can undo: it is a history barrier, so ops minted against + a pre-reset base_version do not replay across it. The check runs BEFORE the + file is read, so an unconfirmed call cannot even fail halfway. + """ + renderer = get_renderer() + renderer.command = "workflow reset-doc" + if not confirm: + renderer.error( + code="workflow_reset_doc_unconfirmed", + message=( + "`workflow reset-doc` erases every node AND the document's replay history; " + "it requires an explicit --confirm. Nothing was written." + ), + hint=( + "re-run with --confirm if that is really what you want — otherwise " + "`comfy workflow clear ` empties the graph while keeping the document's history" + ), + ) + raise typer.Exit(code=1) + p, workflow = _load_workflow_or_fail(renderer, file) + workflow, op = workflow_ops.reset_doc(workflow, actor=actor, base_version=base_version) + _finish(renderer, p, workflow, op, base_version, stdout, "workflow reset-doc") + + # --------------------------------------------------------------------------- # Litegraph node modes worth surfacing on ls-nodes. 0 (always) and 1 (on-event) # are normal execution and are deliberately unlabeled. Mirrors workflow_to_api's diff --git a/comfy_cli/error_codes.py b/comfy_cli/error_codes.py index 22f133911..53f85e3f7 100644 --- a/comfy_cli/error_codes.py +++ b/comfy_cli/error_codes.py @@ -520,6 +520,20 @@ class ErrorCode: "rejected atomically — nothing was applied.", "run the standalone `comfy workflow clear ` first, then apply the remaining ops as a batch", ), + ErrorCode( + "workflow_reset_doc_not_batchable", + "A batch (`workflow apply` / `workflow foreach`) contained a `reset_doc` op. `reset_doc` resets the " + "whole document to the empty baseline and erases its replay history, so it is standalone-only " + "(docs/op-vocabulary-v1.md: batchable = no) and the batch was rejected atomically — nothing was applied.", + "run the standalone `comfy workflow reset-doc --confirm` first, then apply the remaining ops as a batch", + ), + ErrorCode( + "workflow_reset_doc_unconfirmed", + "`comfy workflow reset-doc` was called without `--confirm`. The command fails closed: it erases every " + "node AND the document's replay history, which no later op can undo.", + "re-run with `--confirm` if that is really what you want — otherwise `comfy workflow clear ` " + "empties the graph while keeping the document's history", + ), ErrorCode( "normalized_value", "Warning (not fatal): a set-widget value wasn't an exact COMBO option, so " diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py index c0fa6f416..d7ed9485b 100644 --- a/comfy_cli/workflow_ops.py +++ b/comfy_cli/workflow_ops.py @@ -92,16 +92,32 @@ 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 -#: (``reset_doc`` is specified in op-vocabulary-v1.md; implementation is -#: deferred to the bulk-writers ticket). ``apply_op`` must keep rejecting these. -DEFERRED_OPS: tuple[str, ...] = ("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, ...] = () #: 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") +#: Per-kind rendering for :class:`NotBatchableError` — the registered error code +#: and the standalone command that DOES do the job. One entry per frozen kind +#: outside ``BATCHABLE_OPS``; the contract test pins that correspondence. +_NOT_BATCHABLE: dict[str, dict[str, str]] = { + "clear": { + "code": "workflow_clear_not_batchable", + "command": "comfy workflow clear ", + "does": "wipes the whole graph", + }, + "reset_doc": { + "code": "workflow_reset_doc_not_batchable", + "command": "comfy workflow reset-doc --confirm", + "does": "resets the whole document to the empty baseline and erases its replay history", + }, +} + class NotBatchableError(ValueError): """A frozen op kind that is standalone-only was submitted inside a batch. @@ -110,16 +126,25 @@ class NotBatchableError(ValueError): (see ``comfy_cli/error_codes.py``) instead of the generic ``workflow_edit_invalid``, so a caller learns the exact standalone command to run rather than re-trying the batch. + + ``code``/``hint`` are per-kind INSTANCE attributes; the class attributes are + the ``clear`` values, kept so existing callers that read + ``NotBatchableError.code`` off the class still resolve. """ code = "workflow_clear_not_batchable" hint = "run the standalone `comfy workflow clear ` first, then apply the remaining ops as a batch" - def __init__(self, index: int): + def __init__(self, index: int, kind: str = "clear"): + entry = _NOT_BATCHABLE.get(kind, _NOT_BATCHABLE["clear"]) + command = entry["command"] + self.code = entry["code"] + self.kind = kind + self.hint = f"run the standalone `{command}` first, then apply the remaining ops as a batch" super().__init__( - f"spec #{index}: `clear` wipes the whole graph and is standalone-only (op-vocabulary-v1: " + f"spec #{index}: `{kind}` {entry['does']} and is standalone-only (op-vocabulary-v1: " "batchable = no) — it never rides inside a batch. No changes were applied — the batch was " - "discarded. Run `comfy workflow clear ` as its own command, then apply the remaining ops." + f"discarded. Run `{command}` as its own command, then apply the remaining ops." ) @@ -843,6 +868,207 @@ def clear(workflow: dict, *, actor: str = "cli", base_version: int = 0) -> tuple return apply_op(workflow, op, None), op +def reset_doc(workflow: dict, *, actor: str = "cli", base_version: int = 0) -> tuple[dict, dict]: + """Reset the whole document to the empty baseline (op-vocabulary-v1 §1.6). + + Not ``clear``. ``clear`` empties the graph but PRESERVES the id high-water + marks and the applied-op bookkeeping, so it is an ordinary edit that merges + with concurrent ops. ``reset_doc`` drops those too: it is a **history + barrier**, and ops minted against a pre-reset ``base_version`` do not replay + across it. + + That is why the CLI surface guards it behind an explicit ``--confirm`` and + why it is standalone-only — there is no safe way to fold "forget everything + that ever applied" into the middle of a batch. + """ + removed = [n.get("id") for n in workflow.get("nodes") or [] if isinstance(n, dict)] + op = _new_op("reset_doc", actor, base_version, removed_nodes=removed) + return apply_op(workflow, op, None), op + + +# --------------------------------------------------------------------------- +# Bulk writers — expressing a whole-file replacement as ops (V1-038) +# --------------------------------------------------------------------------- + + +class NotExpressibleError(ValueError): + """A graph uses structure the frozen v1 vocabulary cannot express. + + Raised by :func:`replace_ops` INSTEAD of returning a partial batch. A + partial batch is the dangerous answer: it applies cleanly and leaves a + document that is not the graph the caller asked for. The caller is expected + to fall back to whatever whole-document path it had before (the cloud + agent re-mints), and to say why. + """ + + +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 + positively (a closed list of things we know we CAN'T do) rather than by + trying and checking, so an unexpressible template fails before it has + written anything. + """ + if not isinstance(workflow, dict) or not isinstance(workflow.get("nodes"), list): + return "not a frontend-format workflow (no `nodes` list) — only the save/UI format can be op-ified" + definitions = workflow.get("definitions") + if isinstance(definitions, dict) and definitions.get("subgraphs"): + return "the workflow contains a subgraph definition, which no frozen op kind can create" + if workflow.get("groups"): + return "the workflow contains canvas groups, which no frozen op kind can create" + extra = workflow.get("extra") + 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" + for node in workflow["nodes"]: + if not isinstance(node, dict) or node.get("id") is None or not node.get("type"): + return "the workflow contains a node with no id or no type" + for link in workflow.get("links") or []: + if not isinstance(link, list) or len(link) < 5: + return "the workflow contains a link that is not a [id, from, from_slot, to, to_slot, type] tuple" + return None + + +def _slot_ref(node: dict, slots_key: str, index: Any, alias: str) -> str: + """`$alias.` for a spec-form connect, preferring the slot NAME. + + Names are the canonical reference form and survive slot reordering; the + index is the fallback for a node whose slot list the template omits. + ``_split_ref_slot`` partitions on the FIRST dot, so a name containing one + would resolve wrong — those fall back to the index too. + """ + slots = node.get(slots_key) + if isinstance(slots, list) and isinstance(index, int) and 0 <= index < len(slots): + name = (slots[index] or {}).get("name") if isinstance(slots[index], dict) else None + if isinstance(name, str) and name and "." not in name: + return f"${alias}.{name}" + return f"${alias}.{index}" + + +def _alias_for(class_type: str, used: dict[str, int]) -> str: + """A deterministic, batch-unique alias for a node — `ksampler`, `ksampler_2`.""" + base = re.sub(r"[^a-z0-9_]", "", str(class_type).lower()) or "node" + used[base] = used.get(base, 0) + 1 + return base if used[base] == 1 else f"{base}_{used[base]}" + + +def replace_ops(old: dict, new: dict, *, actor: str = "cli", base_version: int = 0) -> list[dict]: + """The stamped op batch that turns ``old`` into ``new``. + + This is what makes a BULK WRITER (a template fetch, a saved-workflow open) + an attributed, incremental edit instead of a whole-document replacement. + Without it the only way to land a replaced canvas in a shared document is to + re-seed it — and §8.6 is explicit that independently re-seeding a base is + the one thing a replica must never do, because the duplicate identities + only show up on the first merge. + + Shape: ``delete_node`` for everything currently in ``old`` (in order), then + ``add_node`` for every node in ``new``, then ``connect`` for every link. + Widget values need no ``set_widget`` ops — they ride inside the ``add_node`` + payload, which §8.5 makes authoritative at replay. + + **Identity is re-minted, never inherited.** Template graphs are numbered + from small frontend counters (1, 2, 3…); replaying those ids into a live + document would reuse identities a concurrent replica may still hold, which + §1.5 calls out as letting a merge resurrect a deleted node. Every node and + link gets a fresh ``mint_id`` and every interior reference is remapped onto + it. + + **Dual-shape on purpose.** Each returned dict is a fully minted op (``op_id`` + / ``actor`` / ``stamp`` + the kind's minted fields) AND carries that kind's + SPEC keys (``class_type``/``at``/``as``, ``from``/``to``, ``node``). So the + same array replays through :func:`apply_op` losslessly *and* is accepted + verbatim by :func:`apply_specs` — one artifact, both consumers. The two are + not equivalent: ``apply_specs`` re-mints each node from the live catalog, so + it reproduces the STRUCTURE (classes + wiring) while the op path reproduces + the graph exactly, widget values included. + + :raises NotExpressibleError: ``new`` uses structure no frozen op can create. + """ + reason = _inexpressible_reason(new) + if reason: + raise NotExpressibleError(reason) + + ops: list[dict] = [] + old_links = [link for link in (old.get("links") or []) if isinstance(link, list) and len(link) >= 5] + for node in old.get("nodes") or []: + if not isinstance(node, dict) or node.get("id") is None: + continue + nid = node["id"] + ops.append( + _new_op( + "delete_node", + actor, + base_version, + node_id=nid, + removed_links=[link[0] for link in old_links if link[1] == nid or link[3] == nid], + # spec key, so apply_specs dispatches the same entry + node=nid, + ) + ) + + node_ids: dict[Any, int] = {n["id"]: mint_id() for n in new["nodes"]} + link_ids: dict[Any, int] = {link[0]: mint_id() for link in (new.get("links") or [])} + aliases: dict[Any, str] = {} + used: dict[str, int] = {} + + for original in new["nodes"]: + node = copy.deepcopy(original) + node["id"] = node_ids[original["id"]] + for slot in node.get("inputs") or []: + if isinstance(slot, dict) and slot.get("link") is not None: + slot["link"] = link_ids.get(slot["link"]) + for slot in node.get("outputs") or []: + if isinstance(slot, dict) and isinstance(slot.get("links"), list): + slot["links"] = [link_ids[x] for x in slot["links"] if x in link_ids] + alias = _alias_for(original.get("type"), used) + aliases[original["id"]] = alias + pos = original.get("pos") + ops.append( + _new_op( + "add_node", + actor, + base_version, + node_id=node["id"], + class_type=original.get("type"), + pos=pos, + node=node, + # spec keys + **{"at": pos, "as": alias}, + ) + ) + + by_original_id = {n["id"]: n for n in new["nodes"]} + for link in new.get("links") or []: + lid, from_node, from_slot, to_node, to_slot = link[0], link[1], link[2], link[3], link[4] + if from_node not in node_ids or to_node not in node_ids: + # A link to a node the graph does not contain is already broken in + # the source; dropping it is the faithful translation of a graph the + # canvas would render with a dangling edge. + continue + ops.append( + _new_op( + "connect", + actor, + base_version, + link_id=link_ids[lid], + from_node=node_ids[from_node], + from_slot=from_slot, + to_node=node_ids[to_node], + to_slot=to_slot, + link_type=link[5] if len(link) > 5 else None, + # spec keys + **{ + "from": _slot_ref(by_original_id[from_node], "outputs", from_slot, aliases[from_node]), + "to": _slot_ref(by_original_id[to_node], "inputs", to_slot, aliases[to_node]), + }, + ) + ) + return ops + + def delete_node( workflow: dict, graph, @@ -1144,11 +1370,11 @@ def apply_specs( workflow, op = delete_node( workflow, graph, resolve_ref(spec["node"], aliases), actor=actor, base_version=base_version ) - elif kind == "clear": + elif kind in _NOT_BATCHABLE: # In the frozen vocabulary but standalone-only — surfaced with # its own registered code so the caller learns the standalone # command instead of a generic "unknown op". - raise NotBatchableError(i) + raise NotBatchableError(i, kind) else: raise ValueError(f"spec #{i}: unknown op {kind!r}") except KeyError as e: @@ -1185,9 +1411,16 @@ def apply_op(workflow: dict, op: dict, graph) -> dict: _apply_delete_node(workflow, op) elif kind == "clear": _apply_clear(workflow, op) + elif kind == "reset_doc": + _apply_reset_doc(workflow, op) else: raise ValueError(f"unknown op {kind!r}") - applied.append(op["op_id"]) + # NOT ``applied.append`` — ``_apply_reset_doc`` REPLACES ``_applied_ops`` + # with a fresh list (that is what makes it a history barrier), so the local + # binding above is stale for that kind and the reset's own op_id would be + # written into a discarded list. Re-read, so a re-delivered reset_doc is a + # no-op rather than a second wipe. + workflow.setdefault("_applied_ops", []).append(op["op_id"]) return workflow @@ -1382,6 +1615,32 @@ def _apply_clear(workflow: dict, op: dict) -> None: workflow["groups"] = [] +#: Document-identity keys a ``reset_doc`` keeps. Everything else is discarded: +#: the point of the op is that nothing from the old document survives it. The +#: id stays so the reset document is still THIS workflow, not a new one. +_RESET_DOC_KEEP = ("id",) + + +def _apply_reset_doc(workflow: dict, op: dict) -> None: + """Replace the whole document with the empty baseline, bookkeeping included. + + Unlike ``_apply_clear`` this drops ``last_node_id``/``last_link_id``, + ``_applied_ops`` and ``_widget_stamps`` — the history barrier of §1.6. Ids + are minted at random in ``[2**40, 2**53)`` (``mint_id``), never allocated + from the high-water marks, so resetting them to 0 cannot cause id reuse. + """ + kept = {k: workflow[k] for k in _RESET_DOC_KEEP if k in workflow} + workflow.clear() + workflow.update(kept) + workflow["nodes"] = [] + workflow["links"] = [] + workflow["groups"] = [] + workflow["last_node_id"] = 0 + workflow["last_link_id"] = 0 + workflow["_applied_ops"] = [] + workflow["_widget_stamps"] = {} + + # --------------------------------------------------------------------------- # conflict detection + canonicalization (for ask-to-merge / convergence checks) # --------------------------------------------------------------------------- diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md index 5c604599c..9af9facdf 100644 --- a/docs/op-vocabulary-v1.md +++ b/docs/op-vocabulary-v1.md @@ -27,13 +27,14 @@ Six kinds. No other kind is valid in v1: `apply_op` rejects an unknown kind with | `set_widget` | yes | `comfy workflow set-widget` | Set one widget value by name | | `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 | (deferred) | Reset the whole document to an empty baseline | +| `reset_doc` | no | `comfy workflow reset-doc --confirm` | Reset the whole document to an empty baseline | Batchable = the kind is accepted by `apply_specs` (the `workflow apply` / `workflow foreach` batch surface). `clear` and `reset_doc` rewrite the whole -document, so they are standalone-only: a batch containing `clear` is rejected -atomically with error code `workflow_clear_not_batchable` and a hint naming the -standalone `comfy workflow clear` command. Nothing from such a batch is applied. +document, so they are standalone-only: a batch containing either is rejected +atomically with its own registered error code — +`workflow_clear_not_batchable` / `workflow_reset_doc_not_batchable` — and a hint +naming the standalone command. Nothing from such a batch is applied. Every op carries the common envelope stamped by `_new_op`: @@ -141,19 +142,31 @@ monotonic — id reuse would let a merge resurrect a deleted node's identity. names the standalone command. * Idempotency: `op_id` no-op; clearing an empty document changes nothing. -### 1.6 `reset_doc` — standalone only, deferred - -Defined here; **implementation is deferred to the bulk-writers ticket**. -`apply_op` currently rejects it (`unknown op 'reset_doc'`), and the contract -tests pin that it stays rejected until it is un-deferred by amendment. - -Semantics when implemented: replace the entire document with the empty baseline, -including apply bookkeeping — unlike `clear`, which preserves the id high-water -marks and the applied-op history. Because it erases replay history, it is a -history barrier: ops minted against a pre-reset `base_version` do not replay -across it. Guard semantics: the CLI surface requires an explicit `--confirm` -flag; without it the command fails closed and applies nothing. Not batchable, -for the same reason as `clear`. +### 1.6 `reset_doc` — standalone only + +Command: `comfy workflow reset-doc --confirm`. Implemented by amendment +v1.1 (§10); `DEFERRED_OPS` is now empty. Minted op fields: `removed_nodes` (ids +present at mint time), same as `clear`. + +Replaces the entire document with the empty baseline, **including apply +bookkeeping** — unlike `clear`, which preserves the id high-water marks and the +applied-op history. `last_node_id` / `last_link_id` go to 0 (safe: ids come from +`mint_id`, never from the high-water marks — §8.3), `_applied_ops` and +`_widget_stamps` are dropped, and only the document `id` survives. Because it +erases replay history it is a **history barrier**: ops minted against a +pre-reset `base_version` do not replay across it. + +* Guard: the CLI surface requires an explicit `--confirm`; without it the + command fails closed with `workflow_reset_doc_unconfirmed` and writes nothing. + The check runs before the file is read, so an unconfirmed call cannot fail + halfway. It is the only edit command with a guard, because it is the only one + no later op can undo. +* Idempotency: the reset's own `op_id` is written into the freshly-emptied + `_applied_ops`, so a re-delivered `reset_doc` is a no-op, not a second wipe. +* Batchable: **no**, for the same reason as `clear` — rejected with + `workflow_reset_doc_not_batchable`. +* 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. ## 2. Idempotency and identity @@ -390,6 +403,43 @@ Current contract, pinned: to sibling instances, definition garbage collection) is owed before this document's v1.1, together with the FE stable-ID reconciliation (section 6). +### 8.8 Bulk writers emit ops, they do not re-seed + +A **bulk writer** is any command that replaces the working file wholesale rather +than editing it: `comfy templates fetch -o ` today, `workflow get -o` +next. Downstream, such a replacement used to become a new document — the +consumer re-minted a snapshot from the new file. §8.6 forbids exactly that for a +replica, and even for the store owner it throws away the attributed history the +op log exists to keep. + +`workflow_ops.replace_ops(old, new)` is the alternative, and `templates fetch +--emit-ops` is its first caller. The rules: + +* **Shape**: `delete_node` for every node in `old` (in order), then `add_node` + for every node in `new`, then `connect` for every link. No `set_widget` ops — + widget values ride inside the `add_node` payload, which §8.5 makes + authoritative. +* **Identity is re-minted, never inherited.** Template graphs are numbered from + small frontend counters; replaying those ids into a live document reuses + identities a concurrent replica may still hold (§1.5's resurrection hazard). + Every node and link gets a fresh `mint_id` and every interior reference + (`inputs[].link`, `outputs[].links`, the `links` tuples) is remapped onto it. +* **Dual shape.** Each emitted entry is a fully minted op (envelope + the kind's + minted fields) AND carries that kind's spec keys (`class_type`/`at`/`as`, + `from`/`to`, `node`). The same array therefore replays through `apply_op` + losslessly and is accepted verbatim by `apply_specs`. The two are not + equivalent: `apply_specs` re-mints each node from the live catalog, so it + reproduces the structure (classes + wiring) while the op path reproduces the + graph exactly, widget values included. +* **All or nothing.** A graph the vocabulary cannot express — a subgraph + definition, a canvas group, a reroute point, a malformed node or link — emits + **no ops at all** (`NotExpressibleError`, surfaced as `ops_skipped`), never a + partial batch. A partial batch applies cleanly and leaves a document that is + not the graph the caller asked for; the consumer is expected to keep its + whole-document fallback for these cases. +* **`reset_doc` is never part of a bulk batch** (§1.6). Replacing a canvas is + expressed as deletes + adds, which merge; a history barrier does not. + ## 9. Amendments * Post-freeze changes require a **versioned amendment section** appended to @@ -401,3 +451,26 @@ Current contract, pinned: * Adding, removing, or re-scoping an op kind requires updating `FROZEN_OPS` / `DEFERRED_OPS` / `BATCHABLE_OPS`, the dispatch tables, and this document in one commit — `tests/comfy_cli/test_op_vocabulary_contract.py` fails otherwise. + +## 10. Amendment v1.1 — 2026-08-12 (V1-038 / BE-7171) + +**`reset_doc` is un-deferred.** `DEFERRED_OPS` is now empty; `apply_op` +dispatches `reset_doc` and `apply_specs` rejects it as standalone-only with its +own registered code. §1.6 is rewritten from "semantics when implemented" to the +implemented contract, and the frozen table's standalone-command cell names +`comfy workflow reset-doc --confirm` instead of "(deferred)". No frozen kind was +added, removed, or re-scoped: `reset_doc` was already in `FROZEN_OPS` and +already `Batchable = no`. + +*Why now*: the bulk-writers ticket needed a real, guarded "start this document +over" primitive so that "replace the canvas" and "erase the document" stopped +being the same operation. They are now distinct: §8.8's bulk batch replaces the +canvas with merging deletes+adds, and `reset_doc` is the explicit, confirmed +barrier a caller asks for by name. + +**§8.8 is new** and normative for bulk writers (`replace_ops`, +`templates fetch --emit-ops`). It adds no op kind — it constrains how existing +kinds are minted for a whole-file replacement. + +**No change to §§2-7, 8.1-8.7.** Stamping, LWW, abort-remainder, aliases and +replication semantics are untouched. diff --git a/tests/comfy_cli/command/test_templates_fetch_emit_ops.py b/tests/comfy_cli/command/test_templates_fetch_emit_ops.py new file mode 100644 index 000000000..13ad8860c --- /dev/null +++ b/tests/comfy_cli/command/test_templates_fetch_emit_ops.py @@ -0,0 +1,285 @@ +"""``comfy templates fetch --emit-ops`` (V1-038 / BE-7171). + +``templates fetch -o workflow.json`` is a BULK WRITER: it replaces the working +file wholesale. Downstream (the cloud agent's document) that replacement had to +be expressed as a **re-mint** — a brand-new document with no attributed, +incremental history, and §8.6's "one common initial snapshot" rule makes an +independent re-seed the one thing a replica must never do. + +``--emit-ops`` closes that: the fetch also emits ``data.ops`` — the stamped op +batch that turns the file it is replacing INTO the template, in the frozen +vocabulary. Two contracts, both tested here: + +* **the op contract** (what the cloud forwards): replaying the batch with + ``apply_op`` reproduces the template's graph exactly — same node types, same + wiring, same widget values; +* **the spec contract** (what ``nodes path --emit-ops`` already promises): the + same array is accepted by ``apply_specs`` verbatim, so the batch is a legal + ``comfy workflow apply --ops`` input. + +Templates the frozen vocabulary cannot express (subgraph definitions, groups) +emit NO ops and say why, so the consumer falls back to its re-mint path rather +than silently landing a partial graph. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from typer.testing import CliRunner + +from comfy_cli import workflow_ops +from comfy_cli.caller import Caller +from comfy_cli.command import templates as templates_cmd +from comfy_cli.cql.engine import Graph +from comfy_cli.output.renderer import OutputMode, Renderer, reset_renderer_for_testing, set_renderer + + +@pytest.fixture(autouse=True) +def reset_singleton(): + reset_renderer_for_testing() + yield + reset_renderer_for_testing() + + +def _force_json_renderer(): + r = Renderer.resolve( + is_stdout_tty=False, + env={}, + caller=Caller(kind="user", agentic=False, source_env=None), + json_flag=True, + ) + r.mode = OutputMode.JSON + set_renderer(r) + return r + + +def _object_info() -> dict[str, Any]: + return { + "TinyLoader": { + "input": {"required": {"ckpt_name": [["a.safetensors", "b.safetensors"]]}}, + "input_order": {"required": ["ckpt_name"]}, + "output": ["MODEL"], + "output_name": ["MODEL"], + "category": "loaders", + "display_name": "Tiny Loader", + "python_module": "nodes", + }, + "TinySink": { + "input": {"required": {"model": ["MODEL"]}}, + "input_order": {"required": ["model"]}, + "output": [], + "output_name": [], + "category": "test", + "display_name": "Tiny Sink", + "python_module": "nodes", + }, + } + + +def _graph() -> Graph: + return Graph.from_object_info(_object_info()) + + +# A two-node template in frontend/save format: loader -> sink, one link. +def _template() -> dict[str, Any]: + return { + "id": "tpl-1", + "revision": 0, + "last_node_id": 2, + "last_link_id": 1, + "nodes": [ + { + "id": 1, + "type": "TinyLoader", + "pos": [10, 20], + "inputs": [], + "outputs": [{"name": "MODEL", "type": "MODEL", "links": [1]}], + "widgets_values": ["b.safetensors"], + }, + { + "id": 2, + "type": "TinySink", + "pos": [300, 20], + "inputs": [{"name": "model", "type": "MODEL", "link": 1}], + "outputs": [], + "widgets_values": [], + }, + ], + "links": [[1, 1, 0, 2, 0, "MODEL"]], + "groups": [], + } + + +def _existing() -> dict[str, Any]: + """A workflow already on the canvas — what the fetch replaces.""" + return { + "id": "wf-old", + "nodes": [ + { + "id": 77, + "type": "TinyLoader", + "pos": [0, 0], + "inputs": [], + "outputs": [{"name": "MODEL", "type": "MODEL", "links": []}], + "widgets_values": ["a.safetensors"], + } + ], + "links": [], + "last_node_id": 77, + "last_link_id": 0, + } + + +_GALLERY_ROW = { + "name": "tiny_template", + "title": "Tiny Template", + "output_type": "image", + "category": "Basics", + "tags": [], + "models": [], + "providers": [], +} + + +@pytest.fixture +def patched_fetch(monkeypatch: pytest.MonkeyPatch): + """Resolve the gallery + the workflow body locally: no network.""" + monkeypatch.setattr(templates_cmd, "_load_gallery", lambda *a, **kw: [{"templates": []}]) + monkeypatch.setattr(templates_cmd, "_flatten_templates", lambda cats: [dict(_GALLERY_ROW)]) + monkeypatch.setattr( + templates_cmd, + "_fetch_template_workflow", + lambda name, **kw: json.dumps(_template()).encode("utf-8"), + ) + + +def _run(args: list[str], capsys) -> dict[str, Any]: + _force_json_renderer() + runner = CliRunner() + result = runner.invoke(templates_cmd.app, args, standalone_mode=False) + captured = capsys.readouterr().out + if not captured.strip(): + captured = result.stdout or "" + for line in reversed(captured.strip().splitlines()): + try: + return json.loads(line) + except json.JSONDecodeError: + continue + raise AssertionError(f"no JSON envelope (rc={result.exit_code}, exc={result.exception}, out={captured[:600]})") + + +class TestTemplateFetchEmitsOpBatch: + def test_template_fetch_emits_op_batch(self, tmp_path: Path, patched_fetch, capsys): + """The batch is stamped, ordered delete→add→connect, and replaying it + onto the file being replaced reproduces the template exactly.""" + out = tmp_path / "workflow_bulkops.json" + base = _existing() + out.write_text(json.dumps(base), encoding="utf-8") + + env = _run( + ["fetch", "tiny_template", "-o", str(out), "--emit-ops", "--actor", "agent:th_1:7", "--base-version", "4"], + capsys, + ) + + assert env["ok"] is True, env + ops = env["data"]["ops"] + + # Every entry is a real, stamped op — the cloud drops anything without + # an op_id, so an unstamped entry is silently lost, not rejected. + for op in ops: + assert len(op["op_id"]) == 32 and op["op_id"].islower() + assert op["actor"] == "agent:th_1:7" + assert op["base_version"] == 4 + assert op["stamp"] == [4, "agent:th_1:7"] + assert len({op["op_id"] for op in ops}) == len(ops) + + # Replace = delete what was there, then build the template. + assert [op["op"] for op in ops] == ["delete_node", "add_node", "add_node", "connect"] + assert ops[0]["node_id"] == 77 + + # THE OP CONTRACT: replay onto the pre-fetch graph == the template. + replayed: dict[str, Any] = json.loads(json.dumps(base)) + for op in ops: + replayed = workflow_ops.apply_op(replayed, op, _graph()) + workflow_ops.strip_internal(replayed) + + assert [n["type"] for n in replayed["nodes"]] == ["TinyLoader", "TinySink"] + # Widget values ride inside the add_node payload (§8.5), so a fetched + # template keeps its demo values instead of catalog defaults. + loader = next(n for n in replayed["nodes"] if n["type"] == "TinyLoader") + assert loader["widgets_values"] == ["b.safetensors"] + assert loader["pos"] == [10, 20] + assert len(replayed["links"]) == 1 + link = replayed["links"][0] + sink = next(n for n in replayed["nodes"] if n["type"] == "TinySink") + assert link[1] == loader["id"] and link[3] == sink["id"] + assert sink["inputs"][0]["link"] == link[0] + + # Identity is minted, never inherited: the template's small counter ids + # (1, 2) would resurrect ids a concurrent replica may still hold. + assert all(n["id"] >= 1 << 40 for n in replayed["nodes"]) + assert link[0] >= 1 << 40 + + # The file the fetch wrote is still the template itself (unchanged + # behavior — --emit-ops adds a payload, it does not change the write). + assert [n["id"] for n in json.loads(out.read_text(encoding="utf-8"))["nodes"]] == [1, 2] + + def test_emitted_batch_round_trips_through_apply_specs(self, tmp_path: Path, patched_fetch, capsys): + """THE SPEC CONTRACT: the same array is a legal `workflow apply --ops` + batch — apply_specs accepts it verbatim and rebuilds the structure.""" + out = tmp_path / "workflow_bulkspecs.json" + base = _existing() + out.write_text(json.dumps(base), encoding="utf-8") + + env = _run(["fetch", "tiny_template", "-o", str(out), "--emit-ops"], capsys) + specs = env["data"]["ops"] + + wf, ops, aliases = workflow_ops.apply_specs(json.loads(json.dumps(base)), _graph(), specs) + + assert [n["type"] for n in wf["nodes"]] == ["TinyLoader", "TinySink"] + assert 77 not in [n["id"] for n in wf["nodes"]] + assert len(wf["links"]) == 1 + assert wf["links"][0][1] == aliases[specs[1]["as"]] + assert wf["links"][0][3] == aliases[specs[2]["as"]] + + def test_without_the_flag_the_envelope_is_unchanged(self, tmp_path: Path, patched_fetch, capsys): + out = tmp_path / "workflow_noops.json" + out.write_text(json.dumps(_existing()), encoding="utf-8") + env = _run(["fetch", "tiny_template", "-o", str(out)], capsys) + assert env["ok"] is True + assert "ops" not in env["data"] + assert "ops_skipped" not in env["data"] + + def test_emit_ops_on_a_fresh_canvas_has_no_deletes(self, tmp_path: Path, patched_fetch, capsys): + out = tmp_path / "does_not_exist_yet.json" + env = _run(["fetch", "tiny_template", "-o", str(out), "--emit-ops"], capsys) + assert [op["op"] for op in env["data"]["ops"]] == ["add_node", "add_node", "connect"] + + def test_inexpressible_template_emits_no_ops_and_says_why( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, patched_fetch, capsys + ): + """A template the frozen vocabulary cannot express (a subgraph + definition) must emit NOTHING — a partial batch would land a graph that + is not the template. The consumer keeps its re-mint fallback for these.""" + tpl = _template() + tpl["definitions"] = {"subgraphs": [{"id": "sg-1", "nodes": []}]} + monkeypatch.setattr( + templates_cmd, "_fetch_template_workflow", lambda name, **kw: json.dumps(tpl).encode("utf-8") + ) + out = tmp_path / "workflow_subgraph.json" + env = _run(["fetch", "tiny_template", "-o", str(out), "--emit-ops"], capsys) + + assert env["ok"] is True + assert "ops" not in env["data"] + assert "subgraph" in env["data"]["ops_skipped"] + + def test_emit_ops_without_out_still_emits(self, patched_fetch, capsys): + """Without -o there is no file being replaced, so the batch is a pure + build — still emitted, so a caller that materializes the envelope's + workflow itself can use the ops.""" + env = _run(["fetch", "tiny_template", "--emit-ops"], capsys) + assert [op["op"] for op in env["data"]["ops"]] == ["add_node", "add_node", "connect"] diff --git a/tests/comfy_cli/test_reset_doc_op.py b/tests/comfy_cli/test_reset_doc_op.py new file mode 100644 index 000000000..d4ebc7196 --- /dev/null +++ b/tests/comfy_cli/test_reset_doc_op.py @@ -0,0 +1,194 @@ +"""``reset_doc`` — the guarded, standalone-only document reset (V1-038 / BE-7171). + +``reset_doc`` was frozen in ``docs/op-vocabulary-v1.md`` §1.6 but left deferred: +``apply_op`` rejected it and ``DEFERRED_OPS`` pinned that rejection. This ticket +un-defers it, so the guarantees that make it safe move from prose into tests: + +* it is **guarded** — ``comfy workflow reset-doc `` fails closed without an + explicit ``--confirm`` and writes nothing; +* it is **standalone-only** — a batch containing it is rejected atomically with a + registered error code, exactly like ``clear``; +* it is a **history barrier** — unlike ``clear`` it drops the id high-water marks + and the applied-op bookkeeping, so it is not merely "delete every node". +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from typer.testing import CliRunner + +from comfy_cli import error_codes, workflow_ops +from comfy_cli.caller import Caller +from comfy_cli.command import workflow as workflow_cmd +from comfy_cli.command import workflow_edit +from comfy_cli.cql.engine import Graph +from comfy_cli.output.renderer import OutputMode, Renderer, reset_renderer_for_testing, set_renderer + + +@pytest.fixture(autouse=True) +def reset_singleton(): + reset_renderer_for_testing() + yield + reset_renderer_for_testing() + + +def _force_json_renderer(): + r = Renderer.resolve( + is_stdout_tty=False, + env={}, + caller=Caller(kind="user", agentic=False, source_env=None), + json_flag=True, + ) + r.mode = OutputMode.JSON + set_renderer(r) + return r + + +def _object_info() -> dict[str, Any]: + return { + "TinyLoader": { + "input": {"required": {"ckpt_name": [["a.safetensors"]]}}, + "input_order": {"required": ["ckpt_name"]}, + "output": ["MODEL"], + "output_name": ["MODEL"], + "category": "loaders", + "display_name": "Tiny Loader", + "python_module": "nodes", + }, + } + + +def _graph() -> Graph: + return Graph.from_object_info(_object_info()) + + +def _populated() -> dict[str, Any]: + return { + "id": "wf-1", + "revision": 0, + "nodes": [ + {"id": 1, "type": "TinyLoader", "pos": [0, 0], "inputs": [], "outputs": [], "widgets_values": []}, + {"id": 2, "type": "TinyLoader", "pos": [10, 0], "inputs": [], "outputs": [], "widgets_values": []}, + ], + "links": [], + "groups": [{"title": "g"}], + "last_node_id": 2, + "last_link_id": 0, + "_applied_ops": ["deadbeef" * 4], + } + + +def _run(args: list[str], capsys) -> dict[str, Any]: + _force_json_renderer() + runner = CliRunner() + result = runner.invoke(workflow_cmd.app, args, standalone_mode=False) + captured = capsys.readouterr().out + if not captured.strip(): + captured = result.stdout or "" + for line in reversed(captured.strip().splitlines()): + try: + return json.loads(line) + except json.JSONDecodeError: + continue + raise AssertionError(f"no JSON envelope (rc={result.exit_code}, exc={result.exception}, out={captured[:600]})") + + +class TestResetDocGuard: + def test_reset_doc_requires_confirm(self, tmp_path: Path, capsys): + """Without --confirm the command fails closed: nothing is written. + + The guard is the whole reason reset_doc is safe to expose — it erases + replay history, so an accidental invocation is unrecoverable by replay. + """ + wf = tmp_path / "wf_reset_guard.json" + before = _populated() + wf.write_text(json.dumps(before), encoding="utf-8") + + env = _run(["reset-doc", str(wf)], capsys) + + assert env["ok"] is False + assert env["error"]["code"] == "workflow_reset_doc_unconfirmed" + assert "--confirm" in (env["error"].get("hint") or "") + # The file is byte-for-byte untouched — a guard that writes anything is + # not a guard. + assert json.loads(wf.read_text(encoding="utf-8")) == before + + def test_reset_doc_with_confirm_empties_the_document(self, tmp_path: Path, capsys): + wf = tmp_path / "wf_reset_confirm.json" + wf.write_text(json.dumps(_populated()), encoding="utf-8") + + env = _run(["reset-doc", str(wf), "--confirm"], capsys) + + assert env["ok"] is True, env + op = env["data"]["op"] + assert op["op"] == "reset_doc" + assert op["removed_nodes"] == [1, 2] + assert op["stamp"] == [op["base_version"], op["actor"]] + + after = json.loads(wf.read_text(encoding="utf-8")) + assert after["nodes"] == [] + assert after["links"] == [] + assert after["groups"] == [] + # History barrier, not a clear: the high-water marks go back to the + # empty baseline (clear preserves them, §1.5 vs §1.6). + assert after["last_node_id"] == 0 + assert after["last_link_id"] == 0 + + +class TestResetDocIsNotBatchable: + def test_reset_doc_rejected_in_batch(self): + """A batch containing reset_doc is rejected atomically, with its own + registered code naming the standalone command.""" + with pytest.raises(workflow_ops.NotBatchableError) as ei: + workflow_ops.apply_specs( + {"nodes": [], "links": []}, + _graph(), + [{"op": "add_node", "class_type": "TinyLoader"}, {"op": "reset_doc"}], + ) + err = ei.value + assert err.code == "workflow_reset_doc_not_batchable" + assert error_codes.is_registered(err.code) + registered = error_codes.get(err.code) + assert registered is not None and "comfy workflow reset-doc" in (registered.hint or "") + assert "comfy workflow reset-doc" in err.hint + assert "no changes were applied" in str(err).lower() + + def test_reset_doc_rejected_through_the_apply_command( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys + ): + monkeypatch.setattr(workflow_edit, "_get_graph", lambda *a, **kw: _graph()) + wf = tmp_path / "wf_reset_batch.json" + wf.write_text(json.dumps(_populated()), encoding="utf-8") + ops = tmp_path / "reset_batch_ops.json" + ops.write_text(json.dumps([{"op": "reset_doc"}]), encoding="utf-8") + + env = _run(["apply", str(wf), "--ops", str(ops)], capsys) + + assert env["ok"] is False + assert env["error"]["code"] == "workflow_reset_doc_not_batchable" + # Atomic: the graph the batch was rejected against is untouched. + assert len(json.loads(wf.read_text(encoding="utf-8"))["nodes"]) == 2 + + +class TestResetDocReplay: + def test_apply_op_replays_reset_doc_and_records_it(self): + """Un-deferred: apply_op dispatches reset_doc. Its own op_id survives + the wipe, so a re-delivered reset is a no-op rather than a second wipe.""" + wf, op = workflow_ops.reset_doc(_populated(), actor="agent:t:1", base_version=3) + assert wf["nodes"] == [] and wf["links"] == [] + assert wf["_applied_ops"] == [op["op_id"]] + + # Idempotent re-delivery: put a node back, replay the same op — the + # op_id gate drops it, so the node survives. + wf["nodes"].append({"id": 9, "type": "TinyLoader"}) + wf = workflow_ops.apply_op(wf, op, None) + assert [n["id"] for n in wf["nodes"]] == [9] + + def test_reset_doc_is_no_longer_deferred(self): + assert "reset_doc" in workflow_ops.FROZEN_OPS + assert "reset_doc" not in workflow_ops.DEFERRED_OPS + assert "reset_doc" not in workflow_ops.BATCHABLE_OPS