fix(workflow): simplify selection context menu positioning

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.github/CODEOWNERS

@@ -36,7 +36,7 @@
 /api/core/workflow/graph/ @laipz8200 @QuantumGhost
 /api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
 /api/core/workflow/node_events/ @laipz8200 @QuantumGhost
-/api/dify_graph/model_runtime/ @laipz8200 @QuantumGhost
+/api/graphon/model_runtime/ @laipz8200 @WH-2099
 
 # Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
 /api/core/workflow/nodes/agent/ @Nov1c444

api/.importlinter

@@ -1,10 +1,14 @@
 [importlinter]
 root_packages =
     core
-    dify_graph
+    constants
+    context
+    graphon
     configs
     controllers
     extensions
+    factories
+    libs
     models
     tasks
     services
@@ -22,40 +26,30 @@ layers =
     runtime
     entities
 containers =
-    dify_graph
+    graphon
 ignore_imports =
-    dify_graph.nodes.base.node -> dify_graph.graph_events
-    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_events
-    dify_graph.nodes.loop.loop_node -> dify_graph.graph_events
+    graphon.nodes.base.node -> graphon.graph_events
+    graphon.nodes.iteration.iteration_node -> graphon.graph_events
+    graphon.nodes.loop.loop_node -> graphon.graph_events
 
-    dify_graph.nodes.iteration.iteration_node -> dify_graph.graph_engine
-    dify_graph.nodes.loop.loop_node -> dify_graph.graph_engine
+    graphon.nodes.iteration.iteration_node -> graphon.graph_engine
+    graphon.nodes.loop.loop_node -> graphon.graph_engine
     # TODO(QuantumGhost): fix the import violation later
-    dify_graph.entities.pause_reason -> dify_graph.nodes.human_input.entities
-
-[importlinter:contract:workflow-infrastructure-dependencies]
-name = Workflow Infrastructure Dependencies
-type = forbidden
-source_modules =
-    dify_graph
-forbidden_modules =
-    extensions.ext_database
-    extensions.ext_redis
-allow_indirect_imports = True
-ignore_imports =
-    dify_graph.nodes.llm.node -> extensions.ext_database
-    dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
-    dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
+    graphon.entities.pause_reason -> graphon.nodes.human_input.entities
 
 [importlinter:contract:workflow-external-imports]
 name = Workflow External Imports
 type = forbidden
 source_modules =
-    dify_graph
+    graphon
 forbidden_modules =
+    constants
     configs
+    context
     controllers
     extensions
+    factories
+    libs
     models
     services
     tasks
@@ -88,46 +82,14 @@ forbidden_modules =
     core.tools
     core.trigger
     core.variables
-ignore_imports =
-    dify_graph.nodes.llm.llm_utils -> core.model_manager
-    dify_graph.nodes.llm.protocols -> core.model_manager
-    dify_graph.nodes.llm.llm_utils -> dify_graph.model_runtime.model_providers.__base.large_language_model
-    dify_graph.nodes.llm.node -> core.tools.signature
-    dify_graph.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
-    dify_graph.nodes.tool.tool_node -> core.tools.tool_engine
-    dify_graph.nodes.tool.tool_node -> core.tools.tool_manager
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> dify_graph.model_runtime.model_providers.__base.large_language_model
-    dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
-    dify_graph.nodes.question_classifier.question_classifier_node -> core.model_manager
-    dify_graph.nodes.tool.tool_node -> core.tools.utils.message_transformer
-    dify_graph.nodes.llm.node -> core.llm_generator.output_parser.errors
-    dify_graph.nodes.llm.node -> core.llm_generator.output_parser.structured_output
-    dify_graph.nodes.llm.node -> core.model_manager
-    dify_graph.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
-    dify_graph.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
-    dify_graph.nodes.llm.node -> core.prompt.utils.prompt_message_util
-    dify_graph.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
-    dify_graph.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
-    dify_graph.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
-    dify_graph.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
-    dify_graph.nodes.llm.node -> models.dataset
-    dify_graph.nodes.llm.file_saver -> core.tools.signature
-    dify_graph.nodes.llm.file_saver -> core.tools.tool_file_manager
-    dify_graph.nodes.tool.tool_node -> core.tools.errors
-    dify_graph.nodes.llm.node -> extensions.ext_database
-    dify_graph.nodes.llm.node -> models.model
-    dify_graph.nodes.tool.tool_node -> services
-    dify_graph.model_runtime.model_providers.__base.ai_model -> configs
-    dify_graph.model_runtime.model_providers.__base.ai_model -> extensions.ext_redis
-    dify_graph.model_runtime.model_providers.__base.large_language_model -> configs
-    dify_graph.model_runtime.model_providers.__base.text_embedding_model -> core.entities.embedding_type
-    dify_graph.model_runtime.model_providers.model_provider_factory -> configs
-    dify_graph.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
-    dify_graph.model_runtime.model_providers.model_provider_factory -> models.provider_ids
+
+[importlinter:contract:workflow-third-party-imports]
+name = Workflow Third-Party Imports
+type = forbidden
+source_modules =
+    graphon
+forbidden_modules =
+    sqlalchemy
 
 [importlinter:contract:rsc]
 name = RSC
@@ -136,7 +98,7 @@ layers =
     graph_engine
     response_coordinator
 containers =
-    dify_graph.graph_engine
+    graphon.graph_engine
 
 [importlinter:contract:worker]
 name = Worker
@@ -145,7 +107,7 @@ layers =
     graph_engine
     worker
 containers =
-    dify_graph.graph_engine
+    graphon.graph_engine
 
 [importlinter:contract:graph-engine-architecture]
 name = Graph Engine Architecture
@@ -161,28 +123,28 @@ layers =
     worker_management
     domain
 containers =
-    dify_graph.graph_engine
+    graphon.graph_engine
 
 [importlinter:contract:domain-isolation]
 name = Domain Model Isolation
 type = forbidden
 source_modules =
-    dify_graph.graph_engine.domain
+    graphon.graph_engine.domain
 forbidden_modules =
-    dify_graph.graph_engine.worker_management
-    dify_graph.graph_engine.command_channels
-    dify_graph.graph_engine.layers
-    dify_graph.graph_engine.protocols
+    graphon.graph_engine.worker_management
+    graphon.graph_engine.command_channels
+    graphon.graph_engine.layers
+    graphon.graph_engine.protocols
 
 [importlinter:contract:worker-management]
 name = Worker Management
 type = forbidden
 source_modules =
-    dify_graph.graph_engine.worker_management
+    graphon.graph_engine.worker_management
 forbidden_modules =
-    dify_graph.graph_engine.orchestration
-    dify_graph.graph_engine.command_processing
-    dify_graph.graph_engine.event_management
+    graphon.graph_engine.orchestration
+    graphon.graph_engine.command_processing
+    graphon.graph_engine.event_management
 
 
 [importlinter:contract:graph-traversal-components]
@@ -192,11 +154,11 @@ layers =
     edge_processor
     skip_propagator
 containers =
-    dify_graph.graph_engine.graph_traversal
+    graphon.graph_engine.graph_traversal
 
 [importlinter:contract:command-channels]
 name = Command Channels Independence
 type = independence
 modules =
-    dify_graph.graph_engine.command_channels.in_memory_channel
-    dify_graph.graph_engine.command_channels.redis_channel
+    graphon.graph_engine.command_channels.in_memory_channel
+    graphon.graph_engine.command_channels.redis_channel

api/.ruff.toml

@@ -100,7 +100,7 @@ ignore = [
 "configs/*" = [
     "N802", # invalid-function-name
 ]
-"dify_graph/model_runtime/callbacks/base_callback.py" = ["T201"]
+"graphon/model_runtime/callbacks/base_callback.py" = ["T201"]
 "core/workflow/callbacks/workflow_logging_callback.py" = ["T201"]
 "libs/gmpy2_pkcs10aep_cipher.py" = [
     "N803", # invalid-argument-name

api/context/__init__.py

@@ -1,74 +1,36 @@
 """
-Core Context - Framework-agnostic context management.
+Application-layer context adapters.
 
-This module provides context management that is independent of any specific
-web framework. Framework-specific implementations register their context
-capture functions at application initialization time.
-
-This ensures the workflow layer remains completely decoupled from Flask
-or any other web framework.
+Concrete execution-context implementations live here so `graphon` only
+depends on injected context managers rather than framework state capture.
 """
 
-import contextvars
-from collections.abc import Callable
-
-from dify_graph.context.execution_context import (
+from context.execution_context import (
+    AppContext,
+    ContextProviderNotFoundError,
     ExecutionContext,
+    ExecutionContextBuilder,
     IExecutionContext,
     NullAppContext,
+    capture_current_context,
+    read_context,
+    register_context,
+    register_context_capturer,
+    reset_context_provider,
 )
-
-# Global capturer function - set by framework-specific modules
-_capturer: Callable[[], IExecutionContext] | None = None
-
-
-def register_context_capturer(capturer: Callable[[], IExecutionContext]) -> None:
-    """
-    Register a context capture function.
-
-    This should be called by framework-specific modules (e.g., Flask)
-    during application initialization.
-
-    Args:
-        capturer: Function that captures current context and returns IExecutionContext
-    """
-    global _capturer
-    _capturer = capturer
-
-
-def capture_current_context() -> IExecutionContext:
-    """
-    Capture current execution context.
-
-    This function uses the registered context capturer. If no capturer
-    is registered, it returns a minimal context with only contextvars
-    (suitable for non-framework environments like tests or standalone scripts).
-
-    Returns:
-        IExecutionContext with captured context
-    """
-    if _capturer is None:
-        # No framework registered - return minimal context
-        return ExecutionContext(
-            app_context=NullAppContext(),
-            context_vars=contextvars.copy_context(),
-        )
-
-    return _capturer()
-
-
-def reset_context_provider() -> None:
-    """
-    Reset the context capturer.
-
-    This is primarily useful for testing to ensure a clean state.
-    """
-    global _capturer
-    _capturer = None
-
+from context.models import SandboxContext
 
 __all__ = [
+    "AppContext",
+    "ContextProviderNotFoundError",
+    "ExecutionContext",
+    "ExecutionContextBuilder",
+    "IExecutionContext",
+    "NullAppContext",
+    "SandboxContext",
     "capture_current_context",
+    "read_context",
+    "register_context",
     "register_context_capturer",
     "reset_context_provider",
 ]

api/context/execution_context.py

@@ -1,5 +1,8 @@
 """
-Execution Context - Abstracted context management for workflow execution.
+Application-layer execution context adapters.
+
+Concrete context capture lives outside `graphon` so the graph package only
+consumes injected context managers when it needs to preserve thread-local state.
 """
 
 import contextvars
@@ -16,33 +19,33 @@ class AppContext(ABC):
     """
     Abstract application context interface.
 
-    This abstraction allows workflow execution to work with or without Flask
-    by providing a common interface for application context management.
+    Application adapters can implement this to restore framework-specific state
+    such as Flask app context around worker execution.
     """
 
     @abstractmethod
     def get_config(self, key: str, default: Any = None) -> Any:
         """Get configuration value by key."""
-        pass
+        raise NotImplementedError
 
     @abstractmethod
     def get_extension(self, name: str) -> Any:
-        """Get Flask extension by name (e.g., 'db', 'cache')."""
-        pass
+        """Get application extension by name."""
+        raise NotImplementedError
 
     @abstractmethod
     def enter(self) -> AbstractContextManager[None]:
         """Enter the application context."""
-        pass
+        raise NotImplementedError
 
 
 @runtime_checkable
 class IExecutionContext(Protocol):
     """
-    Protocol for execution context.
+    Protocol for enterable execution context objects.
 
-    This protocol defines the interface that all execution contexts must implement,
-    allowing both ExecutionContext and FlaskExecutionContext to be used interchangeably.
+    Concrete implementations may carry extra framework state, but callers only
+    depend on standard context-manager behavior plus optional user metadata.
     """
 
     def __enter__(self) -> "IExecutionContext":
@@ -62,14 +65,10 @@ class IExecutionContext(Protocol):
 @final
 class ExecutionContext:
     """
-    Execution context for workflow execution in worker threads.
+    Generic execution context used by application-layer adapters.
 
-    This class encapsulates all context needed for workflow execution:
-    - Application context (Flask app or standalone)
-    - Context variables for Python contextvars
-    - User information (optional)
-
-    It is designed to be serializable and passable to worker threads.
+    It restores captured `contextvars` and optionally enters an application
+    context before the worker executes graph logic.
     """
 
     def __init__(
@@ -78,14 +77,6 @@ class ExecutionContext:
         context_vars: contextvars.Context | None = None,
         user: Any = None,
     ) -> None:
-        """
-        Initialize execution context.
-
-        Args:
-            app_context: Application context (Flask or standalone)
-            context_vars: Python contextvars to preserve
-            user: User object (optional)
-        """
         self._app_context = app_context
         self._context_vars = context_vars
         self._user = user
@@ -98,27 +89,21 @@ class ExecutionContext:
 
     @property
     def context_vars(self) -> contextvars.Context | None:
-        """Get context variables."""
+        """Get captured context variables."""
         return self._context_vars
 
     @property
     def user(self) -> Any:
-        """Get user object."""
+        """Get captured user object."""
         return self._user
 
     @contextmanager
     def enter(self) -> Generator[None, None, None]:
-        """
-        Enter this execution context.
-
-        This is a convenience method that creates a context manager.
-        """
-        # Restore context variables if provided
+        """Enter this execution context."""
         if self._context_vars:
             for var, val in self._context_vars.items():
                 var.set(val)
 
-        # Enter app context if available
         if self._app_context is not None:
             with self._app_context.enter():
                 yield
@@ -141,18 +126,10 @@ class ExecutionContext:
 
 class NullAppContext(AppContext):
     """
-    Null implementation of AppContext for non-Flask environments.
-
-    This is used when running without Flask (e.g., in tests or standalone mode).
+    Null application context for non-framework environments.
     """
 
     def __init__(self, config: dict[str, Any] | None = None) -> None:
-        """
-        Initialize null app context.
-
-        Args:
-            config: Optional configuration dictionary
-        """
         self._config = config or {}
         self._extensions: dict[str, Any] = {}
 
@@ -165,7 +142,7 @@ class NullAppContext(AppContext):
         return self._extensions.get(name)
 
     def set_extension(self, name: str, extension: Any) -> None:
-        """Set extension by name."""
+        """Register an extension for tests or standalone execution."""
         self._extensions[name] = extension
 
     @contextmanager
@@ -176,9 +153,7 @@ class NullAppContext(AppContext):
 
 class ExecutionContextBuilder:
     """
-    Builder for creating ExecutionContext instances.
-
-    This provides a fluent API for building execution contexts.
+    Builder for creating `ExecutionContext` instances.
     """
 
     def __init__(self) -> None:
@@ -211,63 +186,42 @@ class ExecutionContextBuilder:
 
 
 _capturer: Callable[[], IExecutionContext] | None = None
-
-# Tenant-scoped providers using tuple keys for clarity and constant-time lookup.
-# Key mapping:
-#   (name, tenant_id) -> provider
-# - name: namespaced identifier (recommend prefixing, e.g. "workflow.sandbox")
-# - tenant_id: tenant identifier string
-# Value:
-#   provider: Callable[[], BaseModel] returning the typed context value
-# Type-safety note:
-#   - This registry cannot enforce that all providers for a given name return the same BaseModel type.
-#   - Implementors SHOULD provide typed wrappers around register/read (like Go's context best practice),
-#     e.g. def register_sandbox_ctx(tenant_id: str, p: Callable[[], SandboxContext]) and
-#          def read_sandbox_ctx(tenant_id: str) -> SandboxContext.
 _tenant_context_providers: dict[tuple[str, str], Callable[[], BaseModel]] = {}
 
 T = TypeVar("T", bound=BaseModel)
 
 
 class ContextProviderNotFoundError(KeyError):
-    """Raised when a tenant-scoped context provider is missing for a given (name, tenant_id)."""
+    """Raised when a tenant-scoped context provider is missing."""
 
     pass
 
 
 def register_context_capturer(capturer: Callable[[], IExecutionContext]) -> None:
-    """Register a single enterable execution context capturer (e.g., Flask)."""
+    """Register an enterable execution context capturer."""
     global _capturer
     _capturer = capturer
 
 
 def register_context(name: str, tenant_id: str, provider: Callable[[], BaseModel]) -> None:
-    """Register a tenant-specific provider for a named context.
-
-    Tip: use a namespaced "name" (e.g., "workflow.sandbox") to avoid key collisions.
-    Consider adding a typed wrapper for this registration in your feature module.
-    """
+    """Register a tenant-specific provider for a named context."""
     _tenant_context_providers[(name, tenant_id)] = provider
 
 
 def read_context(name: str, *, tenant_id: str) -> BaseModel:
-    """
-    Read a context value for a specific tenant.
-
-    Raises KeyError if the provider for (name, tenant_id) is not registered.
-    """
-    prov = _tenant_context_providers.get((name, tenant_id))
-    if prov is None:
+    """Read a context value for a specific tenant."""
+    provider = _tenant_context_providers.get((name, tenant_id))
+    if provider is None:
         raise ContextProviderNotFoundError(f"Context provider '{name}' not registered for tenant '{tenant_id}'")
-    return prov()
+    return provider()
 
 
 def capture_current_context() -> IExecutionContext:
     """
     Capture current execution context from the calling environment.
 
-    If a capturer is registered (e.g., Flask), use it. Otherwise, return a minimal
-    context with NullAppContext + copy of current contextvars.
+    If no framework adapter is registered, return a minimal context that only
+    restores `contextvars`.
     """
     if _capturer is None:
         return ExecutionContext(
@@ -278,7 +232,22 @@ def capture_current_context() -> IExecutionContext:
 
 
 def reset_context_provider() -> None:
-    """Reset the capturer and all tenant-scoped context providers (primarily for tests)."""
+    """Reset the capturer and tenant-scoped providers."""
     global _capturer
     _capturer = None
     _tenant_context_providers.clear()
+
+
+__all__ = [
+    "AppContext",
+    "ContextProviderNotFoundError",
+    "ExecutionContext",
+    "ExecutionContextBuilder",
+    "IExecutionContext",
+    "NullAppContext",
+    "capture_current_context",
+    "read_context",
+    "register_context",
+    "register_context_capturer",
+    "reset_context_provider",
+]

api/context/flask_app_context.py

@@ -10,11 +10,7 @@ from typing import Any, final
 
 from flask import Flask, current_app, g
 
-from dify_graph.context import register_context_capturer
-from dify_graph.context.execution_context import (
-    AppContext,
-    IExecutionContext,
-)
+from context.execution_context import AppContext, IExecutionContext, register_context_capturer
 
 
 @final

api/contexts/__init__.py

@@ -6,7 +6,6 @@ from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
     from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
-    from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
     from core.trigger.provider import PluginTriggerProviderController
 
@@ -20,14 +19,6 @@ plugin_tool_providers: RecyclableContextVar[dict[str, "PluginToolProviderControl
 
 plugin_tool_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(ContextVar("plugin_tool_providers_lock"))
 
-plugin_model_providers: RecyclableContextVar[list["PluginModelProviderEntity"] | None] = RecyclableContextVar(
-    ContextVar("plugin_model_providers")
-)
-
-plugin_model_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
-    ContextVar("plugin_model_providers_lock")
-)
-
 datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
     RecyclableContextVar(ContextVar("datasource_plugin_providers"))
 )

api/controllers/common/fields.py

@@ -4,7 +4,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, computed_field
 
-from dify_graph.file import helpers as file_helpers
+from graphon.file import helpers as file_helpers
 from models.model import IconType
 
 JSONValue: TypeAlias = str | int | float | bool | None | dict[str, Any] | list[Any]

api/controllers/console/app/app.py

@@ -26,9 +26,9 @@ from controllers.console.wraps import (
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.trigger.constants import TRIGGER_NODE_TYPES
-from dify_graph.enums import WorkflowExecutionStatus
-from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
+from graphon.enums import WorkflowExecutionStatus
+from graphon.file import helpers as file_helpers
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType

api/controllers/console/app/audio.py

@@ -22,7 +22,7 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import login_required
 from models import App, AppMode
 from services.audio_service import AudioService

api/controllers/console/app/completion.py

@@ -26,7 +26,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import current_user, login_required

api/controllers/console/app/generator.py

@@ -18,8 +18,8 @@ from core.helper.code_executor.javascript.javascript_code_provider import Javasc
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
 from core.llm_generator.llm_generator import LLMGenerator
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import current_account_with_tenant, login_required
 from models import App
 from services.workflow_service import WorkflowService

api/controllers/console/app/message.py

@@ -24,9 +24,9 @@ from controllers.console.wraps import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from fields.raws import FilesContainedField
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs.helper import TimestampField, uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required

api/controllers/console/app/model_config.py

@@ -88,6 +88,7 @@ class ModelConfigResource(Resource):
                         tenant_id=current_tenant_id,
                         app_id=app_model.id,
                         agent_tool=agent_tool_entity,
+                        user_id=current_user.id,
                     )
                     manager = ToolParameterConfigurationManager(
                         tenant_id=current_tenant_id,
@@ -127,6 +128,7 @@ class ModelConfigResource(Resource):
                             tenant_id=current_tenant_id,
                             app_id=app_model.id,
                             agent_tool=agent_tool_entity,
+                            user_id=current_user.id,
                         )
                     except Exception:
                         continue

api/controllers/console/app/workflow.py

@@ -20,6 +20,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.file_access import DatabaseFileAccessController
 from core.helper.trace_id_helper import get_external_trace_id
 from core.plugin.impl.exc import PluginInvokeError
 from core.trigger.constants import TRIGGER_SCHEDULE_NODE_TYPE
@@ -29,15 +30,15 @@ from core.trigger.debug.event_selectors import (
     create_event_poller,
     select_trigger_debug_events,
 )
-from dify_graph.enums import NodeType
-from dify_graph.file.models import File
-from dify_graph.graph_engine.manager import GraphEngineManager
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
 from fields.member_fields import simple_account_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
+from graphon.enums import NodeType
+from graphon.file.models import File
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
 from libs.datetime_utils import naive_utc_now
 from libs.helper import TimestampField, uuid_value
@@ -51,6 +52,7 @@ from services.errors.llm import InvokeRateLimitError
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
+_file_access_controller = DatabaseFileAccessController()
 LISTENING_RETRY_IN = 2000
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE = "source workflow must be published"
@@ -204,6 +206,7 @@ def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence
         mappings=files,
         tenant_id=workflow.tenant_id,
         config=file_extra_config,
+        access_controller=_file_access_controller,
     )
     return file_objs
 

api/controllers/console/app/workflow_app_log.py

@@ -9,12 +9,12 @@ from sqlalchemy.orm import Session
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from dify_graph.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import (
     build_workflow_app_log_pagination_model,
     build_workflow_archived_log_pagination_model,
 )
+from graphon.enums import WorkflowExecutionStatus
 from libs.login import login_required
 from models import App
 from models.model import AppMode

api/controllers/console/app/workflow_draft_variable.py

@@ -15,14 +15,15 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
-from dify_graph.file import helpers as file_helpers
-from dify_graph.variables.segment_group import SegmentGroup
-from dify_graph.variables.segments import ArrayFileSegment, FileSegment, Segment
-from dify_graph.variables.types import SegmentType
+from core.app.file_access import DatabaseFileAccessController
+from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
+from graphon.file import helpers as file_helpers
+from graphon.variables.segment_group import SegmentGroup
+from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
+from graphon.variables.types import SegmentType
 from libs.login import current_user, login_required
 from models import App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -30,6 +31,7 @@ from services.workflow_draft_variable_service import WorkflowDraftVariableList,
 from services.workflow_service import WorkflowService
 
 logger = logging.getLogger(__name__)
+_file_access_controller = DatabaseFileAccessController()
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
@@ -389,13 +391,21 @@ class VariableApi(Resource):
             if variable.value_type == SegmentType.FILE:
                 if not isinstance(raw_value, dict):
                     raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                raw_value = build_from_mapping(mapping=raw_value, tenant_id=app_model.tenant_id)
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=app_model.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             elif variable.value_type == SegmentType.ARRAY_FILE:
                 if not isinstance(raw_value, list):
                     raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=app_model.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

api/controllers/console/app/workflow_run.py

@@ -12,8 +12,7 @@ from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import NotFoundError
-from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.enums import WorkflowExecutionStatus
+from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id
 from extensions.ext_database import db
 from fields.end_user_fields import simple_end_user_fields
 from fields.member_fields import simple_account_fields
@@ -27,6 +26,8 @@ from fields.workflow_run_fields import (
     workflow_run_node_execution_list_fields,
     workflow_run_pagination_fields,
 )
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.enums import WorkflowExecutionStatus
 from libs.archive_storage import ArchiveStorageNotConfiguredError, get_archive_storage
 from libs.custom_inputs import time_duration
 from libs.helper import uuid_value
@@ -496,6 +497,9 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
 
         pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
         pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
+        form_tokens_by_form_id = _load_form_tokens_by_form_id(
+            [reason.form_id for reason in pause_reasons if isinstance(reason, HumanInputRequired)]
+        )
 
         # Build response
         paused_at = pause_entity.paused_at if pause_entity else None
@@ -514,7 +518,9 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
                         "pause_type": {
                             "type": "human_input",
                             "form_id": reason.form_id,
-                            "backstage_input_url": _build_backstage_input_url(reason.form_token),
+                            "backstage_input_url": _build_backstage_input_url(
+                                form_tokens_by_form_id.get(reason.form_id)
+                            ),
                         },
                     }
                 )

api/controllers/console/auth/oauth_server.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel
 from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.console.wraps import account_initialization_required, setup_required
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from models.model import OAuthProviderApp

api/controllers/console/datasets/datasets.py

@@ -25,13 +25,12 @@ from controllers.console.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
-from core.provider_manager import ProviderManager
+from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
-from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.app_fields import app_detail_kernel_fields, related_app_list
 from fields.dataset_fields import (
@@ -52,6 +51,7 @@ from fields.dataset_fields import (
     weighted_score_fields,
 )
 from fields.document_fields import document_status_fields
+from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_account_with_tenant, login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermission, DatasetPermissionEnum
@@ -332,7 +332,7 @@ class DatasetListApi(Resource):
             )
 
         # check embedding setting
-        provider_manager = ProviderManager()
+        provider_manager = create_plugin_provider_manager(tenant_id=current_tenant_id)
         configurations = provider_manager.get_configurations(tenant_id=current_tenant_id)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)
@@ -446,7 +446,7 @@ class DatasetApi(Resource):
             data.update({"partial_member_list": part_users_list})
 
         # check embedding setting
-        provider_manager = ProviderManager()
+        provider_manager = create_plugin_provider_manager(tenant_id=current_tenant_id)
         configurations = provider_manager.get_configurations(tenant_id=current_tenant_id)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)

api/controllers/console/datasets/datasets_document.py

@@ -28,8 +28,6 @@ from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
-from dify_graph.model_runtime.entities.model_entities import ModelType
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from fields.dataset_fields import dataset_fields
 from fields.document_fields import (
@@ -39,6 +37,8 @@ from fields.document_fields import (
     document_status_fields,
     document_with_segments_fields,
 )
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
@@ -454,7 +454,7 @@ class DatasetInitApi(Resource):
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
                 raise ValueError("embedding model and embedding model provider are required for high quality indexing.")
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=knowledge_config.embedding_model_provider,

api/controllers/console/datasets/datasets_segments.py

@@ -27,10 +27,10 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
-from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
+from graphon.model_runtime.entities.model_entities import ModelType
 from libs.helper import escape_like_pattern
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import ChildChunk, DocumentSegment
@@ -283,7 +283,7 @@ class DatasetDocumentSegmentApi(Resource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -336,7 +336,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -387,7 +387,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -572,7 +572,7 @@ class ChildChunkAddApi(Resource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,

api/controllers/console/datasets/external.py

@@ -25,7 +25,7 @@ from libs.login import current_account_with_tenant, login_required
 from services.dataset_service import DatasetService
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
-from services.knowledge_service import ExternalDatasetTestService
+from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
 
 
 def _build_dataset_detail_model():
@@ -86,7 +86,7 @@ class ExternalHitTestingPayload(BaseModel):
 
 
 class BedrockRetrievalPayload(BaseModel):
-    retrieval_setting: dict[str, object]
+    retrieval_setting: "BedrockRetrievalSetting"
     query: str
     knowledge_id: str
 

api/controllers/console/datasets/hit_testing_base.py

@@ -19,8 +19,8 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.hit_testing_fields import hit_testing_record_fields
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs.login import current_user
 from models.account import Account
 from services.dataset_service import DatasetService

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -10,8 +10,8 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from core.plugin.impl.oauth import OAuthHandler
-from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import DatasourceProviderID
 from services.datasource_provider_service import DatasourceProviderService

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -21,11 +21,12 @@ from controllers.console.app.workflow_draft_variable import (
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
-from dify_graph.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
-from dify_graph.variables.types import SegmentType
+from core.app.file_access import DatabaseFileAccessController
+from core.workflow.variable_prefixes import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
+from graphon.variables.types import SegmentType
 from libs.login import current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -33,6 +34,7 @@ from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
 
 logger = logging.getLogger(__name__)
+_file_access_controller = DatabaseFileAccessController()
 
 
 def _create_pagination_parser():
@@ -223,13 +225,21 @@ class RagPipelineVariableApi(Resource):
             if variable.value_type == SegmentType.FILE:
                 if not isinstance(raw_value, dict):
                     raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                raw_value = build_from_mapping(mapping=raw_value, tenant_id=pipeline.tenant_id)
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=pipeline.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             elif variable.value_type == SegmentType.ARRAY_FILE:
                 if not isinstance(raw_value, list):
                     raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                raw_value = build_from_mappings(mappings=raw_value, tenant_id=pipeline.tenant_id)
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=pipeline.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -37,9 +37,9 @@ from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpErr
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs import helper
 from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required

api/controllers/console/explore/audio.py

@@ -19,7 +19,7 @@ from controllers.console.app.error import (
 )
 from controllers.console.explore.wraps import InstalledAppResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from services.audio_service import AudioService
 from services.errors.audio import (
     AudioTooLargeServiceError,

api/controllers/console/explore/completion.py

@@ -24,8 +24,8 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_user

api/controllers/console/explore/message.py

@@ -21,9 +21,9 @@ from controllers.console.explore.error import (
 from controllers.console.explore.wraps import InstalledAppResource
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
 from libs.login import current_account_with_tenant

api/controllers/console/explore/trial.py

@@ -42,8 +42,6 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.graph_engine.manager import GraphEngineManager
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.app_fields import (
@@ -61,6 +59,8 @@ from fields.workflow_fields import (
     workflow_fields,
     workflow_partial_fields,
 )
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from libs.login import current_user

api/controllers/console/explore/workflow.py

@@ -21,9 +21,9 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.graph_engine.manager import GraphEngineManager
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.login import current_account_with_tenant
 from models.model import AppMode, InstalledApp

api/controllers/console/remote_files.py

@@ -13,9 +13,9 @@ from controllers.common.errors import (
 )
 from controllers.console import console_ns
 from core.helper import ssrf_proxy
-from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
+from graphon.file import helpers as file_helpers
 from libs.login import current_account_with_tenant, login_required
 from services.file_service import FileService
 

api/controllers/console/workspace/agent_providers.py

@@ -2,7 +2,7 @@ from flask_restx import Resource, fields
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.agent_service import AgentService
 

api/controllers/console/workspace/endpoint.py

@@ -8,7 +8,7 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.plugin.impl.exc import PluginPermissionDeniedError
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from services.plugin.endpoint_service import EndpointService
 

api/controllers/console/workspace/load_balancing_config.py

@@ -5,8 +5,8 @@ from werkzeug.exceptions import Forbidden
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
-from dify_graph.model_runtime.entities.model_entities import ModelType
-from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
 from libs.login import current_account_with_tenant, login_required
 from models import TenantAccountRole
 from services.model_load_balancing_service import ModelLoadBalancingService

api/controllers/console/workspace/model_providers.py

@@ -7,9 +7,9 @@ from pydantic import BaseModel, Field, field_validator
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
-from dify_graph.model_runtime.entities.model_entities import ModelType
-from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService

api/controllers/console/workspace/models.py

@@ -8,9 +8,9 @@ from pydantic import BaseModel, Field, field_validator
 from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
-from dify_graph.model_runtime.entities.model_entities import ModelType
-from dify_graph.model_runtime.errors.validate import CredentialsValidateFailedError
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import current_account_with_tenant, login_required
 from services.model_load_balancing_service import ModelLoadBalancingService
@@ -282,14 +282,18 @@ class ModelProviderModelCredentialApi(Resource):
         )
 
         if args.config_from == "predefined-model":
-            available_credentials = model_provider_service.provider_manager.get_provider_available_credentials(
-                tenant_id=tenant_id, provider_name=provider
+            available_credentials = model_provider_service.get_provider_available_credentials(
+                tenant_id=tenant_id,
+                provider=provider,
             )
         else:
             # Normalize model_type to the origin value stored in DB (e.g., "text-generation" for LLM)
             normalized_model_type = args.model_type.to_origin_model_type()
-            available_credentials = model_provider_service.provider_manager.get_provider_model_available_credentials(
-                tenant_id=tenant_id, provider_name=provider, model_type=normalized_model_type, model_name=args.model
+            available_credentials = model_provider_service.get_provider_model_available_credentials(
+                tenant_id=tenant_id,
+                provider=provider,
+                model_type=normalized_model_type,
+                model=args.model,
             )
 
         return jsonable_encoder(

api/controllers/console/workspace/plugin.py

@@ -14,7 +14,7 @@ from controllers.console import console_ns
 from controllers.console.workspace import plugin_permission_required
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.plugin.impl.exc import PluginDaemonClientSideError
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_account_with_tenant, login_required
 from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermission
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService

api/controllers/console/workspace/tool_providers.py

@@ -26,8 +26,8 @@ from core.mcp.mcp_client import MCPClient
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
 from core.tools.entities.tool_entities import ApiProviderSchemaType, WorkflowToolParameterConfiguration
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import alphanumeric, uuid_value
 from libs.login import current_account_with_tenant, login_required
 from models.provider_ids import ToolProviderID

api/controllers/console/workspace/trigger_providers.py

@@ -14,8 +14,8 @@ from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
 from core.trigger.entities.entities import SubscriptionBuilderUpdater
 from core.trigger.trigger_manager import TriggerManager
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import current_user, login_required
 from models.account import Account
 from models.provider_ids import TriggerProviderID

api/controllers/files/tool_files.py

@@ -70,22 +70,25 @@ class ToolFileApi(Resource):
         except Exception:
             raise UnsupportedFileTypeError()
 
+        mime_type = tool_file.mime_type
+        filename = tool_file.filename
+
         response = Response(
             stream,
-            mimetype=tool_file.mimetype,
+            mimetype=mime_type,
             direct_passthrough=True,
             headers={},
         )
         if tool_file.size > 0:
             response.headers["Content-Length"] = str(tool_file.size)
-        if args.as_attachment:
-            encoded_filename = quote(tool_file.name)
+        if args.as_attachment and filename:
+            encoded_filename = quote(filename)
             response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
 
         enforce_download_for_html(
             response,
-            mime_type=tool_file.mimetype,
-            filename=tool_file.name,
+            mime_type=mime_type,
+            filename=filename,
             extension=extension,
         )
 

api/controllers/files/upload.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
 import services
+from core.tools.signature import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
-from dify_graph.file.helpers import verify_plugin_file_signature
 from fields.file_fields import FileResponse
 
 from ..common.errors import (

api/controllers/inner_api/__init__.py

@@ -16,12 +16,14 @@ api = ExternalApi(
 inner_api_ns = Namespace("inner_api", description="Internal API operations", path="/")
 
 from . import mail as _mail
+from .app import dsl as _app_dsl
 from .plugin import plugin as _plugin
 from .workspace import workspace as _workspace
 
 api.add_namespace(inner_api_ns)
 
 __all__ = [
+    "_app_dsl",
     "_mail",
     "_plugin",
     "_workspace",

api/controllers/inner_api/app/__init__.py

@@ -0,0 +1 @@
+

api/controllers/inner_api/app/dsl.py

@@ -0,0 +1,110 @@
+"""Inner API endpoints for app DSL import/export.
+
+Called by the enterprise admin-api service. Import requires ``creator_email``
+to attribute the created app; workspace/membership validation is done by the
+Go admin-api caller.
+"""
+
+from flask import request
+from flask_restx import Resource
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from controllers.common.schema import register_schema_model
+from controllers.console.wraps import setup_required
+from controllers.inner_api import inner_api_ns
+from controllers.inner_api.wraps import enterprise_inner_api_only
+from extensions.ext_database import db
+from models import Account, App
+from models.account import AccountStatus
+from services.app_dsl_service import AppDslService, ImportMode, ImportStatus
+
+
+class InnerAppDSLImportPayload(BaseModel):
+    yaml_content: str = Field(description="YAML DSL content")
+    creator_email: str = Field(description="Email of the workspace member who will own the imported app")
+    name: str | None = Field(default=None, description="Override app name from DSL")
+    description: str | None = Field(default=None, description="Override app description from DSL")
+
+
+register_schema_model(inner_api_ns, InnerAppDSLImportPayload)
+
+
+@inner_api_ns.route("/enterprise/workspaces/<string:workspace_id>/dsl/import")
+class EnterpriseAppDSLImport(Resource):
+    @setup_required
+    @enterprise_inner_api_only
+    @inner_api_ns.doc("enterprise_app_dsl_import")
+    @inner_api_ns.expect(inner_api_ns.models[InnerAppDSLImportPayload.__name__])
+    @inner_api_ns.doc(
+        responses={
+            200: "Import completed",
+            202: "Import pending (DSL version mismatch requires confirmation)",
+            400: "Import failed (business error)",
+            404: "Creator account not found or inactive",
+        }
+    )
+    def post(self, workspace_id: str):
+        """Import a DSL into a workspace on behalf of a specified creator."""
+        args = InnerAppDSLImportPayload.model_validate(inner_api_ns.payload or {})
+
+        account = _get_active_account(args.creator_email)
+        if account is None:
+            return {"message": f"account '{args.creator_email}' not found or inactive"}, 404
+
+        account.set_tenant_id(workspace_id)
+
+        with Session(db.engine) as session:
+            dsl_service = AppDslService(session)
+            result = dsl_service.import_app(
+                account=account,
+                import_mode=ImportMode.YAML_CONTENT,
+                yaml_content=args.yaml_content,
+                name=args.name,
+                description=args.description,
+            )
+            session.commit()
+
+        if result.status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        if result.status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+@inner_api_ns.route("/enterprise/apps/<string:app_id>/dsl")
+class EnterpriseAppDSLExport(Resource):
+    @setup_required
+    @enterprise_inner_api_only
+    @inner_api_ns.doc(
+        "enterprise_app_dsl_export",
+        responses={
+            200: "Export successful",
+            404: "App not found",
+        },
+    )
+    def get(self, app_id: str):
+        """Export an app's DSL as YAML."""
+        include_secret = request.args.get("include_secret", "false").lower() == "true"
+
+        app_model = db.session.query(App).filter_by(id=app_id).first()
+        if not app_model:
+            return {"message": "app not found"}, 404
+
+        data = AppDslService.export_dsl(
+            app_model=app_model,
+            include_secret=include_secret,
+        )
+
+        return {"data": data}, 200
+
+
+def _get_active_account(email: str) -> Account | None:
+    """Look up an active account by email.
+
+    Workspace membership is already validated by the Go admin-api caller.
+    """
+    account = db.session.query(Account).filter_by(email=email).first()
+    if account is None or account.status != AccountStatus.ACTIVE:
+        return None
+    return account

api/controllers/inner_api/plugin/plugin.py

@@ -28,8 +28,8 @@ from core.plugin.entities.request import (
     RequestRequestUploadFile,
 )
 from core.tools.entities.tool_entities import ToolProviderType
-from dify_graph.file.helpers import get_signed_file_url_for_plugin
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from core.tools.signature import get_signed_file_url_for_plugin
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import length_prefixed_response
 from models import Account, Tenant
 from models.model import EndUser

api/controllers/mcp/mcp.py

@@ -9,8 +9,8 @@ from controllers.common.schema import register_schema_model
 from controllers.mcp import mcp_ns
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request
-from dify_graph.variables.input_entities import VariableEntity
 from extensions.ext_database import db
+from graphon.variables.input_entities import VariableEntity
 from libs import helper
 from models.enums import AppMCPServerStatus
 from models.model import App, AppMCPServer, AppMode, EndUser

api/controllers/service_api/app/audio.py

@@ -21,7 +21,7 @@ from controllers.service_api.app.error import (
 )
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from models.model import App, EndUser
 from services.audio_service import AudioService
 from services.errors.audio import (

api/controllers/service_api/app/completion.py

@@ -28,7 +28,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser

api/controllers/service_api/app/file_preview.py

@@ -4,6 +4,7 @@ from urllib.parse import quote
 from flask import Response, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
+from sqlalchemy import select
 
 from controllers.common.file_response import enforce_download_for_html
 from controllers.common.schema import register_schema_model
@@ -102,27 +103,27 @@ class FilePreviewApi(Resource):
                 raise FileAccessDeniedError("Invalid file or app identifier")
 
             # First, find the MessageFile that references this upload file
-            message_file = db.session.query(MessageFile).where(MessageFile.upload_file_id == file_id).first()
+            message_file = db.session.scalar(select(MessageFile).where(MessageFile.upload_file_id == file_id).limit(1))
 
             if not message_file:
                 raise FileNotFoundError("File not found in message context")
 
             # Get the message and verify it belongs to the requesting app
-            message = (
-                db.session.query(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).first()
+            message = db.session.scalar(
+                select(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).limit(1)
             )
 
             if not message:
                 raise FileAccessDeniedError("File access denied: not owned by requesting app")
 
             # Get the actual upload file record
-            upload_file = db.session.query(UploadFile).where(UploadFile.id == file_id).first()
+            upload_file = db.session.get(UploadFile, file_id)
 
             if not upload_file:
                 raise FileNotFoundError("Upload file record not found")
 
             # Additional security: verify tenant isolation
-            app = db.session.query(App).where(App.id == app_id).first()
+            app = db.session.get(App, app_id)
             if app and upload_file.tenant_id != app.tenant_id:
                 raise FileAccessDeniedError("File access denied: tenant mismatch")
 

api/controllers/service_api/app/site.py

@@ -1,4 +1,5 @@
 from flask_restx import Resource
+from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from controllers.common.fields import Site as SiteResponse
@@ -28,7 +29,7 @@ class AppSiteApi(Resource):
 
         Returns the site configuration for the application including theme, icons, and text.
         """
-        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
+        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
 
         if not site:
             raise Forbidden()

api/controllers/service_api/app/workflow.py

@@ -27,12 +27,12 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id
-from dify_graph.enums import WorkflowExecutionStatus
-from dify_graph.graph_engine.manager import GraphEngineManager
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
+from graphon.enums import WorkflowExecutionStatus
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import OptionalTimestampField, TimestampField
 from models.model import App, AppMode, EndUser

api/controllers/service_api/dataset/dataset.py

@@ -14,11 +14,11 @@ from controllers.service_api.wraps import (
     DatasetApiResource,
     cloud_edition_billing_rate_limit_check,
 )
-from core.provider_manager import ProviderManager
+from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
-from dify_graph.model_runtime.entities.model_entities import ModelType
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import DataSetTag
+from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -140,10 +140,10 @@ class DatasetListApi(DatasetApiResource):
             query.page, query.limit, tenant_id, current_user, query.keyword, query.tag_ids, query.include_all
         )
         # check embedding setting
-        provider_manager = ProviderManager()
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
+        provider_manager = create_plugin_provider_manager(tenant_id=cid)
         configurations = provider_manager.get_configurations(tenant_id=cid)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)
@@ -259,10 +259,10 @@ class DatasetApi(DatasetApiResource):
             raise Forbidden(str(e))
         data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         # check embedding setting
-        provider_manager = ProviderManager()
         assert isinstance(current_user, Account)
         cid = current_user.current_tenant_id
         assert cid is not None
+        provider_manager = create_plugin_provider_manager(tenant_id=cid)
         configurations = provider_manager.get_configurations(tenant_id=cid)
 
         embedding_models = configurations.get_models(model_type=ModelType.TEXT_EMBEDDING, only_active=True)

api/controllers/service_api/dataset/segment.py

@@ -18,9 +18,9 @@ from controllers.service_api.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
-from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.segment_fields import child_chunk_fields, segment_fields
+from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from services.dataset_service import DatasetService, DocumentService, SegmentService
@@ -106,7 +106,7 @@ class SegmentApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -160,7 +160,7 @@ class SegmentApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -266,7 +266,7 @@ class DatasetSegmentApi(DatasetApiResource):
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             # check embedding model setting
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,
@@ -361,7 +361,7 @@ class ChildChunkApi(DatasetApiResource):
         # check embedding model setting
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
             try:
-                model_manager = ModelManager()
+                model_manager = ModelManager.for_tenant(tenant_id=current_tenant_id)
                 model_manager.get_model_instance(
                     tenant_id=current_tenant_id,
                     provider=dataset.embedding_model_provider,

api/controllers/service_api/workspace/models.py

@@ -3,7 +3,7 @@ from flask_restx import Resource
 
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_dataset_token
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 from services.model_provider_service import ModelProviderService
 
 

api/controllers/service_api/wraps.py

@@ -9,6 +9,7 @@ from flask import current_app, request
 from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
+from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from enums.cloud_plan import CloudPlan
@@ -62,7 +63,7 @@ def validate_app_token(
         def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
             api_token = validate_and_get_api_token("app")
 
-            app_model = db.session.query(App).where(App.id == api_token.app_id).first()
+            app_model = db.session.get(App, api_token.app_id)
             if not app_model:
                 raise Forbidden("The app no longer exists.")
 
@@ -72,7 +73,7 @@ def validate_app_token(
             if not app_model.enable_api:
                 raise Forbidden("The app's API service has been disabled.")
 
-            tenant = db.session.query(Tenant).where(Tenant.id == app_model.tenant_id).first()
+            tenant = db.session.get(Tenant, app_model.tenant_id)
             if tenant is None:
                 raise ValueError("Tenant does not exist.")
             if tenant.status == TenantStatus.ARCHIVE:
@@ -106,8 +107,8 @@ def validate_app_token(
             else:
                 # For service API without end-user context, ensure an Account is logged in
                 # so services relying on current_account_with_tenant() work correctly.
-                tenant_owner_info = (
-                    db.session.query(Tenant, Account)
+                tenant_owner_info = db.session.execute(
+                    select(Tenant, Account)
                     .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
                     .join(Account, TenantAccountJoin.account_id == Account.id)
                     .where(
@@ -115,8 +116,7 @@ def validate_app_token(
                         TenantAccountJoin.role == "owner",
                         Tenant.status == TenantStatus.NORMAL,
                     )
-                    .one_or_none()
-                )
+                ).one_or_none()
 
                 if tenant_owner_info:
                     tenant_model, account = tenant_owner_info
@@ -277,29 +277,28 @@ def validate_dataset_token(
             # Validate dataset if dataset_id is provided
             if dataset_id:
                 dataset_id = str(dataset_id)
-                dataset = (
-                    db.session.query(Dataset)
+                dataset = db.session.scalar(
+                    select(Dataset)
                     .where(
                         Dataset.id == dataset_id,
                         Dataset.tenant_id == api_token.tenant_id,
                     )
-                    .first()
+                    .limit(1)
                 )
                 if not dataset:
                     raise NotFound("Dataset not found.")
                 if not dataset.enable_api:
                     raise Forbidden("Dataset api access is not enabled.")
-            tenant_account_join = (
-                db.session.query(Tenant, TenantAccountJoin)
+            tenant_account_join = db.session.execute(
+                select(Tenant, TenantAccountJoin)
                 .where(Tenant.id == api_token.tenant_id)
                 .where(TenantAccountJoin.tenant_id == Tenant.id)
                 .where(TenantAccountJoin.role.in_(["owner"]))
                 .where(Tenant.status == TenantStatus.NORMAL)
-                .one_or_none()
-            )  # TODO: only owner information is required, so only one is returned.
+            ).one_or_none()  # TODO: only owner information is required, so only one is returned.
             if tenant_account_join:
                 tenant, ta = tenant_account_join
-                account = db.session.query(Account).where(Account.id == ta.account_id).first()
+                account = db.session.get(Account, ta.account_id)
                 # Login admin
                 if account:
                     account.current_tenant = tenant
@@ -360,7 +359,9 @@ class DatasetApiResource(Resource):
     method_decorators = [validate_dataset_token]
 
     def get_dataset(self, dataset_id: str, tenant_id: str) -> Dataset:
-        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id, Dataset.tenant_id == tenant_id).first()
+        dataset = db.session.scalar(
+            select(Dataset).where(Dataset.id == dataset_id, Dataset.tenant_id == tenant_id).limit(1)
+        )
 
         if not dataset:
             raise NotFound("Dataset not found.")

api/controllers/web/audio.py

@@ -20,7 +20,7 @@ from controllers.web.error import (
 )
 from controllers.web.wraps import WebApiResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs.helper import uuid_value
 from models.model import App
 from services.audio_service import AudioService

api/controllers/web/completion.py

@@ -25,7 +25,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from models.model import AppMode

api/controllers/web/message.py

@@ -20,9 +20,9 @@ from controllers.web.error import (
 from controllers.web.wraps import WebApiResource
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfiniteScrollPagination, WebMessageListItem
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
 from models.enums import FeedbackRating

api/controllers/web/remote_files.py

@@ -11,9 +11,9 @@ from controllers.common.errors import (
     UnsupportedFileTypeError,
 )
 from core.helper import ssrf_proxy
-from dify_graph.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
+from graphon.file import helpers as file_helpers
 from services.file_service import FileService
 
 from ..common.schema import register_schema_models

api/controllers/web/workflow.py

@@ -22,9 +22,9 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
-from dify_graph.graph_engine.manager import GraphEngineManager
-from dify_graph.model_runtime.errors.invoke import InvokeError
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
+from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService

api/core/agent/base_agent_runner.py

@@ -15,6 +15,7 @@ from core.app.entities.app_invoke_entities import (
     AgentChatAppGenerateEntity,
     ModelConfigWithCredentialsEntity,
 )
+from core.app.file_access import DatabaseFileAccessController
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
 from core.memory.token_buffer_memory import TokenBufferMemory
@@ -26,8 +27,10 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
-from dify_graph.file import file_manager
-from dify_graph.model_runtime.entities import (
+from extensions.ext_database import db
+from factories import file_factory
+from graphon.file import file_manager
+from graphon.model_runtime.entities import (
     AssistantPromptMessage,
     LLMUsage,
     PromptMessage,
@@ -37,15 +40,14 @@ from dify_graph.model_runtime.entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
-from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from dify_graph.model_runtime.entities.model_entities import ModelFeature
-from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from extensions.ext_database import db
-from factories import file_factory
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from graphon.model_runtime.entities.model_entities import ModelFeature
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from models.enums import CreatorUserRole
 from models.model import Conversation, Message, MessageAgentThought, MessageFile
 
 logger = logging.getLogger(__name__)
+_file_access_controller = DatabaseFileAccessController()
 
 
 class BaseAgentRunner(AppRunner):
@@ -138,6 +140,7 @@ class BaseAgentRunner(AppRunner):
             tenant_id=self.tenant_id,
             app_id=self.app_config.app_id,
             agent_tool=tool,
+            user_id=self.user_id,
             invoke_from=self.application_generate_entity.invoke_from,
         )
         assert tool_entity.entity.description
@@ -524,7 +527,10 @@ class BaseAgentRunner(AppRunner):
         image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
         file_objs = file_factory.build_from_message_files(
-            message_files=files, tenant_id=self.tenant_id, config=file_extra_config
+            message_files=files,
+            tenant_id=self.tenant_id,
+            config=file_extra_config,
+            access_controller=_file_access_controller,
         )
         if not file_objs:
             return UserPromptMessage(content=message.query)

api/core/agent/cot_agent_runner.py

@@ -15,8 +15,8 @@ from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransfo
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
-from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from dify_graph.model_runtime.entities.message_entities import (
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from graphon.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     PromptMessage,
     PromptMessageTool,
@@ -122,7 +122,6 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                 tools=[],
                 stop=app_generate_entity.model_conf.stop,
                 stream=True,
-                user=self.user_id,
                 callbacks=[],
             )
 

api/core/agent/cot_chat_agent_runner.py

@@ -1,16 +1,16 @@
 import json
 
 from core.agent.cot_agent_runner import CotAgentRunner
-from dify_graph.file import file_manager
-from dify_graph.model_runtime.entities import (
+from graphon.file import file_manager
+from graphon.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
     SystemPromptMessage,
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 
 class CotChatAgentRunner(CotAgentRunner):

api/core/agent/cot_completion_agent_runner.py

@@ -1,13 +1,13 @@
 import json
 
 from core.agent.cot_agent_runner import CotAgentRunner
-from dify_graph.model_runtime.entities.message_entities import (
+from graphon.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     PromptMessage,
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
+from graphon.model_runtime.utils.encoders import jsonable_encoder
 
 
 class CotCompletionAgentRunner(CotAgentRunner):

api/core/agent/fc_agent_runner.py

@@ -11,8 +11,8 @@ from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueMessag
 from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
-from dify_graph.file import file_manager
-from dify_graph.model_runtime.entities import (
+from graphon.file import file_manager
+from graphon.model_runtime.entities import (
     AssistantPromptMessage,
     LLMResult,
     LLMResultChunk,
@@ -25,7 +25,7 @@ from dify_graph.model_runtime.entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
-from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from models.model import Message
 
 logger = logging.getLogger(__name__)
@@ -96,7 +96,6 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                 tools=prompt_messages_tools,
                 stop=app_generate_entity.model_conf.stop,
                 stream=self.stream_tool_call,
-                user=self.user_id,
                 callbacks=[],
             )
 

api/core/agent/output_parser/cot_output_parser.py

@@ -4,7 +4,7 @@ from collections.abc import Generator
 from typing import Union
 
 from core.agent.entities import AgentScratchpadUnit
-from dify_graph.model_runtime.entities.llm_entities import LLMResultChunk
+from graphon.model_runtime.entities.llm_entities import LLMResultChunk
 
 
 class CotAgentOutputParser:

api/core/app/app_config/easy_ui_based_app/model_config/converter.py

@@ -4,10 +4,10 @@ from core.app.app_config.entities import EasyUIBasedAppConfig
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities.model_entities import ModelStatus
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from core.provider_manager import ProviderManager
-from dify_graph.model_runtime.entities.llm_entities import LLMMode
-from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
-from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
+from graphon.model_runtime.entities.llm_entities import LLMMode
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 
 
 class ModelConfigConverter:
@@ -21,7 +21,7 @@ class ModelConfigConverter:
         """
         model_config = app_config.model
 
-        provider_manager = ProviderManager()
+        provider_manager = create_plugin_provider_manager(tenant_id=app_config.tenant_id)
         provider_model_bundle = provider_manager.get_provider_model_bundle(
             tenant_id=app_config.tenant_id, provider=model_config.provider, model_type=ModelType.LLM
         )

api/core/app/app_config/easy_ui_based_app/model_config/manager.py

@@ -2,9 +2,8 @@ from collections.abc import Mapping
 from typing import Any
 
 from core.app.app_config.entities import ModelConfigEntity
-from core.provider_manager import ProviderManager
-from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
-from dify_graph.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
+from core.plugin.impl.model_runtime_factory import create_plugin_model_assembly
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from models.model import AppModelConfigDict
 from models.provider_ids import ModelProviderID
 
@@ -54,9 +53,12 @@ class ModelConfigManager:
         if not isinstance(config["model"], dict):
             raise ValueError("model must be of object type")
 
+        # Keep provider discovery and provider-backed model listing on the same
+        # request-scoped runtime so caller scope and provider caches stay aligned.
+        assembly = create_plugin_model_assembly(tenant_id=tenant_id)
+
         # model.provider
-        model_provider_factory = ModelProviderFactory(tenant_id)
-        provider_entities = model_provider_factory.get_providers()
+        provider_entities = assembly.model_provider_factory.get_providers()
         model_provider_names = [provider.provider for provider in provider_entities]
         if "provider" not in config["model"]:
             raise ValueError(f"model.provider is required and must be in {str(model_provider_names)}")
@@ -71,8 +73,7 @@ class ModelConfigManager:
         if "name" not in config["model"]:
             raise ValueError("model.name is required")
 
-        provider_manager = ProviderManager()
-        models = provider_manager.get_configurations(tenant_id).get_models(
+        models = assembly.provider_manager.get_configurations(tenant_id).get_models(
             provider=config["model"]["provider"], model_type=ModelType.LLM
         )
 

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -7,7 +7,7 @@ from core.app.app_config.entities import (
     PromptTemplateEntity,
 )
 from core.prompt.simple_prompt_transform import ModelMode
-from dify_graph.model_runtime.entities.message_entities import PromptMessageRole
+from graphon.model_runtime.entities.message_entities import PromptMessageRole
 from models.model import AppMode, AppModelConfigDict
 
 

api/core/app/app_config/easy_ui_based_app/variables/manager.py

@@ -3,7 +3,7 @@ from typing import cast
 
 from core.app.app_config.entities import ExternalDataVariableEntity
 from core.external_data_tool.factory import ExternalDataToolFactory
-from dify_graph.variables.input_entities import VariableEntity, VariableEntityType
+from graphon.variables.input_entities import VariableEntity, VariableEntityType
 from models.model import AppModelConfigDict
 
 _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(

api/core/app/app_config/entities.py

@@ -5,10 +5,10 @@ from typing import Any, Literal
 from pydantic import BaseModel, Field
 
 from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
-from dify_graph.file import FileUploadConfig
-from dify_graph.model_runtime.entities.llm_entities import LLMMode
-from dify_graph.model_runtime.entities.message_entities import PromptMessageRole
-from dify_graph.variables.input_entities import VariableEntity as WorkflowVariableEntity
+from graphon.file import FileUploadConfig
+from graphon.model_runtime.entities.llm_entities import LLMMode
+from graphon.model_runtime.entities.message_entities import PromptMessageRole
+from graphon.variables.input_entities import VariableEntity as WorkflowVariableEntity
 from models.model import AppMode
 
 

api/core/app/app_config/features/file_upload/manager.py

@@ -2,7 +2,7 @@ from collections.abc import Mapping
 from typing import Any
 
 from constants import DEFAULT_FILE_NUMBER_LIMITS
-from dify_graph.file import FileUploadConfig
+from graphon.file import FileUploadConfig
 
 
 class FileUploadConfigManager:

api/core/app/app_config/workflow_ui_based_app/variables/manager.py

@@ -1,7 +1,7 @@
 import re
 
 from core.app.app_config.entities import RagPipelineVariableEntity
-from dify_graph.variables.input_entities import VariableEntity
+from graphon.variables.input_entities import VariableEntity
 from models.workflow import Workflow
 
 

api/core/app/apps/advanced_chat/app_generator.py

@@ -24,6 +24,7 @@ from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
 from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter
 from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
@@ -34,17 +35,13 @@ from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.graph_engine.layers.base import GraphEngineLayer
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
-from dify_graph.repositories.draft_variable_repository import (
-    DraftVariableSaverFactory,
-)
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.runtime import GraphRuntimeState
-from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.graph_engine.layers.base import GraphEngineLayer
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.runtime import GraphRuntimeState
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.base import Base
@@ -150,85 +147,87 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        files = args["files"] if args.get("files") else []
-        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
-        if file_extra_config:
-            file_objs = file_factory.build_from_mappings(
-                mappings=files,
-                tenant_id=app_model.tenant_id,
-                config=file_extra_config,
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            files = args["files"] if args.get("files") else []
+            file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+            if file_extra_config:
+                file_objs = file_factory.build_from_mappings(
+                    mappings=files,
+                    tenant_id=app_model.tenant_id,
+                    config=file_extra_config,
+                    access_controller=self._file_access_controller,
+                )
+            else:
+                file_objs = []
+
+            # convert to app config
+            app_config = AdvancedChatAppConfigManager.get_app_config(app_model=app_model, workflow=workflow)
+
+            # get tracing instance
+            trace_manager = TraceQueueManager(
+                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
             )
-        else:
-            file_objs = []
 
-        # convert to app config
-        app_config = AdvancedChatAppConfigManager.get_app_config(app_model=app_model, workflow=workflow)
+            if invoke_from == InvokeFrom.DEBUGGER:
+                # always enable retriever resource in debugger mode
+                app_config.additional_features.show_retrieve_source = True  # type: ignore
 
-        # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-        )
+            # init application generate entity
+            application_generate_entity = AdvancedChatAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                file_upload_config=file_extra_config,
+                conversation_id=conversation.id if conversation else None,
+                inputs=self._prepare_user_inputs(
+                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+                ),
+                query=query,
+                files=list(file_objs),
+                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+                user_id=user.id,
+                stream=streaming,
+                invoke_from=invoke_from,
+                extras=extras,
+                trace_manager=trace_manager,
+                workflow_run_id=str(workflow_run_id),
+            )
+            contexts.plugin_tool_providers.set({})
+            contexts.plugin_tool_providers_lock.set(threading.Lock())
 
-        if invoke_from == InvokeFrom.DEBUGGER:
-            # always enable retriever resource in debugger mode
-            app_config.additional_features.show_retrieve_source = True  # type: ignore
+            # Create repositories
+            #
+            # Create session factory
+            session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+            # Create workflow execution(aka workflow run) repository
+            if invoke_from == InvokeFrom.DEBUGGER:
+                workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+            else:
+                workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+            workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
+                session_factory=session_factory,
+                user=user,
+                app_id=application_generate_entity.app_config.app_id,
+                triggered_from=workflow_triggered_from,
+            )
+            # Create workflow node execution repository
+            workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
+                session_factory=session_factory,
+                user=user,
+                app_id=application_generate_entity.app_config.app_id,
+                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+            )
 
-        # init application generate entity
-        application_generate_entity = AdvancedChatAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            file_upload_config=file_extra_config,
-            conversation_id=conversation.id if conversation else None,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-            ),
-            query=query,
-            files=list(file_objs),
-            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=invoke_from,
-            extras=extras,
-            trace_manager=trace_manager,
-            workflow_run_id=str(workflow_run_id),
-        )
-        contexts.plugin_tool_providers.set({})
-        contexts.plugin_tool_providers_lock.set(threading.Lock())
-
-        # Create repositories
-        #
-        # Create session factory
-        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-        # Create workflow execution(aka workflow run) repository
-        if invoke_from == InvokeFrom.DEBUGGER:
-            workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
-        else:
-            workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
-        workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=workflow_triggered_from,
-        )
-        # Create workflow node execution repository
-        workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
-
-        return self._generate(
-            workflow=workflow,
-            user=user,
-            invoke_from=invoke_from,
-            application_generate_entity=application_generate_entity,
-            workflow_execution_repository=workflow_execution_repository,
-            workflow_node_execution_repository=workflow_node_execution_repository,
-            conversation=conversation,
-            stream=streaming,
-            pause_state_config=pause_state_config,
-        )
+            return self._generate(
+                workflow=workflow,
+                user=user,
+                invoke_from=invoke_from,
+                application_generate_entity=application_generate_entity,
+                workflow_execution_repository=workflow_execution_repository,
+                workflow_node_execution_repository=workflow_node_execution_repository,
+                conversation=conversation,
+                stream=streaming,
+                pause_state_config=pause_state_config,
+            )
 
     def resume(
         self,
@@ -460,94 +459,90 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         :param conversation: conversation
         :param stream: is stream
         """
-        is_first_conversation = conversation is None
+        with self._bind_file_access_scope(
+            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
+            invoke_from=invoke_from,
+        ):
+            is_first_conversation = conversation is None
 
-        if conversation is not None and message is not None:
-            pass
-        else:
-            conversation, message = self._init_generate_records(application_generate_entity, conversation)
+            if conversation is not None and message is not None:
+                pass
+            else:
+                conversation, message = self._init_generate_records(application_generate_entity, conversation)
 
-        if is_first_conversation:
-            # update conversation features
-            conversation.override_model_configs = workflow.features
-            db.session.commit()
-            db.session.refresh(conversation)
+            if is_first_conversation:
+                # update conversation features
+                conversation.override_model_configs = workflow.features
+                db.session.commit()
+                db.session.refresh(conversation)
 
-        # get conversation dialogue count
-        # NOTE: dialogue_count should not start from 0,
-        # because during the first conversation, dialogue_count should be 1.
-        self._dialogue_count = get_thread_messages_length(conversation.id) + 1
+            # get conversation dialogue count
+            # NOTE: dialogue_count should not start from 0,
+            # because during the first conversation, dialogue_count should be 1.
+            self._dialogue_count = get_thread_messages_length(conversation.id) + 1
 
-        # init queue manager
-        queue_manager = MessageBasedAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation.id,
-            app_mode=conversation.mode,
-            message_id=message.id,
-        )
-
-        graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
-        if pause_state_config is not None:
-            graph_layers.append(
-                PauseStatePersistenceLayer(
-                    session_factory=pause_state_config.session_factory,
-                    generate_entity=application_generate_entity,
-                    state_owner_user_id=pause_state_config.state_owner_user_id,
-                )
+            # init queue manager
+            queue_manager = MessageBasedAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
+                conversation_id=conversation.id,
+                app_mode=conversation.mode,
+                message_id=message.id,
             )
 
-        # new thread with request context and contextvars
-        context = contextvars.copy_context()
+            graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+            if pause_state_config is not None:
+                graph_layers.append(
+                    PauseStatePersistenceLayer(
+                        session_factory=pause_state_config.session_factory,
+                        generate_entity=application_generate_entity,
+                        state_owner_user_id=pause_state_config.state_owner_user_id,
+                    )
+                )
 
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "conversation_id": conversation.id,
-                "message_id": message.id,
-                "context": context,
-                "variable_loader": variable_loader,
-                "workflow_execution_repository": workflow_execution_repository,
-                "workflow_node_execution_repository": workflow_node_execution_repository,
-                "graph_engine_layers": tuple(graph_layers),
-                "graph_runtime_state": graph_runtime_state,
-            },
-        )
+            # new thread with request context and contextvars
+            context = contextvars.copy_context()
 
-        worker_thread.start()
+            worker_thread = threading.Thread(
+                target=self._generate_worker,
+                kwargs={
+                    "flask_app": current_app._get_current_object(),  # type: ignore
+                    "application_generate_entity": application_generate_entity,
+                    "queue_manager": queue_manager,
+                    "conversation_id": conversation.id,
+                    "message_id": message.id,
+                    "context": context,
+                    "variable_loader": variable_loader,
+                    "workflow_execution_repository": workflow_execution_repository,
+                    "workflow_node_execution_repository": workflow_node_execution_repository,
+                    "graph_engine_layers": tuple(graph_layers),
+                    "graph_runtime_state": graph_runtime_state,
+                },
+            )
 
-        # release database connection, because the following new thread operations may take a long time
-        with Session(bind=db.engine, expire_on_commit=False) as session:
-            workflow = _refresh_model(session, workflow)
-            message = _refresh_model(session, message)
-        #     workflow_ = session.get(Workflow, workflow.id)
-        #     assert workflow_ is not None
-        #     workflow = workflow_
-        #     message_ = session.get(Message, message.id)
-        #     assert message_ is not None
-        #     message = message_
-        # db.session.refresh(workflow)
-        # db.session.refresh(message)
-        # db.session.refresh(user)
-        db.session.close()
+            worker_thread.start()
 
-        # return response or stream generator
-        response = self._handle_advanced_chat_response(
-            application_generate_entity=application_generate_entity,
-            workflow=workflow,
-            queue_manager=queue_manager,
-            conversation=conversation,
-            message=message,
-            user=user,
-            stream=stream,
-            draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
-        )
+            # release database connection, because the following new thread operations may take a long time
+            with Session(bind=db.engine, expire_on_commit=False) as session:
+                workflow = _refresh_model(session, workflow)
+                message = _refresh_model(session, message)
+            db.session.close()
 
-        return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            # return response or stream generator
+            response = self._handle_advanced_chat_response(
+                application_generate_entity=application_generate_entity,
+                workflow=workflow,
+                queue_manager=queue_manager,
+                conversation=conversation,
+                message=message,
+                user=user,
+                stream=stream,
+                draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
+            )
+
+            return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,

api/core/app/apps/advanced_chat/app_runner.py

@@ -25,19 +25,24 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.node_factory import get_default_root_node_id
+from core.workflow.system_variables import (
+    build_bootstrap_variables,
+    build_system_variables,
+    system_variables_to_mapping,
+)
+from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from dify_graph.enums import WorkflowType
-from dify_graph.graph_engine.command_channels.redis_channel import RedisChannel
-from dify_graph.graph_engine.layers.base import GraphEngineLayer
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.runtime import GraphRuntimeState, VariablePool
-from dify_graph.system_variable import SystemVariable
-from dify_graph.variable_loader import VariableLoader
-from dify_graph.variables.variables import Variable
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
+from graphon.enums import WorkflowType
+from graphon.graph_engine.command_channels.redis_channel import RedisChannel
+from graphon.graph_engine.layers.base import GraphEngineLayer
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
+from graphon.variables.variables import Variable
 from models import Workflow
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
@@ -90,7 +95,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
-        system_inputs = SystemVariable(
+        system_inputs = build_system_variables(
             query=self.application_generate_entity.query,
             files=self.application_generate_entity.files,
             conversation_id=self.conversation.id,
@@ -150,7 +155,10 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
             self.application_generate_entity.inputs = new_inputs
             self.application_generate_entity.query = new_query
-            system_inputs.query = new_query
+            system_inputs = build_system_variables(
+                system_variables_to_mapping(system_inputs),
+                query=new_query,
+            )
 
             # annotation reply
             if self.handle_annotation_reply(
@@ -166,14 +174,17 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
             # Create a variable pool.
             # init variable pool
-            variable_pool = VariablePool(
-                system_variables=system_inputs,
-                user_inputs=new_inputs,
-                environment_variables=self._workflow.environment_variables,
-                # Based on the definition of `Variable`,
-                # `VariableBase` instances can be safely used as `Variable` since they are compatible.
-                conversation_variables=conversation_variables,
+            variable_pool = VariablePool()
+            add_variables_to_pool(
+                variable_pool,
+                build_bootstrap_variables(
+                    system_variables=system_inputs,
+                    environment_variables=self._workflow.environment_variables,
+                    conversation_variables=conversation_variables,
+                ),
             )
+            root_node_id = get_default_root_node_id(self._workflow.graph_dict)
+            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=new_inputs)
 
             # init graph
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
@@ -185,6 +196,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 user_id=self.application_generate_entity.user_id,
                 user_from=user_from,
                 invoke_from=invoke_from,
+                root_node_id=root_node_id,
             )
 
         db.session.close()

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -14,6 +14,7 @@ from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.common.graph_runtime_state_support import GraphRuntimeStateSupport
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     InvokeFrom,
@@ -65,15 +66,15 @@ from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
-from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.enums import WorkflowExecutionStatus
-from dify_graph.model_runtime.entities.llm_entities import LLMUsage
-from dify_graph.model_runtime.utils.encoders import jsonable_encoder
-from dify_graph.nodes import BuiltinNodeTypes
-from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
-from dify_graph.runtime import GraphRuntimeState
-from dify_graph.system_variable import SystemVariable
+from core.workflow.file_reference import resolve_file_record_id
+from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.enums import WorkflowExecutionStatus
+from graphon.model_runtime.entities.llm_entities import LLMUsage
+from graphon.model_runtime.utils.encoders import jsonable_encoder
+from graphon.nodes import BuiltinNodeTypes
+from graphon.runtime import GraphRuntimeState
 from libs.datetime_utils import naive_utc_now
 from models import Account, Conversation, EndUser, Message, MessageFile
 from models.enums import CreatorUserRole, MessageFileBelongsTo, MessageStatus
@@ -117,7 +118,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         else:
             raise NotImplementedError(f"User type not supported: {type(user)}")
 
-        self._workflow_system_variables = SystemVariable(
+        self._workflow_system_variables = build_system_variables(
             query=message.query,
             files=application_generate_entity.files,
             conversation_id=conversation.id,
@@ -741,8 +742,9 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     def _load_human_input_form_id(self, *, node_id: str) -> str | None:
         form_repository = HumanInputFormRepositoryImpl(
             tenant_id=self._workflow_tenant_id,
+            workflow_execution_id=self._workflow_run_id,
         )
-        form = form_repository.get_form(self._workflow_run_id, node_id)
+        form = form_repository.get_form(node_id)
         if form is None:
             return None
         return form.id
@@ -933,21 +935,23 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
 
         metadata = self._task_state.metadata.model_dump()
         message.message_metadata = json.dumps(jsonable_encoder(metadata))
-        message_files = [
-            MessageFile(
-                message_id=message.id,
-                type=file["type"],
-                transfer_method=file["transfer_method"],
-                url=file["remote_url"],
-                belongs_to=MessageFileBelongsTo.ASSISTANT,
-                upload_file_id=file["related_id"],
-                created_by_role=CreatorUserRole.ACCOUNT
-                if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else CreatorUserRole.END_USER,
-                created_by=message.from_account_id or message.from_end_user_id or "",
+        message_files: list[MessageFile] = []
+        for file in self._recorded_files:
+            reference = file.get("reference") or file.get("related_id")
+            message_files.append(
+                MessageFile(
+                    message_id=message.id,
+                    type=file["type"],
+                    transfer_method=file["transfer_method"],
+                    url=file["remote_url"],
+                    belongs_to=MessageFileBelongsTo.ASSISTANT,
+                    upload_file_id=resolve_file_record_id(reference if isinstance(reference, str) else None),
+                    created_by_role=CreatorUserRole.ACCOUNT
+                    if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
+                    else CreatorUserRole.END_USER,
+                    created_by=message.from_account_id or message.from_end_user_id or "",
+                )
             )
-            for file in self._recorded_files
-        ]
         session.add_all(message_files)
 
     def _seed_graph_runtime_state_from_queue_manager(self) -> None:
@@ -1003,13 +1007,11 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         return message
 
     def _save_output_for_event(self, event: QueueNodeSucceededEvent | QueueNodeExceptionEvent, node_execution_id: str):
-        with Session(db.engine) as session, session.begin():
-            saver = self._draft_var_saver_factory(
-                session=session,
-                app_id=self._application_generate_entity.app_config.app_id,
-                node_id=event.node_id,
-                node_type=event.node_type,
-                node_execution_id=node_execution_id,
-                enclosing_node_id=event.in_loop_id or event.in_iteration_id,
-            )
-            saver.save(event.process_data, event.outputs)
+        saver = self._draft_var_saver_factory(
+            app_id=self._application_generate_entity.app_config.app_id,
+            node_id=event.node_id,
+            node_type=event.node_type,
+            node_execution_id=node_execution_id,
+            enclosing_node_id=event.in_loop_id or event.in_iteration_id,
+        )
+        saver.save(event.process_data, event.outputs)

api/core/app/apps/agent_chat/app_generator.py

@@ -21,9 +21,9 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, EndUser
 from services.conversation_service import ConversationService
@@ -129,89 +129,93 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        files = args.get("files") or []
-        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
-        if file_extra_config:
-            file_objs = file_factory.build_from_mappings(
-                mappings=files,
-                tenant_id=app_model.tenant_id,
-                config=file_extra_config,
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            files = args.get("files") or []
+            file_extra_config = FileUploadConfigManager.convert(
+                override_model_config_dict or app_model_config.to_dict()
             )
-        else:
-            file_objs = []
+            if file_extra_config:
+                file_objs = file_factory.build_from_mappings(
+                    mappings=files,
+                    tenant_id=app_model.tenant_id,
+                    config=file_extra_config,
+                    access_controller=self._file_access_controller,
+                )
+            else:
+                file_objs = []
 
-        # convert to app config
-        app_config = AgentChatAppConfigManager.get_app_config(
-            app_model=app_model,
-            app_model_config=app_model_config,
-            conversation=conversation,
-            override_config_dict=override_model_config_dict,
-        )
+            # convert to app config
+            app_config = AgentChatAppConfigManager.get_app_config(
+                app_model=app_model,
+                app_model_config=app_model_config,
+                conversation=conversation,
+                override_config_dict=override_model_config_dict,
+            )
 
-        # get tracing instance
-        trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
+            # get tracing instance
+            trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
 
-        # init application generate entity
-        application_generate_entity = AgentChatAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            model_conf=ModelConfigConverter.convert(app_config),
-            file_upload_config=file_extra_config,
-            conversation_id=conversation.id if conversation else None,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-            ),
-            query=query,
-            files=list(file_objs),
-            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=invoke_from,
-            extras=extras,
-            call_depth=0,
-            trace_manager=trace_manager,
-        )
+            # init application generate entity
+            application_generate_entity = AgentChatAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                model_conf=ModelConfigConverter.convert(app_config),
+                file_upload_config=file_extra_config,
+                conversation_id=conversation.id if conversation else None,
+                inputs=self._prepare_user_inputs(
+                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+                ),
+                query=query,
+                files=list(file_objs),
+                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+                user_id=user.id,
+                stream=streaming,
+                invoke_from=invoke_from,
+                extras=extras,
+                call_depth=0,
+                trace_manager=trace_manager,
+            )
 
-        # init generate records
-        (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
+            # init generate records
+            (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
 
-        # init queue manager
-        queue_manager = MessageBasedAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation.id,
-            app_mode=conversation.mode,
-            message_id=message.id,
-        )
+            # init queue manager
+            queue_manager = MessageBasedAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
+                conversation_id=conversation.id,
+                app_mode=conversation.mode,
+                message_id=message.id,
+            )
 
-        # new thread with request context and contextvars
-        context = contextvars.copy_context()
+            # new thread with request context and contextvars
+            context = contextvars.copy_context()
 
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "context": context,
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "conversation_id": conversation.id,
-                "message_id": message.id,
-            },
-        )
+            worker_thread = threading.Thread(
+                target=self._generate_worker,
+                kwargs={
+                    "flask_app": current_app._get_current_object(),  # type: ignore
+                    "context": context,
+                    "application_generate_entity": application_generate_entity,
+                    "queue_manager": queue_manager,
+                    "conversation_id": conversation.id,
+                    "message_id": message.id,
+                },
+            )
 
-        worker_thread.start()
+            worker_thread.start()
 
-        # return response or stream generator
-        response = self._handle_response(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            conversation=conversation,
-            message=message,
-            user=user,
-            stream=streaming,
-        )
-        return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            # return response or stream generator
+            response = self._handle_response(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation=conversation,
+                message=message,
+                user=user,
+                stream=streaming,
+            )
+            return AgentChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,

api/core/app/apps/agent_chat/app_runner.py

@@ -15,10 +15,10 @@ from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
-from dify_graph.model_runtime.entities.llm_entities import LLMMode
-from dify_graph.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from dify_graph.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from extensions.ext_database import db
+from graphon.model_runtime.entities.llm_entities import LLMMode
+from graphon.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from models.model import App, Conversation, Message
 
 logger = logging.getLogger(__name__)

api/core/app/apps/base_app_generate_response_converter.py

@@ -6,7 +6,7 @@ from typing import Any, Union
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
-from dify_graph.model_runtime.errors.invoke import InvokeError
+from graphon.model_runtime.errors.invoke import InvokeError
 
 logger = logging.getLogger(__name__)
 

api/core/app/apps/base_app_generator.py

@@ -1,27 +1,89 @@
 from collections.abc import Generator, Mapping, Sequence
+from contextlib import AbstractContextManager, nullcontext
 from typing import TYPE_CHECKING, Any, Union, final
 
 from sqlalchemy.orm import Session
 
-from core.app.entities.app_invoke_entities import InvokeFrom
-from dify_graph.enums import NodeType
-from dify_graph.file import File, FileUploadConfig
-from dify_graph.repositories.draft_variable_repository import (
+from core.app.apps.draft_variable_saver import (
     DraftVariableSaver,
     DraftVariableSaverFactory,
     NoopDraftVariableSaver,
 )
-from dify_graph.variables.input_entities import VariableEntityType
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
+from core.app.file_access import DatabaseFileAccessController, FileAccessScope, bind_file_access_scope
+from extensions.ext_database import db
 from factories import file_factory
+from graphon.enums import NodeType
+from graphon.file import File, FileUploadConfig
+from graphon.variables.input_entities import VariableEntityType
 from libs.orjson import orjson_dumps
 from models import Account, EndUser
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
-    from dify_graph.variables.input_entities import VariableEntity
+    from graphon.variables.input_entities import VariableEntity
+
+
+@final
+class _DebuggerDraftVariableSaver:
+    """Adapter that binds SQLAlchemy session setup outside the saver port."""
+
+    def __init__(
+        self,
+        *,
+        account: Account,
+        app_id: str,
+        node_id: str,
+        node_type: NodeType,
+        node_execution_id: str,
+        enclosing_node_id: str | None = None,
+    ) -> None:
+        self._account = account
+        self._app_id = app_id
+        self._node_id = node_id
+        self._node_type = node_type
+        self._node_execution_id = node_execution_id
+        self._enclosing_node_id = enclosing_node_id
+
+    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None) -> None:
+        with Session(db.engine) as session, session.begin():
+            DraftVariableSaverImpl(
+                session=session,
+                app_id=self._app_id,
+                node_id=self._node_id,
+                node_type=self._node_type,
+                node_execution_id=self._node_execution_id,
+                enclosing_node_id=self._enclosing_node_id,
+                user=self._account,
+            ).save(process_data, outputs)
 
 
 class BaseAppGenerator:
+    _file_access_controller: DatabaseFileAccessController = DatabaseFileAccessController()
+
+    @staticmethod
+    def _bind_file_access_scope(
+        *,
+        tenant_id: str,
+        user: Account | EndUser,
+        invoke_from: InvokeFrom,
+    ) -> AbstractContextManager[None]:
+        """Bind request-scoped file ownership markers for downstream file lookups."""
+
+        user_id = getattr(user, "id", None)
+        if not isinstance(user_id, str) or not user_id:
+            return nullcontext()
+
+        user_from = UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER
+        return bind_file_access_scope(
+            FileAccessScope(
+                tenant_id=tenant_id,
+                user_id=user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+            )
+        )
+
     def _prepare_user_inputs(
         self,
         *,
@@ -50,6 +112,7 @@ class BaseAppGenerator:
                     allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
                 strict_type_validation=strict_type_validation,
+                access_controller=self._file_access_controller,
             )
             for k, v in user_inputs.items()
             if isinstance(v, dict) and entity_dictionary[k].type == VariableEntityType.FILE
@@ -64,6 +127,7 @@ class BaseAppGenerator:
                     allowed_file_extensions=entity_dictionary[k].allowed_file_extensions or [],
                     allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
+                access_controller=self._file_access_controller,
             )
             for k, v in user_inputs.items()
             if isinstance(v, list)
@@ -226,32 +290,30 @@ class BaseAppGenerator:
             assert isinstance(account, Account)
 
             def draft_var_saver_factory(
-                session: Session,
                 app_id: str,
                 node_id: str,
                 node_type: NodeType,
                 node_execution_id: str,
                 enclosing_node_id: str | None = None,
             ) -> DraftVariableSaver:
-                return DraftVariableSaverImpl(
-                    session=session,
+                return _DebuggerDraftVariableSaver(
+                    account=account,
                     app_id=app_id,
                     node_id=node_id,
                     node_type=node_type,
                     node_execution_id=node_execution_id,
                     enclosing_node_id=enclosing_node_id,
-                    user=account,
                 )
         else:
 
             def draft_var_saver_factory(
-                session: Session,
                 app_id: str,
                 node_id: str,
                 node_type: NodeType,
                 node_execution_id: str,
                 enclosing_node_id: str | None = None,
             ) -> DraftVariableSaver:
+                _ = app_id, node_id, node_type, node_execution_id, enclosing_node_id
                 return NoopDraftVariableSaver()
 
         return draft_var_saver_factory

api/core/app/apps/base_app_queue_manager.py

@@ -20,8 +20,8 @@ from core.app.entities.queue_entities import (
     QueueStopEvent,
     WorkflowQueueMessage,
 )
-from dify_graph.runtime import GraphRuntimeState
 from extensions.ext_redis import redis_client
+from graphon.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)
 
@@ -61,27 +61,30 @@ class AppQueueManager(ABC):
         listen_timeout = dify_config.APP_MAX_EXECUTION_TIME
         start_time = time.time()
         last_ping_time: int | float = 0
-        while True:
-            try:
-                message = self._q.get(timeout=1)
-                if message is None:
-                    break
+        try:
+            while True:
+                try:
+                    message = self._q.get(timeout=1)
+                    if message is None:
+                        break
 
-                yield message
-            except queue.Empty:
-                continue
-            finally:
-                elapsed_time = time.time() - start_time
-                if elapsed_time >= listen_timeout or self._is_stopped():
-                    # publish two messages to make sure the client can receive the stop signal
-                    # and stop listening after the stop signal processed
-                    self.publish(
-                        QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL), PublishFrom.TASK_PIPELINE
-                    )
+                    yield message
+                except queue.Empty:
+                    continue
+                finally:
+                    elapsed_time = time.time() - start_time
+                    if elapsed_time >= listen_timeout or self._is_stopped():
+                        # publish two messages to make sure the client can receive the stop signal
+                        # and stop listening after the stop signal processed
+                        self.publish(
+                            QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL), PublishFrom.TASK_PIPELINE
+                        )
 
-                if elapsed_time // 10 > last_ping_time:
-                    self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
-                    last_ping_time = elapsed_time // 10
+                    if elapsed_time // 10 > last_ping_time:
+                        self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
+                        last_ping_time = elapsed_time // 10
+        finally:
+            self._graph_runtime_state = None  # Release reference once consumers finish or close the generator.
 
     def stop_listen(self):
         """
@@ -90,7 +93,6 @@ class AppQueueManager(ABC):
         """
         self._clear_task_belong_cache()
         self._q.put(None)
-        self._graph_runtime_state = None  # Release reference to allow GC to reclaim memory
 
     def _clear_task_belong_cache(self) -> None:
         """

api/core/app/apps/base_app_runner.py

@@ -29,22 +29,22 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
-from dify_graph.file.enums import FileTransferMethod, FileType
-from dify_graph.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
-from dify_graph.model_runtime.entities.message_entities import (
+from extensions.ext_database import db
+from graphon.file.enums import FileTransferMethod, FileType
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from graphon.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     ImagePromptMessageContent,
     PromptMessage,
     TextPromptMessageContent,
 )
-from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey
-from dify_graph.model_runtime.errors.invoke import InvokeBadRequestError
-from extensions.ext_database import db
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
 from models.enums import CreatorUserRole, MessageFileBelongsTo
 from models.model import App, AppMode, Message, MessageAnnotation, MessageFile
 
 if TYPE_CHECKING:
-    from dify_graph.file.models import File
+    from graphon.file.models import File
 
 _logger = logging.getLogger(__name__)
 

api/core/app/apps/chat/app_generator.py

@@ -1,3 +1,4 @@
+import contextvars
 import logging
 import threading
 import uuid
@@ -20,9 +21,9 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from models import Account
 from models.model import App, EndUser
 from services.conversation_service import ConversationService
@@ -120,89 +121,96 @@ class ChatAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        files = args["files"] if args.get("files") else []
-        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
-        if file_extra_config:
-            file_objs = file_factory.build_from_mappings(
-                mappings=files,
-                tenant_id=app_model.tenant_id,
-                config=file_extra_config,
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            files = args["files"] if args.get("files") else []
+            file_extra_config = FileUploadConfigManager.convert(
+                override_model_config_dict or app_model_config.to_dict()
             )
-        else:
-            file_objs = []
+            if file_extra_config:
+                file_objs = file_factory.build_from_mappings(
+                    mappings=files,
+                    tenant_id=app_model.tenant_id,
+                    config=file_extra_config,
+                    access_controller=self._file_access_controller,
+                )
+            else:
+                file_objs = []
 
-        # convert to app config
-        app_config = ChatAppConfigManager.get_app_config(
-            app_model=app_model,
-            app_model_config=app_model_config,
-            conversation=conversation,
-            override_config_dict=override_model_config_dict,
-        )
+            # convert to app config
+            app_config = ChatAppConfigManager.get_app_config(
+                app_model=app_model,
+                app_model_config=app_model_config,
+                conversation=conversation,
+                override_config_dict=override_model_config_dict,
+            )
 
-        # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-        )
+            # get tracing instance
+            trace_manager = TraceQueueManager(
+                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+            )
 
-        # init application generate entity
-        application_generate_entity = ChatAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            model_conf=ModelConfigConverter.convert(app_config),
-            file_upload_config=file_extra_config,
-            conversation_id=conversation.id if conversation else None,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-            ),
-            query=query,
-            files=list(file_objs),
-            parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
-            user_id=user.id,
-            invoke_from=invoke_from,
-            extras=extras,
-            trace_manager=trace_manager,
-            stream=streaming,
-        )
+            # init application generate entity
+            application_generate_entity = ChatAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                model_conf=ModelConfigConverter.convert(app_config),
+                file_upload_config=file_extra_config,
+                conversation_id=conversation.id if conversation else None,
+                inputs=self._prepare_user_inputs(
+                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+                ),
+                query=query,
+                files=list(file_objs),
+                parent_message_id=args.get("parent_message_id") if invoke_from != InvokeFrom.SERVICE_API else UUID_NIL,
+                user_id=user.id,
+                invoke_from=invoke_from,
+                extras=extras,
+                trace_manager=trace_manager,
+                stream=streaming,
+            )
 
-        # init generate records
-        (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
+            # init generate records
+            (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
 
-        # init queue manager
-        queue_manager = MessageBasedAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation.id,
-            app_mode=conversation.mode,
-            message_id=message.id,
-        )
-
-        # new thread with request context
-        @copy_current_request_context
-        def worker_with_context():
-            return self._generate_worker(
-                flask_app=current_app._get_current_object(),  # type: ignore
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
+            # init queue manager
+            queue_manager = MessageBasedAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
                 conversation_id=conversation.id,
+                app_mode=conversation.mode,
                 message_id=message.id,
             )
 
-        worker_thread = threading.Thread(target=worker_with_context)
+            context = contextvars.copy_context()
 
-        worker_thread.start()
+            # new thread with request context
+            @copy_current_request_context
+            def worker_with_context():
+                return context.run(
+                    self._generate_worker,
+                    flask_app=current_app._get_current_object(),  # type: ignore
+                    application_generate_entity=application_generate_entity,
+                    queue_manager=queue_manager,
+                    conversation_id=conversation.id,
+                    message_id=message.id,
+                )
 
-        # return response or stream generator
-        response = self._handle_response(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            conversation=conversation,
-            message=message,
-            user=user,
-            stream=streaming,
-        )
+            worker_thread = threading.Thread(target=worker_with_context)
 
-        return ChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            worker_thread.start()
+
+            # return response or stream generator
+            response = self._handle_response(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation=conversation,
+                message=message,
+                user=user,
+                stream=streaming,
+            )
+
+            return ChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,

api/core/app/apps/chat/app_runner.py

@@ -15,9 +15,9 @@ from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
-from dify_graph.file import File
-from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent
 from extensions.ext_database import db
+from graphon.file import File
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from models.model import App, Conversation, Message
 
 logger = logging.getLogger(__name__)
@@ -223,7 +223,6 @@ class ChatAppRunner(AppRunner):
             model_parameters=application_generate_entity.model_conf.parameters,
             stop=stop,
             stream=application_generate_entity.stream,
-            user=application_generate_entity.user_id,
         )
 
         # handle invoke result

api/core/app/apps/common/graph_runtime_state_support.py

@@ -4,7 +4,8 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from dify_graph.runtime import GraphRuntimeState
+from core.workflow.system_variables import SystemVariableKey, get_system_text
+from graphon.runtime import GraphRuntimeState
 
 if TYPE_CHECKING:
     from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
@@ -30,10 +31,10 @@ class GraphRuntimeStateSupport:
         return self._resolve_graph_runtime_state(graph_runtime_state)
 
     def _extract_workflow_run_id(self, graph_runtime_state: GraphRuntimeState) -> str:
-        system_variables = graph_runtime_state.variable_pool.system_variables
-        if not system_variables or not system_variables.workflow_execution_id:
+        workflow_run_id = get_system_text(graph_runtime_state.variable_pool, SystemVariableKey.WORKFLOW_EXECUTION_ID)
+        if not workflow_run_id:
             raise ValueError("workflow_execution_id missing from runtime state")
-        return str(system_variables.workflow_execution_id)
+        return workflow_run_id
 
     def _resolve_graph_runtime_state(
         self,

api/core/app/apps/common/workflow_response_converter.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import time
 from collections.abc import Mapping, Sequence
@@ -50,22 +51,23 @@ from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
 from core.trigger.trigger_manager import TriggerManager
+from core.workflow.human_input_forms import load_form_tokens_by_form_id
+from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
-from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.entities.workflow_start_reason import WorkflowStartReason
-from dify_graph.enums import (
+from extensions.ext_database import db
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.entities.workflow_start_reason import WorkflowStartReason
+from graphon.enums import (
     BuiltinNodeTypes,
-    SystemVariableKey,
     WorkflowExecutionStatus,
     WorkflowNodeExecutionMetadataKey,
     WorkflowNodeExecutionStatus,
 )
-from dify_graph.file import FILE_MODEL_IDENTITY, File
-from dify_graph.runtime import GraphRuntimeState
-from dify_graph.system_variable import SystemVariable
-from dify_graph.variables.segments import ArrayFileSegment, FileSegment, Segment
-from dify_graph.workflow_type_encoder import WorkflowRuntimeTypeConverter
-from extensions.ext_database import db
+from graphon.file import FILE_MODEL_IDENTITY, File
+from graphon.runtime import GraphRuntimeState
+from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
+from graphon.variables.variables import Variable
+from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import Account, EndUser
 from models.human_input import HumanInputForm
@@ -111,11 +113,11 @@ class WorkflowResponseConverter:
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
         user: Union[Account, EndUser],
-        system_variables: SystemVariable,
+        system_variables: Sequence[Variable],
     ):
         self._application_generate_entity = application_generate_entity
         self._user = user
-        self._system_variables = system_variables
+        self._system_variables = system_variables_to_mapping(system_variables)
         self._workflow_inputs = self._prepare_workflow_inputs()
 
         # Disable truncation for SERVICE_API calls to keep backward compatibility.
@@ -133,7 +135,7 @@ class WorkflowResponseConverter:
     # ------------------------------------------------------------------
     def _prepare_workflow_inputs(self) -> Mapping[str, Any]:
         inputs = dict(self._application_generate_entity.inputs)
-        for field_name, value in self._system_variables.to_dict().items():
+        for field_name, value in self._system_variables.items():
             # TODO(@future-refactor): store system variables separately from user inputs so we don't
             # need to flatten `sys.*` entries into the input payload just for rerun/export tooling.
             if field_name == SystemVariableKey.CONVERSATION_ID:
@@ -318,13 +320,23 @@ class WorkflowResponseConverter:
         pause_reasons = [reason.model_dump(mode="json") for reason in event.reasons]
         human_input_form_ids = [reason.form_id for reason in event.reasons if isinstance(reason, HumanInputRequired)]
         expiration_times_by_form_id: dict[str, datetime] = {}
+        display_in_ui_by_form_id: dict[str, bool] = {}
+        form_token_by_form_id: dict[str, str] = {}
         if human_input_form_ids:
-            stmt = select(HumanInputForm.id, HumanInputForm.expiration_time).where(
-                HumanInputForm.id.in_(human_input_form_ids)
-            )
+            stmt = select(
+                HumanInputForm.id,
+                HumanInputForm.expiration_time,
+                HumanInputForm.form_definition,
+            ).where(HumanInputForm.id.in_(human_input_form_ids))
             with Session(bind=db.engine) as session:
-                for form_id, expiration_time in session.execute(stmt):
+                for form_id, expiration_time, form_definition in session.execute(stmt):
                     expiration_times_by_form_id[str(form_id)] = expiration_time
+                    try:
+                        definition_payload = json.loads(form_definition) if form_definition else {}
+                    except (TypeError, json.JSONDecodeError):
+                        definition_payload = {}
+                    display_in_ui_by_form_id[str(form_id)] = bool(definition_payload.get("display_in_ui"))
+                form_token_by_form_id = load_form_tokens_by_form_id(human_input_form_ids, session=session)
 
         responses: list[StreamResponse] = []
 
@@ -344,8 +356,8 @@ class WorkflowResponseConverter:
                             form_content=reason.form_content,
                             inputs=reason.inputs,
                             actions=reason.actions,
-                            display_in_ui=reason.display_in_ui,
-                            form_token=reason.form_token,
+                            display_in_ui=display_in_ui_by_form_id.get(reason.form_id, False),
+                            form_token=form_token_by_form_id.get(reason.form_id),
                             resolved_default_values=reason.resolved_default_values,
                             expiration_time=int(expiration_time.timestamp()),
                         ),

api/core/app/apps/completion/app_generator.py

@@ -1,3 +1,4 @@
+import contextvars
 import logging
 import threading
 import uuid
@@ -20,9 +21,9 @@ from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, InvokeFrom
 from core.ops.ops_trace_manager import TraceQueueManager
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 from models import Account, App, EndUser, Message
 from services.errors.app import MoreLikeThisDisabledError
 from services.errors.message import MessageNotExistsError
@@ -108,83 +109,90 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         #
         # For implementation reference, see the `_parse_file` function and
         # `DraftWorkflowNodeRunApi` class which handle this properly.
-        files = args["files"] if args.get("files") else []
-        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
-        if file_extra_config:
-            file_objs = file_factory.build_from_mappings(
-                mappings=files,
-                tenant_id=app_model.tenant_id,
-                config=file_extra_config,
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            files = args["files"] if args.get("files") else []
+            file_extra_config = FileUploadConfigManager.convert(
+                override_model_config_dict or app_model_config.to_dict()
             )
-        else:
-            file_objs = []
+            if file_extra_config:
+                file_objs = file_factory.build_from_mappings(
+                    mappings=files,
+                    tenant_id=app_model.tenant_id,
+                    config=file_extra_config,
+                    access_controller=self._file_access_controller,
+                )
+            else:
+                file_objs = []
 
-        # convert to app config
-        app_config = CompletionAppConfigManager.get_app_config(
-            app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
-        )
+            # convert to app config
+            app_config = CompletionAppConfigManager.get_app_config(
+                app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
+            )
 
-        # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
-        )
+            # get tracing instance
+            trace_manager = TraceQueueManager(
+                app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+            )
 
-        # init application generate entity
-        application_generate_entity = CompletionAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            model_conf=ModelConfigConverter.convert(app_config),
-            file_upload_config=file_extra_config,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
-            ),
-            query=query,
-            files=list(file_objs),
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=invoke_from,
-            extras={},
-            trace_manager=trace_manager,
-        )
+            # init application generate entity
+            application_generate_entity = CompletionAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                model_conf=ModelConfigConverter.convert(app_config),
+                file_upload_config=file_extra_config,
+                inputs=self._prepare_user_inputs(
+                    user_inputs=inputs, variables=app_config.variables, tenant_id=app_model.tenant_id
+                ),
+                query=query,
+                files=list(file_objs),
+                user_id=user.id,
+                stream=streaming,
+                invoke_from=invoke_from,
+                extras={},
+                trace_manager=trace_manager,
+            )
 
-        # init generate records
-        (conversation, message) = self._init_generate_records(application_generate_entity)
+            # init generate records
+            (conversation, message) = self._init_generate_records(application_generate_entity)
 
-        # init queue manager
-        queue_manager = MessageBasedAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation.id,
-            app_mode=conversation.mode,
-            message_id=message.id,
-        )
-
-        # new thread with request context
-        @copy_current_request_context
-        def worker_with_context():
-            return self._generate_worker(
-                flask_app=current_app._get_current_object(),  # type: ignore
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
+            # init queue manager
+            queue_manager = MessageBasedAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
+                conversation_id=conversation.id,
+                app_mode=conversation.mode,
                 message_id=message.id,
             )
 
-        worker_thread = threading.Thread(target=worker_with_context)
+            context = contextvars.copy_context()
 
-        worker_thread.start()
+            # new thread with request context
+            @copy_current_request_context
+            def worker_with_context():
+                return context.run(
+                    self._generate_worker,
+                    flask_app=current_app._get_current_object(),  # type: ignore
+                    application_generate_entity=application_generate_entity,
+                    queue_manager=queue_manager,
+                    message_id=message.id,
+                )
 
-        # return response or stream generator
-        response = self._handle_response(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            conversation=conversation,
-            message=message,
-            user=user,
-            stream=streaming,
-        )
+            worker_thread = threading.Thread(target=worker_with_context)
 
-        return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            worker_thread.start()
+
+            # return response or stream generator
+            response = self._handle_response(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation=conversation,
+                message=message,
+                user=user,
+                stream=streaming,
+            )
+
+            return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def _generate_worker(
         self,
@@ -280,71 +288,76 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         model_dict["completion_params"] = completion_params
         override_model_config_dict["model"] = model_dict
 
-        # parse files
-        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict)
-        if file_extra_config:
-            file_objs = file_factory.build_from_mappings(
-                mappings=message.message_files,
-                tenant_id=app_model.tenant_id,
-                config=file_extra_config,
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            # parse files
+            file_extra_config = FileUploadConfigManager.convert(override_model_config_dict)
+            if file_extra_config:
+                file_objs = file_factory.build_from_mappings(
+                    mappings=message.message_files,
+                    tenant_id=app_model.tenant_id,
+                    config=file_extra_config,
+                    access_controller=self._file_access_controller,
+                )
+            else:
+                file_objs = []
+
+            # convert to app config
+            app_config = CompletionAppConfigManager.get_app_config(
+                app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
             )
-        else:
-            file_objs = []
 
-        # convert to app config
-        app_config = CompletionAppConfigManager.get_app_config(
-            app_model=app_model, app_model_config=app_model_config, override_config_dict=override_model_config_dict
-        )
+            # init application generate entity
+            application_generate_entity = CompletionAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                model_conf=ModelConfigConverter.convert(app_config),
+                inputs=message.inputs,
+                query=message.query,
+                files=list(file_objs),
+                user_id=user.id,
+                stream=stream,
+                invoke_from=invoke_from,
+                extras={},
+            )
 
-        # init application generate entity
-        application_generate_entity = CompletionAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            model_conf=ModelConfigConverter.convert(app_config),
-            inputs=message.inputs,
-            query=message.query,
-            files=list(file_objs),
-            user_id=user.id,
-            stream=stream,
-            invoke_from=invoke_from,
-            extras={},
-        )
+            # init generate records
+            (conversation, message) = self._init_generate_records(application_generate_entity)
 
-        # init generate records
-        (conversation, message) = self._init_generate_records(application_generate_entity)
-
-        # init queue manager
-        queue_manager = MessageBasedAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation.id,
-            app_mode=conversation.mode,
-            message_id=message.id,
-        )
-
-        # new thread with request context
-        @copy_current_request_context
-        def worker_with_context():
-            return self._generate_worker(
-                flask_app=current_app._get_current_object(),  # type: ignore
-                application_generate_entity=application_generate_entity,
-                queue_manager=queue_manager,
+            # init queue manager
+            queue_manager = MessageBasedAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
+                conversation_id=conversation.id,
+                app_mode=conversation.mode,
                 message_id=message.id,
             )
 
-        worker_thread = threading.Thread(target=worker_with_context)
+            context = contextvars.copy_context()
 
-        worker_thread.start()
+            # new thread with request context
+            @copy_current_request_context
+            def worker_with_context():
+                return context.run(
+                    self._generate_worker,
+                    flask_app=current_app._get_current_object(),  # type: ignore
+                    application_generate_entity=application_generate_entity,
+                    queue_manager=queue_manager,
+                    message_id=message.id,
+                )
 
-        # return response or stream generator
-        response = self._handle_response(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            conversation=conversation,
-            message=message,
-            user=user,
-            stream=stream,
-        )
+            worker_thread = threading.Thread(target=worker_with_context)
 
-        return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            worker_thread.start()
+
+            # return response or stream generator
+            response = self._handle_response(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation=conversation,
+                message=message,
+                user=user,
+                stream=stream,
+            )
+
+            return CompletionAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)

api/core/app/apps/completion/app_runner.py

@@ -13,9 +13,9 @@ from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCa
 from core.model_manager import ModelInstance
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
-from dify_graph.file import File
-from dify_graph.model_runtime.entities.message_entities import ImagePromptMessageContent
 from extensions.ext_database import db
+from graphon.file import File
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
 from models.model import App, Message
 
 logger = logging.getLogger(__name__)
@@ -181,7 +181,6 @@ class CompletionAppRunner(AppRunner):
             model_parameters=application_generate_entity.model_conf.parameters,
             stop=stop,
             stream=application_generate_entity.stream,
-            user=application_generate_entity.user_id,
         )
 
         # handle invoke result

api/core/app/apps/draft_variable_saver.py

@@ -4,31 +4,30 @@ import abc
 from collections.abc import Mapping
 from typing import Any, Protocol
 
-from sqlalchemy.orm import Session
-
-from dify_graph.enums import NodeType
+from graphon.enums import NodeType
 
 
 class DraftVariableSaver(Protocol):
     @abc.abstractmethod
-    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None):
-        pass
+    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None) -> None:
+        """Persist node draft variables for a completed execution."""
+        raise NotImplementedError
 
 
 class DraftVariableSaverFactory(Protocol):
     @abc.abstractmethod
     def __call__(
         self,
-        session: Session,
         app_id: str,
         node_id: str,
         node_type: NodeType,
         node_execution_id: str,
         enclosing_node_id: str | None = None,
     ) -> DraftVariableSaver:
-        pass
+        """Build a saver bound to a concrete node execution."""
+        raise NotImplementedError
 
 
 class NoopDraftVariableSaver(DraftVariableSaver):
-    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None):
-        pass
+    def save(self, process_data: Mapping[str, Any] | None, outputs: Mapping[str, Any] | None) -> None:
+        return None

api/core/app/apps/message_based_app_generator.py

@@ -28,6 +28,7 @@ from core.app.entities.task_entities import (
 )
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.file_reference import resolve_file_record_id
 from extensions.ext_database import db
 from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic
@@ -227,7 +228,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
                     transfer_method=file.transfer_method,
                     belongs_to=MessageFileBelongsTo.USER,
                     url=file.remote_url,
-                    upload_file_id=file.related_id,
+                    upload_file_id=resolve_file_record_id(file.reference),
                     created_by_role=(CreatorUserRole.ACCOUNT if account_id else CreatorUserRole.END_USER),
                     created_by=account_id or end_user_id or "",
                 )

api/core/app/apps/pipeline/pipeline_generator.py

@@ -18,6 +18,7 @@ import contexts
 from configs import dify_config
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.pipeline.pipeline_config_manager import PipelineConfigManager
 from core.app.apps.pipeline.pipeline_queue_manager import PipelineQueueManager
@@ -34,13 +35,14 @@ from core.datasource.entities.datasource_entities import (
 from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
 from core.entities.knowledge_entities import PipelineDataset, PipelineDocument
 from core.rag.index_processor.constant.built_in_field import BuiltInField
-from core.repositories.factory import DifyCoreRepositoryFactory
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
-from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
+from core.repositories.factory import (
+    DifyCoreRepositoryFactory,
+    WorkflowExecutionRepository,
+    WorkflowNodeExecutionRepository,
+)
 from extensions.ext_database import db
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline

api/core/app/apps/pipeline/pipeline_runner.py

@@ -12,19 +12,19 @@ from core.app.entities.app_invoke_entities import (
     build_dify_run_context,
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id
+from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
+from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from dify_graph.entities.graph_init_params import GraphInitParams
-from dify_graph.enums import WorkflowType
-from dify_graph.graph import Graph
-from dify_graph.graph_events import GraphEngineEvent, GraphRunFailedEvent
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.runtime import GraphRuntimeState, VariablePool
-from dify_graph.system_variable import SystemVariable
-from dify_graph.variable_loader import VariableLoader
-from dify_graph.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from extensions.ext_database import db
+from graphon.entities.graph_init_params import GraphInitParams
+from graphon.enums import WorkflowType
+from graphon.graph import Graph
+from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
+from graphon.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from models.dataset import Document, Pipeline
 from models.model import EndUser
 from models.workflow import Workflow
@@ -112,7 +112,7 @@ class PipelineRunner(WorkflowBasedAppRunner):
             files = self.application_generate_entity.files
 
             # Create a variable pool.
-            system_inputs = SystemVariable(
+            system_inputs = build_system_variables(
                 files=files,
                 user_id=user_id,
                 app_id=app_config.app_id,
@@ -142,19 +142,25 @@ class PipelineRunner(WorkflowBasedAppRunner):
                             )
                         )
 
-            variable_pool = VariablePool(
-                system_variables=system_inputs,
-                user_inputs=inputs,
-                environment_variables=workflow.environment_variables,
-                conversation_variables=[],
-                rag_pipeline_variables=rag_pipeline_variables,
+            variable_pool = VariablePool()
+            add_variables_to_pool(
+                variable_pool,
+                build_bootstrap_variables(
+                    system_variables=system_inputs,
+                    environment_variables=workflow.environment_variables,
+                    rag_pipeline_variables=rag_pipeline_variables,
+                ),
             )
+            root_node_id = self.application_generate_entity.start_node_id or get_default_root_node_id(
+                workflow.graph_dict
+            )
+            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=inputs)
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
 
             # init graph
             graph = self._init_rag_pipeline_graph(
                 graph_runtime_state=graph_runtime_state,
-                start_node_id=self.application_generate_entity.start_node_id,
+                start_node_id=root_node_id,
                 workflow=workflow,
                 user_from=user_from,
                 invoke_from=invoke_from,

api/core/app/apps/workflow/app_generator.py

@@ -17,6 +17,7 @@ from configs import dify_config
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
@@ -30,15 +31,13 @@ from core.db.session_factory import session_factory
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
-from dify_graph.graph_engine.layers.base import GraphEngineLayer
-from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
-from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.runtime import GraphRuntimeState
-from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
+from graphon.graph_engine.layers.base import GraphEngineLayer
+from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
+from graphon.runtime import GraphRuntimeState
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from libs.flask_utils import preserve_flask_contexts
 from models.account import Account
 from models.enums import WorkflowRunTriggeredFrom
@@ -129,107 +128,109 @@ class WorkflowAppGenerator(BaseAppGenerator):
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
-        files: Sequence[Mapping[str, Any]] = args.get("files") or []
+        with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
-        # parse files
-        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
-        # for better separation of concerns.
-        #
-        # For implementation reference, see the `_parse_file` function and
-        # `DraftWorkflowNodeRunApi` class which handle this properly.
-        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
-        system_files = file_factory.build_from_mappings(
-            mappings=files,
-            tenant_id=app_model.tenant_id,
-            config=file_extra_config,
-            strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-        )
-
-        # convert to app config
-        app_config = WorkflowAppConfigManager.get_app_config(
-            app_model=app_model,
-            workflow=workflow,
-        )
-
-        # get tracing instance
-        trace_manager = TraceQueueManager(
-            app_id=app_model.id,
-            user_id=user.id if isinstance(user, Account) else user.session_id,
-        )
-
-        inputs: Mapping[str, Any] = args["inputs"]
-
-        extras = {
-            **extract_external_trace_id_from_args(args),
-        }
-        workflow_run_id = str(workflow_run_id or uuid.uuid4())
-        # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
-        # trigger shouldn't prepare user inputs
-        if self._should_prepare_user_inputs(args):
-            inputs = self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
+            # parse files
+            # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+            # for better separation of concerns.
+            #
+            # For implementation reference, see the `_parse_file` function and
+            # `DraftWorkflowNodeRunApi` class which handle this properly.
+            file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+            system_files = file_factory.build_from_mappings(
+                mappings=files,
                 tenant_id=app_model.tenant_id,
+                config=file_extra_config,
                 strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+                access_controller=self._file_access_controller,
             )
-        # init application generate entity
-        application_generate_entity = WorkflowAppGenerateEntity(
-            task_id=str(uuid.uuid4()),
-            app_config=app_config,
-            file_upload_config=file_extra_config,
-            inputs=inputs,
-            files=list(system_files),
-            user_id=user.id,
-            stream=streaming,
-            invoke_from=invoke_from,
-            call_depth=call_depth,
-            trace_manager=trace_manager,
-            workflow_execution_id=workflow_run_id,
-            extras=extras,
-        )
 
-        contexts.plugin_tool_providers.set({})
-        contexts.plugin_tool_providers_lock.set(threading.Lock())
+            # convert to app config
+            app_config = WorkflowAppConfigManager.get_app_config(
+                app_model=app_model,
+                workflow=workflow,
+            )
 
-        # Create repositories
-        #
-        # Create session factory
-        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
-        # Create workflow execution(aka workflow run) repository
-        if triggered_from is not None:
-            # Use explicitly provided triggered_from (for async triggers)
-            workflow_triggered_from = triggered_from
-        elif invoke_from == InvokeFrom.DEBUGGER:
-            workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
-        else:
-            workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
-        workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=workflow_triggered_from,
-        )
-        # Create workflow node execution repository
-        workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
-            session_factory=session_factory,
-            user=user,
-            app_id=application_generate_entity.app_config.app_id,
-            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-        )
+            # get tracing instance
+            trace_manager = TraceQueueManager(
+                app_id=app_model.id,
+                user_id=user.id if isinstance(user, Account) else user.session_id,
+            )
 
-        return self._generate(
-            app_model=app_model,
-            workflow=workflow,
-            user=user,
-            application_generate_entity=application_generate_entity,
-            invoke_from=invoke_from,
-            workflow_execution_repository=workflow_execution_repository,
-            workflow_node_execution_repository=workflow_node_execution_repository,
-            streaming=streaming,
-            root_node_id=root_node_id,
-            graph_engine_layers=graph_engine_layers,
-            pause_state_config=pause_state_config,
-        )
+            inputs: Mapping[str, Any] = args["inputs"]
+
+            extras = {
+                **extract_external_trace_id_from_args(args),
+            }
+            workflow_run_id = str(workflow_run_id or uuid.uuid4())
+            # FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
+            # trigger shouldn't prepare user inputs
+            if self._should_prepare_user_inputs(args):
+                inputs = self._prepare_user_inputs(
+                    user_inputs=inputs,
+                    variables=app_config.variables,
+                    tenant_id=app_model.tenant_id,
+                    strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+                )
+            # init application generate entity
+            application_generate_entity = WorkflowAppGenerateEntity(
+                task_id=str(uuid.uuid4()),
+                app_config=app_config,
+                file_upload_config=file_extra_config,
+                inputs=inputs,
+                files=list(system_files),
+                user_id=user.id,
+                stream=streaming,
+                invoke_from=invoke_from,
+                call_depth=call_depth,
+                trace_manager=trace_manager,
+                workflow_execution_id=workflow_run_id,
+                extras=extras,
+            )
+
+            contexts.plugin_tool_providers.set({})
+            contexts.plugin_tool_providers_lock.set(threading.Lock())
+
+            # Create repositories
+            #
+            # Create session factory
+            session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+            # Create workflow execution(aka workflow run) repository
+            if triggered_from is not None:
+                # Use explicitly provided triggered_from (for async triggers)
+                workflow_triggered_from = triggered_from
+            elif invoke_from == InvokeFrom.DEBUGGER:
+                workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+            else:
+                workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+            workflow_execution_repository = DifyCoreRepositoryFactory.create_workflow_execution_repository(
+                session_factory=session_factory,
+                user=user,
+                app_id=application_generate_entity.app_config.app_id,
+                triggered_from=workflow_triggered_from,
+            )
+            # Create workflow node execution repository
+            workflow_node_execution_repository = DifyCoreRepositoryFactory.create_workflow_node_execution_repository(
+                session_factory=session_factory,
+                user=user,
+                app_id=application_generate_entity.app_config.app_id,
+                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+            )
+
+            return self._generate(
+                app_model=app_model,
+                workflow=workflow,
+                user=user,
+                application_generate_entity=application_generate_entity,
+                invoke_from=invoke_from,
+                workflow_execution_repository=workflow_execution_repository,
+                workflow_node_execution_repository=workflow_node_execution_repository,
+                streaming=streaming,
+                root_node_id=root_node_id,
+                graph_engine_layers=graph_engine_layers,
+                pause_state_config=pause_state_config,
+            )
 
     def resume(
         self,
@@ -292,62 +293,67 @@ class WorkflowAppGenerator(BaseAppGenerator):
         :param workflow_node_execution_repository: repository for workflow node execution
         :param streaming: is stream
         """
-        graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
+        with self._bind_file_access_scope(
+            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
+            invoke_from=invoke_from,
+        ):
+            graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
 
-        # init queue manager
-        queue_manager = WorkflowAppQueueManager(
-            task_id=application_generate_entity.task_id,
-            user_id=application_generate_entity.user_id,
-            invoke_from=application_generate_entity.invoke_from,
-            app_mode=app_model.mode,
-        )
-
-        if pause_state_config is not None:
-            graph_layers.append(
-                PauseStatePersistenceLayer(
-                    session_factory=pause_state_config.session_factory,
-                    generate_entity=application_generate_entity,
-                    state_owner_user_id=pause_state_config.state_owner_user_id,
-                )
+            # init queue manager
+            queue_manager = WorkflowAppQueueManager(
+                task_id=application_generate_entity.task_id,
+                user_id=application_generate_entity.user_id,
+                invoke_from=application_generate_entity.invoke_from,
+                app_mode=app_model.mode,
             )
 
-        # new thread with request context and contextvars
-        context = contextvars.copy_context()
+            if pause_state_config is not None:
+                graph_layers.append(
+                    PauseStatePersistenceLayer(
+                        session_factory=pause_state_config.session_factory,
+                        generate_entity=application_generate_entity,
+                        state_owner_user_id=pause_state_config.state_owner_user_id,
+                    )
+                )
 
-        # release database connection, because the following new thread operations may take a long time
-        db.session.close()
+            # new thread with request context and contextvars
+            context = contextvars.copy_context()
 
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "context": context,
-                "variable_loader": variable_loader,
-                "root_node_id": root_node_id,
-                "workflow_execution_repository": workflow_execution_repository,
-                "workflow_node_execution_repository": workflow_node_execution_repository,
-                "graph_engine_layers": tuple(graph_layers),
-                "graph_runtime_state": graph_runtime_state,
-            },
-        )
+            # release database connection, because the following new thread operations may take a long time
+            db.session.close()
 
-        worker_thread.start()
+            worker_thread = threading.Thread(
+                target=self._generate_worker,
+                kwargs={
+                    "flask_app": current_app._get_current_object(),  # type: ignore
+                    "application_generate_entity": application_generate_entity,
+                    "queue_manager": queue_manager,
+                    "context": context,
+                    "variable_loader": variable_loader,
+                    "root_node_id": root_node_id,
+                    "workflow_execution_repository": workflow_execution_repository,
+                    "workflow_node_execution_repository": workflow_node_execution_repository,
+                    "graph_engine_layers": tuple(graph_layers),
+                    "graph_runtime_state": graph_runtime_state,
+                },
+            )
 
-        draft_var_saver_factory = self._get_draft_var_saver_factory(invoke_from, user)
+            worker_thread.start()
 
-        # return response or stream generator
-        response = self._handle_response(
-            application_generate_entity=application_generate_entity,
-            workflow=workflow,
-            queue_manager=queue_manager,
-            user=user,
-            draft_var_saver_factory=draft_var_saver_factory,
-            stream=streaming,
-        )
+            draft_var_saver_factory = self._get_draft_var_saver_factory(invoke_from, user)
 
-        return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            # return response or stream generator
+            response = self._handle_response(
+                application_generate_entity=application_generate_entity,
+                workflow=workflow,
+                queue_manager=queue_manager,
+                user=user,
+                draft_var_saver_factory=draft_var_saver_factory,
+                stream=streaming,
+            )
+
+            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
 
     def single_iteration_generate(
         self,

api/core/app/apps/workflow/app_runner.py

@@ -8,17 +8,18 @@ from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
+from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.node_factory import get_default_root_node_id
+from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
+from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from dify_graph.enums import WorkflowType
-from dify_graph.graph_engine.command_channels.redis_channel import RedisChannel
-from dify_graph.graph_engine.layers.base import GraphEngineLayer
-from dify_graph.repositories.workflow_execution_repository import WorkflowExecutionRepository
-from dify_graph.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from dify_graph.runtime import GraphRuntimeState, VariablePool
-from dify_graph.system_variable import SystemVariable
-from dify_graph.variable_loader import VariableLoader
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
+from graphon.enums import WorkflowType
+from graphon.graph_engine.command_channels.redis_channel import RedisChannel
+from graphon.graph_engine.layers.base import GraphEngineLayer
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import VariableLoader
 from libs.datetime_utils import naive_utc_now
 from models.workflow import Workflow
 
@@ -96,7 +97,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             inputs = self.application_generate_entity.inputs
 
             # Create a variable pool.
-            system_inputs = SystemVariable(
+            system_inputs = build_system_variables(
                 files=self.application_generate_entity.files,
                 user_id=self._sys_user_id,
                 app_id=app_config.app_id,
@@ -104,12 +105,16 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_id=app_config.workflow_id,
                 workflow_execution_id=self.application_generate_entity.workflow_execution_id,
             )
-            variable_pool = VariablePool(
-                system_variables=system_inputs,
-                user_inputs=inputs,
-                environment_variables=self._workflow.environment_variables,
-                conversation_variables=[],
+            variable_pool = VariablePool()
+            add_variables_to_pool(
+                variable_pool,
+                build_bootstrap_variables(
+                    system_variables=system_inputs,
+                    environment_variables=self._workflow.environment_variables,
+                ),
             )
+            root_node_id = self._root_node_id or get_default_root_node_id(self._workflow.graph_dict)
+            add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=inputs)
 
             graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
             graph = self._init_graph(
@@ -120,7 +125,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 user_id=self.application_generate_entity.user_id,
                 user_from=user_from,
                 invoke_from=invoke_from,
-                root_node_id=self._root_node_id,
+                root_node_id=root_node_id,
             )
 
         # RUN WORKFLOW

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -10,6 +10,7 @@ from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.common.graph_runtime_state_support import GraphRuntimeStateSupport
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+from core.app.apps.draft_variable_saver import DraftVariableSaverFactory
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.queue_entities import (
     AppQueueEvent,
@@ -55,12 +56,11 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
-from dify_graph.entities.workflow_start_reason import WorkflowStartReason
-from dify_graph.enums import WorkflowExecutionStatus
-from dify_graph.repositories.draft_variable_repository import DraftVariableSaverFactory
-from dify_graph.runtime import GraphRuntimeState
-from dify_graph.system_variable import SystemVariable
+from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
+from graphon.entities.workflow_start_reason import WorkflowStartReason
+from graphon.enums import WorkflowExecutionStatus
+from graphon.runtime import GraphRuntimeState
 from models import Account
 from models.enums import CreatorUserRole
 from models.model import EndUser
@@ -104,7 +104,7 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         self._invoke_from = queue_manager.invoke_from
         self._draft_var_saver_factory = draft_var_saver_factory
         self._workflow = workflow
-        self._workflow_system_variables = SystemVariable(
+        self._workflow_system_variables = build_system_variables(
             files=application_generate_entity.files,
             user_id=user_session_id,
             app_id=application_generate_entity.app_config.app_id,
@@ -728,13 +728,11 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         return response
 
     def _save_output_for_event(self, event: QueueNodeSucceededEvent | QueueNodeExceptionEvent, node_execution_id: str):
-        with Session(db.engine) as session, session.begin():
-            saver = self._draft_var_saver_factory(
-                session=session,
-                app_id=self._application_generate_entity.app_config.app_id,
-                node_id=event.node_id,
-                node_type=event.node_type,
-                node_execution_id=node_execution_id,
-                enclosing_node_id=event.in_loop_id or event.in_iteration_id,
-            )
-            saver.save(event.process_data, event.outputs)
+        saver = self._draft_var_saver_factory(
+            app_id=self._application_generate_entity.app_config.app_id,
+            node_id=event.node_id,
+            node_type=event.node_type,
+            node_execution_id=node_execution_id,
+            enclosing_node_id=event.in_loop_id or event.in_iteration_id,
+        )
+        saver.save(event.process_data, event.outputs)

api/core/app/apps/workflow_app_runner.py

@@ -34,13 +34,22 @@ from core.app.entities.queue_entities import (
 )
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id, resolve_workflow_node_class
+from core.workflow.system_variables import (
+    build_bootstrap_variables,
+    default_system_variables,
+    get_node_creation_preload_selectors,
+    inject_default_system_variable_mappings,
+    preload_node_creation_variables,
+)
+from core.workflow.variable_pool_initializer import add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
-from dify_graph.entities import GraphInitParams
-from dify_graph.entities.graph_config import NodeConfigDictAdapter
-from dify_graph.entities.pause_reason import HumanInputRequired
-from dify_graph.graph import Graph
-from dify_graph.graph_engine.layers.base import GraphEngineLayer
-from dify_graph.graph_events import (
+from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
+from graphon.entities import GraphInitParams
+from graphon.entities.graph_config import NodeConfigDictAdapter
+from graphon.entities.pause_reason import HumanInputRequired
+from graphon.graph import Graph
+from graphon.graph_engine.layers.base import GraphEngineLayer
+from graphon.graph_events import (
     GraphEngineEvent,
     GraphRunFailedEvent,
     GraphRunPartialSucceededEvent,
@@ -66,10 +75,9 @@ from dify_graph.graph_events import (
     NodeRunStreamChunkEvent,
     NodeRunSucceededEvent,
 )
-from dify_graph.graph_events.graph import GraphRunAbortedEvent
-from dify_graph.runtime import GraphRuntimeState, VariablePool
-from dify_graph.system_variable import SystemVariable
-from dify_graph.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
+from graphon.graph_events.graph import GraphRunAbortedEvent
+from graphon.runtime import GraphRuntimeState, VariablePool
+from graphon.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from models.workflow import Workflow
 from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
 
@@ -173,14 +181,15 @@ class WorkflowBasedAppRunner:
             ValueError: If neither single_iteration_run nor single_loop_run is specified
         """
         # Create initial runtime state with variable pool containing environment variables
-        graph_runtime_state = GraphRuntimeState(
-            variable_pool=VariablePool(
-                system_variables=SystemVariable.default(),
-                user_inputs={},
+        variable_pool = VariablePool()
+        add_variables_to_pool(
+            variable_pool,
+            build_bootstrap_variables(
+                system_variables=default_system_variables(),
                 environment_variables=workflow.environment_variables,
             ),
-            start_at=time.time(),
         )
+        graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
 
         # Determine which type of single node execution and get graph/variable_pool
         if single_iteration_run:
@@ -272,6 +281,8 @@ class WorkflowBasedAppRunner:
 
         graph_config["edges"] = edge_configs
 
+        typed_node_configs = [NodeConfigDictAdapter.validate_python(node) for node in node_configs]
+
         # Create required parameters for Graph.init
         graph_init_params = GraphInitParams(
             workflow_id=workflow.id,
@@ -291,26 +302,15 @@ class WorkflowBasedAppRunner:
             graph_runtime_state=graph_runtime_state,
         )
 
-        # init graph
-        graph = Graph.init(
-            graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, skip_validation=True
-        )
-
-        if not graph:
-            raise ValueError("graph not found in workflow")
-
-        # fetch node config from node id
         target_node_config = None
-        for node in node_configs:
-            if node.get("id") == node_id:
+        for node in typed_node_configs:
+            if node["id"] == node_id:
                 target_node_config = node
                 break
 
         if not target_node_config:
             raise ValueError(f"{node_type_label} node id not found in workflow graph")
 
-        target_node_config = NodeConfigDictAdapter.validate_python(target_node_config)
-
         # Get node class
         node_type = target_node_config["data"].type
         node_version = str(target_node_config["data"].version)
@@ -319,12 +319,31 @@ class WorkflowBasedAppRunner:
         # Use the variable pool from graph_runtime_state instead of creating a new one
         variable_pool = graph_runtime_state.variable_pool
 
+        preload_node_creation_variables(
+            variable_loader=self._variable_loader,
+            variable_pool=variable_pool,
+            selectors=[
+                selector
+                for node_config in typed_node_configs
+                for selector in get_node_creation_preload_selectors(
+                    node_type=node_config["data"].type,
+                    node_data=node_config["data"],
+                )
+            ],
+        )
+
         try:
             variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
                 graph_config=workflow.graph_dict, config=target_node_config
             )
         except NotImplementedError:
             variable_mapping = {}
+        variable_mapping = inject_default_system_variable_mappings(
+            node_id=target_node_config["id"],
+            node_type=node_type,
+            node_data=target_node_config["data"],
+            variable_mapping=variable_mapping,
+        )
 
         load_into_variable_pool(
             variable_loader=self._variable_loader,
@@ -340,6 +359,14 @@ class WorkflowBasedAppRunner:
             tenant_id=workflow.tenant_id,
         )
 
+        # init graph after constructor-time context has been loaded
+        graph = Graph.init(
+            graph_config=graph_config, node_factory=node_factory, root_node_id=node_id, skip_validation=True
+        )
+
+        if not graph:
+            raise ValueError("graph not found in workflow")
+
         return graph, variable_pool
 
     @staticmethod
@@ -408,7 +435,11 @@ class WorkflowBasedAppRunner:
             node_run_result = event.node_run_result
             inputs = node_run_result.inputs
             process_data = node_run_result.process_data
-            outputs = node_run_result.outputs
+            outputs = project_node_outputs_for_workflow_run(
+                node_type=event.node_type,
+                inputs=inputs,
+                outputs=node_run_result.outputs,
+            )
             execution_metadata = node_run_result.metadata
             self._publish_event(
                 QueueNodeRetryEvent(
@@ -448,7 +479,11 @@ class WorkflowBasedAppRunner:
             node_run_result = event.node_run_result
             inputs = node_run_result.inputs
             process_data = node_run_result.process_data
-            outputs = node_run_result.outputs
+            outputs = project_node_outputs_for_workflow_run(
+                node_type=event.node_type,
+                inputs=inputs,
+                outputs=node_run_result.outputs,
+            )
             execution_metadata = node_run_result.metadata
             self._publish_event(
                 QueueNodeSucceededEvent(
@@ -466,6 +501,11 @@ class WorkflowBasedAppRunner:
                 )
             )
         elif isinstance(event, NodeRunFailedEvent):
+            outputs = project_node_outputs_for_workflow_run(
+                node_type=event.node_type,
+                inputs=event.node_run_result.inputs,
+                outputs=event.node_run_result.outputs,
+            )
             self._publish_event(
                 QueueNodeFailedEvent(
                     node_execution_id=event.id,
@@ -475,7 +515,7 @@ class WorkflowBasedAppRunner:
                     finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
                     process_data=event.node_run_result.process_data,
-                    outputs=event.node_run_result.outputs,
+                    outputs=outputs,
                     error=event.node_run_result.error or "Unknown error",
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,
@@ -483,6 +523,11 @@ class WorkflowBasedAppRunner:
                 )
             )
         elif isinstance(event, NodeRunExceptionEvent):
+            outputs = project_node_outputs_for_workflow_run(
+                node_type=event.node_type,
+                inputs=event.node_run_result.inputs,
+                outputs=event.node_run_result.outputs,
+            )
             self._publish_event(
                 QueueNodeExceptionEvent(
                     node_execution_id=event.id,
@@ -492,7 +537,7 @@ class WorkflowBasedAppRunner:
                     finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
                     process_data=event.node_run_result.process_data,
-                    outputs=event.node_run_result.outputs,
+                    outputs=outputs,
                     error=event.node_run_result.error or "Unknown error",
                     execution_metadata=event.node_run_result.metadata,
                     in_iteration_id=event.in_iteration_id,

api/core/app/entities/app_invoke_entities.py

@@ -7,14 +7,16 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validat
 from constants import UUID_NIL
 from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAppConfig
 from core.entities.provider_configuration import ProviderModelBundle
-from dify_graph.entities.graph_init_params import DIFY_RUN_CONTEXT_KEY
-from dify_graph.file import File, FileUploadConfig
-from dify_graph.model_runtime.entities.model_entities import AIModelEntity
+from graphon.file import File, FileUploadConfig
+from graphon.model_runtime.entities.model_entities import AIModelEntity
 
 if TYPE_CHECKING:
     from core.ops.ops_trace_manager import TraceQueueManager
 
 
+DIFY_RUN_CONTEXT_KEY = "_dify"
+
+
 class UserFrom(StrEnum):
     ACCOUNT = "account"
     END_USER = "end-user"