dify/api/services/workflow_runtime/base_runner.py

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)