feat(workflow): emit source_address on promoted subgraph widget rows

CHANGELOG.md

@@ -17,6 +17,13 @@ history.
 
 ### Added
 
+- `comfy workflow slots` rows for a subgraph's promoted widgets carry
+  `source_address`, the interior node and widget the row stands for
+  (`57/28.unet_name`), resolved through nested promotions and renamed inputs.
+  The key is optional: a row omits it when its interior widget does not
+  resolve, when its value comes from an interior PrimitiveNode, or when the
+  input is backed only by a legacy `proxyWidgets` entry.
+  The row's `address` is unchanged and stays the one to edit.
 - DynamicGroup workflow conversion, row editing and validation, including row
   templates in node discovery and the widget catalog. Refuse incompatible saved
   widget layouts and ambiguous names instead of editing a different widget.

comfy_cli/cql/engine.py

@@ -4702,6 +4702,10 @@ def _declared_subgraph_slots(
         }
         if linked_from is not None:
             slot["linked_from"] = linked_from
+        if pi is not None and pi.is_widget:
+            source_address = _promoted_source_address(instance_id, sg, pi, defs, graph)
+            if source_address is not None:
+                slot["source_address"] = source_address
         declared.append(slot)
     if workflow is not None:
         # Legacy ``proxyWidgets`` promotions the definition does not back
@@ -4719,38 +4723,87 @@ def _declared_subgraph_slots(
             if current is _promoted.UNSET:
                 all_resolved = False
                 continue
-            declared.append(
-                {
-                    "address": f"{instance_id}.{entry.name}",
-                    "name": entry.name,
-                    "type": str(entry.type),
-                    "current_value": current,
-                    "instance_id": instance_id,
-                    "node_type": instance.get("type", ""),
-                }
-            )
+            slot = {
+                "address": f"{instance_id}.{entry.name}",
+                "name": entry.name,
+                "type": str(entry.type),
+                "current_value": current,
+                "instance_id": instance_id,
+                "node_type": instance.get("type", ""),
+            }
+            # A primitive bypass's value lives on an interior PrimitiveNode
+            # that may feed several widgets; source_address names one concrete
+            # widget, so these rows omit it (documented in SKILL.md).
+            if entry.plan == _promoted.PLAN_CREATE:
+                source_address = _promoted_source_address(instance_id, sg, entry.as_promoted_input(sg), defs, graph)
+                if source_address is not None:
+                    slot["source_address"] = source_address
+            declared.append(slot)
     return declared, (any_declared and all_resolved)
 
 
-def promoted_source_port(sg: dict, pi: Any, defs: dict[str, dict], graph: Graph) -> tuple[str, Port | None]:
-    """``(interior class, Port)`` of the concrete widget behind promoted input
-    ``pi`` — the schema a host value is validated against."""
+def _promoted_source_address(instance_id: str, sg: dict, pi: Any, defs: dict[str, dict], graph: Graph) -> str | None:
+    """``<instance>/<interior>[/<deeper>…].<widget>`` — the concrete interior
+    widget a promoted row stands for, through any nested promotion chain.
+    Informational only: the row is still edited at ``<instance>.<name>``.
+    ``None`` when the chain does not resolve to a live interior widget: a
+    missing interior node, or a widget the catalog says that class lacks (a
+    class the catalog does not know is trusted as saved)."""
     from comfy_cli.cql import promoted as _promoted
 
     target = _promoted.deepest_source(sg, pi, defs)
     if target is None:
-        return "", None
+        return None
     path, widget = target
+    class_type, port = promoted_source_port(sg, pi, defs, graph)
+    if port is None:
+        if not class_type:
+            return None
+        m = graph.node(class_type)
+        if m is not None:
+            # A dynamic-combo sub-widget (``model.size_preset``) is not a
+            # top-level port; its selector is.
+            head = widget.split(".", 1)[0]
+            if head == widget or not any(p.name == head and p.is_dynamic_combo for p in m.inputs):
+                return None
+            # …and only the SELECTED option's sub-widgets exist on the node.
+            node = _promoted_source_node(sg, path, defs)
+            values = _widgets_as_positional(node.get("widgets_values") if node else None, graph, class_type)
+            if widget not in graph.widget_order_for_node(class_type, values):
+                return None
+    return f"{_SUBGRAPH_PATH_SEP.join([instance_id, *path])}.{widget}"
+
+
+def _promoted_source_node(sg: dict, path: list[str], defs: dict[str, dict]) -> dict | None:
+    """The terminal interior node at the end of a promoted source ``path``,
+    descending into nested subgraph definitions; ``None`` if any hop is missing."""
+    from comfy_cli.cql import promoted as _promoted
+
     current_sg = sg
     node: dict | None = None
     for seg in path:
-        node = next((n for n in current_sg.get("nodes") or [] if str(n.get("id")) == seg), None)
+        node = _promoted._inner_node(current_sg, seg)
         if node is None:
-            return "", None
+            return None
         nxt = defs.get(str(node.get("type", "")))
         if nxt is not None:
             current_sg = nxt
-    class_type = str(node.get("type", "")) if node else ""
+    return node
+
+
+def promoted_source_port(sg: dict, pi: Any, defs: dict[str, dict], graph: Graph) -> tuple[str, Port | None]:
+    """``(interior class, Port)`` of the concrete widget behind promoted input
+    ``pi`` — the schema a host value is validated against."""
+    from comfy_cli.cql import promoted as _promoted
+
+    target = _promoted.deepest_source(sg, pi, defs)
+    if target is None:
+        return "", None
+    path, widget = target
+    node = _promoted_source_node(sg, path, defs)
+    if node is None:
+        return "", None
+    class_type = str(node.get("type", ""))
     m = graph.node(class_type)
     if m is None:
         return class_type, None

comfy_cli/cql/promoted.py

@@ -270,7 +270,7 @@ def source_value(workflow: dict, sg: dict, pi: PromotedInput, graph, defs: dict[
     if pi.source_node is None:
         return UNSET
     defs = defs if defs is not None else defs_by_id(workflow)
-    inner = next((n for n in sg.get("nodes") or [] if str(n.get("id")) == pi.source_node), None)
+    inner = _inner_node(sg, pi.source_node)
     if inner is None:
         return UNSET
     if pi.nested:
@@ -389,7 +389,7 @@ def deepest_source(sg: dict, pi: PromotedInput, defs: dict[str, dict]) -> tuple[
         path.append(current.source_node)
         if not current.nested:
             return path, str(current.source_widget)
-        inner = next((n for n in current_sg.get("nodes") or [] if str(n.get("id")) == current.source_node), None)
+        inner = _inner_node(current_sg, current.source_node)
         inner_def = defs.get(str(inner.get("type", ""))) if inner is not None else None
         if inner_def is None or current.source_input is None:
             return None
@@ -549,7 +549,7 @@ def resolve_write(
                     )
             if link is not None:
                 origin = link.get("origin_id")
-                origin_node = next((n for n in parent_def.get("nodes") or [] if str(n.get("id")) == str(origin)), None)
+                origin_node = _inner_node(parent_def, origin)
                 raise ValueError(
                     f"{given} is fed by interior node {'/'.join(segments[:-1])}/{origin} "
                     f"({origin_node.get('type') if origin_node else '?'}) — that link supplies the value, so a widget "
@@ -792,7 +792,7 @@ def as_promoted_input(self, sg: dict) -> PromotedInput:
         validation/source-port lookup a linked promotion gets applies."""
         if self.plan == PLAN_PRIMITIVE:
             tid, slot = self.targets[0]
-            target = next((n for n in sg.get("nodes") or [] if str(n.get("id")) == tid), None)
+            target = _inner_node(sg, tid)
             entry = (target.get("inputs") or [])[slot] if target is not None else None
             marker = entry.get("widget") if isinstance(entry, dict) else None
             widget = marker.get("name") if isinstance(marker, dict) and marker.get("name") else entry.get("name")

comfy_cli/skills/comfy/SKILL.md

@@ -1007,6 +1007,13 @@ comfy --json workflow slots path.json
 #   --node <id> lists one node's slots, subgraph interior included (a
 #   template id like 94 works on an inserted canvas; that node's own rows
 #   carry template_node_id)
+#   a promoted subgraph widget row (<instance_id>.<name>) carries
+#   source_address when its interior widget resolves — the interior widget
+#   it stands for, e.g. 57/28.unet_name (the real widget name, even when
+#   the promotion is renamed). Optional: a row omits it when its source
+#   does not resolve, when the value comes from an interior PrimitiveNode,
+#   or when the input is only a legacy proxyWidgets entry. Read-only
+#   context: edit at the row's own address
 
 # 2. Set a single slot
 comfy workflow set-slot path.json 6.text="a cat"

tests/comfy_cli/command/test_workflow_slots.py

@@ -1101,3 +1101,151 @@ def test_set_slot_reports_a_domain_error_instead_of_crashing(self, patched_graph
         env = _run(["set-slot", str(path), "3.seed=1"], capsys)
         assert env["ok"] is False
         assert env["error"]["code"] == "workflow_slot_invalid"
+
+
+# ---------------------------------------------------------------------------
+# slots — source_address on promoted subgraph widget rows
+# ---------------------------------------------------------------------------
+#
+# A promoted widget is advertised at ``<instance>.<name>`` and its interior
+# widget is hidden, so the row names the interior node + widget it stands for
+# in ``source_address`` — the only way a consumer recovers the real widget
+# behind a renamed promotion or tells two same-class loaders apart.
+
+
+def _gallery_workflow(name: str) -> dict:
+    return json.loads((_FIXTURES / "gallery" / name).read_text(encoding="utf-8"))
+
+
+@pytest.fixture
+def patched_promoted_graph(monkeypatch):
+    oi = json.loads((_FIXTURES / "object_info_subgraph_promoted.json").read_text(encoding="utf-8"))
+    monkeypatch.setattr(workflow_cmd, "_get_graph", lambda *a, **kw: Graph.from_object_info(oi))
+
+
+class TestSlotsSourceAddress:
+    def test_promoted_row_names_its_interior_widget(self, patched_promoted_graph, tmp_path, capsys):
+        path = _write_workflow(tmp_path, _gallery_workflow("image_z_image_turbo.json"))
+        env = _run(["slots", str(path)], capsys)
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        row = by_addr["57.unet_name"]
+        assert row["source_address"] == "57/28.unet_name"
+        # Additive only: the editable address and the UUID node_type stay.
+        assert row["node_type"] == "f2fdebf6-dfaf-43b6-9eb2-7f70613cfdc1"
+        assert "57/28.unet_name" not in by_addr
+        # A labelled promotion still names the interior widget.
+        assert by_addr["57.text"]["source_address"] == "57/27.text"
+
+    def test_minimax_promoted_loader(self, patched_promoted_graph, tmp_path, capsys):
+        path = _write_workflow(tmp_path, _gallery_workflow("audio_minimax_music_3.json"))
+        env = _run(["slots", str(path)], capsys)
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        assert by_addr["37.unet_name"]["source_address"] == "37/6.unet_name"
+
+    def test_renamed_promotion_carries_the_real_widget_name(self, patched_promoted_graph, tmp_path, capsys):
+        wf = _gallery_workflow("image_z_image_turbo.json")
+        sg = wf["definitions"]["subgraphs"][0]
+        next(i for i in sg["inputs"] if i["name"] == "unet_name")["name"] = "diffusion_model"
+        path = _write_workflow(tmp_path, wf)
+        env = _run(["slots", str(path)], capsys)
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        assert "57.unet_name" not in by_addr
+        assert by_addr["57.diffusion_model"]["source_address"] == "57/28.unet_name"
+
+    def test_top_level_row_has_no_source_address(self, patched_promoted_graph, tmp_path, capsys):
+        path = _write_workflow(tmp_path, _gallery_workflow("image_z_image_turbo.json"))
+        env = _run(["slots", str(path)], capsys)
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        assert "source_address" not in by_addr["9.filename_prefix"]
+
+    def test_nested_promotion_resolves_to_the_deepest_widget(self, patched_subgraph_graph, tmp_path, capsys):
+        path = _write_workflow(tmp_path, _subgraph_workflow())
+        env = _run(["slots", str(path)], capsys)
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        # 10.value -> inner instance 3's promoted value -> node 7's widget.
+        assert by_addr["10.value"]["source_address"] == "10/3/7.value"
+        # A legacy proxyWidgets promotion the frontend repairs on load.
+        assert by_addr["10.seed"]["source_address"] == "10/9.seed"
+        # Interior rows are not promotions.
+        assert "source_address" not in by_addr["10/9.prompt"]
+
+    def test_select_projects_source_address(self, patched_promoted_graph, tmp_path, capsys):
+        path = _write_workflow(tmp_path, _gallery_workflow("audio_minimax_music_3.json"))
+        env = _run(["slots", str(path), "--select", "slots.#.source_address"], capsys)
+        assert env["ok"] is True
+        # Rows without the key (plain top-level widgets) are dropped, not null.
+        assert "37/6.unet_name" in env["data"]
+        assert all(isinstance(a, str) for a in env["data"])
+
+    @staticmethod
+    def _materialized_loader(interior_type: str) -> dict:
+        # Host value materialized for every promotion, so the unet_name row is
+        # emitted even when the interior widget no longer resolves.
+        wf = _gallery_workflow("image_z_image_turbo.json")
+        next(n for n in wf["nodes"] if n["id"] == 57)["widgets_values"] = [
+            "a cat",
+            1024,
+            1024,
+            0,
+            8,
+            "z.safetensors",
+            "qwen.safetensors",
+            "ae.safetensors",
+        ]
+        sg = wf["definitions"]["subgraphs"][0]
+        next(n for n in sg["nodes"] if n["id"] == 28)["type"] = interior_type
+        return wf
+
+    def test_stale_interior_widget_omits_source_address(self, patched_promoted_graph, tmp_path, capsys):
+        # The catalog's VAELoader has no unet_name: the row stays (the host
+        # value runs) but names no dangling interior widget.
+        path = _write_workflow(tmp_path, self._materialized_loader("VAELoader"))
+        env = _run(["slots", str(path)], capsys)
+        row = {s["address"]: s for s in env["data"]["slots"]}["57.unet_name"]
+        assert row["current_value"] == "z.safetensors"
+        assert "source_address" not in row
+
+    def test_uncatalogued_interior_class_keeps_source_address(self, patched_promoted_graph, tmp_path, capsys):
+        # A custom node the catalog does not know is trusted as saved.
+        path = _write_workflow(tmp_path, self._materialized_loader("SomeCustomUnetLoader"))
+        env = _run(["slots", str(path)], capsys)
+        row = {s["address"]: s for s in env["data"]["slots"]}["57.unet_name"]
+        assert row["source_address"] == "57/28.unet_name"
+
+    def test_null_interior_node_does_not_crash_the_chain(self, patched_subgraph_graph, tmp_path, capsys):
+        # Every other walk skips a non-dict ``nodes`` entry; the source chain
+        # must too rather than raising out of ``slots`` with no envelope.
+        wf = _subgraph_workflow()
+        for sg in wf["definitions"]["subgraphs"]:
+            sg["nodes"].insert(0, None)
+        path = _write_workflow(tmp_path, wf)
+        env = _run(["slots", str(path)], capsys)
+        assert env["ok"] is True
+        by_addr = {s["address"]: s for s in env["data"]["slots"]}
+        assert by_addr["10.value"]["source_address"] == "10/3/7.value"
+
+    @pytest.mark.parametrize(
+        ("selection", "widget", "expected"),
+        [
+            ("seedream 5.0 pro", "model.size_preset", "50/1.model.size_preset"),
+            # max_images exists only on the lite option: the saved pro
+            # selection has no such widget, so no dangling address.
+            ("seedream 5.0 pro", "model.max_images", None),
+            ("seedream 5.0 lite", "model.max_images", "50/1.model.max_images"),
+            ("seedream 5.0 pro", "model.bogus", None),
+        ],
+    )
+    def test_dynamic_combo_sub_widget_follows_saved_selection(self, selection, widget, expected):
+        from comfy_cli.cql import engine as cql_engine
+        from comfy_cli.cql.promoted import PromotedInput
+
+        inner = _seedream_workflow()["nodes"][0]
+        if selection == "seedream 5.0 lite":
+            # lite inlines size_preset, width, height, max_images, fail_on_partial.
+            inner["widgets_values"] = ["p", selection, "(1K) 1024x1024 (1:1)", 2048, 2048, 3, False, 0, "fixed", False]
+        else:
+            assert inner["widgets_values"][1] == selection
+        pi = PromotedInput(
+            name=widget, type="INT", index=0, value_index=0, source_node="1", source_input=widget, source_widget=widget
+        )
+        assert cql_engine._promoted_source_address("50", {"nodes": [inner]}, pi, {}, _seedream_graph()) == expected