dify/api/services/workflow_runtime/node_factory.py

import importlib
import pkgutil
import secrets
from collections.abc import Callable, Iterator, Mapping, MutableMapping, Sequence
from copy import copy
from dataclasses import dataclass
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Protocol, cast, final, override, runtime_checkable

from sqlalchemy import select

from configs import dify_config
from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
from core.app.llm.model_access import build_dify_model_access, fetch_model_config
from core.credit_usage import created_by_from_app_type
from core.db.session_factory import get_session_maker, session_factory
from core.dify_builder.input_schema import start_schema
from core.dify_builder.real_call_consent import ConfirmedToolGuard
from core.file import remote_fetcher
from core.helper.code_executor.code_executor import (
    CodeExecutionError,
    CodeExecutor,
)
from core.helper.ssrf_proxy import graphon_ssrf_proxy
from core.memory.token_buffer_memory import TokenBufferMemory
from core.model_manager import ModelInstance
from core.prompt.entities.advanced_prompt_entities import MemoryConfig
from core.trigger.constants import TRIGGER_NODE_TYPES
from core.workflow.human_input_adapter import adapt_node_config_for_graph, parse_human_input_delivery_methods
from core.workflow.llm_environment_variable import (
    parse_llm_model_selector,
    resolve_llm_model_config,
    should_resolve_llm_model_selector,
)
from core.workflow.llm_node import DifyLLMNode
from core.workflow.node_runtime import (
    DifyFileReferenceFactory,
    DifyHumanInputNodeRuntime,
    DifyPreparedLLM,
    DifyPreparedPollingLLM,
    DifyPromptMessageSerializer,
    DifyRetrieverAttachmentLoader,
    DifyToolFileManager,
    DifyToolNodeRuntime,
    build_dify_llm_file_saver,
    build_knowledge_fs_node_dependencies,
)
from core.workflow.nodes.agent.message_transformer import AgentMessageTransformer
from core.workflow.nodes.agent.plugin_strategy_adapter import (
    PluginAgentStrategyPresentationProvider,
    PluginAgentStrategyResolver,
)
from core.workflow.nodes.agent.runtime_support import AgentRuntimeSupport
from core.workflow.nodes.agent_v2.dify_tools_builder import WorkflowAgentDifyToolsBuilder
from core.workflow.nodes.agent_v2.output_adapter import WorkflowAgentOutputAdapter
from core.workflow.nodes.agent_v2.runtime_request_builder import WorkflowAgentRuntimeRequestBuilder
from core.workflow.nodes.human_input.callback import DifyHITLCallback
from core.workflow.nodes.human_input.entities import HumanInputNodeData as DifyHumanInputNodeData
from core.workflow.nodes.human_input_v2.node import HumanInputNode as HumanInputV2Node
from core.workflow.system_variables import SystemVariableKey, get_system_text, system_variable_selector
from core.workflow.template_rendering import CodeExecutorJinja2TemplateRenderer
from core.workflow.workflow_tool_node import DifyWorkflowToolNode
from domain.workflow.agent_node import is_dify_agent_node_data
from extensions.application_services.data_sources import build_data_source_credentials
from graphon.entities.base_node_data import BaseNodeData
from graphon.entities.graph_config import NodeConfigDict, NodeConfigDictAdapter
from graphon.enums import BuiltinNodeTypes, NodeExecutionType, NodeType
from graphon.file.file_manager import file_manager
from graphon.graph.graph import NodeFactory
from graphon.model_runtime.memory import PromptMessageMemory
from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
from graphon.nodes.base.node import Node
from graphon.nodes.code.code_node import CodeExecutorProtocol
from graphon.nodes.code.entities import CodeLanguage
from graphon.nodes.code.limits import CodeNodeLimits
from graphon.nodes.document_extractor import UnstructuredApiConfig
from graphon.nodes.end.end_node import EndNode
from graphon.nodes.http_request import build_http_request_config
from graphon.nodes.llm.entities import LLMNodeData
from graphon.nodes.parameter_extractor.entities import ParameterExtractorNodeData
from graphon.nodes.question_classifier.entities import QuestionClassifierNodeData
from graphon.nodes.start.start_node import StartNode
from graphon.nodes.tool.entities import ToolProviderType
from graphon.nodes.tool.tool_node import ToolNode
from graphon.variables.segments import ArrayObjectSegment, ObjectSegment
from models.model import Conversation
from runtime_contracts.builder.execution import (
    BuilderExecutionContext,
    BuilderExecutionPolicyError,
    BuilderExecutionRecorder,
    ConfirmedToolExecutionContext,
    JsonToolExecutionContext,
    validate_execution_context,
)
from services.dify_builder.application import gmail_read_policy
from services.dify_builder.application.execution_policy import (
    RestrictedAdmissionSnapshot,
    admit_restricted_workflow,
)
from services.dify_builder.effect_mock import mocked_node_class
from services.dify_builder.json_tool_execution import admit_json_workflow
from services.dify_builder.json_tool_runtime import QualifiedJsonToolNodeRuntime
from services.dify_builder.native_testing.context_binding import NativeContextUnavailable
from services.dify_builder.native_testing.runtime_bridge import NativeRuntimeBridge
from services.dify_builder.restricted_execution import (
    RESTRICTED_HTTP_CONFIG,
    DeniedHttpFiles,
    RestrictedHttpClient,
    validate_restricted_node,
    validate_restricted_raw_node,
)
from services.dify_builder.revision import node_config_digest
from services.dify_builder.template_execution import MAX_OUTPUT_BYTES, BoundedJinja2TemplateRenderer
from services.human_input_v2.composition import WorkflowHumanInputRuntime
from services.knowledge_fs.knowledge_retrieval_v2.automatic_metadata_filter import (
    KnowledgeFSAutomaticMetadataFilterExtractor,
)
from services.knowledge_fs.knowledge_retrieval_v2.entities import KNOWLEDGE_RETRIEVAL_V2_NODE_TYPE
from services.workflow_runtime.agent_binding import WorkflowAgentBindingResolver
from services.workflow_runtime.agent_node import DifyAgentNode

if TYPE_CHECKING:
    from core.tools.workflow_as_tool.tool import WorkflowTool
    from graphon.runtime import InitParams, RuntimeState
    from services.workflow_runtime.base_runner import WorkflowRunDriver

LATEST_VERSION = "latest"
_START_NODE_TYPES: frozenset[NodeType] = frozenset(
    (BuiltinNodeTypes.START, BuiltinNodeTypes.DATASOURCE, *TRIGGER_NODE_TYPES)
)


@dataclass(frozen=True, slots=True)
class DifyGraphInitContext:
    """Explicit graph-init values owned by the workflow layer.

    Dify is gradually removing direct `InitParams` construction from its
    production call sites. Keep the translation here until `graphon` exposes an
    equivalent explicit API.
    """

    workflow_id: str
    graph_config: Mapping[str, Any]
    run_context: Mapping[str, Any]
    call_depth: int
    execution_driver: "WorkflowRunDriver | None" = None

    def to_graph_init_params(self) -> "InitParams":
        from graphon.runtime import InitParams

        return InitParams(
            workflow_id=self.workflow_id,
            graph_config=self.graph_config,
            run_context=self.run_context,
            call_depth=self.call_depth,
        )


def _import_node_package(package_name: str, *, excluded_modules: frozenset[str] = frozenset()) -> None:
    package = importlib.import_module(package_name)
    for _, module_name, _ in pkgutil.walk_packages(package.__path__, package.__name__ + "."):
        if module_name in excluded_modules:
            continue
        importlib.import_module(module_name)


@lru_cache(maxsize=1)
def register_nodes() -> None:
    """Import production node modules so they self-register with ``Node``."""
    _import_node_package("graphon.nodes")
    _import_node_package("core.workflow.nodes")
    _import_node_package("services.knowledge_fs.knowledge_retrieval_v2")


def get_node_type_classes_mapping() -> Mapping[NodeType, Mapping[str, type[Node]]]:
    """Return a read-only snapshot of the current production node registry.

    The workflow layer owns node bootstrap because it must compose built-in
    `graphon.nodes.*` implementations with workflow-local nodes under
    `core.workflow.nodes.*`. Keeping this import side effect here avoids
    reintroducing registry bootstrapping into lower-level graph primitives.
    """
    register_nodes()
    return Node.get_node_type_classes_mapping()


def resolve_workflow_node_class(
    *,
    node_type: NodeType,
    node_version: str,
    node_data: Mapping[str, Any] | BaseNodeData | None = None,
) -> type[Node]:
    """Resolve the production node class for the requested type/version."""
    node_mapping = get_node_type_classes_mapping().get(node_type)
    if not node_mapping:
        raise ValueError(f"No class mapping found for node type: {node_type}")

    # Historical Agent nodes used version=2 for their tool-parameter format.
    # Only the explicit kind marker identifies the newer Dify Agent node.
    if (
        node_data is not None
        and node_type == BuiltinNodeTypes.AGENT
        and node_version == "2"
        and not is_dify_agent_node_data(node_data)
    ):
        node_version = "1"

    latest_node_class = node_mapping.get(LATEST_VERSION)
    matched_node_class = node_mapping.get(node_version)
    node_class = matched_node_class or latest_node_class
    if not node_class:
        raise ValueError(f"No latest version class found for node type: {node_type}")
    return node_class


def is_start_node_type(node_type: NodeType) -> bool:
    """Return True when the node type can serve as a workflow entry point."""
    return node_type in _START_NODE_TYPES


def get_default_root_node_id(graph_config: Mapping[str, Any]) -> str:
    """Resolve the default entry node for a persisted top-level workflow graph.

    This workflow-layer helper depends on start-node semantics defined by
    `is_start_node_type`, so it intentionally lives next to the node registry
    instead of in the raw `graphon.entities.graph_config` schema module.
    """
    nodes = graph_config.get("nodes")
    if not isinstance(nodes, list):
        raise ValueError("nodes in workflow graph must be a list")

    for node in nodes:
        if not isinstance(node, Mapping):
            continue

        if node.get("type") == "custom-note":
            continue

        node_id = node.get("id")
        data = node.get("data")
        if not isinstance(node_id, str) or not isinstance(data, Mapping):
            continue

        node_type = data.get("type")
        if isinstance(node_type, str) and is_start_node_type(node_type):
            return node_id

    raise ValueError("Unable to determine default root node ID from workflow graph")


class _LazyNodeTypeClassesMapping(MutableMapping[NodeType, Mapping[str, type[Node]]]):
    """Mutable dict-like view over the current node registry."""

    def __init__(self) -> None:
        self._cached_snapshot: dict[NodeType, Mapping[str, type[Node]]] = {}
        self._cached_version = -1
        self._deleted: set[NodeType] = set()
        self._overrides: dict[NodeType, Mapping[str, type[Node]]] = {}

    def _snapshot(self) -> dict[NodeType, Mapping[str, type[Node]]]:
        current_version = Node.get_registry_version()
        if self._cached_version != current_version:
            self._cached_snapshot = dict(get_node_type_classes_mapping())
            self._cached_version = current_version
        if not self._deleted and not self._overrides:
            return self._cached_snapshot

        snapshot = {key: value for key, value in self._cached_snapshot.items() if key not in self._deleted}
        snapshot.update(self._overrides)
        return snapshot

    @override
    def __getitem__(self, key: NodeType) -> Mapping[str, type[Node]]:
        return self._snapshot()[key]

    @override
    def __setitem__(self, key: NodeType, value: Mapping[str, type[Node]]) -> None:
        self._deleted.discard(key)
        self._overrides[key] = value

    @override
    def __delitem__(self, key: NodeType) -> None:
        if key in self._overrides:
            del self._overrides[key]
            return
        if key in self._cached_snapshot:
            self._deleted.add(key)
            return
        raise KeyError(key)

    @override
    def __iter__(self) -> Iterator[NodeType]:
        return iter(self._snapshot())

    @override
    def __len__(self) -> int:
        return len(self._snapshot())


# Keep the canonical node-class mapping in the workflow layer that also bootstraps
# legacy `core.workflow.nodes.*` registrations.
NODE_TYPE_CLASSES_MAPPING: MutableMapping[NodeType, Mapping[str, type[Node]]] = _LazyNodeTypeClassesMapping()


type LLMCompatibleNodeData = LLMNodeData | QuestionClassifierNodeData | ParameterExtractorNodeData


def fetch_memory(
    *,
    conversation_id: str | None,
    app_id: str,
    node_data_memory: MemoryConfig | None,
    model_instance: ModelInstance,
) -> TokenBufferMemory | None:
    """Build prompt memory for node construction without requiring Flask-local state."""
    if not node_data_memory or not conversation_id:
        return None

    # Node construction can happen in graph initialization paths where Flask's
    # app context is not active. Use the app-configured session factory instead
    # of resolving db.engine through Flask-SQLAlchemy's current_app proxy.
    with session_factory.create_session() as session:
        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
        conversation = session.scalar(stmt)
        if not conversation:
            return None

    return TokenBufferMemory(conversation=conversation, model_instance=model_instance)


# The exact classes a confirmed Builder Tool run may construct.
_CONFIRMED_CLASSES: Mapping[str, type[Node]] = {"start": StartNode, "tool": ToolNode, "end": EndNode}


@runtime_checkable
class ConfirmedToolGuardIssuer(BuilderExecutionRecorder, Protocol):
    """A launch recorder that can also issue the service-owned guard for this Run.

    Implemented by the Builder service for one claimed launch identity: it
    claims the grant at the given actual native IDs and returns the service's
    guard, or refuses. The factory never builds or reconstructs a guard itself.
    """

    def confirmed_guard(
        self, *, node_id: str, native_run_id: str, native_node_execution_id: str
    ) -> ConfirmedToolGuard: ...


class _DeniedConfirmedToolFiles:
    """Tool file owner for a confirmed run: file messages are refused, never stored."""

    def create_file_by_raw(self, *, file_binary: bytes, mimetype: str, filename: str | None = None) -> Any:
        raise BuilderExecutionPolicyError("unsupported_tool_file")

    def get_file_generator_by_tool_file_id(self, tool_file_id: str) -> Any:
        raise BuilderExecutionPolicyError("unsupported_tool_file")


class DefaultWorkflowCodeExecutor:
    def execute(
        self,
        *,
        language: CodeLanguage,
        code: str,
        inputs: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        return CodeExecutor.execute_workflow_code_template(
            language=language,
            code=code,
            inputs=inputs,
        )

    def is_execution_error(self, error: Exception) -> bool:
        return isinstance(error, CodeExecutionError)


@final
class DifyNodeFactory(NodeFactory):
    """
    Compose native nodes from the production registry. Builder restricted launches
    receive only exact admitted implementations and their restricted collaborators.
    """

    @classmethod
    def from_graph_init_context(
        cls,
        *,
        graph_init_context: DifyGraphInitContext,
        runtime_state: "RuntimeState",
        execution_recorder: BuilderExecutionRecorder | None = None,
        native_runtime: NativeRuntimeBridge | None = None,
        human_input_run_context: Mapping[str, Any] | DifyRunContext | None = None,
        use_workflow_tool_containers: bool = True,
        workflow_tools: dict[str, "WorkflowTool"] | None = None,
    ) -> "DifyNodeFactory":
        """Bridge Dify's explicit init context into the current `graphon` API."""
        return cls(
            init_params=graph_init_context.to_graph_init_params(),
            runtime_state=runtime_state,
            human_input_run_context=human_input_run_context,
            use_workflow_tool_containers=use_workflow_tool_containers,
            execution_driver=graph_init_context.execution_driver,
            execution_recorder=execution_recorder,
            native_runtime=native_runtime,
            workflow_tools=workflow_tools,
        )

    def __init__(
        self,
        init_params: "InitParams",
        runtime_state: "RuntimeState",
        human_input_run_context: Mapping[str, Any] | DifyRunContext | None = None,
        use_workflow_tool_containers: bool = True,
        workflow_tools: dict[str, "WorkflowTool"] | None = None,
        execution_driver: "WorkflowRunDriver | None" = None,
        *,
        execution_recorder: BuilderExecutionRecorder | None = None,
        native_runtime: NativeRuntimeBridge | None = None,
        _request_hmac_key: bytes | None = None,
    ) -> None:
        self.init_params = init_params
        self._builder_init_params = init_params
        self.runtime_state = runtime_state
        self._dify_context = self._resolve_dify_context(init_params.run_context)
        self._use_workflow_tool_containers = use_workflow_tool_containers
        self._execution_driver = execution_driver
        self.workflow_tools = workflow_tools if workflow_tools is not None else {}
        self._human_input_run_context = (
            self._dify_context
            if human_input_run_context is None
            else (
                human_input_run_context
                if isinstance(human_input_run_context, DifyRunContext)
                else self._resolve_dify_context(human_input_run_context)
            )
        )
        self.execution_recorder = execution_recorder
        self._native_runtime = native_runtime
        if native_runtime is not None and self._dify_context.native_test_ref is None:
            raise NativeContextUnavailable("native_context_missing")
        if self._dify_context.native_test_required or self._dify_context.native_test_ref is not None:
            ref = self._dify_context.native_test_ref
            if ref is None:
                raise NativeContextUnavailable("native_context_missing")
            if native_runtime is None:
                raise NativeContextUnavailable("native_runtime_unavailable")
            context = native_runtime.resolve(ref)
            if self._dify_context.builder_execution is not None or (
                self._dify_context.tenant_id,
                self._dify_context.app_id,
                self._dify_context.user_id,
                init_params.workflow_id,
                init_params.call_depth,
                dict(init_params.graph_config),
            ) != (
                context.tenant_id,
                context.app_id,
                context.actor_id,
                context.workflow_id,
                context.call_depth,
                context.graph,
            ):
                raise NativeContextUnavailable("native_context_mismatch")
            return
        context = self._dify_context.builder_execution

        # Exhaustive context dispatch, before ordinary eager setup. A Builder
        # context never falls through to native unrestricted construction.
        match context:
            case ConfirmedToolExecutionContext():
                self._initialize_confirmed_tool_runtime(context)
                return
            case JsonToolExecutionContext():
                self._initialize_json_tool_runtime(context)
                return
            case BuilderExecutionContext():
                self._initialize_restricted_runtime(context, _request_hmac_key)
                return
            case None:
                pass
            case _:
                raise BuilderExecutionPolicyError("unsupported_execution_policy")
        self._code_executor: CodeExecutorProtocol = DefaultWorkflowCodeExecutor()
        self._code_limits = CodeNodeLimits(
            max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
            max_number=dify_config.CODE_MAX_NUMBER,
            min_number=dify_config.CODE_MIN_NUMBER,
            max_precision=dify_config.CODE_MAX_PRECISION,
            max_depth=dify_config.CODE_MAX_DEPTH,
            max_number_array_length=dify_config.CODE_MAX_NUMBER_ARRAY_LENGTH,
            max_string_array_length=dify_config.CODE_MAX_STRING_ARRAY_LENGTH,
            max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
        )
        self._jinja2_template_renderer = CodeExecutorJinja2TemplateRenderer()
        self._template_transform_max_output_length = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
        self._http_request_http_client = graphon_ssrf_proxy
        self._remote_file_http_client = remote_fetcher.graphon_remote_file_fetcher
        self._bound_tool_file_manager_factory = lambda: DifyToolFileManager(
            self._dify_context,
            conversation_id_getter=self._conversation_id,
        )
        self._file_reference_factory = DifyFileReferenceFactory(self._dify_context)
        self._prompt_message_serializer = DifyPromptMessageSerializer()
        self._retriever_attachment_loader = DifyRetrieverAttachmentLoader(
            file_reference_factory=self._file_reference_factory,
        )
        self._llm_file_saver = build_dify_llm_file_saver(
            run_context=self._dify_context,
            http_client=self._remote_file_http_client,
            conversation_id_getter=self._conversation_id,
        )
        self._human_input_runtime = DifyHumanInputNodeRuntime(
            self._human_input_run_context,
            workflow_execution_id_getter=lambda: get_system_text(
                self.runtime_state.variable_pool,
                SystemVariableKey.WORKFLOW_EXECUTION_ID,
            ),
            conversation_id_getter=self._conversation_id,
        )
        self._tool_runtime = DifyToolNodeRuntime(
            self._dify_context,
            workflow_id=self.init_params.workflow_id,
            execution_driver=execution_driver,
            workflow_tools=self.workflow_tools,
        )
        self._http_request_file_manager = file_manager
        self._document_extractor_unstructured_api_config = UnstructuredApiConfig(
            api_url=dify_config.UNSTRUCTURED_API_URL,
            api_key=dify_config.UNSTRUCTURED_API_KEY or "",
        )
        self._http_request_config = build_http_request_config(
            max_connect_timeout=dify_config.HTTP_REQUEST_MAX_CONNECT_TIMEOUT,
            max_read_timeout=dify_config.HTTP_REQUEST_MAX_READ_TIMEOUT,
            max_write_timeout=dify_config.HTTP_REQUEST_MAX_WRITE_TIMEOUT,
            max_binary_size=dify_config.HTTP_REQUEST_NODE_MAX_BINARY_SIZE,
            max_text_size=dify_config.HTTP_REQUEST_NODE_MAX_TEXT_SIZE,
            ssl_verify=dify_config.HTTP_REQUEST_NODE_SSL_VERIFY,
            ssrf_default_max_retries=dify_config.SSRF_DEFAULT_MAX_RETRIES,
        )

        self._llm_credentials_provider, self._llm_model_factory = build_dify_model_access(self._dify_context)
        self._agent_strategy_resolver = PluginAgentStrategyResolver()
        self._agent_strategy_presentation_provider = PluginAgentStrategyPresentationProvider()
        self._agent_runtime_support = AgentRuntimeSupport()
        self._agent_message_transformer = AgentMessageTransformer()

    def _require_builder_owner(
        self, context: BuilderExecutionContext | ConfirmedToolExecutionContext | JsonToolExecutionContext
    ) -> None:
        recorder = self.execution_recorder
        if recorder is None or not recorder.healthy:
            raise BuilderExecutionPolicyError("missing_execution_recorder")
        if (
            self._dify_context.tenant_id,
            self._dify_context.app_id,
            self._dify_context.user_id,
            self.init_params.workflow_id,
            self.init_params.call_depth,
        ) != (context.tenant_id, context.app_id, context.actor_id, context.workflow_id, 0):
            raise BuilderExecutionPolicyError("execution_owner_mismatch")

    def _admission_snapshot(self) -> RestrictedAdmissionSnapshot:
        graph = dict(self.init_params.graph_config)
        return RestrictedAdmissionSnapshot(
            app_mode="workflow",
            workflow_kind="standard",
            graph=graph,
            input_schema=start_schema(graph),
            features={},
            has_environment_variables=False,
            has_conversation_variables=False,
            has_external_tracing=False,
        )

    def _initialize_restricted_runtime(self, context: BuilderExecutionContext, request_hmac_key: bytes | None) -> None:
        self._require_builder_owner(context)
        snapshot = self._admission_snapshot()
        # Check every raw class/config binding before concrete validation.
        for node_config in snapshot.graph.get("nodes", []):
            self._validate_restricted_raw_config(node_config)
        admit_restricted_workflow(context, snapshot)
        self._request_hmac_key = request_hmac_key if request_hmac_key is not None else secrets.token_bytes(32)

    def _initialize_confirmed_tool_runtime(self, context: ConfirmedToolExecutionContext) -> None:
        """Admit exactly the reserved Start → Tool → End graph and one request-scoped Tool runtime.

        Only ``DifyToolNodeRuntime`` is constructed, with the service-owned guard
        issuer from the launch recorder. Graphon binds a node execution ID only
        at ``get_runtime``; the closure then reads this run's actual native Run
        ID from the graph runtime state and asks the service to claim the grant
        and build its guard, before any ToolManager resolution.
        """
        checked = validate_execution_context(context.model_dump(mode="json"))
        if not isinstance(checked, ConfirmedToolExecutionContext) or checked != context:
            raise BuilderExecutionPolicyError("invalid_execution_context")
        self._require_builder_owner(context)
        recorder = self.execution_recorder
        if not isinstance(recorder, ConfirmedToolGuardIssuer):
            raise BuilderExecutionPolicyError("missing_confirmed_tool_guard")
        snapshot = self._admission_snapshot()
        for node_config in snapshot.graph.get("nodes", []):
            self._validate_confirmed_raw_config(node_config)
        admission = gmail_read_policy.admit_get_message_workflow(snapshot)
        if admission.nodes != context.admitted_nodes:
            raise BuilderExecutionPolicyError("node_binding_mismatch")

        def issue_guard(node_id: str, node_execution_id: str) -> ConfirmedToolGuard:
            native_run_id = get_system_text(self.runtime_state.variable_pool, SystemVariableKey.WORKFLOW_EXECUTION_ID)
            if not native_run_id:
                raise BuilderExecutionPolicyError("execution_launch_mismatch")
            return recorder.confirmed_guard(
                node_id=node_id, native_run_id=native_run_id, native_node_execution_id=node_execution_id
            )

        self._confirmed_tool_runtime = DifyToolNodeRuntime(self._dify_context, confirmed_guard_factory=issue_guard)

    def _initialize_json_tool_runtime(self, context: JsonToolExecutionContext) -> None:
        """Admit the original three-node graph before native Tool data adaptation."""
        from services.dify_builder.execution_policy_service import JsonLaunchRecorder

        checked = validate_execution_context(context.model_dump(mode="json"))
        if not isinstance(checked, JsonToolExecutionContext) or checked != context:
            raise BuilderExecutionPolicyError("invalid_execution_context")
        self._require_builder_owner(context)
        recorder = self.execution_recorder
        if not isinstance(recorder, JsonLaunchRecorder):
            raise BuilderExecutionPolicyError("missing_json_tool_issuer")
        snapshot = self._admission_snapshot()
        for node_config in snapshot.graph.get("nodes", []):
            self._validate_json_raw_config(node_config)
        if admit_json_workflow(snapshot).nodes != context.admitted_nodes:
            raise BuilderExecutionPolicyError("node_binding_mismatch")
        native_run_id = get_system_text(self.runtime_state.variable_pool, SystemVariableKey.WORKFLOW_EXECUTION_ID)
        if not native_run_id:
            raise BuilderExecutionPolicyError("execution_launch_mismatch")
        self._json_tool_runtime = QualifiedJsonToolNodeRuntime(
            self._dify_context,
            native_run_id=native_run_id,
            native_task_id=recorder.native_task_id,
            issuer=recorder,
        )

    @override
    def with_runtime_state(self, runtime_state: "RuntimeState") -> "DifyNodeFactory":
        return DifyNodeFactory(
            init_params=self._builder_init_params
            if self._dify_context.builder_execution is not None or self._native_runtime is not None
            else self.init_params,
            runtime_state=runtime_state,
            human_input_run_context=self._human_input_run_context,
            use_workflow_tool_containers=self._use_workflow_tool_containers,
            workflow_tools=self.workflow_tools,
            execution_driver=self._execution_driver,
            execution_recorder=self.execution_recorder,
            _request_hmac_key=(
                self._request_hmac_key
                if isinstance(self._dify_context.builder_execution, BuilderExecutionContext)
                else None
            ),
            native_runtime=self._native_runtime,
        ).with_graph_config(self.init_params.graph_config)

    @override
    def with_graph_config(self, graph_config: Mapping[str, Any]) -> "DifyNodeFactory":
        factory = copy(self)
        factory.init_params = self.init_params.model_copy(update={"graph_config": graph_config})
        return factory

    @property
    def execution_driver(self) -> "WorkflowRunDriver | None":
        return self._execution_driver

    @property
    def human_input_run_context(self) -> DifyRunContext:
        return self._human_input_run_context

    @staticmethod
    def _resolve_dify_context(run_context: Mapping[str, Any]) -> DifyRunContext:
        raw_ctx = run_context.get(DIFY_RUN_CONTEXT_KEY)
        if raw_ctx is None:
            raise ValueError(f"run_context missing required key: {DIFY_RUN_CONTEXT_KEY}")
        if isinstance(raw_ctx, DifyRunContext):
            return raw_ctx
        return DifyRunContext.model_validate(raw_ctx)

    def _conversation_id(self) -> str | None:
        return get_system_text(self.runtime_state.variable_pool, SystemVariableKey.CONVERSATION_ID)

    def _raw_builder_node_config(self, node_config: Mapping[str, Any]) -> dict[str, Any]:
        """Remove only Graphon's generated root ownership field before checking the original binding."""
        data = node_config.get("data")
        if self._dify_context.builder_execution is None or not isinstance(data, BaseNodeData):
            return dict(node_config)
        raw_data = data.model_dump(mode="json", by_alias=True, exclude_unset=True)
        original = next(
            (node for node in self._builder_init_params.graph_config["nodes"] if node["id"] == node_config["id"]),
            None,
        )
        if original is None:
            raise BuilderExecutionPolicyError("execution_binding_mismatch")
        if "container_id" not in original["data"] and raw_data.get("container_id") == "":
            del raw_data["container_id"]
        return {**node_config, "data": raw_data}

    @override
    def create_node(self, node_config: dict[str, Any] | NodeConfigDict) -> Node:
        """
        Create a Node instance from node configuration data using the traditional mapping.

        :param node_config: node configuration dictionary containing type and other data
        :return: initialized Node instance
        :raises ValueError: if node_config fails NodeConfigDict/BaseNodeData validation
            (including pydantic ValidationError, which subclasses ValueError),
            if node type is unknown, or if no implementation exists for the resolved version
        """
        context = self._dify_context.builder_execution
        node_config = self._raw_builder_node_config(node_config)
        if isinstance(context, ConfirmedToolExecutionContext):
            return self._create_confirmed_node(node_config)
        if isinstance(context, JsonToolExecutionContext):
            return self._create_json_node(node_config)
        if context is not None:
            self._validate_restricted_raw_config(node_config)
        adapted_node_config = adapt_node_config_for_graph(node_config)
        node_id, node_data, node_class, resolved_node_data = _validated_node_config(
            adapted_node_config,
            resolve_node_class=self._resolve_node_class_for_factory,
            validate_resolved_node_data=self._validate_resolved_node_data,
            pre_validate_node_data=self._validate_restricted_raw_config if context is not None else None,
        )
        if self._native_runtime is not None:
            ref = self._dify_context.native_test_ref
            if ref is None:
                raise NativeContextUnavailable("native_context_missing")
            prepared = self._native_runtime.prepare_node(
                ref=ref,
                node_id=node_id,
                node_data=resolved_node_data,
                node_class=node_class,
                runtime_state=self.runtime_state,
            )
            node = prepared.node_class(
                node_id=node_id,
                data=resolved_node_data.model_dump(mode="python", by_alias=True),
                init_params=self.init_params,
                runtime_state=self.runtime_state,
                **prepared.arguments,
            )
            prepared.attach(node)
            return node
        if isinstance(context, BuilderExecutionContext):
            validate_restricted_node(
                context=context, node_id=node_id, node_class=node_class, resolved_node_data=resolved_node_data
            )
            kwargs: dict[str, object] = {}
            if node_data.type == BuiltinNodeTypes.HTTP_REQUEST:
                assert self.execution_recorder is not None
                client = RestrictedHttpClient(
                    context=context,
                    node_id=node_id,
                    recorder=self.execution_recorder,
                    request_hmac_key=self._request_hmac_key,
                )
                files = DeniedHttpFiles(client)
                kwargs = {
                    "http_request_config": RESTRICTED_HTTP_CONFIG,
                    "http_client": client,
                    "tool_file_manager_factory": files,
                    "file_manager": files,
                    "file_reference_factory": files,
                }
            elif node_data.type == BuiltinNodeTypes.TEMPLATE_TRANSFORM:
                kwargs = {
                    "jinja2_template_renderer": BoundedJinja2TemplateRenderer(),
                    "max_output_length": MAX_OUTPUT_BYTES,
                }
            return node_class(
                node_id=node_id,
                data=resolved_node_data.model_dump(mode="python", by_alias=True),
                init_params=self.init_params,
                runtime_state=self.runtime_state,
                **kwargs,
            )
        if node_id in self._dify_context.builder_mocked_node_ids:
            # Validate native configuration, then stop before any dependency
            # factory or native constructor can acquire execution resources.
            return mocked_node_class(node_class)(
                node_id=node_id,
                data=resolved_node_data.model_dump(mode="python", by_alias=True),
                init_params=self.init_params,
                runtime_state=self.runtime_state,
            )
        node_type = node_data.type
        if node_type == BuiltinNodeTypes.LLM:
            resolved_node_data = self._resolve_llm_model_reference(cast(LLMNodeData, resolved_node_data))
        node: Node | None = None
        node_init_kwargs_factories: Mapping[NodeType, Callable[[], dict[str, object]]] = {
            BuiltinNodeTypes.DATASOURCE: lambda: {
                "datasource_credentials": build_data_source_credentials(
                    database_client=get_session_maker()
                ).providers.get_datasource_credentials,
            },
            BuiltinNodeTypes.CODE: lambda: {
                "code_executor": self._code_executor,
                "code_limits": self._code_limits,
            },
            BuiltinNodeTypes.TEMPLATE_TRANSFORM: lambda: {
                "jinja2_template_renderer": self._jinja2_template_renderer,
                "max_output_length": self._template_transform_max_output_length,
            },
            BuiltinNodeTypes.HTTP_REQUEST: lambda: {
                "http_request_config": self._http_request_config,
                "http_client": self._http_request_http_client,
                "tool_file_manager_factory": self._bound_tool_file_manager_factory,
                "file_manager": self._http_request_file_manager,
                "file_reference_factory": self._file_reference_factory,
            },
            BuiltinNodeTypes.HUMAN_INPUT: lambda: (
                {
                    "human_input_runtime": WorkflowHumanInputRuntime(self._dify_context),
                }
                if node_class is HumanInputV2Node
                else {
                    "hitl_callback": self._build_human_input_callback(
                        node_data=DifyHumanInputNodeData.model_validate(adapted_node_config["data"]),
                        execution_id_getter=lambda: node.execution_id if node is not None else None,
                    ),
                }
            ),
            BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
                node_class=node_class,
                node_data=resolved_node_data,
                wrap_model_instance=True,
                include_http_client=True,
                include_llm_file_saver=True,
                include_prompt_message_serializer=True,
                include_retriever_attachment_loader=True,
                include_jinja2_template_renderer=True,
            ),
            BuiltinNodeTypes.DOCUMENT_EXTRACTOR: lambda: {
                "unstructured_api_config": self._document_extractor_unstructured_api_config,
                "http_client": self._remote_file_http_client,
            },
            BuiltinNodeTypes.QUESTION_CLASSIFIER: lambda: self._build_llm_compatible_node_init_kwargs(
                node_class=node_class,
                node_data=resolved_node_data,
                wrap_model_instance=True,
                include_http_client=True,
                include_llm_file_saver=True,
                include_prompt_message_serializer=True,
                include_retriever_attachment_loader=False,
                include_jinja2_template_renderer=False,
            ),
            BuiltinNodeTypes.PARAMETER_EXTRACTOR: lambda: self._build_llm_compatible_node_init_kwargs(
                node_class=node_class,
                node_data=resolved_node_data,
                wrap_model_instance=True,
                include_http_client=False,
                include_llm_file_saver=False,
                include_prompt_message_serializer=True,
                include_retriever_attachment_loader=False,
                include_jinja2_template_renderer=False,
            ),
            BuiltinNodeTypes.TOOL: lambda: {
                "tool_file_manager": self._bound_tool_file_manager_factory(),
                "runtime": self._tool_runtime,
            },
            BuiltinNodeTypes.AGENT: lambda: self._build_agent_node_init_kwargs(node_class=node_class),
            KNOWLEDGE_RETRIEVAL_V2_NODE_TYPE: lambda: {
                **build_knowledge_fs_node_dependencies(),
                "metadata_filter_extractor": KnowledgeFSAutomaticMetadataFilterExtractor(
                    credentials_provider=self._llm_credentials_provider,
                    model_factory=self._llm_model_factory,
                ),
            },
        }
        node_init_kwargs = node_init_kwargs_factories.get(node_type, lambda: {})()
        constructor_node_data = resolved_node_data.model_dump(mode="python", by_alias=True)
        node = node_class(
            node_id=node_id,
            data=constructor_node_data,
            init_params=self.init_params,
            runtime_state=self.runtime_state,
            **node_init_kwargs,
        )
        return node

    def _create_confirmed_node(self, node_config: Mapping[str, Any]) -> Node:
        """Only the exact admitted Start/Tool v1/End classes; the Tool gets the confirmed runtime only."""
        self._validate_confirmed_raw_config(node_config)
        adapted_node_config = adapt_node_config_for_graph(node_config)
        node_id, _node_data, node_class, resolved_node_data = _validated_node_config(
            adapted_node_config,
            resolve_node_class=self._resolve_node_class_for_factory,
            validate_resolved_node_data=self._validate_resolved_node_data,
            pre_validate_node_data=self._validate_confirmed_raw_config,
        )
        if node_class is not _CONFIRMED_CLASSES.get(resolved_node_data.type) or resolved_node_data.version != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        kwargs: dict[str, object] = {}
        if node_class is ToolNode:
            kwargs = {"tool_file_manager": _DeniedConfirmedToolFiles(), "runtime": self._confirmed_tool_runtime}
        return node_class(
            node_id=node_id,
            data=resolved_node_data.model_dump(mode="python", by_alias=True),
            init_params=self.init_params,
            runtime_state=self.runtime_state,
            **kwargs,
        )

    def _create_json_node(self, node_config: Mapping[str, Any]) -> Node:
        """Construct only audited native classes, with the qualified Tool runtime."""
        self._validate_json_raw_config(node_config)
        adapted = adapt_node_config_for_graph(node_config)
        node_id, _node_data, node_class, resolved = _validated_node_config(
            adapted,
            resolve_node_class=self._resolve_node_class_for_factory,
            validate_resolved_node_data=self._validate_resolved_node_data,
        )
        if node_class is not _CONFIRMED_CLASSES.get(resolved.type) or resolved.version != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        kwargs: dict[str, object] = {}
        if node_class is ToolNode:
            kwargs = {"tool_file_manager": _DeniedConfirmedToolFiles(), "runtime": self._json_tool_runtime}
        return node_class(
            node_id=node_id,
            data=resolved.model_dump(mode="python", by_alias=True),
            init_params=self.init_params,
            runtime_state=self.runtime_state,
            **kwargs,
        )

    def _validate_json_raw_config(self, node_config: Mapping[str, Any], node_class: type[Node] | None = None) -> None:
        context = self._dify_context.builder_execution
        if not isinstance(context, JsonToolExecutionContext):
            raise BuilderExecutionPolicyError("unsupported_execution_policy")
        data, node_id = node_config.get("data"), node_config.get("id")
        if not isinstance(data, Mapping) or not isinstance(node_id, str):
            raise BuilderExecutionPolicyError("invalid_node_configuration")
        node_type, version = data.get("type"), data.get("version", "1")
        if not isinstance(node_type, str) or node_type not in _CONFIRMED_CLASSES or version != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        if node_class is None:
            node_class = self._resolve_node_class(node_type=node_type, node_version=version, node_data=data)
        if node_class is not _CONFIRMED_CLASSES[node_type] or node_class.version() != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        binding = next((item for item in context.admitted_nodes if item.node_id == node_id), None)
        if (
            binding is None
            or binding.implementation != f"{node_class.__module__}.{node_class.__qualname__}"
            or binding.node_version != "1"
            or binding.normalized_config_digest != node_config_digest(data)
        ):
            raise BuilderExecutionPolicyError("execution_binding_mismatch")

    def _validate_confirmed_raw_config(
        self, node_config: Mapping[str, Any], node_class: type[Node] | None = None
    ) -> None:
        context = self._dify_context.builder_execution
        if not isinstance(context, ConfirmedToolExecutionContext):
            raise BuilderExecutionPolicyError("unsupported_execution_policy")
        data, node_id = node_config.get("data"), node_config.get("id")
        if not isinstance(data, Mapping) or not isinstance(node_id, str):
            raise BuilderExecutionPolicyError("invalid_node_configuration")
        node_type, version = data.get("type"), data.get("version", "1")
        if not isinstance(node_type, str) or node_type not in _CONFIRMED_CLASSES or version != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        if node_class is None:
            node_class = self._resolve_node_class(node_type=node_type, node_version=version, node_data=data)
        if node_class is not _CONFIRMED_CLASSES[node_type] or node_class.version() != "1":
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        binding = next((b for b in context.admitted_nodes if b.node_id == node_id), None)
        if (
            binding is None
            or binding.implementation != f"{node_class.__module__}.{node_class.__qualname__}"
            or binding.node_version != "1"
            or binding.normalized_config_digest != node_config_digest(data)
        ):
            raise BuilderExecutionPolicyError("execution_binding_mismatch")

    def _validate_restricted_raw_config(
        self, node_config: Mapping[str, Any], node_class: type[Node] | None = None
    ) -> None:
        context = self._dify_context.builder_execution
        assert isinstance(context, BuilderExecutionContext)
        data, node_id = node_config.get("data"), node_config.get("id")
        if not isinstance(data, Mapping) or not isinstance(node_id, str):
            raise BuilderExecutionPolicyError("invalid_node_configuration")
        node_type, version = data.get("type"), data.get("version", "1")
        if (
            not isinstance(node_type, str)
            or node_type not in {"start", "end", "http-request", "template-transform"}
            or version != "1"
        ):
            raise BuilderExecutionPolicyError("unsupported_node_implementation")
        if node_class is None:
            node_class = self._resolve_node_class(node_type=node_type, node_version=version, node_data=data)
        validate_restricted_raw_node(context=context, node_id=node_id, node_class=node_class, raw_node_data=data)

    @override
    def validate_node(self, node_config: NodeConfigDict) -> NodeExecutionType:
        context = self._dify_context.builder_execution
        raw_node_config = self._raw_builder_node_config(node_config)
        validator = None
        if isinstance(context, ConfirmedToolExecutionContext):
            validator = self._validate_confirmed_raw_config
        elif isinstance(context, JsonToolExecutionContext):
            validator = self._validate_json_raw_config
        elif isinstance(context, BuilderExecutionContext):
            validator = self._validate_restricted_raw_config
        if validator is not None:
            validator(raw_node_config, None)
        node_id, _, node_class, resolved_node_data = _validated_node_config(
            adapt_node_config_for_graph(raw_node_config),
            resolve_node_class=self._resolve_node_class_for_factory,
            validate_resolved_node_data=self._validate_resolved_node_data,
            # Compatibility adaptation can move wrapped Tool configurations into
            # parameters. Bind the saved raw configuration to the actual resolved
            # implementation, rather than comparing that adapted representation
            # with the raw digest admitted before graph initialization.
            pre_validate_node_data=(
                (lambda _adapted_config, resolved_class: validator(raw_node_config, resolved_class))
                if validator is not None
                else None
            ),
        )
        if isinstance(context, BuilderExecutionContext):
            validate_restricted_node(
                context=context, node_id=node_id, node_class=node_class, resolved_node_data=resolved_node_data
            )
        return node_class.execution_type

    @staticmethod
    def _validate_resolved_node_data(node_class: type[Node], node_data: BaseNodeData) -> BaseNodeData:
        """
        Re-validate the permissive graph payload with the concrete NodeData model declared by the resolved node class.
        """
        validate_node_data = getattr(node_class, "validate_node_data", None)
        if callable(validate_node_data):
            return cast("BaseNodeData", validate_node_data(node_data))
        return node_data

    def _resolve_node_class_for_factory(
        self,
        *,
        node_type: NodeType,
        node_version: str,
        node_data: BaseNodeData,
    ) -> type[Node]:
        if (
            self._use_workflow_tool_containers
            and self._dify_context.builder_execution is None
            and self._native_runtime is None
            and node_type == BuiltinNodeTypes.TOOL
            and node_data.model_dump().get("provider_type") == ToolProviderType.WORKFLOW
        ):
            return DifyWorkflowToolNode
        return self._resolve_node_class(
            node_type=node_type,
            node_version=node_version,
            node_data=node_data,
        )

    @staticmethod
    def _resolve_node_class(
        *,
        node_type: NodeType,
        node_version: str,
        node_data: Mapping[str, Any] | BaseNodeData | None = None,
    ) -> type[Node]:
        if node_type == BuiltinNodeTypes.LLM:
            return DifyLLMNode
        return resolve_workflow_node_class(
            node_type=node_type,
            node_version=node_version,
            node_data=node_data,
        )

    def _resolve_llm_model_reference(self, node_data: LLMNodeData) -> LLMNodeData:
        """Resolve an optional shared model selector from the workflow variable pool."""

        model_selector = (node_data.model_extra or {}).get("model_selector")
        if not should_resolve_llm_model_selector(model_selector):
            return node_data

        selector = parse_llm_model_selector(model_selector)
        variable = self.runtime_state.variable_pool.get(selector)
        if not isinstance(variable, ObjectSegment):
            raise ValueError(f"LLM environment variable '{selector[1]}' was not found or is not an LLM variable")

        resolved_model = resolve_llm_model_config(
            node_model=node_data.model,
            variable_name=selector[1],
            variable_value=variable.value,
        )
        return node_data.model_copy(update={"model": resolved_model})

    def _build_agent_node_init_kwargs(self, *, node_class: type[Node]) -> dict[str, object]:
        if issubclass(node_class, DifyAgentNode):
            from clients.agent_backend import AgentBackendRunEventAdapter, AgentBackendRunRequestBuilder
            from clients.agent_backend.factory import create_agent_backend_run_client
            from core.workflow.nodes.agent_v2.file_tenant_validator import AgentOutputFileTenantValidator
            from core.workflow.nodes.agent_v2.output_failure_orchestrator import OutputFailureOrchestrator
            from core.workflow.nodes.agent_v2.output_file_rebacker import reback_tool_file_output
            from core.workflow.nodes.agent_v2.output_type_checker import PerOutputTypeChecker
            from core.workflow.nodes.agent_v2.session_store import WorkflowAgentWorkspaceStore
            from services.dify_builder.native_testing.agent_tool_declarations import AgentToolDeclarationService

            kwargs: dict[str, object] = {
                "binding_resolver": WorkflowAgentBindingResolver(),
                "runtime_request_builder": WorkflowAgentRuntimeRequestBuilder(
                    request_builder=AgentBackendRunRequestBuilder(),
                    dify_tools_builder=WorkflowAgentDifyToolsBuilder(
                        native_declaration_preparer=AgentToolDeclarationService(),
                    ),
                ),
                "agent_backend_client": create_agent_backend_run_client(
                    base_url=dify_config.AGENT_BACKEND_BASE_URL,
                    api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                    use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                    fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
                    stream_max_reconnects=dify_config.AGENT_BACKEND_STREAM_MAX_RECONNECTS,
                ),
                "event_adapter": AgentBackendRunEventAdapter(),
                # Agent Files §4.6: reback file outputs from the ToolFile row so
                # downstream metadata is authoritative, not sandbox-provided.
                "output_adapter": WorkflowAgentOutputAdapter(tool_file_rebacker=reback_tool_file_output),
                # Stage 4 §5/§7: per-output validation + failure orchestration. The
                # tenant validator resolves ToolFile (canonical) + UploadFile refs.
                "type_checker": PerOutputTypeChecker(file_validator=AgentOutputFileTenantValidator()),
                "failure_orchestrator": OutputFailureOrchestrator(),
                "session_store": WorkflowAgentWorkspaceStore(),
            }
            if self._dify_context.builder_agent_execution is not None:
                from core.dify_builder.models import Actor
                from services.dify_builder.wiring import build_confirmed_agent_node_owners

                preparation, commands, recovery = build_confirmed_agent_node_owners(
                    actor=Actor(tenant_id=self._dify_context.tenant_id, account_id=self._dify_context.user_id),
                    backend=kwargs["agent_backend_client"],
                )
                kwargs.update(agent_preparation=preparation, agent_commands=commands, agent_run_recovery=recovery)
            return kwargs
        return {
            "strategy_resolver": self._agent_strategy_resolver,
            "presentation_provider": self._agent_strategy_presentation_provider,
            "runtime_support": self._agent_runtime_support,
            "message_transformer": self._agent_message_transformer,
        }

    def _build_human_input_callback(
        self,
        *,
        node_data: DifyHumanInputNodeData,
        execution_id_getter: Callable[[], str | None],
    ) -> DifyHITLCallback:
        return DifyHITLCallback(
            form_repository=self._human_input_runtime.build_form_repository(),
            node_data=node_data,
            conversation_id=self._conversation_id(),
            delivery_methods=self._human_input_runtime._resolve_delivery_methods(node_data=node_data),
            display_in_ui=self._human_input_runtime._display_in_ui(node_data=node_data),
            file_reference_factory=self._file_reference_factory,
            execution_id_getter=execution_id_getter,
        )

    def _build_llm_compatible_node_init_kwargs(
        self,
        *,
        node_class: type[Node],
        node_data: BaseNodeData,
        wrap_model_instance: bool,
        include_http_client: bool,
        include_llm_file_saver: bool,
        include_prompt_message_serializer: bool,
        include_retriever_attachment_loader: bool,
        include_jinja2_template_renderer: bool,
    ) -> dict[str, object]:
        validated_node_data = cast(LLMCompatibleNodeData, node_data)
        model_instance = self._build_model_instance_for_llm_node(validated_node_data)
        request_metadata: dict[str, object] = {"app_id": self._dify_context.app_id}
        app_type = self._dify_context.app_type
        created_by = self._dify_context.created_by
        if app_type is not None:
            request_metadata["app_type"] = app_type
            request_metadata["created_by"] = created_by_from_app_type(app_type)
        elif created_by is not None:
            request_metadata["created_by"] = created_by

        node_model_instance = (
            self._wrap_model_instance_for_node(
                node_data=validated_node_data,
                model_instance=model_instance,
                request_metadata=request_metadata,
            )
            if wrap_model_instance
            else model_instance
        )
        node_init_kwargs: dict[str, object] = {
            "credentials_provider": self._llm_credentials_provider,
            "model_factory": self._llm_model_factory,
            "model_instance": node_model_instance,
            "memory": self._build_memory_for_llm_node(
                node_data=validated_node_data,
                model_instance=model_instance,
            ),
        }
        if validated_node_data.type == BuiltinNodeTypes.QUESTION_CLASSIFIER:
            node_init_kwargs["template_renderer"] = self._jinja2_template_renderer
        if include_http_client:
            node_init_kwargs["http_client"] = self._remote_file_http_client
        if include_llm_file_saver:
            node_init_kwargs["llm_file_saver"] = self._llm_file_saver
        if include_prompt_message_serializer:
            node_init_kwargs["prompt_message_serializer"] = self._prompt_message_serializer
        if include_retriever_attachment_loader:
            node_init_kwargs["retriever_attachment_loader"] = self._build_retriever_attachment_loader(
                cast(LLMNodeData, validated_node_data)
            )
        if include_jinja2_template_renderer:
            node_init_kwargs["jinja2_template_renderer"] = self._jinja2_template_renderer
        if validated_node_data.type == BuiltinNodeTypes.LLM:
            node_init_kwargs["default_query_selector"] = system_variable_selector(SystemVariableKey.QUERY)
            node_init_kwargs["polling_finalizer"] = cast(DifyPreparedLLM, node_model_instance).finalize_llm_polling
        return node_init_kwargs

    @staticmethod
    def _wrap_model_instance_for_node(
        *,
        node_data: LLMCompatibleNodeData,
        model_instance: ModelInstance,
        request_metadata: Mapping[str, object] | None = None,
    ) -> DifyPreparedLLM:
        # Only graphon's LLM node consumes the polling protocol. Keep classifier
        # and extractor nodes on the existing wrapper even if the same model
        # advertises polling support.
        if node_data.type == BuiltinNodeTypes.LLM and DifyNodeFactory._supports_plugin_llm_polling(model_instance):
            return DifyPreparedPollingLLM(model_instance, request_metadata=request_metadata)
        return DifyPreparedLLM(model_instance, request_metadata=request_metadata)

    @staticmethod
    def _supports_plugin_llm_polling(model_instance: ModelInstance) -> bool:
        return model_instance.get_model_schema().support_polling

    def _build_retriever_attachment_loader(self, node_data: LLMNodeData) -> DifyRetrieverAttachmentLoader:
        return DifyRetrieverAttachmentLoader(
            file_reference_factory=self._file_reference_factory,
            segment_access_checker=self._build_retriever_segment_access_checker(
                node_data.context.variable_selector if node_data.context.enabled else None
            ),
        )

    def _build_retriever_segment_access_checker(
        self,
        context_variable_selector: Sequence[str] | None,
    ) -> Callable[[str], bool]:
        def checker(segment_id: str) -> bool:
            if not context_variable_selector:
                return False

            context_value = self.runtime_state.variable_pool.get(context_variable_selector)
            if not isinstance(context_value, ArrayObjectSegment):
                return False

            for item in context_value.value:
                if not isinstance(item, Mapping):
                    continue
                metadata = item.get("metadata")
                if not isinstance(metadata, Mapping):
                    continue
                if metadata.get("_source") == "knowledge" and str(metadata.get("segment_id")) == str(segment_id):
                    return True
            return False

        return checker

    def _build_model_instance_for_llm_node(self, node_data: LLMCompatibleNodeData) -> ModelInstance:
        node_data_model = node_data.model
        model_instance, _ = fetch_model_config(
            node_data_model=node_data_model,
            credentials_provider=self._llm_credentials_provider,
            model_factory=self._llm_model_factory,
        )
        model_instance.model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
        return model_instance

    def _build_memory_for_llm_node(
        self,
        *,
        node_data: LLMCompatibleNodeData,
        model_instance: ModelInstance,
    ) -> PromptMessageMemory | None:
        if node_data.memory is None:
            return None

        conversation_id = get_system_text(self.runtime_state.variable_pool, SystemVariableKey.CONVERSATION_ID)
        return fetch_memory(
            conversation_id=conversation_id,
            app_id=self._dify_context.app_id,
            node_data_memory=node_data.memory,
            model_instance=model_instance,
        )


def _node_config_error(node_config: Mapping[str, Any], exc: Exception) -> ValueError:
    """``exc`` re-raised as a ``ValueError`` whose message starts with the
    offending node's id and type, e.g. ``node 'node4' (http-request): ...``."""
    data = node_config.get("data")
    node_type = str(data.get("type") or "") if isinstance(data, Mapping) else ""
    node_id = str(node_config.get("id") or "")
    return ValueError(f"node {node_id!r} ({node_type or 'unknown type'}): {exc}")


def _validated_node_config(
    adapted_node_config: Mapping[str, Any],
    *,
    resolve_node_class: Callable[..., type[Node]] = DifyNodeFactory._resolve_node_class,
    validate_resolved_node_data: Callable[[type[Node], BaseNodeData], BaseNodeData] = (
        DifyNodeFactory._validate_resolved_node_data
    ),
    pre_validate_node_data: Callable[[Mapping[str, Any], type[Node] | None], None] | None = None,
) -> tuple[str, BaseNodeData, type[Node], BaseNodeData]:
    """Adapt-shaped node config in, fully validated node out -- the single
    sequence ``DifyNodeFactory.create_node`` and ``validate_node_config`` both
    run: shared-shape validation, node-class resolution, then re-validation
    against the concrete NodeData model the resolved class declares, then
    Human Input delivery-method validation through the runtime's pure parser.

    ``validate_node_config``'s whole contract is "whatever raises here would
    raise at ``Graph.init``", which only holds while the two callers share this
    one sequence. ``create_node`` passes its own bound ``_resolve_node_class`` /
    ``_validate_resolved_node_data`` (both ``@staticmethod``s in production, but
    tests monkeypatch them per-instance); ``validate_node_config`` has no
    instance, so it defaults to the class's own.

    Restricted callers supply a callback to check raw bindings before common DTO
    validation and the actual selected class immediately before concrete validation.
    Its typed policy refusal is preserved without generic node error wrapping.

    Returns ``(node_id, shared node_data, resolved node_class, resolved_node_data)``.
    Ordinary validation raises ``ValueError`` prefixed ``node '<id>' (<type>): ``.
    """
    if pre_validate_node_data is not None:
        pre_validate_node_data(adapted_node_config, None)
    try:
        typed_node_config = NodeConfigDictAdapter.validate_python(adapted_node_config)
        node_id = typed_node_config["id"]
        node_data = typed_node_config["data"]
        node_class = resolve_node_class(
            node_type=node_data.type,
            node_version=str(node_data.version),
            node_data=node_data,
        )
        if pre_validate_node_data is not None:
            # Guard the actual selected class too: the registry may change between lookups.
            pre_validate_node_data(adapted_node_config, node_class)
        # Graph configs are initially validated against permissive shared node data.
        # Re-validate using the resolved node class so workflow-local node schemas
        # stay explicit and constructors receive the concrete typed payload.
        resolved_node_data = validate_resolved_node_data(node_class, node_data)
        if node_data.type == BuiltinNodeTypes.HUMAN_INPUT:
            parse_human_input_delivery_methods(resolved_node_data)
    except BuilderExecutionPolicyError:
        # Preserve the typed, sanitized policy refusal for native routing.
        raise
    except ValueError as exc:
        # pydantic's ValidationError subclasses ValueError. Its message names
        # only the model class ("... for HttpRequestNodeData"), so a run that
        # dies at Graph.init could not say WHICH node was wrong (ESQ1-302).
        # Every workflow start goes through here, so this changes the error
        # text of every graph-init failure, not only the Builder's.
        raise _node_config_error(adapted_node_config, exc) from exc
    return node_id, node_data, node_class, resolved_node_data


def validate_node_config(node_config: Mapping[str, Any] | NodeConfigDict) -> BaseNodeData:
    """Validate ONE raw node config exactly as ``DifyNodeFactory.create_node``
    does before constructing the node -- adapt, shared-shape validation,
    node-class resolution, concrete NodeData re-validation, Human Input delivery
    parsing -- with no factory instance and no runtime state.

    This is the dry check a draft can be put through BEFORE it is run: whatever
    raises here would raise at ``Graph.init`` and kill the run before its first
    node. Raises ``ValueError`` prefixed ``node <id> (<type>):``.
    """
    adapted_node_config = adapt_node_config_for_graph(node_config)
    _, _, _, resolved_node_data = _validated_node_config(adapted_node_config)
    return resolved_node_data