feat(dify-builder): graph_ops.filter_applicable dry-run validation

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

api/services/dify_builder/graph_ops.py

@@ -164,8 +164,7 @@ def apply_delete_node(graph: Graph, node_id: str) -> tuple[Graph, list[str]]:
 
     new_graph["nodes"] = [n for n in nodes if n.get("id") != node_id]
     new_graph["edges"] = [
-        e for e in new_graph.get("edges", [])
-        if e.get("source") != node_id and e.get("target") != node_id
+        e for e in new_graph.get("edges", []) if e.get("source") != node_id and e.get("target") != node_id
     ]
     return new_graph, [node_id]
 
@@ -235,9 +234,7 @@ def apply_insert_between(
     old_target = edge.get("target")
     new_graph = copy.deepcopy(graph)
     edges = new_graph.get("edges", [])
-    matched = next(
-        (e for e in edges if e.get("source") == old_source and e.get("target") == old_target), None
-    )
+    matched = next((e for e in edges if e.get("source") == old_source and e.get("target") == old_target), None)
     if matched is None:
         raise ValueError(f"edge not found: {old_source} -> {old_target}")
 
@@ -287,16 +284,13 @@ def diff_graphs(before: Graph, after: Graph) -> tuple[list[str], str]:
     changes.extend(f"added node {node_id}" for node_id in added_nodes)
     changes.extend(f"removed node {node_id}" for node_id in removed_nodes)
     changes.extend(f"added {source} → {target}" for source, target, _sh, _th in added_edges)
-    changes.extend(
-        f"removed {source} → {target}" for source, target, _sh, _th in removed_edges
-    )
+    changes.extend(f"removed {source} → {target}" for source, target, _sh, _th in removed_edges)
 
     for node_id in sorted(before_nodes.keys() & after_nodes.keys()):
         before_data = before_nodes[node_id].get("data", {})
         after_data = after_nodes[node_id].get("data", {})
         changed_keys = sorted(
-            key for key in set(before_data) | set(after_data)
-            if before_data.get(key) != after_data.get(key)
+            key for key in set(before_data) | set(after_data) if before_data.get(key) != after_data.get(key)
         )
         changes.extend(f"{node_id}: {key} updated" for key in changed_keys)
 
@@ -311,10 +305,7 @@ def structural_fingerprint(graph: Graph) -> str:
     removing, or reconnecting a node changes it. Used by recovery (C-1) to
     tell a config-only hand-edit from a structural one. Order-independent
     (nodes and edges are sorted before hashing)."""
-    nodes = sorted(
-        (str(n.get("id", "")), str((n.get("data") or {}).get("type", "")))
-        for n in graph.get("nodes", [])
-    )
+    nodes = sorted((str(n.get("id", "")), str((n.get("data") or {}).get("type", ""))) for n in graph.get("nodes", []))
     edges = sorted(
         (
             str(e.get("source", "")),
@@ -332,3 +323,53 @@ def node_ids(graph: Graph) -> list[str]:
     """The ids of every node in ``graph``, in document order (missing-id
     nodes skipped). Used by recovery's target-presence check."""
     return [str(n["id"]) for n in graph.get("nodes", []) if n.get("id") is not None]
+
+
+STRUCTURAL_OPS: frozenset[str] = frozenset({"create_node", "delete_node", "connect", "insert_between"})
+
+_APPLY_FNS = {
+    "set_node_config": apply_set_node_config,
+    "create_node": apply_create_node,
+    "delete_node": apply_delete_node,
+    "connect": apply_connect,
+    "insert_between": apply_insert_between,
+}
+
+
+def filter_applicable(
+    graph: Graph,
+    intents: list[MutationIntent],
+    allowed_node_types: set[str] | None = None,
+) -> tuple[list[MutationIntent], list[tuple[MutationIntent, str]]]:
+    """Dry-run each intent through the real validate_intent_args + apply_* on a
+    working deep copy, in order (so a connect sees a node an earlier create_node
+    added). Returns (applicable, rejected) where each rejected entry is
+    (intent, reason).
+
+    Catches every failure the live apply_repair would hit -- unknown op, missing
+    required arg / extra arg key, dangling node/edge ref, duplicate id -- and
+    additionally rejects a create_node / insert_between whose node_type is not in
+    allowed_node_types (a check apply_* does not do). Applicable intents advance
+    the working copy; a rejected intent does not.
+    """
+    working = copy.deepcopy(graph)
+    applicable: list[MutationIntent] = []
+    rejected: list[tuple[MutationIntent, str]] = []
+    for intent in intents:
+        apply_fn = _APPLY_FNS.get(intent.op)
+        if apply_fn is None:
+            rejected.append((intent, f"unknown mutation op: {intent.op!r}"))
+            continue
+        if allowed_node_types is not None and intent.op in ("create_node", "insert_between"):
+            node_type = intent.args.get("node_type")
+            if node_type not in allowed_node_types:
+                rejected.append((intent, f"node_type not allowed: {node_type!r}"))
+                continue
+        try:
+            validate_intent_args(intent)
+            working, _changed = apply_fn(working, **intent.args)
+        except (ValueError, TypeError) as exc:
+            rejected.append((intent, str(exc)))
+            continue
+        applicable.append(intent)
+    return applicable, rejected

api/tests/unit_tests/services/dify_builder/test_graph_ops.py

@@ -12,6 +12,7 @@ import copy
 import pytest
 
 from core.dify_builder.models import MutationIntent
+from services.dify_builder import graph_ops
 from services.dify_builder.graph_ops import (
     MUTATION_ARG_KEYS,
     apply_connect,
@@ -475,3 +476,67 @@ def test_node_ids():
     g = {"nodes": [{"id": "a", "data": {"type": "llm"}}, {"id": "b", "data": {"type": "end"}}], "edges": []}
     assert node_ids(g) == ["a", "b"]
     assert node_ids({}) == []
+
+
+# ---- filter_applicable -------------------------------------------------------
+
+
+_G = {
+    "nodes": [
+        {"id": "a", "type": "custom", "data": {"type": "code", "title": "Code"}},
+        {"id": "b", "type": "custom", "data": {"type": "end", "title": "End"}},
+    ],
+    "edges": [{"id": "a-b", "source": "a", "target": "b"}],
+}
+
+
+def test_filter_applicable_all_valid():
+    intents = [MutationIntent(op="set_node_config", args={"node_id": "a", "path": "code", "value": "x"})]
+    ok, bad = graph_ops.filter_applicable(_G, intents)
+    assert ok == intents
+    assert bad == []
+
+
+def test_filter_rejects_unknown_op():
+    intents = [MutationIntent(op="frobnicate", args={})]
+    ok, bad = graph_ops.filter_applicable(_G, intents)
+    assert ok == []
+    assert "unknown mutation op" in bad[0][1]
+
+
+def test_filter_rejects_missing_required_arg():
+    intents = [MutationIntent(op="set_node_config", args={"node_id": "a"})]
+    ok, bad = graph_ops.filter_applicable(_G, intents)
+    assert ok == []
+    assert "missing required arg" in bad[0][1]
+
+
+def test_filter_rejects_extra_arg_key():
+    intents = [MutationIntent(op="delete_node", args={"node_id": "a", "bogus": 1})]
+    ok, bad = graph_ops.filter_applicable(_G, intents)
+    assert ok == []
+    assert "bogus" in bad[0][1]
+
+
+def test_filter_rejects_dangling_ref():
+    intents = [MutationIntent(op="connect", args={"from_node": "a", "to_node": "zzz"})]
+    ok, bad = graph_ops.filter_applicable(_G, intents)
+    assert ok == []
+    assert "node not found" in bad[0][1]
+
+
+def test_filter_rejects_unknown_node_type():
+    intents = [MutationIntent(op="create_node", args={"node_type": "made-up", "config": {}})]
+    ok, bad = graph_ops.filter_applicable(_G, intents, allowed_node_types={"llm", "end"})
+    assert ok == []
+    assert "node_type not allowed" in bad[0][1]
+
+
+def test_filter_create_then_connect_ordering():
+    intents = [
+        MutationIntent(op="create_node", args={"node_type": "llm", "config": {}, "node_id": "c"}),
+        MutationIntent(op="connect", args={"from_node": "c", "to_node": "b"}),
+    ]
+    ok, bad = graph_ops.filter_applicable(_G, intents, allowed_node_types={"llm", "end"})
+    assert ok == intents  # the connect sees the node the prior create added
+    assert bad == []