diff --git a/comfy_cli/cql/engine.py b/comfy_cli/cql/engine.py
index 0af9338f..7e9df564 100644
--- a/comfy_cli/cql/engine.py
+++ b/comfy_cli/cql/engine.py
@@ -489,6 +489,11 @@ def annotate(
     # -- Lookup --
 
     def node(self, name: str) -> Morphism | None:
+        # A malformed workflow can supply an unhashable class_type (list/dict);
+        # dict.get on an unhashable key raises TypeError, so screen it out here
+        # rather than crash the reachability walk / lookups (BE-3406 hardening).
+        if not isinstance(name, str):
+            return None
         return self._nodes.get(name)
 
     def all_nodes(self) -> list[Morphism]:
@@ -703,6 +708,15 @@ def validate_workflow(self, workflow: dict[str, Any]) -> dict[str, Any]:
         # (execution.py:1155-1162, prompt_no_outputs). Track whether any
         # recognized node is an output node.
         has_output_node = False
+        # Server parity: ComfyUI's validate_prompt only validates output nodes
+        # and their transitive input ancestors — any node not reachable from an
+        # output is pruned and never validated (execution.py). Restrict the
+        # promoted hard checks (required-input presence, autogrow-required, and
+        # below_min/above_max ranges) to that reachable set, so a
+        # disconnected/incomplete node the server would silently drop isn't
+        # hard-rejected here. Edge/shape/enum checks are left as-is (pre-existing
+        # behavior, out of scope for this change).
+        reachable = _output_reachable_node_ids(workflow, self)
 
         for node_id, node_data in workflow.items():
             # `_meta` is the compose/run provenance block (schema/blueprint/items),
@@ -721,6 +735,11 @@ def validate_workflow(self, workflow: dict[str, Any]) -> dict[str, Any]:
                 )
                 continue
             class_type = node_data.get("class_type", "")
+            # A non-string class_type (list/dict from malformed JSON) is unhashable
+            # and would crash the self._nodes.get(class_type) lookup below; treat it
+            # as absent so it flows to the structured non_node_key path instead.
+            if not isinstance(class_type, str):
+                class_type = ""
             if not class_type:
                 warnings.append(
                     {
@@ -758,7 +777,13 @@ def validate_workflow(self, workflow: dict[str, Any]) -> dict[str, Any]:
             # case surfaces here instead of as a cryptic server reject.
             autogrow_ports = {p.name: p for p in m.inputs if p.is_autogrow}
             autogrow_seen: set[str] = set()
-            for input_name, value in (node_data.get("inputs") or {}).items():
+            node_inputs = node_data.get("inputs")
+            # A truthy non-dict `inputs` (e.g. a string/list from malformed JSON)
+            # sails through `or {}` and crashes `.items()`; treat it as empty so
+            # required-input checks flag the absence instead of raising.
+            if not isinstance(node_inputs, dict):
+                node_inputs = {}
+            for input_name, value in node_inputs.items():
                 if autogrow_ports and "." in input_name:
                     base = input_name.split(".", 1)[0]
                     if base in autogrow_ports:
@@ -801,7 +826,10 @@ def validate_workflow(self, workflow: dict[str, Any]) -> dict[str, Any]:
                         continue
 
                     src_class = src_data["class_type"]
-                    src_m = self._nodes.get(src_class)
+                    # Route through the guarded lookup: a referenced node with an
+                    # unhashable class_type (malformed JSON) would otherwise crash
+                    # the dict.get here.
+                    src_m = self.node(src_class)
                     if src_m is None:
                         # Source class_type already flagged by the outer loop
                         continue
@@ -868,13 +896,22 @@ def validate_workflow(self, workflow: dict[str, Any]) -> dict[str, Any]:
                         }
                     )
                     continue
-                # Catalog checks (enum membership, etc.)
-                cat_errors, cat_warnings = _validate_catalog_value(node_id, class_type, input_name, port, value)
+                # Catalog checks (enum membership, etc.). Range violations are a
+                # hard reject only on a node the server will actually run; on a
+                # pruned (output-unreachable) node they stay advisory warnings.
+                cat_errors, cat_warnings = _validate_catalog_value(
+                    node_id, class_type, input_name, port, value, range_is_error=node_id in reachable
+                )
                 errors.extend(cat_errors)
                 warnings.extend(cat_warnings)
 
-            errors.extend(_check_autogrow_required(node_id, autogrow_ports, autogrow_seen, node_data))
-            errors.extend(_check_required_present(node_id, m, node_data))
+            # Required-presence checks apply only to output-reachable nodes: the
+            # server prunes unreachable nodes without validating them, so
+            # enforcing required inputs on a disconnected node over-rejects a
+            # prompt the server would run (BE-3406).
+            if node_id in reachable:
+                errors.extend(_check_autogrow_required(node_id, autogrow_ports, autogrow_seen, node_data))
+                errors.extend(_check_required_present(node_id, m, node_data))
 
         # No-outputs check: the server rejects any prompt with zero output
         # nodes (execution.py:1155-1162, prompt_no_outputs) — including an
@@ -1044,14 +1081,61 @@ def morphism_to_dict(self, m: Morphism) -> dict[str, Any]:
 # error/warning dicts for the caller to append — no shared state is threaded.
 
 
+def _output_reachable_node_ids(workflow: dict[str, Any], graph: Graph) -> set[str]:
+    """Node ids the server would actually validate: output nodes and their
+    transitive input ancestors.
+
+    Mirrors ComfyUI's execution.py::validate_prompt, which validates only the
+    output nodes (``OUTPUT_NODE``) and everything reachable by walking their
+    input links backward — any node not reachable from an output is pruned and
+    never validated. We reproduce that reachable set so the promoted hard checks
+    (required_input_missing, autogrow_no_slots, below_min/above_max) don't
+    reject a disconnected node the server would silently drop.
+
+    An input value shaped ``[source_node_id, output_index]`` is a link edge (the
+    same predicate the per-input link walk uses); we follow those edges backward
+    from every output node. A reference to a node absent from the workflow (a
+    dangling edge, flagged separately) simply isn't traversed.
+    """
+    reachable: set[str] = set()
+    stack: list[str] = []
+    for node_id, node_data in workflow.items():
+        if node_id == "_meta" or not isinstance(node_data, dict):
+            continue
+        m = graph.node(node_data.get("class_type", ""))
+        if m is not None and m.is_output_node and node_id not in reachable:
+            reachable.add(node_id)
+            stack.append(node_id)
+    while stack:
+        node_data = workflow.get(stack.pop())
+        if not isinstance(node_data, dict):
+            continue
+        node_inputs = node_data.get("inputs")
+        # Guard against a truthy non-dict `inputs` from malformed JSON, which
+        # would slip past `or {}` and raise AttributeError on `.values()`.
+        if not isinstance(node_inputs, dict):
+            continue
+        for value in node_inputs.values():
+            if isinstance(value, list) and len(value) == 2:
+                src_id = str(value[0])
+                if src_id in workflow and src_id not in reachable:
+                    reachable.add(src_id)
+                    stack.append(src_id)
+    return reachable
+
+
 def _validate_catalog_value(
-    node_id: str, class_type: str, input_name: str, port: Port, value: Any
+    node_id: str, class_type: str, input_name: str, port: Port, value: Any, *, range_is_error: bool = True
 ) -> tuple[list[dict], list[dict]]:
     """Enum-membership and other catalog checks for one scalar input value.
 
     Returns (errors, warnings): unknown-enum and out-of-range values are hard
     errors (the server rejects them); every other catalog finding is a
     namespaced warning.
+
+    ``range_is_error`` gates the below_min/above_max promotion (BE-3406): the
+    server only range-checks nodes it actually runs, so on a pruned
+    (output-unreachable) node these are demoted back to advisory warnings.
     """
     errors: list[dict] = []
     warnings: list[dict] = []
@@ -1086,15 +1170,23 @@ def _validate_catalog_value(
             # unknown_enum_value hard error does.
             bound = port.options.min if w["code"] == "below_min" else port.options.max
             op = ">=" if w["code"] == "below_min" else "<="
-            errors.append(
-                {
-                    "node_id": node_id,
-                    "field": input_name,
-                    "code": w["code"],
-                    "message": w["message"],
-                    "hint": f"use a value {op} {bound}",
-                }
-            )
+            entry = {
+                "node_id": node_id,
+                "field": input_name,
+                "code": w["code"],
+                "message": w["message"],
+                "hint": f"use a value {op} {bound}",
+            }
+            # Only a hard reject on a node the server will run; on a pruned node
+            # it stays advisory (matching pre-promotion behavior for that node).
+            if range_is_error:
+                errors.append(entry)
+            else:
+                # Demoted to an advisory warning: match the fully-qualified
+                # `field` schema every other warning uses (preflight renders
+                # w["field"]), rather than leaking the bare input_name.
+                entry["field"] = f"{node_id}.{class_type}.{input_name}"
+                warnings.append(entry)
         else:
             w["field"] = f"{node_id}.{class_type}.{w['field']}"
             warnings.append(w)
diff --git a/tests/comfy_cli/cql/test_engine.py b/tests/comfy_cli/cql/test_engine.py
index b35adf4d..4d4f03f0 100644
--- a/tests/comfy_cli/cql/test_engine.py
+++ b/tests/comfy_cli/cql/test_engine.py
@@ -704,12 +704,15 @@ def test_multiple_edge_errors_reported(self, graph: Graph):
 
     def test_below_min_error(self, graph: Graph):
         """A value below the catalog min is a hard error (the server rejects it
-        with value_smaller_than_min) — was a warning before BE-3357."""
+        with value_smaller_than_min) — was a warning before BE-3357. Node "1" is
+        wired to a SaveImage output so it is server-reachable (BE-3406); an
+        unreachable node would be pruned and the range demoted to a warning."""
         wf = {
             "1": {
                 "class_type": "EmptyLatentImage",
                 "inputs": {"width": 0, "height": 512, "batch_size": 1},
             },
+            "2": {"class_type": "SaveImage", "inputs": {"images": ["1", 0], "filename_prefix": "out"}},
         }
         result = graph.validate_workflow(wf)
         assert result["valid"] is False
@@ -720,12 +723,14 @@ def test_below_min_error(self, graph: Graph):
         assert "below_min" not in [w["code"] for w in result["warnings"]]
 
     def test_above_max_error(self, graph: Graph):
-        """A value above the catalog max is a hard error (value_bigger_than_max)."""
+        """A value above the catalog max is a hard error (value_bigger_than_max).
+        Node "1" is wired to a SaveImage output so it is server-reachable."""
         wf = {
             "1": {
                 "class_type": "EmptyLatentImage",
                 "inputs": {"width": 999999, "height": 512, "batch_size": 1},
             },
+            "2": {"class_type": "SaveImage", "inputs": {"images": ["1", 0], "filename_prefix": "out"}},
         }
         result = graph.validate_workflow(wf)
         assert result["valid"] is False
@@ -782,8 +787,11 @@ def test_bare_link_wiring_errors_with_slot_hint(self, graph: Graph):
         assert "images.image0" in err["hint"]
 
     def test_required_autogrow_with_no_slots_errors(self, graph: Graph):
+        # BatchImagesNode "20" is wired to a SaveImage output so it is
+        # server-reachable (BE-3406) — an unreachable node would be pruned.
         wf = {
             "20": {"class_type": "BatchImagesNode", "inputs": {}},
+            "30": {"class_type": "SaveImage", "inputs": {"images": ["20", 0], "filename_prefix": "out"}},
         }
         result = graph.validate_workflow(wf)
         assert result["valid"] is False
@@ -996,6 +1004,120 @@ def test_meta_key_still_exempt(self, graph_sd15: Graph):
         assert result["valid"] is True, result["errors"]
         assert [e for e in result["errors"] if e.get("node_id") == "_meta"] == []
 
+    # -- Output-reachability pruning (BE-3406): match the server, which only
+    # validates output nodes and their transitive input ancestors. --
+
+    def test_disconnected_node_missing_required_is_pruned(self, graph_sd15: Graph):
+        """Acceptance (a): a disconnected KSampler missing all its required
+        inputs, alongside a valid connected output chain, does not fail
+        validation — the server prunes it (never reachable from an output), so
+        we must not hard-reject the whole prompt on it."""
+        wf = self._sd15_full()
+        # A stray KSampler wired to nothing and referenced by nothing: not
+        # reachable from SaveImage, so the server never validates it.
+        wf["99"] = {"class_type": "KSampler", "inputs": {"seed": 1}}
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is True, result["errors"]
+        assert [e for e in result["errors"] if e["node_id"] == "99"] == []
+
+    def test_reachable_node_missing_required_still_errors(self, graph_sd15: Graph):
+        """Acceptance (b): a node ON the output chain that is missing a required
+        input still hard-errors — reachability doesn't weaken real validation."""
+        wf = self._sd15_full()
+        del wf["3"]["inputs"]["seed"]  # KSampler feeds VAEDecode → SaveImage
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is False
+        missing = [e for e in result["errors"] if e["code"] == "required_input_missing" and e["node_id"] == "3"]
+        assert {e["field"] for e in missing} == {"seed"}
+
+    def test_transitive_ancestor_missing_required_still_errors(self, graph_sd15: Graph):
+        """A *transitive* ancestor (CLIPTextEncode, two hops upstream of the
+        SaveImage output) is reachable and its missing required input errors —
+        proving the backward walk follows link edges, not just direct parents."""
+        wf = self._sd15_full()
+        del wf["6"]["inputs"]["text"]  # "6" → KSampler.positive → VAEDecode → SaveImage
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is False
+        missing = [e for e in result["errors"] if e["code"] == "required_input_missing" and e["node_id"] == "6"]
+        assert {e["field"] for e in missing} == {"text"}
+
+    def test_output_node_missing_required_still_errors(self, graph_sd15: Graph):
+        """The output node itself seeds the reachable set, so a required input
+        missing on SaveImage still errors."""
+        wf = self._sd15_full()
+        del wf["9"]["inputs"]["filename_prefix"]
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is False
+        assert any(
+            e["code"] == "required_input_missing" and e["node_id"] == "9" and e["field"] == "filename_prefix"
+            for e in result["errors"]
+        )
+
+    def test_disconnected_out_of_range_demoted_to_warning(self, graph_sd15: Graph):
+        """A below_min value on a disconnected node is a warning, not a hard
+        error — the server never range-checks a pruned node. The connected chain
+        stays valid."""
+        wf = self._sd15_full()
+        # A stray EmptyLatentImage (all required inputs present) with an
+        # out-of-range width, wired to nothing.
+        wf["99"] = {"class_type": "EmptyLatentImage", "inputs": {"width": 1, "height": 512, "batch_size": 1}}
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is True, result["errors"]
+        assert [e for e in result["errors"] if e["node_id"] == "99"] == []
+        warned = [w for w in result["warnings"] if w.get("code") == "below_min" and w.get("node_id") == "99"]
+        assert len(warned) == 1
+
+    def test_reachable_out_of_range_still_errors(self, graph_sd15: Graph):
+        """The connected EmptyLatentImage feeding the output chain still
+        hard-errors on an out-of-range width (reachability preserves the #551
+        promotion where it matters)."""
+        wf = self._sd15_full()
+        wf["5"]["inputs"]["width"] = 1  # "5" → KSampler.latent_image → … → SaveImage
+        result = graph_sd15.validate_workflow(wf)
+        assert result["valid"] is False
+        assert [e for e in result["errors"] if e["code"] == "below_min" and e["node_id"] == "5"]
+
+    def test_demoted_range_warning_field_is_qualified(self, graph_sd15: Graph):
+        """A range violation demoted to a warning on a pruned node uses the same
+        fully-qualified `field` (`node.class.input`) as every other warning, so
+        consumers (e.g. preflight renders w["field"]) see one schema."""
+        wf = self._sd15_full()
+        wf["99"] = {"class_type": "EmptyLatentImage", "inputs": {"width": 1, "height": 512, "batch_size": 1}}
+        result = graph_sd15.validate_workflow(wf)
+        warned = [w for w in result["warnings"] if w.get("code") == "below_min" and w.get("node_id") == "99"]
+        assert len(warned) == 1
+        assert warned[0]["field"] == "99.EmptyLatentImage.width"
+
+
+class TestValidateMalformedInputs:
+    """Malformed workflow JSON must yield structured output, never an unhandled
+    traceback (BE-3406 hardening) — the validator's whole contract is to catch
+    bad prompts, so it may not crash on the shapes it's meant to reject."""
+
+    def test_non_dict_inputs_does_not_crash(self, graph: Graph):
+        """A truthy non-dict `inputs` (string/list from malformed JSON) slips
+        past `or {}` and would crash `.items()`/`.values()`; validation must
+        instead return a result. Node is wired to a SaveImage so it's reachable
+        (exercises both the per-input loop and the reachability walk)."""
+        wf = {
+            "1": {"class_type": "EmptyLatentImage", "inputs": "not-a-dict"},
+            "2": {"class_type": "SaveImage", "inputs": {"images": ["1", 0], "filename_prefix": "out"}},
+        }
+        result = graph.validate_workflow(wf)  # must not raise
+        assert isinstance(result["errors"], list)
+        assert isinstance(result["warnings"], list)
+
+    def test_unhashable_class_type_does_not_crash(self, graph: Graph):
+        """An unhashable class_type (list/dict) would raise TypeError in the
+        `self._nodes.get(class_type)` lookup and the reachability walk's
+        `graph.node(...)`; both are screened so validation returns a result."""
+        wf = {
+            "1": {"class_type": ["EmptyLatentImage"], "inputs": {"width": 512}},
+            "2": {"class_type": "SaveImage", "inputs": {"images": ["1", 0], "filename_prefix": "out"}},
+        }
+        result = graph.validate_workflow(wf)  # must not raise
+        assert isinstance(result["errors"], list)
+
 
 # ===========================================================================
 # TestDirectModeSlots