fix typo: Scrapper -> Scraper in webscraper.yaml

Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com>

api/.importlinter

@@ -29,8 +29,6 @@ ignore_imports =
 
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
     core.workflow.nodes.loop.loop_node -> core.app.workflow.node_factory
-    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.layers.llm_quota
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
 
     core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
     core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
@@ -54,6 +52,7 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
     core.workflow.nodes.llm.file_saver -> extensions.ext_database
+    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
     core.workflow.nodes.llm.node -> extensions.ext_database
     core.workflow.nodes.tool.tool_node -> extensions.ext_database
     # TODO(QuantumGhost): use DI to avoid depending on global DB.
@@ -108,11 +107,14 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> core.tools.tool_manager
     core.workflow.nodes.document_extractor.node -> core.helper.ssrf_proxy
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
-    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.layers.llm_quota
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
+    core.workflow.nodes.llm.llm_utils -> configs
     core.workflow.nodes.llm.llm_utils -> core.model_manager
     core.workflow.nodes.llm.protocols -> core.model_manager
     core.workflow.nodes.llm.llm_utils -> core.model_runtime.model_providers.__base.large_language_model
+    core.workflow.nodes.llm.llm_utils -> models.model
+    core.workflow.nodes.llm.llm_utils -> models.provider
+    core.workflow.nodes.llm.llm_utils -> services.credit_pool_service
     core.workflow.nodes.llm.node -> core.tools.signature
     core.workflow.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
     core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
@@ -129,10 +131,12 @@ ignore_imports =
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.model_providers.__base.large_language_model
     core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
+    core.workflow.nodes.start.entities -> core.app.app_config.entities
+    core.workflow.nodes.start.start_node -> core.app.app_config.entities
     core.workflow.workflow_entry -> core.app.apps.exc
     core.workflow.workflow_entry -> core.app.entities.app_invoke_entities
-    core.workflow.workflow_entry -> core.app.workflow.layers.llm_quota
     core.workflow.workflow_entry -> core.app.workflow.node_factory
+    core.workflow.nodes.llm.llm_utils -> core.entities.provider_entities
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
     core.workflow.nodes.question_classifier.question_classifier_node -> core.model_manager
     core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
@@ -146,6 +150,7 @@ ignore_imports =
     core.workflow.nodes.llm.node -> core.model_manager
     core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
     core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.llm_utils -> core.prompt.entities.advanced_prompt_entities
     core.workflow.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
     core.workflow.nodes.llm.node -> core.prompt.utils.prompt_message_util
     core.workflow.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
@@ -167,6 +172,7 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
     core.workflow.nodes.llm.file_saver -> extensions.ext_database
+    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
     core.workflow.nodes.llm.node -> extensions.ext_database
     core.workflow.nodes.tool.tool_node -> extensions.ext_database
     core.workflow.nodes.human_input.human_input_node -> extensions.ext_database
@@ -174,7 +180,7 @@ ignore_imports =
     core.workflow.workflow_entry -> extensions.otel.runtime
     core.workflow.nodes.agent.agent_node -> models
     core.workflow.nodes.base.node -> models.enums
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
+    core.workflow.nodes.llm.llm_utils -> models.provider_ids
     core.workflow.nodes.llm.node -> models.model
     core.workflow.workflow_entry -> models.enums
     core.workflow.nodes.agent.agent_node -> services

api/controllers/mcp/mcp.py

@@ -8,9 +8,9 @@ from sqlalchemy.orm import Session
 from controllers.common.schema import register_schema_model
 from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
+from core.app.app_config.entities import VariableEntity
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request
-from core.workflow.variables.input_entities import VariableEntity
 from extensions.ext_database import db
 from libs import helper
 from models.model import App, AppMCPServer, AppMode, EndUser

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

@@ -1,8 +1,7 @@
 import re
 
-from core.app.app_config.entities import ExternalDataVariableEntity
+from core.app.app_config.entities import ExternalDataVariableEntity, VariableEntity, VariableEntityType
 from core.external_data_tool.factory import ExternalDataToolFactory
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 
 _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(
     [

api/core/app/app_config/entities.py

@@ -2,12 +2,12 @@ from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field
+from jsonschema import Draft7Validator, SchemaError
+from pydantic import BaseModel, Field, field_validator
 
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.workflow.file import FileUploadConfig
-from core.workflow.variables.input_entities import VariableEntity as WorkflowVariableEntity
+from core.workflow.file import FileTransferMethod, FileType, FileUploadConfig
 from models.model import AppMode
 
 
@@ -90,7 +90,61 @@ class PromptTemplateEntity(BaseModel):
     advanced_completion_prompt_template: AdvancedCompletionPromptTemplateEntity | None = None
 
 
-class RagPipelineVariableEntity(WorkflowVariableEntity):
+class VariableEntityType(StrEnum):
+    TEXT_INPUT = "text-input"
+    SELECT = "select"
+    PARAGRAPH = "paragraph"
+    NUMBER = "number"
+    EXTERNAL_DATA_TOOL = "external_data_tool"
+    FILE = "file"
+    FILE_LIST = "file-list"
+    CHECKBOX = "checkbox"
+    JSON_OBJECT = "json_object"
+
+
+class VariableEntity(BaseModel):
+    """
+    Variable Entity.
+    """
+
+    # `variable` records the name of the variable in user inputs.
+    variable: str
+    label: str
+    description: str = ""
+    type: VariableEntityType
+    required: bool = False
+    hide: bool = False
+    default: Any = None
+    max_length: int | None = None
+    options: Sequence[str] = Field(default_factory=list)
+    allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
+    allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
+    allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
+    json_schema: dict | None = Field(default=None)
+
+    @field_validator("description", mode="before")
+    @classmethod
+    def convert_none_description(cls, v: Any) -> str:
+        return v or ""
+
+    @field_validator("options", mode="before")
+    @classmethod
+    def convert_none_options(cls, v: Any) -> Sequence[str]:
+        return v or []
+
+    @field_validator("json_schema")
+    @classmethod
+    def validate_json_schema(cls, schema: dict | None) -> dict | None:
+        if schema is None:
+            return None
+        try:
+            Draft7Validator.check_schema(schema)
+        except SchemaError as e:
+            raise ValueError(f"Invalid JSON schema: {e.message}")
+        return schema
+
+
+class RagPipelineVariableEntity(VariableEntity):
     """
     Rag Pipeline Variable Entity.
     """
@@ -260,7 +314,7 @@ class AppConfig(BaseModel):
     app_id: str
     app_mode: AppMode
     additional_features: AppAdditionalFeatures | None = None
-    variables: list[WorkflowVariableEntity] = []
+    variables: list[VariableEntity] = []
     sensitive_word_avoidance: SensitiveWordAvoidanceEntity | None = None
 
 

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

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

api/core/app/apps/base_app_generator.py

@@ -3,6 +3,7 @@ from typing import TYPE_CHECKING, Any, Union, final
 
 from sqlalchemy.orm import Session
 
+from core.app.app_config.entities import VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.workflow.enums import NodeType
 from core.workflow.file import File, FileUploadConfig
@@ -11,14 +12,13 @@ from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaverFactory,
     NoopDraftVariableSaver,
 )
-from core.workflow.variables.input_entities import VariableEntityType
 from factories import file_factory
 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 core.workflow.variables.input_entities import VariableEntity
+    from core.app.app_config.entities import VariableEntity
 
 
 class BaseAppGenerator:

api/core/app/llm/__init__.py

@@ -1,5 +1 @@
 """LLM-related application services."""
-
-from .quota import deduct_llm_quota, ensure_llm_quota_available
-
-__all__ = ["deduct_llm_quota", "ensure_llm_quota_available"]

api/core/app/llm/quota.py

@@ -1,93 +0,0 @@
-from sqlalchemy import update
-from sqlalchemy.orm import Session
-
-from configs import dify_config
-from core.entities.model_entities import ModelStatus
-from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
-from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
-from core.model_runtime.entities.llm_entities import LLMUsage
-from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
-from models.provider import Provider, ProviderType
-from models.provider_ids import ModelProviderID
-
-
-def ensure_llm_quota_available(*, model_instance: ModelInstance) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
-
-    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
-        return
-
-    provider_model = provider_configuration.get_provider_model(
-        model_type=model_instance.model_type_instance.model_type,
-        model=model_instance.model_name,
-    )
-    if provider_model and provider_model.status == ModelStatus.QUOTA_EXCEEDED:
-        raise QuotaExceededError(f"Model provider {model_instance.provider} quota exceeded.")
-
-
-def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LLMUsage) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
-
-    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
-        return
-
-    system_configuration = provider_configuration.system_configuration
-
-    quota_unit = None
-    for quota_configuration in system_configuration.quota_configurations:
-        if quota_configuration.quota_type == system_configuration.current_quota_type:
-            quota_unit = quota_configuration.quota_unit
-
-            if quota_configuration.quota_limit == -1:
-                return
-
-            break
-
-    used_quota = None
-    if quota_unit:
-        if quota_unit == QuotaUnit.TOKENS:
-            used_quota = usage.total_tokens
-        elif quota_unit == QuotaUnit.CREDITS:
-            used_quota = dify_config.get_model_credits(model_instance.model_name)
-        else:
-            used_quota = 1
-
-    if used_quota is not None and system_configuration.current_quota_type is not None:
-        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
-            from services.credit_pool_service import CreditPoolService
-
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-            )
-        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
-            from services.credit_pool_service import CreditPoolService
-
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-                pool_type="paid",
-            )
-        else:
-            with Session(db.engine) as session:
-                stmt = (
-                    update(Provider)
-                    .where(
-                        Provider.tenant_id == tenant_id,
-                        # TODO: Use provider name with prefix after the data migration.
-                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                        Provider.provider_type == ProviderType.SYSTEM.value,
-                        Provider.quota_type == system_configuration.current_quota_type.value,
-                        Provider.quota_limit > Provider.quota_used,
-                    )
-                    .values(
-                        quota_used=Provider.quota_used + used_quota,
-                        last_used=naive_utc_now(),
-                    )
-                )
-                session.execute(stmt)
-                session.commit()

api/core/app/workflow/layers/__init__.py

@@ -1,11 +1,9 @@
 """Workflow-level GraphEngine layers that depend on outer infrastructure."""
 
-from .llm_quota import LLMQuotaLayer
 from .observability import ObservabilityLayer
 from .persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 
 __all__ = [
-    "LLMQuotaLayer",
     "ObservabilityLayer",
     "PersistenceWorkflowInfo",
     "WorkflowPersistenceLayer",

api/core/app/workflow/layers/llm_quota.py

@@ -1,128 +0,0 @@
-"""
-LLM quota deduction layer for GraphEngine.
-
-This layer centralizes model-quota deduction outside node implementations.
-"""
-
-import logging
-from typing import TYPE_CHECKING, cast, final
-
-from typing_extensions import override
-
-from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
-from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
-from core.workflow.enums import NodeType
-from core.workflow.graph_engine.entities.commands import AbortCommand, CommandType
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events import GraphEngineEvent, GraphNodeEventBase
-from core.workflow.graph_events.node import NodeRunSucceededEvent
-from core.workflow.nodes.base.node import Node
-
-if TYPE_CHECKING:
-    from core.workflow.nodes.llm.node import LLMNode
-    from core.workflow.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
-    from core.workflow.nodes.question_classifier.question_classifier_node import QuestionClassifierNode
-
-logger = logging.getLogger(__name__)
-
-
-@final
-class LLMQuotaLayer(GraphEngineLayer):
-    """Graph layer that applies LLM quota deduction after node execution."""
-
-    def __init__(self) -> None:
-        super().__init__()
-        self._abort_sent = False
-
-    @override
-    def on_graph_start(self) -> None:
-        self._abort_sent = False
-
-    @override
-    def on_event(self, event: GraphEngineEvent) -> None:
-        _ = event
-
-    @override
-    def on_graph_end(self, error: Exception | None) -> None:
-        _ = error
-
-    @override
-    def on_node_run_start(self, node: Node) -> None:
-        if self._abort_sent:
-            return
-
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
-            return
-
-        try:
-            ensure_llm_quota_available(model_instance=model_instance)
-        except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
-            logger.warning("LLM quota check failed, node_id=%s, error=%s", node.id, exc)
-
-    @override
-    def on_node_run_end(
-        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
-    ) -> None:
-        if error is not None or not isinstance(result_event, NodeRunSucceededEvent):
-            return
-
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
-            return
-
-        try:
-            deduct_llm_quota(
-                tenant_id=node.tenant_id,
-                model_instance=model_instance,
-                usage=result_event.node_run_result.llm_usage,
-            )
-        except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
-            logger.warning("LLM quota deduction exceeded, node_id=%s, error=%s", node.id, exc)
-        except Exception:
-            logger.exception("LLM quota deduction failed, node_id=%s", node.id)
-
-    @staticmethod
-    def _set_stop_event(node: Node) -> None:
-        stop_event = getattr(node.graph_runtime_state, "stop_event", None)
-        if stop_event is not None:
-            stop_event.set()
-
-    def _send_abort_command(self, *, reason: str) -> None:
-        if not self.command_channel or self._abort_sent:
-            return
-
-        try:
-            self.command_channel.send_command(
-                AbortCommand(
-                    command_type=CommandType.ABORT,
-                    reason=reason,
-                )
-            )
-            self._abort_sent = True
-        except Exception:
-            logger.exception("Failed to send quota abort command")
-
-    @staticmethod
-    def _extract_model_instance(node: Node) -> ModelInstance | None:
-        try:
-            match node.node_type:
-                case NodeType.LLM:
-                    return cast("LLMNode", node).model_instance
-                case NodeType.PARAMETER_EXTRACTOR:
-                    return cast("ParameterExtractorNode", node).model_instance
-                case NodeType.QUESTION_CLASSIFIER:
-                    return cast("QuestionClassifierNode", node).model_instance
-                case _:
-                    return None
-        except AttributeError:
-            logger.warning(
-                "LLMQuotaLayer skipped quota deduction because node does not expose a model instance, node_id=%s",
-                node.id,
-            )
-            return None

api/core/app/workflow/node_factory.py

@@ -1,8 +1,6 @@
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, cast, final
 
-from sqlalchemy import select
-from sqlalchemy.orm import Session
 from typing_extensions import override
 
 from configs import dify_config
@@ -13,16 +11,14 @@ from core.helper.code_executor.code_executor import (
     CodeExecutor,
 )
 from core.helper.ssrf_proxy import ssrf_proxy
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.model_entities import ModelType
-from core.model_runtime.memory import PromptMessageMemory
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.entities.graph_config import NodeConfigDict
-from core.workflow.enums import NodeType, SystemVariableKey
+from core.workflow.enums import NodeType
 from core.workflow.file.file_manager import file_manager
 from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
@@ -33,9 +29,11 @@ from core.workflow.nodes.datasource import DatasourceNode
 from core.workflow.nodes.document_extractor import DocumentExtractorNode, UnstructuredApiConfig
 from core.workflow.nodes.http_request import HttpRequestNode, build_http_request_config
 from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
+from core.workflow.nodes.llm import llm_utils
 from core.workflow.nodes.llm.entities import ModelConfig
 from core.workflow.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
 from core.workflow.nodes.llm.node import LLMNode
+from core.workflow.nodes.llm.protocols import PromptMessageMemory
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from core.workflow.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
 from core.workflow.nodes.question_classifier.question_classifier_node import QuestionClassifierNode
@@ -43,34 +41,12 @@ from core.workflow.nodes.template_transform.template_renderer import (
     CodeExecutorJinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
-from core.workflow.variables.segments import StringSegment
-from extensions.ext_database import db
-from models.model import Conversation
 
 if TYPE_CHECKING:
     from core.workflow.entities import GraphInitParams
     from core.workflow.runtime import GraphRuntimeState
 
 
-def fetch_memory(
-    *,
-    conversation_id: str | None,
-    app_id: str,
-    node_data_memory: MemoryConfig | None,
-    model_instance: ModelInstance,
-) -> TokenBufferMemory | None:
-    if not node_data_memory or not conversation_id:
-        return None
-
-    with Session(db.engine, expire_on_commit=False) as session:
-        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
-        conversation = session.scalar(stmt)
-        if not conversation:
-            return None
-
-    return TokenBufferMemory(conversation=conversation, model_instance=model_instance)
-
-
 class DefaultWorkflowCodeExecutor:
     def execute(
         self,
@@ -245,7 +221,6 @@ class DifyNodeFactory(NodeFactory):
 
         if node_type == NodeType.QUESTION_CLASSIFIER:
             model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
             return QuestionClassifierNode(
                 id=node_id,
                 config=node_config,
@@ -254,12 +229,10 @@ class DifyNodeFactory(NodeFactory):
                 credentials_provider=self._llm_credentials_provider,
                 model_factory=self._llm_model_factory,
                 model_instance=model_instance,
-                memory=memory,
             )
 
         if node_type == NodeType.PARAMETER_EXTRACTOR:
             model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
             return ParameterExtractorNode(
                 id=node_id,
                 config=node_config,
@@ -268,7 +241,6 @@ class DifyNodeFactory(NodeFactory):
                 credentials_provider=self._llm_credentials_provider,
                 model_factory=self._llm_model_factory,
                 model_instance=model_instance,
-                memory=memory,
             )
 
         return node_class(
@@ -323,14 +295,8 @@ class DifyNodeFactory(NodeFactory):
             return None
 
         node_memory = MemoryConfig.model_validate(raw_memory_config)
-        conversation_id_variable = self.graph_runtime_state.variable_pool.get(
-            ["sys", SystemVariableKey.CONVERSATION_ID]
-        )
-        conversation_id = (
-            conversation_id_variable.value if isinstance(conversation_id_variable, StringSegment) else None
-        )
-        return fetch_memory(
-            conversation_id=conversation_id,
+        return llm_utils.fetch_memory(
+            variable_pool=self.graph_runtime_state.variable_pool,
             app_id=self.graph_init_params.app_id,
             node_data_memory=node_memory,
             model_instance=model_instance,

api/core/mcp/server/streamable_http.py

@@ -4,10 +4,10 @@ from collections.abc import Mapping
 from typing import Any, cast
 
 from configs import dify_config
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
 from core.mcp import types as mcp_types
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from models.model import App, AppMCPServer, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 

api/core/model_runtime/memory/__init__.py

@@ -1,3 +0,0 @@
-from .prompt_message_memory import DEFAULT_MEMORY_MAX_TOKEN_LIMIT, PromptMessageMemory
-
-__all__ = ["DEFAULT_MEMORY_MAX_TOKEN_LIMIT", "PromptMessageMemory"]

api/core/model_runtime/memory/prompt_message_memory.py

@@ -1,18 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Sequence
-from typing import Protocol
-
-from core.model_runtime.entities import PromptMessage
-
-DEFAULT_MEMORY_MAX_TOKEN_LIMIT = 2000
-
-
-class PromptMessageMemory(Protocol):
-    """Port for loading memory as prompt messages."""
-
-    def get_history_prompt_messages(
-        self, max_token_limit: int = DEFAULT_MEMORY_MAX_TOKEN_LIMIT, message_limit: int | None = None
-    ) -> Sequence[PromptMessage]:
-        """Return historical prompt messages constrained by token/message limits."""
-        ...

api/core/plugin/backwards_invocation/model.py

@@ -2,7 +2,6 @@ import tempfile
 from binascii import hexlify, unhexlify
 from collections.abc import Generator
 
-from core.app.llm import deduct_llm_quota
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import (
@@ -30,6 +29,7 @@ from core.plugin.entities.request import (
 )
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.utils.model_invocation_utils import ModelInvocationUtils
+from core.workflow.nodes.llm import llm_utils
 from models.account import Tenant
 
 
@@ -63,14 +63,16 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
             def handle() -> Generator[LLMResultChunk, None, None]:
                 for chunk in response:
                     if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
+                        llm_utils.deduct_llm_quota(
+                            tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage
+                        )
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
             if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
+                llm_utils.deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(response: LLMResult) -> Generator[LLMResultChunk, None, None]:
                 yield LLMResultChunk(
@@ -124,14 +126,16 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
             def handle() -> Generator[LLMResultChunkWithStructuredOutput, None, None]:
                 for chunk in response:
                     if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
+                        llm_utils.deduct_llm_quota(
+                            tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage
+                        )
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
             if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
+                llm_utils.deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(
                 response: LLMResultWithStructuredOutput,

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -8,7 +8,6 @@ from typing import Any, cast
 
 logger = logging.getLogger(__name__)
 
-from core.app.llm import deduct_llm_quota
 from core.entities.knowledge_entities import PreviewDetail
 from core.llm_generator.prompts import DEFAULT_GENERATOR_SUMMARY_PROMPT
 from core.model_manager import ModelInstance
@@ -36,6 +35,7 @@ from core.rag.models.document import AttachmentDocument, Document, MultimodalGen
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
 from core.workflow.file import File, FileTransferMethod, FileType, file_manager
+from core.workflow.nodes.llm import llm_utils
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping
 from libs import helper
@@ -474,7 +474,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
 
         # Deduct quota for summary generation (same as workflow nodes)
         try:
-            deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
+            llm_utils.deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
         except Exception as e:
             # Log but don't fail summary generation if quota deduction fails
             logger.warning("Failed to deduct quota for summary generation: %s", str(e))

api/core/rag/retrieval/router/multi_dataset_react_route.py

@@ -2,7 +2,6 @@ from collections.abc import Generator, Sequence
 from typing import Union
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.app.llm import deduct_llm_quota
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
@@ -10,6 +9,7 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
 from core.rag.retrieval.output_parser.react_output import ReactAction
 from core.rag.retrieval.output_parser.structured_chat import StructuredChatOutputParser
+from core.workflow.nodes.llm import llm_utils
 
 PREFIX = """Respond to the human as helpfully and accurately as possible. You have access to the following tools:"""
 
@@ -162,7 +162,7 @@ class ReactMultiDatasetRouter:
         text, usage = self._handle_invoke_result(invoke_result=invoke_result)
 
         # deduct quota
-        deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
+        llm_utils.deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
 
         return text, usage
 

api/core/tools/builtin_tool/providers/webscraper/webscraper.yaml

@@ -6,9 +6,9 @@ identity:
     zh_Hans: 网页抓取
     pt_BR: WebScraper
   description:
-    en_US: Web Scrapper tool kit is used to scrape web
+    en_US: Web Scraper tool kit is used to scrape web
     zh_Hans: 一个用于抓取网页的工具。
-    pt_BR: Web Scrapper tool kit is used to scrape web
+    pt_BR: Web Scraper tool kit is used to scrape web
   icon: icon.svg
   tags:
     - productivity

api/core/tools/utils/workflow_configuration_sync.py

@@ -1,11 +1,11 @@
 from collections.abc import Mapping, Sequence
 from typing import Any
 
+from core.app.app_config.entities import VariableEntity
 from core.tools.entities.tool_entities import WorkflowToolParameterConfiguration
 from core.tools.errors import WorkflowToolHumanInputNotSupportedError
 from core.workflow.enums import NodeType
 from core.workflow.nodes.base.entities import OutputVariableEntity
-from core.workflow.variables.input_entities import VariableEntity
 
 
 class WorkflowToolConfigurationUtils:

api/core/tools/workflow_as_tool/provider.py

@@ -5,6 +5,7 @@ from collections.abc import Mapping
 from pydantic import Field
 from sqlalchemy.orm import Session
 
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.db.session_factory import session_factory
 from core.plugin.entities.parameters import PluginParameterOption
@@ -22,7 +23,6 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.utils.workflow_configuration_sync import WorkflowToolConfigurationUtils
 from core.tools.workflow_as_tool.tool import WorkflowTool
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from extensions.ext_database import db
 from models.account import Account
 from models.model import App, AppMode

api/core/workflow/graph_engine/graph_engine.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import logging
 import queue
+import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, cast, final
 
@@ -76,10 +77,13 @@ class GraphEngine:
         config: GraphEngineConfig = _DEFAULT_CONFIG,
     ) -> None:
         """Initialize the graph engine with all subsystems and dependencies."""
+        # stop event
+        self._stop_event = threading.Event()
 
         # Bind runtime state to current workflow context
         self._graph = graph
         self._graph_runtime_state = graph_runtime_state
+        self._graph_runtime_state.stop_event = self._stop_event
         self._graph_runtime_state.configure(graph=cast("GraphProtocol", graph))
         self._command_channel = command_channel
         self._config = config
@@ -159,6 +163,7 @@ class GraphEngine:
             layers=self._layers,
             execution_context=execution_context,
             config=self._config,
+            stop_event=self._stop_event,
         )
 
         # === Orchestration ===
@@ -189,6 +194,7 @@ class GraphEngine:
             event_handler=self._event_handler_registry,
             execution_coordinator=self._execution_coordinator,
             event_emitter=self._event_manager,
+            stop_event=self._stop_event,
         )
 
         # === Validation ===
@@ -308,6 +314,7 @@ class GraphEngine:
 
     def _start_execution(self, *, resume: bool = False) -> None:
         """Start execution subsystems."""
+        self._stop_event.clear()
         paused_nodes: list[str] = []
         deferred_nodes: list[str] = []
         if resume:
@@ -341,6 +348,7 @@ class GraphEngine:
 
     def _stop_execution(self) -> None:
         """Stop execution subsystems."""
+        self._stop_event.set()
         self._dispatcher.stop()
         self._worker_pool.stop()
         # Don't mark complete here as the dispatcher already does it

api/core/workflow/graph_engine/orchestration/dispatcher.py

@@ -44,6 +44,7 @@ class Dispatcher:
         event_queue: queue.Queue[GraphNodeEventBase],
         event_handler: "EventHandler",
         execution_coordinator: ExecutionCoordinator,
+        stop_event: threading.Event,
         event_emitter: EventManager | None = None,
     ) -> None:
         """
@@ -61,7 +62,7 @@ class Dispatcher:
         self._event_emitter = event_emitter
 
         self._thread: threading.Thread | None = None
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._start_time: float | None = None
 
     def start(self) -> None:
@@ -69,14 +70,12 @@ class Dispatcher:
         if self._thread and self._thread.is_alive():
             return
 
-        self._stop_event.clear()
         self._start_time = time.time()
         self._thread = threading.Thread(target=self._dispatcher_loop, name="GraphDispatcher", daemon=True)
         self._thread.start()
 
     def stop(self) -> None:
         """Stop the dispatcher thread."""
-        self._stop_event.set()
         if self._thread and self._thread.is_alive():
             self._thread.join(timeout=2.0)
 

api/core/workflow/graph_engine/worker.py

@@ -42,6 +42,7 @@ class Worker(threading.Thread):
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: Sequence[GraphEngineLayer],
+        stop_event: threading.Event,
         worker_id: int = 0,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -62,13 +63,16 @@ class Worker(threading.Thread):
         self._graph = graph
         self._worker_id = worker_id
         self._execution_context = execution_context
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._layers = layers if layers is not None else []
         self._last_task_time = time.time()
 
     def stop(self) -> None:
-        """Signal the worker to stop processing."""
-        self._stop_event.set()
+        """Worker is controlled via shared stop_event from GraphEngine.
+
+        This method is a no-op retained for backward compatibility.
+        """
+        pass
 
     @property
     def is_idle(self) -> bool:

api/core/workflow/graph_engine/worker_management/worker_pool.py

@@ -37,6 +37,7 @@ class WorkerPool:
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: list[GraphEngineLayer],
+        stop_event: threading.Event,
         config: GraphEngineConfig,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -63,6 +64,7 @@ class WorkerPool:
         self._worker_counter = 0
         self._lock = threading.RLock()
         self._running = False
+        self._stop_event = stop_event
 
         # No longer tracking worker states with callbacks to avoid lock contention
 
@@ -133,6 +135,7 @@ class WorkerPool:
             layers=self._layers,
             worker_id=worker_id,
             execution_context=self._execution_context,
+            stop_event=self._stop_event,
         )
 
         worker.start()

api/core/workflow/nodes/base/node.py

@@ -302,6 +302,10 @@ class Node(Generic[NodeDataT]):
         """
         raise NotImplementedError
 
+    def _should_stop(self) -> bool:
+        """Check if execution should be stopped."""
+        return self.graph_runtime_state.stop_event.is_set()
+
     def run(self) -> Generator[GraphNodeEventBase, None, None]:
         execution_id = self.ensure_execution_id()
         self._start_at = naive_utc_now()
@@ -370,6 +374,21 @@ class Node(Generic[NodeDataT]):
                     yield event
                 else:
                     yield event
+
+                if self._should_stop():
+                    error_message = "Execution cancelled"
+                    yield NodeRunFailedEvent(
+                        id=self.execution_id,
+                        node_id=self._node_id,
+                        node_type=self.node_type,
+                        start_at=self._start_at,
+                        node_run_result=NodeRunResult(
+                            status=WorkflowNodeExecutionStatus.FAILED,
+                            error=error_message,
+                        ),
+                        error=error_message,
+                    )
+                    return
         except Exception as e:
             logger.exception("Node %s failed to run", self._node_id)
             result = NodeRunResult(

api/core/workflow/nodes/iteration/iteration_node.py

@@ -588,7 +588,6 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
     def _create_graph_engine(self, index: int, item: object):
         # Import dependencies
-        from core.app.workflow.layers.llm_quota import LLMQuotaLayer
         from core.app.workflow.node_factory import DifyNodeFactory
         from core.workflow.entities import GraphInitParams
         from core.workflow.graph import Graph
@@ -643,6 +642,5 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
             command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
             config=GraphEngineConfig(),
         )
-        graph_engine.layer(LLMQuotaLayer())
 
         return graph_engine

api/core/workflow/nodes/llm/llm_utils.py

@@ -1,21 +1,28 @@
 from collections.abc import Sequence
 from typing import cast
 
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+
+from configs import dify_config
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
+from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
-from core.model_runtime.entities import PromptMessageRole
-from core.model_runtime.entities.message_entities import (
-    ImagePromptMessageContent,
-    PromptMessage,
-    TextPromptMessageContent,
-)
+from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.entities.model_entities import AIModelEntity
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.workflow.enums import SystemVariableKey
 from core.workflow.file.models import File
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment
+from core.workflow.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment, StringSegment
+from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
+from models.model import Conversation
+from models.provider import Provider, ProviderType
+from models.provider_ids import ModelProviderID
 
 from .exc import InvalidVariableTypeError
-from .protocols import PromptMessageMemory
 
 
 def fetch_model_schema(*, model_instance: ModelInstance) -> AIModelEntity:
@@ -41,51 +48,88 @@ def fetch_files(variable_pool: VariablePool, selector: Sequence[str]) -> Sequenc
     raise InvalidVariableTypeError(f"Invalid variable type: {type(variable)}")
 
 
-def convert_history_messages_to_text(
-    *,
-    history_messages: Sequence[PromptMessage],
-    human_prefix: str,
-    ai_prefix: str,
-) -> str:
-    string_messages: list[str] = []
-    for message in history_messages:
-        if message.role == PromptMessageRole.USER:
-            role = human_prefix
-        elif message.role == PromptMessageRole.ASSISTANT:
-            role = ai_prefix
+def fetch_memory(
+    variable_pool: VariablePool, app_id: str, node_data_memory: MemoryConfig | None, model_instance: ModelInstance
+) -> TokenBufferMemory | None:
+    if not node_data_memory:
+        return None
+
+    # get conversation id
+    conversation_id_variable = variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
+    if not isinstance(conversation_id_variable, StringSegment):
+        return None
+    conversation_id = conversation_id_variable.value
+
+    with Session(db.engine, expire_on_commit=False) as session:
+        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
+        conversation = session.scalar(stmt)
+        if not conversation:
+            return None
+
+    memory = TokenBufferMemory(conversation=conversation, model_instance=model_instance)
+    return memory
+
+
+def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUsage):
+    provider_model_bundle = model_instance.provider_model_bundle
+    provider_configuration = provider_model_bundle.configuration
+
+    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
+        return
+
+    system_configuration = provider_configuration.system_configuration
+
+    quota_unit = None
+    for quota_configuration in system_configuration.quota_configurations:
+        if quota_configuration.quota_type == system_configuration.current_quota_type:
+            quota_unit = quota_configuration.quota_unit
+
+            if quota_configuration.quota_limit == -1:
+                return
+
+            break
+
+    used_quota = None
+    if quota_unit:
+        if quota_unit == QuotaUnit.TOKENS:
+            used_quota = usage.total_tokens
+        elif quota_unit == QuotaUnit.CREDITS:
+            used_quota = dify_config.get_model_credits(model_instance.model_name)
         else:
-            continue
+            used_quota = 1
 
-        if isinstance(message.content, list):
-            content_parts = []
-            for content in message.content:
-                if isinstance(content, TextPromptMessageContent):
-                    content_parts.append(content.data)
-                elif isinstance(content, ImagePromptMessageContent):
-                    content_parts.append("[image]")
+    if used_quota is not None and system_configuration.current_quota_type is not None:
+        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+            from services.credit_pool_service import CreditPoolService
 
-            inner_msg = "\n".join(content_parts)
-            string_messages.append(f"{role}: {inner_msg}")
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+            )
+        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
+            from services.credit_pool_service import CreditPoolService
+
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+                pool_type="paid",
+            )
         else:
-            string_messages.append(f"{role}: {message.content}")
-
-    return "\n".join(string_messages)
-
-
-def fetch_memory_text(
-    *,
-    memory: PromptMessageMemory,
-    max_token_limit: int,
-    message_limit: int | None = None,
-    human_prefix: str = "Human",
-    ai_prefix: str = "Assistant",
-) -> str:
-    history_messages = memory.get_history_prompt_messages(
-        max_token_limit=max_token_limit,
-        message_limit=message_limit,
-    )
-    return convert_history_messages_to_text(
-        history_messages=history_messages,
-        human_prefix=human_prefix,
-        ai_prefix=ai_prefix,
-    )
+            with Session(db.engine) as session:
+                stmt = (
+                    update(Provider)
+                    .where(
+                        Provider.tenant_id == tenant_id,
+                        # TODO: Use provider name with prefix after the data migration.
+                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                        Provider.provider_type == ProviderType.SYSTEM.value,
+                        Provider.quota_type == system_configuration.current_quota_type.value,
+                        Provider.quota_limit > Provider.quota_used,
+                    )
+                    .values(
+                        quota_used=Provider.quota_used + used_quota,
+                        last_used=naive_utc_now(),
+                    )
+                )
+                session.execute(stmt)
+                session.commit()

api/core/workflow/nodes/llm/node.py

@@ -37,7 +37,6 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from core.model_runtime.memory import PromptMessageMemory
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.prompt.entities.advanced_prompt_entities import CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
@@ -63,7 +62,7 @@ from core.workflow.node_events import (
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
-from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
+from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory, PromptMessageMemory
 from core.workflow.runtime import VariablePool
 from core.workflow.variables import (
     ArrayFileSegment,
@@ -279,6 +278,8 @@ class LLMNode(Node[LLMNodeData]):
                         else None
                     )
 
+                    # deduct quota
+                    llm_utils.deduct_llm_quota(tenant_id=self.tenant_id, model_instance=model_instance, usage=usage)
                     break
                 elif isinstance(event, LLMStructuredOutput):
                     structured_output = event
@@ -1233,10 +1234,6 @@ class LLMNode(Node[LLMNodeData]):
     def retry(self) -> bool:
         return self.node_data.retry_config.retry_enabled
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
-
 
 def _combine_message_content_with_role(
     *, contents: str | list[PromptMessageContentUnionTypes] | None = None, role: PromptMessageRole
@@ -1339,16 +1336,48 @@ def _handle_memory_completion_mode(
         )
         if not memory_config.role_prefix:
             raise MemoryRolePrefixRequiredError("Memory role prefix is required for completion model.")
-        memory_text = llm_utils.fetch_memory_text(
-            memory=memory,
+        memory_messages = memory.get_history_prompt_messages(
             max_token_limit=rest_tokens,
             message_limit=memory_config.window.size if memory_config.window.enabled else None,
+        )
+        memory_text = _convert_history_messages_to_text(
+            history_messages=memory_messages,
             human_prefix=memory_config.role_prefix.user,
             ai_prefix=memory_config.role_prefix.assistant,
         )
     return memory_text
 
 
+def _convert_history_messages_to_text(
+    *,
+    history_messages: Sequence[PromptMessage],
+    human_prefix: str,
+    ai_prefix: str,
+) -> str:
+    string_messages: list[str] = []
+    for message in history_messages:
+        if message.role == PromptMessageRole.USER:
+            role = human_prefix
+        elif message.role == PromptMessageRole.ASSISTANT:
+            role = ai_prefix
+        else:
+            continue
+
+        if isinstance(message.content, list):
+            content_parts = []
+            for content in message.content:
+                if isinstance(content, TextPromptMessageContent):
+                    content_parts.append(content.data)
+                elif isinstance(content, ImagePromptMessageContent):
+                    content_parts.append("[image]")
+
+            inner_msg = "\n".join(content_parts)
+            string_messages.append(f"{role}: {inner_msg}")
+        else:
+            string_messages.append(f"{role}: {message.content}")
+    return "\n".join(string_messages)
+
+
 def _handle_completion_template(
     *,
     template: LLMNodeCompletionModelPromptTemplate,

api/core/workflow/nodes/llm/protocols.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import Any, Protocol
 
 from core.model_manager import ModelInstance
+from core.model_runtime.entities import PromptMessage
 
 
 class CredentialsProvider(Protocol):
@@ -19,3 +21,13 @@ class ModelFactory(Protocol):
     def init_model_instance(self, provider_name: str, model_name: str) -> ModelInstance:
         """Create a model instance that is ready for schema lookup and invocation."""
         ...
+
+
+class PromptMessageMemory(Protocol):
+    """Port for loading memory as prompt messages for LLM nodes."""
+
+    def get_history_prompt_messages(
+        self, max_token_limit: int = 2000, message_limit: int | None = None
+    ) -> Sequence[PromptMessage]:
+        """Return historical prompt messages constrained by token/message limits."""
+        ...

api/core/workflow/nodes/loop/loop_node.py

@@ -413,7 +413,6 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
 
     def _create_graph_engine(self, start_at: datetime, root_node_id: str):
         # Import dependencies
-        from core.app.workflow.layers.llm_quota import LLMQuotaLayer
         from core.app.workflow.node_factory import DifyNodeFactory
         from core.workflow.entities import GraphInitParams
         from core.workflow.graph import Graph
@@ -455,6 +454,5 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
             config=GraphEngineConfig(),
         )
-        graph_engine.layer(LLMQuotaLayer())
 
         return graph_engine

api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py

@@ -5,6 +5,7 @@ import uuid
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, cast
 
+from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import ImagePromptMessageContent
 from core.model_runtime.entities.llm_entities import LLMUsage
@@ -23,17 +24,12 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
 from core.prompt.simple_prompt_transform import ModelMode
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
-from core.workflow.enums import (
-    NodeType,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.llm import llm_utils
-from core.workflow.nodes.llm.protocols import PromptMessageMemory
 from core.workflow.runtime import VariablePool
 from core.workflow.variables.types import ArrayValidation, SegmentType
 from factories.variable_factory import build_segment_with_type
@@ -101,7 +97,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
     _model_instance: ModelInstance
     _credentials_provider: "CredentialsProvider"
     _model_factory: "ModelFactory"
-    _memory: PromptMessageMemory | None
 
     def __init__(
         self,
@@ -113,7 +108,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         credentials_provider: "CredentialsProvider",
         model_factory: "ModelFactory",
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None = None,
     ) -> None:
         super().__init__(
             id=id,
@@ -124,7 +118,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         self._credentials_provider = credentials_provider
         self._model_factory = model_factory
         self._model_instance = model_instance
-        self._memory = memory
 
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
@@ -170,7 +163,13 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
         except ValueError as exc:
             raise ModelSchemaNotFoundError("Model schema not found") from exc
-        memory = self._memory
+        # fetch memory
+        memory = llm_utils.fetch_memory(
+            variable_pool=variable_pool,
+            app_id=self.app_id,
+            node_data_memory=node_data.memory,
+            model_instance=model_instance,
+        )
 
         if (
             set(model_schema.features or []) & {ModelFeature.TOOL_CALL, ModelFeature.MULTI_TOOL_CALL}
@@ -309,6 +308,9 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         usage = invoke_result.usage
         tool_call = invoke_result.message.tool_calls[0] if invoke_result.message.tool_calls else None
 
+        # deduct quota
+        llm_utils.deduct_llm_quota(tenant_id=self.tenant_id, model_instance=model_instance, usage=usage)
+
         return text, usage, tool_call
 
     def _generate_function_call_prompt(
@@ -317,7 +319,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> tuple[list[PromptMessage], list[PromptMessageTool]]:
@@ -405,7 +407,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -443,7 +445,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -468,8 +470,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             files=files,
             context="",
             memory_config=node_data.memory,
-            # AdvancedPromptTransform is still typed against TokenBufferMemory.
-            memory=cast(Any, memory),
+            memory=memory,
             model_instance=model_instance,
             image_detail_config=vision_detail,
         )
@@ -482,7 +483,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         query: str,
         variable_pool: VariablePool,
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -714,7 +715,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ) -> list[ChatModelMessage]:
         model_mode = ModelMode(node_data.model.mode)
@@ -723,8 +724,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         instruction = variable_pool.convert_template(node_data.instruction or "").text
 
         if memory and node_data.memory and node_data.memory.window:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory, max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
+            memory_str = memory.get_history_prompt_text(
+                max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
             )
         if model_mode == ModelMode.CHAT:
             system_prompt_messages = ChatModelMessage(
@@ -741,7 +742,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)
@@ -750,8 +751,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         instruction = variable_pool.convert_template(node_data.instruction or "").text
 
         if memory and node_data.memory and node_data.memory.window:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory, max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
+            memory_str = memory.get_history_prompt_text(
+                max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
             )
         if model_mode == ModelMode.CHAT:
             system_prompt_messages = ChatModelMessage(
@@ -827,10 +828,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
 
         return rest_tokens
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
-
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,

api/core/workflow/nodes/question_classifier/question_classifier_node.py

@@ -3,6 +3,7 @@ import re
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import LLMUsage, ModelPropertyKey, PromptMessageRole
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -26,7 +27,7 @@ from core.workflow.nodes.llm import (
     llm_utils,
 )
 from core.workflow.nodes.llm.file_saver import FileSaverImpl, LLMFileSaver
-from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory, PromptMessageMemory
+from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from libs.json_in_md_parser import parse_and_check_json_markdown
 
 from .entities import QuestionClassifierNodeData
@@ -55,7 +56,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
     _credentials_provider: "CredentialsProvider"
     _model_factory: "ModelFactory"
     _model_instance: ModelInstance
-    _memory: PromptMessageMemory | None
 
     def __init__(
         self,
@@ -67,7 +67,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         credentials_provider: "CredentialsProvider",
         model_factory: "ModelFactory",
         model_instance: ModelInstance,
-        memory: PromptMessageMemory | None = None,
         llm_file_saver: LLMFileSaver | None = None,
     ):
         super().__init__(
@@ -82,7 +81,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self._credentials_provider = credentials_provider
         self._model_factory = model_factory
         self._model_instance = model_instance
-        self._memory = memory
 
         if llm_file_saver is None:
             llm_file_saver = FileSaverImpl(
@@ -105,7 +103,13 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         variables = {"query": query}
         # fetch model instance
         model_instance = self._model_instance
-        memory = self._memory
+        # fetch memory
+        memory = llm_utils.fetch_memory(
+            variable_pool=variable_pool,
+            app_id=self.app_id,
+            node_data_memory=node_data.memory,
+            model_instance=model_instance,
+        )
         # fetch instruction
         node_data.instruction = node_data.instruction or ""
         node_data.instruction = variable_pool.convert_template(node_data.instruction).text
@@ -236,10 +240,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
                 llm_usage=usage,
             )
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
-
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,
@@ -323,7 +323,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self,
         node_data: QuestionClassifierNodeData,
         query: str,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)
@@ -336,8 +336,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         input_text = query
         memory_str = ""
         if memory:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory,
+            memory_str = memory.get_history_prompt_text(
                 max_token_limit=max_token_limit,
                 message_limit=node_data.memory.window.size if node_data.memory and node_data.memory.window else None,
             )

api/core/workflow/nodes/start/entities.py

@@ -2,8 +2,8 @@ from collections.abc import Sequence
 
 from pydantic import Field
 
+from core.app.app_config.entities import VariableEntity
 from core.workflow.nodes.base import BaseNodeData
-from core.workflow.variables.input_entities import VariableEntity
 
 
 class StartNodeData(BaseNodeData):

api/core/workflow/nodes/start/start_node.py

@@ -2,12 +2,12 @@ from typing import Any
 
 from jsonschema import Draft7Validator, ValidationError
 
+from core.app.app_config.entities import VariableEntityType
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.start.entities import StartNodeData
-from core.workflow.variables.input_entities import VariableEntityType
 
 
 class StartNode(Node[StartNodeData]):

api/core/workflow/runtime/graph_runtime_state.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import importlib
 import json
+import threading
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
@@ -218,6 +219,8 @@ class GraphRuntimeState:
         self._pending_graph_node_states: dict[str, NodeState] | None = None
         self._pending_graph_edge_states: dict[str, NodeState] | None = None
 
+        self.stop_event: threading.Event = threading.Event()
+
         if graph is not None:
             self.attach_graph(graph)
 

api/core/workflow/variables/__init__.py

@@ -1,4 +1,3 @@
-from .input_entities import VariableEntity, VariableEntityType
 from .segment_group import SegmentGroup
 from .segments import (
     ArrayAnySegment,
@@ -65,6 +64,4 @@ __all__ = [
     "StringVariable",
     "Variable",
     "VariableBase",
-    "VariableEntity",
-    "VariableEntityType",
 ]

api/core/workflow/variables/input_entities.py

@@ -1,62 +0,0 @@
-from collections.abc import Sequence
-from enum import StrEnum
-from typing import Any
-
-from jsonschema import Draft7Validator, SchemaError
-from pydantic import BaseModel, Field, field_validator
-
-from core.workflow.file import FileTransferMethod, FileType
-
-
-class VariableEntityType(StrEnum):
-    TEXT_INPUT = "text-input"
-    SELECT = "select"
-    PARAGRAPH = "paragraph"
-    NUMBER = "number"
-    EXTERNAL_DATA_TOOL = "external_data_tool"
-    FILE = "file"
-    FILE_LIST = "file-list"
-    CHECKBOX = "checkbox"
-    JSON_OBJECT = "json_object"
-
-
-class VariableEntity(BaseModel):
-    """
-    Shared variable entity used by workflow runtime and app configuration.
-    """
-
-    # `variable` records the name of the variable in user inputs.
-    variable: str
-    label: str
-    description: str = ""
-    type: VariableEntityType
-    required: bool = False
-    hide: bool = False
-    default: Any = None
-    max_length: int | None = None
-    options: Sequence[str] = Field(default_factory=list)
-    allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
-    allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
-    allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
-    json_schema: dict[str, Any] | None = Field(default=None)
-
-    @field_validator("description", mode="before")
-    @classmethod
-    def convert_none_description(cls, value: Any) -> str:
-        return value or ""
-
-    @field_validator("options", mode="before")
-    @classmethod
-    def convert_none_options(cls, value: Any) -> Sequence[str]:
-        return value or []
-
-    @field_validator("json_schema")
-    @classmethod
-    def validate_json_schema(cls, schema: dict[str, Any] | None) -> dict[str, Any] | None:
-        if schema is None:
-            return None
-        try:
-            Draft7Validator.check_schema(schema)
-        except SchemaError as error:
-            raise ValueError(f"Invalid JSON schema: {error.message}")
-        return schema

api/core/workflow/workflow_entry.py

@@ -6,7 +6,6 @@ from typing import Any, cast
 from configs import dify_config
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.app.workflow.layers.llm_quota import LLMQuotaLayer
 from core.app.workflow.layers.observability import ObservabilityLayer
 from core.app.workflow.node_factory import DifyNodeFactory
 from core.workflow.constants import ENVIRONMENT_VARIABLE_NODE_ID
@@ -107,7 +106,6 @@ class WorkflowEntry:
             max_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS, max_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME
         )
         self.graph_engine.layer(limits_layer)
-        self.graph_engine.layer(LLMQuotaLayer())
 
         # Add observability layer when OTel is enabled
         if dify_config.ENABLE_OTEL or is_instrument_flag_enabled():

api/services/workflow/workflow_converter.py

@@ -8,6 +8,7 @@ from core.app.app_config.entities import (
     ExternalDataVariableEntity,
     ModelConfigEntity,
     PromptTemplateEntity,
+    VariableEntity,
 )
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfigManager
 from core.app.apps.chat.app_config_manager import ChatAppConfigManager
@@ -19,7 +20,6 @@ from core.prompt.simple_prompt_transform import SimplePromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file.models import FileUploadConfig
 from core.workflow.nodes import NodeType
-from core.workflow.variables.input_entities import VariableEntity
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from models import Account

api/services/workflow_service.py

@@ -9,6 +9,7 @@ from sqlalchemy import exists, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from configs import dify_config
+from core.app.app_config.entities import VariableEntityType
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -39,7 +40,6 @@ from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import load_into_variable_pool
 from core.workflow.variables import VariableBase
-from core.workflow.variables.input_entities import VariableEntityType
 from core.workflow.variables.variables import Variable
 from core.workflow.workflow_entry import WorkflowEntry
 from enums.cloud_plan import CloudPlan

api/tasks/document_indexing_sync_task.py

@@ -1,4 +1,3 @@
-import json
 import logging
 import time
 
@@ -126,7 +125,7 @@ def document_indexing_sync_task(dataset_id: str, document_id: str):
 
         data_source_info = document.data_source_info_dict
         data_source_info["last_edited_time"] = last_edited_time
-        document.data_source_info = json.dumps(data_source_info)
+        document.data_source_info = data_source_info
 
         document.indexing_status = "parsing"
         document.processing_started_at = naive_utc_now()

api/tests/integration_tests/workflow/nodes/test_parameter_extractor.py

@@ -5,7 +5,7 @@ from unittest.mock import MagicMock
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.model_manager import ModelInstance
-from core.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
+from core.model_runtime.entities import AssistantPromptMessage
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import WorkflowNodeExecutionStatus
 from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
@@ -22,17 +22,19 @@ from tests.integration_tests.model_runtime.__mock.plugin_daemon import setup_mod
 
 def get_mocked_fetch_memory(memory_text: str):
     class MemoryMock:
-        def get_history_prompt_messages(
+        def get_history_prompt_text(
             self,
+            human_prefix: str = "Human",
+            ai_prefix: str = "Assistant",
             max_token_limit: int = 2000,
             message_limit: int | None = None,
         ):
-            return [UserPromptMessage(content=memory_text), AssistantPromptMessage(content="mocked answer")]
+            return memory_text
 
     return MagicMock(return_value=MemoryMock())
 
 
-def init_parameter_extractor_node(config: dict, memory=None):
+def init_parameter_extractor_node(config: dict):
     graph_config = {
         "edges": [
             {
@@ -77,7 +79,6 @@ def init_parameter_extractor_node(config: dict, memory=None):
         credentials_provider=MagicMock(spec=CredentialsProvider),
         model_factory=MagicMock(spec=ModelFactory),
         model_instance=MagicMock(spec=ModelInstance),
-        memory=memory,
     )
     return node
 
@@ -349,7 +350,7 @@ def test_extract_json_from_tool_call():
     assert result["location"] == "kawaii"
 
 
-def test_chat_parameter_extractor_with_memory(setup_model_mock):
+def test_chat_parameter_extractor_with_memory(setup_model_mock, monkeypatch):
     """
     Test chat parameter extractor with memory.
     """
@@ -372,7 +373,6 @@ def test_chat_parameter_extractor_with_memory(setup_model_mock):
                 "memory": {"window": {"enabled": True, "size": 50}},
             },
         },
-        memory=get_mocked_fetch_memory("customized memory")(),
     )
 
     node._model_instance = get_mocked_fetch_model_instance(
@@ -381,6 +381,8 @@ def test_chat_parameter_extractor_with_memory(setup_model_mock):
         mode="chat",
         credentials={"openai_api_key": os.environ.get("OPENAI_API_KEY")},
     )()
+    # Test the mock before running the actual test
+    monkeypatch.setattr("core.workflow.nodes.llm.llm_utils.fetch_memory", get_mocked_fetch_memory("customized memory"))
     db.session.close = MagicMock()
 
     result = node._run()

api/tests/test_containers_integration_tests/services/workflow/test_workflow_converter.py

@@ -10,10 +10,11 @@ from core.app.app_config.entities import (
     ExternalDataVariableEntity,
     ModelConfigEntity,
     PromptTemplateEntity,
+    VariableEntity,
+    VariableEntityType,
 )
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from models import Account, Tenant
 from models.api_based_extension import APIBasedExtension
 from models.model import App, AppMode, AppModelConfig

api/tests/test_containers_integration_tests/tasks/test_document_indexing_sync_task.py

@@ -12,6 +12,8 @@ from unittest.mock import Mock, patch
 from uuid import uuid4
 
 import pytest
+from psycopg2.extensions import register_adapter
+from psycopg2.extras import Json
 
 from core.indexing_runner import DocumentIsPausedError, IndexingRunner
 from models import Account, Tenant, TenantAccountJoin, TenantAccountRole
@@ -19,6 +21,12 @@ from models.dataset import Dataset, Document, DocumentSegment
 from tasks.document_indexing_sync_task import document_indexing_sync_task
 
 
+@pytest.fixture(autouse=True)
+def _register_dict_adapter_for_psycopg2():
+    """Align test DB adapter behavior with dict payloads used in task update flow."""
+    register_adapter(dict, Json)
+
+
 class DocumentIndexingSyncTaskTestDataFactory:
     """Create real DB entities for document indexing sync integration tests."""
 

api/tests/unit_tests/core/app/apps/test_base_app_generator.py

@@ -1,7 +1,7 @@
 import pytest
 
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 
 
 def test_validate_inputs_with_zero():

api/tests/unit_tests/core/mcp/server/test_streamable_http.py

@@ -4,6 +4,7 @@ from unittest.mock import Mock, patch
 import jsonschema
 import pytest
 
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
 from core.mcp import types
 from core.mcp.server.streamable_http import (
@@ -18,7 +19,6 @@ from core.mcp.server.streamable_http import (
     prepare_tool_arguments,
     process_mapping_response,
 )
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from models.model import App, AppMCPServer, AppMode, EndUser
 
 

api/tests/unit_tests/core/workflow/graph_engine/layers/test_llm_quota.py

@@ -1,174 +0,0 @@
-import threading
-from datetime import datetime
-from unittest.mock import MagicMock, patch
-
-from core.app.workflow.layers.llm_quota import LLMQuotaLayer
-from core.errors.error import QuotaExceededError
-from core.model_runtime.entities.llm_entities import LLMUsage
-from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
-from core.workflow.graph_engine.entities.commands import CommandType
-from core.workflow.graph_events.node import NodeRunSucceededEvent
-from core.workflow.node_events import NodeRunResult
-
-
-def _build_succeeded_event() -> NodeRunSucceededEvent:
-    return NodeRunSucceededEvent(
-        id="execution-id",
-        node_id="llm-node-id",
-        node_type=NodeType.LLM,
-        start_at=datetime.now(),
-        node_run_result=NodeRunResult(
-            status=WorkflowNodeExecutionStatus.SUCCEEDED,
-            inputs={"question": "hello"},
-            llm_usage=LLMUsage.empty_usage(),
-        ),
-    )
-
-
-def test_deduct_quota_called_for_successful_llm_node() -> None:
-    layer = LLMQuotaLayer()
-    node = MagicMock()
-    node.id = "llm-node-id"
-    node.execution_id = "execution-id"
-    node.node_type = NodeType.LLM
-    node.tenant_id = "tenant-id"
-    node.model_instance = object()
-
-    result_event = _build_succeeded_event()
-    with patch("core.app.workflow.layers.llm_quota.deduct_llm_quota", autospec=True) as mock_deduct:
-        layer.on_node_run_end(node=node, error=None, result_event=result_event)
-
-    mock_deduct.assert_called_once_with(
-        tenant_id="tenant-id",
-        model_instance=node.model_instance,
-        usage=result_event.node_run_result.llm_usage,
-    )
-
-
-def test_deduct_quota_called_for_question_classifier_node() -> None:
-    layer = LLMQuotaLayer()
-    node = MagicMock()
-    node.id = "question-classifier-node-id"
-    node.execution_id = "execution-id"
-    node.node_type = NodeType.QUESTION_CLASSIFIER
-    node.tenant_id = "tenant-id"
-    node.model_instance = object()
-
-    result_event = _build_succeeded_event()
-    with patch("core.app.workflow.layers.llm_quota.deduct_llm_quota", autospec=True) as mock_deduct:
-        layer.on_node_run_end(node=node, error=None, result_event=result_event)
-
-    mock_deduct.assert_called_once_with(
-        tenant_id="tenant-id",
-        model_instance=node.model_instance,
-        usage=result_event.node_run_result.llm_usage,
-    )
-
-
-def test_non_llm_node_is_ignored() -> None:
-    layer = LLMQuotaLayer()
-    node = MagicMock()
-    node.id = "start-node-id"
-    node.execution_id = "execution-id"
-    node.node_type = NodeType.START
-    node.tenant_id = "tenant-id"
-    node._model_instance = object()
-
-    result_event = _build_succeeded_event()
-    with patch("core.app.workflow.layers.llm_quota.deduct_llm_quota", autospec=True) as mock_deduct:
-        layer.on_node_run_end(node=node, error=None, result_event=result_event)
-
-    mock_deduct.assert_not_called()
-
-
-def test_quota_error_is_handled_in_layer() -> None:
-    layer = LLMQuotaLayer()
-    node = MagicMock()
-    node.id = "llm-node-id"
-    node.execution_id = "execution-id"
-    node.node_type = NodeType.LLM
-    node.tenant_id = "tenant-id"
-    node.model_instance = object()
-
-    result_event = _build_succeeded_event()
-    with patch(
-        "core.app.workflow.layers.llm_quota.deduct_llm_quota",
-        autospec=True,
-        side_effect=ValueError("quota exceeded"),
-    ):
-        layer.on_node_run_end(node=node, error=None, result_event=result_event)
-
-
-def test_quota_deduction_exceeded_aborts_workflow_immediately() -> None:
-    layer = LLMQuotaLayer()
-    stop_event = threading.Event()
-    layer.command_channel = MagicMock()
-
-    node = MagicMock()
-    node.id = "llm-node-id"
-    node.execution_id = "execution-id"
-    node.node_type = NodeType.LLM
-    node.tenant_id = "tenant-id"
-    node.model_instance = object()
-    node.graph_runtime_state = MagicMock()
-    node.graph_runtime_state.stop_event = stop_event
-
-    result_event = _build_succeeded_event()
-    with patch(
-        "core.app.workflow.layers.llm_quota.deduct_llm_quota",
-        autospec=True,
-        side_effect=QuotaExceededError("No credits remaining"),
-    ):
-        layer.on_node_run_end(node=node, error=None, result_event=result_event)
-
-    assert stop_event.is_set()
-    layer.command_channel.send_command.assert_called_once()
-    abort_command = layer.command_channel.send_command.call_args.args[0]
-    assert abort_command.command_type == CommandType.ABORT
-    assert abort_command.reason == "No credits remaining"
-
-
-def test_quota_precheck_failure_aborts_workflow_immediately() -> None:
-    layer = LLMQuotaLayer()
-    stop_event = threading.Event()
-    layer.command_channel = MagicMock()
-
-    node = MagicMock()
-    node.id = "llm-node-id"
-    node.node_type = NodeType.LLM
-    node.model_instance = object()
-    node.graph_runtime_state = MagicMock()
-    node.graph_runtime_state.stop_event = stop_event
-
-    with patch(
-        "core.app.workflow.layers.llm_quota.ensure_llm_quota_available",
-        autospec=True,
-        side_effect=QuotaExceededError("Model provider openai quota exceeded."),
-    ):
-        layer.on_node_run_start(node)
-
-    assert stop_event.is_set()
-    layer.command_channel.send_command.assert_called_once()
-    abort_command = layer.command_channel.send_command.call_args.args[0]
-    assert abort_command.command_type == CommandType.ABORT
-    assert abort_command.reason == "Model provider openai quota exceeded."
-
-
-def test_quota_precheck_passes_without_abort() -> None:
-    layer = LLMQuotaLayer()
-    stop_event = threading.Event()
-    layer.command_channel = MagicMock()
-
-    node = MagicMock()
-    node.id = "llm-node-id"
-    node.node_type = NodeType.LLM
-    node.model_instance = object()
-    node.graph_runtime_state = MagicMock()
-    node.graph_runtime_state.stop_event = stop_event
-
-    with patch("core.app.workflow.layers.llm_quota.ensure_llm_quota_available", autospec=True) as mock_check:
-        layer.on_node_run_start(node)
-
-    assert not stop_event.is_set()
-    mock_check.assert_called_once_with(model_instance=node.model_instance)
-    layer.command_channel.send_command.assert_not_called()

api/tests/unit_tests/core/workflow/graph_engine/orchestration/test_dispatcher.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import queue
+import threading
 from unittest import mock
 
 from core.workflow.entities.pause_reason import SchedulingPause
@@ -36,6 +37,7 @@ def test_dispatcher_should_consume_remains_events_after_pause():
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=execution_coordinator,
+        stop_event=threading.Event(),
     )
     dispatcher._dispatcher_loop()
     assert event_queue.empty()
@@ -96,6 +98,7 @@ def _run_dispatcher_for_event(event) -> int:
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=coordinator,
+        stop_event=threading.Event(),
     )
 
     dispatcher._dispatcher_loop()
@@ -181,6 +184,7 @@ def test_dispatcher_drain_event_queue():
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=coordinator,
+        stop_event=threading.Event(),
     )
 
     dispatcher._dispatcher_loop()

api/tests/unit_tests/core/workflow/graph_engine/test_dispatcher_pause_drain.py

@@ -1,4 +1,5 @@
 import queue
+import threading
 from datetime import datetime
 
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -64,6 +65,7 @@ def test_dispatcher_drains_events_when_paused() -> None:
         event_handler=handler,
         execution_coordinator=coordinator,
         event_emitter=None,
+        stop_event=threading.Event(),
     )
 
     dispatcher._dispatcher_loop()

api/tests/unit_tests/core/workflow/graph_engine/test_stop_event.py

@@ -0,0 +1,550 @@
+"""
+Unit tests for stop_event functionality in GraphEngine.
+
+Tests the unified stop_event management by GraphEngine and its propagation
+to WorkerPool, Worker, Dispatcher, and Nodes.
+"""
+
+import threading
+import time
+from unittest.mock import MagicMock, Mock, patch
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow.entities.graph_init_params import GraphInitParams
+from core.workflow.graph import Graph
+from core.workflow.graph_engine import GraphEngine, GraphEngineConfig
+from core.workflow.graph_engine.command_channels import InMemoryChannel
+from core.workflow.graph_events import (
+    GraphRunStartedEvent,
+    GraphRunSucceededEvent,
+    NodeRunStartedEvent,
+)
+from core.workflow.nodes.answer.answer_node import AnswerNode
+from core.workflow.nodes.start.start_node import StartNode
+from core.workflow.runtime import GraphRuntimeState, VariablePool
+from models.enums import UserFrom
+
+
+class TestStopEventPropagation:
+    """Test suite for stop_event propagation through GraphEngine components."""
+
+    def test_graph_engine_creates_stop_event(self):
+        """Test that GraphEngine creates a stop_event on initialization."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Verify stop_event was created
+        assert engine._stop_event is not None
+        assert isinstance(engine._stop_event, threading.Event)
+
+        # Verify it was set in graph_runtime_state
+        assert runtime_state.stop_event is not None
+        assert runtime_state.stop_event is engine._stop_event
+
+    def test_stop_event_cleared_on_start(self):
+        """Test that stop_event is cleared when execution starts."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Set the stop_event before running
+        engine._stop_event.set()
+        assert engine._stop_event.is_set()
+
+        # Run the engine (should clear the stop_event)
+        events = list(engine.run())
+
+        # After running, stop_event should be set again (by _stop_execution)
+        # But during start it was cleared
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        assert any(isinstance(e, GraphRunSucceededEvent) for e in events)
+
+    def test_stop_event_set_on_stop(self):
+        """Test that stop_event is set when execution stops."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Initially not set
+        assert not engine._stop_event.is_set()
+
+        # Run the engine
+        list(engine.run())
+
+        # After execution completes, stop_event should be set
+        assert engine._stop_event.is_set()
+
+    def test_stop_event_passed_to_worker_pool(self):
+        """Test that stop_event is passed to WorkerPool."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Verify WorkerPool has the stop_event
+        assert engine._worker_pool._stop_event is not None
+        assert engine._worker_pool._stop_event is engine._stop_event
+
+    def test_stop_event_passed_to_dispatcher(self):
+        """Test that stop_event is passed to Dispatcher."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Verify Dispatcher has the stop_event
+        assert engine._dispatcher._stop_event is not None
+        assert engine._dispatcher._stop_event is engine._stop_event
+
+
+class TestNodeStopCheck:
+    """Test suite for Node._should_stop() functionality."""
+
+    def test_node_should_stop_checks_runtime_state(self):
+        """Test that Node._should_stop() checks GraphRuntimeState.stop_event."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "{{#start.result#}}"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        # Initially stop_event is not set
+        assert not answer_node._should_stop()
+
+        # Set the stop_event
+        runtime_state.stop_event.set()
+
+        # Now _should_stop should return True
+        assert answer_node._should_stop()
+
+    def test_node_run_checks_stop_event_between_yields(self):
+        """Test that Node.run() checks stop_event between yielding events."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        # Create a simple node
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "hello"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        # Set stop_event BEFORE running the node
+        runtime_state.stop_event.set()
+
+        # Run the node - should yield start event then detect stop
+        # The node should check stop_event before processing
+        assert answer_node._should_stop(), "stop_event should be set"
+
+        # Run and collect events
+        events = list(answer_node.run())
+
+        # Since stop_event is set at the start, we should get:
+        # 1. NodeRunStartedEvent (always yielded first)
+        # 2. Either NodeRunFailedEvent (if detected early) or NodeRunSucceededEvent (if too fast)
+        assert len(events) >= 2
+        assert isinstance(events[0], NodeRunStartedEvent)
+
+        # Note: AnswerNode is very simple and might complete before stop check
+        # The important thing is that _should_stop() returns True when stop_event is set
+        assert answer_node._should_stop()
+
+
+class TestStopEventIntegration:
+    """Integration tests for stop_event in workflow execution."""
+
+    def test_simple_workflow_respects_stop_event(self):
+        """Test that a simple workflow respects stop_event."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"
+
+        # Create start and answer nodes
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "hello"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        mock_graph.nodes["start"] = start_node
+        mock_graph.nodes["answer"] = answer_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Set stop_event before running
+        runtime_state.stop_event.set()
+
+        # Run the engine
+        events = list(engine.run())
+
+        # Should get started event but not succeeded (due to stop)
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        # The workflow should still complete (start node runs quickly)
+        # but answer node might be cancelled depending on timing
+
+    def test_stop_event_with_concurrent_nodes(self):
+        """Test stop_event behavior with multiple concurrent nodes."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        # Create multiple nodes
+        for i in range(3):
+            answer_node = AnswerNode(
+                id=f"answer_{i}",
+                config={"id": f"answer_{i}", "data": {"title": f"answer_{i}", "answer": f"test{i}"}},
+                graph_init_params=GraphInitParams(
+                    tenant_id="test_tenant",
+                    app_id="test_app",
+                    workflow_id="test_workflow",
+                    graph_config={},
+                    user_id="test_user",
+                    user_from=UserFrom.ACCOUNT,
+                    invoke_from=InvokeFrom.DEBUGGER,
+                    call_depth=0,
+                ),
+                graph_runtime_state=runtime_state,
+            )
+            mock_graph.nodes[f"answer_{i}"] = answer_node
+
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # All nodes should share the same stop_event
+        for node in mock_graph.nodes.values():
+            assert node.graph_runtime_state.stop_event is runtime_state.stop_event
+            assert node.graph_runtime_state.stop_event is engine._stop_event
+
+
+class TestStopEventTimeoutBehavior:
+    """Test stop_event behavior with join timeouts."""
+
+    @patch("core.workflow.graph_engine.orchestration.dispatcher.threading.Thread", autospec=True)
+    def test_dispatcher_uses_shorter_timeout(self, mock_thread_cls: MagicMock):
+        """Test that Dispatcher uses 2s timeout instead of 10s."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        dispatcher = engine._dispatcher
+        dispatcher.start()  # This will create and start the mocked thread
+
+        mock_thread_instance = mock_thread_cls.return_value
+        mock_thread_instance.is_alive.return_value = True
+
+        dispatcher.stop()
+
+        mock_thread_instance.join.assert_called_once_with(timeout=2.0)
+
+    @patch("core.workflow.graph_engine.worker_management.worker_pool.Worker", autospec=True)
+    def test_worker_pool_uses_shorter_timeout(self, mock_worker_cls: MagicMock):
+        """Test that WorkerPool uses 2s timeout instead of 10s."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        worker_pool = engine._worker_pool
+        worker_pool.start(initial_count=1)  # Start with one worker
+
+        mock_worker_instance = mock_worker_cls.return_value
+        mock_worker_instance.is_alive.return_value = True
+
+        worker_pool.stop()
+
+        mock_worker_instance.join.assert_called_once_with(timeout=2.0)
+
+
+class TestStopEventResumeBehavior:
+    """Test stop_event behavior during workflow resume."""
+
+    def test_stop_event_cleared_on_resume(self):
+        """Test that stop_event is cleared when resuming a paused workflow."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Simulate a previous execution that set stop_event
+        engine._stop_event.set()
+        assert engine._stop_event.is_set()
+
+        # Run the engine (should clear stop_event in _start_execution)
+        events = list(engine.run())
+
+        # Execution should complete successfully
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        assert any(isinstance(e, GraphRunSucceededEvent) for e in events)
+
+
+class TestWorkerStopBehavior:
+    """Test Worker behavior with shared stop_event."""
+
+    def test_worker_uses_shared_stop_event(self):
+        """Test that Worker uses shared stop_event from GraphEngine."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+            config=GraphEngineConfig(),
+        )
+
+        # Get the worker pool and check workers
+        worker_pool = engine._worker_pool
+
+        # Start the worker pool to create workers
+        worker_pool.start()
+
+        # Check that at least one worker was created
+        assert len(worker_pool._workers) > 0
+
+        # Verify workers use the shared stop_event
+        for worker in worker_pool._workers:
+            assert worker._stop_event is engine._stop_event
+
+        # Clean up
+        worker_pool.stop()
+
+    def test_worker_stop_is_noop(self):
+        """Test that Worker.stop() is now a no-op."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        # Create a mock worker
+        from core.workflow.graph_engine.ready_queue import InMemoryReadyQueue
+        from core.workflow.graph_engine.worker import Worker
+
+        ready_queue = InMemoryReadyQueue()
+        event_queue = MagicMock()
+
+        # Create a proper mock graph with real dict
+        mock_graph = Mock(spec=Graph)
+        mock_graph.nodes = {}  # Use real dict
+
+        stop_event = threading.Event()
+
+        worker = Worker(
+            ready_queue=ready_queue,
+            event_queue=event_queue,
+            graph=mock_graph,
+            layers=[],
+            stop_event=stop_event,
+        )
+
+        # Calling stop() should do nothing (no-op)
+        # and should NOT set the stop_event
+        worker.stop()
+        assert not stop_event.is_set()

api/tests/unit_tests/core/workflow/nodes/test_start_node_json_object.py

@@ -4,12 +4,12 @@ import time
 import pytest
 from pydantic import ValidationError as PydanticValidationError
 
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.workflow.entities import GraphInitParams
 from core.workflow.nodes.start.entities import StartNodeData
 from core.workflow.nodes.start.start_node import StartNode
 from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 
 
 def make_start_node(user_inputs, variables):

api/tests/unit_tests/services/workflow/test_workflow_converter.py

@@ -13,11 +13,12 @@ from core.app.app_config.entities import (
     ExternalDataVariableEntity,
     ModelConfigEntity,
     PromptTemplateEntity,
+    VariableEntity,
+    VariableEntityType,
 )
 from core.helper import encrypter
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from models.api_based_extension import APIBasedExtension, APIBasedExtensionPoint
 from models.model import AppMode
 from services.workflow.workflow_converter import WorkflowConverter

api/tests/unit_tests/tasks/test_document_indexing_sync_task.py

@@ -5,7 +5,6 @@ These tests intentionally stay in unit scope because they validate call argument
 for external collaborators rather than SQL-backed state transitions.
 """
 
-import json
 import uuid
 from unittest.mock import MagicMock, Mock, patch
 
@@ -197,78 +196,3 @@ class TestDocumentIndexingSyncTaskCollaboratorParams:
             provider="notion_datasource",
             plugin_id="langgenius/notion_datasource",
         )
-
-
-class TestDataSourceInfoSerialization:
-    """Regression test: data_source_info must be written as a JSON string, not a raw dict.
-
-    See https://github.com/langgenius/dify/issues/32705
-    psycopg2 raises ``ProgrammingError: can't adapt type 'dict'`` when a Python
-    dict is passed directly to a text/LongText column.
-    """
-
-    def test_data_source_info_serialized_as_json_string(
-        self,
-        mock_document,
-        mock_dataset,
-        dataset_id,
-        document_id,
-    ):
-        """data_source_info must be serialized with json.dumps before DB write."""
-        with (
-            patch("tasks.document_indexing_sync_task.session_factory") as mock_session_factory,
-            patch("tasks.document_indexing_sync_task.DatasourceProviderService") as mock_service_class,
-            patch("tasks.document_indexing_sync_task.NotionExtractor") as mock_extractor_class,
-            patch("tasks.document_indexing_sync_task.IndexProcessorFactory") as mock_ipf,
-            patch("tasks.document_indexing_sync_task.IndexingRunner") as mock_runner_class,
-        ):
-            # External collaborators
-            mock_service = MagicMock()
-            mock_service.get_datasource_credentials.return_value = {"integration_secret": "token"}
-            mock_service_class.return_value = mock_service
-
-            mock_extractor = MagicMock()
-            # Return a *different* timestamp so the task enters the sync/update branch
-            mock_extractor.get_notion_last_edited_time.return_value = "2024-02-01T00:00:00Z"
-            mock_extractor_class.return_value = mock_extractor
-
-            mock_ip = MagicMock()
-            mock_ipf.return_value.init_index_processor.return_value = mock_ip
-
-            mock_runner = MagicMock()
-            mock_runner_class.return_value = mock_runner
-
-            # DB session mock — shared across all ``session_factory.create_session()`` calls
-            session = MagicMock()
-            session.scalars.return_value.all.return_value = []
-            # .where() path: session 1 reads document + dataset, session 2 reads dataset
-            session.query.return_value.where.return_value.first.side_effect = [
-                mock_document,
-                mock_dataset,
-                mock_dataset,
-            ]
-            # .filter_by() path: session 3 (update), session 4 (indexing)
-            session.query.return_value.filter_by.return_value.first.side_effect = [
-                mock_document,
-                mock_document,
-            ]
-
-            begin_cm = MagicMock()
-            begin_cm.__enter__.return_value = session
-            begin_cm.__exit__.return_value = False
-            session.begin.return_value = begin_cm
-
-            session_cm = MagicMock()
-            session_cm.__enter__.return_value = session
-            session_cm.__exit__.return_value = False
-            mock_session_factory.create_session.return_value = session_cm
-
-            # Act
-            document_indexing_sync_task(dataset_id, document_id)
-
-            # Assert: data_source_info must be a JSON *string*, not a dict
-            assert isinstance(mock_document.data_source_info, str), (
-                f"data_source_info should be a JSON string, got {type(mock_document.data_source_info).__name__}"
-            )
-            parsed = json.loads(mock_document.data_source_info)
-            assert parsed["last_edited_time"] == "2024-02-01T00:00:00Z"

web/vite.config.ts

@@ -1,78 +1,15 @@
 import type { Plugin } from 'vite'
-import fs from 'node:fs'
 import path from 'node:path'
 import { fileURLToPath } from 'node:url'
 import react from '@vitejs/plugin-react'
-import { codeInspectorPlugin } from 'code-inspector-plugin'
 import vinext from 'vinext'
 import { defineConfig } from 'vite'
 import tsconfigPaths from 'vite-tsconfig-paths'
 
 const __dirname = path.dirname(fileURLToPath(import.meta.url))
 const isCI = !!process.env.CI
-const inspectorPort = 5678
-const inspectorInjectTarget = path.resolve(__dirname, 'app/components/browser-initializer.tsx')
-const inspectorRuntimeFile = path.resolve(
-  __dirname,
-  `node_modules/code-inspector-plugin/dist/append-code-${inspectorPort}.js`,
-)
-
-const getInspectorRuntimeSnippet = (): string => {
-  if (!fs.existsSync(inspectorRuntimeFile))
-    return ''
-
-  const raw = fs.readFileSync(inspectorRuntimeFile, 'utf-8')
-  // Remove the helper module default export from append file to avoid duplicate default exports.
-  return raw.replace(
-    /\s*export default function CodeInspectorEmptyElement\(\)\s*\{[\s\S]*$/,
-    '',
-  )
-}
-
-const normalizeInspectorModuleId = (id: string): string => {
-  const withoutQuery = id.split('?', 1)[0]
-
-  // Vite/vinext may pass absolute fs modules as "/@fs/<abs-path>".
-  if (withoutQuery.startsWith('/@fs/'))
-    return withoutQuery.slice('/@fs'.length)
-
-  return withoutQuery
-}
-
-const createCodeInspectorPlugin = (): Plugin => {
-  return codeInspectorPlugin({
-    bundler: 'vite',
-    port: inspectorPort,
-    injectTo: inspectorInjectTarget,
-    exclude: [/^(?!.*\.(?:js|ts|mjs|mts|jsx|tsx|vue|svelte|html)(?:$|\?)).*/],
-  }) as Plugin
-}
-
-const createForceInspectorClientInjectionPlugin = (): Plugin => {
-  const clientSnippet = getInspectorRuntimeSnippet()
-
-  return {
-    name: 'vinext-force-code-inspector-client',
-    apply: 'serve',
-    enforce: 'pre',
-    transform(code, id) {
-      if (!clientSnippet)
-        return null
-
-      const cleanId = normalizeInspectorModuleId(id)
-      if (cleanId !== inspectorInjectTarget)
-        return null
-      if (code.includes('code-inspector-component'))
-        return null
-
-      return `${clientSnippet}\n${code}`
-    },
-  }
-}
 
 export default defineConfig(({ mode }) => {
-  const isDev = mode === 'development'
-
   return {
     plugins: mode === 'test'
       ? [
@@ -89,12 +26,6 @@ export default defineConfig(({ mode }) => {
           } as Plugin,
         ]
       : [
-          ...(isDev
-            ? [
-                createCodeInspectorPlugin(),
-                createForceInspectorClientInjectionPlugin(),
-              ]
-            : []),
           vinext(),
         ],
     resolve: {