From 39080eed1056814eb1c8bf9f5ae9b83578cc8f8b Mon Sep 17 00:00:00 2001
From: jyong <718720800@qq.com>
Date: Thu, 28 Aug 2025 22:01:49 +0800
Subject: [PATCH] fix preview run

---
 api/core/workflow/nodes/datasource/datasource_node.py | 3 ++-
 api/services/rag_pipeline/rag_pipeline.py             | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/api/core/workflow/nodes/datasource/datasource_node.py b/api/core/workflow/nodes/datasource/datasource_node.py
index d9c2750ec5..7f603dbff8 100644
--- a/api/core/workflow/nodes/datasource/datasource_node.py
+++ b/api/core/workflow/nodes/datasource/datasource_node.py
@@ -21,7 +21,7 @@ from core.variables.segments import ArrayAnySegment
 from core.variables.variables import ArrayAnyVariable
 from core.workflow.entities.variable_pool import VariablePool, VariableValue
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
-from core.workflow.enums import ErrorStrategy, NodeType, SystemVariableKey
+from core.workflow.enums import ErrorStrategy, NodeType, SystemVariableKey, NodeExecutionType
 from core.workflow.node_events import NodeRunResult, StreamChunkEvent, StreamCompletedEvent
 from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig
 from core.workflow.nodes.base.node import Node
@@ -44,6 +44,7 @@ class DatasourceNode(Node):
 
     _node_data: DatasourceNodeData
     node_type = NodeType.DATASOURCE
+    execution_type = NodeExecutionType.ROOT
 
     def init_node_data(self, data: Mapping[str, Any]) -> None:
         self._node_data = DatasourceNodeData.model_validate(data)
diff --git a/api/services/rag_pipeline/rag_pipeline.py b/api/services/rag_pipeline/rag_pipeline.py
index 6ae9d68e5f..ce3954b315 100644
--- a/api/services/rag_pipeline/rag_pipeline.py
+++ b/api/services/rag_pipeline/rag_pipeline.py
@@ -41,6 +41,7 @@ from core.workflow.entities.workflow_node_execution import (
 )
 from core.workflow.enums import ErrorStrategy, NodeType, SystemVariableKey
 from core.workflow.errors import WorkflowNodeRunFailedError
+from core.workflow.graph_events import NodeRunSucceededEvent, NodeRunFailedEvent
 from core.workflow.graph_events.base import GraphNodeEventBase
 from core.workflow.node_events.base import NodeRunResult
 from core.workflow.node_events.node import StreamCompletedEvent
@@ -756,7 +757,7 @@ class RagPipelineService:
 
             node_run_result: NodeRunResult | None = None
             for event in generator:
-                if isinstance(event, StreamCompletedEvent):
+                if isinstance(event, (NodeRunSucceededEvent, NodeRunFailedEvent)):
                     node_run_result = event.node_run_result
                     # sign output files
                     node_run_result.outputs = WorkflowEntry.handle_special_values(node_run_result.outputs) or {}