fix(graph_engine): Cannot run single iteration or loop node (#31470)

Signed-off-by: -LAN- <laipz8200@outlook.com>
Co-authored-by: Yeuoly <45712896+Yeuoly@users.noreply.github.com>

Author: laipz8200

api/core/app/apps/workflow_app_runner.py

@@ -157,7 +157,7 @@ def _prepare_single_node_execution(
         # Create initial runtime state with variable pool containing environment variables
         graph_runtime_state = GraphRuntimeState(
             variable_pool=VariablePool(
-                system_variables=SystemVariable.empty(),
+                system_variables=SystemVariable.default(),
                 user_inputs={},
                 environment_variables=workflow.environment_variables,
             ),
@@ -272,7 +272,9 @@ def _get_graph_and_variable_pool_for_single_node_run(
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=node_id)
+        graph = Graph.init(
+            graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, skip_validation=True
+        )
 
         if not graph:
             raise ValueError("graph not found in workflow")

api/core/workflow/graph/graph.py

@@ -288,6 +288,7 @@ def init(
         graph_config: Mapping[str, object],
         node_factory: NodeFactory,
         root_node_id: str | None = None,
+        skip_validation: bool = False,
     ) -> Graph:
         """
         Initialize graph
@@ -339,8 +340,9 @@ def init(
             root_node=root_node,
         )
 
-        # Validate the graph structure using built-in validators
-        get_graph_validator().validate(graph)
+        if not skip_validation:
+            # Validate the graph structure using built-in validators
+            get_graph_validator().validate(graph)
 
         return graph
 

api/core/workflow/runtime/variable_pool.py

@@ -44,7 +44,7 @@ class VariablePool(BaseModel):
     )
     system_variables: SystemVariable = Field(
         description="System variables",
-        default_factory=SystemVariable.empty,
+        default_factory=SystemVariable.default,
     )
     environment_variables: Sequence[Variable] = Field(
         description="Environment variables.",
@@ -271,4 +271,4 @@ def _add_system_variables(self, system_variable: SystemVariable):
     @classmethod
     def empty(cls) -> VariablePool:
         """Create an empty variable pool."""
-        return cls(system_variables=SystemVariable.empty())
+        return cls(system_variables=SystemVariable.default())

api/core/workflow/system_variable.py

@@ -3,6 +3,7 @@
 from collections.abc import Mapping, Sequence
 from types import MappingProxyType
 from typing import Any
+from uuid import uuid4
 
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
 
@@ -72,8 +73,8 @@ def validate_json_fields(cls, data):
         return data
 
     @classmethod
-    def empty(cls) -> SystemVariable:
-        return cls()
+    def default(cls) -> SystemVariable:
+        return cls(workflow_execution_id=str(uuid4()))
 
     def to_dict(self) -> dict[SystemVariableKey, Any]:
         # NOTE: This method is provided for compatibility with legacy code.

api/core/workflow/workflow_entry.py

@@ -277,7 +277,7 @@ def run_free_node(
 
         # init variable pool
         variable_pool = VariablePool(
-            system_variables=SystemVariable.empty(),
+            system_variables=SystemVariable.default(),
             user_inputs={},
             environment_variables=[],
         )

api/services/rag_pipeline/rag_pipeline.py

@@ -436,7 +436,7 @@ def run_draft_workflow_node(
                 user_inputs=user_inputs,
                 user_id=account.id,
                 variable_pool=VariablePool(
-                    system_variables=SystemVariable.empty(),
+                    system_variables=SystemVariable.default(),
                     user_inputs=user_inputs,
                     environment_variables=[],
                     conversation_variables=[],

api/services/workflow_service.py

@@ -675,7 +675,7 @@ def run_draft_workflow_node(
 
         else:
             variable_pool = VariablePool(
-                system_variables=SystemVariable.empty(),
+                system_variables=SystemVariable.default(),
                 user_inputs=user_inputs,
                 environment_variables=draft_workflow.environment_variables,
                 conversation_variables=[],
@@ -1063,7 +1063,7 @@ def _setup_variable_pool(
             system_variable.conversation_id = conversation_id
             system_variable.dialogue_count = 1
     else:
-        system_variable = SystemVariable.empty()
+        system_variable = SystemVariable.default()
 
     # init variable pool
     variable_pool = VariablePool(

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

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.workflow.app_runner import WorkflowAppRunner
+from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.workflow.runtime import GraphRuntimeState, VariablePool
+from core.workflow.system_variable import SystemVariable
+from models.workflow import Workflow
+
+
+def _make_graph_state():
+    variable_pool = VariablePool(
+        system_variables=SystemVariable.default(),
+        user_inputs={},
+        environment_variables=[],
+        conversation_variables=[],
+    )
+    return MagicMock(), variable_pool, GraphRuntimeState(variable_pool=variable_pool, start_at=0.0)
+
+
+@pytest.mark.parametrize(
+    ("single_iteration_run", "single_loop_run"),
+    [
+        (WorkflowAppGenerateEntity.SingleIterationRunEntity(node_id="iter", inputs={}), None),
+        (None, WorkflowAppGenerateEntity.SingleLoopRunEntity(node_id="loop", inputs={})),
+    ],
+)
+def test_run_uses_single_node_execution_branch(
+    single_iteration_run: Any,
+    single_loop_run: Any,
+) -> None:
+    app_config = MagicMock()
+    app_config.app_id = "app"
+    app_config.tenant_id = "tenant"
+    app_config.workflow_id = "workflow"
+
+    app_generate_entity = MagicMock(spec=WorkflowAppGenerateEntity)
+    app_generate_entity.app_config = app_config
+    app_generate_entity.inputs = {}
+    app_generate_entity.files = []
+    app_generate_entity.user_id = "user"
+    app_generate_entity.invoke_from = InvokeFrom.SERVICE_API
+    app_generate_entity.workflow_execution_id = "execution-id"
+    app_generate_entity.task_id = "task-id"
+    app_generate_entity.call_depth = 0
+    app_generate_entity.trace_manager = None
+    app_generate_entity.single_iteration_run = single_iteration_run
+    app_generate_entity.single_loop_run = single_loop_run
+
+    workflow = MagicMock(spec=Workflow)
+    workflow.tenant_id = "tenant"
+    workflow.app_id = "app"
+    workflow.id = "workflow"
+    workflow.type = "workflow"
+    workflow.version = "v1"
+    workflow.graph_dict = {"nodes": [], "edges": []}
+    workflow.environment_variables = []
+
+    runner = WorkflowAppRunner(
+        application_generate_entity=app_generate_entity,
+        queue_manager=MagicMock(spec=AppQueueManager),
+        variable_loader=MagicMock(),
+        workflow=workflow,
+        system_user_id="system-user",
+        workflow_execution_repository=MagicMock(),
+        workflow_node_execution_repository=MagicMock(),
+    )
+
+    graph, variable_pool, graph_runtime_state = _make_graph_state()
+    mock_workflow_entry = MagicMock()
+    mock_workflow_entry.graph_engine = MagicMock()
+    mock_workflow_entry.graph_engine.layer = MagicMock()
+    mock_workflow_entry.run.return_value = iter([])
+
+    with (
+        patch("core.app.apps.workflow.app_runner.RedisChannel"),
+        patch("core.app.apps.workflow.app_runner.redis_client"),
+        patch("core.app.apps.workflow.app_runner.WorkflowEntry", return_value=mock_workflow_entry) as entry_class,
+        patch.object(
+            runner,
+            "_prepare_single_node_execution",
+            return_value=(
+                graph,
+                variable_pool,
+                graph_runtime_state,
+            ),
+        ) as prepare_single,
+        patch.object(runner, "_init_graph") as init_graph,
+    ):
+        runner.run()
+
+    prepare_single.assert_called_once_with(
+        workflow=workflow,
+        single_iteration_run=single_iteration_run,
+        single_loop_run=single_loop_run,
+    )
+    init_graph.assert_not_called()
+
+    entry_kwargs = entry_class.call_args.kwargs
+    assert entry_kwargs["invoke_from"] == InvokeFrom.DEBUGGER
+    assert entry_kwargs["variable_pool"] is variable_pool
+    assert entry_kwargs["graph_runtime_state"] is graph_runtime_state

api/tests/unit_tests/core/workflow/graph/test_graph_skip_validation.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.workflow.node_factory import DifyNodeFactory
+from core.workflow.entities import GraphInitParams
+from core.workflow.graph import Graph
+from core.workflow.graph.validation import GraphValidationError
+from core.workflow.nodes import NodeType
+from core.workflow.runtime import GraphRuntimeState, VariablePool
+from core.workflow.system_variable import SystemVariable
+from models.enums import UserFrom
+
+
+def _build_iteration_graph(node_id: str) -> dict[str, Any]:
+    return {
+        "nodes": [
+            {
+                "id": node_id,
+                "data": {
+                    "type": "iteration",
+                    "title": "Iteration",
+                    "iterator_selector": ["start", "items"],
+                    "output_selector": [node_id, "output"],
+                },
+            }
+        ],
+        "edges": [],
+    }
+
+
+def _build_loop_graph(node_id: str) -> dict[str, Any]:
+    return {
+        "nodes": [
+            {
+                "id": node_id,
+                "data": {
+                    "type": "loop",
+                    "title": "Loop",
+                    "loop_count": 1,
+                    "break_conditions": [],
+                    "logical_operator": "and",
+                    "loop_variables": [],
+                    "outputs": {},
+                },
+            }
+        ],
+        "edges": [],
+    }
+
+
+def _make_factory(graph_config: dict[str, Any]) -> DifyNodeFactory:
+    graph_init_params = GraphInitParams(
+        tenant_id="tenant",
+        app_id="app",
+        workflow_id="workflow",
+        graph_config=graph_config,
+        user_id="user",
+        user_from=UserFrom.ACCOUNT,
+        invoke_from=InvokeFrom.DEBUGGER,
+        call_depth=0,
+    )
+    graph_runtime_state = GraphRuntimeState(
+        variable_pool=VariablePool(
+            system_variables=SystemVariable.default(),
+            user_inputs={},
+            environment_variables=[],
+        ),
+        start_at=0.0,
+    )
+    return DifyNodeFactory(graph_init_params=graph_init_params, graph_runtime_state=graph_runtime_state)
+
+
+def test_iteration_root_requires_skip_validation():
+    node_id = "iteration-node"
+    graph_config = _build_iteration_graph(node_id)
+    node_factory = _make_factory(graph_config)
+
+    with pytest.raises(GraphValidationError):
+        Graph.init(
+            graph_config=graph_config,
+            node_factory=node_factory,
+            root_node_id=node_id,
+        )
+
+    graph = Graph.init(
+        graph_config=graph_config,
+        node_factory=node_factory,
+        root_node_id=node_id,
+        skip_validation=True,
+    )
+
+    assert graph.root_node.id == node_id
+    assert graph.root_node.node_type == NodeType.ITERATION
+
+
+def test_loop_root_requires_skip_validation():
+    node_id = "loop-node"
+    graph_config = _build_loop_graph(node_id)
+    node_factory = _make_factory(graph_config)
+
+    with pytest.raises(GraphValidationError):
+        Graph.init(
+            graph_config=graph_config,
+            node_factory=node_factory,
+            root_node_id=node_id,
+        )
+
+    graph = Graph.init(
+        graph_config=graph_config,
+        node_factory=node_factory,
+        root_node_id=node_id,
+        skip_validation=True,
+    )
+
+    assert graph.root_node.id == node_id
+    assert graph.root_node.node_type == NodeType.LOOP