Merge branch 'refs/heads/feat/workflow-parallel-support' into deploy/dev

api/core/workflow/graph_engine/entities/graph.py

@@ -174,6 +174,7 @@ def init(cls,
         node_parallel_mapping: dict[str, str] = {}
         cls._recursively_add_parallels(
             edge_mapping=edge_mapping,
+            reverse_edge_mapping=reverse_edge_mapping,
             start_node_id=root_node_id,
             parallel_mapping=parallel_mapping,
             node_parallel_mapping=node_parallel_mapping
@@ -310,6 +311,7 @@ def _check_connected_to_previous_node(
     @classmethod
     def _recursively_add_parallels(cls,
                                    edge_mapping: dict[str, list[GraphEdge]],
+                                   reverse_edge_mapping: dict[str, list[GraphEdge]],
                                    start_node_id: str,
                                    parallel_mapping: dict[str, GraphParallel],
                                    node_parallel_mapping: dict[str, str]) -> None:
@@ -365,6 +367,7 @@ def _recursively_add_parallels(cls,
 
                 in_branch_node_ids = cls._fetch_all_node_ids_in_parallels(
                     edge_mapping=edge_mapping,
+                    reverse_edge_mapping=reverse_edge_mapping,
                     parallel_branch_node_ids=parallel_branch_node_ids
                 )
 
@@ -412,6 +415,7 @@ def _recursively_add_parallels(cls,
         for graph_edge in target_node_edges:
             cls._recursively_add_parallels(
                 edge_mapping=edge_mapping,
+                reverse_edge_mapping=reverse_edge_mapping,
                 start_node_id=graph_edge.target_node_id,
                 parallel_mapping=parallel_mapping,
                 node_parallel_mapping=node_parallel_mapping
@@ -472,6 +476,7 @@ def _recursively_add_parallel_node_ids(cls,
     @classmethod
     def _fetch_all_node_ids_in_parallels(cls,
                                          edge_mapping: dict[str, list[GraphEdge]],
+                                         reverse_edge_mapping: dict[str, list[GraphEdge]],
                                          parallel_branch_node_ids: list[str]) -> dict[str, list[str]]:
         """
         Fetch all node ids in parallels
@@ -499,7 +504,11 @@ def _fetch_all_node_ids_in_parallels(cls,
                     leaf_node_ids[branch_node_id].append(node_id)
 
                 for branch_node_id2, inner_route2 in routes_node_ids.items():
-                    if branch_node_id != branch_node_id2 and node_id in inner_route2:
+                    if (
+                        branch_node_id != branch_node_id2 
+                        and node_id in inner_route2
+                        and len(reverse_edge_mapping.get(node_id, [])) > 1
+                    ):
                         if node_id not in merge_branch_node_ids:
                             merge_branch_node_ids[node_id] = []
 

web/app/components/workflow/hooks/use-workflow-run.ts

@@ -332,6 +332,8 @@ export const useWorkflowRun = () => {
               const currentIndex = draft.tracing!.findIndex((trace) => {
                 if (!trace.execution_metadata?.parallel_id)
                   return trace.node_id === data.node_id
+                if (trace.execution_metadata?.parallel_id === data.execution_metadata?.parallel_id)
+                  return trace.node_id === data.node_id && trace.execution_metadata?.parallel_start_node_id === data.execution_metadata?.parallel_start_node_id
 
                 return trace.node_id === data.node_id && trace.execution_metadata?.parallel_id === data.execution_metadata?.parallel_id
               })

web/app/components/workflow/panel/debug-and-preview/hooks.ts

@@ -385,6 +385,8 @@ export const useChat = (
             const currentIndex = responseItem.workflowProcess!.tracing!.findIndex((item) => {
               if (!item.execution_metadata?.parallel_id)
                 return item.node_id === data.node_id
+              if (item.execution_metadata?.parallel_id === data.execution_metadata?.parallel_id)
+                return item.node_id === data.node_id && item.execution_metadata?.parallel_start_node_id === data.execution_metadata?.parallel_start_node_id
 
               return item.node_id === data.node_id && item.execution_metadata?.parallel_id === data.execution_metadata?.parallel_id
             })

web/app/components/workflow/run/node.tsx

@@ -76,7 +76,7 @@ const NodePanel: FC<Props> = ({
       <div className='group transition-all bg-background-default border border-components-panel-border rounded-[10px] shadows-shadow-xs hover:shadow-md'>
         <div
           className={cn(
-            'flex items-center pl-1 pr-2 cursor-pointer',
+            'flex items-center pl-1 pr-3 cursor-pointer',
             hideInfo ? 'py-2' : 'py-1.5',
             !collapseState && (hideInfo ? '!pb-1' : '!pb-1.5'),
           )}

web/app/components/workflow/run/tracing-panel.tsx

@@ -37,7 +37,7 @@ function buildLogTree(nodes: NodeTracing[]): TracingNodeProps[] {
   const levelCounts: { [key: string]: number } = {}
   const parallelChildCounts: { [key: string]: Set<string> } = {}
 
-  function getParallelTitle(parentId: string | null): string {
+  const getParallelTitle = (parentId: string | null): string => {
     const levelKey = parentId || 'root'
     if (!levelCounts[levelKey])
       levelCounts[levelKey] = 0
@@ -50,7 +50,7 @@ function buildLogTree(nodes: NodeTracing[]): TracingNodeProps[] {
     return `PARALLEL-${levelNumber}${letter}`
   }
 
-  function getBranchTitle(parentId: string | null, branchNum: number): string {
+  const getBranchTitle = (parentId: string | null, branchNum: number): string => {
     const levelKey = parentId || 'root'
     const parentTitle = parentId ? parallelStacks[parentId].parallelTitle : ''
     const levelNumber = parentTitle ? parseInt(parentTitle.split('-')[1]) + 1 : 1
@@ -153,6 +153,7 @@ const TracingPanel: FC<TracingPanelProps> = ({ list, onShowIterationDetail }) =>
       const newSet = new Set(prev)
       if (newSet.has(id))
         newSet.delete(id)
+
       else
         newSet.add(id)
 
@@ -165,12 +166,18 @@ const TracingPanel: FC<TracingPanelProps> = ({ list, onShowIterationDetail }) =>
   }, [])
 
   const handleParallelMouseLeave = useCallback((e: React.MouseEvent) => {
-    const relatedTarget = e.relatedTarget as HTMLElement
-    const closestParallel = relatedTarget?.closest('[data-parallel-id]')
-    if (closestParallel)
-      setHoveredParallel(closestParallel.getAttribute('data-parallel-id'))
-    else
+    const relatedTarget = e.relatedTarget as Element | null
+    if (relatedTarget && 'closest' in relatedTarget) {
+      const closestParallel = relatedTarget.closest('[data-parallel-id]')
+      if (closestParallel)
+        setHoveredParallel(closestParallel.getAttribute('data-parallel-id'))
+
+      else
+        setHoveredParallel(null)
+    }
+    else {
       setHoveredParallel(null)
+    }
   }, [])
 
   const renderNode = (node: TracingNodeProps) => {