Merge remote-tracking branch 'origin/main' into feat/support-agent-sandbox

# Conflicts:
#	api/core/workflow/graph_events/__init__.py

Author: lyzno1

api/controllers/console/datasets/datasets.py

@@ -286,7 +286,7 @@ class DatasetListApi(Resource):
     @enterprise_license_required
     def get(self):
         current_user, current_tenant_id = current_account_with_tenant()
-        query = ConsoleDatasetListQuery.model_validate(request.args.to_dict(flat=False))
+        query = ConsoleDatasetListQuery.model_validate(request.args.to_dict())
         # provider = request.args.get("provider", default="vendor")
         if query.ids:
             datasets, total = DatasetService.get_datasets_by_ids(query.ids, current_tenant_id)

api/controllers/service_api/dataset/dataset.py

@@ -122,7 +122,7 @@ class DatasetListApi(DatasetApiResource):
     )
     def get(self, tenant_id):
         """Resource for getting datasets."""
-        query = DatasetListQuery.model_validate(request.args.to_dict(flat=False))
+        query = DatasetListQuery.model_validate(request.args.to_dict())
         # provider = request.args.get("provider", default="vendor")
 
         datasets, total = DatasetService.get_datasets(

api/core/workflow/graph_engine/layers/base.py

@@ -8,7 +8,7 @@
 from abc import ABC, abstractmethod
 
 from core.workflow.graph_engine.protocols.command_channel import CommandChannel
-from core.workflow.graph_events import GraphEngineEvent
+from core.workflow.graph_events import GraphEngineEvent, GraphNodeEventBase
 from core.workflow.nodes.base.node import Node
 from core.workflow.runtime import ReadOnlyGraphRuntimeState
 
@@ -98,7 +98,7 @@ def on_graph_end(self, error: Exception | None) -> None:
         """
         pass
 
-    def on_node_run_start(self, node: Node) -> None:  # noqa: B027
+    def on_node_run_start(self, node: Node) -> None:
         """
         Called immediately before a node begins execution.
 
@@ -109,9 +109,11 @@ def on_node_run_start(self, node: Node) -> None:  # noqa: B027
         Args:
             node: The node instance about to be executed
         """
-        pass
+        return
 
-    def on_node_run_end(self, node: Node, error: Exception | None) -> None:  # noqa: B027
+    def on_node_run_end(
+        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
+    ) -> None:
         """
         Called after a node finishes execution.
 
@@ -121,5 +123,6 @@ def on_node_run_end(self, node: Node, error: Exception | None) -> None:  # noqa:
         Args:
             node: The node instance that just finished execution
             error: Exception instance if the node failed, otherwise None
+            result_event: The final result event from node execution (succeeded/failed/paused), if any
         """
-        pass
+        return

api/core/workflow/graph_engine/layers/node_parsers.py

@@ -18,12 +18,15 @@
 from configs import dify_config
 from core.workflow.enums import NodeType
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_engine.layers.node_parsers import (
+from core.workflow.graph_events import GraphNodeEventBase
+from core.workflow.nodes.base.node import Node
+from extensions.otel.parser import (
     DefaultNodeOTelParser,
+    LLMNodeOTelParser,
     NodeOTelParser,
+    RetrievalNodeOTelParser,
     ToolNodeOTelParser,
 )
-from core.workflow.nodes.base.node import Node
 from extensions.otel.runtime import is_instrument_flag_enabled
 
 logger = logging.getLogger(__name__)
@@ -72,6 +75,8 @@ def _build_parser_registry(self) -> None:
         """Initialize parser registry for node types."""
         self._parsers = {
             NodeType.TOOL: ToolNodeOTelParser(),
+            NodeType.LLM: LLMNodeOTelParser(),
+            NodeType.KNOWLEDGE_RETRIEVAL: RetrievalNodeOTelParser(),
         }
 
     def _get_parser(self, node: Node) -> NodeOTelParser:
@@ -119,7 +124,9 @@ def on_node_run_start(self, node: Node) -> None:
             logger.warning("Failed to create OpenTelemetry span for node %s: %s", node.id, e)
 
     @override
-    def on_node_run_end(self, node: Node, error: Exception | None) -> None:
+    def on_node_run_end(
+        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
+    ) -> None:
         """
         Called when a node finishes execution.
 
@@ -139,7 +146,7 @@ def on_node_run_end(self, node: Node, error: Exception | None) -> None:
             span = node_context.span
             parser = self._get_parser(node)
             try:
-                parser.parse(node=node, span=span, error=error)
+                parser.parse(node=node, span=span, error=error, result_event=result_event)
                 span.end()
             finally:
                 token = node_context.token

api/core/workflow/graph_engine/layers/observability.py

@@ -17,7 +17,7 @@
 from core.workflow.context import IExecutionContext
 from core.workflow.graph import Graph
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events import GraphNodeEventBase, NodeRunFailedEvent
+from core.workflow.graph_events import GraphNodeEventBase, NodeRunFailedEvent, is_node_result_event
 from core.workflow.nodes.base.node import Node
 
 from .ready_queue import ReadyQueue
@@ -131,6 +131,7 @@ def _execute_node(self, node: Node) -> None:
         node.ensure_execution_id()
 
         error: Exception | None = None
+        result_event: GraphNodeEventBase | None = None
 
         # Execute the node with preserved context if execution context is provided
         if self._execution_context is not None:
@@ -140,22 +141,26 @@ def _execute_node(self, node: Node) -> None:
                     node_events = node.run()
                     for event in node_events:
                         self._event_queue.put(event)
+                        if is_node_result_event(event):
+                            result_event = event
                 except Exception as exc:
                     error = exc
                     raise
                 finally:
-                    self._invoke_node_run_end_hooks(node, error)
+                    self._invoke_node_run_end_hooks(node, error, result_event)
         else:
             self._invoke_node_run_start_hooks(node)
             try:
                 node_events = node.run()
                 for event in node_events:
                     self._event_queue.put(event)
+                    if is_node_result_event(event):
+                        result_event = event
             except Exception as exc:
                 error = exc
                 raise
             finally:
-                self._invoke_node_run_end_hooks(node, error)
+                self._invoke_node_run_end_hooks(node, error, result_event)
 
     def _invoke_node_run_start_hooks(self, node: Node) -> None:
         """Invoke on_node_run_start hooks for all layers."""
@@ -166,11 +171,13 @@ def _invoke_node_run_start_hooks(self, node: Node) -> None:
                 # Silently ignore layer errors to prevent disrupting node execution
                 continue
 
-    def _invoke_node_run_end_hooks(self, node: Node, error: Exception | None) -> None:
+    def _invoke_node_run_end_hooks(
+        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
+    ) -> None:
         """Invoke on_node_run_end hooks for all layers."""
         for layer in self._layers:
             try:
-                layer.on_node_run_end(node, error)
+                layer.on_node_run_end(node, error, result_event)
             except Exception:
                 # Silently ignore layer errors to prevent disrupting node execution
                 continue

api/core/workflow/graph_engine/worker.py

@@ -47,6 +47,7 @@
     NodeRunSucceededEvent,
     ToolCall,
     ToolResult,
+    is_node_result_event,
 )
 
 __all__ = [
@@ -79,4 +80,5 @@
     "NodeRunSucceededEvent",
     "ToolCall",
     "ToolResult",
+    "is_node_result_event",
 ]

api/core/workflow/graph_events/__init__.py

@@ -83,3 +83,26 @@ class NodeRunRetryEvent(NodeRunStartedEvent):
 
 class NodeRunPauseRequestedEvent(GraphNodeEventBase):
     reason: PauseReason = Field(..., description="pause reason")
+
+
+def is_node_result_event(event: GraphNodeEventBase) -> bool:
+    """
+    Check if an event is a final result event from node execution.
+
+    A result event indicates the completion of a node execution and contains
+    runtime information such as inputs, outputs, or error details.
+
+    Args:
+        event: The event to check
+
+    Returns:
+        True if the event is a node result event (succeeded/failed/paused), False otherwise
+    """
+    return isinstance(
+        event,
+        (
+            NodeRunSucceededEvent,
+            NodeRunFailedEvent,
+            NodeRunPauseRequestedEvent,
+        ),
+    )

api/core/workflow/graph_events/node.py

@@ -0,0 +1,20 @@
+"""
+OpenTelemetry node parsers for workflow nodes.
+
+This module provides parsers that extract node-specific metadata and set
+OpenTelemetry span attributes according to semantic conventions.
+"""
+
+from extensions.otel.parser.base import DefaultNodeOTelParser, NodeOTelParser, safe_json_dumps
+from extensions.otel.parser.llm import LLMNodeOTelParser
+from extensions.otel.parser.retrieval import RetrievalNodeOTelParser
+from extensions.otel.parser.tool import ToolNodeOTelParser
+
+__all__ = [
+    "DefaultNodeOTelParser",
+    "LLMNodeOTelParser",
+    "NodeOTelParser",
+    "RetrievalNodeOTelParser",
+    "ToolNodeOTelParser",
+    "safe_json_dumps",
+]