diff --git a/api/core/workflow/graph_normalizers.py b/api/core/workflow/graph_normalizers.py index f9f957fe83f..3b8ee5a6ed3 100644 --- a/api/core/workflow/graph_normalizers.py +++ b/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 diff --git a/api/tests/unit_tests/core/workflow/test_graph_normalizers.py b/api/tests/unit_tests/core/workflow/test_graph_normalizers.py index b3acb9975ad..12593d39f2f 100644 --- a/api/tests/unit_tests/core/workflow/test_graph_normalizers.py +++ b/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) == []