feat(api): enhance workflow validation and structure checks

- Added a new validation class to ensure that trigger nodes do not coexist with UserInput (start) nodes in the workflow graph.
- Implemented a method in WorkflowService to validate the graph structure before persisting workflows, leveraging the new validation logic.
- Updated unit tests to cover the new validation scenarios and ensure proper error propagation.

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

@@ -64,6 +64,15 @@ class _TestNode(Node):
         )
         self.data = dict(data)
 
+        node_type_value = data.get("type")
+        if isinstance(node_type_value, NodeType):
+            self.node_type = node_type_value
+        elif isinstance(node_type_value, str):
+            try:
+                self.node_type = NodeType(node_type_value)
+            except ValueError:
+                pass
+
     def _run(self):
         raise NotImplementedError
 
@@ -179,3 +188,22 @@ def test_graph_promotes_fail_branch_nodes_to_branch_execution_type(
     graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
 
     assert graph.nodes["branch"].execution_type == NodeExecutionType.BRANCH
+
+
+def test_graph_validation_blocks_start_and_trigger_coexistence(
+    graph_init_dependencies: tuple[_SimpleNodeFactory, dict[str, object]],
+) -> None:
+    node_factory, graph_config = graph_init_dependencies
+    graph_config["nodes"] = [
+        {"id": "start", "data": {"type": NodeType.START, "title": "Start", "execution_type": NodeExecutionType.ROOT}},
+        {
+            "id": "trigger",
+            "data": {"type": NodeType.TRIGGER_WEBHOOK, "title": "Webhook", "execution_type": NodeExecutionType.ROOT},
+        },
+    ]
+    graph_config["edges"] = []
+
+    with pytest.raises(GraphValidationError) as exc_info:
+        Graph.init(graph_config=graph_config, node_factory=node_factory)
+
+    assert any(issue.code == "TRIGGER_START_NODE_CONFLICT" for issue in exc_info.value.issues)

api/tests/unit_tests/services/workflow/test_workflow_service.py

@@ -1,7 +1,8 @@
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
+from core.workflow.graph.validation import GraphValidationError, GraphValidationIssue
 from models.model import App
 from models.workflow import Workflow
 from services.workflow_service import WorkflowService
@@ -161,3 +162,25 @@ class TestWorkflowService:
         assert workflows == []
         assert has_more is False
         mock_session.scalars.assert_called_once()
+
+    def test_validate_graph_structure_invokes_graph_init(self, workflow_service, mock_app):
+        graph = {"nodes": [], "edges": []}
+
+        with patch("services.workflow_service.Graph.init") as mock_graph_init:
+            workflow_service.validate_graph_structure(mock_app, graph)
+
+        mock_graph_init.assert_called_once()
+        assert mock_graph_init.call_args.kwargs["graph_config"] is graph
+        assert "node_factory" in mock_graph_init.call_args.kwargs
+
+    def test_validate_graph_structure_propagates_graph_errors(self, workflow_service, mock_app):
+        graph = {"nodes": [], "edges": []}
+        issue = GraphValidationIssue(code="ERR", message="invalid")
+
+        with patch("services.workflow_service.Graph.init", side_effect=GraphValidationError([issue])):
+            with pytest.raises(GraphValidationError):
+                workflow_service.validate_graph_structure(mock_app, graph)
+
+    def test_validate_graph_structure_requires_nodes_and_edges(self, workflow_service, mock_app):
+        with pytest.raises(ValueError, match="must include 'nodes' and 'edges'"):
+            workflow_service.validate_graph_structure(mock_app, {"nodes": []})