feat(workflow): fail-closed branch-handle repair and declared-handle check

The planner names a branch's outgoing source_handle values in one LLM
call; the node builder chooses the case / class / action ids in another.
ESQ1-303's edges left its if-else on "score_equals_60" and "else"
against a case named "true", so the engine skipped both arms and the
run succeeded with nothing.

declared_branch_handles lists what a node routes on; repair_branch_edge_
handles re-homes an edge only when it is forced (exact / alias / name /
stem match, one arm fanning out under a single invented name, default-
source edges onto unused handles in order) and returns everything
ambiguous untouched; undeclared_branch_handles reports what is left so a
validator can reject it. A wrong guess would silently swap IF and ELSE.
Aliases are trusted as names -- the documented limit -- which is why the
node-builder handle contract, not this repair, is the primary fix.

A fail-branch node's success path stays on "source" (what graphon's
NodeRunResult.edge_source_handle actually emits by default), not an
invented "success" handle nothing ever routes on; a fail-branch
branch-type node (if-else / question-classifier / human-input) just
gains the extra "fail-branch" arm. human-input also always declares its
implicit "__timeout" arm, since every human-input node has a timeout
config and the FE always renders that handle; that arm is excluded from
the "unused declared handles" pool the fan-out/default-order repair
rules count against, since it is routinely left unwired by design and
must not block or absorb repair of the arms the author did configure.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

api/core/workflow/graph_normalizers.py

@@ -316,3 +316,237 @@ def http_request_body_errors(nodes: list[Any]) -> list[tuple[str, str]]:
                 )
             )
     return errors
+
+
+# ---------------------------------------------------------------------------
+# Branch edge handles (if-else / question-classifier / human-input / fail-branch)
+# ---------------------------------------------------------------------------
+
+# Names the planner uses for the arms of a two-arm if-else before the node
+# builder has chosen a case id. Trusted as NAMES: an ELSE alias always means
+# the implicit ``false`` arm, an IF alias the single declared case.
+_ELSE_ALIASES = frozenset(
+    {"else", "otherwise", "default", "no", "not", "elsebranch", "else_branch", "false_branch", "falsebranch"}
+)
+_IF_ALIASES = frozenset(
+    {"if", "then", "yes", "true_branch", "truebranch", "if_branch", "ifbranch", "case", "case1", "case_1"}
+)
+_DEFAULT_HANDLES = frozenset({"", "source"})
+# A handle shorter than this never claims a declared id by prefix ("no" must
+# not swallow "nothing").
+_MIN_STEM = 3
+
+# Node types that route on named handles the node itself declares (if-else's
+# case ids + the implicit "false" arm, question-classifier's class ids,
+# human-input's action ids + the implicit "__timeout" arm) rather than the
+# single default "source" handle every other node type uses.
+_NAMED_BRANCH_TYPES = frozenset({"if-else", "question-classifier", "human-input"})
+
+# human-input's implicit timeout arm. This is
+# ``core.workflow.nodes.human_input.constants.TIMEOUT_HANDLE`` ("__timeout"),
+# not imported: that package's __init__ pulls in graphon/pydantic entities,
+# which this module's own contract (see the module docstring) keeps out --
+# "no engine imports beyond the graph dict shape".
+_HUMAN_INPUT_TIMEOUT_HANDLE = "__timeout"
+
+# Declared handles that exist on every node of their type regardless of what
+# the author configured (unlike a case id, a class id or an action id, which
+# only exist because the author added them). An author routinely leaves them
+# unwired, so they must not count as "the one remaining slot" when repair is
+# deciding whether an unnamed/misnamed edge unambiguously belongs to the last
+# free arm -- otherwise that arm would never resolve. An edge already ON one
+# of them still matches it exactly (the pass above, not this set).
+_UNCOUNTED_UNUSED_HANDLES = frozenset({_HUMAN_INPUT_TIMEOUT_HANDLE})
+
+
+def _canon(handle: Any) -> str:
+    return "".join(ch for ch in str(handle or "").lower() if ch.isalnum() or ch == "_")
+
+
+def declared_branch_handles(node: Mapping[str, Any]) -> list[str]:
+    """The source handles ``node`` exposes, in declaration order; ``[]`` for a
+    node that has a single ``source`` handle.
+
+    A node with ``error_strategy: "fail-branch"`` gains a ``"fail-branch"``
+    arm. Its success path stays on whatever handle the node already routes
+    on: graphon's ``NodeRunResult.edge_source_handle`` defaults to
+    ``"source"``, so a plain node (http-request, code, ...) declares
+    ``["source", "fail-branch"]`` -- not an invented ``"success"`` handle
+    nothing ever emits, which would strand its real ``"source"`` edge as
+    undeclared and skip it at run time. A branch-type node (if-else,
+    question-classifier, human-input) keeps routing on its own handles and
+    simply gains the extra ``"fail-branch"`` arm.
+    """
+    data = node.get("data")
+    if not isinstance(data, Mapping):
+        return []
+    node_type = data.get("type")
+    handles: list[str] = []
+    if node_type == "if-else":
+        handles = [str(c["case_id"]) for c in data.get("cases") or [] if isinstance(c, Mapping) and c.get("case_id")]
+        handles.append("false")  # the implicit ELSE arm
+    elif node_type == "question-classifier":
+        handles = [str(k["id"]) for k in data.get("classes") or [] if isinstance(k, Mapping) and k.get("id")]
+    elif node_type == "human-input":
+        handles = [str(a["id"]) for a in data.get("user_actions") or [] if isinstance(a, Mapping) and a.get("id")]
+        handles.append(_HUMAN_INPUT_TIMEOUT_HANDLE)  # the implicit timeout arm
+    if data.get("error_strategy") == "fail-branch":
+        if node_type in _NAMED_BRANCH_TYPES:
+            handles.append("fail-branch")
+        else:
+            handles = ["source", "fail-branch"]
+    return handles
+
+
+def _name_aliases(node: Mapping[str, Any], handles: list[str]) -> dict[str, str]:
+    """canonical name -> declared handle, for the names a planner might use."""
+    data = node.get("data") or {}
+    names: dict[str, str] = {}
+    if data.get("type") == "if-else":
+        for alias in _ELSE_ALIASES:
+            names[alias] = "false"
+        if len(handles) == 2:  # one case + ELSE: an IF alias is unambiguous
+            for alias in _IF_ALIASES:
+                names[alias] = handles[0]
+    elif data.get("type") == "question-classifier":
+        for klass in data.get("classes") or []:
+            if isinstance(klass, Mapping) and klass.get("id") and klass.get("name"):
+                names[_canon(klass["name"])] = str(klass["id"])
+    elif data.get("type") == "human-input":
+        for action in data.get("user_actions") or []:
+            if isinstance(action, Mapping) and action.get("id"):
+                for key in ("title", "label", "name"):
+                    if action.get(key):
+                        names[_canon(action[key])] = str(action["id"])
+    return names
+
+
+def _match_handle(handle: str, handles: list[str], names: Mapping[str, str]) -> str | None:
+    """A declared handle that ``handle`` unambiguously means, else ``None``."""
+    canon = _canon(handle)
+    for declared in handles:
+        if _canon(declared) == canon:
+            return declared
+    if canon in names:
+        return names[canon]
+    if len(canon) >= _MIN_STEM:
+        stems = [
+            declared
+            for declared in handles
+            if len(_canon(declared)) >= _MIN_STEM
+            and (canon.startswith(_canon(declared)) or _canon(declared).startswith(canon))
+        ]
+        if len(stems) == 1:
+            return stems[0]
+    return None
+
+
+def repair_branch_edge_handles(nodes: list[Any], edges: list[Any]) -> list[dict[str, Any]]:
+    """Re-home edges leaving a branch node onto the handles the node declares.
+
+    Three passes per branch node, each only when it is forced:
+
+    1. exact / alias / name / stem match (``else`` -> ``false``, a class name
+       -> its id, ``approved`` -> ``approve``);
+    2. edges still unmatched, all carrying the SAME unknown handle, when
+       exactly one declared handle is unused -> that handle (one arm fanning
+       out under an invented name);
+    3. edges on the default ``source`` handle, when there are at least as
+       many unused declared handles -> unused handles in declaration order
+       (the pre-existing behaviour).
+
+    "Unused" for (2) and (3) never counts a handle in
+    ``_UNCOUNTED_UNUSED_HANDLES`` (human-input's implicit ``"__timeout"``
+    arm): it exists on every node of its type whether or not the author
+    wired it, so it must not silently absorb an edge meant for the one real
+    arm that IS left unused, and must not block that arm's resolution either.
+
+    Anything else -- two different unknown names, one unknown edge with two
+    free arms, a three-case node with invented names -- is left exactly as it
+    was and returned, so ``undeclared_branch_handles`` fails the graph closed.
+    A wrong guess would silently swap the IF and ELSE arms; a visible rejection
+    is better.
+    """
+    unresolved: list[dict[str, Any]] = []
+    for node in nodes:
+        if not isinstance(node, Mapping):
+            continue
+        handles = declared_branch_handles(node)
+        if not handles:
+            continue
+        node_id = str(node.get("id") or "")
+        outgoing = [e for e in edges if isinstance(e, MutableMapping) and e.get("source") == node_id]
+        if not outgoing:
+            continue
+        names = _name_aliases(node, handles)
+        unknown: list[MutableMapping[str, Any]] = []
+        for edge in outgoing:
+            handle = edge.get("sourceHandle")
+            if handle in handles:
+                continue
+            if handle is None or handle in _DEFAULT_HANDLES:
+                unknown.append(edge)
+                continue
+            matched = _match_handle(str(handle), handles, names)
+            if matched is None:
+                unknown.append(edge)
+            else:
+                edge["sourceHandle"] = matched
+        if not unknown:
+            continue
+        taken = {e.get("sourceHandle") for e in outgoing if e.get("sourceHandle") in handles}
+        unused = [h for h in handles if h not in taken and h not in _UNCOUNTED_UNUSED_HANDLES]
+        named = [
+            e
+            for e in unknown
+            if e.get("sourceHandle") not in (None, "") and e.get("sourceHandle") not in _DEFAULT_HANDLES
+        ]
+        defaulted = [e for e in unknown if e not in named]
+        distinct_names = {str(e.get("sourceHandle")) for e in named}
+        if named and not defaulted and len(distinct_names) == 1 and len(unused) == 1:
+            for edge in named:
+                edge["sourceHandle"] = unused[0]
+            continue
+        if defaulted and not named and len(defaulted) <= len(unused):
+            for edge, handle in zip(defaulted, unused):
+                edge["sourceHandle"] = handle
+            continue
+        for edge in unknown:
+            unresolved.append(
+                {
+                    "node_id": node_id,
+                    "target": str(edge.get("target") or ""),
+                    "handle": edge.get("sourceHandle"),
+                    "declared": list(handles),
+                }
+            )
+    return unresolved
+
+
+def undeclared_branch_handles(nodes: list[Any], edges: list[Any]) -> list[dict[str, Any]]:
+    """Every edge leaving a branch node on a handle the node does not declare
+    (including the default ``source``): the edge would hang off a handle that
+    does not exist and its arm would silently never run."""
+    bad: list[dict[str, Any]] = []
+    for node in nodes:
+        if not isinstance(node, Mapping):
+            continue
+        handles = declared_branch_handles(node)
+        if not handles:
+            continue
+        node_id = str(node.get("id") or "")
+        for edge in edges:
+            if not isinstance(edge, Mapping) or edge.get("source") != node_id:
+                continue
+            handle = edge.get("sourceHandle")
+            if handle in handles:
+                continue
+            bad.append(
+                {
+                    "node_id": node_id,
+                    "target": str(edge.get("target") or ""),
+                    "handle": "source" if handle is None else handle,
+                    "declared": list(handles),
+                }
+            )
+    return bad

api/tests/unit_tests/core/workflow/test_graph_normalizers.py

@@ -511,3 +511,271 @@ class TestHttpRequestBodyErrors:
         result = normalize_http_request_bodies(nodes)
         assert result == []
         assert nodes == before
+
+
+def _esq1_303_graph() -> dict:
+    return json.loads((_FIXTURES / "esq1_303_draft_graph.json").read_text(encoding="utf-8"))
+
+
+def _if_else(node_id: str, case_ids: tuple[str, ...] = ("true",)) -> dict:
+    return {
+        "id": node_id,
+        "data": {
+            "type": "if-else",
+            "cases": [{"case_id": c, "logical_operator": "and", "conditions": []} for c in case_ids],
+        },
+    }
+
+
+def _edges(source: str, pairs: list[tuple]) -> list[dict]:
+    return [{"source": source, "target": target, "sourceHandle": handle} for handle, target in pairs]
+
+
+def _handles(edges: list[dict]) -> list[tuple]:
+    return [(e["target"], e.get("sourceHandle")) for e in edges]
+
+
+class TestDeclaredBranchHandles:
+    def test_if_else_declares_its_case_ids_plus_the_implicit_else(self):
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        assert declared_branch_handles(_if_else("n", ("true",))) == ["true", "false"]
+        assert declared_branch_handles(_if_else("n", ("c1", "c2"))) == ["c1", "c2", "false"]
+
+    def test_question_classifier_and_human_input_declare_their_ids(self):
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        qc = {"id": "q", "data": {"type": "question-classifier", "classes": [{"id": "1", "name": "A"}, {"id": "2"}]}}
+        hi = {"id": "h", "data": {"type": "human-input", "user_actions": [{"id": "approve"}, {"id": "deny"}]}}
+        assert declared_branch_handles(qc) == ["1", "2"]
+        # human-input always routes an implicit timeout arm too (R1(b)).
+        assert declared_branch_handles(hi) == ["approve", "deny", "__timeout"]
+
+    def test_human_input_declares_its_action_ids_plus_the_implicit_timeout(self):
+        # (i) R1(b): human-input routes an implicit "__timeout" arm
+        # (TIMEOUT_HANDLE, core/workflow/nodes/human_input/constants.py) in
+        # addition to its declared user actions.
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        hi = {"id": "h", "data": {"type": "human-input", "user_actions": [{"id": "approve"}, {"id": "deny"}]}}
+        assert declared_branch_handles(hi) == ["approve", "deny", "__timeout"]
+
+    def test_fail_branch_nodes_declare_source_and_fail_branch(self):
+        # R1(a): a fail-branch node's SUCCESS path stays on the handle
+        # graphon actually emits (NodeRunResult.edge_source_handle defaults
+        # to "source"), not an invented "success" handle nothing emits.
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        node = {"id": "h", "data": {"type": "http-request", "error_strategy": "fail-branch"}}
+        assert declared_branch_handles(node) == ["source", "fail-branch"]
+
+    def test_if_else_with_fail_branch_declares_cases_false_and_fail_branch(self):
+        # (iv) a branch-type node with error_strategy fail-branch appends
+        # "fail-branch" to its own declared handles rather than replacing them.
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        node = _if_else("n", ("c1", "c2"))
+        node["data"]["error_strategy"] = "fail-branch"
+        assert declared_branch_handles(node) == ["c1", "c2", "false", "fail-branch"]
+
+    def test_plain_nodes_declare_nothing(self):
+        from core.workflow.graph_normalizers import declared_branch_handles
+
+        assert declared_branch_handles({"id": "l", "data": {"type": "llm"}}) == []
+
+
+class TestRepairBranchEdgeHandles:
+    """The planner names handles before the node builder picks case ids
+    (ESQ1-303: ``score_equals_60`` / ``else`` against a case named ``true``).
+    Repair only what is forced; leave anything ambiguous for the validator."""
+
+    def test_heals_the_esq1_303_dev_edges_else_alias_pins_the_other_arm(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        graph = _esq1_303_graph()
+
+        unresolved = repair_branch_edge_handles(graph["nodes"], graph["edges"])
+
+        assert unresolved == []
+        by_target = {e["target"]: e["sourceHandle"] for e in graph["edges"] if e["source"] == "node2"}
+        assert by_target == {"node3": "true", "node4": "false"}
+
+    def test_legacy_default_handle_edges_still_take_unused_handles_in_order(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("source", "a"), (None, "b")])
+        assert repair_branch_edge_handles([_if_else("n")], edges) == []
+        assert _handles(edges) == [("a", "true"), ("b", "false")]
+
+    def test_three_default_edges_on_two_handles_stay_untouched(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("source", "a"), ("source", "b"), ("source", "c")])
+        unresolved = repair_branch_edge_handles([_if_else("n")], edges)
+        assert len(unresolved) == 3
+        assert all(h == "source" for _, h in _handles(edges))
+
+    def test_both_edges_unknown_with_no_alias_is_ambiguous_and_untouched(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("pass", "a"), ("fail", "b")])
+        unresolved = repair_branch_edge_handles([_if_else("n")], edges)
+        assert [u["handle"] for u in unresolved] == ["pass", "fail"]
+        assert _handles(edges) == [("a", "pass"), ("b", "fail")]
+
+    def test_a_single_unknown_edge_with_two_free_arms_is_ambiguous(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("maybe", "a")])
+        assert len(repair_branch_edge_handles([_if_else("n")], edges)) == 1
+        assert _handles(edges) == [("a", "maybe")]
+
+    def test_three_case_if_else_with_invented_names_is_ambiguous(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("low", "a"), ("mid", "b"), ("high", "c")])
+        assert len(repair_branch_edge_handles([_if_else("n", ("case1", "case2"))], edges)) == 3
+        assert _handles(edges) == [("a", "low"), ("b", "mid"), ("c", "high")]
+
+    def test_one_exact_plus_else_alias_forces_the_last_unknown_onto_the_last_free_case(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("case1", "a"), ("whatever", "b"), ("else", "c")])
+        assert repair_branch_edge_handles([_if_else("n", ("case1", "case2"))], edges) == []
+        assert _handles(edges) == [("a", "case1"), ("b", "case2"), ("c", "false")]
+
+    def test_an_if_alias_while_true_is_taken_is_a_fan_out_not_the_else_arm(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("true", "a"), ("if", "b")])
+        repair_branch_edge_handles([_if_else("n")], edges)
+        assert _handles(edges) == [("a", "true"), ("b", "true")]
+
+    def test_the_same_unknown_handle_twice_is_one_arm_fanning_out(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = _edges("n", [("false", "a"), ("matched", "b"), ("matched", "c")])
+        assert repair_branch_edge_handles([_if_else("n")], edges) == []
+        assert _handles(edges) == [("a", "false"), ("b", "true"), ("c", "true")]
+
+    def test_question_classifier_matches_class_names_case_and_space_insensitively(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        nodes = [
+            {
+                "id": "q",
+                "data": {
+                    "type": "question-classifier",
+                    "classes": [{"id": "1", "name": "Billing"}, {"id": "2", "name": "Tech Support"}],
+                },
+            }
+        ]
+        edges = _edges("q", [("billing", "a"), ("tech_support", "b")])
+        assert repair_branch_edge_handles(nodes, edges) == []
+        assert _handles(edges) == [("a", "1"), ("b", "2")]
+
+    def test_human_input_matches_action_stems_then_eliminates(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        nodes = [
+            {
+                "id": "h",
+                "data": {
+                    "type": "human-input",
+                    "user_actions": [{"id": "approve", "title": "Approve"}, {"id": "deny", "title": "Deny"}],
+                },
+            }
+        ]
+        edges = _edges("h", [("approved", "a"), ("rejected", "b")])
+        assert repair_branch_edge_handles(nodes, edges) == []
+        assert _handles(edges) == [("a", "approve"), ("b", "deny")]
+
+    def test_a_short_stem_never_matches_by_prefix(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        # "no" is too short a stem to claim "nothing"; both arms are free, so refuse.
+        nodes = [{"id": "h", "data": {"type": "human-input", "user_actions": [{"id": "no"}, {"id": "later"}]}}]
+        edges = _edges("h", [("nothing", "a")])
+        assert len(repair_branch_edge_handles(nodes, edges)) == 1
+
+    def test_alias_trust_documented_limit_a_mislabelled_else_alias_swaps_the_arms(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        # The planner put an ELSE alias on the IF-arm edge and "true" on the
+        # other. Names are trusted as names: this maps "default" -> false. No
+        # handle repair can detect a planner/builder semantic inversion; the
+        # node-builder handle contract (prompt) is what prevents it.
+        edges = _edges("n", [("default", "a"), ("true", "b")])
+        repair_branch_edge_handles([_if_else("n")], edges)
+        assert _handles(edges) == [("a", "false"), ("b", "true")]
+
+    def test_non_branch_nodes_are_untouched(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        edges = [{"source": "llm1", "target": "end", "sourceHandle": "source"}]
+        assert repair_branch_edge_handles([{"id": "llm1", "data": {"type": "llm"}}], edges) == []
+        assert edges[0]["sourceHandle"] == "source"
+
+    def test_fail_branch_node_success_on_source_and_failure_on_fail_branch_is_untouched(self):
+        # (ii) R1(a): the success edge is already on "source" (what graphon
+        # actually emits) and the failure edge on "fail-branch"; both are
+        # exact matches, so nothing is re-homed.
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        nodes = [{"id": "h", "data": {"type": "http-request", "error_strategy": "fail-branch"}}]
+        edges = _edges("h", [("source", "ok"), ("fail-branch", "err")])
+        assert repair_branch_edge_handles(nodes, edges) == []
+        assert _handles(edges) == [("ok", "source"), ("err", "fail-branch")]
+
+    def test_human_input_timeout_edge_is_untouched(self):
+        # (iii) the implicit "__timeout" arm is a declared handle; an edge
+        # already on it is an exact match and is left alone.
+        from core.workflow.graph_normalizers import repair_branch_edge_handles
+
+        nodes = [{"id": "h", "data": {"type": "human-input", "user_actions": [{"id": "approve"}, {"id": "deny"}]}}]
+        edges = _edges("h", [("approve", "a"), ("deny", "b"), ("__timeout", "c")])
+        assert repair_branch_edge_handles(nodes, edges) == []
+        assert _handles(edges) == [("a", "approve"), ("b", "deny"), ("c", "__timeout")]
+
+
+class TestUndeclaredBranchHandles:
+    def test_reports_every_edge_whose_handle_the_node_does_not_declare(self):
+        from core.workflow.graph_normalizers import undeclared_branch_handles
+
+        graph = _esq1_303_graph()  # before repair: score_equals_60 / else
+        bad = undeclared_branch_handles(graph["nodes"], graph["edges"])
+
+        assert [(b["node_id"], b["handle"], b["target"]) for b in bad] == [
+            ("node2", "score_equals_60", "node3"),
+            ("node2", "else", "node4"),
+        ]
+        assert bad[0]["declared"] == ["true", "false"]
+
+    def test_default_handles_on_branch_nodes_are_undeclared_too(self):
+        from core.workflow.graph_normalizers import undeclared_branch_handles
+
+        edges = [{"source": "n", "target": "a"}, {"source": "n", "target": "b", "sourceHandle": "source"}]
+        assert [b["handle"] for b in undeclared_branch_handles([_if_else("n")], edges)] == ["source", "source"]
+
+    def test_a_clean_graph_reports_nothing(self):
+        from core.workflow.graph_normalizers import repair_branch_edge_handles, undeclared_branch_handles
+
+        graph = _esq1_303_graph()
+        repair_branch_edge_handles(graph["nodes"], graph["edges"])
+        assert undeclared_branch_handles(graph["nodes"], graph["edges"]) == []
+
+    def test_fail_branch_node_clean_edges_report_nothing(self):
+        # (ii) source + fail-branch is exactly the declared set; nothing to report.
+        from core.workflow.graph_normalizers import undeclared_branch_handles
+
+        nodes = [{"id": "h", "data": {"type": "http-request", "error_strategy": "fail-branch"}}]
+        edges = _edges("h", [("source", "ok"), ("fail-branch", "err")])
+        assert undeclared_branch_handles(nodes, edges) == []
+
+    def test_human_input_timeout_edge_is_not_reported(self):
+        # (iii) "__timeout" is declared, so it is not flagged as undeclared.
+        from core.workflow.graph_normalizers import undeclared_branch_handles
+
+        nodes = [{"id": "h", "data": {"type": "human-input", "user_actions": [{"id": "approve"}, {"id": "deny"}]}}]
+        edges = _edges("h", [("approve", "a"), ("deny", "b"), ("__timeout", "c")])
+        assert undeclared_branch_handles(nodes, edges) == []