From 59538fd5cdc5eb21ff5521650a7dcce13e853ca9 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Thu, 2 Jul 2026 00:51:01 +0800 Subject: [PATCH] enforce workflow data edge order --- src/langbot/pkg/api/http/service/workflow.py | 93 ++++++++++++++++-- .../WorkflowEditorComponent.tsx | 24 ++++- .../home/workflows/store/useWorkflowStore.ts | 94 +++++++++++++++++++ 3 files changed, 200 insertions(+), 11 deletions(-) diff --git a/src/langbot/pkg/api/http/service/workflow.py b/src/langbot/pkg/api/http/service/workflow.py index 92e2b00b7..91436ef71 100644 --- a/src/langbot/pkg/api/http/service/workflow.py +++ b/src/langbot/pkg/api/http/service/workflow.py @@ -116,6 +116,16 @@ class WorkflowService: workflow_uuid = str(uuid.uuid4()) + definition = workflow_data.get( + 'definition', + { + 'nodes': [], + 'edges': [], + 'variables': {}, + }, + ) + self._validate_workflow_definition(definition) + # Prepare workflow data new_workflow = { 'uuid': workflow_uuid, @@ -124,14 +134,7 @@ class WorkflowService: 'emoji': workflow_data.get('emoji', '💼'), 'version': 1, 'is_enabled': workflow_data.get('is_enabled', True), - 'definition': workflow_data.get( - 'definition', - { - 'nodes': [], - 'edges': [], - 'variables': {}, - }, - ), + 'definition': definition, 'global_config': workflow_data.get('global_config', {}), 'extensions_preferences': workflow_data.get( 'extensions_preferences', @@ -190,6 +193,7 @@ class WorkflowService: # Increment version if definition changed if 'definition' in workflow_data: + self._validate_workflow_definition(workflow_data['definition']) workflow_data['version'] = current.get('version', 0) + 1 # Save version history @@ -877,6 +881,79 @@ class WorkflowService: return result + @staticmethod + def _definition_edge_type(edge: dict) -> str: + edge_type = edge.get('edge_type') + if not edge_type: + edge_type = (edge.get('data', {}) or {}).get('edgeType') + if edge_type in {'control', 'data', 'legacy'}: + return edge_type + return 'legacy' + + @staticmethod + def _definition_edge_source(edge: dict) -> str: + return edge.get('source_node', '') or edge.get('source', '') + + @staticmethod + def _definition_edge_target(edge: dict) -> str: + return edge.get('target_node', '') or edge.get('target', '') + + @classmethod + def _definition_has_control_path( + cls, + source_node: str, + target_node: str, + edges: list[dict], + ) -> bool: + if not source_node or not target_node or source_node == target_node: + return False + + adjacency: dict[str, list[str]] = {} + for edge in edges: + if cls._definition_edge_type(edge) not in {'control', 'legacy'}: + continue + source = cls._definition_edge_source(edge) + target = cls._definition_edge_target(edge) + if not source or not target: + continue + adjacency.setdefault(source, []).append(target) + + visited: set[str] = set() + stack = list(adjacency.get(source_node, [])) + while stack: + current = stack.pop() + if current in visited: + continue + if current == target_node: + return True + + visited.add(current) + stack.extend(adjacency.get(current, [])) + + return False + + def _validate_workflow_definition(self, definition: dict) -> None: + if not isinstance(definition, dict): + raise ValueError('Workflow definition must be an object') + + edges = definition.get('edges', []) + if not isinstance(edges, list): + raise ValueError('Workflow definition edges must be a list') + + for edge in edges: + if not isinstance(edge, dict): + continue + if self._definition_edge_type(edge) != 'data': + continue + + source = self._definition_edge_source(edge) + target = self._definition_edge_target(edge) + if not self._definition_has_control_path(source, target, edges): + raise ValueError( + 'Invalid data edge: data outputs can only connect from a node ' + 'that appears earlier in the control flow to a later node input' + ) + def _serialize_execution(self, execution) -> dict: data = self.ap.persistence_mgr.serialize_model( persistence_workflow.WorkflowExecution, diff --git a/web/src/app/home/workflows/components/workflow-editor/WorkflowEditorComponent.tsx b/web/src/app/home/workflows/components/workflow-editor/WorkflowEditorComponent.tsx index 0a18cb52b..54d609c79 100644 --- a/web/src/app/home/workflows/components/workflow-editor/WorkflowEditorComponent.tsx +++ b/web/src/app/home/workflows/components/workflow-editor/WorkflowEditorComponent.tsx @@ -11,7 +11,12 @@ import { SelectionMode, MarkerType, } from '@xyflow/react'; -import type { Node, NodeTypes, OnSelectionChangeParams } from '@xyflow/react'; +import type { + Connection, + Node, + NodeTypes, + OnSelectionChangeParams, +} from '@xyflow/react'; import '@xyflow/react/dist/style.css'; import { useWorkflowStore, @@ -78,6 +83,7 @@ function WorkflowEditorInner() { onNodesChange, onEdgesChange, onConnect, + isConnectionValid, addNode, selectNode, selectEdge, @@ -388,6 +394,17 @@ function WorkflowEditorInner() { return categoryColors[category] || '#6b7280'; }, []); + const handleIsValidConnection = useCallback( + (connection: Connection | WorkflowEdge) => + isConnectionValid({ + source: connection.source, + target: connection.target, + sourceHandle: connection.sourceHandle ?? null, + targetHandle: connection.targetHandle ?? null, + }), + [isConnectionValid], + ); + const displayNodes = nodes.map((node) => { const executionResult = nodeExecutionResults[node.id]; @@ -441,6 +458,7 @@ function WorkflowEditorInner() { onNodesChange={onNodesChange} onEdgesChange={onEdgesChange} onConnect={onConnect} + isValidConnection={handleIsValidConnection} onNodeClick={handleNodeClick} onEdgeClick={handleEdgeClick} onPaneClick={handlePaneClick} @@ -452,8 +470,8 @@ function WorkflowEditorInner() { snapToGrid snapGrid={[15, 15]} selectionMode={SelectionMode.Partial} - selectionOnDrag - panOnDrag={[2]} // Right click to pan (left click is for node selection) + selectionOnDrag={false} + panOnDrag selectNodesOnDrag={false} defaultEdgeOptions={{ type: 'default', diff --git a/web/src/app/home/workflows/store/useWorkflowStore.ts b/web/src/app/home/workflows/store/useWorkflowStore.ts index 1917f3e5d..913c1a684 100644 --- a/web/src/app/home/workflows/store/useWorkflowStore.ts +++ b/web/src/app/home/workflows/store/useWorkflowStore.ts @@ -128,6 +128,7 @@ interface WorkflowState { onNodesChange: (changes: NodeChange[]) => void; onEdgesChange: (changes: EdgeChange[]) => void; onConnect: (connection: Connection) => void; + isConnectionValid: (connection: Connection) => boolean; addNode: (type: string, position: { x: number; y: number }) => string; updateNodeConfig: (nodeId: string, config: Record) => void; @@ -206,6 +207,91 @@ const generateEdgeId = () => `edge_${generateUuidLikeId()}`; const getWorkflowEdgeType = (edge: WorkflowEdge): WorkflowEdgeType => normalizeWorkflowEdgeType(edge.data?.edgeType); +const isControlOrderEdge = (edge: WorkflowEdge): boolean => { + const edgeType = getWorkflowEdgeType(edge); + return edgeType === 'control' || edgeType === 'legacy'; +}; + +const hasForwardControlPath = ( + sourceNodeId: string, + targetNodeId: string, + edges: WorkflowEdge[], +): boolean => { + if (!sourceNodeId || !targetNodeId || sourceNodeId === targetNodeId) { + return false; + } + + const adjacency = new Map(); + edges.forEach((edge) => { + if (!isControlOrderEdge(edge)) { + return; + } + if (!edge.source || !edge.target) { + return; + } + + const targets = adjacency.get(edge.source) || []; + targets.push(edge.target); + adjacency.set(edge.source, targets); + }); + + const visited = new Set(); + const stack = [...(adjacency.get(sourceNodeId) || [])]; + + while (stack.length > 0) { + const current = stack.pop(); + if (!current || visited.has(current)) { + continue; + } + if (current === targetNodeId) { + return true; + } + + visited.add(current); + stack.push(...(adjacency.get(current) || [])); + } + + return false; +}; + +const isWorkflowConnectionValid = ( + connection: Connection, + edges: WorkflowEdge[], +): boolean => { + if (!connection.source || !connection.target) { + return false; + } + if (connection.source === connection.target) { + return false; + } + + const edgeType = inferWorkflowEdgeType( + connection.sourceHandle, + connection.targetHandle, + ); + if (!edgeType) { + return false; + } + + const duplicateEdge = edges.some( + (edge) => + edge.source === connection.source && + edge.target === connection.target && + (edge.sourceHandle || null) === (connection.sourceHandle || null) && + (edge.targetHandle || null) === (connection.targetHandle || null) && + getWorkflowEdgeType(edge) === edgeType, + ); + if (duplicateEdge) { + return false; + } + + if (edgeType === 'data') { + return hasForwardControlPath(connection.source, connection.target, edges); + } + + return true; +}; + const withWorkflowEdgeStyle = (edge: WorkflowEdge): WorkflowEdge => { const edgeType = getWorkflowEdgeType(edge); @@ -327,6 +413,10 @@ export const useWorkflowStore = create((set, get) => ({ }, onConnect: (connection) => { + if (!get().isConnectionValid(connection)) { + return; + } + const edgeType = inferWorkflowEdgeType( connection.sourceHandle, connection.targetHandle, @@ -355,6 +445,10 @@ export const useWorkflowStore = create((set, get) => ({ get().pushHistory(); }, + isConnectionValid: (connection) => { + return isWorkflowConnectionValid(connection, get().edges); + }, + // Add new node addNode: (type, position) => { const { nodeTypes } = get();