fix(cql): shape-check the definitions/subgraphs containers before reading them

comfy_cli/command/templates.py

@@ -1621,10 +1621,12 @@ def _workflow_node_types(workflow: Any) -> set[str]:
         return types
     if isinstance(workflow.get("nodes"), list):
         node_lists = [workflow.get("nodes") or []]
-        subgraphs = (workflow.get("definitions") or {}).get("subgraphs") or []
-        for sg in subgraphs:
-            if isinstance(sg, dict):
-                node_lists.append(sg.get("nodes") or [])
+        definitions = workflow.get("definitions")
+        subgraphs = definitions.get("subgraphs") if isinstance(definitions, dict) else None
+        if isinstance(subgraphs, list):
+            for sg in subgraphs:
+                if isinstance(sg, dict):
+                    node_lists.append(sg.get("nodes") or [])
         for nodes in node_lists:
             for node in nodes:
                 if isinstance(node, dict) and isinstance(node.get("type"), str):

comfy_cli/cql/engine.py

@@ -2794,8 +2794,18 @@ def _subgraph_defs_by_id(workflow: dict) -> dict[str, dict]:
     always wins. We still register ``name`` as a *fallback* key (only when it
     doesn't shadow an id and isn't ambiguous across defs) to support older
     name-typed templates that predate UUID ids.
+
+    Containers that are not a ``dict``/``list`` read as "no definitions", so a
+    corrupt file degrades to an empty index rather than raising (a truthy
+    non-dict ``definitions`` or non-list ``subgraphs``) or walking a string
+    per-character.
     """
-    defs = (workflow.get("definitions") or {}).get("subgraphs") or []
+    definitions = workflow.get("definitions")
+    if not isinstance(definitions, dict):
+        return {}
+    defs = definitions.get("subgraphs")
+    if not isinstance(defs, list):
+        return {}
     by_id: dict[str, dict] = {}
     name_counts: dict[str, int] = {}
     name_first: dict[str, dict] = {}
@@ -3493,11 +3503,14 @@ def _count_instances(workflow: dict, def_id: str) -> int:
     for n in workflow.get("nodes") or []:
         if isinstance(n, dict) and str(n.get("type", "")) == def_id:
             count += 1
-    for sg in (workflow.get("definitions") or {}).get("subgraphs") or []:
-        if isinstance(sg, dict):
-            for n in sg.get("nodes") or []:
-                if isinstance(n, dict) and str(n.get("type", "")) == def_id:
-                    count += 1
+    definitions = workflow.get("definitions")
+    subgraphs = definitions.get("subgraphs") if isinstance(definitions, dict) else None
+    if isinstance(subgraphs, list):
+        for sg in subgraphs:
+            if isinstance(sg, dict):
+                for n in sg.get("nodes") or []:
+                    if isinstance(n, dict) and str(n.get("type", "")) == def_id:
+                        count += 1
     return count
 
 

tests/comfy_cli/command/test_templates.py

@@ -1517,3 +1517,32 @@ def test_ls_self_heals_from_a_cache_poisoned_by_an_older_build(cache_file, monke
     assert result.exit_code == 0, result.output
     assert _envelope(result.output)["data"]["total_in_gallery"] > 0
     assert json.loads(cache_file.read_bytes()) == FIXTURE  # healed on disk too
+
+
+# ---------------------------------------------------------------------------
+# _workflow_node_types — corrupt definitions containers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "definitions",
+    [5, "definitions", ["x"], {"subgraphs": 5}, {"subgraphs": "abc"}],
+    ids=["int", "str", "list", "non-list-subgraphs", "str-subgraphs"],
+)
+def test_workflow_node_types_tolerates_corrupt_definitions(definitions):
+    """A wrong-typed ``definitions``/``subgraphs`` reads as "no definitions".
+
+    ``or`` only replaced falsy values, so a truthy wrong-typed one reached
+    ``.get``/the loop and raised. Top-level nodes are still collected.
+    """
+    wf = {"nodes": [{"id": 1, "type": "KSampler"}], "definitions": definitions}
+    assert templates_cmd._workflow_node_types(wf) == {"KSampler"}
+
+
+def test_workflow_node_types_still_reads_well_formed_subgraph_nodes():
+    """Positive control: the shape checks must not cost the happy path."""
+    wf = {
+        "nodes": [{"id": 1, "type": "KSampler"}],
+        "definitions": {"subgraphs": [{"id": "u1", "nodes": [{"id": 9, "type": "CLIPTextEncode"}]}]},
+    }
+    assert templates_cmd._workflow_node_types(wf) == {"KSampler", "CLIPTextEncode"}

tests/comfy_cli/command/test_workflow_slots.py

@@ -1040,3 +1040,51 @@ def test_input_path_short_circuits_routing(self, tmp_path, monkeypatch, capsys):
         path = _write_workflow(tmp_path, _direct_workflow())
         env = _run(["slots", str(path), "--input", str(oi)], capsys)
         assert env["ok"] is True, env
+
+
+class TestSlotsCorruptDefinitions:
+    """End-to-end: a corrupt ``definitions`` block must still yield an envelope.
+
+    Before the shape checks in ``_subgraph_defs_by_id`` a truthy wrong-typed
+    container exited ``comfy workflow slots`` on a rich traceback
+    (``AttributeError`` on ``definitions``, ``TypeError`` on ``subgraphs``)
+    with NOTHING on stdout — the JSON caller saw a hard crash instead of a
+    parseable result. ``slots_cmd``'s ``except (ValueError, KeyError)`` does
+    not catch either, and there is no catch-all above it.
+    """
+
+    @pytest.mark.parametrize(
+        "definitions",
+        [5, {"subgraphs": 5}, {"subgraphs": "abc"}, ["x"]],
+        ids=["non-dict-definitions", "non-list-subgraphs", "str-subgraphs", "list-definitions"],
+    )
+    def test_slots_degrades_to_an_empty_index(self, patched_graph, tmp_path, capsys, definitions):
+        path = _write_workflow(tmp_path, {"nodes": [], "links": [], "definitions": definitions})
+        captured, _err, result = _invoke(["slots", str(path)], capsys)
+        assert result.exception is None, f"crashed instead of degrading: {result.exception!r}"
+        assert result.exit_code == 0, captured
+        env = json.loads([ln for ln in captured.strip().splitlines() if ln.strip()][-1])
+        assert env["ok"] is True
+        assert env["data"]["count"] == 0
+        assert env["data"]["slots"] == []
+
+    def test_top_level_nodes_are_still_read_past_a_corrupt_definitions_block(self, patched_graph, tmp_path, capsys):
+        """Degrading is not the same as giving up: only the definitions block is
+        unreadable, so the top-level graph's slots must still come back."""
+        wf = dict(_direct_workflow(), definitions=5)
+        path = _write_workflow(tmp_path, wf)
+        env = _run(["slots", str(path)], capsys)
+        assert env["ok"] is True
+        assert env["data"]["count"] > 0
+        assert "6.text" in {s["address"] for s in env["data"]["slots"]}
+
+    def test_set_slot_reports_a_domain_error_instead_of_crashing(self, patched_graph, tmp_path, capsys):
+        """``set-slot`` reaches the same helper before any of its own guards, so
+        it crashed on the same input. It should now fail on the real problem —
+        the address doesn't resolve — through the normal error envelope."""
+        path = _write_workflow(tmp_path, {"nodes": [], "links": [], "definitions": {"subgraphs": 5}})
+        _captured, _err, result = _invoke(["set-slot", str(path), "3.seed=1"], capsys)
+        assert result.exception is None or isinstance(result.exception, SystemExit), repr(result.exception)
+        env = _run(["set-slot", str(path), "3.seed=1"], capsys)
+        assert env["ok"] is False
+        assert env["error"]["code"] == "workflow_slot_invalid"

tests/comfy_cli/cql/test_engine.py

@@ -17,7 +17,9 @@
     Graph,
     Port,
     _apply_one_slot,
+    _count_instances,
     _extract_frontend_slots,
+    _subgraph_defs_by_id,
     _write_widget,
 )
 
@@ -4364,3 +4366,83 @@ def test_no_output_node_at_all_does_not_double_report(self, graph: Graph):
         assert any(e["code"] == "prompt_no_outputs" for e in result["errors"])
         warns = [w for w in result["warnings"] if w["code"] == "node_not_reachable_from_output"]
         assert warns == [], "prompt_no_outputs already says it; don't pile on"
+
+
+# ---------------------------------------------------------------------------
+# _subgraph_defs_by_id / _count_instances — corrupt definitions containers
+# ---------------------------------------------------------------------------
+
+
+class TestSubgraphDefsByIdShapeChecks:
+    """A hand-edited or truncated save can put anything under ``definitions``.
+
+    The old ``(workflow.get("definitions") or {}).get("subgraphs") or []`` only
+    replaced FALSY values, so a truthy wrong-typed one survived and crashed the
+    caller (``comfy workflow slots`` exited on a traceback with no envelope).
+    The agreed degradation is an empty index: a document with no readable
+    definitions has no subgraph instances to resolve, and every caller already
+    handles ``{}``.
+    """
+
+    @pytest.mark.parametrize(
+        "definitions",
+        [5, "definitions", ["x"], 0.0, True],
+        ids=["int", "str", "list", "float", "bool"],
+    )
+    def test_non_dict_definitions_reads_as_no_definitions(self, definitions):
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": definitions}) == {}
+
+    @pytest.mark.parametrize(
+        "subgraphs",
+        [5, {"a": 1}, 3.5, True],
+        ids=["int", "dict", "float", "bool"],
+    )
+    def test_non_list_subgraphs_reads_as_no_definitions(self, subgraphs):
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": {"subgraphs": subgraphs}}) == {}
+
+    def test_string_subgraphs_is_not_walked_per_character(self):
+        """A string is iterable, so a naive guard silently walks it character by
+        character. It happens to yield ``{}`` too (single chars aren't dicts),
+        so this case documents INTENT: the isinstance branch must reject the
+        string outright rather than arrive at ``{}`` by accident."""
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": {"subgraphs": "abc"}}) == {}
+
+    def test_missing_containers_still_read_as_empty(self):
+        assert _subgraph_defs_by_id({"nodes": []}) == {}
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": None}) == {}
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": {}}) == {}
+        assert _subgraph_defs_by_id({"nodes": [], "definitions": {"subgraphs": []}}) == {}
+        # A top-level ``subgraphs`` with no ``definitions`` wrapper is not the
+        # block this helper reads; it must not be picked up by accident.
+        assert _subgraph_defs_by_id({"subgraphs": {"a": 1}}) == {}
+        assert _subgraph_defs_by_id({"subgraphs": [{"id": "u1"}]}) == {}
+
+    def test_well_formed_definitions_still_index_by_id_and_name(self):
+        """Positive control: the shape checks must not cost the happy path."""
+        wf = {"definitions": {"subgraphs": [{"id": "u1", "name": "A", "nodes": []}]}}
+        by_id = _subgraph_defs_by_id(wf)
+        assert by_id["u1"] is wf["definitions"]["subgraphs"][0]
+        assert by_id["A"] is wf["definitions"]["subgraphs"][0], "unambiguous name stays a fallback key"
+
+    def test_non_dict_entries_are_still_skipped(self):
+        """The per-entry guard is unchanged: junk beside a real def is dropped,
+        the real def is kept."""
+        wf = {"definitions": {"subgraphs": ["junk", None, 7, {"id": "u1", "name": "A", "nodes": []}]}}
+        assert set(_subgraph_defs_by_id(wf)) == {"u1", "A"}
+
+
+class TestCountInstancesShapeChecks:
+    def test_non_dict_definitions_counts_top_level_only(self):
+        wf = {"nodes": [{"id": 1, "type": "u1"}], "definitions": 5}
+        assert _count_instances(wf, "u1") == 1
+
+    def test_non_list_subgraphs_counts_top_level_only(self):
+        wf = {"nodes": [{"id": 1, "type": "u1"}], "definitions": {"subgraphs": "abc"}}
+        assert _count_instances(wf, "u1") == 1
+
+    def test_well_formed_definitions_still_count_interior_instances(self):
+        wf = {
+            "nodes": [{"id": 1, "type": "u1"}],
+            "definitions": {"subgraphs": [{"id": "u2", "nodes": [{"id": 9, "type": "u1"}]}]},
+        }
+        assert _count_instances(wf, "u1") == 2