From d962ed5f44bf4cbc7d3537e0bf7502845a3b9c82 Mon Sep 17 00:00:00 2001 From: bymyself Date: Mon, 31 Aug 2026 04:41:18 +0000 Subject: [PATCH 1/3] fix(workflow): rank concurrent autogrow names --- comfy_cli/workflow_ops.py | 68 ++++++++++++++++--- docs/op-vocabulary-v1.md | 2 +- tests/comfy_cli/command/test_workflow_edit.py | 25 +++++++ 3 files changed, 83 insertions(+), 12 deletions(-) diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py index 36ab4d85..3fc48c0a 100644 --- a/comfy_cli/workflow_ops.py +++ b/comfy_cli/workflow_ops.py @@ -40,13 +40,10 @@ (the link id). Two concurrent autogrow connects both survive and ``canonical`` compares grown slots by ``grow_id``, not by list position. -The one thing a leaderless writer genuinely *cannot* converge is a *sequence -decision*: the human-visible ordering/numbering of concurrently-grown autogrow -slots (a batch's element order) and of concurrent interior writes to the same -shared subgraph definition. Those are surfaced by :func:`detect_conflict` for the -merge consumer / ask-to-merge to resolve — the ops still never lose data, and -``canonical`` treats the order as immaterial, so the semantic graph converges even -while the display order does not. +Concurrent autogrow display order is resolved by the same total op rank used for +LWW writes: ``[base_version, actor, op_id]``. Concurrent interior writes to the +same shared subgraph definition remain a sequence decision and are surfaced by +:func:`detect_conflict` for the merge consumer / ask-to-merge to resolve. """ from __future__ import annotations @@ -484,6 +481,40 @@ def _next_autogrow_name(ins: list, requested: str, template: dict | None = None) return f"{base}.{_autogrow_elem_name(base, n, template)}" +def _rank_autogrow_group(workflow: dict, dst: dict, base: str, template: dict | None) -> None: + """Give replayed grows stable names and positions by their total op rank.""" + ins = dst.get("inputs") or [] + ranks = workflow.get("_autogrow_ranks") or {} + indexed = [ + (idx, inp) + for idx, inp in enumerate(ins) + if isinstance(inp, dict) + and str(inp.get("grow_id")) in ranks + and _autogrow_base(str(inp.get("name", ""))) == base + ] + if not indexed: + return + + ranked = sorted((inp for _, inp in indexed), key=lambda inp: ranks[str(inp["grow_id"])]) + fixed_names = {inp.get("name") for inp in ins if inp not in ranked} + names: list[str] = [] + n = 0 + while len(names) < len(ranked): + name = f"{base}.{_autogrow_elem_name(base, n, template)}" + if name not in fixed_names: + names.append(name) + n += 1 + for inp, name in zip(ranked, names, strict=True): + inp["name"] = name + for (idx, _), inp in zip(indexed, ranked, strict=True): + ins[idx] = inp + + slot_by_link = {inp.get("link"): idx for idx, inp in enumerate(ins) if inp.get("link") is not None} + for link in workflow.get("links") or []: + if len(link) >= 5 and str(link[3]) == str(dst.get("id")) and link[0] in slot_by_link: + link[4] = slot_by_link[link[0]] + + def _autogrow_base(slot_name: str) -> str: """The group a grown slot name belongs to: everything before the LAST dot. Element names never contain a dot, but a group nested under a @@ -1791,6 +1822,7 @@ def apply_op(workflow: dict, op: dict, graph) -> dict: # the poison state: a retry of the identical op loses to the failed # attempt's own stamp and is silently dropped forever. stamps_before = dict(workflow.get("_widget_stamps") or {}) + autogrow_ranks_before = dict(workflow.get("_autogrow_ranks") or {}) try: if kind == "add_node": _apply_add_node(workflow, op) @@ -1809,6 +1841,8 @@ def apply_op(workflow: dict, op: dict, graph) -> dict: except BaseException: if stamps_before or "_widget_stamps" in workflow: workflow["_widget_stamps"] = stamps_before + if autogrow_ranks_before or "_autogrow_ranks" in workflow: + workflow["_autogrow_ranks"] = autogrow_ranks_before raise # NOT ``applied.append`` — ``_apply_reset_doc`` REPLACES ``_applied_ops`` # with a fresh list (that is what makes it a history barrier), so the local @@ -2053,6 +2087,8 @@ def _apply_connect(workflow: dict, op: dict, graph) -> None: entry["widget"] = {"name": grow["widget"]} ins.append(entry) to_idx = len(ins) - 1 + if not grow.get("promoted") and not grow.get("widget") and inputcount is None: + workflow.setdefault("_autogrow_ranks", {})[str(op["link_id"])] = _stamp_key(op) if inputcount is not None: # Bump using the op's mint-time-planned value (NOT re-derived # from a post-collision-renamed slot number): every op's @@ -2142,6 +2178,8 @@ def _apply_connect(workflow: dict, op: dict, graph) -> None: out_links = out_port["links"] if op["link_id"] not in out_links: out_links.append(op["link_id"]) + if grow is not None and not grow.get("promoted") and not grow.get("widget") and grow.get("inputcount") is None: + _rank_autogrow_group(workflow, dst, _autogrow_base(str(grow["name"])), template) def _remove_link(workflow: dict, link_id: Any) -> None: @@ -2207,6 +2245,7 @@ def _apply_reset_doc(workflow: dict, op: dict) -> None: workflow["last_link_id"] = 0 workflow["_applied_ops"] = [] workflow["_widget_stamps"] = {} + workflow["_autogrow_ranks"] = {} # --------------------------------------------------------------------------- @@ -2256,12 +2295,17 @@ def _write_target(op: dict) -> tuple: def detect_conflict(a: dict, b: dict) -> bool: """True iff two ops write the same target incompatibly — the signal V0's - ask-to-merge raises instead of silently clobbering. Two autogrow connects to - the same base conflict here (their batch order is undecidable leaderlessly) - even though :func:`apply_op` keeps both connections and ``canonical`` treats - their order as immaterial.""" + ask-to-merge raises instead of silently clobbering. Ordinary autogrow + connects are not conflicts: their names and positions use ``_stamp_key``'s + deterministic total order.""" if _write_target(a) != _write_target(b): return False + if all(op.get("op") == "connect" and op.get("grow") for op in (a, b)): + if all( + not op["grow"].get("promoted") and not op["grow"].get("widget") and op["grow"].get("inputcount") is None + for op in (a, b) + ): + return False if a["op"] == "set_widget" and b["op"] == "set_widget": return a.get("value") != b.get("value") return True @@ -2289,6 +2333,7 @@ def canonical(workflow: dict) -> dict: w = copy.deepcopy(workflow) w.pop("_applied_ops", None) w.pop("_widget_stamps", None) + w.pop("_autogrow_ranks", None) nodes = w.get("nodes") # Capture each node's original index -> slot identity BEFORE reordering # inputs, so links (which reference the raw index) can be rewritten. @@ -2393,6 +2438,7 @@ def strip_internal(workflow: dict) -> dict: """ workflow.pop("_applied_ops", None) workflow.pop("_widget_stamps", None) + workflow.pop("_autogrow_ranks", None) return complete_save_format(workflow) diff --git a/docs/op-vocabulary-v1.md b/docs/op-vocabulary-v1.md index 69edcf4a..8a037b59 100644 --- a/docs/op-vocabulary-v1.md +++ b/docs/op-vocabulary-v1.md @@ -220,7 +220,7 @@ and `"7"` two registers for one node. | concurrent moves | no `move` op exists in v1 — positions are decided once at `add_node` mint time and frozen into the op; live position editing is frontend view state, out of scope until the FE stable-ID reconciliation (section 6) | `add_node` / `layout.cascade_pos` | | edges referencing deleted nodes | the connect no-ops (delete wins); a delete removes incident links and scrubs every dangling input/output reference, so no dangling edge survives either order | `_apply_connect`, `_apply_delete_node` | | duplicate entity creation | impossible by construction across writers (random 53-bit `mint_id`, no shared counter); a replayed `add_node` whose `node_id` already exists is a no-op; a re-sent op is dropped by `op_id` | `mint_id`, `_apply_add_node` | -| concurrent autogrow connects to one base | both survive: each grows a fresh slot keyed by `grow_id`; their display order is the one sequence decision a leaderless writer cannot make and is surfaced by `detect_conflict` for the merge consumer | `_apply_connect` (grow path), `detect_conflict` | +| concurrent autogrow connects to one base | both survive: each grows a fresh slot keyed by `grow_id`; names and positions are ranked by `[base_version, actor, op_id]`, so display order converges and `detect_conflict` does not surface an ask-to-merge conflict | `_apply_connect` (grow path), `_rank_autogrow_group`, `detect_conflict` | | invalid / inapplicable ops | explicit per kind — unknown kind: **reject** (`apply_op` raises); malformed op (missing required field): **reject**; well-formed op whose target node is gone: **no-op** (delete wins); `set_widget` naming a widget the live schema does not have: **reject**; `clear`/`reset_doc` inside a batch: **reject** with `workflow_clear_not_batchable` / `unknown op`. Rejection is never silent | `apply_op`, `apply_specs`, `_widget_index` | ## 4. Partial batches: abort-remainder diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py index fddbb4aa..99d6bf6d 100644 --- a/tests/comfy_cli/command/test_workflow_edit.py +++ b/tests/comfy_cli/command/test_workflow_edit.py @@ -2056,6 +2056,31 @@ def test_p9_autogrow_connects_are_commutative(self): # ...and the two orders converge. assert ops.canonical(ab) == ops.canonical(ba) + def test_p9_autogrow_display_order_uses_total_stamp_rank(self): + """Concurrent grows get identical names/order in both replay orders.""" + ops = self._ops() + g = _graph() + base = _autogrow_workflow() + _, lower = ops.connect(copy.deepcopy(base), g, 20, "IMAGE", 10, "images", actor="a", base_version=4) + _, higher = ops.connect(copy.deepcopy(base), g, 21, "IMAGE", 10, "images", actor="a", base_version=4) + lower["op_id"] = "0" * 32 + higher["op_id"] = "f" * 32 + + ab = ops.apply_op(ops.apply_op(copy.deepcopy(base), lower, g), higher, g) + ba = ops.apply_op(ops.apply_op(copy.deepcopy(base), higher, g), lower, g) + + def display(workflow): + node = next(n for n in workflow["nodes"] if n["id"] == 10) + return [(i["name"], i.get("grow_id"), i.get("link")) for i in node["inputs"]] + + assert display(ab) == display(ba) + assert [name for name, grow_id, _ in display(ab) if grow_id is not None] == [ + "images.image0", + "images.image1", + ] + assert ops.canonical(ab) == ops.canonical(ba) + assert ops.detect_conflict(lower, higher) is False + def test_autogrow_uses_schema_prefix_zero_based(self): """A ``{"prefix": "frame"}`` template names grown slots verbatim from the schema, 0-based (images.frame0, images.frame1) — a prefix that From c9de1a76441365256da60e092c3a1b14bb8bf6ce Mon Sep 17 00:00:00 2001 From: bymyself Date: Mon, 31 Aug 2026 05:41:18 +0000 Subject: [PATCH 2/3] fix: detect exited watcher processes on Windows --- comfy_cli/command/jobs.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 92ccbea6..b3e322f4 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -58,18 +58,21 @@ def _is_pid_alive(pid: int) -> bool: """Check if a process with the given PID is still running. - Uses ``psutil.pid_exists`` — never ``os.kill(pid, 0)``, which on Windows + Uses ``psutil.Process.is_running`` — never ``os.kill(pid, 0)``, which on Windows routes through ``GenerateConsoleCtrlEvent`` (0 == CTRL_C_EVENT) and, on Python <= 3.13.1, can fall through to ``TerminateProcess`` and kill the - probed process (python/cpython gh-58689). + probed process (python/cpython gh-58689). ``pid_exists`` alone is not + sufficient on Windows because an exited process remains addressable while + another process still holds an open handle to it. """ if pid <= 0: return False import psutil try: - return psutil.pid_exists(pid) - except (OverflowError, ValueError, OSError): + process = psutil.Process(pid) + return process.is_running() and process.status() != psutil.STATUS_ZOMBIE + except (psutil.Error, OverflowError, ValueError, OSError): # `watcher_pid` comes off a deliberately tolerant JSON load with no # range check, so a corrupt or hand-edited state file can carry a pid # psutil can't even look up (out-of-range -> OverflowError; Windows From 32a94f42443729d5e043d67a94cfbfa614f48515 Mon Sep 17 00:00:00 2001 From: bymyself Date: Wed, 2 Sep 2026 10:06:33 +0000 Subject: [PATCH 3/3] fix(workflow): keep autogrow replay idempotent Addresses https://github.com/Comfy-Org/comfy-cli/pull/826#discussion_r3891755677 --- comfy_cli/workflow_ops.py | 11 +++++++---- tests/comfy_cli/command/test_workflow_edit.py | 13 +++++++++++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/comfy_cli/workflow_ops.py b/comfy_cli/workflow_ops.py index 3fc48c0a..adc594e1 100644 --- a/comfy_cli/workflow_ops.py +++ b/comfy_cli/workflow_ops.py @@ -2034,6 +2034,13 @@ def _apply_connect(workflow: dict, op: dict, graph) -> None: # schema's own element names, when the catalog carries a template). ins = dst.setdefault("inputs", []) to_idx = next((k for k, i in enumerate(ins) if i.get("grow_id") == op["link_id"]), None) + inputcount = grow.get("inputcount") + port = None + template = None + if not grow.get("promoted") and not grow.get("widget") and inputcount is None: + base = _autogrow_base(str(grow["name"])) + port = _autogrow_group_port(graph, dst, base) + template = None if port is None else port.autogrow_template if grow.get("promoted"): # A promoted subgraph input is ONE register named by the definition # (``("input", to_node, "grow", name)``), not a fresh slot per @@ -2052,7 +2059,6 @@ def _apply_connect(workflow: dict, op: dict, graph) -> None: _remove_link(workflow, prev) ins[to_idx]["grow_id"] = op["link_id"] # the register follows the winner if to_idx is None: - inputcount = grow.get("inputcount") if grow.get("promoted"): name = grow["name"] elif inputcount is not None: @@ -2061,9 +2067,6 @@ def _apply_connect(workflow: dict, op: dict, graph) -> None: # base.elemN fallback — that name is meaningless for this family. name = _next_inputcount_name(ins, grow["name"]) else: - base = _autogrow_base(str(grow["name"])) - port = None if grow.get("widget") else _autogrow_group_port(graph, dst, base) - template = None if port is None else port.autogrow_template name = _next_autogrow_name(ins, grow["name"], template) if port is not None and name != grow["name"]: # A replay collision renamed the slot: never mint one past diff --git a/tests/comfy_cli/command/test_workflow_edit.py b/tests/comfy_cli/command/test_workflow_edit.py index 99d6bf6d..5d936f66 100644 --- a/tests/comfy_cli/command/test_workflow_edit.py +++ b/tests/comfy_cli/command/test_workflow_edit.py @@ -2081,6 +2081,19 @@ def display(workflow): assert ops.canonical(ab) == ops.canonical(ba) assert ops.detect_conflict(lower, higher) is False + def test_autogrow_replay_with_existing_slot_is_idempotent(self): + """A replay can reach an existing grow slot after applied-op history is lost.""" + ops = self._ops() + graph = _graph_with_autogrow_template({"prefix": "frame"}) + base = _autogrow_workflow() + applied, op = ops.connect(base, graph, 20, "IMAGE", 10, "images", actor="a") + expected = ops.canonical(applied) + + applied.pop("_applied_ops", None) + replayed = ops.apply_op(applied, op, graph) + + assert ops.canonical(replayed) == expected + def test_autogrow_uses_schema_prefix_zero_based(self): """A ``{"prefix": "frame"}`` template names grown slots verbatim from the schema, 0-based (images.frame0, images.frame1) — a prefix that