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 '' (): ``. """ 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 ():``. """ adapted_node_config = adapt_node_config_for_graph(node_config) _, _, _, resolved_node_data = _validated_node_config(adapted_node_config) return resolved_node_data