dify/api/core/workflow/node_runtime.py

from __future__ import annotations

from collections.abc import Callable, Generator, Mapping, Sequence
from dataclasses import dataclass
from enum import Enum
from typing import TYPE_CHECKING, Any, Literal, Protocol, cast, overload, override

from pydantic import JsonValue
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
from core.app.file_access import (
    DatabaseFileAccessController,
    grant_upload_file_access,
    is_retriever_segment_access_granted,
)
from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
from core.db.session_factory import session_factory
from core.dify_builder.execution_policy import BuilderExecutionPolicyError
from core.dify_builder.real_call_consent import ConfirmedToolGuard
from core.helper.trace_id_helper import ParentTraceContext
from core.llm_generator.output_parser.errors import OutputParserError
from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
from core.model_context import use_credit_usage_metadata
from core.model_manager import ModelInstance, QuotaManagedModelInstance
from core.plugin.impl.exc import PluginDaemonClientSideError, PluginInvokeError
from core.plugin.impl.plugin import PluginInstaller
from core.prompt.utils.prompt_message_util import PromptMessageUtil
from core.repositories.human_input_repository import (
    FormCreateParams,
    HumanInputFormEntity,
    HumanInputFormRepository,
    HumanInputFormRepositoryImpl,
)
from core.tools.entities.tool_entities import ToolProviderType as CoreToolProviderType
from core.tools.errors import ToolInvokeError
from core.tools.tool_engine import ToolEngine
from core.tools.tool_file_manager import ToolFileManager
from core.tools.tool_manager import ToolManager
from core.tools.utils.message_transformer import ToolFileMessageTransformer
from core.workflow.file_reference import build_file_reference
from core.workflow.nodes.human_input.entities import (
    FileInputConfig,
    FileListInputConfig,
    FormInputConfig,
    HumanInputNodeData,
)
from extensions.ext_database import db
from factories import file_factory
from graphon.file import File, FileTransferMethod, FileType
from graphon.model_runtime.entities import LLMMode
from graphon.model_runtime.entities.llm_entities import (
    LLMPollingResult,
    LLMPollingStatus,
    LLMResult,
    LLMResultChunk,
    LLMResultChunkWithStructuredOutput,
    LLMResultWithStructuredOutput,
    LLMUsage,
)
from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageTool
from graphon.model_runtime.entities.model_entities import AIModelEntity
from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
from graphon.nodes.llm.runtime_protocols import (
    LLMPollingCapableProtocol,
    LLMProtocol,
    PromptMessageSerializerProtocol,
    RetrieverAttachmentLoaderProtocol,
)
from graphon.nodes.protocols import FileReferenceFactoryProtocol, HttpClientProtocol, ToolFileManagerProtocol
from graphon.nodes.runtime import ToolNodeRuntimeProtocol
from graphon.nodes.tool.exc import ToolNodeError, ToolRuntimeInvocationError, ToolRuntimeResolutionError
from graphon.nodes.tool_runtime_entities import (
    ToolRuntimeHandle,
    ToolRuntimeMessage,
    ToolRuntimeParameter,
)
from models.dataset import SegmentAttachmentBinding
from models.model import UploadFile
from services.tools.builtin_tools_manage_service import BuiltinToolManageService

from .human_input_adapter import (
    BoundRecipient,
    DeliveryChannelConfig,
    DeliveryMethodType,
    EmailDeliveryMethod,
    EmailRecipients,
    is_human_input_webapp_enabled,
    parse_human_input_delivery_methods,
)
from .system_variables import SystemVariableKey, get_system_text


class PollingLLMRuntimeProtocol(Protocol):
    """Runtime capability required by the workflow polling adapter."""

    def start_llm_polling(
        self,
        *,
        provider: str,
        model: str,
        credentials: dict[str, Any],
        model_parameters: dict[str, Any],
        prompt_messages: Sequence[PromptMessage],
        tools: Sequence[PromptMessageTool] | None,
        stop: Sequence[str] | None,
        json_schema: dict[str, Any] | None,
    ) -> LLMPollingResult: ...

    def check_llm_polling(
        self,
        *,
        provider: str,
        model: str,
        credentials: dict[str, Any],
        plugin_state: dict[str, JsonValue],
    ) -> LLMPollingResult: ...


if TYPE_CHECKING:
    from core.app.apps.workflow_app_runner import WorkflowRunDriver
    from core.tools.__base.tool import Tool
    from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage
    from graphon.nodes.llm.file_saver import LLMFileSaver
    from graphon.nodes.tool.entities import ToolNodeData


_file_access_controller = DatabaseFileAccessController()


def resolve_dify_run_context(run_context: Mapping[str, Any] | DifyRunContext) -> DifyRunContext:
    if isinstance(run_context, DifyRunContext):
        return run_context

    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 apply_dify_debug_email_recipient(
    method: DeliveryChannelConfig,
    *,
    enabled: bool,
    actor_id: str | None,
) -> DeliveryChannelConfig:
    """Apply the Dify debugger-specific email recipient override outside `graphon`."""
    if not enabled:
        return method
    if not isinstance(method, EmailDeliveryMethod):
        return method
    if not method.config.debug_mode:
        return method

    if actor_id is None:
        debug_recipients = EmailRecipients(include_bound_group=False, items=[])
    else:
        debug_recipients = EmailRecipients(
            include_bound_group=False,
            items=[BoundRecipient(reference_id=actor_id)],
        )
    debug_config = method.config.with_recipients(debug_recipients)
    return method.model_copy(update={"config": debug_config})


class DifyFileReferenceFactory(FileReferenceFactoryProtocol):
    def __init__(self, run_context: Mapping[str, Any] | DifyRunContext) -> None:
        self._run_context = resolve_dify_run_context(run_context)

    @override
    def build_from_mapping(self, *, mapping: Mapping[str, Any]):
        return file_factory.build_from_mapping(
            mapping=mapping,
            tenant_id=self._run_context.tenant_id,
            access_controller=_file_access_controller,
        )


class DifyPreparedLLM(LLMProtocol):
    """Workflow-layer adapter that hides the full `ModelInstance` API from `graphon` nodes."""

    def __init__(self, model_instance: ModelInstance, request_metadata: Mapping[str, object] | None = None) -> None:
        self._model_instance = model_instance
        self._request_metadata = request_metadata

    @property
    @override
    def provider(self) -> str:
        return self._model_instance.provider

    @property
    @override
    def model_name(self) -> str:
        return self._model_instance.model_name

    @property
    @override
    def parameters(self) -> Mapping[str, Any]:
        return self._model_instance.parameters

    @parameters.setter
    @override
    def parameters(self, value: Mapping[str, Any]) -> None:
        self._model_instance.parameters = value

    @property
    @override
    def stop(self) -> Sequence[str] | None:
        return self._model_instance.stop

    @override
    def get_model_schema(self) -> AIModelEntity:
        model_schema = cast(LargeLanguageModel, self._model_instance.model_type_instance).get_model_schema(
            self._model_instance.model_name,
            self._model_instance.credentials,
        )
        if model_schema is None:
            raise ValueError(f"Model schema not found for {self._model_instance.model_name}")
        return model_schema

    @override
    def get_llm_num_tokens(self, prompt_messages: Sequence[PromptMessage]) -> int:
        return self._model_instance.get_llm_num_tokens(prompt_messages)

    @overload
    def invoke_llm(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        model_parameters: Mapping[str, Any],
        tools: Sequence[PromptMessageTool] | None,
        stop: Sequence[str] | None,
        stream: Literal[False],
    ) -> LLMResult: ...

    @overload
    def invoke_llm(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        model_parameters: Mapping[str, Any],
        tools: Sequence[PromptMessageTool] | None,
        stop: Sequence[str] | None,
        stream: Literal[True],
    ) -> Generator[LLMResultChunk]: ...

    @override
    def invoke_llm(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        model_parameters: Mapping[str, Any],
        tools: Sequence[PromptMessageTool] | None,
        stop: Sequence[str] | None,
        stream: bool,
    ) -> LLMResult | Generator[LLMResultChunk]:
        return self._model_instance.invoke_llm(
            prompt_messages=list(prompt_messages),
            model_parameters=dict(model_parameters),
            tools=list(tools or []),
            stop=list(stop or []),
            stream=stream,
            request_metadata=self._request_metadata,
        )

    @overload
    def invoke_llm_with_structured_output(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        json_schema: Mapping[str, Any],
        model_parameters: Mapping[str, Any],
        stop: Sequence[str] | None,
        stream: Literal[False],
    ) -> LLMResultWithStructuredOutput: ...

    @overload
    def invoke_llm_with_structured_output(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        json_schema: Mapping[str, Any],
        model_parameters: Mapping[str, Any],
        stop: Sequence[str] | None,
        stream: Literal[True],
    ) -> Generator[LLMResultChunkWithStructuredOutput]: ...

    @override
    def invoke_llm_with_structured_output(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        json_schema: Mapping[str, Any],
        model_parameters: Mapping[str, Any],
        stop: Sequence[str] | None,
        stream: bool,
    ) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput]:
        return invoke_llm_with_structured_output(
            provider=self.provider,
            model_schema=self.get_model_schema(),
            model_instance=self._model_instance,
            prompt_messages=prompt_messages,
            json_schema=json_schema,
            model_parameters=model_parameters,
            stop=list(stop or []),
            stream=stream,
            request_metadata=self._request_metadata,
        )

    @override
    def is_structured_output_parse_error(self, error: Exception) -> bool:
        return isinstance(error, OutputParserError)

    def finalize_llm_polling(self) -> None:
        """Finalize resources held by a polling invocation, if any."""


class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
    """Prepared workflow LLM adapter that exposes Graphon's polling protocol."""

    def __init__(self, model_instance: ModelInstance, request_metadata: Mapping[str, object] | None = None) -> None:
        super().__init__(model_instance, request_metadata=request_metadata)
        model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
        self._polling_runtime = cast(PollingLLMRuntimeProtocol, model_type_instance.model_runtime)
        self._polling_quota_reservation = None

    @override
    def finalize_llm_polling(self) -> None:
        reservation = self._polling_quota_reservation
        self._polling_quota_reservation = None
        if reservation is not None:
            QuotaManagedModelInstance.release_quota_safely(reservation)

    def _settle_polling_quota(self, polling_result: LLMPollingResult) -> LLMPollingResult:
        reservation = self._polling_quota_reservation
        if reservation is None or polling_result.status == LLMPollingStatus.RUNNING:
            return polling_result

        try:
            if polling_result.status == LLMPollingStatus.SUCCEEDED:
                if polling_result.result is None:
                    raise ValueError("A successful LLM polling result must include a model result.")
                reservation.commit(polling_result.result.usage)
            else:
                reservation.release()
        except Exception:
            QuotaManagedModelInstance.release_quota_safely(reservation)
            raise
        finally:
            self._polling_quota_reservation = None

        return polling_result

    @override
    def start_llm_polling(
        self,
        *,
        prompt_messages: Sequence[PromptMessage],
        model_parameters: Mapping[str, Any],
        tools: Sequence[PromptMessageTool] | None,
        stop: Sequence[str] | None,
        json_schema: Mapping[str, Any] | None,
    ) -> LLMPollingResult:
        self.finalize_llm_polling()

        if isinstance(self._model_instance, QuotaManagedModelInstance):
            self._polling_quota_reservation = self._model_instance._reserve_quota_for_request(self._request_metadata)

        try:
            polling_result = self._polling_runtime.start_llm_polling(
                provider=self.provider,
                model=self.model_name,
                credentials=self._model_instance.credentials,
                prompt_messages=prompt_messages,
                model_parameters=dict(model_parameters),
                tools=tools,
                stop=stop,
                json_schema=dict(json_schema) if json_schema is not None else None,
            )
            return self._settle_polling_quota(polling_result)
        except Exception:
            self.finalize_llm_polling()
            raise

    @override
    def check_llm_polling(
        self,
        *,
        plugin_state: Mapping[str, JsonValue],
    ) -> LLMPollingResult:
        try:
            polling_result = self._polling_runtime.check_llm_polling(
                provider=self.provider,
                model=self.model_name,
                credentials=self._model_instance.credentials,
                plugin_state=dict(plugin_state),
            )
            return self._settle_polling_quota(polling_result)
        except Exception:
            self.finalize_llm_polling()
            raise


class DifyPromptMessageSerializer(PromptMessageSerializerProtocol):
    @override
    def serialize(
        self,
        *,
        model_mode: LLMMode,
        prompt_messages: Sequence[PromptMessage],
    ) -> Any:
        return PromptMessageUtil.prompt_messages_to_prompt_for_saving(
            model_mode=model_mode,
            prompt_messages=prompt_messages,
        )


class DifyRetrieverAttachmentLoader(RetrieverAttachmentLoaderProtocol):
    """Resolve retriever attachments through Dify persistence and return graph file references."""

    _segment_access_checker: Callable[[str], bool] | None

    def __init__(
        self,
        *,
        file_reference_factory: FileReferenceFactoryProtocol,
        segment_access_checker: Callable[[str], bool] | None = None,
    ) -> None:
        self._file_reference_factory = file_reference_factory
        self._segment_access_checker = segment_access_checker

    @override
    def load(self, *, segment_id: str) -> Sequence[File]:
        if not is_retriever_segment_access_granted(segment_id):
            return []
        if self._segment_access_checker is not None and not self._segment_access_checker(segment_id):
            return []

        with Session(db.engine, expire_on_commit=False) as session:
            attachments_with_bindings = session.execute(
                select(SegmentAttachmentBinding, UploadFile)
                .join(UploadFile, UploadFile.id == SegmentAttachmentBinding.attachment_id)
                .where(SegmentAttachmentBinding.segment_id == segment_id)
            ).all()

        grant_upload_file_access(str(upload_file.id) for _, upload_file in attachments_with_bindings)
        return [
            self._file_reference_factory.build_from_mapping(
                mapping={
                    "id": upload_file.id,
                    "filename": upload_file.name,
                    "extension": "." + upload_file.extension,
                    "mime_type": upload_file.mime_type,
                    "type": FileType.IMAGE,
                    "transfer_method": FileTransferMethod.LOCAL_FILE,
                    "remote_url": upload_file.source_url,
                    "reference": build_file_reference(record_id=str(upload_file.id)),
                    "size": upload_file.size,
                }
            )
            for _, upload_file in attachments_with_bindings
        ]


class DifyToolFileManager(ToolFileManagerProtocol):
    """Workflow adapter that resolves conversation scope outside `graphon`."""

    _conversation_id_getter: Callable[[], str | None] | None

    def __init__(
        self,
        run_context: Mapping[str, Any] | DifyRunContext,
        *,
        conversation_id_getter: Callable[[], str | None] | None = None,
    ) -> None:
        self._run_context = resolve_dify_run_context(run_context)
        self._manager = ToolFileManager()
        self._conversation_id_getter = conversation_id_getter

    @override
    def create_file_by_raw(
        self,
        *,
        file_binary: bytes,
        mimetype: str,
        filename: str | None = None,
    ) -> Any:
        conversation_id = self._conversation_id_getter() if self._conversation_id_getter is not None else None
        return self._manager.create_file_by_raw(
            user_id=self._run_context.user_id,
            tenant_id=self._run_context.tenant_id,
            conversation_id=conversation_id,
            file_binary=file_binary,
            mimetype=mimetype,
            filename=filename,
        )

    @override
    def get_file_generator_by_tool_file_id(self, tool_file_id: str):
        return self._manager.get_file_generator_by_tool_file_id(tool_file_id)


@dataclass(frozen=True, slots=True)
class _WorkflowToolRuntimeSpec:
    provider_type: CoreToolProviderType
    provider_id: str
    tool_name: str
    tool_configurations: dict[str, Any]
    credential_id: str | None = None


@dataclass(frozen=True, slots=True)
class _WorkflowToolRuntimeBinding:
    """Workflow-private runtime state stored inside the opaque graph handle.

    The binding keeps conversation scope in `core.workflow` while `graphon`
    continues to treat the handle as an opaque token.
    """

    tool: Tool
    conversation_id: str | None = None
    parent_trace_context: ParentTraceContext | None = None
    trace_session_id: str | None = None


class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
    def __init__(
        self,
        run_context: Mapping[str, Any] | DifyRunContext,
        session_maker: sessionmaker[Session] | None = None,
        *,
        workflow_id: str | None = None,
        confirmed_guard: ConfirmedToolGuard | None = None,
        confirmed_guard_factory: Callable[[str, str], ConfirmedToolGuard] | None = None,
        execution_driver: WorkflowRunDriver | None = None,
    ) -> None:
        """Both confirmed arguments are ``None`` for every ordinary run; at most one may be set.

        A consented Builder run supplies either a ready, already claimed guard
        (internal/preclaimed use) or a server-owned ``confirmed_guard_factory``.
        Graphon only knows the native node execution ID when it calls
        ``get_runtime``, so the factory is called there, once, with the node ID
        and that actual ID to claim the grant and build the guard. A configured
        factory means confirmed mode even before it is called: nothing falls
        through to ordinary resolution or invocation. Only the guard's exact node
        resolves, and it is consumed with a session that never holds a
        transaction (see ``invoke``). Ordinary node factories supply their
        actual workflow ID for Tool-time draft setup checks; confirmed factories
        never use it, and published child versions retain native delegation.
        """
        if confirmed_guard is not None and confirmed_guard_factory is not None:
            raise BuilderExecutionPolicyError("invalid_confirmed_configuration")
        self._run_context = resolve_dify_run_context(run_context)
        self._file_reference_factory = DifyFileReferenceFactory(self._run_context)
        self._workflow_id = workflow_id
        self._session_maker = session_maker
        self._execution_driver = execution_driver
        self._confirmed_guard = confirmed_guard
        self._confirmed_guard_factory = confirmed_guard_factory
        self._confirmed_guard_factory_called = False

    @property
    def _confirmed(self) -> bool:
        return self._confirmed_guard is not None or self._confirmed_guard_factory is not None

    @property
    def file_reference_factory(self) -> FileReferenceFactoryProtocol:
        return self._file_reference_factory

    @override
    def build_file_reference(self, *, mapping: Mapping[str, Any]):
        return self._file_reference_factory.build_from_mapping(mapping=mapping)

    @override
    def get_runtime(
        self,
        *,
        node_id: str,
        node_data: ToolNodeData,
        variable_pool,
        node_execution_id: str | None = None,
    ) -> ToolRuntimeHandle:
        try:
            guard = self._bind_confirmed_guard(
                node_id=node_id, node_execution_id=node_execution_id, node_data=node_data
            )
            if guard is None:
                tool_runtime = ToolManager.get_workflow_tool_runtime(
                    self._run_context.tenant_id,
                    self._run_context.app_id,
                    node_id,
                    self._build_tool_runtime_spec(node_data),
                    self._run_context.user_id,
                    self._run_context.invoke_from,
                    variable_pool,
                    execution_driver=self._execution_driver,
                    draft_tool_workflow_id=self._workflow_id,
                )
            else:
                self._require_confirmed_node(
                    guard, node_id=node_id, node_execution_id=node_execution_id, node_data=node_data
                )
                tool_runtime = ToolManager.get_workflow_tool_runtime(
                    self._run_context.tenant_id,
                    self._run_context.app_id,
                    node_id,
                    self._build_tool_runtime_spec(node_data),
                    self._run_context.user_id,
                    self._run_context.invoke_from,
                    variable_pool,
                    confirmed_guard=guard,
                )
        except BuilderExecutionPolicyError as exc:
            # Declared refusals carry only their fixed reason code.
            raise ToolRuntimeResolutionError(str(exc)) from exc
        except Exception as exc:
            if self._confirmed:
                # Provider, cache or credential text must not leave the confirmed boundary.
                raise ToolRuntimeResolutionError(
                    str(BuilderExecutionPolicyError("confirmed_tool_resolution_failed"))
                ) from None
            if isinstance(exc, ToolNodeError):
                raise
            raise ToolRuntimeResolutionError(str(exc)) from exc

        conversation_id = (
            None if variable_pool is None else get_system_text(variable_pool, SystemVariableKey.CONVERSATION_ID)
        )
        parent_trace_context: ParentTraceContext | None = None
        trace_session_id: str | None = None
        if self._is_workflow_tool_provider(node_data):
            outer_workflow_run_id = (
                None
                if variable_pool is None
                else get_system_text(variable_pool, SystemVariableKey.WORKFLOW_EXECUTION_ID)
            )
            if isinstance(outer_workflow_run_id, str) and isinstance(node_execution_id, str):
                parent_trace_context = ParentTraceContext(
                    parent_workflow_run_id=outer_workflow_run_id,
                    parent_node_execution_id=node_execution_id,
                )
            if isinstance(self._run_context.trace_session_id, str) and self._run_context.trace_session_id:
                trace_session_id = self._run_context.trace_session_id
        return ToolRuntimeHandle(
            raw=_WorkflowToolRuntimeBinding(
                tool=tool_runtime,
                conversation_id=conversation_id,
                parent_trace_context=parent_trace_context,
                trace_session_id=trace_session_id,
            )
        )

    @override
    def get_runtime_parameters(
        self,
        *,
        tool_runtime: ToolRuntimeHandle,
    ) -> Sequence[ToolRuntimeParameter]:
        tool = self._tool_from_handle(tool_runtime)
        return [
            ToolRuntimeParameter(name=parameter.name, required=parameter.required)
            for parameter in (tool.get_merged_runtime_parameters() or [])
        ]

    @override
    def invoke(
        self,
        *,
        tool_runtime: ToolRuntimeHandle,
        tool_parameters: Mapping[str, Any],
        workflow_call_depth: int,
        provider_name: str,
    ) -> Generator[ToolRuntimeMessage]:
        runtime_binding = self._binding_from_handle(tool_runtime)
        tool = runtime_binding.tool
        if self._confirmed:
            # Before any callback or transport: a configured factory that never bound a
            # guard, or a handle for another tool, is refused rather than run ordinarily.
            guard = self._confirmed_guard
            if guard is None:
                raise self._map_invocation_exception(
                    BuilderExecutionPolicyError("confirmed_guard_unavailable"), provider_name=provider_name
                )
            if getattr(tool, "confirmed_guard", None) is not guard:
                raise self._map_invocation_exception(
                    BuilderExecutionPolicyError("execution_launch_mismatch"), provider_name=provider_name
                )
        callback = DifyWorkflowCallbackHandler()
        if runtime_binding.parent_trace_context and hasattr(tool, "set_parent_trace_context"):
            tool.set_parent_trace_context(
                parent_workflow_run_id=runtime_binding.parent_trace_context.parent_workflow_run_id,
                parent_node_execution_id=runtime_binding.parent_trace_context.parent_node_execution_id,
            )
        elif hasattr(tool, "clear_parent_trace_context"):
            tool.clear_parent_trace_context()
        if runtime_binding.trace_session_id and hasattr(tool, "set_trace_session_id"):
            tool.set_trace_session_id(runtime_binding.trace_session_id)
        elif hasattr(tool, "clear_trace_session_id"):
            tool.clear_trace_session_id()

        try:
            request_metadata = (
                {"app_type": self._run_context.app_type} if self._run_context.app_type is not None else None
            )
            with use_credit_usage_metadata(request_metadata):
                session_maker = self._session_maker or session_factory.get_session_maker()
                if self._confirmed:
                    yield from self._invoke_confirmed(
                        session_maker,
                        tool=tool,
                        tool_parameters=tool_parameters,
                        workflow_call_depth=workflow_call_depth,
                        callback=callback,
                        conversation_id=runtime_binding.conversation_id,
                        provider_name=provider_name,
                    )
                    return
                with session_maker.begin() as session:
                    messages = ToolEngine.generic_invoke(
                        session=session,
                        tool=tool,
                        tool_parameters=dict(tool_parameters),
                        user_id=self._run_context.user_id,
                        workflow_tool_callback=callback,
                        workflow_call_depth=workflow_call_depth,
                        app_id=self._run_context.app_id,
                        conversation_id=runtime_binding.conversation_id,
                    )
                    transformed_messages = ToolFileMessageTransformer.transform_tool_invoke_messages(
                        messages=messages,
                        user_id=self._run_context.user_id,
                        tenant_id=self._run_context.tenant_id,
                        conversation_id=runtime_binding.conversation_id,
                    )
                    yield from self._adapt_messages(transformed_messages, provider_name=provider_name)
        except Exception as exc:
            raise self._map_invocation_exception(exc, provider_name=provider_name) from exc

    def _invoke_confirmed(
        self,
        session_maker: sessionmaker[Session],
        *,
        tool: Tool,
        tool_parameters: Mapping[str, Any],
        workflow_call_depth: int,
        callback: DifyWorkflowCallbackHandler,
        conversation_id: str | None,
        provider_name: str,
    ) -> Generator[ToolRuntimeMessage]:
        """Consume one consented native Tool call through the same native owners.

        Unlike the ordinary ``begin()`` branch the session never starts a
        transaction: the guard's phase claims commit in their own short
        transactions and decryption, refresh and dispatch must run outside SQL.
        Nothing here issues SQL on it; it is re-checked at every native callback.
        The tool itself refuses non-data messages before file transformation.
        """
        with session_maker() as session:
            self._require_no_transaction(session)
            messages = ToolEngine.generic_invoke(
                session=session,
                tool=tool,
                tool_parameters=dict(tool_parameters),
                user_id=self._run_context.user_id,
                workflow_tool_callback=callback,
                workflow_call_depth=workflow_call_depth,
                app_id=self._run_context.app_id,
                conversation_id=conversation_id,
            )
            transformed_messages = ToolFileMessageTransformer.transform_tool_invoke_messages(
                messages=messages,
                user_id=self._run_context.user_id,
                tenant_id=self._run_context.tenant_id,
                conversation_id=conversation_id,
            )
            try:
                for message in self._adapt_messages(transformed_messages, provider_name=provider_name):
                    self._require_no_transaction(session)
                    yield message
                self._require_no_transaction(session)
            finally:
                # Close the native chain deterministically so cancellation reaches the
                # plugin adapter's single outcome callback now; it is never rebuilt.
                transformed_messages.close()
                messages.close()

    def _bind_confirmed_guard(
        self, *, node_id: str, node_execution_id: str | None, node_data: ToolNodeData
    ) -> ConfirmedToolGuard | None:
        """The confirmed guard for this resolution, claimed lazily through the factory at most once.

        Returns ``None`` only for an ordinary runtime. The factory is called with
        Graphon's actual execution ID, its guard is validated and retained before
        any ToolManager resolution, and it is never rebuilt: any later
        resolution through the factory path refuses.
        """
        factory = self._confirmed_guard_factory
        if factory is None:
            return self._confirmed_guard
        if self._confirmed_guard_factory_called:
            raise BuilderExecutionPolicyError("confirmed_runtime_replayed")
        if type(node_execution_id) is not str or not node_execution_id:
            raise BuilderExecutionPolicyError("execution_launch_mismatch")
        if node_data.tool_configurations:
            # Refused before claiming, so a misconfigured node never spends the grant.
            raise BuilderExecutionPolicyError("unsupported_tool_configuration")
        # Spent before the call: a failing, mismatched or replayed claim is never attempted again.
        self._confirmed_guard_factory_called = True
        try:
            guard = factory(node_id, node_execution_id)
        except BuilderExecutionPolicyError:
            raise
        except Exception:
            # Claim/service text may echo stored data.
            raise BuilderExecutionPolicyError("confirmed_guard_unavailable") from None
        if guard is None:
            raise BuilderExecutionPolicyError("confirmed_guard_unavailable")
        binding = guard.binding
        if (
            binding.node_id != node_id
            or binding.native_node_execution_id != node_execution_id
            or binding.context.tenant_id != self._run_context.tenant_id
            or binding.context.app_id != self._run_context.app_id
        ):
            raise BuilderExecutionPolicyError("execution_launch_mismatch")
        self._confirmed_guard = guard
        return guard

    @staticmethod
    def _require_no_transaction(session: Session) -> None:
        if session.in_transaction():
            raise BuilderExecutionPolicyError("confirmed_session_in_transaction")

    def _require_confirmed_node(
        self,
        guard: ConfirmedToolGuard,
        *,
        node_id: str,
        node_execution_id: str | None,
        node_data: ToolNodeData,
    ) -> None:
        """Only the exact claimed node of this run, with no tool configuration, resolves under a guard."""
        binding = guard.binding
        if (
            node_id != binding.node_id
            or node_execution_id is None
            or node_execution_id != binding.native_node_execution_id
            or binding.context.tenant_id != self._run_context.tenant_id
            or binding.context.app_id != self._run_context.app_id
        ):
            raise BuilderExecutionPolicyError("execution_launch_mismatch")
        if node_data.tool_configurations:
            raise BuilderExecutionPolicyError("unsupported_tool_configuration")

    @override
    def get_usage(
        self,
        *,
        tool_runtime: ToolRuntimeHandle,
    ) -> LLMUsage:
        latest = getattr(self._binding_from_handle(tool_runtime).tool, "latest_usage", None)
        if isinstance(latest, LLMUsage):
            return latest
        if isinstance(latest, dict):
            return LLMUsage.model_validate(latest)
        return LLMUsage.empty_usage()

    def resolve_provider_icons(
        self,
        *,
        provider_name: str,
        default_icon: str | None = None,
    ) -> tuple[str | Mapping[str, str] | None, str | Mapping[str, str] | None]:
        icon: str | Mapping[str, str] | None = default_icon
        icon_dark: str | Mapping[str, str] | None = None

        manager = PluginInstaller()
        plugins = manager.list_plugins(self._run_context.tenant_id)
        try:
            current_plugin = next(plugin for plugin in plugins if f"{plugin.plugin_id}/{plugin.name}" == provider_name)
            icon = current_plugin.declaration.icon
        except StopIteration:
            pass

        try:
            builtin_tool = next(
                provider
                for provider in BuiltinToolManageService.list_builtin_tools(
                    self._run_context.user_id,
                    self._run_context.tenant_id,
                )
                if provider.name == provider_name
            )
            icon = builtin_tool.icon
            icon_dark = builtin_tool.icon_dark
        except StopIteration:
            pass

        return icon, icon_dark

    @staticmethod
    def _tool_from_handle(tool_runtime: ToolRuntimeHandle) -> Tool:
        return DifyToolNodeRuntime._binding_from_handle(tool_runtime).tool

    @staticmethod
    def _binding_from_handle(tool_runtime: ToolRuntimeHandle) -> _WorkflowToolRuntimeBinding:
        if isinstance(tool_runtime.raw, _WorkflowToolRuntimeBinding):
            return tool_runtime.raw
        return _WorkflowToolRuntimeBinding(tool=cast("Tool", tool_runtime.raw))

    @staticmethod
    def _build_tool_runtime_spec(node_data: ToolNodeData) -> _WorkflowToolRuntimeSpec:
        tool_configurations = dict(node_data.tool_configurations)
        tool_configurations.update(
            {name: tool_input.model_dump(mode="python") for name, tool_input in node_data.tool_parameters.items()}
        )
        return _WorkflowToolRuntimeSpec(
            provider_type=CoreToolProviderType(node_data.provider_type.value),
            provider_id=node_data.provider_id,
            tool_name=node_data.tool_name,
            tool_configurations=tool_configurations,
            credential_id=node_data.credential_id,
        )

    @staticmethod
    def _is_workflow_tool_provider(node_data: ToolNodeData) -> bool:
        return node_data.provider_type.value == CoreToolProviderType.WORKFLOW.value

    def _adapt_messages(
        self,
        messages: Generator[CoreToolInvokeMessage],
        *,
        provider_name: str,
    ) -> Generator[ToolRuntimeMessage]:
        try:
            for message in messages:
                yield self._convert_message(message)
        except Exception as exc:
            raise self._map_invocation_exception(exc, provider_name=provider_name) from exc

    def _convert_message(self, message: CoreToolInvokeMessage) -> ToolRuntimeMessage:
        graph_message_type = ToolRuntimeMessage.MessageType(message.type.value)
        graph_message = self._convert_message_payload(message.message)
        graph_meta = message.meta.copy() if message.meta is not None else None
        return ToolRuntimeMessage(type=graph_message_type, message=graph_message, meta=graph_meta)

    def _convert_message_payload(
        self,
        message: CoreToolInvokeMessage.TextMessage
        | CoreToolInvokeMessage.JsonMessage
        | CoreToolInvokeMessage.BlobChunkMessage
        | CoreToolInvokeMessage.BlobMessage
        | CoreToolInvokeMessage.LogMessage
        | CoreToolInvokeMessage.FileMessage
        | CoreToolInvokeMessage.VariableMessage
        | CoreToolInvokeMessage.RetrieverResourceMessage
        | None,
    ) -> (
        ToolRuntimeMessage.TextMessage
        | ToolRuntimeMessage.JsonMessage
        | ToolRuntimeMessage.BlobChunkMessage
        | ToolRuntimeMessage.BlobMessage
        | ToolRuntimeMessage.LogMessage
        | ToolRuntimeMessage.FileMessage
        | ToolRuntimeMessage.VariableMessage
        | ToolRuntimeMessage.RetrieverResourceMessage
        | None
    ):
        if message is None:
            return None

        from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage

        match message:
            case CoreToolInvokeMessage.TextMessage():
                return ToolRuntimeMessage.TextMessage(text=message.text)
            case CoreToolInvokeMessage.JsonMessage():
                return ToolRuntimeMessage.JsonMessage(
                    json_object=message.json_object,
                    suppress_output=message.suppress_output,
                )
            case CoreToolInvokeMessage.BlobMessage():
                return ToolRuntimeMessage.BlobMessage(blob=message.blob)
            case CoreToolInvokeMessage.BlobChunkMessage():
                return ToolRuntimeMessage.BlobChunkMessage(
                    id=message.id,
                    sequence=message.sequence,
                    total_length=message.total_length,
                    blob=message.blob,
                    end=message.end,
                )
            case CoreToolInvokeMessage.FileMessage():
                return ToolRuntimeMessage.FileMessage(file_marker=message.file_marker)
            case CoreToolInvokeMessage.VariableMessage():
                return ToolRuntimeMessage.VariableMessage(
                    variable_name=message.variable_name,
                    variable_value=message.variable_value,
                    stream=message.stream,
                )
            case CoreToolInvokeMessage.LogMessage():
                return ToolRuntimeMessage.LogMessage(
                    id=message.id,
                    label=message.label,
                    parent_id=message.parent_id,
                    error=message.error,
                    status=ToolRuntimeMessage.LogMessage.LogStatus(message.status.value),
                    data=dict(message.data),
                    metadata=dict(message.metadata),
                )
            case CoreToolInvokeMessage.RetrieverResourceMessage():
                retriever_resources = [
                    resource.model_dump() if hasattr(resource, "model_dump") else dict(resource)
                    for resource in message.retriever_resources
                ]
                return ToolRuntimeMessage.RetrieverResourceMessage(
                    retriever_resources=retriever_resources,
                    context=message.context,
                )
            case _:
                raise TypeError(f"unsupported tool message payload: {type(message).__name__}")

    @staticmethod
    def _map_invocation_exception(exc: Exception, *, provider_name: str) -> ToolNodeError:
        match exc:
            case ToolNodeError():
                return exc
            case PluginInvokeError():
                return ToolRuntimeInvocationError(exc.to_user_friendly_error(plugin_name=provider_name))
            case PluginDaemonClientSideError():
                return ToolRuntimeInvocationError(f"Failed to invoke tool, error: {exc.description}")
            case ToolInvokeError():
                return ToolRuntimeInvocationError(f"Failed to invoke tool {provider_name}: {exc}")
            case _:
                return ToolRuntimeInvocationError(str(exc))


class DifyHumanInputNodeRuntime:
    def __init__(
        self,
        run_context: Mapping[str, Any] | DifyRunContext,
        *,
        workflow_execution_id_getter: Callable[[], str | None] | None = None,
        conversation_id_getter: Callable[[], str | None] | None = None,
        form_repository: HumanInputFormRepository | None = None,
    ) -> None:
        self._run_context = resolve_dify_run_context(run_context)
        self._workflow_execution_id_getter = workflow_execution_id_getter
        self._conversation_id_getter = conversation_id_getter
        self._form_repository = form_repository
        self._file_reference_factory = DifyFileReferenceFactory(self._run_context)

    def _invoke_source(self) -> str:
        invoke_from = self._run_context.invoke_from
        if isinstance(invoke_from, str):
            return invoke_from
        if isinstance(invoke_from, Enum):
            return str(invoke_from.value)
        return str(invoke_from)

    def _resolve_delivery_methods(self, *, node_data: HumanInputNodeData) -> Sequence[DeliveryChannelConfig]:
        invoke_source = self._invoke_source()
        methods = [method for method in parse_human_input_delivery_methods(node_data) if method.enabled]
        if invoke_source in {"debugger", "explore"}:
            methods = [method for method in methods if method.type != DeliveryMethodType.WEBAPP]
        return [
            apply_dify_debug_email_recipient(
                method,
                enabled=invoke_source == "debugger",
                actor_id=self._run_context.user_id,
            )
            for method in methods
        ]

    def _display_in_ui(self, *, node_data: HumanInputNodeData) -> bool:
        if self._invoke_source() == "debugger":
            return True
        return is_human_input_webapp_enabled(node_data)

    def build_form_repository(self) -> HumanInputFormRepository:
        if self._form_repository is not None:
            return self._form_repository

        return self._build_form_repository()

    def _build_form_repository(self) -> HumanInputFormRepository:
        invoke_source = self._invoke_source()
        return HumanInputFormRepositoryImpl(
            tenant_id=self._run_context.tenant_id,
            app_id=self._run_context.app_id,
            workflow_execution_id=self._workflow_execution_id_getter() if self._workflow_execution_id_getter else None,
            invoke_source=invoke_source,
            submission_actor_id=self._run_context.user_id if invoke_source in {"debugger", "explore"} else None,
        )

    def with_form_repository(self, form_repository: HumanInputFormRepository) -> DifyHumanInputNodeRuntime:
        return DifyHumanInputNodeRuntime(
            self._run_context,
            workflow_execution_id_getter=self._workflow_execution_id_getter,
            conversation_id_getter=self._conversation_id_getter,
            form_repository=form_repository,
        )

    def get_form(self, *, node_id: str) -> HumanInputFormEntity | None:
        repo = self.build_form_repository()
        return repo.get_form(node_id)

    def restore_submitted_data(
        self,
        *,
        inputs: Sequence[FormInputConfig],
        submitted_data: Mapping[str, Any],
    ) -> Mapping[str, Any]:
        restored_data: dict[str, Any] = dict(submitted_data)
        for input_config in inputs:
            output_variable_name = input_config.output_variable_name
            if output_variable_name not in submitted_data:
                continue
            restored_data[output_variable_name] = self._restore_submitted_value(
                input_config=input_config,
                value=submitted_data[output_variable_name],
            )
        return restored_data

    def create_form(
        self,
        *,
        node_id: str,
        node_data: HumanInputNodeData,
        rendered_content: str,
        resolved_default_values: Mapping[str, Any],
    ) -> HumanInputFormEntity:
        repo = self.build_form_repository()
        params = FormCreateParams(
            workflow_execution_id=self._workflow_execution_id_getter() if self._workflow_execution_id_getter else None,
            # A chatflow (advanced-chat) run carries a conversation; tag the form with
            # it too so it is queryable per conversation. None for a pure workflow run.
            conversation_id=self._conversation_id_getter() if self._conversation_id_getter else None,
            node_id=node_id,
            form_config=node_data,
            rendered_content=rendered_content,
            delivery_methods=self._resolve_delivery_methods(node_data=node_data),
            display_in_ui=self._display_in_ui(node_data=node_data),
            resolved_default_values=resolved_default_values,
        )
        return repo.create_form(params)

    def _restore_submitted_value(
        self,
        *,
        input_config: FormInputConfig,
        value: Any,
    ) -> Any:
        if isinstance(input_config, FileInputConfig):
            return self._restore_submitted_file_value(
                output_variable_name=input_config.output_variable_name,
                value=value,
            )
        if isinstance(input_config, FileListInputConfig):
            return self._restore_submitted_file_list_value(
                output_variable_name=input_config.output_variable_name,
                value=value,
            )
        return value

    def _restore_submitted_file_value(
        self,
        *,
        output_variable_name: str,
        value: Any,
    ) -> Any:
        if not isinstance(value, Mapping):
            msg = (
                "HumanInput file submission must be persisted as a mapping, "
                f"output_variable_name={output_variable_name}"
            )
            raise ValueError(msg)
        return self._file_reference_factory.build_from_mapping(mapping=value)

    def _restore_submitted_file_list_value(
        self,
        *,
        output_variable_name: str,
        value: Any,
    ) -> list[Any]:
        if not isinstance(value, list):
            msg = (
                "HumanInput file-list submission must be persisted as a list, "
                f"output_variable_name={output_variable_name}"
            )
            raise ValueError(msg)
        if any(not isinstance(item, Mapping) for item in value):
            msg = f"HumanInput file-list submission must contain mappings, output_variable_name={output_variable_name}"
            raise ValueError(msg)
        return [self._file_reference_factory.build_from_mapping(mapping=item) for item in value]


def build_dify_llm_file_saver(
    *,
    run_context: Mapping[str, Any] | DifyRunContext,
    http_client: HttpClientProtocol,
    conversation_id_getter: Callable[[], str | None] | None = None,
) -> LLMFileSaver:
    from graphon.nodes.llm.file_saver import FileSaverImpl

    return FileSaverImpl(
        tool_file_manager=DifyToolFileManager(run_context, conversation_id_getter=conversation_id_getter),
        file_reference_factory=DifyFileReferenceFactory(run_context),
        http_client=http_client,
    )