fix(workflow): rank concurrent autogrow names

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

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
@@ -2000,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
@@ -2018,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:
@@ -2027,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
@@ -2053,6 +2090,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 +2181,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 +2248,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 +2298,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 +2336,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 +2441,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)
 
 

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

tests/comfy_cli/command/test_workflow_edit.py

@@ -2056,6 +2056,44 @@ 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_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