dify/api/services/workflow_runtime/entry.py

from __future__ import annotations

import logging
import time
from collections.abc import Callable, Generator, Mapping, Sequence
from contextlib import ExitStack
from contextvars import copy_context
from functools import partial
from typing import TYPE_CHECKING, Any, TypedDict
from uuid import uuid4

from configs import dify_config
from core.app.apps.exc import GenerateTaskStoppedError
from core.app.entities.app_invoke_entities import (
    InvokeFrom,
    UserFrom,
    build_dify_run_context,
)
from core.app.file_access import DatabaseFileAccessController
from core.app.layers.execution_context_layer import ExecutionContextLayer
from core.app.workflow.file_runtime import create_dify_workflow_file_runtime
from core.app.workflow.layers.observability import ObservabilityLayer
from core.credit_usage import CreditUsageAppType
from core.repositories.human_input_repository import HumanInputFormSubmissionRepository
from core.tools.workflow_as_tool.repository import WorkflowToolSource
from core.workflow.nodes.human_input.boundary import HumanInputFormEventFilter
from core.workflow.system_variables import (
    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_node_inputs_to_pool, add_variables_to_pool
from core.workflow.variable_prefixes import ENVIRONMENT_VARIABLE_NODE_ID
from core.workflow.workflow_tool_container_handler import (
    WorkflowToolContainerHandler,
    WorkflowToolEventListenerFactory,
    WorkflowToolNestedContainerHandler,
)
from domain.workflow.agent_node import is_dify_agent_node_data
from extensions.otel.runtime import is_instrument_flag_enabled
from factories import file_factory
from graphon.engine import Engine
from graphon.engine.command import CommandChannel, InMemoryChannel
from graphon.engine.container_handler.builtin.iteration import IterationContainerHandler
from graphon.engine.container_handler.builtin.loop import LoopContainerHandler
from graphon.engine.filter import EngineEventFilterContext, ResponseStreamFilter, filter_engine_events
from graphon.engine.layer import ExecutionLimitsLayer, Layer
from graphon.engine_events import EngineEvent, GraphRunFailedEvent, NodeEvent, is_node_result_event
from graphon.entities.graph_config import NodeConfigDictAdapter
from graphon.errors import WorkflowNodeRunFailedError
from graphon.file import File
from graphon.file.runtime import use_workflow_file_runtime
from graphon.graph import Graph
from graphon.nodes import BuiltinNodeTypes
from graphon.nodes.base.node import Node
from graphon.nodes.container_effects import ContainerAwaitRequest
from graphon.runtime import ReadOnlyRuntimeStateWrapper, RuntimeState, VariablePool
from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
from libs.workflow_template_values import template_value_selectors
from models.agent_config_entities import AgentSoulConfig
from models.workflow import Workflow
from services.workflow_node_variables import add_node_input_to_variable_pool, load_additional_node_variables
from services.workflow_runtime.agent_binding import WorkflowAgentBindingResolver
from services.workflow_runtime.node_factory import (
    DifyGraphInitContext,
    DifyNodeFactory,
    is_start_node_type,
    resolve_workflow_node_class,
)

if TYPE_CHECKING:
    from services.dify_builder.native_testing.host_execution import NativeHostLaunch
    from services.workflow_runtime.base_runner import WorkflowRunDriver

logger = logging.getLogger(__name__)
_file_access_controller = DatabaseFileAccessController()


def iter_dify_graph_engine_events(
    engine: Engine,
    response_stream_filter: ResponseStreamFilter | None = None,
) -> Generator[EngineEvent]:
    """
    Apply Dify's response streaming compatibility filter to Engine events.

    Graphon emits raw variable stream chunks and requires callers to opt
    into the legacy response-ordered stream behavior that Dify exposes to its
    workflow runners and tests.

    ``response_stream_filter``, when supplied, must be the same instance a
    caller intends to persist on pause (see ``WorkflowRunAgg``) so
    the filter's ``paths_map`` reflects everything the engine has actually
    streamed for this run.
    """
    yield from filter_engine_events(
        engine.run(),
        context=EngineEventFilterContext.from_engine(engine),
        filters=[
            HumanInputFormEventFilter(form_repository=HumanInputFormSubmissionRepository()),
            response_stream_filter or ResponseStreamFilter(),
        ],
    )


class _NodeConfigDict(TypedDict):
    id: str
    width: int
    height: int
    type: str
    data: dict[str, Any]


class _EdgeConfigDict(TypedDict):
    source: str
    target: str
    sourceHandle: str
    targetHandle: str


class SingleNodeGraphDict(TypedDict):
    nodes: list[_NodeConfigDict]
    edges: list[_EdgeConfigDict]


class WorkflowEntry:
    def __init__(
        self,
        graph_config: Mapping[str, Any],
        graph: Graph,
        call_depth: int,
        graph_runtime_state: RuntimeState,
        command_channel: CommandChannel | None = None,
        response_stream_filter: ResponseStreamFilter | None = None,
        native_host: NativeHostLaunch | None = None,
    ) -> None:
        """
        Init workflow entry
        :param graph_config: workflow graph config
        :param graph: workflow graph
        :param call_depth: call depth
        :param graph_runtime_state: pre-created graph runtime state
        :param command_channel: command channel for external control (optional, defaults to InMemoryChannel)
        :param response_stream_filter: pre-restored filter for resumed runs (optional, defaults to a fresh
            ResponseStreamFilter for runs with no prior pause)
        """
        # check call depth
        workflow_call_max_depth = dify_config.WORKFLOW_CALL_MAX_DEPTH
        if call_depth > workflow_call_max_depth:
            raise ValueError(f"Max workflow call depth {workflow_call_max_depth} reached.")

        # Use provided command channel or default to InMemoryChannel
        if command_channel is None:
            command_channel = InMemoryChannel()

        self.command_channel = command_channel
        self.graph_config = graph_config
        self.workflow_tool_sources: dict[str, WorkflowToolSource] = {}
        self.workflow_tool_event_listener_factory: WorkflowToolEventListenerFactory | None = None
        self._response_stream_filter = response_stream_filter or ResponseStreamFilter()
        file_runtime = create_dify_workflow_file_runtime()
        execution_context_layer = ExecutionContextLayer()
        limits_layer = ExecutionLimitsLayer(
            max_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS, max_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME
        )
        workflow_tool_event_listeners: dict[str, Callable[[NodeEvent], None]] = {}
        self.graph_engine = Engine(
            graph=graph,
            runtime_state=graph_runtime_state,
            command_channel=command_channel,
            workers=dify_config.GRAPH_ENGINE_MAX_WORKERS,
            file_runtime=file_runtime,
            container_handler_factories=(
                partial(
                    WorkflowToolNestedContainerHandler,
                    handler_factory=LoopContainerHandler,
                    hidden_event_listener=limits_layer.on_event,
                    event_listeners=workflow_tool_event_listeners,
                ),
                partial(
                    WorkflowToolNestedContainerHandler,
                    handler_factory=IterationContainerHandler,
                    hidden_event_listener=limits_layer.on_event,
                    event_listeners=workflow_tool_event_listeners,
                ),
                partial(
                    WorkflowToolContainerHandler,
                    sources=self.workflow_tool_sources,
                    hidden_event_listener=limits_layer.on_event,
                    event_listener_factory=self._create_workflow_tool_event_listener,
                    event_listeners=workflow_tool_event_listeners,
                    execution_context_factory=execution_context_layer.enter_context,
                ),
            ),
        )

        if native_host is not None:
            self.graph_engine.add_layer(native_host.observation_layer())
        else:
            self.graph_engine.add_layer(execution_context_layer)
        # Add execution limits layer
        self.graph_engine.add_layer(limits_layer)

        # Add observability layer when OTel is enabled
        if native_host is None and (dify_config.ENABLE_OTEL or is_instrument_flag_enabled()):
            self.graph_engine.add_layer(ObservabilityLayer())

    def _create_workflow_tool_event_listener(
        self, source: WorkflowToolSource, caller_workflow_id: str, caller_node_execution_id: str
    ) -> Callable[[NodeEvent], None]:
        if self.workflow_tool_event_listener_factory is None:
            return lambda event: None
        return self.workflow_tool_event_listener_factory(source, caller_workflow_id, caller_node_execution_id)

    @property
    def response_stream_filter(self) -> ResponseStreamFilter:
        return self._response_stream_filter

    def run(self) -> Generator[EngineEvent]:
        try:
            yield from iter_dify_graph_engine_events(self.graph_engine, self._response_stream_filter)
        except GenerateTaskStoppedError:
            pass
        except Exception as e:
            logger.exception("Unknown Error when workflow entry running")
            yield GraphRunFailedEvent(error=str(e))
            return

    @classmethod
    def single_step_run(
        cls,
        *,
        workflow: Workflow,
        node_id: str,
        user_id: str,
        user_inputs: Mapping[str, Any],
        variable_pool: VariablePool,
        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
        execution_driver: WorkflowRunDriver | None,
    ) -> tuple[Node, Generator[NodeEvent | ContainerAwaitRequest]]:
        """
        Single step run workflow node.

        Preconditions: The workflow owns the requested node and its Agent binding.
        Postconditions: Preload node references, including manual Agent knowledge
        filters, into the supplied pool before constructing the node. Value-free
        filter operators do not introduce variable dependencies. Legacy Agent
        nodes do not resolve an Agent Soul binding.
        :param workflow: Workflow instance
        :param node_id: node id
        :param user_id: user id
        :param user_inputs: user inputs
        :return:
        """
        node_config = workflow.get_node_config_by_id(node_id)
        node_config_data = node_config["data"]

        # Get node type
        node_type = node_config_data.type
        if node_type in {BuiltinNodeTypes.LOOP, BuiltinNodeTypes.ITERATION}:
            raise ValueError("Loop and Iteration nodes must use their engine-backed debug endpoints")
        node_version = str(node_config_data.version)
        node_cls = resolve_workflow_node_class(
            node_type=node_type,
            node_version=node_version,
            node_data=node_config_data,
        )

        # init graph context and runtime state
        run_context = build_dify_run_context(
            tenant_id=workflow.tenant_id,
            app_id=workflow.app_id,
            user_id=user_id,
            user_from=UserFrom.ACCOUNT,
            invoke_from=InvokeFrom.DEBUGGER,
            app_type=CreditUsageAppType.WORKFLOW,
        )
        graph_init_context = DifyGraphInitContext(
            workflow_id=workflow.id,
            graph_config=workflow.graph_dict,
            run_context=run_context,
            call_depth=0,
            execution_driver=execution_driver,
        )
        graph_runtime_state = RuntimeState(
            variable_pool=variable_pool,
            start_at=time.perf_counter(),
            workflow_id=workflow.id,
        )

        if is_start_node_type(node_type):
            add_node_inputs_to_pool(variable_pool, node_id=node_id, inputs=user_inputs)

        preload_node_creation_variables(
            variable_loader=variable_loader,
            variable_pool=variable_pool,
            selectors=get_node_creation_preload_selectors(
                node_type=node_type,
                node_data=node_config_data,
            ),
        )

        try:
            # variable selector to variable mapping
            variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
                graph_config=workflow.graph_dict, config=node_config
            )
        except NotImplementedError:
            variable_mapping = {}
        if is_dify_agent_node_data(node_config_data):
            # Agent Soul lives in its owned snapshot, outside graph node data.
            # Single-step execution must also load references used only by its
            # retrieval filters, which are absent from the task prompt.
            bundle = WorkflowAgentBindingResolver().resolve(
                tenant_id=workflow.tenant_id,
                app_id=workflow.app_id,
                workflow_id=workflow.id,
                node_id=node_id,
            )
            agent_soul = AgentSoulConfig.model_validate(bundle.snapshot.config_snapshot_dict)
            variable_mapping = {
                **variable_mapping,
                **{
                    f"{node_id}.metadata_filter.{'.'.join(selector)}": selector
                    for knowledge_set in agent_soul.knowledge.sets
                    if knowledge_set.metadata_filtering.mode == "manual"
                    and knowledge_set.metadata_filtering.conditions is not None
                    for condition in knowledge_set.metadata_filtering.conditions.conditions
                    if condition.comparison_operator not in ("empty", "not empty")
                    for selector in template_value_selectors(condition.value)
                },
            }
        variable_mapping = inject_default_system_variable_mappings(
            node_id=node_id,
            node_type=node_type,
            node_data=node_config_data,
            variable_mapping=variable_mapping,
        )

        # Loading missing variable from draft var here, and set it into
        # variable_pool.
        load_into_variable_pool(
            variable_loader=variable_loader,
            variable_pool=variable_pool,
            variable_mapping=variable_mapping,
            user_inputs=user_inputs,
        )
        variable_mapping = load_additional_node_variables(
            node_config=node_config,
            graph_config=workflow.graph_dict,
            variable_mapping=variable_mapping,
            variable_loader=variable_loader,
            variable_pool=variable_pool,
            user_inputs=user_inputs,
        )
        if node_type != BuiltinNodeTypes.DATASOURCE:
            cls.mapping_user_inputs_to_variable_pool(
                variable_mapping=variable_mapping,
                user_inputs=user_inputs,
                variable_pool=variable_pool,
                tenant_id=workflow.tenant_id,
            )

        # init workflow run state
        node_factory = DifyNodeFactory.from_graph_init_context(
            graph_init_context=graph_init_context,
            runtime_state=graph_runtime_state,
            use_workflow_tool_containers=False,
        )
        node = node_factory.create_node(node_config)

        try:
            generator = cls._run_node_with_layers(node)
        except Exception as e:
            logger.exception(
                "error while running node, workflow_id=%s, node_id=%s, node_type=%s, node_version=%s",
                workflow.id,
                node.id,
                node.node_type,
                node.version(),
            )
            raise WorkflowNodeRunFailedError(node=node, err_msg=str(e)) from e
        return node, generator

    @staticmethod
    def _create_single_node_graph(
        node_id: str,
        node_data: dict[str, Any],
        node_width: int = 114,
        node_height: int = 514,
    ) -> SingleNodeGraphDict:
        """
        Create a minimal graph structure for testing a single node in isolation.

        :param node_id: ID of the target node
        :param node_data: configuration data for the target node
        :param node_width: width for UI layout (default: 200)
        :param node_height: height for UI layout (default: 100)
        :return: graph dictionary with start node and target node
        """
        node_config: _NodeConfigDict = {
            "id": node_id,
            "width": node_width,
            "height": node_height,
            "type": "custom",
            "data": node_data,
        }
        start_node_config: _NodeConfigDict = {
            "id": "start",
            "width": node_width,
            "height": node_height,
            "type": "custom",
            "data": {
                "type": BuiltinNodeTypes.START,
                "title": "Start",
                "desc": "Start",
            },
        }
        return SingleNodeGraphDict(
            nodes=[start_node_config, node_config],
            edges=[
                {
                    "source": "start",
                    "target": node_id,
                    "sourceHandle": "source",
                    "targetHandle": "target",
                }
            ],
        )

    @classmethod
    def run_free_node(
        cls, node_data: dict[str, Any], node_id: str, tenant_id: str, user_id: str, user_inputs: dict[str, Any]
    ) -> tuple[Node, Generator[NodeEvent | ContainerAwaitRequest]]:
        """
        Run free node

        NOTE: only parameter_extractor/question_classifier are supported

        :param node_data: node data
        :param node_id: node id
        :param tenant_id: tenant id
        :param user_id: user id
        :param user_inputs: user inputs
        :return:
        """
        # Create a minimal graph for single node execution
        graph_dict = cls._create_single_node_graph(node_id, node_data)

        node_type = node_data.get("type", "")
        if node_type not in {BuiltinNodeTypes.PARAMETER_EXTRACTOR, BuiltinNodeTypes.QUESTION_CLASSIFIER}:
            raise ValueError(f"Node type {node_type} not supported")

        node_cls = resolve_workflow_node_class(node_type=node_type, node_version="1")
        if not node_cls:
            raise ValueError(f"Node class not found for node type {node_type}")

        # init variable pool
        variable_pool = VariablePool()
        add_variables_to_pool(variable_pool, default_system_variables())

        # init graph context and runtime state
        run_context = build_dify_run_context(
            tenant_id=tenant_id,
            app_id="",
            user_id=user_id,
            user_from=UserFrom.ACCOUNT,
            invoke_from=InvokeFrom.DEBUGGER,
            app_type=CreditUsageAppType.WORKFLOW,
        )
        graph_init_context = DifyGraphInitContext(
            workflow_id="",
            graph_config=graph_dict,
            run_context=run_context,
            call_depth=0,
        )
        graph_runtime_state = RuntimeState(
            variable_pool=variable_pool,
            start_at=time.perf_counter(),
            workflow_id="",
        )

        # init workflow run state
        node_config = NodeConfigDictAdapter.validate_python({"id": node_id, "data": node_data})
        node_factory = DifyNodeFactory.from_graph_init_context(
            graph_init_context=graph_init_context,
            runtime_state=graph_runtime_state,
        )
        node = node_factory.create_node(node_config)

        try:
            # variable selector to variable mapping
            try:
                variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
                    graph_config=graph_dict, config=node_config
                )
            except NotImplementedError:
                variable_mapping = {}
            variable_mapping = inject_default_system_variable_mappings(
                node_id=node_id,
                node_type=node_type,
                node_data=node_data,
                variable_mapping=variable_mapping,
            )

            cls.mapping_user_inputs_to_variable_pool(
                variable_mapping=variable_mapping,
                user_inputs=user_inputs,
                variable_pool=variable_pool,
                tenant_id=tenant_id,
            )

            generator = cls._run_node_with_layers(node)

            return node, generator
        except Exception as e:
            logger.exception(
                "error while running node, node_id=%s, node_type=%s, node_version=%s",
                node.id,
                node.node_type,
                node.version(),
            )
            raise WorkflowNodeRunFailedError(node=node, err_msg=str(e)) from e

    @staticmethod
    def handle_special_values(value: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
        # NOTE(QuantumGhost): Avoid using this function in new code.
        # Keep values structured as long as possible and only convert to dict
        # immediately before serialization (e.g., JSON serialization) to maintain
        # data integrity and type information.
        result = WorkflowEntry._handle_special_values(value)
        return result if isinstance(result, Mapping) or result is None else dict(result)

    @staticmethod
    def _handle_special_values(value: Any):
        if value is None:
            return value
        if isinstance(value, dict):
            res = {}
            for k, v in value.items():
                res[k] = WorkflowEntry._handle_special_values(v)
            return res
        if isinstance(value, list):
            res_list = []
            for item in value:
                res_list.append(WorkflowEntry._handle_special_values(item))
            return res_list
        if isinstance(value, File):
            return value.to_dict()
        return value

    @classmethod
    def mapping_user_inputs_to_variable_pool(
        cls,
        *,
        variable_mapping: Mapping[str, Sequence[str]],
        user_inputs: Mapping[str, Any],
        variable_pool: VariablePool,
        tenant_id: str,
    ):
        # NOTE(QuantumGhost): This logic should remain synchronized with
        # the implementation of `load_into_variable_pool`, specifically the logic about
        # variable existence checking.

        # WARNING(QuantumGhost): The semantics of this method are not clearly defined,
        # and multiple parts of the codebase depend on its current behavior.
        # Modify with caution.
        # Apply whole objects before nested overrides, independent of editor field order.
        for node_variable, variable_selector in sorted(variable_mapping.items(), key=lambda item: len(item[1])):
            # fetch node id and variable key from node_variable
            node_variable_list = node_variable.split(".")
            if len(node_variable_list) < 1:
                raise ValueError(f"Invalid node variable {node_variable}")

            node_variable_key = ".".join(node_variable_list[1:])

            if (node_variable_key not in user_inputs and node_variable not in user_inputs) and not variable_pool.get(
                variable_selector
            ):
                raise ValueError(f"Variable key {node_variable} not found in user inputs.")

            # environment variable already exist in variable pool, not from user inputs
            if variable_pool.get(variable_selector) and variable_selector[0] == ENVIRONMENT_VARIABLE_NODE_ID:
                continue

            # fetch variable node id from variable selector
            variable_node_id = variable_selector[0]

            # get input value
            input_value = (
                user_inputs[node_variable] if node_variable in user_inputs else user_inputs.get(node_variable_key)
            )
            if input_value is None:
                continue

            if isinstance(input_value, dict) and "type" in input_value and "transfer_method" in input_value:
                input_value = file_factory.build_from_mapping(
                    mapping=input_value,
                    tenant_id=tenant_id,
                    access_controller=_file_access_controller,
                )
            if (
                isinstance(input_value, list)
                and all(isinstance(item, dict) for item in input_value)
                and all("type" in item and "transfer_method" in item for item in input_value)
            ):
                input_value = file_factory.build_from_mappings(
                    mappings=input_value,
                    tenant_id=tenant_id,
                    access_controller=_file_access_controller,
                )

            # append variable and value to variable pool
            if variable_node_id != ENVIRONMENT_VARIABLE_NODE_ID:
                add_node_input_to_variable_pool(variable_pool, variable_selector, input_value)

    @staticmethod
    def _run_node_with_layers(node: Node) -> Generator[NodeEvent | ContainerAwaitRequest]:
        """
        Run a standalone node with the execution-context and observability layers used by workflow runs.
        """
        file_runtime = create_dify_workflow_file_runtime()
        command_channel = InMemoryChannel()
        runtime_state = ReadOnlyRuntimeStateWrapper(node.runtime_state)
        with use_workflow_file_runtime(file_runtime):
            layers: Sequence[Layer] = (ExecutionContextLayer(), ObservabilityLayer())
            for layer in layers:
                layer.initialize(runtime_state, command_channel)
                layer.on_graph_start()
            # Keep Flask and tracing scopes inside the snippet across suspension,
            # without exposing them to callers interleaving multiple streams.
            snippet_context = copy_context()

        node.bind_execution_id(str(uuid4()))

        def _gen():
            error: Exception | None = None
            result_event: NodeEvent | None = None
            layers_finished = False
            node_contexts = ExitStack()

            def finish_layers() -> None:
                nonlocal layers_finished
                if layers_finished:
                    return
                layers_finished = True
                for layer in layers:
                    layer.on_node_run_end(node, error, result_event)
                node_contexts.close()
                for layer in layers:
                    layer.on_graph_end(error)

            try:
                for layer in layers:
                    node_contexts.enter_context(layer.node_run_context(node))
                for layer in layers:
                    layer.on_node_run_start(node)
                for event in node.run():
                    if isinstance(event, NodeEvent) and is_node_result_event(event):
                        result_event = event
                        finish_layers()
                    yield event
            except Exception as exc:
                error = exc
                raise
            finally:
                finish_layers()

        def scoped_events():
            events = _gen()
            error: BaseException | None = None
            try:
                while True:
                    try:
                        event = (
                            snippet_context.run(next, events)
                            if error is None
                            else snippet_context.run(events.throw, error)
                        )
                    except StopIteration:
                        return
                    try:
                        yield event
                    except BaseException as exc:
                        error = exc
                    else:
                        error = None
            finally:
                snippet_context.run(events.close)

        return scoped_events()