chore: use pydantic to verify (#38365)

Co-authored-by: Byron.wang <byron@dify.ai>

api/core/app/apps/workflow_app_runner.py

@@ -3,7 +3,7 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
-from pydantic import ValidationError
+from pydantic import BaseModel, ConfigDict, ValidationError
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.agent_strategy import AgentStrategyInfo
@@ -56,7 +56,8 @@ from core.workflow.system_variables import (
 from core.workflow.variable_pool_initializer import add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
-from graphon.entities.graph_config import NodeConfigDictAdapter
+from graphon.entities.base_node_data import BaseNodeData
+from graphon.entities.graph_config import NodeConfigDict
 from graphon.entities.pause_reason import HitlRequired
 from graphon.enums import BuiltinNodeTypes
 from graphon.graph import Graph
@@ -96,6 +97,67 @@ from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
 logger = logging.getLogger(__name__)
 
 
+class _WorkflowGraphNodeData(BaseNodeData):
+    """Node data fields the runner needs before concrete node validation."""
+
+    start_node_id: str | None = None
+    iteration_id: str | None = None
+    loop_id: str | None = None
+
+    def parent_id_for(self, key: str) -> str | None:
+        match key:
+            case "iteration_id":
+                return self.iteration_id
+            case "loop_id":
+                return self.loop_id
+            case _:
+                return None
+
+
+class _WorkflowGraphNodeConfig(BaseModel):
+    """Top-level node wrapper used for runner graph filtering."""
+
+    id: str
+    data: _WorkflowGraphNodeData
+
+    model_config = ConfigDict(extra="allow", strict=True)
+
+    def to_graph_node_config(self) -> dict[str, Any]:
+        return {
+            **self.model_dump(mode="python", exclude={"data"}, exclude_unset=True),
+            "data": self.data.model_dump(mode="python", exclude_unset=True),
+        }
+
+    def to_typed_node_config(self) -> NodeConfigDict:
+        return cast(NodeConfigDict, {**self.model_dump(mode="python", exclude={"data"}), "data": self.data})
+
+
+class _WorkflowGraphEdgeConfig(BaseModel):
+    """Top-level edge wrapper used for selecting a single-node subgraph."""
+
+    source: str | None = None
+    target: str | None = None
+
+    model_config = ConfigDict(extra="allow", strict=True)
+
+
+class _WorkflowGraphConfig(BaseModel):
+    """Validated graph config boundary for runner methods.
+
+    Workflow graph payloads are persisted JSON mappings with extra metadata used
+    by graphon, so validation is intentionally limited to the top-level graph
+    shape this runner needs before passing the mapping onward.
+    """
+
+    nodes: list[_WorkflowGraphNodeConfig]
+    edges: list[_WorkflowGraphEdgeConfig]
+
+    model_config = ConfigDict(extra="allow", strict=True)
+
+    def to_graph_config(self) -> dict[str, Any]:
+        return self.model_dump(mode="python", exclude_unset=True)
+
+
 class WorkflowBasedAppRunner:
     def __init__(
         self,
@@ -132,14 +194,7 @@ class WorkflowBasedAppRunner:
         """
         Init graph
         """
-        if "nodes" not in graph_config or "edges" not in graph_config:
-            raise ValueError("nodes or edges not found in workflow graph")
-
-        if not isinstance(graph_config.get("nodes"), list):
-            raise ValueError("nodes in workflow graph must be a list")
-
-        if not isinstance(graph_config.get("edges"), list):
-            raise ValueError("edges in workflow graph must be a list")
+        graph_config = _WorkflowGraphConfig.model_validate(graph_config).to_graph_config()
 
         # Create explicit graph init context for Graph.init.
         run_context = build_dify_run_context(
@@ -259,6 +314,8 @@ class WorkflowBasedAppRunner:
     ) -> tuple[Graph, VariablePool]:
         """
         Get graph and variable pool for single node execution (iteration or loop).
+        The workflow graph and user inputs are treated as caller-owned data; the
+        narrowed graph config used for execution is built locally.
 
         Args:
             workflow: The workflow instance
@@ -272,47 +329,39 @@ class WorkflowBasedAppRunner:
             A tuple containing (graph, variable_pool)
         """
         # fetch workflow graph
-        graph_config = workflow.graph_dict
-        if not graph_config:
-            raise ValueError("workflow graph not found")
+        source_graph_config = _WorkflowGraphConfig.model_validate(workflow.graph_dict)
 
-        graph_config = cast(dict[str, Any], graph_config)
-
-        if "nodes" not in graph_config or "edges" not in graph_config:
-            raise ValueError("nodes or edges not found in workflow graph")
-
-        if not isinstance(graph_config.get("nodes"), list):
-            raise ValueError("nodes in workflow graph must be a list")
-
-        if not isinstance(graph_config.get("edges"), list):
-            raise ValueError("edges in workflow graph must be a list")
+        node_user_inputs = dict(user_inputs)
 
         # filter nodes only in the specified node type (iteration or loop)
-        main_node_config = next((n for n in graph_config.get("nodes", []) if n.get("id") == node_id), None)
-        start_node_id = main_node_config.get("data", {}).get("start_node_id") if main_node_config else None
-        node_configs = [
+        main_node_config = next((node for node in source_graph_config.nodes if node.id == node_id), None)
+        start_node_id = main_node_config.data.start_node_id if main_node_config else None
+        selected_node_configs = [
             node
-            for node in graph_config.get("nodes", [])
-            if node.get("id") == node_id
-            or node.get("data", {}).get(node_type_filter_key, "") == node_id
-            or (start_node_id and node.get("id") == start_node_id)
+            for node in source_graph_config.nodes
+            if node.id == node_id
+            or node.data.parent_id_for(node_type_filter_key) == node_id
+            or (start_node_id and node.id == start_node_id)
         ]
 
-        graph_config["nodes"] = node_configs
-
-        node_ids = [node.get("id") for node in node_configs]
+        node_ids = [node.id for node in selected_node_configs]
 
         # filter edges only in the specified node type
-        edge_configs = [
+        selected_edge_configs = [
             edge
-            for edge in graph_config.get("edges", [])
-            if (edge.get("source") is None or edge.get("source") in node_ids)
-            and (edge.get("target") is None or edge.get("target") in node_ids)
+            for edge in source_graph_config.edges
+            if (edge.source is None or edge.source in node_ids) and (edge.target is None or edge.target in node_ids)
         ]
 
-        graph_config["edges"] = edge_configs
+        node_configs = [node.to_graph_node_config() for node in selected_node_configs]
+        edge_configs = [edge.model_dump(mode="python", exclude_unset=True) for edge in selected_edge_configs]
+        graph_config = {
+            **source_graph_config.to_graph_config(),
+            "nodes": node_configs,
+            "edges": edge_configs,
+        }
 
-        typed_node_configs = [NodeConfigDictAdapter.validate_python(node) for node in node_configs]
+        typed_node_configs = selected_node_configs
 
         # Create explicit graph init context for Graph.init.
         run_context = build_dify_run_context(
@@ -338,7 +387,7 @@ class WorkflowBasedAppRunner:
 
         target_node_config = None
         for node in typed_node_configs:
-            if node["id"] == node_id:
+            if node.id == node_id:
                 target_node_config = node
                 break
 
@@ -346,8 +395,8 @@ class WorkflowBasedAppRunner:
             raise ValueError(f"{node_type_label} node id not found in workflow graph")
 
         # Get node class
-        node_type = target_node_config["data"].type
-        node_version = str(target_node_config["data"].version)
+        node_type = target_node_config.data.type
+        node_version = str(target_node_config.data.version)
         node_cls = resolve_workflow_node_class(node_type=node_type, node_version=node_version)
 
         # Use the variable pool from graph_runtime_state instead of creating a new one
@@ -360,22 +409,22 @@ class WorkflowBasedAppRunner:
                 selector
                 for node_config in typed_node_configs
                 for selector in get_node_creation_preload_selectors(
-                    node_type=node_config["data"].type,
-                    node_data=node_config["data"],
+                    node_type=node_config.data.type,
+                    node_data=node_config.data,
                 )
             ],
         )
 
         try:
             variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
-                graph_config=workflow.graph_dict, config=target_node_config
+                graph_config=graph_config, config=target_node_config.to_typed_node_config()
             )
         except NotImplementedError:
             variable_mapping = {}
         variable_mapping = inject_default_system_variable_mappings(
-            node_id=target_node_config["id"],
+            node_id=target_node_config.id,
             node_type=node_type,
-            node_data=target_node_config["data"],
+            node_data=target_node_config.data,
             variable_mapping=variable_mapping,
         )
 
@@ -383,12 +432,12 @@ class WorkflowBasedAppRunner:
             variable_loader=self._variable_loader,
             variable_pool=variable_pool,
             variable_mapping=variable_mapping,
-            user_inputs=user_inputs,
+            user_inputs=node_user_inputs,
         )
 
         WorkflowEntry.mapping_user_inputs_to_variable_pool(
             variable_mapping=variable_mapping,
-            user_inputs=user_inputs,
+            user_inputs=node_user_inputs,
             variable_pool=variable_pool,
             tenant_id=workflow.tenant_id,
         )

api/tests/unit_tests/core/app/apps/test_workflow_app_runner_core.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import json
+from copy import deepcopy
 from datetime import UTC, datetime
 from types import SimpleNamespace
 
 import pytest
+from pydantic import ValidationError
 
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, InvokeFrom, UserFrom
@@ -78,7 +80,7 @@ class TestWorkflowBasedAppRunner:
             start_at=0.0,
         )
 
-        with pytest.raises(ValueError, match="nodes or edges not found"):
+        with pytest.raises(ValidationError, match="nodes"):
             runner._init_graph(
                 graph_config={},
                 graph_runtime_state=runtime_state,
@@ -86,7 +88,7 @@ class TestWorkflowBasedAppRunner:
                 invoke_from=InvokeFrom.DEBUGGER,
             )
 
-        with pytest.raises(ValueError, match="nodes in workflow graph must be a list"):
+        with pytest.raises(ValidationError, match="nodes"):
             runner._init_graph(
                 graph_config={"nodes": {}, "edges": []},
                 graph_runtime_state=runtime_state,
@@ -94,7 +96,7 @@ class TestWorkflowBasedAppRunner:
                 invoke_from=InvokeFrom.DEBUGGER,
             )
 
-        with pytest.raises(ValueError, match="edges in workflow graph must be a list"):
+        with pytest.raises(ValidationError, match="edges"):
             runner._init_graph(
                 graph_config={"nodes": [], "edges": {}},
                 graph_runtime_state=runtime_state,
@@ -192,6 +194,71 @@ class TestWorkflowBasedAppRunner:
         assert graph is not None
         assert variable_pool is graph_runtime_state.variable_pool
 
+    def test_get_graph_and_variable_pool_for_single_node_run_does_not_mutate_inputs(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        runner = WorkflowBasedAppRunner(queue_manager=SimpleNamespace(), app_id="app")
+        graph_runtime_state = GraphRuntimeState(
+            variable_pool=VariablePool.from_bootstrap(system_variables=default_system_variables()),
+            start_at=0.0,
+        )
+        graph_config = {
+            "nodes": [
+                {"id": "node-1", "data": {"type": "start", "version": "1"}},
+                {"id": "node-child", "data": {"type": "start", "version": "1", "iteration_id": "node-1"}},
+                {"id": "outside", "data": {"type": "start", "version": "1"}},
+            ],
+            "edges": [
+                {"id": "edge-1", "source": "node-1", "target": "node-child"},
+                {"id": "edge-2", "source": "outside", "target": "node-1"},
+            ],
+        }
+        expected_graph_config = deepcopy(graph_config)
+        workflow = SimpleNamespace(tenant_id="tenant", id="workflow", graph_dict=graph_config)
+        user_inputs = {"question": "hello"}
+        expected_user_inputs = dict(user_inputs)
+        captured_graph_config = {}
+
+        def fake_graph_init(**kwargs):
+            captured_graph_config["value"] = kwargs["graph_config"]
+            return SimpleNamespace()
+
+        class _NodeCls:
+            @staticmethod
+            def extract_variable_selector_to_variable_mapping(graph_config, config):
+                return {}
+
+        from core.app.apps import workflow_app_runner
+
+        monkeypatch.setattr("core.app.apps.workflow_app_runner.Graph.init", fake_graph_init)
+        monkeypatch.setattr(workflow_app_runner, "resolve_workflow_node_class", lambda **_kwargs: _NodeCls)
+        monkeypatch.setattr(
+            "core.app.apps.workflow_app_runner.load_into_variable_pool",
+            lambda **kwargs: kwargs["user_inputs"].update({"from_loader": "local-only"}),
+        )
+        monkeypatch.setattr(
+            "core.app.apps.workflow_app_runner.WorkflowEntry.mapping_user_inputs_to_variable_pool",
+            lambda **kwargs: kwargs["user_inputs"].update({"from_mapping": "local-only"}),
+        )
+
+        graph, variable_pool = runner._get_graph_and_variable_pool_for_single_node_run(
+            workflow=workflow,
+            node_id="node-1",
+            user_inputs=user_inputs,
+            graph_runtime_state=graph_runtime_state,
+            node_type_filter_key="iteration_id",
+            node_type_label="iteration",
+            user_id="00000000-0000-0000-0000-000000000001",
+        )
+
+        assert graph is not None
+        assert variable_pool is graph_runtime_state.variable_pool
+        assert workflow.graph_dict == expected_graph_config
+        assert user_inputs == expected_user_inputs
+        assert captured_graph_config["value"] is not workflow.graph_dict
+        assert [node["id"] for node in captured_graph_config["value"]["nodes"]] == ["node-1", "node-child"]
+        assert captured_graph_config["value"]["edges"] == [{"id": "edge-1", "source": "node-1", "target": "node-child"}]
+
     def test_get_graph_and_variable_pool_for_single_node_run_includes_trace_session_id(
         self, monkeypatch: pytest.MonkeyPatch
     ):
@@ -294,18 +361,11 @@ class TestWorkflowBasedAppRunner:
             def extract_variable_selector_to_variable_mapping(graph_config, config):
                 return {}
 
-        def _validate_node_config(value):
-            return {"id": value["id"], "data": SimpleNamespace(**value["data"])}
-
         def _graph_init(**kwargs):
             variable_pool = graph_runtime_state.variable_pool
             assert variable_pool.get(["sys", "conversation_id"]) is not None
             return SimpleNamespace()
 
-        monkeypatch.setattr(
-            "core.app.apps.workflow_app_runner.NodeConfigDictAdapter.validate_python",
-            _validate_node_config,
-        )
         monkeypatch.setattr(
             "core.app.apps.workflow_app_runner.Graph.init",
             _graph_init,

api/tests/unit_tests/core/app/apps/test_workflow_app_runner_single_node.py

@@ -5,6 +5,7 @@ from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
+from pydantic import ValidationError
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_runner import WorkflowAppRunner
@@ -12,7 +13,6 @@ from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.credit_usage import CreditUsageAppType
 from core.workflow.system_variables import default_system_variables
-from graphon.entities.graph_config import NodeConfigDictAdapter
 from graphon.runtime import GraphRuntimeState, VariablePool
 from models.model import AppMode
 from models.workflow import Workflow, WorkflowKind
@@ -118,7 +118,7 @@ def test_run_uses_single_node_execution_branch(
     assert entry_kwargs["graph_runtime_state"] is graph_runtime_state
 
 
-def test_single_node_run_validates_target_node_config(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_single_node_run_validates_target_node_config() -> None:
     runner = WorkflowBasedAppRunner(
         queue_manager=MagicMock(spec=AppQueueManager),
         variable_loader=MagicMock(),
@@ -133,14 +133,7 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey
                 "nodes": [
                     {
                         "id": "loop-node",
-                        "data": {
-                            "type": "loop",
-                            "title": "Loop",
-                            "loop_count": 1,
-                            "start_node_id": "loop-start",
-                            "break_conditions": [],
-                            "logical_operator": "and",
-                        },
+                        "data": {},
                     }
                 ],
                 "edges": [],
@@ -149,21 +142,8 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey
     )
 
     _, _, graph_runtime_state = _make_graph_state()
-    seen_configs: list[object] = []
-    original_validate_python = NodeConfigDictAdapter.validate_python
 
-    def record_validate_python(value: object):
-        seen_configs.append(value)
-        return original_validate_python(value)
-
-    monkeypatch.setattr(NodeConfigDictAdapter, "validate_python", record_validate_python)
-
-    with (
-        patch("core.app.apps.workflow_app_runner.DifyNodeFactory"),
-        patch("core.app.apps.workflow_app_runner.Graph.init", return_value=MagicMock()),
-        patch("core.app.apps.workflow_app_runner.load_into_variable_pool"),
-        patch("core.app.apps.workflow_app_runner.WorkflowEntry.mapping_user_inputs_to_variable_pool"),
-    ):
+    with pytest.raises(ValidationError, match="nodes.0.data.type"):
         runner._get_graph_and_variable_pool_for_single_node_run(
             workflow=workflow,
             node_id="loop-node",
@@ -174,8 +154,6 @@ def test_single_node_run_validates_target_node_config(monkeypatch: pytest.Monkey
             user_id="00000000-0000-0000-0000-000000000001",
         )
 
-    assert seen_configs == [workflow.graph_dict["nodes"][0]]
-
 
 def test_run_adds_inputs_with_snippet_compatible_start_aliases() -> None:
     app_config = MagicMock()