Move prompt module under model runtime

Signed-off-by: majiayu000 <1835304752@qq.com>

Makefile

@@ -68,10 +68,9 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type checks (basedpyright + mypy + ty)..."
+	@echo "📝 Running type checks (basedpyright + mypy)..."
 	@./dev/basedpyright-check $(PATH_TO_CHECK)
 	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
-	@cd api && uv run ty check
 	@echo "✅ Type checks complete"
 
 test:
@@ -132,7 +131,7 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checks (basedpyright, mypy, ty)"
+	@echo "  make type-check     - Run type checks (basedpyright, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

api/.importlinter

@@ -91,7 +91,7 @@ forbidden_modules =
     core.moderation
     core.ops
     core.plugin
-    core.prompt
+    core.model_runtime.prompt
     core.provider_manager
     core.rag
     core.repositories
@@ -106,7 +106,6 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> core.provider_manager
     core.workflow.nodes.agent.agent_node -> core.tools.tool_manager
     core.workflow.nodes.document_extractor.node -> core.helper.ssrf_proxy
-    core.workflow.nodes.http_request.node -> core.tools.tool_file_manager
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
     core.workflow.nodes.llm.llm_utils -> configs
@@ -128,10 +127,10 @@ ignore_imports =
     core.workflow.nodes.human_input.human_input_node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.app.entities.app_invoke_entities
     core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.advanced_prompt_transform
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.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.question_classifier.question_classifier_node -> core.model_runtime.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
@@ -143,28 +142,22 @@ ignore_imports =
     core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
     core.workflow.nodes.tool.tool_node -> models
     core.workflow.nodes.agent.agent_node -> models.model
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.code_node_provider
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.javascript.javascript_code_provider
-    core.workflow.nodes.code.code_node -> core.helper.code_executor.python3.python3_code_provider
-    core.workflow.nodes.code.entities -> core.helper.code_executor.code_executor
-    core.workflow.nodes.http_request.executor -> core.helper.ssrf_proxy
-    core.workflow.nodes.http_request.node -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.file_saver -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.node -> core.helper.code_executor
     core.workflow.nodes.template_transform.template_renderer -> core.helper.code_executor.code_executor
     core.workflow.nodes.llm.node -> core.llm_generator.output_parser.errors
     core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
     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
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.utils.prompt_message_util
-    core.workflow.nodes.question_classifier.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.utils.prompt_message_util
+    core.workflow.nodes.agent.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.llm_utils -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.node -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.llm.node -> core.model_runtime.prompt.utils.prompt_message_util
+    core.workflow.nodes.parameter_extractor.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.prompt.utils.prompt_message_util
+    core.workflow.nodes.question_classifier.entities -> core.model_runtime.prompt.entities.advanced_prompt_entities
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_runtime.prompt.utils.prompt_message_util
     core.workflow.nodes.knowledge_index.entities -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.retrieval.retrieval_methods
     core.workflow.nodes.knowledge_index.knowledge_index_node -> models.dataset
@@ -197,7 +190,12 @@ ignore_imports =
 name = Model Runtime Internal Imports
 type = forbidden
 source_modules =
-    core.model_runtime
+    core.model_runtime.callbacks
+    core.model_runtime.entities
+    core.model_runtime.errors
+    core.model_runtime.model_providers
+    core.model_runtime.schema_validators
+    core.model_runtime.utils
 forbidden_modules =
     configs
     controllers
@@ -227,7 +225,7 @@ forbidden_modules =
     core.moderation
     core.ops
     core.plugin
-    core.prompt
+    core.model_runtime.prompt
     core.provider_manager
     core.rag
     core.repositories

api/controllers/console/wraps.py

@@ -36,9 +36,9 @@ ERROR_MSG_INVALID_ENCRYPTED_DATA = "Invalid encrypted data"
 ERROR_MSG_INVALID_ENCRYPTED_CODE = "Invalid encrypted code"
 
 
-def account_initialization_required(view: Callable[P, R]):
+def account_initialization_required(view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
         # check account initialization
         current_user, _ = current_account_with_tenant()
         if current_user.status == AccountStatus.UNINITIALIZED:
@@ -214,9 +214,9 @@ def cloud_utm_record(view: Callable[P, R]):
     return decorated
 
 
-def setup_required(view: Callable[P, R]):
+def setup_required(view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
         # check setup
         if (
             dify_config.EDITION == "SELF_HOSTED"

api/core/agent/base_agent_runner.py

@@ -32,7 +32,7 @@ from core.model_runtime.entities import (
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.entities.model_entities import ModelFeature
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (
     ToolParameter,

api/core/agent/cot_agent_runner.py

@@ -17,8 +17,8 @@ from core.model_runtime.entities.message_entities import (
     ToolPromptMessage,
     UserPromptMessage,
 )
+from core.model_runtime.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine

api/core/agent/fc_agent_runner.py

@@ -21,7 +21,7 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.model_runtime.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from core.tools.entities.tool_entities import ToolInvokeMeta
 from core.tools.tool_engine import ToolEngine
 from core.workflow.file import file_manager

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

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

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

@@ -32,8 +32,8 @@ from core.app.entities.task_entities import ChatbotAppBlockingResponse, ChatbotA
 from core.app.layers.pause_state_persist_layer import PauseStateLayerConfig, PauseStatePersistenceLayer
 from core.helper.trace_id_helper import extract_external_trace_id_from_args
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
+from core.model_runtime.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.ops.ops_trace_manager import TraceQueueManager
-from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.repositories.draft_variable_repository import (

api/core/app/apps/base_app_runner.py

@@ -33,10 +33,14 @@ from core.model_runtime.entities.message_entities import (
 )
 from core.model_runtime.entities.model_entities import ModelPropertyKey
 from core.model_runtime.errors.invoke import InvokeBadRequestError
+from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.model_runtime.prompt.entities.advanced_prompt_entities import (
+    ChatModelMessage,
+    CompletionModelPromptTemplate,
+    MemoryConfig,
+)
+from core.model_runtime.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.moderation.input_moderation import InputModeration
-from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
-from core.prompt.simple_prompt_transform import ModelMode, SimplePromptTransform
 from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.file.enums import FileTransferMethod, FileType
 from extensions.ext_database import db

api/core/app/apps/message_based_app_generator.py

@@ -27,7 +27,7 @@ from core.app.entities.task_entities import (
     CompletionAppStreamResponse,
 )
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
 from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -52,10 +52,10 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.prompt.utils.prompt_message_util import PromptMessageUtil
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
-from core.prompt.utils.prompt_message_util import PromptMessageUtil
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.tools.signature import sign_tool_file
 from core.workflow.file import helpers as file_helpers
 from core.workflow.file.enums import FileTransferMethod

api/core/app/workflow/node_factory.py

@@ -6,13 +6,15 @@ from typing_extensions import override
 from configs import dify_config
 from core.app.llm.model_access import build_dify_model_access
 from core.datasource.datasource_manager import DatasourceManager
-from core.helper.code_executor.code_executor import CodeExecutionError, CodeExecutor
-from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.helper.code_executor.code_executor import (
+    CodeExecutionError,
+    CodeExecutor,
+)
 from core.helper.ssrf_proxy import ssrf_proxy
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.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
@@ -80,7 +82,6 @@ class DifyNodeFactory(NodeFactory):
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
         self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
-        self._code_providers: tuple[type[CodeNodeProvider], ...] = CodeNode.default_code_providers()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
@@ -152,7 +153,6 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 code_executor=self._code_executor,
-                code_providers=self._code_providers,
                 code_limits=self._code_limits,
             )
 

api/core/helper/code_executor/code_executor.py

@@ -1,6 +1,5 @@
 import logging
 from collections.abc import Mapping
-from enum import StrEnum
 from threading import Lock
 from typing import Any
 
@@ -14,6 +13,7 @@ from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTr
 from core.helper.code_executor.python3.python3_transformer import Python3TemplateTransformer
 from core.helper.code_executor.template_transformer import TemplateTransformer
 from core.helper.http_client_pooling import get_pooled_http_client
+from core.workflow.nodes.code.entities import CodeLanguage
 
 logger = logging.getLogger(__name__)
 code_execution_endpoint_url = URL(str(dify_config.CODE_EXECUTION_ENDPOINT))
@@ -40,12 +40,6 @@ class CodeExecutionResponse(BaseModel):
     data: Data
 
 
-class CodeLanguage(StrEnum):
-    PYTHON3 = "python3"
-    JINJA2 = "jinja2"
-    JAVASCRIPT = "javascript"
-
-
 def _build_code_executor_client() -> httpx.Client:
     return httpx.Client(
         verify=CODE_EXECUTION_SSL_VERIFY,

api/core/llm_generator/llm_generator.py

@@ -27,10 +27,10 @@ from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
 from extensions.ext_database import db
 from extensions.ext_storage import storage

api/core/memory/token_buffer_memory.py

@@ -14,7 +14,7 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
-from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.workflow.file import file_manager
 from extensions.ext_database import db
 from factories import file_factory

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -83,19 +83,21 @@ def _merge_tool_call_delta(
         tool_call.function.arguments += delta.function.arguments
 
 
-def _build_llm_result_from_first_chunk(
+def _build_llm_result_from_chunks(
     model: str,
     prompt_messages: Sequence[PromptMessage],
     chunks: Iterator[LLMResultChunk],
 ) -> LLMResult:
     """
-    Build a single `LLMResult` from the first returned chunk.
+    Build a single `LLMResult` by accumulating all returned chunks.
 
-    This is used for `stream=False` because the plugin side may still implement the response via a chunked stream.
+    Some models only support streaming output (e.g. Qwen3 open-source edition)
+    and the plugin side may still implement the response via a chunked stream,
+    so all chunks must be consumed and concatenated into a single ``LLMResult``.
 
-    Note:
-        This function always drains the `chunks` iterator after reading the first chunk to ensure any underlying
-        streaming resources are released (e.g., HTTP connections owned by the plugin runtime).
+    The ``usage`` is taken from the last chunk that carries it, which is the
+    typical convention for streaming responses (the final chunk contains the
+    aggregated token counts).
     """
     content = ""
     content_list: list[PromptMessageContentUnionTypes] = []
@@ -104,24 +106,27 @@ def _build_llm_result_from_first_chunk(
     tools_calls: list[AssistantPromptMessage.ToolCall] = []
 
     try:
-        first_chunk = next(chunks, None)
-        if first_chunk is not None:
-            if isinstance(first_chunk.delta.message.content, str):
-                content += first_chunk.delta.message.content
-            elif isinstance(first_chunk.delta.message.content, list):
-                content_list.extend(first_chunk.delta.message.content)
+        for chunk in chunks:
+            if isinstance(chunk.delta.message.content, str):
+                content += chunk.delta.message.content
+            elif isinstance(chunk.delta.message.content, list):
+                content_list.extend(chunk.delta.message.content)
 
-            if first_chunk.delta.message.tool_calls:
-                _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
+            if chunk.delta.message.tool_calls:
+                _increase_tool_call(chunk.delta.message.tool_calls, tools_calls)
 
-            usage = first_chunk.delta.usage or LLMUsage.empty_usage()
-            system_fingerprint = first_chunk.system_fingerprint
+            if chunk.delta.usage:
+                usage = chunk.delta.usage
+            if chunk.system_fingerprint:
+                system_fingerprint = chunk.system_fingerprint
+    except Exception:
+        logger.exception("Error while consuming non-stream plugin chunk iterator.")
+        raise
     finally:
-        try:
-            for _ in chunks:
-                pass
-        except Exception:
-            logger.debug("Failed to drain non-stream plugin chunk iterator.", exc_info=True)
+        # Drain any remaining chunks to release underlying streaming resources (e.g. HTTP connections).
+        close = getattr(chunks, "close", None)
+        if callable(close):
+            close()
 
     return LLMResult(
         model=model,
@@ -174,7 +179,7 @@ def _normalize_non_stream_plugin_result(
 ) -> LLMResult:
     if isinstance(result, LLMResult):
         return result
-    return _build_llm_result_from_first_chunk(model=model, prompt_messages=prompt_messages, chunks=result)
+    return _build_llm_result_from_chunks(model=model, prompt_messages=prompt_messages, chunks=result)
 
 
 def _increase_tool_call(

api/core/model_runtime/prompt/advanced_prompt_transform.py

@@ -14,9 +14,13 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
-from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
-from core.prompt.prompt_transform import PromptTransform
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.model_runtime.prompt.entities.advanced_prompt_entities import (
+    ChatModelMessage,
+    CompletionModelPromptTemplate,
+    MemoryConfig,
+)
+from core.model_runtime.prompt.prompt_transform import PromptTransform
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file import file_manager
 from core.workflow.file.models import File
 from core.workflow.runtime import VariablePool

api/core/model_runtime/prompt/agent_history_prompt_transform.py

@@ -10,7 +10,7 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.prompt_transform import PromptTransform
+from core.model_runtime.prompt.prompt_transform import PromptTransform
 
 
 class AgentHistoryPromptTransform(PromptTransform):

api/core/model_runtime/prompt/prompt_transform.py

@@ -5,7 +5,7 @@ from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import PromptMessage
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 
 
 class PromptTransform:

api/core/model_runtime/prompt/simple_prompt_transform.py

@@ -15,9 +15,9 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
     UserPromptMessage,
 )
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
-from core.prompt.prompt_transform import PromptTransform
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.prompt_transform import PromptTransform
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file import file_manager
 from models.model import AppMode
 

api/core/model_runtime/prompt/utils/get_thread_messages_length.py

@@ -1,6 +1,6 @@
 from sqlalchemy import select
 
-from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
 from extensions.ext_database import db
 from models.model import Message
 

api/core/model_runtime/prompt/utils/prompt_message_util.py

@@ -10,7 +10,7 @@ from core.model_runtime.entities import (
     PromptMessageRole,
     TextPromptMessageContent,
 )
-from core.prompt.simple_prompt_transform import ModelMode
+from core.model_runtime.prompt.simple_prompt_transform import ModelMode
 
 
 class PromptMessageUtil:

api/core/ops/base_trace_instance.py

@@ -14,10 +14,9 @@ class BaseTraceInstance(ABC):
     Base trace instance for ops trace services
     """
 
-    @abstractmethod
     def __init__(self, trace_config: BaseTracingConfig):
         """
-        Abstract initializer for the trace instance.
+        Initializer for the trace instance.
         Distribute trace tasks by matching entities
         """
         self.trace_config = trace_config

api/core/ops/ops_trace_manager.py

@@ -41,8 +41,8 @@ logger = logging.getLogger(__name__)
 
 
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
-    def __getitem__(self, provider: str) -> dict[str, Any]:
-        match provider:
+    def __getitem__(self, key: str) -> dict[str, Any]:
+        match key:
             case TracingProviderEnum.LANGFUSE:
                 from core.ops.entities.config_entity import LangfuseConfig
                 from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
@@ -149,7 +149,7 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
                 }
 
             case _:
-                raise KeyError(f"Unsupported tracing provider: {provider}")
+                raise KeyError(f"Unsupported tracing provider: {key}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py

@@ -192,8 +192,8 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             metrics=self.config.metrics,
             include_values=True,
-            vector=None,  # ty: ignore [invalid-argument-type]
-            content=None,  # ty: ignore [invalid-argument-type]
+            vector=None,
+            content=None,
             top_k=1,
             filter=f"ref_doc_id='{id}'",
         )
@@ -211,7 +211,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,  # ty: ignore [invalid-argument-type]
+            collection_data=None,
             collection_data_filter=f"ref_doc_id IN {ids_str}",
         )
         self._client.delete_collection_data(request)
@@ -225,7 +225,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,  # ty: ignore [invalid-argument-type]
+            collection_data=None,
             collection_data_filter=f"metadata_ ->> '{key}' = '{value}'",
         )
         self._client.delete_collection_data(request)
@@ -249,7 +249,7 @@ class AnalyticdbVectorOpenAPI:
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
             vector=query_vector,
-            content=None,  # ty: ignore [invalid-argument-type]
+            content=None,
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,
         )
@@ -285,7 +285,7 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
-            vector=None,  # ty: ignore [invalid-argument-type]
+            vector=None,
             content=query,
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,

api/core/rag/datasource/vdb/couchbase/couchbase_vector.py

@@ -306,7 +306,7 @@ class CouchbaseVector(BaseVector):
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         top_k = kwargs.get("top_k", 4)
         try:
-            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))  # ty: ignore [too-many-positional-arguments]
+            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))
             search_iter = self._scope.search(
                 self._collection_name + "_search", CBrequest, SearchOptions(limit=top_k, fields=["*"])
             )

api/core/rag/index_processor/index_processor_base.py

@@ -75,15 +75,15 @@ class BaseIndexProcessor(ABC):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ):
+    ) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
         raise NotImplementedError
 
     @abstractmethod

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

@@ -115,7 +115,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ):
+    ) -> None:
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
@@ -130,7 +130,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             else:
                 keyword.add_texts(documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -196,7 +196,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
         documents: list[Any] = []
         all_multimodal_documents: list[Any] = []
         if isinstance(chunks, list):

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

@@ -126,7 +126,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ):
+    ) -> None:
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             for document in documents:
@@ -139,7 +139,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
         # node_ids is segment's node_ids
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
@@ -272,7 +272,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     child_nodes.append(child_document)
         return child_nodes
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
         parent_childs = ParentChildStructureChunk.model_validate(chunks)
         documents = []
         for parent_child in parent_childs.parent_child_chunks:

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

@@ -139,14 +139,14 @@ class QAIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ):
+    ) -> None:
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -206,7 +206,7 @@ class QAIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
         qa_chunks = QAStructureChunk.model_validate(chunks)
         documents = []
         for qa_chunk in qa_chunks.qa_chunks:

api/core/rag/retrieval/dataset_retrieval.py

@@ -29,12 +29,12 @@ from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 from core.model_runtime.entities.model_entities import ModelFeature, ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.model_runtime.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
+from core.model_runtime.prompt.simple_prompt_transform import ModelMode
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
-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.rag.data_post_processor.data_post_processor import DataPostProcessor
 from core.rag.datasource.keyword.jieba.jieba_keyword_table_handler import JiebaKeywordTableHandler
 from core.rag.datasource.retrieval_service import RetrievalService

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

@@ -5,8 +5,8 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 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
-from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
+from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.model_runtime.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

api/core/workflow/graph_engine/graph_engine.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 
 import logging
 import queue
-import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, cast, final
 
@@ -77,13 +76,10 @@ 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
@@ -163,7 +159,6 @@ class GraphEngine:
             layers=self._layers,
             execution_context=execution_context,
             config=self._config,
-            stop_event=self._stop_event,
         )
 
         # === Orchestration ===
@@ -194,7 +189,6 @@ class GraphEngine:
             event_handler=self._event_handler_registry,
             execution_coordinator=self._execution_coordinator,
             event_emitter=self._event_manager,
-            stop_event=self._stop_event,
         )
 
         # === Validation ===
@@ -314,7 +308,6 @@ 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:
@@ -348,7 +341,6 @@ 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,7 +44,6 @@ class Dispatcher:
         event_queue: queue.Queue[GraphNodeEventBase],
         event_handler: "EventHandler",
         execution_coordinator: ExecutionCoordinator,
-        stop_event: threading.Event,
         event_emitter: EventManager | None = None,
     ) -> None:
         """
@@ -62,7 +61,7 @@ class Dispatcher:
         self._event_emitter = event_emitter
 
         self._thread: threading.Thread | None = None
-        self._stop_event = stop_event
+        self._stop_event = threading.Event()
         self._start_time: float | None = None
 
     def start(self) -> None:
@@ -70,12 +69,14 @@ 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,7 +42,6 @@ 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:
@@ -63,16 +62,13 @@ class Worker(threading.Thread):
         self._graph = graph
         self._worker_id = worker_id
         self._execution_context = execution_context
-        self._stop_event = stop_event
+        self._stop_event = threading.Event()
         self._layers = layers if layers is not None else []
         self._last_task_time = time.time()
 
     def stop(self) -> None:
-        """Worker is controlled via shared stop_event from GraphEngine.
-
-        This method is a no-op retained for backward compatibility.
-        """
-        pass
+        """Signal the worker to stop processing."""
+        self._stop_event.set()
 
     @property
     def is_idle(self) -> bool:

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

@@ -37,7 +37,6 @@ class WorkerPool:
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: list[GraphEngineLayer],
-        stop_event: threading.Event,
         config: GraphEngineConfig,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -64,7 +63,6 @@ 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
 
@@ -135,7 +133,6 @@ 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/agent/entities.py

@@ -3,7 +3,7 @@ from typing import Any, Literal, Union
 
 from pydantic import BaseModel
 
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.tools.entities.tool_entities import ToolSelector
 from core.workflow.nodes.base.entities import BaseNodeData
 

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

@@ -302,10 +302,6 @@ 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()
@@ -374,21 +370,6 @@ 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/code/code_node.py

@@ -1,10 +1,8 @@
 from collections.abc import Mapping, Sequence
 from decimal import Decimal
-from typing import TYPE_CHECKING, Any, ClassVar, Protocol, cast
+from textwrap import dedent
+from typing import TYPE_CHECKING, Any, Protocol, cast
 
-from core.helper.code_executor.code_node_provider import CodeNodeProvider
-from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
-from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
@@ -36,12 +34,44 @@ class WorkflowCodeExecutor(Protocol):
     def is_execution_error(self, error: Exception) -> bool: ...
 
 
+def _build_default_config(*, language: CodeLanguage, code: str) -> Mapping[str, object]:
+    return {
+        "type": "code",
+        "config": {
+            "variables": [
+                {"variable": "arg1", "value_selector": []},
+                {"variable": "arg2", "value_selector": []},
+            ],
+            "code_language": language,
+            "code": code,
+            "outputs": {"result": {"type": "string", "children": None}},
+        },
+    }
+
+
+_DEFAULT_CODE_BY_LANGUAGE: Mapping[CodeLanguage, str] = {
+    CodeLanguage.PYTHON3: dedent(
+        """
+        def main(arg1: str, arg2: str):
+            return {
+                "result": arg1 + arg2,
+            }
+        """
+    ),
+    CodeLanguage.JAVASCRIPT: dedent(
+        """
+        function main({arg1, arg2}) {
+            return {
+                result: arg1 + arg2
+            }
+        }
+        """
+    ),
+}
+
+
 class CodeNode(Node[CodeNodeData]):
     node_type = NodeType.CODE
-    _DEFAULT_CODE_PROVIDERS: ClassVar[tuple[type[CodeNodeProvider], ...]] = (
-        Python3CodeProvider,
-        JavascriptCodeProvider,
-    )
     _limits: CodeNodeLimits
 
     def __init__(
@@ -52,7 +82,6 @@ class CodeNode(Node[CodeNodeData]):
         graph_runtime_state: "GraphRuntimeState",
         *,
         code_executor: WorkflowCodeExecutor,
-        code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits,
     ) -> None:
         super().__init__(
@@ -62,9 +91,6 @@ class CodeNode(Node[CodeNodeData]):
             graph_runtime_state=graph_runtime_state,
         )
         self._code_executor: WorkflowCodeExecutor = code_executor
-        self._code_providers: tuple[type[CodeNodeProvider], ...] = (
-            tuple(code_providers) if code_providers else self._DEFAULT_CODE_PROVIDERS
-        )
         self._limits = code_limits
 
     @classmethod
@@ -78,15 +104,10 @@ class CodeNode(Node[CodeNodeData]):
         if filters:
             code_language = cast(CodeLanguage, filters.get("code_language", CodeLanguage.PYTHON3))
 
-        code_provider: type[CodeNodeProvider] = next(
-            provider for provider in cls._DEFAULT_CODE_PROVIDERS if provider.is_accept_language(code_language)
-        )
-
-        return code_provider.get_default_config()
-
-    @classmethod
-    def default_code_providers(cls) -> tuple[type[CodeNodeProvider], ...]:
-        return cls._DEFAULT_CODE_PROVIDERS
+        default_code = _DEFAULT_CODE_BY_LANGUAGE.get(code_language)
+        if default_code is None:
+            raise CodeNodeError(f"Unsupported code language: {code_language}")
+        return _build_default_config(language=code_language, code=default_code)
 
     @classmethod
     def version(cls) -> str:
@@ -108,7 +129,6 @@ class CodeNode(Node[CodeNodeData]):
                 variables[variable_name] = variable.to_object() if variable else None
         # Run code
         try:
-            _ = self._select_code_provider(code_language)
             result = self._code_executor.execute(
                 language=code_language,
                 code=code,
@@ -130,12 +150,6 @@ class CodeNode(Node[CodeNodeData]):
 
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=variables, outputs=result)
 
-    def _select_code_provider(self, code_language: CodeLanguage) -> type[CodeNodeProvider]:
-        for provider in self._code_providers:
-            if provider.is_accept_language(code_language):
-                return provider
-        raise CodeNodeError(f"Unsupported code language: {code_language}")
-
     def _check_string(self, value: str | None, variable: str) -> str | None:
         """
         Check string

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

@@ -1,12 +1,19 @@
+from enum import StrEnum
 from typing import Annotated, Literal
 
 from pydantic import AfterValidator, BaseModel
 
-from core.helper.code_executor.code_executor import CodeLanguage
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.variables.types import SegmentType
 
+
+class CodeLanguage(StrEnum):
+    PYTHON3 = "python3"
+    JINJA2 = "jinja2"
+    JAVASCRIPT = "javascript"
+
+
 _ALLOWED_OUTPUT_FROM_CODE = frozenset(
     [
         SegmentType.STRING,

api/core/workflow/nodes/http_request/executor.py

@@ -10,9 +10,7 @@ from urllib.parse import urlencode, urlparse
 import httpx
 from json_repair import repair_json
 
-from core.helper.ssrf_proxy import ssrf_proxy
 from core.workflow.file.enums import FileTransferMethod
-from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.runtime import VariablePool
 from core.workflow.variables.segments import ArrayFileSegment, FileSegment
 
@@ -81,8 +79,8 @@ class Executor:
         http_request_config: HttpRequestNodeConfig,
         max_retries: int | None = None,
         ssl_verify: bool | None = None,
-        http_client: HttpClientProtocol | None = None,
-        file_manager: FileManagerProtocol | None = None,
+        http_client: HttpClientProtocol,
+        file_manager: FileManagerProtocol,
     ):
         self._http_request_config = http_request_config
         # If authorization API key is present, convert the API key using the variable pool
@@ -116,8 +114,8 @@ class Executor:
         self.max_retries = (
             max_retries if max_retries is not None else self._http_request_config.ssrf_default_max_retries
         )
-        self._http_client = http_client or ssrf_proxy
-        self._file_manager = file_manager or default_file_manager
+        self._http_client = http_client
+        self._file_manager = file_manager
 
         # init template
         self.variable_pool = variable_pool

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

@@ -3,17 +3,14 @@ import mimetypes
 from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
-from core.helper.ssrf_proxy import ssrf_proxy
-from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.file import File, FileTransferMethod
-from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.http_request.executor import Executor
-from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
+from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol, ToolFileManagerProtocol
 from core.workflow.variables.segments import ArrayFileSegment
 from factories import file_factory
 
@@ -45,9 +42,9 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         graph_runtime_state: "GraphRuntimeState",
         *,
         http_request_config: HttpRequestNodeConfig,
-        http_client: HttpClientProtocol | None = None,
-        tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        file_manager: FileManagerProtocol | None = None,
+        http_client: HttpClientProtocol,
+        tool_file_manager_factory: Callable[[], ToolFileManagerProtocol],
+        file_manager: FileManagerProtocol,
     ) -> None:
         super().__init__(
             id=id,
@@ -55,10 +52,11 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
+
         self._http_request_config = http_request_config
-        self._http_client = http_client or ssrf_proxy
+        self._http_client = http_client
         self._tool_file_manager_factory = tool_file_manager_factory
-        self._file_manager = file_manager or default_file_manager
+        self._file_manager = file_manager
 
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:

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

@@ -4,7 +4,11 @@ from typing import Any, Literal
 from pydantic import BaseModel, Field, field_validator
 
 from core.model_runtime.entities import ImagePromptMessageContent, LLMMode
-from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import (
+    ChatModelMessage,
+    CompletionModelPromptTemplate,
+    MemoryConfig,
+)
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.base.entities import VariableSelector
 

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

@@ -11,7 +11,7 @@ from core.model_manager import ModelInstance
 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.model_runtime.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

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

@@ -37,9 +37,9 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
+from core.model_runtime.prompt.entities.advanced_prompt_entities import CompletionModelPromptTemplate, MemoryConfig
+from core.model_runtime.prompt.utils.prompt_message_util import PromptMessageUtil
 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
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.tools.signature import sign_upload_file
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID

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

@@ -7,7 +7,7 @@ from pydantic import (
     field_validator,
 )
 
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.llm.entities import ModelConfig, VisionConfig
 from core.workflow.variables.types import SegmentType

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

@@ -19,11 +19,11 @@ from core.model_runtime.entities.message_entities import (
 )
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.model_runtime.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
+from core.model_runtime.prompt.simple_prompt_transform import ModelMode
+from core.model_runtime.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.model_runtime.utils.encoders import jsonable_encoder
-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.file import File
 from core.workflow.node_events import NodeRunResult

api/core/workflow/nodes/protocols.py

@@ -27,3 +27,16 @@ class HttpClientProtocol(Protocol):
 
 class FileManagerProtocol(Protocol):
     def download(self, f: File, /) -> bytes: ...
+
+
+class ToolFileManagerProtocol(Protocol):
+    def create_file_by_raw(
+        self,
+        *,
+        user_id: str,
+        tenant_id: str,
+        conversation_id: str | None,
+        file_binary: bytes,
+        mimetype: str,
+        filename: str | None = None,
+    ) -> Any: ...

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

@@ -1,6 +1,6 @@
 from pydantic import BaseModel, Field
 
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.llm import ModelConfig, VisionConfig
 

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

@@ -6,9 +6,9 @@ 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.prompt.simple_prompt_transform import ModelMode
+from core.model_runtime.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.prompt.simple_prompt_transform import ModelMode
-from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import (
     NodeExecutionType,

api/core/workflow/runtime/graph_runtime_state.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import importlib
 import json
-import threading
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
@@ -219,8 +218,6 @@ 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/libs/login.py

@@ -13,6 +13,8 @@ from libs.token import check_csrf_token
 from models import Account
 
 if TYPE_CHECKING:
+    from flask.typing import ResponseReturnValue
+
     from models.model import EndUser
 
 
@@ -38,7 +40,7 @@ P = ParamSpec("P")
 R = TypeVar("R")
 
 
-def login_required(func: Callable[P, R]):
+def login_required(func: Callable[P, R]) -> Callable[P, R | ResponseReturnValue]:
     """
     If you decorate a view with this, it will ensure that the current user is
     logged in and authenticated before calling the actual view. (If they are
@@ -73,7 +75,7 @@ def login_required(func: Callable[P, R]):
     """
 
     @wraps(func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+    def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R | ResponseReturnValue:
         if request.method in EXEMPT_METHODS or dify_config.LOGIN_DISABLED:
             pass
         elif current_user is not None and not current_user.is_authenticated:

api/pyproject.toml

@@ -116,7 +116,6 @@ dev = [
     "dotenv-linter~=0.5.0",
     "faker~=38.2.0",
     "lxml-stubs~=0.5.1",
-    "ty>=0.0.14",
     "basedpyright~=1.31.0",
     "ruff~=0.14.0",
     "pytest~=8.3.2",

api/services/account_service.py

@@ -289,6 +289,12 @@ class AccountService:
 
         TenantService.create_owner_tenant_if_not_exist(account=account)
 
+        # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
+        if dify_config.ENTERPRISE_ENABLED:
+            from services.enterprise.enterprise_service import try_join_default_workspace
+
+            try_join_default_workspace(str(account.id))
+
         return account
 
     @staticmethod
@@ -1407,6 +1413,12 @@ class RegisterService:
                 tenant_was_created.send(tenant)
 
             db.session.commit()
+
+            # Enterprise-only: best-effort add the account to the default workspace (does not switch current workspace).
+            if dify_config.ENTERPRISE_ENABLED:
+                from services.enterprise.enterprise_service import try_join_default_workspace
+
+                try_join_default_workspace(str(account.id))
         except WorkSpaceNotAllowedCreateError:
             db.session.rollback()
             logger.exception("Register failed")

api/services/advanced_prompt_template_service.py

@@ -1,6 +1,6 @@
 import copy
 
-from core.prompt.prompt_templates.advanced_prompt_templates import (
+from core.model_runtime.prompt.prompt_templates.advanced_prompt_templates import (
     BAICHUAN_CHAT_APP_CHAT_PROMPT_CONFIG,
     BAICHUAN_CHAT_APP_COMPLETION_PROMPT_CONFIG,
     BAICHUAN_COMPLETION_APP_CHAT_PROMPT_CONFIG,

api/services/conversation_service.py

@@ -180,6 +180,14 @@ class ConversationService:
 
     @classmethod
     def delete(cls, app_model: App, conversation_id: str, user: Union[Account, EndUser] | None):
+        """
+        Delete a conversation only if it belongs to the given user and app context.
+
+        Raises:
+            ConversationNotExistsError: When the conversation is not visible to the current user.
+        """
+        conversation = cls.get_conversation(app_model, conversation_id, user)
+
         try:
             logger.info(
                 "Initiating conversation deletion for app_name %s, conversation_id: %s",
@@ -187,10 +195,10 @@ class ConversationService:
                 conversation_id,
             )
 
-            db.session.query(Conversation).where(Conversation.id == conversation_id).delete(synchronize_session=False)
+            db.session.delete(conversation)
             db.session.commit()
 
-            delete_conversation_related_data.delay(conversation_id)
+            delete_conversation_related_data.delay(conversation.id)
 
         except Exception as e:
             db.session.rollback()

api/services/enterprise/base.py

@@ -39,6 +39,9 @@ class BaseRequest:
         endpoint: str,
         json: Any | None = None,
         params: Mapping[str, Any] | None = None,
+        *,
+        timeout: float | httpx.Timeout | None = None,
+        raise_for_status: bool = False,
     ) -> Any:
         headers = {"Content-Type": "application/json", cls.secret_key_header: cls.secret_key}
         url = f"{cls.base_url}{endpoint}"
@@ -53,7 +56,16 @@ class BaseRequest:
             logger.debug("Failed to generate traceparent header", exc_info=True)
 
         with httpx.Client(mounts=mounts) as client:
-            response = client.request(method, url, json=json, params=params, headers=headers)
+            # IMPORTANT:
+            # - In httpx, passing timeout=None disables timeouts (infinite) and overrides the library default.
+            # - To preserve httpx's default timeout behavior for existing call sites, only pass the kwarg when set.
+            request_kwargs: dict[str, Any] = {"json": json, "params": params, "headers": headers}
+            if timeout is not None:
+                request_kwargs["timeout"] = timeout
+
+            response = client.request(method, url, **request_kwargs)
+            if raise_for_status:
+                response.raise_for_status()
         return response.json()
 
 

api/services/enterprise/enterprise_service.py

@@ -1,9 +1,16 @@
+import logging
+import uuid
 from datetime import datetime
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from configs import dify_config
 from services.enterprise.base import EnterpriseRequest
 
+logger = logging.getLogger(__name__)
+
+DEFAULT_WORKSPACE_JOIN_TIMEOUT_SECONDS = 1.0
+
 
 class WebAppSettings(BaseModel):
     access_mode: str = Field(
@@ -30,6 +37,55 @@ class WorkspacePermission(BaseModel):
     )
 
 
+class DefaultWorkspaceJoinResult(BaseModel):
+    """
+    Result of ensuring an account is a member of the enterprise default workspace.
+
+    - joined=True is idempotent (already a member also returns True)
+    - joined=False means enterprise default workspace is not configured or invalid/archived
+    """
+
+    workspace_id: str = Field(default="", alias="workspaceId")
+    joined: bool
+    message: str
+
+    model_config = ConfigDict(extra="forbid", populate_by_name=True)
+
+    @model_validator(mode="after")
+    def _check_workspace_id_when_joined(self) -> "DefaultWorkspaceJoinResult":
+        if self.joined and not self.workspace_id:
+            raise ValueError("workspace_id must be non-empty when joined is True")
+        return self
+
+
+def try_join_default_workspace(account_id: str) -> None:
+    """
+    Enterprise-only side-effect: ensure account is a member of the default workspace.
+
+    This is a best-effort integration. Failures must not block user registration.
+    """
+
+    if not dify_config.ENTERPRISE_ENABLED:
+        return
+
+    try:
+        result = EnterpriseService.join_default_workspace(account_id=account_id)
+        if result.joined:
+            logger.info(
+                "Joined enterprise default workspace for account %s (workspace_id=%s)",
+                account_id,
+                result.workspace_id,
+            )
+        else:
+            logger.info(
+                "Skipped joining enterprise default workspace for account %s (message=%s)",
+                account_id,
+                result.message,
+            )
+    except Exception:
+        logger.warning("Failed to join enterprise default workspace for account %s", account_id, exc_info=True)
+
+
 class EnterpriseService:
     @classmethod
     def get_info(cls):
@@ -39,6 +95,34 @@ class EnterpriseService:
     def get_workspace_info(cls, tenant_id: str):
         return EnterpriseRequest.send_request("GET", f"/workspace/{tenant_id}/info")
 
+    @classmethod
+    def join_default_workspace(cls, *, account_id: str) -> DefaultWorkspaceJoinResult:
+        """
+        Call enterprise inner API to add an account to the default workspace.
+
+        NOTE: EnterpriseRequest.base_url is expected to already include the `/inner/api` prefix,
+        so the endpoint here is `/default-workspace/members`.
+        """
+
+        # Ensure we are sending a UUID-shaped string (enterprise side validates too).
+        try:
+            uuid.UUID(account_id)
+        except ValueError as e:
+            raise ValueError(f"account_id must be a valid UUID: {account_id}") from e
+
+        data = EnterpriseRequest.send_request(
+            "POST",
+            "/default-workspace/members",
+            json={"account_id": account_id},
+            timeout=DEFAULT_WORKSPACE_JOIN_TIMEOUT_SECONDS,
+            raise_for_status=True,
+        )
+        if not isinstance(data, dict):
+            raise ValueError("Invalid response format from enterprise default workspace API")
+        if "joined" not in data or "message" not in data:
+            raise ValueError("Invalid response payload from enterprise default workspace API")
+        return DefaultWorkspaceJoinResult.model_validate(data)
+
     @classmethod
     def get_app_sso_settings_last_update_time(cls) -> datetime:
         data = EnterpriseRequest.send_request("GET", "/sso/app/last-update-time")

api/services/workflow/workflow_converter.py

@@ -15,9 +15,9 @@ from core.app.apps.chat.app_config_manager import ChatAppConfigManager
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
 from core.helper import encrypter
 from core.model_runtime.entities.llm_entities import LLMMode
+from core.model_runtime.prompt.simple_prompt_transform import SimplePromptTransform
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.model_runtime.utils.encoders import jsonable_encoder
-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 events.app_event import app_was_created

api/tasks/document_indexing_sync_task.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import time
 
@@ -125,7 +126,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 = data_source_info
+        document.data_source_info = json.dumps(data_source_info)
 
         document.indexing_status = "parsing"
         document.processing_started_at = naive_utc_now()

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

@@ -7,8 +7,11 @@ import pytest
 from configs import dify_config
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.workflow.node_factory import DifyNodeFactory
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import WorkflowNodeExecutionStatus
+from core.workflow.file.file_manager import file_manager
 from core.workflow.graph import Graph
 from core.workflow.nodes.http_request import HttpRequestNode, HttpRequestNodeConfig
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -76,6 +79,9 @@ def init_http_node(config: dict):
         graph_init_params=init_params,
         graph_runtime_state=graph_runtime_state,
         http_request_config=HTTP_REQUEST_CONFIG,
+        http_client=ssrf_proxy,
+        tool_file_manager_factory=ToolFileManager,
+        file_manager=file_manager,
     )
 
     return node
@@ -229,6 +235,8 @@ def test_custom_auth_with_empty_api_key_raises_error(setup_http_mock):
             timeout=HttpRequestNodeTimeout(connect=10, read=30, write=10),
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=variable_pool,
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
 
@@ -716,6 +724,9 @@ def test_nested_object_variable_selector(setup_http_mock):
         graph_init_params=init_params,
         graph_runtime_state=graph_runtime_state,
         http_request_config=HTTP_REQUEST_CONFIG,
+        http_client=ssrf_proxy,
+        tool_file_manager_factory=ToolFileManager,
+        file_manager=file_manager,
     )
 
     result = node._run()

api/tests/test_containers_integration_tests/services/test_advanced_prompt_template_service.py

@@ -3,7 +3,7 @@ import copy
 import pytest
 from faker import Faker
 
-from core.prompt.prompt_templates.advanced_prompt_templates import (
+from core.model_runtime.prompt.prompt_templates.advanced_prompt_templates import (
     BAICHUAN_CHAT_APP_CHAT_PROMPT_CONFIG,
     BAICHUAN_CHAT_APP_COMPLETION_PROMPT_CONFIG,
     BAICHUAN_COMPLETION_APP_CHAT_PROMPT_CONFIG,

api/tests/test_containers_integration_tests/services/test_conversation_service.py

@@ -1034,3 +1034,34 @@ class TestConversationServiceExport:
         # Step 2: Async cleanup task triggered
         # The Celery task will handle cleanup of messages, annotations, etc.
         mock_delete_task.delay.assert_called_once_with(conversation_id)
+
+    @patch("services.conversation_service.delete_conversation_related_data")
+    def test_delete_conversation_not_owned_by_account(self, mock_delete_task, db_session_with_containers):
+        """
+        Test deletion is denied when conversation belongs to a different account.
+        """
+        # Arrange
+        app_model, owner_account = ConversationServiceIntegrationTestDataFactory.create_app_and_account(
+            db_session_with_containers
+        )
+        _, other_account = ConversationServiceIntegrationTestDataFactory.create_app_and_account(
+            db_session_with_containers
+        )
+        conversation = ConversationServiceIntegrationTestDataFactory.create_conversation(
+            db_session_with_containers,
+            app_model,
+            owner_account,
+        )
+
+        # Act & Assert
+        with pytest.raises(ConversationNotExistsError):
+            ConversationService.delete(
+                app_model=app_model,
+                conversation_id=conversation.id,
+                user=other_account,
+            )
+
+        # Verify no deletion and no async cleanup trigger
+        not_deleted = db_session_with_containers.scalar(select(Conversation).where(Conversation.id == conversation.id))
+        assert not_deleted is not None
+        mock_delete_task.delay.assert_not_called()

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

@@ -14,7 +14,7 @@ from core.app.app_config.entities import (
     VariableEntityType,
 )
 from core.model_runtime.entities.llm_entities import LLMMode
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 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,8 +12,6 @@ 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
@@ -21,12 +19,6 @@ 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/model_runtime/__base/test_large_language_model_non_stream_parsing.py

@@ -103,16 +103,16 @@ def test__normalize_non_stream_plugin_result__empty_iterator_defaults():
     assert result.system_fingerprint is None
 
 
-def test__normalize_non_stream_plugin_result__closes_chunk_iterator():
+def test__normalize_non_stream_plugin_result__accumulates_all_chunks():
+    """All chunks are accumulated from the iterator."""
     prompt_messages = [UserPromptMessage(content="hi")]
 
-    chunk = _make_chunk(content="hello", usage=LLMUsage.empty_usage())
     closed: list[bool] = []
 
     def _chunk_iter():
         try:
-            yield chunk
-            yield _make_chunk(content="ignored", usage=LLMUsage.empty_usage())
+            yield _make_chunk(content="hello", usage=LLMUsage.empty_usage())
+            yield _make_chunk(content=" world", usage=LLMUsage.empty_usage())
         finally:
             closed.append(True)
 
@@ -122,5 +122,5 @@ def test__normalize_non_stream_plugin_result__closes_chunk_iterator():
         result=_chunk_iter(),
     )
 
-    assert result.message.content == "hello"
+    assert result.message.content == "hello world"
     assert closed == [True]

api/tests/unit_tests/core/model_runtime/prompt/test_advanced_prompt_transform.py

@@ -11,9 +11,13 @@ from core.model_runtime.entities.message_entities import (
     PromptMessageRole,
     UserPromptMessage,
 )
-from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
-from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
-from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.model_runtime.prompt.advanced_prompt_transform import AdvancedPromptTransform
+from core.model_runtime.prompt.entities.advanced_prompt_entities import (
+    ChatModelMessage,
+    CompletionModelPromptTemplate,
+    MemoryConfig,
+)
+from core.model_runtime.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.file import File, FileTransferMethod, FileType
 from models.model import Conversation
 

api/tests/unit_tests/core/model_runtime/prompt/test_agent_history_prompt_transform.py

@@ -12,7 +12,7 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
+from core.model_runtime.prompt.agent_history_prompt_transform import AgentHistoryPromptTransform
 from models.model import Conversation
 
 

api/tests/unit_tests/core/model_runtime/prompt/test_extract_thread_messages.py

@@ -1,7 +1,7 @@
 from uuid import uuid4
 
 from constants import UUID_NIL
-from core.prompt.utils.extract_thread_messages import extract_thread_messages
+from core.model_runtime.prompt.utils.extract_thread_messages import extract_thread_messages
 
 
 class MockMessage:

api/tests/unit_tests/core/model_runtime/prompt/test_prompt_transform.py

@@ -6,7 +6,7 @@
 # from core.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey, ParameterRule
 # from core.model_runtime.entities.provider_entities import ProviderEntity
 # from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-# from core.prompt.prompt_transform import PromptTransform
+# from core.model_runtime.prompt.prompt_transform import PromptTransform
 
 
 # def test__calculate_rest_token():

api/tests/unit_tests/core/model_runtime/prompt/test_simple_prompt_transform.py

@@ -3,7 +3,7 @@ from unittest.mock import MagicMock
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.entities.message_entities import AssistantPromptMessage, UserPromptMessage
-from core.prompt.simple_prompt_transform import SimplePromptTransform
+from core.model_runtime.prompt.simple_prompt_transform import SimplePromptTransform
 from models.model import AppMode, Conversation
 
 

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

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import queue
-import threading
 from unittest import mock
 
 from core.workflow.entities.pause_reason import SchedulingPause
@@ -37,7 +36,6 @@ 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()
@@ -98,7 +96,6 @@ 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()
@@ -184,7 +181,6 @@ 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,5 +1,4 @@
 import queue
-import threading
 from datetime import datetime
 
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -65,7 +64,6 @@ 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_mock_factory.py

@@ -112,7 +112,6 @@ class MockNodeFactory(DifyNodeFactory):
                     graph_runtime_state=self.graph_runtime_state,
                     mock_config=self.mock_config,
                     code_executor=self._code_executor,
-                    code_providers=self._code_providers,
                     code_limits=self._code_limits,
                 )
             elif node_type == NodeType.HTTP_REQUEST:
@@ -123,6 +122,9 @@ class MockNodeFactory(DifyNodeFactory):
                     graph_runtime_state=self.graph_runtime_state,
                     mock_config=self.mock_config,
                     http_request_config=self._http_request_config,
+                    http_client=self._http_request_http_client,
+                    tool_file_manager_factory=self._http_request_tool_file_manager_factory,
+                    file_manager=self._http_request_file_manager,
                 )
             elif node_type in {NodeType.LLM, NodeType.QUESTION_CLASSIFIER, NodeType.PARAMETER_EXTRACTOR}:
                 mock_instance = mock_class(

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

@@ -1,550 +0,0 @@
-"""
-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/code/code_node_spec.py

@@ -1,7 +1,6 @@
 from configs import dify_config
-from core.helper.code_executor.code_executor import CodeLanguage
 from core.workflow.nodes.code.code_node import CodeNode
-from core.workflow.nodes.code.entities import CodeNodeData
+from core.workflow.nodes.code.entities import CodeLanguage, CodeNodeData
 from core.workflow.nodes.code.exc import (
     CodeNodeError,
     DepthLimitError,
@@ -438,7 +437,7 @@ class TestCodeNodeInitialization:
             "outputs": {"x": {"type": "number"}},
         }
 
-        node.init_node_data(data)
+        node._node_data = node._hydrate_node_data(data)
 
         assert node._node_data.title == "Test Node"
         assert node._node_data.code_language == CodeLanguage.PYTHON3
@@ -454,7 +453,7 @@ class TestCodeNodeInitialization:
             "outputs": {"x": {"type": "number"}},
         }
 
-        node.init_node_data(data)
+        node._node_data = node._hydrate_node_data(data)
 
         assert node._node_data.code_language == CodeLanguage.JAVASCRIPT
 

api/tests/unit_tests/core/workflow/nodes/code/entities_spec.py

@@ -1,8 +1,7 @@
 import pytest
 from pydantic import ValidationError
 
-from core.helper.code_executor.code_executor import CodeLanguage
-from core.workflow.nodes.code.entities import CodeNodeData
+from core.workflow.nodes.code.entities import CodeLanguage, CodeNodeData
 from core.workflow.variables.types import SegmentType
 
 

api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_executor.py

@@ -1,6 +1,8 @@
 import pytest
 
 from configs import dify_config
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.workflow.file.file_manager import file_manager
 from core.workflow.nodes.http_request import (
     BodyData,
     HttpRequestNodeAuthorization,
@@ -59,6 +61,8 @@ def test_executor_with_json_body_and_number_variable():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # Check the executor's data
@@ -113,6 +117,8 @@ def test_executor_with_json_body_and_object_variable():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # Check the executor's data
@@ -169,6 +175,8 @@ def test_executor_with_json_body_and_nested_object_variable():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # Check the executor's data
@@ -213,6 +221,8 @@ def test_extract_selectors_from_template_with_newline():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     assert executor.params == [("test", "line1\nline2")]
@@ -258,6 +268,8 @@ def test_executor_with_form_data():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # Check the executor's data
@@ -309,6 +321,8 @@ def test_init_headers():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=VariablePool(system_variables=SystemVariable.default()),
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
     executor = create_executor("aa\n cc:")
@@ -344,6 +358,8 @@ def test_init_params():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=VariablePool(system_variables=SystemVariable.default()),
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
     # Test basic key-value pairs
@@ -394,6 +410,8 @@ def test_empty_api_key_raises_error_bearer():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=variable_pool,
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
 
@@ -419,6 +437,8 @@ def test_empty_api_key_raises_error_basic():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=variable_pool,
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
 
@@ -444,6 +464,8 @@ def test_empty_api_key_raises_error_custom():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=variable_pool,
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
 
@@ -469,6 +491,8 @@ def test_whitespace_only_api_key_raises_error():
             timeout=timeout,
             http_request_config=HTTP_REQUEST_CONFIG,
             variable_pool=variable_pool,
+            http_client=ssrf_proxy,
+            file_manager=file_manager,
         )
 
 
@@ -493,6 +517,8 @@ def test_valid_api_key_works():
         timeout=timeout,
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # Should not raise an error
@@ -541,6 +567,8 @@ def test_executor_with_json_body_and_unquoted_uuid_variable():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # The UUID should be preserved in full, not truncated
@@ -586,6 +614,8 @@ def test_executor_with_json_body_and_unquoted_uuid_with_newlines():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     # The UUID should be preserved in full
@@ -625,6 +655,8 @@ def test_executor_with_json_body_preserves_numbers_and_strings():
         timeout=HttpRequestNodeTimeout(connect=10, read=30, write=30),
         http_request_config=HTTP_REQUEST_CONFIG,
         variable_pool=variable_pool,
+        http_client=ssrf_proxy,
+        file_manager=file_manager,
     )
 
     assert executor.json["count"] == 42

api/tests/unit_tests/core/workflow/nodes/http_request/test_http_request_node.py

@@ -5,8 +5,11 @@ import httpx
 import pytest
 
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.tools.tool_file_manager import ToolFileManager
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import WorkflowNodeExecutionStatus
+from core.workflow.file.file_manager import file_manager
 from core.workflow.nodes.http_request import HTTP_REQUEST_CONFIG_FILTER_KEY, HttpRequestNode, HttpRequestNodeConfig
 from core.workflow.nodes.http_request.entities import HttpRequestNodeTimeout, Response
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -116,6 +119,9 @@ def _build_http_node(
         graph_init_params=graph_init_params,
         graph_runtime_state=graph_runtime_state,
         http_request_config=HTTP_REQUEST_CONFIG,
+        http_client=ssrf_proxy,
+        tool_file_manager_factory=ToolFileManager,
+        file_manager=file_manager,
     )
 
 

api/tests/unit_tests/core/workflow/nodes/llm/test_node.py

@@ -21,7 +21,7 @@ from core.model_runtime.entities.message_entities import (
 )
 from core.model_runtime.entities.model_entities import AIModelEntity, FetchFrom, ModelType
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.model_runtime.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.workflow.entities import GraphInitParams
 from core.workflow.file import File, FileTransferMethod, FileType
 from core.workflow.nodes.llm import llm_utils

api/tests/unit_tests/services/enterprise/test_enterprise_service.py

@@ -0,0 +1,141 @@
+"""Unit tests for enterprise service integrations.
+
+This module covers the enterprise-only default workspace auto-join behavior:
+- Enterprise mode disabled: no external calls
+- Successful join / skipped join: no errors
+- Failures (network/invalid response/invalid UUID): soft-fail wrapper must not raise
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+from services.enterprise.enterprise_service import (
+    DefaultWorkspaceJoinResult,
+    EnterpriseService,
+    try_join_default_workspace,
+)
+
+
+class TestJoinDefaultWorkspace:
+    def test_join_default_workspace_success(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+        response = {"workspace_id": "22222222-2222-2222-2222-222222222222", "joined": True, "message": "ok"}
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = response
+
+            result = EnterpriseService.join_default_workspace(account_id=account_id)
+
+            assert isinstance(result, DefaultWorkspaceJoinResult)
+            assert result.workspace_id == response["workspace_id"]
+            assert result.joined is True
+            assert result.message == "ok"
+
+            mock_send_request.assert_called_once_with(
+                "POST",
+                "/default-workspace/members",
+                json={"account_id": account_id},
+                timeout=1.0,
+                raise_for_status=True,
+            )
+
+    def test_join_default_workspace_invalid_response_format_raises(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = "not-a-dict"
+
+            with pytest.raises(ValueError, match="Invalid response format"):
+                EnterpriseService.join_default_workspace(account_id=account_id)
+
+    def test_join_default_workspace_invalid_account_id_raises(self):
+        with pytest.raises(ValueError):
+            EnterpriseService.join_default_workspace(account_id="not-a-uuid")
+
+    def test_join_default_workspace_missing_required_fields_raises(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+        response = {"workspace_id": "", "message": "ok"}  # missing "joined"
+
+        with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request") as mock_send_request:
+            mock_send_request.return_value = response
+
+            with pytest.raises(ValueError, match="Invalid response payload"):
+                EnterpriseService.join_default_workspace(account_id=account_id)
+
+    def test_join_default_workspace_joined_without_workspace_id_raises(self):
+        with pytest.raises(ValueError, match="workspace_id must be non-empty when joined is True"):
+            DefaultWorkspaceJoinResult(workspace_id="", joined=True, message="ok")
+
+
+class TestTryJoinDefaultWorkspace:
+    def test_try_join_default_workspace_enterprise_disabled_noop(self):
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = False
+
+            try_join_default_workspace("11111111-1111-1111-1111-111111111111")
+
+            mock_join.assert_not_called()
+
+    def test_try_join_default_workspace_successful_join_does_not_raise(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.return_value = DefaultWorkspaceJoinResult(
+                workspace_id="22222222-2222-2222-2222-222222222222",
+                joined=True,
+                message="ok",
+            )
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_skipped_join_does_not_raise(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.return_value = DefaultWorkspaceJoinResult(
+                workspace_id="",
+                joined=False,
+                message="no default workspace configured",
+            )
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_api_failure_soft_fails(self):
+        account_id = "11111111-1111-1111-1111-111111111111"
+
+        with (
+            patch("services.enterprise.enterprise_service.dify_config") as mock_config,
+            patch("services.enterprise.enterprise_service.EnterpriseService.join_default_workspace") as mock_join,
+        ):
+            mock_config.ENTERPRISE_ENABLED = True
+            mock_join.side_effect = Exception("network failure")
+
+            # Should not raise
+            try_join_default_workspace(account_id)
+
+            mock_join.assert_called_once_with(account_id=account_id)
+
+    def test_try_join_default_workspace_invalid_account_id_soft_fails(self):
+        with patch("services.enterprise.enterprise_service.dify_config") as mock_config:
+            mock_config.ENTERPRISE_ENABLED = True
+
+            # Should not raise even though UUID parsing fails inside join_default_workspace
+            try_join_default_workspace("not-a-uuid")

api/tests/unit_tests/services/test_account_service.py

@@ -1064,6 +1064,67 @@ class TestRegisterService:
 
     # ==================== Registration Tests ====================
 
+    def test_create_account_and_tenant_calls_default_workspace_join_when_enterprise_enabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should be invoked when ENTERPRISE_ENABLED is True."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", True, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_workspace,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            result = AccountService.create_account_and_tenant(
+                email="test@example.com",
+                name="Test User",
+                interface_language="en-US",
+                password=None,
+            )
+
+            assert result == mock_account
+            mock_create_workspace.assert_called_once_with(account=mock_account)
+            mock_join_default_workspace.assert_called_once_with(str(mock_account.id))
+
+    def test_create_account_and_tenant_does_not_call_default_workspace_join_when_enterprise_disabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should not be invoked when ENTERPRISE_ENABLED is False."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", False, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_workspace,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            AccountService.create_account_and_tenant(
+                email="test@example.com",
+                name="Test User",
+                interface_language="en-US",
+                password=None,
+            )
+
+            mock_create_workspace.assert_called_once_with(account=mock_account)
+            mock_join_default_workspace.assert_not_called()
+
     def test_register_success(self, mock_db_dependencies, mock_external_service_dependencies):
         """Test successful account registration."""
         # Setup mocks
@@ -1115,6 +1176,65 @@ class TestRegisterService:
                 mock_event.send.assert_called_once_with(mock_tenant)
                 self._assert_database_operations_called(mock_db_dependencies["db"])
 
+    def test_register_calls_default_workspace_join_when_enterprise_enabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should be invoked after successful register commit."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", True, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            result = RegisterService.register(
+                email="test@example.com",
+                name="Test User",
+                password="password123",
+                language="en-US",
+                create_workspace_required=False,
+            )
+
+            assert result == mock_account
+            mock_join_default_workspace.assert_called_once_with(str(mock_account.id))
+
+    def test_register_does_not_call_default_workspace_join_when_enterprise_disabled(
+        self, mock_db_dependencies, mock_external_service_dependencies, monkeypatch
+    ):
+        """Enterprise-only side effect should not be invoked when ENTERPRISE_ENABLED is False."""
+        monkeypatch.setattr(dify_config, "ENTERPRISE_ENABLED", False, raising=False)
+
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock(
+            account_id="11111111-1111-1111-1111-111111111111"
+        )
+
+        with (
+            patch("services.account_service.AccountService.create_account") as mock_create_account,
+            patch("services.enterprise.enterprise_service.try_join_default_workspace") as mock_join_default_workspace,
+        ):
+            mock_create_account.return_value = mock_account
+
+            RegisterService.register(
+                email="test@example.com",
+                name="Test User",
+                password="password123",
+                language="en-US",
+                create_workspace_required=False,
+            )
+
+            mock_join_default_workspace.assert_not_called()
+
     def test_register_with_oauth(self, mock_db_dependencies, mock_external_service_dependencies):
         """Test account registration with OAuth integration."""
         # Setup mocks

api/tests/unit_tests/tasks/test_document_indexing_sync_task.py

@@ -5,6 +5,7 @@ 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
 
@@ -196,3 +197,78 @@ 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"

api/ty.toml

@@ -1,50 +0,0 @@
-[src]
-exclude = [
-    # deps groups (A1/A2/B/C/D/E)
-    # B: app runner + prompt
-    "core/prompt",
-    "core/app/apps/base_app_runner.py",
-    "core/app/apps/workflow_app_runner.py",
-    "core/agent",
-    "core/plugin",
-    # C: services/controllers/fields/libs
-    "services",
-    "controllers/inner_api",
-    "controllers/console/app",
-    "controllers/console/explore",
-    "controllers/console/datasets",
-    "controllers/console/workspace",
-    "controllers/service_api/wraps.py",
-    "fields/conversation_fields.py",
-    "libs/external_api.py",
-    # D: observability + integrations
-    "core/ops",
-    "extensions",
-    # E: vector DB integrations
-    "core/rag/datasource/vdb",
-    # non-producition or generated code
-    "migrations",
-    "tests",
-    # targeted ignores for current type-check errors
-    # TODO(QuantumGhost): suppress type errors in HITL related code.
-    # fix the type error later
-    "configs/middleware/cache/redis_pubsub_config.py",
-    "extensions/ext_redis.py",
-    "models/execution_extra_content.py",
-    "tasks/workflow_execution_tasks.py",
-    "core/workflow/nodes/base/node.py",
-    "services/human_input_delivery_test_service.py",
-    "core/app/apps/advanced_chat/app_generator.py",
-    "controllers/console/human_input_form.py",
-    "controllers/console/app/workflow_run.py",
-    "repositories/sqlalchemy_api_workflow_node_execution_repository.py",
-    "extensions/logstore/repositories/logstore_api_workflow_run_repository.py",
-    "controllers/web/workflow_events.py",
-    "tasks/app_generate/workflow_execute_task.py",
-]
-
-
-[rules]
-deprecated = "ignore"
-unused-ignore-comment = "ignore"
-# possibly-missing-attribute = "ignore"

api/uv.lock

@@ -1483,7 +1483,6 @@ dev = [
     { name = "scipy-stubs" },
     { name = "sseclient-py" },
     { name = "testcontainers" },
-    { name = "ty" },
     { name = "types-aiofiles" },
     { name = "types-beautifulsoup4" },
     { name = "types-cachetools" },
@@ -1684,7 +1683,6 @@ dev = [
     { name = "scipy-stubs", specifier = ">=1.15.3.0" },
     { name = "sseclient-py", specifier = ">=1.8.0" },
     { name = "testcontainers", specifier = "~=4.13.2" },
-    { name = "ty", specifier = ">=0.0.14" },
     { name = "types-aiofiles", specifier = "~=24.1.0" },
     { name = "types-beautifulsoup4", specifier = "~=4.12.0" },
     { name = "types-cachetools", specifier = "~=5.5.0" },
@@ -6278,30 +6276,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/70/26/2591b48412bde75e33bfd292034103ffe41743cacd03120e3242516cd143/transformers-4.56.2-py3-none-any.whl", hash = "sha256:79c03d0e85b26cb573c109ff9eafa96f3c8d4febfd8a0774e8bba32702dd6dde", size = 11608055, upload-time = "2025-09-19T15:16:23.736Z" },
 ]
 
-[[package]]
-name = "ty"
-version = "0.0.14"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/af/57/22c3d6bf95c2229120c49ffc2f0da8d9e8823755a1c3194da56e51f1cc31/ty-0.0.14.tar.gz", hash = "sha256:a691010565f59dd7f15cf324cdcd1d9065e010c77a04f887e1ea070ba34a7de2", size = 5036573, upload-time = "2026-01-27T00:57:31.427Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/99/cb/cc6d1d8de59beb17a41f9a614585f884ec2d95450306c173b3b7cc090d2e/ty-0.0.14-py3-none-linux_armv6l.whl", hash = "sha256:32cf2a7596e693094621d3ae568d7ee16707dce28c34d1762947874060fdddaa", size = 10034228, upload-time = "2026-01-27T00:57:53.133Z" },
-    { url = "https://files.pythonhosted.org/packages/f3/96/dd42816a2075a8f31542296ae687483a8d047f86a6538dfba573223eaf9a/ty-0.0.14-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:f971bf9805f49ce8c0968ad53e29624d80b970b9eb597b7cbaba25d8a18ce9a2", size = 9939162, upload-time = "2026-01-27T00:57:43.857Z" },
-    { url = "https://files.pythonhosted.org/packages/ff/b4/73c4859004e0f0a9eead9ecb67021438b2e8e5fdd8d03e7f5aca77623992/ty-0.0.14-py3-none-macosx_11_0_arm64.whl", hash = "sha256:45448b9e4806423523268bc15e9208c4f3f2ead7c344f615549d2e2354d6e924", size = 9418661, upload-time = "2026-01-27T00:58:03.411Z" },
-    { url = "https://files.pythonhosted.org/packages/58/35/839c4551b94613db4afa20ee555dd4f33bfa7352d5da74c5fa416ffa0fd2/ty-0.0.14-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ee94a9b747ff40114085206bdb3205a631ef19a4d3fb89e302a88754cbbae54c", size = 9837872, upload-time = "2026-01-27T00:57:23.718Z" },
-    { url = "https://files.pythonhosted.org/packages/41/2b/bbecf7e2faa20c04bebd35fc478668953ca50ee5847ce23e08acf20ea119/ty-0.0.14-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:6756715a3c33182e9ab8ffca2bb314d3c99b9c410b171736e145773ee0ae41c3", size = 9848819, upload-time = "2026-01-27T00:57:58.501Z" },
-    { url = "https://files.pythonhosted.org/packages/be/60/3c0ba0f19c0f647ad9d2b5b5ac68c0f0b4dc899001bd53b3a7537fb247a2/ty-0.0.14-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:89d0038a2f698ba8b6fec5cf216a4e44e2f95e4a5095a8c0f57fe549f87087c2", size = 10324371, upload-time = "2026-01-27T00:57:29.291Z" },
-    { url = "https://files.pythonhosted.org/packages/24/32/99d0a0b37d0397b0a989ffc2682493286aa3bc252b24004a6714368c2c3d/ty-0.0.14-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:2c64a83a2d669b77f50a4957039ca1450626fb474619f18f6f8a3eb885bf7544", size = 10865898, upload-time = "2026-01-27T00:57:33.542Z" },
-    { url = "https://files.pythonhosted.org/packages/1a/88/30b583a9e0311bb474269cfa91db53350557ebec09002bfc3fb3fc364e8c/ty-0.0.14-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:242488bfb547ef080199f6fd81369ab9cb638a778bb161511d091ffd49c12129", size = 10555777, upload-time = "2026-01-27T00:58:05.853Z" },
-    { url = "https://files.pythonhosted.org/packages/cd/a2/cb53fb6325dcf3d40f2b1d0457a25d55bfbae633c8e337bde8ec01a190eb/ty-0.0.14-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4790c3866f6c83a4f424fc7d09ebdb225c1f1131647ba8bdc6fcdc28f09ed0ff", size = 10412913, upload-time = "2026-01-27T00:57:38.834Z" },
-    { url = "https://files.pythonhosted.org/packages/42/8f/f2f5202d725ed1e6a4e5ffaa32b190a1fe70c0b1a2503d38515da4130b4c/ty-0.0.14-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:950f320437f96d4ea9a2332bbfb5b68f1c1acd269ebfa4c09b6970cc1565bd9d", size = 9837608, upload-time = "2026-01-27T00:57:55.898Z" },
-    { url = "https://files.pythonhosted.org/packages/f7/ba/59a2a0521640c489dafa2c546ae1f8465f92956fede18660653cce73b4c5/ty-0.0.14-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:4a0ec3ee70d83887f86925bbc1c56f4628bd58a0f47f6f32ddfe04e1f05466df", size = 9884324, upload-time = "2026-01-27T00:57:46.786Z" },
-    { url = "https://files.pythonhosted.org/packages/03/95/8d2a49880f47b638743212f011088552ecc454dd7a665ddcbdabea25772a/ty-0.0.14-py3-none-musllinux_1_2_i686.whl", hash = "sha256:a1a4e6b6da0c58b34415955279eff754d6206b35af56a18bb70eb519d8d139ef", size = 10033537, upload-time = "2026-01-27T00:58:01.149Z" },
-    { url = "https://files.pythonhosted.org/packages/e9/40/4523b36f2ce69f92ccf783855a9e0ebbbd0f0bb5cdce6211ee1737159ed3/ty-0.0.14-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:dc04384e874c5de4c5d743369c277c8aa73d1edea3c7fc646b2064b637db4db3", size = 10495910, upload-time = "2026-01-27T00:57:26.691Z" },
-    { url = "https://files.pythonhosted.org/packages/08/d5/655beb51224d1bfd4f9ddc0bb209659bfe71ff141bcf05c418ab670698f0/ty-0.0.14-py3-none-win32.whl", hash = "sha256:b20e22cf54c66b3e37e87377635da412d9a552c9bf4ad9fc449fed8b2e19dad2", size = 9507626, upload-time = "2026-01-27T00:57:41.43Z" },
-    { url = "https://files.pythonhosted.org/packages/b6/d9/c569c9961760e20e0a4bc008eeb1415754564304fd53997a371b7cf3f864/ty-0.0.14-py3-none-win_amd64.whl", hash = "sha256:e312ff9475522d1a33186657fe74d1ec98e4a13e016d66f5758a452c90ff6409", size = 10437980, upload-time = "2026-01-27T00:57:36.422Z" },
-    { url = "https://files.pythonhosted.org/packages/ad/0c/186829654f5bfd9a028f6648e9caeb11271960a61de97484627d24443f91/ty-0.0.14-py3-none-win_arm64.whl", hash = "sha256:b6facdbe9b740cb2c15293a1d178e22ffc600653646452632541d01c36d5e378", size = 9885831, upload-time = "2026-01-27T00:57:49.747Z" },
-]
-
 [[package]]
 name = "typer"
 version = "0.20.0"