import logging import time from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from typing import TYPE_CHECKING, Any, cast from pydantic import BaseModel, ConfigDict, ValidationError from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom from core.app.entities.agent_strategy import AgentStrategyInfo from core.app.entities.app_invoke_entities import ( AdvancedChatAppGenerateEntity, InvokeFrom, UserFrom, WorkflowAppGenerateEntity, build_dify_run_context, ) from core.app.entities.queue_entities import ( AppQueueEvent, NodeExecutionSnapshot, QueueAgentLogEvent, QueueHumanInputFormFilledEvent, QueueHumanInputFormTimeoutEvent, QueueIterationCompletedEvent, QueueIterationNextEvent, QueueIterationStartEvent, QueueLoopCompletedEvent, QueueLoopNextEvent, QueueLoopStartEvent, QueueNodeExceptionEvent, QueueNodeFailedEvent, QueueNodeRetryEvent, QueueNodeStartedEvent, QueueNodeSucceededEvent, QueueReasoningChunkEvent, QueueRetrieverResourcesEvent, QueueStopEvent, QueueTextChunkEvent, QueueWorkflowFailedEvent, QueueWorkflowPartialSuccessEvent, QueueWorkflowPausedEvent, QueueWorkflowStartedEvent, QueueWorkflowSucceededEvent, ) from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig from core.credit_usage import CreditUsageAppType from core.rag.entities import RetrievalSourceMetadata from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository from core.repositories.human_input_repository import HumanInputFormRecord from core.workflow.nodes.agent.events import NodeRunAgentLogEvent from core.workflow.nodes.human_input.boundary import resolve_human_input_node_id from core.workflow.nodes.human_input.callback import DifyHITLCallback from core.workflow.nodes.human_input.enums import HumanInputFormStatus from core.workflow.nodes.human_input.pause_reason import HumanInputRequired, PauseReason from core.workflow.nodes.human_input_v2.events import ( NodeRunHumanInputV2FormFilledEvent, NodeRunHumanInputV2FormTimeoutEvent, ) from core.workflow.nodes.human_input_v2.runtime import PreparedForm from core.workflow.system_variables import ( build_bootstrap_variables, default_system_variables, get_node_creation_preload_selectors, inject_default_system_variable_mappings, preload_node_creation_variables, ) from core.workflow.variable_pool_initializer import add_variables_to_pool from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run from graphon.engine.layer import Layer from graphon.engine_events import ( EngineEvent, GraphRunAbortedEvent, GraphRunFailedEvent, GraphRunPartialSucceededEvent, GraphRunPausedEvent, GraphRunStartedEvent, GraphRunSucceededEvent, NodeEvent, NodeRunExceptionEvent, NodeRunFailedEvent, NodeRunHumanInputFormFilledEvent, NodeRunHumanInputFormTimeoutEvent, NodeRunIterationFailedEvent, NodeRunIterationNextEvent, NodeRunIterationStartedEvent, NodeRunIterationSucceededEvent, NodeRunLoopFailedEvent, NodeRunLoopNextEvent, NodeRunLoopStartedEvent, NodeRunLoopSucceededEvent, NodeRunReasoningChunkEvent, NodeRunRetrieverResourceEvent, NodeRunRetryEvent, NodeRunStartedEvent, NodeRunStreamChunkEvent, NodeRunSucceededEvent, ) from graphon.entities.base_node_data import BaseNodeData from graphon.entities.graph_config import NodeConfigDict from graphon.entities.pause_reason import HitlRequired from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowType from graphon.graph import Graph from graphon.graph.scoping import resolve_container_id from graphon.runtime import RuntimeState, VariablePool from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool from graphon.variables.factory import build_segment from models.workflow import Workflow from runtime_contracts.builder.execution import ( BuilderExecutionContext, BuilderExecutionPolicyError, BuilderExecutionRecorder, ConfirmedToolExecutionContext, JsonToolExecutionContext, ) from runtime_contracts.builder.native.agent_context import BuilderAgentExecutionContext from runtime_contracts.builder.native.contracts import NativeTestContextRef from services.dify_builder.native_testing.runtime_bridge import NativeRuntimeBridge from services.workflow_node_variables import load_additional_node_variables from services.workflow_runtime.entry import WorkflowEntry from services.workflow_runtime.node_factory import ( DifyGraphInitContext, DifyNodeFactory, get_default_root_node_id, resolve_workflow_node_class, ) from tasks.mail_human_input_delivery_task import dispatch_human_input_form_delivery_task if TYPE_CHECKING: from services.dify_builder.native_testing.host_execution import NativeHostLaunch logger = logging.getLogger(__name__) class _WorkflowGraphNodeData(BaseNodeData): """Node data fields the runner needs before concrete node validation.""" start_node_id: str | None = None iteration_id: str | None = None loop_id: str | None = None class _WorkflowGraphNodeConfig(BaseModel): """Top-level node wrapper used for runner graph filtering.""" id: str data: _WorkflowGraphNodeData model_config = ConfigDict(extra="allow", strict=True) def to_graph_node_config(self) -> dict[str, Any]: return { **self.model_dump(mode="python", exclude={"data"}, exclude_unset=True), "data": self.data.model_dump(mode="python", exclude_unset=True), } def to_typed_node_config(self) -> NodeConfigDict: return cast(NodeConfigDict, {**self.model_dump(mode="python", exclude={"data"}), "data": self.data}) class _WorkflowGraphEdgeConfig(BaseModel): """Top-level edge wrapper used for selecting a single-node subgraph.""" source: str | None = None target: str | None = None model_config = ConfigDict(extra="allow", strict=True) class _WorkflowGraphConfig(BaseModel): """Validated graph config boundary for runner methods. Workflow graph payloads are persisted JSON mappings with extra metadata used by graphon, so validation is intentionally limited to the top-level graph shape this runner needs before passing the mapping onward. """ nodes: list[_WorkflowGraphNodeConfig] edges: list[_WorkflowGraphEdgeConfig] model_config = ConfigDict(extra="allow", strict=True) def to_graph_config(self) -> dict[str, Any]: return self.model_dump(mode="python", exclude_unset=True) def _merge_container_metadata( lifecycle_metadata: Mapping[str, Any], node_metadata: Mapping[str, Any] ) -> dict[str, Any]: # Keep lifecycle details and ancestor ownership together, preserving the # existing node-result precedence when both sources contain the same key. return {**lifecycle_metadata, **node_metadata} @dataclass(frozen=True, slots=True) class PreparedWorkflowRun: """Prepared engine and workflow facts handed to the persistence-owning orchestrator.""" entry: WorkflowEntry workflow_type: WorkflowType workflow_version: str generate_entity: AdvancedChatAppGenerateEntity | WorkflowAppGenerateEntity application_layers: Sequence[Layer] """Layers that also receive terminal failures raised after the engine closes.""" native_host: "NativeHostLaunch | None" = None """Host that owns native observation and engine cleanup, when present.""" type WorkflowRunDriver = Callable[ [ WorkflowBasedAppRunner, PauseStateLayerConfig | None, WorkflowExecutionRepository, WorkflowNodeExecutionRepository, ], None, ] class WorkflowBasedAppRunner: def __init__( self, *, queue_manager: AppQueueManager, variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER, app_id: str, graph_engine_layers: Sequence[Layer] = (), execution_driver: WorkflowRunDriver | None = None, ): self._queue_manager = queue_manager self._variable_loader = variable_loader self._app_id = app_id self._graph_engine_layers = graph_engine_layers self._execution_driver = execution_driver def prepare(self) -> PreparedWorkflowRun | None: raise NotImplementedError @staticmethod def _resolve_user_from(invoke_from: InvokeFrom) -> UserFrom: if invoke_from.runs_as_account(): return UserFrom.ACCOUNT return UserFrom.END_USER def _init_graph( self, graph_config: Mapping[str, Any], graph_runtime_state: RuntimeState, user_from: UserFrom, invoke_from: InvokeFrom, call_depth: int, workflow_id: str = "", tenant_id: str = "", user_id: str = "", root_node_id: str | None = None, app_type: CreditUsageAppType | None = None, trace_session_id: str | None = None, builder_execution: BuilderExecutionContext | ConfirmedToolExecutionContext | JsonToolExecutionContext | None = None, execution_recorder: BuilderExecutionRecorder | None = None, native_test_required: bool = False, native_test_ref: NativeTestContextRef | None = None, builder_agent_execution: BuilderAgentExecutionContext | None = None, native_runtime: NativeRuntimeBridge | None = None, builder_mocked_node_ids: frozenset[str] = frozenset(), builder_tool_credential_bindings: Mapping[str, str] | None = None, builder_api_tool_bindings: Mapping[str, str] | None = None, ) -> Graph: """ Init graph """ if builder_execution is not None and (execution_recorder is None or not execution_recorder.healthy): raise BuilderExecutionPolicyError("missing_execution_recorder") if builder_execution is None: graph_config = _WorkflowGraphConfig.model_validate(graph_config).to_graph_config() # Restricted factory admission must inspect the original JSON, including # extras, before Graph.init introduces typed node data. # Create explicit graph init context for Graph.init. if ( builder_mocked_node_ids or builder_tool_credential_bindings is not None or builder_api_tool_bindings is not None ) and (builder_execution is not None or native_test_ref is not None): raise BuilderExecutionPolicyError("unsupported_execution_policy") run_context = build_dify_run_context( builder_execution=builder_execution, native_test_required=native_test_required, native_test_ref=native_test_ref, builder_agent_execution=builder_agent_execution, builder_mocked_node_ids=builder_mocked_node_ids, builder_tool_credential_bindings=builder_tool_credential_bindings, builder_api_tool_bindings=builder_api_tool_bindings, tenant_id=tenant_id or "", app_id=self._app_id, user_id=user_id, user_from=user_from, invoke_from=invoke_from, app_type=app_type, trace_session_id=trace_session_id, ) graph_init_context = DifyGraphInitContext( execution_driver=self._execution_driver, workflow_id=workflow_id, graph_config=graph_config, run_context=run_context, call_depth=call_depth, ) # Use the provided graph_runtime_state for consistent state management node_factory = DifyNodeFactory.from_graph_init_context( execution_recorder=execution_recorder, native_runtime=native_runtime, graph_init_context=graph_init_context, runtime_state=graph_runtime_state, ) if root_node_id is None: root_node_id = get_default_root_node_id(graph_config) # init graph graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id) if not graph: raise ValueError("graph not found in workflow") return graph def _prepare_single_node_execution( self, workflow: Workflow, single_iteration_run: Any | None = None, single_loop_run: Any | None = None, *, user_id: str, app_type: CreditUsageAppType | None = None, trace_session_id: str | None = None, ) -> tuple[Graph, VariablePool, RuntimeState]: """ Prepare graph, variable pool, and runtime state for single node execution (either single iteration or single loop). Args: workflow: The workflow instance single_iteration_run: SingleIterationRunEntity if running single iteration, None otherwise single_loop_run: SingleLoopRunEntity if running single loop, None otherwise Returns: A tuple containing (graph, variable_pool, graph_runtime_state) Raises: ValueError: If neither single_iteration_run nor single_loop_run is specified """ # Create initial runtime state with variable pool containing environment variables variable_pool = VariablePool() add_variables_to_pool( variable_pool, build_bootstrap_variables( system_variables=default_system_variables(), environment_variables=workflow.environment_variables, ), ) graph_runtime_state = RuntimeState(workflow_id=workflow.id, variable_pool=variable_pool, start_at=time.time()) # Determine which type of single node execution and get graph/variable_pool if single_iteration_run: graph, variable_pool = self._get_graph_and_variable_pool_for_single_node_run( workflow=workflow, node_id=single_iteration_run.node_id, user_inputs=dict(single_iteration_run.inputs), graph_runtime_state=graph_runtime_state, node_type_label="iteration", user_id=user_id, app_type=app_type, trace_session_id=trace_session_id, ) elif single_loop_run: graph, variable_pool = self._get_graph_and_variable_pool_for_single_node_run( workflow=workflow, node_id=single_loop_run.node_id, user_inputs=dict(single_loop_run.inputs), graph_runtime_state=graph_runtime_state, node_type_label="loop", user_id=user_id, app_type=app_type, trace_session_id=trace_session_id, ) else: raise ValueError("Neither single_iteration_run nor single_loop_run is specified") # Return the graph, variable_pool, and the same graph_runtime_state used during graph creation # This ensures all nodes in the graph reference the same RuntimeState instance return graph, variable_pool, graph_runtime_state def _get_graph_and_variable_pool_for_single_node_run( self, workflow: Workflow, node_id: str, user_inputs: dict[str, Any], graph_runtime_state: RuntimeState, node_type_label: str = "node", # 'iteration' or 'loop' for error messages *, user_id: str = "", app_type: CreditUsageAppType | None = None, trace_session_id: str | None = None, ) -> tuple[Graph, VariablePool]: """ Get graph and variable pool for single node execution (iteration or loop). The workflow graph and user inputs are treated as caller-owned data; the narrowed graph config used for execution is built locally. Args: workflow: The workflow instance node_id: The node ID to execute user_inputs: User inputs for the node graph_runtime_state: The graph runtime state node_type_label: Label for error messages ('iteration' or 'loop') Returns: A tuple containing (graph, variable_pool) """ # fetch workflow graph source_graph_config = _WorkflowGraphConfig.model_validate(workflow.graph_dict) node_user_inputs = dict(user_inputs) graph_config = source_graph_config.to_graph_config() # filter nodes only in the specified node type (iteration or loop) all_node_configs = graph_config["nodes"] nodes_by_id = {node["id"]: node for node in all_node_configs} container_ids = {node["id"]: resolve_container_id(node, nodes_by_id=nodes_by_id) for node in all_node_configs} node_ids = {node_id} while True: child_node_ids = {node["id"] for node in all_node_configs if container_ids[node["id"]] in node_ids} child_node_ids -= node_ids if not child_node_ids: break node_ids.update(child_node_ids) # Preserve ownership resolved with the complete ancestry before removing it # from the debug subtree. Nested legacy IDs can otherwise become ambiguous. selected_node_configs = [] for node in all_node_configs: if node["id"] not in node_ids: continue selected_node = dict(node) selected_node["data"] = dict(node["data"]) selected_node["data"]["container_id"] = container_ids[node["id"]] selected_node_configs.append(_WorkflowGraphNodeConfig.model_validate(selected_node)) # filter edges only in the specified node type selected_edge_configs = [ edge for edge in source_graph_config.edges if (edge.source is None or edge.source in node_ids) and (edge.target is None or edge.target in node_ids) ] node_configs = [node.to_graph_node_config() for node in selected_node_configs] edge_configs = [edge.model_dump(mode="python", exclude_unset=True) for edge in selected_edge_configs] graph_config = { **graph_config, "nodes": node_configs, "edges": edge_configs, } typed_node_configs = selected_node_configs # Create explicit graph init context for Graph.init. run_context = build_dify_run_context( tenant_id=workflow.tenant_id, app_id=self._app_id, user_id=user_id, user_from=UserFrom.ACCOUNT, invoke_from=InvokeFrom.DEBUGGER, app_type=app_type, trace_session_id=trace_session_id, ) graph_init_context = DifyGraphInitContext( execution_driver=self._execution_driver, workflow_id=workflow.id, graph_config=graph_config, run_context=run_context, call_depth=0, ) node_factory = DifyNodeFactory.from_graph_init_context( graph_init_context=graph_init_context, runtime_state=graph_runtime_state, ) target_node_config = None for node in typed_node_configs: if node.id == node_id: target_node_config = node break if not target_node_config: raise ValueError(f"{node_type_label} node id not found in workflow graph") # Get node class node_type = target_node_config.data.type node_version = str(target_node_config.data.version) node_cls = resolve_workflow_node_class(node_type=node_type, node_version=node_version) # Use the variable pool from graph_runtime_state instead of creating a new one variable_pool = graph_runtime_state.variable_pool preload_node_creation_variables( variable_loader=self._variable_loader, variable_pool=variable_pool, selectors=[ selector for node_config in typed_node_configs for selector in get_node_creation_preload_selectors( node_type=node_config.data.type, node_data=node_config.data, ) ], ) try: variable_mapping = node_cls.extract_variable_selector_to_variable_mapping( graph_config=graph_config, config=target_node_config.to_typed_node_config() ) except NotImplementedError: variable_mapping = dict[str, Sequence[str]]() variable_mapping = inject_default_system_variable_mappings( node_id=target_node_config.id, node_type=node_type, node_data=target_node_config.data, variable_mapping=variable_mapping, ) load_into_variable_pool( variable_loader=self._variable_loader, variable_pool=variable_pool, variable_mapping=variable_mapping, user_inputs=node_user_inputs, ) variable_mapping = load_additional_node_variables( node_config=target_node_config.to_typed_node_config(), graph_config=graph_config, variable_mapping=variable_mapping, variable_loader=self._variable_loader, variable_pool=variable_pool, user_inputs=node_user_inputs, ) WorkflowEntry.mapping_user_inputs_to_variable_pool( variable_mapping=variable_mapping, user_inputs=node_user_inputs, variable_pool=variable_pool, tenant_id=workflow.tenant_id, ) # init graph after constructor-time context has been loaded graph = Graph.init( graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, container_id=container_ids[node_id], skip_validation=True, ) if not graph: raise ValueError("graph not found in workflow") return graph, variable_pool @staticmethod def _build_agent_strategy_info(event: NodeRunStartedEvent) -> AgentStrategyInfo | None: raw_agent_strategy = event.extras.get("agent_strategy") if raw_agent_strategy is None: return None try: return AgentStrategyInfo.model_validate(raw_agent_strategy) except ValidationError: logger.warning("Invalid agent strategy payload for node %s", event.node_id, exc_info=True) return None def publish_human_input_results( self, workflow_entry: WorkflowEntry, pending_forms: Mapping[str, HitlRequired], forms: Mapping[str, HumanInputFormRecord], published_form_ids: set[str], ) -> None: """Publish materialized completions supplied by run orchestration.""" variable_pool = workflow_entry.graph_engine.runtime_state.variable_pool for form_id, reason in pending_forms.items(): form = forms[form_id] node_id = resolve_human_input_node_id(node_id=reason.node_id, form_id=form_id, variable_pool=variable_pool) node_title = reason.node_title or form.definition.node_title or form.node_id if form.status == HumanInputFormStatus.TIMEOUT: self._publish_event( QueueHumanInputFormTimeoutEvent( form_id=form_id, node_id=node_id, node_type=BuiltinNodeTypes.HUMAN_INPUT, node_title=node_title, expiration_time=form.expiration_time, ) ) published_form_ids.add(form_id) elif form.status == HumanInputFormStatus.SUBMITTED: action = next( (item for item in form.definition.user_actions if item.id == form.selected_action_id), None ) if action is None: raise ValueError(f"Submitted human input form has no matching action: {form_id}") submitted_data = {name: build_segment(value) for name, value in (form.submitted_data or {}).items()} self._publish_event( QueueHumanInputFormFilledEvent( form_id=form_id, node_id=node_id, node_type=BuiltinNodeTypes.HUMAN_INPUT, node_title=node_title, rendered_content=DifyHITLCallback.render_form_content_with_outputs( form.rendered_content, submitted_data, [item.output_variable_name for item in form.definition.inputs], form.definition.inputs, ), action_id=action.id, action_text=action.title, submitted_data=submitted_data, ) ) published_form_ids.add(form_id) def handle_event( self, workflow_entry: WorkflowEntry, event: EngineEvent, *, node_run_index: int = 1, node_execution_snapshots: tuple[NodeExecutionSnapshot, ...] = (), pause_reasons: Sequence[PauseReason] = (), human_input_v2_forms: Mapping[str, PreparedForm] | None = None, ) -> None: """ Handle event :param workflow_entry: workflow entry :param event: event """ node_metadata: dict[str, Any] = ( {str(key): value for key, value in event.node_run_result.metadata.items()} if isinstance(event, NodeEvent) else {} ) iteration_id = node_metadata.get(WorkflowNodeExecutionMetadataKey.ITERATION_ID.value) loop_id = node_metadata.get(WorkflowNodeExecutionMetadataKey.LOOP_ID.value) match event: case NodeRunHumanInputFormFilledEvent(): form_id = event.form_id if isinstance(event, NodeRunHumanInputV2FormFilledEvent) else event.id self._publish_event( QueueHumanInputFormFilledEvent( form_id=form_id, node_id=resolve_human_input_node_id( node_id=event.node_id, form_id=form_id, variable_pool=workflow_entry.graph_engine.runtime_state.variable_pool, ), node_type=event.node_type, node_title=event.node_title, rendered_content=event.rendered_content, action_id=event.action_id, action_text=event.action_text, submitted_data=event.submitted_data, ) ) case NodeRunHumanInputFormTimeoutEvent(): form_id = event.form_id if isinstance(event, NodeRunHumanInputV2FormTimeoutEvent) else event.id self._publish_event( QueueHumanInputFormTimeoutEvent( form_id=form_id, node_id=resolve_human_input_node_id( node_id=event.node_id, form_id=form_id, variable_pool=workflow_entry.graph_engine.runtime_state.variable_pool, ), node_type=event.node_type, node_title=event.node_title, expiration_time=event.expiration_time, ) ) case GraphRunStartedEvent(): self._publish_event( QueueWorkflowStartedEvent(reason=event.reason, node_execution_snapshots=node_execution_snapshots) ) case GraphRunSucceededEvent(): self._publish_event(QueueWorkflowSucceededEvent(outputs=event.outputs)) case GraphRunPartialSucceededEvent(): self._publish_event( QueueWorkflowPartialSuccessEvent(outputs=event.outputs, exceptions_count=event.exceptions_count) ) case GraphRunFailedEvent(): self._publish_event( QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count) ) case GraphRunAbortedEvent(): self._publish_event( QueueStopEvent( stopped_by=QueueStopEvent.StopBy.USER_MANUAL, reason=event.reason or "Workflow execution aborted", ) ) case GraphRunPausedEvent(): paused_nodes = list( dict.fromkeys(reason.node_id for reason in pause_reasons if isinstance(reason, HumanInputRequired)) ) forms = { reason.form_id: (human_input_v2_forms or {})[reason.form_id] for reason in pause_reasons if isinstance(reason, HumanInputRequired) and reason.form_version == "2" } self._enqueue_human_input_v1_notifications( [ reason for reason in pause_reasons if isinstance(reason, HumanInputRequired) and reason.form_version == "1" ] ) self._publish_event( QueueWorkflowPausedEvent( reasons=list(pause_reasons), outputs=event.outputs, paused_nodes=paused_nodes, human_input_v2_forms=forms, ) ) case NodeRunRetryEvent(): node_run_result = event.node_run_result inputs = node_run_result.inputs process_data = node_run_result.process_data outputs = project_node_outputs_for_workflow_run( node_type=event.node_type, inputs=inputs, outputs=node_run_result.outputs, ) execution_metadata = node_run_result.metadata self._publish_event( QueueNodeRetryEvent( node_execution_id=event.id, node_id=event.node_id, node_title=event.node_title, node_type=event.node_type, start_at=event.start_at, in_iteration_id=iteration_id, in_loop_id=loop_id, inputs=inputs, process_data=process_data, outputs=outputs, error=event.error, execution_metadata=execution_metadata, retry_index=event.retry_index, provider_type=event.provider_type, provider_id=event.provider_id, ) ) case NodeRunStartedEvent(): self._publish_event( QueueNodeStartedEvent( node_execution_id=event.id, node_run_index=node_run_index, node_id=event.node_id, node_title=event.node_title, node_type=event.node_type, start_at=event.start_at, in_iteration_id=iteration_id, in_loop_id=loop_id, agent_strategy=self._build_agent_strategy_info(event), provider_type=event.provider_type, provider_id=event.provider_id, ) ) case NodeRunSucceededEvent(): node_run_result = event.node_run_result inputs = node_run_result.inputs process_data = node_run_result.process_data outputs = project_node_outputs_for_workflow_run( node_type=event.node_type, inputs=inputs, outputs=node_run_result.outputs, ) execution_metadata = node_run_result.metadata if event.node_type == BuiltinNodeTypes.ANSWER and (iteration_id or loop_id): answer = outputs.get("answer") if isinstance(answer, str) and answer: self._publish_event( QueueTextChunkEvent( text=answer, from_variable_selector=[event.node_id, "answer"], in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) self._publish_event( QueueNodeSucceededEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, start_at=event.start_at, finished_at=event.finished_at, inputs=inputs, process_data=process_data, outputs=outputs, execution_metadata=execution_metadata, in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunFailedEvent(): outputs = project_node_outputs_for_workflow_run( node_type=event.node_type, inputs=event.node_run_result.inputs, outputs=event.node_run_result.outputs, ) self._publish_event( QueueNodeFailedEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, start_at=event.start_at, finished_at=event.finished_at, inputs=event.node_run_result.inputs, process_data=event.node_run_result.process_data, outputs=outputs, error=event.node_run_result.error or "Unknown error", execution_metadata=event.node_run_result.metadata, in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunExceptionEvent(): outputs = project_node_outputs_for_workflow_run( node_type=event.node_type, inputs=event.node_run_result.inputs, outputs=event.node_run_result.outputs, ) self._publish_event( QueueNodeExceptionEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, start_at=event.start_at, finished_at=event.finished_at, inputs=event.node_run_result.inputs, process_data=event.node_run_result.process_data, outputs=outputs, error=event.node_run_result.error or "Unknown error", execution_metadata=event.node_run_result.metadata, in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunStreamChunkEvent(): self._publish_event( QueueTextChunkEvent( text=event.chunk, from_variable_selector=list(event.selector), in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunReasoningChunkEvent(): self._publish_event( QueueReasoningChunkEvent( reasoning=event.chunk, from_node_id=event.node_id, is_final=event.is_final, in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunRetrieverResourceEvent(): self._publish_event( QueueRetrieverResourcesEvent( retriever_resources=[ RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources ], in_iteration_id=iteration_id, in_loop_id=loop_id, ) ) case NodeRunAgentLogEvent(): self._publish_event( QueueAgentLogEvent( id=event.message_id, label=event.label, node_execution_id=event.node_execution_id, parent_id=event.parent_id, error=event.error, status=event.status, data=event.data, metadata=event.metadata, node_id=event.node_id, ) ) case NodeRunIterationStartedEvent(): self._publish_event( QueueIterationStartEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, start_at=event.start_at, inputs=event.inputs, metadata=_merge_container_metadata(event.metadata, node_metadata), ) ) case NodeRunIterationNextEvent(): self._publish_event( QueueIterationNextEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, index=event.index, output=event.pre_iteration_output, ) ) case NodeRunIterationSucceededEvent() | NodeRunIterationFailedEvent(): self._publish_event( QueueIterationCompletedEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, start_at=event.start_at, inputs=event.inputs, outputs=event.outputs, metadata=_merge_container_metadata(event.metadata, node_metadata), steps=event.steps, error=event.error if isinstance(event, NodeRunIterationFailedEvent) else None, ) ) case NodeRunLoopStartedEvent(): self._publish_event( QueueLoopStartEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, start_at=event.start_at, inputs=event.inputs, metadata=_merge_container_metadata(event.metadata, node_metadata), ) ) case NodeRunLoopNextEvent(): self._publish_event( QueueLoopNextEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, index=event.index, output=event.pre_loop_output, ) ) case NodeRunLoopSucceededEvent() | NodeRunLoopFailedEvent(): self._publish_event( QueueLoopCompletedEvent( node_execution_id=event.id, node_id=event.node_id, node_type=event.node_type, node_title=event.node_title, start_at=event.start_at, inputs=event.inputs, outputs=event.outputs, metadata=_merge_container_metadata(event.metadata, node_metadata), steps=event.steps, error=event.error if isinstance(event, NodeRunLoopFailedEvent) else None, ) ) def _enqueue_human_input_v1_notifications(self, reasons: Sequence[HumanInputRequired]) -> None: for reason in reasons: if not reason.form_id: continue try: dispatch_human_input_form_delivery_task.apply_async( kwargs={"form_id": reason.form_id, "node_title": reason.node_title}, queue="mail", ) except Exception: # pragma: no cover - defensive logging logger.exception("Failed to enqueue human input form delivery task for form %s", reason.form_id) def _publish_event(self, event: AppQueueEvent): self._queue_manager.publish(event, PublishFrom.APPLICATION_MANAGER)