feat(oauth): improve error handling for provider retrieval and clarify logging messages

# Conflicts:
#	api/controllers/console/workspace/tool_providers.py
#	api/core/tools/entities/api_entities.py
#	api/core/tools/tool_manager.py
#	api/core/tools/utils/configuration.py
#	api/services/tools/tools_transform_service.py

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -8,15 +8,13 @@ body:
       label: Self Checks
       description: "To make sure we get to you in time, please check the following :)"
       options:
-        - label: I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542).
-          required: true
         - label: This is only for bug report, if you would like to ask a question, please head to [Discussions](https://github.com/langgenius/dify/discussions/categories/general).
           required: true
         - label: I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
           required: true
-        - label: I confirm that I am using English to submit this report, otherwise it will be closed.
+        - label: I confirm that I am using English to submit this report (我已阅读并同意 [Language Policy](https://github.com/langgenius/dify/issues/1542)).
           required: true
-        - label: 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
+        - label: "[FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)"
           required: true
         - label: "Please do not modify this template :) and fill in all the required fields."
           required: true
@@ -44,22 +42,20 @@ body:
     attributes:
       label: Steps to reproduce
       description: We highly suggest including screenshots and a bug report log. Please use the right markdown syntax for code blocks.
-      placeholder: Having detailed steps helps us reproduce the bug. If you have logs, please use fenced code blocks (triple backticks ```) to format them.
+      placeholder: Having detailed steps helps us reproduce the bug.
     validations:
       required: true
 
   - type: textarea
     attributes:
       label: ✔️ Expected Behavior
-      description: Describe what you expected to happen.
-      placeholder: What were you expecting? Please do not copy and paste the steps to reproduce here.
+      placeholder: What were you expecting?
     validations:
-      required: true
+      required: false
 
   - type: textarea
     attributes:
       label: ❌ Actual Behavior
-      description: Describe what actually happened.
-      placeholder: What happened instead? Please do not copy and paste the steps to reproduce here.
+      placeholder: What happened instead?
     validations:
       required: false

.github/ISSUE_TEMPLATE/config.yml

@@ -1,11 +1,5 @@
 blank_issues_enabled: false
 contact_links:
-  - name: "\U0001F4A1 Model Providers & Plugins"
-    url: "https://github.com/langgenius/dify-official-plugins/issues/new/choose"
-    about: Report issues with official plugins or model providers, you will need to provide the plugin version and other relevant details.
-  - name: "\U0001F4AC Documentation Issues"
-    url: "https://github.com/langgenius/dify-docs/issues/new"
-    about: Report issues with the documentation, such as typos, outdated information, or missing content. Please provide the specific section and details of the issue.
   - name: "\U0001F4E7 Discussions"
     url: https://github.com/langgenius/dify/discussions/categories/general
-    about: General discussions and seek help from the community
+    about: General discussions and request help from the community

.github/ISSUE_TEMPLATE/document_issue.yml

@@ -0,0 +1,24 @@
+name: "📚 Documentation Issue"
+description: Report issues in our documentation
+labels:
+  - documentation
+body:
+  - type: checkboxes
+    attributes:
+      label: Self Checks
+      description: "To make sure we get to you in time, please check the following :)"
+      options:
+        - label: I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
+          required: true
+        - label: I confirm that I am using English to submit report (我已阅读并同意 [Language Policy](https://github.com/langgenius/dify/issues/1542)).
+          required: true
+        - label: "[FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)"
+          required: true
+        - label: "Please do not modify this template :) and fill in all the required fields."
+          required: true
+  - type: textarea
+    attributes:
+      label: Provide a description of requested docs changes
+      placeholder: Briefly describe which document needs to be corrected and why.
+    validations:
+      required: true

.github/ISSUE_TEMPLATE/feature_request.yml

@@ -8,11 +8,11 @@ body:
       label: Self Checks
       description: "To make sure we get to you in time, please check the following :)"
       options:
-        - label: I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542).
-          required: true
         - label: I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
           required: true
-        - label: I confirm that I am using English to submit this report, otherwise it will be closed.
+        - label: I confirm that I am using English to submit this report (我已阅读并同意 [Language Policy](https://github.com/langgenius/dify/issues/1542)).
+          required: true
+        - label: "[FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)"
           required: true
         - label: "Please do not modify this template :) and fill in all the required fields."
           required: true

.github/ISSUE_TEMPLATE/translation_issue.yml

@@ -0,0 +1,55 @@
+name: "🌐 Localization/Translation issue"
+description: Report incorrect translations. [please use English :)]
+labels:
+  - translation
+body:
+  - type: checkboxes
+    attributes:
+      label: Self Checks
+      description: "To make sure we get to you in time, please check the following :)"
+      options:
+        - label: I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
+          required: true
+        - label: I confirm that I am using English to submit this report (我已阅读并同意 [Language Policy](https://github.com/langgenius/dify/issues/1542)).
+          required: true
+        - label: "[FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)"
+          required: true
+        - label: "Please do not modify this template :) and fill in all the required fields."
+          required: true
+  - type: input
+    attributes:
+      label: Dify version
+      description: Hover over system tray icon or look at Settings
+    validations:
+      required: true
+  - type: input
+    attributes:
+      label: Utility with translation issue
+      placeholder: Some area
+      description: Please input here the utility with the translation issue
+    validations:
+      required: true
+  - type: input
+    attributes:
+      label: 🌐 Language affected
+      placeholder: "German"
+    validations:
+      required: true
+  - type: textarea
+    attributes:
+      label: ❌ Actual phrase(s)
+      placeholder: What is there? Please include a screenshot as that is extremely helpful.
+    validations:
+      required: true
+  - type: textarea
+    attributes:
+      label: ✔️ Expected phrase(s)
+      placeholder: What was expected?
+    validations:
+      required: true
+  - type: textarea
+    attributes:
+      label: ℹ Why is the current translation wrong
+      placeholder: Why do you feel this is incorrect?
+    validations:
+      required: true

.github/workflows/build-push.yml

@@ -6,7 +6,6 @@ on:
       - "main"
       - "deploy/dev"
       - "deploy/enterprise"
-      - "build/**"
     tags:
       - "*"
 

api/.env.example

@@ -495,8 +495,6 @@ ENDPOINT_URL_TEMPLATE=http://localhost:5002/e/{hook_id}
 
 # Reset password token expiry minutes
 RESET_PASSWORD_TOKEN_EXPIRY_MINUTES=5
-CHANGE_EMAIL_TOKEN_EXPIRY_MINUTES=5
-OWNER_TRANSFER_TOKEN_EXPIRY_MINUTES=5
 
 CREATE_TIDB_SERVICE_JOB_ENABLED=false
 
@@ -507,8 +505,6 @@ LOGIN_LOCKOUT_DURATION=86400
 
 # Enable OpenTelemetry
 ENABLE_OTEL=false
-OTLP_TRACE_ENDPOINT=
-OTLP_METRIC_ENDPOINT=
 OTLP_BASE_ENDPOINT=http://localhost:4318
 OTLP_API_KEY=
 OTEL_EXPORTER_OTLP_PROTOCOL=

api/commands.py

@@ -2,11 +2,10 @@ import base64
 import json
 import logging
 import secrets
-from typing import Any, Optional
+from typing import Optional
 
 import click
 from flask import current_app
-from pydantic import TypeAdapter
 from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
@@ -1175,12 +1174,12 @@ def setup_system_tool_oauth_client(provider, client_params):
     try:
         # json validate
         click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
-        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
+        json.loads(client_params)
         click.echo(click.style("Client params validated successfully.", fg="green"))
 
         click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
         click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
-        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
+        oauth_client_params = encrypt_system_oauth_params(client_params)
         click.echo(click.style("Client params encrypted successfully.", fg="green"))
     except Exception as e:
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))

api/configs/feature/__init__.py

@@ -31,15 +31,6 @@ class SecurityConfig(BaseSettings):
         description="Duration in minutes for which a password reset token remains valid",
         default=5,
     )
-    CHANGE_EMAIL_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
-        description="Duration in minutes for which a change email token remains valid",
-        default=5,
-    )
-
-    OWNER_TRANSFER_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
-        description="Duration in minutes for which a owner transfer token remains valid",
-        default=5,
-    )
 
     LOGIN_DISABLED: bool = Field(
         description="Whether to disable login checks",
@@ -623,16 +614,6 @@ class AuthConfig(BaseSettings):
         default=86400,
     )
 
-    CHANGE_EMAIL_LOCKOUT_DURATION: PositiveInt = Field(
-        description="Time (in seconds) a user must wait before retrying change email after exceeding the rate limit.",
-        default=86400,
-    )
-
-    OWNER_TRANSFER_LOCKOUT_DURATION: PositiveInt = Field(
-        description="Time (in seconds) a user must wait before retrying owner transfer after exceeding the rate limit.",
-        default=86400,
-    )
-
 
 class ModerationConfig(BaseSettings):
     """

api/configs/middleware/__init__.py

@@ -162,11 +162,6 @@ class DatabaseConfig(BaseSettings):
         default=3600,
     )
 
-    SQLALCHEMY_POOL_USE_LIFO: bool = Field(
-        description="If True, SQLAlchemy will use last-in-first-out way to retrieve connections from pool.",
-        default=False,
-    )
-
     SQLALCHEMY_POOL_PRE_PING: bool = Field(
         description="If True, enables connection pool pre-ping feature to check connections.",
         default=False,
@@ -204,7 +199,6 @@ class DatabaseConfig(BaseSettings):
             "pool_recycle": self.SQLALCHEMY_POOL_RECYCLE,
             "pool_pre_ping": self.SQLALCHEMY_POOL_PRE_PING,
             "connect_args": connect_args,
-            "pool_use_lifo": self.SQLALCHEMY_POOL_USE_LIFO,
         }
 
 

api/configs/observability/otel/otel_config.py

@@ -12,16 +12,6 @@ class OTelConfig(BaseSettings):
         default=False,
     )
 
-    OTLP_TRACE_ENDPOINT: str = Field(
-        description="OTLP trace endpoint",
-        default="",
-    )
-
-    OTLP_METRIC_ENDPOINT: str = Field(
-        description="OTLP metric endpoint",
-        default="",
-    )
-
     OTLP_BASE_ENDPOINT: str = Field(
         description="OTLP base endpoint",
         default="http://localhost:4318",

api/controllers/console/app/app.py

@@ -151,7 +151,6 @@ class AppApi(Resource):
         parser.add_argument("icon", type=str, location="json")
         parser.add_argument("icon_background", type=str, location="json")
         parser.add_argument("use_icon_as_answer_icon", type=bool, location="json")
-        parser.add_argument("max_active_requests", type=int, location="json")
         args = parser.parse_args()
 
         app_service = AppService()

api/controllers/console/app/mcp_server.py

@@ -35,20 +35,16 @@ class AppMCPServerController(Resource):
     @get_app_model
     @marshal_with(app_server_fields)
     def post(self, app_model):
+        # The role of the current user in the ta table must be editor, admin, or owner
         if not current_user.is_editor:
             raise NotFound()
         parser = reqparse.RequestParser()
-        parser.add_argument("description", type=str, required=False, location="json")
+        parser.add_argument("description", type=str, required=True, location="json")
         parser.add_argument("parameters", type=dict, required=True, location="json")
         args = parser.parse_args()
-
-        description = args.get("description")
-        if not description:
-            description = app_model.description or ""
-
         server = AppMCPServer(
             name=app_model.name,
-            description=description,
+            description=args["description"],
             parameters=json.dumps(args["parameters"], ensure_ascii=False),
             status=AppMCPServerStatus.ACTIVE,
             app_id=app_model.id,
@@ -69,22 +65,14 @@ class AppMCPServerController(Resource):
             raise NotFound()
         parser = reqparse.RequestParser()
         parser.add_argument("id", type=str, required=True, location="json")
-        parser.add_argument("description", type=str, required=False, location="json")
+        parser.add_argument("description", type=str, required=True, location="json")
         parser.add_argument("parameters", type=dict, required=True, location="json")
         parser.add_argument("status", type=str, required=False, location="json")
         args = parser.parse_args()
         server = db.session.query(AppMCPServer).filter(AppMCPServer.id == args["id"]).first()
         if not server:
             raise NotFound()
-
-        description = args.get("description")
-        if description is None:
-            pass
-        elif not description:
-            server.description = app_model.description or ""
-        else:
-            server.description = description
-
+        server.description = args["description"]
         server.parameters = json.dumps(args["parameters"], ensure_ascii=False)
         if args["status"]:
             if args["status"] not in [status.value for status in AppMCPServerStatus]:

api/controllers/console/app/workflow_draft_variable.py

@@ -68,18 +68,13 @@ def _create_pagination_parser():
     return parser
 
 
-def _serialize_variable_type(workflow_draft_var: WorkflowDraftVariable) -> str:
-    value_type = workflow_draft_var.value_type
-    return value_type.exposed_type().value
-
-
 _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "id": fields.String,
     "type": fields.String(attribute=lambda model: model.get_variable_type()),
     "name": fields.String,
     "description": fields.String,
     "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
-    "value_type": fields.String(attribute=_serialize_variable_type),
+    "value_type": fields.String,
     "edited": fields.Boolean(attribute=lambda model: model.edited),
     "visible": fields.Boolean,
 }
@@ -95,7 +90,7 @@ _WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
     "name": fields.String,
     "description": fields.String,
     "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
-    "value_type": fields.String(attribute=_serialize_variable_type),
+    "value_type": fields.String,
     "edited": fields.Boolean(attribute=lambda model: model.edited),
     "visible": fields.Boolean,
 }
@@ -401,7 +396,7 @@ class EnvironmentVariableCollectionApi(Resource):
                     "name": v.name,
                     "description": v.description,
                     "selector": v.selector,
-                    "value_type": v.value_type.exposed_type().value,
+                    "value_type": v.value_type.value,
                     "value": v.value,
                     # Do not track edited for env vars.
                     "edited": False,

api/controllers/console/auth/error.py

@@ -27,19 +27,7 @@ class InvalidTokenError(BaseHTTPException):
 
 class PasswordResetRateLimitExceededError(BaseHTTPException):
     error_code = "password_reset_rate_limit_exceeded"
-    description = "Too many password reset emails have been sent. Please try again in 1 minute."
-    code = 429
-
-
-class EmailChangeRateLimitExceededError(BaseHTTPException):
-    error_code = "email_change_rate_limit_exceeded"
-    description = "Too many email change emails have been sent. Please try again in 1 minute."
-    code = 429
-
-
-class OwnerTransferRateLimitExceededError(BaseHTTPException):
-    error_code = "owner_transfer_rate_limit_exceeded"
-    description = "Too many owner transfer emails have been sent. Please try again in 1 minute."
+    description = "Too many password reset emails have been sent. Please try again in 1 minutes."
     code = 429
 
 
@@ -77,39 +65,3 @@ class EmailPasswordResetLimitError(BaseHTTPException):
     error_code = "email_password_reset_limit"
     description = "Too many failed password reset attempts. Please try again in 24 hours."
     code = 429
-
-
-class EmailChangeLimitError(BaseHTTPException):
-    error_code = "email_change_limit"
-    description = "Too many failed email change attempts. Please try again in 24 hours."
-    code = 429
-
-
-class EmailAlreadyInUseError(BaseHTTPException):
-    error_code = "email_already_in_use"
-    description = "A user with this email already exists."
-    code = 400
-
-
-class OwnerTransferLimitError(BaseHTTPException):
-    error_code = "owner_transfer_limit"
-    description = "Too many failed owner transfer attempts. Please try again in 24 hours."
-    code = 429
-
-
-class NotOwnerError(BaseHTTPException):
-    error_code = "not_owner"
-    description = "You are not the owner of the workspace."
-    code = 400
-
-
-class CannotTransferOwnerToSelfError(BaseHTTPException):
-    error_code = "cannot_transfer_owner_to_self"
-    description = "You cannot transfer ownership to yourself."
-    code = 400
-
-
-class MemberNotInTenantError(BaseHTTPException):
-    error_code = "member_not_in_tenant"
-    description = "The member is not in the workspace."
-    code = 400

api/controllers/console/datasets/error.py

@@ -25,6 +25,12 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
+class HighQualityDatasetOnlyError(BaseHTTPException):
+    error_code = "high_quality_dataset_only"
+    description = "Current operation only supports 'high-quality' datasets."
+    code = 400
+
+
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/console/workspace/account.py

@@ -4,20 +4,10 @@ import pytz
 from flask import request
 from flask_login import current_user
 from flask_restful import Resource, fields, marshal_with, reqparse
-from sqlalchemy import select
-from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
 from controllers.console import api
-from controllers.console.auth.error import (
-    EmailAlreadyInUseError,
-    EmailChangeLimitError,
-    EmailCodeError,
-    InvalidEmailError,
-    InvalidTokenError,
-)
-from controllers.console.error import AccountNotFound, EmailSendIpLimitError
 from controllers.console.workspace.error import (
     AccountAlreadyInitedError,
     CurrentPasswordIncorrectError,
@@ -28,17 +18,15 @@ from controllers.console.workspace.error import (
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_enabled,
-    enable_change_email,
     enterprise_license_required,
     only_edition_cloud,
     setup_required,
 )
 from extensions.ext_database import db
 from fields.member_fields import account_fields
-from libs.helper import TimestampField, email, extract_remote_ip, timezone
+from libs.helper import TimestampField, timezone
 from libs.login import login_required
 from models import AccountIntegrate, InvitationCode
-from models.account import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
@@ -381,134 +369,6 @@ class EducationAutoCompleteApi(Resource):
         return BillingService.EducationIdentity.autocomplete(args["keywords"], args["page"], args["limit"])
 
 
-class ChangeEmailSendEmailApi(Resource):
-    @enable_change_email
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("email", type=email, required=True, location="json")
-        parser.add_argument("language", type=str, required=False, location="json")
-        parser.add_argument("phase", type=str, required=False, location="json")
-        parser.add_argument("token", type=str, required=False, location="json")
-        args = parser.parse_args()
-
-        ip_address = extract_remote_ip(request)
-        if AccountService.is_email_send_ip_limit(ip_address):
-            raise EmailSendIpLimitError()
-
-        if args["language"] is not None and args["language"] == "zh-Hans":
-            language = "zh-Hans"
-        else:
-            language = "en-US"
-        account = None
-        user_email = args["email"]
-        if args["phase"] is not None and args["phase"] == "new_email":
-            if args["token"] is None:
-                raise InvalidTokenError()
-
-            reset_data = AccountService.get_change_email_data(args["token"])
-            if reset_data is None:
-                raise InvalidTokenError()
-            user_email = reset_data.get("email", "")
-
-            if user_email != current_user.email:
-                raise InvalidEmailError()
-        else:
-            with Session(db.engine) as session:
-                account = session.execute(select(Account).filter_by(email=args["email"])).scalar_one_or_none()
-            if account is None:
-                raise AccountNotFound()
-
-        token = AccountService.send_change_email_email(
-            account=account, email=args["email"], old_email=user_email, language=language, phase=args["phase"]
-        )
-        return {"result": "success", "data": token}
-
-
-class ChangeEmailCheckApi(Resource):
-    @enable_change_email
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("email", type=email, required=True, location="json")
-        parser.add_argument("code", type=str, required=True, location="json")
-        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-
-        user_email = args["email"]
-
-        is_change_email_error_rate_limit = AccountService.is_change_email_error_rate_limit(args["email"])
-        if is_change_email_error_rate_limit:
-            raise EmailChangeLimitError()
-
-        token_data = AccountService.get_change_email_data(args["token"])
-        if token_data is None:
-            raise InvalidTokenError()
-
-        if user_email != token_data.get("email"):
-            raise InvalidEmailError()
-
-        if args["code"] != token_data.get("code"):
-            AccountService.add_change_email_error_rate_limit(args["email"])
-            raise EmailCodeError()
-
-        # Verified, revoke the first token
-        AccountService.revoke_change_email_token(args["token"])
-
-        # Refresh token data by generating a new token
-        _, new_token = AccountService.generate_change_email_token(
-            user_email, code=args["code"], old_email=token_data.get("old_email"), additional_data={}
-        )
-
-        AccountService.reset_change_email_error_rate_limit(args["email"])
-        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
-
-
-class ChangeEmailResetApi(Resource):
-    @enable_change_email
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(account_fields)
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("new_email", type=email, required=True, location="json")
-        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-
-        reset_data = AccountService.get_change_email_data(args["token"])
-        if not reset_data:
-            raise InvalidTokenError()
-
-        AccountService.revoke_change_email_token(args["token"])
-
-        if not AccountService.check_email_unique(args["new_email"]):
-            raise EmailAlreadyInUseError()
-
-        old_email = reset_data.get("old_email", "")
-        if current_user.email != old_email:
-            raise AccountNotFound()
-
-        updated_account = AccountService.update_account(current_user, email=args["new_email"])
-
-        return updated_account
-
-
-class CheckEmailUnique(Resource):
-    @setup_required
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("email", type=email, required=True, location="json")
-        args = parser.parse_args()
-        if not AccountService.check_email_unique(args["email"]):
-            raise EmailAlreadyInUseError()
-        return {"result": "success"}
-
-
 # Register API resources
 api.add_resource(AccountInitApi, "/account/init")
 api.add_resource(AccountProfileApi, "/account/profile")
@@ -525,10 +385,5 @@ api.add_resource(AccountDeleteUpdateFeedbackApi, "/account/delete/feedback")
 api.add_resource(EducationVerifyApi, "/account/education/verify")
 api.add_resource(EducationApi, "/account/education")
 api.add_resource(EducationAutoCompleteApi, "/account/education/autocomplete")
-# Change email
-api.add_resource(ChangeEmailSendEmailApi, "/account/change-email")
-api.add_resource(ChangeEmailCheckApi, "/account/change-email/validity")
-api.add_resource(ChangeEmailResetApi, "/account/change-email/reset")
-api.add_resource(CheckEmailUnique, "/account/change-email/check-email-unique")
 # api.add_resource(AccountEmailApi, '/account/email')
 # api.add_resource(AccountEmailVerifyApi, '/account/email-verify')

api/controllers/console/workspace/error.py

@@ -13,6 +13,12 @@ class CurrentPasswordIncorrectError(BaseHTTPException):
     code = 400
 
 
+class ProviderRequestFailedError(BaseHTTPException):
+    error_code = "provider_request_failed"
+    description = None
+    code = 400
+
+
 class InvalidInvitationCodeError(BaseHTTPException):
     error_code = "invalid_invitation_code"
     description = "Invalid invitation code."

api/controllers/console/workspace/members.py

@@ -1,34 +1,22 @@
 from urllib import parse
 
-from flask import request
 from flask_login import current_user
 from flask_restful import Resource, abort, marshal_with, reqparse
 
 import services
 from configs import dify_config
 from controllers.console import api
-from controllers.console.auth.error import (
-    CannotTransferOwnerToSelfError,
-    EmailCodeError,
-    InvalidEmailError,
-    InvalidTokenError,
-    MemberNotInTenantError,
-    NotOwnerError,
-    OwnerTransferLimitError,
-)
-from controllers.console.error import EmailSendIpLimitError, WorkspaceMembersLimitExceeded
+from controllers.console.error import WorkspaceMembersLimitExceeded
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
-    is_allow_transfer_owner,
     setup_required,
 )
 from extensions.ext_database import db
 from fields.member_fields import account_with_role_list_fields
-from libs.helper import extract_remote_ip
 from libs.login import login_required
 from models.account import Account, TenantAccountRole
-from services.account_service import AccountService, RegisterService, TenantService
+from services.account_service import RegisterService, TenantService
 from services.errors.account import AccountAlreadyInTenantError
 from services.feature_service import FeatureService
 
@@ -168,146 +156,8 @@ class DatasetOperatorMemberListApi(Resource):
         return {"result": "success", "accounts": members}, 200
 
 
-class SendOwnerTransferEmailApi(Resource):
-    """Send owner transfer email."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_allow_transfer_owner
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("language", type=str, required=False, location="json")
-        args = parser.parse_args()
-        ip_address = extract_remote_ip(request)
-        if AccountService.is_email_send_ip_limit(ip_address):
-            raise EmailSendIpLimitError()
-
-        # check if the current user is the owner of the workspace
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
-            raise NotOwnerError()
-
-        if args["language"] is not None and args["language"] == "zh-Hans":
-            language = "zh-Hans"
-        else:
-            language = "en-US"
-
-        email = current_user.email
-
-        token = AccountService.send_owner_transfer_email(
-            account=current_user,
-            email=email,
-            language=language,
-            workspace_name=current_user.current_tenant.name,
-        )
-
-        return {"result": "success", "data": token}
-
-
-class OwnerTransferCheckApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_allow_transfer_owner
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("code", type=str, required=True, location="json")
-        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-        # check if the current user is the owner of the workspace
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
-            raise NotOwnerError()
-
-        user_email = current_user.email
-
-        is_owner_transfer_error_rate_limit = AccountService.is_owner_transfer_error_rate_limit(user_email)
-        if is_owner_transfer_error_rate_limit:
-            raise OwnerTransferLimitError()
-
-        token_data = AccountService.get_owner_transfer_data(args["token"])
-        if token_data is None:
-            raise InvalidTokenError()
-
-        if user_email != token_data.get("email"):
-            raise InvalidEmailError()
-
-        if args["code"] != token_data.get("code"):
-            AccountService.add_owner_transfer_error_rate_limit(user_email)
-            raise EmailCodeError()
-
-        # Verified, revoke the first token
-        AccountService.revoke_owner_transfer_token(args["token"])
-
-        # Refresh token data by generating a new token
-        _, new_token = AccountService.generate_owner_transfer_token(user_email, code=args["code"], additional_data={})
-
-        AccountService.reset_owner_transfer_error_rate_limit(user_email)
-        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
-
-
-class OwnerTransfer(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @is_allow_transfer_owner
-    def post(self, member_id):
-        parser = reqparse.RequestParser()
-        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-
-        # check if the current user is the owner of the workspace
-        if not TenantService.is_owner(current_user, current_user.current_tenant):
-            raise NotOwnerError()
-
-        if current_user.id == str(member_id):
-            raise CannotTransferOwnerToSelfError()
-
-        transfer_token_data = AccountService.get_owner_transfer_data(args["token"])
-        if not transfer_token_data:
-            raise InvalidTokenError()
-
-        if transfer_token_data.get("email") != current_user.email:
-            raise InvalidEmailError()
-
-        AccountService.revoke_owner_transfer_token(args["token"])
-
-        member = db.session.get(Account, str(member_id))
-        if not member:
-            abort(404)
-        else:
-            member_account = member
-        if not TenantService.is_member(member_account, current_user.current_tenant):
-            raise MemberNotInTenantError()
-
-        try:
-            assert member is not None, "Member not found"
-            TenantService.update_member_role(current_user.current_tenant, member, "owner", current_user)
-
-            AccountService.send_new_owner_transfer_notify_email(
-                account=member,
-                email=member.email,
-                workspace_name=current_user.current_tenant.name,
-            )
-
-            AccountService.send_old_owner_transfer_notify_email(
-                account=current_user,
-                email=current_user.email,
-                workspace_name=current_user.current_tenant.name,
-                new_owner_email=member.email,
-            )
-
-        except Exception as e:
-            raise ValueError(str(e))
-
-        return {"result": "success"}
-
-
 api.add_resource(MemberListApi, "/workspaces/current/members")
 api.add_resource(MemberInviteEmailApi, "/workspaces/current/members/invite-email")
 api.add_resource(MemberCancelInviteApi, "/workspaces/current/members/<uuid:member_id>")
 api.add_resource(MemberUpdateRoleApi, "/workspaces/current/members/<uuid:member_id>/update-role")
 api.add_resource(DatasetOperatorMemberListApi, "/workspaces/current/dataset-operators")
-# owner transfer
-api.add_resource(SendOwnerTransferEmailApi, "/workspaces/current/members/send-owner-transfer-confirm-email")
-api.add_resource(OwnerTransferCheckApi, "/workspaces/current/members/owner-transfer-check")
-api.add_resource(OwnerTransfer, "/workspaces/current/members/<uuid:member_id>/owner-transfer")

api/controllers/console/workspace/tool_providers.py

@@ -807,7 +807,7 @@ class ToolOAuthCustomClient(Resource):
                 tenant_id=current_user.current_tenant_id, provider=provider
             )
         )
-
+    
     @setup_required
     @login_required
     @account_initialization_required
@@ -845,7 +845,6 @@ class ToolBuiltinProviderGetCredentialInfoApi(Resource):
             )
         )
 
-
 class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required

api/controllers/console/wraps.py

@@ -235,29 +235,3 @@ def email_password_login_enabled(view):
         abort(403)
 
     return decorated
-
-
-def enable_change_email(view):
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if features.enable_change_email:
-            return view(*args, **kwargs)
-
-        # otherwise, return 403
-        abort(403)
-
-    return decorated
-
-
-def is_allow_transfer_owner(view):
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_features(current_user.current_tenant_id)
-        if features.is_allow_transfer_workspace:
-            return view(*args, **kwargs)
-
-        # otherwise, return 403
-        abort(403)
-
-    return decorated

api/controllers/inner_api/plugin/plugin.py

@@ -175,7 +175,7 @@ class PluginInvokeToolApi(Resource):
                     provider=payload.provider,
                     tool_name=payload.tool,
                     tool_parameters=payload.tool_parameters,
-                    credential_id=payload.credential_id,
+                    credential_id=payload.credential_id
                 ),
             )
 

api/controllers/service_api/dataset/error.py

@@ -25,6 +25,12 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
+class HighQualityDatasetOnlyError(BaseHTTPException):
+    error_code = "high_quality_dataset_only"
+    description = "Current operation only supports 'high-quality' datasets."
+    code = 400
+
+
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/core/agent/plugin_entities.py

@@ -41,7 +41,6 @@ class AgentStrategyParameter(PluginParameter):
         APP_SELECTOR = CommonParameterType.APP_SELECTOR.value
         MODEL_SELECTOR = CommonParameterType.MODEL_SELECTOR.value
         TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR.value
-        ANY = CommonParameterType.ANY.value
 
         # deprecated, should not use.
         SYSTEM_FILES = CommonParameterType.SYSTEM_FILES.value

api/core/agent/strategy/plugin.py

@@ -60,5 +60,7 @@ class PluginAgentStrategy(BaseAgentStrategy):
             conversation_id=conversation_id,
             app_id=app_id,
             message_id=message_id,
-            context=PluginInvokeContext(credentials=credentials or InvokeCredentials()),
+            context=PluginInvokeContext(
+                credentials=credentials or InvokeCredentials()
+            ),
         )

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

@@ -16,10 +16,9 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
 )
 from core.moderation.base import ModerationError
-from core.variables.variables import VariableUnion
 from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
 from core.workflow.entities.variable_pool import VariablePool
-from core.workflow.system_variable import SystemVariable
+from core.workflow.enums import SystemVariableKey
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
@@ -65,7 +64,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         if not workflow:
             raise ValueError("Workflow not initialized")
 
-        user_id: str | None = None
+        user_id = None
         if self.application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
             end_user = db.session.query(EndUser).filter(EndUser.id == self.application_generate_entity.user_id).first()
             if end_user:
@@ -137,25 +136,23 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 session.commit()
 
             # Create a variable pool.
-            system_inputs = SystemVariable(
-                query=query,
-                files=files,
-                conversation_id=self.conversation.id,
-                user_id=user_id,
-                dialogue_count=self._dialogue_count,
-                app_id=app_config.app_id,
-                workflow_id=app_config.workflow_id,
-                workflow_execution_id=self.application_generate_entity.workflow_run_id,
-            )
+            system_inputs = {
+                SystemVariableKey.QUERY: query,
+                SystemVariableKey.FILES: files,
+                SystemVariableKey.CONVERSATION_ID: self.conversation.id,
+                SystemVariableKey.USER_ID: user_id,
+                SystemVariableKey.DIALOGUE_COUNT: self._dialogue_count,
+                SystemVariableKey.APP_ID: app_config.app_id,
+                SystemVariableKey.WORKFLOW_ID: app_config.workflow_id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: self.application_generate_entity.workflow_run_id,
+            }
 
             # init variable pool
             variable_pool = VariablePool(
                 system_variables=system_inputs,
                 user_inputs=inputs,
                 environment_variables=workflow.environment_variables,
-                # Based on the definition of `VariableUnion`,
-                # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-                conversation_variables=cast(list[VariableUnion], conversation_variables),
+                conversation_variables=conversation_variables,
             )
 
             # init graph

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

@@ -61,12 +61,12 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.entities.workflow_execution import WorkflowExecutionStatus, WorkflowType
+from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
@@ -116,16 +116,16 @@ class AdvancedChatAppGenerateTaskPipeline:
 
         self._workflow_cycle_manager = WorkflowCycleManager(
             application_generate_entity=application_generate_entity,
-            workflow_system_variables=SystemVariable(
-                query=message.query,
-                files=application_generate_entity.files,
-                conversation_id=conversation.id,
-                user_id=user_session_id,
-                dialogue_count=dialogue_count,
-                app_id=application_generate_entity.app_config.app_id,
-                workflow_id=workflow.id,
-                workflow_execution_id=application_generate_entity.workflow_run_id,
-            ),
+            workflow_system_variables={
+                SystemVariableKey.QUERY: message.query,
+                SystemVariableKey.FILES: application_generate_entity.files,
+                SystemVariableKey.CONVERSATION_ID: conversation.id,
+                SystemVariableKey.USER_ID: user_session_id,
+                SystemVariableKey.DIALOGUE_COUNT: dialogue_count,
+                SystemVariableKey.APP_ID: application_generate_entity.app_config.app_id,
+                SystemVariableKey.WORKFLOW_ID: workflow.id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: application_generate_entity.workflow_run_id,
+            },
             workflow_info=CycleManagerWorkflowInfo(
                 workflow_id=workflow.id,
                 workflow_type=WorkflowType(workflow.type),

api/core/app/apps/base_app_runner.py

@@ -38,6 +38,69 @@ _logger = logging.getLogger(__name__)
 
 
 class AppRunner:
+    def get_pre_calculate_rest_tokens(
+        self,
+        app_record: App,
+        model_config: ModelConfigWithCredentialsEntity,
+        prompt_template_entity: PromptTemplateEntity,
+        inputs: Mapping[str, str],
+        files: Sequence["File"],
+        query: Optional[str] = None,
+    ) -> int:
+        """
+        Get pre calculate rest tokens
+        :param app_record: app record
+        :param model_config: model config entity
+        :param prompt_template_entity: prompt template entity
+        :param inputs: inputs
+        :param files: files
+        :param query: query
+        :return:
+        """
+        # Invoke model
+        model_instance = ModelInstance(
+            provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
+        )
+
+        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+
+        max_tokens = 0
+        for parameter_rule in model_config.model_schema.parameter_rules:
+            if parameter_rule.name == "max_tokens" or (
+                parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
+            ):
+                max_tokens = (
+                    model_config.parameters.get(parameter_rule.name)
+                    or model_config.parameters.get(parameter_rule.use_template or "")
+                ) or 0
+
+        if model_context_tokens is None:
+            return -1
+
+        if max_tokens is None:
+            max_tokens = 0
+
+        # get prompt messages without memory and context
+        prompt_messages, stop = self.organize_prompt_messages(
+            app_record=app_record,
+            model_config=model_config,
+            prompt_template_entity=prompt_template_entity,
+            inputs=inputs,
+            files=files,
+            query=query,
+        )
+
+        prompt_tokens = model_instance.get_llm_num_tokens(prompt_messages)
+
+        rest_tokens: int = model_context_tokens - max_tokens - prompt_tokens
+        if rest_tokens < 0:
+            raise InvokeBadRequestError(
+                "Query or prefix prompt is too long, you can reduce the prefix prompt, "
+                "or shrink the max token, or switch to a llm with a larger token limit size."
+            )
+
+        return rest_tokens
+
     def recalc_llm_max_tokens(
         self, model_config: ModelConfigWithCredentialsEntity, prompt_messages: list[PromptMessage]
     ):

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

@@ -11,7 +11,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
 from core.workflow.entities.variable_pool import VariablePool
-from core.workflow.system_variable import SystemVariable
+from core.workflow.enums import SystemVariableKey
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
@@ -95,14 +95,13 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             files = self.application_generate_entity.files
 
             # Create a variable pool.
-
-            system_inputs = SystemVariable(
-                files=files,
-                user_id=user_id,
-                app_id=app_config.app_id,
-                workflow_id=app_config.workflow_id,
-                workflow_execution_id=self.application_generate_entity.workflow_execution_id,
-            )
+            system_inputs = {
+                SystemVariableKey.FILES: files,
+                SystemVariableKey.USER_ID: user_id,
+                SystemVariableKey.APP_ID: app_config.app_id,
+                SystemVariableKey.WORKFLOW_ID: app_config.workflow_id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: self.application_generate_entity.workflow_execution_id,
+            }
 
             variable_pool = VariablePool(
                 system_variables=system_inputs,

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

@@ -54,10 +54,10 @@ from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTas
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.entities.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
+from core.workflow.enums import SystemVariableKey
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from extensions.ext_database import db
 from models.account import Account
@@ -107,13 +107,13 @@ class WorkflowAppGenerateTaskPipeline:
 
         self._workflow_cycle_manager = WorkflowCycleManager(
             application_generate_entity=application_generate_entity,
-            workflow_system_variables=SystemVariable(
-                files=application_generate_entity.files,
-                user_id=user_session_id,
-                app_id=application_generate_entity.app_config.app_id,
-                workflow_id=workflow.id,
-                workflow_execution_id=application_generate_entity.workflow_execution_id,
-            ),
+            workflow_system_variables={
+                SystemVariableKey.FILES: application_generate_entity.files,
+                SystemVariableKey.USER_ID: user_session_id,
+                SystemVariableKey.APP_ID: application_generate_entity.app_config.app_id,
+                SystemVariableKey.WORKFLOW_ID: workflow.id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: application_generate_entity.workflow_execution_id,
+            },
             workflow_info=CycleManagerWorkflowInfo(
                 workflow_id=workflow.id,
                 workflow_type=WorkflowType(workflow.type),

api/core/app/apps/workflow_app_runner.py

@@ -62,7 +62,6 @@ from core.workflow.graph_engine.entities.event import (
 from core.workflow.graph_engine.entities.graph import Graph
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
-from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
@@ -167,7 +166,7 @@ class WorkflowBasedAppRunner(AppRunner):
 
         # init variable pool
         variable_pool = VariablePool(
-            system_variables=SystemVariable.empty(),
+            system_variables={},
             user_inputs={},
             environment_variables=workflow.environment_variables,
         )
@@ -264,7 +263,7 @@ class WorkflowBasedAppRunner(AppRunner):
 
         # init variable pool
         variable_pool = VariablePool(
-            system_variables=SystemVariable.empty(),
+            system_variables={},
             user_inputs={},
             environment_variables=workflow.environment_variables,
         )

api/core/app/task_pipeline/exc.py

@@ -10,3 +10,8 @@ class RecordNotFoundError(TaskPipilineError):
 class WorkflowRunNotFoundError(RecordNotFoundError):
     def __init__(self, workflow_run_id: str):
         super().__init__("WorkflowRun", workflow_run_id)
+
+
+class WorkflowNodeExecutionNotFoundError(RecordNotFoundError):
+    def __init__(self, workflow_node_execution_id: str):
+        super().__init__("WorkflowNodeExecution", workflow_node_execution_id)

api/core/entities/parameter_entities.py

@@ -14,7 +14,6 @@ class CommonParameterType(StrEnum):
     APP_SELECTOR = "app-selector"
     MODEL_SELECTOR = "model-selector"
     TOOLS_SELECTOR = "array[tools]"
-    ANY = "any"
 
     # Dynamic select parameter
     # Once you are not sure about the available options until authorization is done

api/core/file/tool_file_parser.py

@@ -7,6 +7,13 @@ if TYPE_CHECKING:
 _tool_file_manager_factory: Callable[[], "ToolFileManager"] | None = None
 
 
+class ToolFileParser:
+    @staticmethod
+    def get_tool_file_manager() -> "ToolFileManager":
+        assert _tool_file_manager_factory is not None
+        return _tool_file_manager_factory()
+
+
 def set_tool_file_manager_factory(factory: Callable[[], "ToolFileManager"]) -> None:
     global _tool_file_manager_factory
     _tool_file_manager_factory = factory

api/core/helper/url_signer.py

@@ -0,0 +1,52 @@
+import base64
+import hashlib
+import hmac
+import os
+import time
+
+from pydantic import BaseModel, Field
+
+from configs import dify_config
+
+
+class SignedUrlParams(BaseModel):
+    sign_key: str = Field(..., description="The sign key")
+    timestamp: str = Field(..., description="Timestamp")
+    nonce: str = Field(..., description="Nonce")
+    sign: str = Field(..., description="Signature")
+
+
+class UrlSigner:
+    @classmethod
+    def get_signed_url(cls, url: str, sign_key: str, prefix: str) -> str:
+        signed_url_params = cls.get_signed_url_params(sign_key, prefix)
+        return (
+            f"{url}?timestamp={signed_url_params.timestamp}"
+            f"&nonce={signed_url_params.nonce}&sign={signed_url_params.sign}"
+        )
+
+    @classmethod
+    def get_signed_url_params(cls, sign_key: str, prefix: str) -> SignedUrlParams:
+        timestamp = str(int(time.time()))
+        nonce = os.urandom(16).hex()
+        sign = cls._sign(sign_key, timestamp, nonce, prefix)
+
+        return SignedUrlParams(sign_key=sign_key, timestamp=timestamp, nonce=nonce, sign=sign)
+
+    @classmethod
+    def verify(cls, sign_key: str, timestamp: str, nonce: str, sign: str, prefix: str) -> bool:
+        recalculated_sign = cls._sign(sign_key, timestamp, nonce, prefix)
+
+        return sign == recalculated_sign
+
+    @classmethod
+    def _sign(cls, sign_key: str, timestamp: str, nonce: str, prefix: str) -> str:
+        if not dify_config.SECRET_KEY:
+            raise Exception("SECRET_KEY is not set")
+
+        data_to_sign = f"{prefix}|{sign_key}|{timestamp}|{nonce}"
+        secret_key = dify_config.SECRET_KEY.encode()
+        sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
+        encoded_sign = base64.urlsafe_b64encode(sign).decode()
+
+        return encoded_sign

api/core/llm_generator/llm_generator.py

@@ -148,11 +148,9 @@ class LLMGenerator:
 
             model_manager = ModelManager()
 
-            model_instance = model_manager.get_model_instance(
+            model_instance = model_manager.get_default_model_instance(
                 tenant_id=tenant_id,
                 model_type=ModelType.LLM,
-                provider=model_config.get("provider", ""),
-                model=model_config.get("name", ""),
             )
 
             try:

api/core/mcp/auth/auth_flow.py

@@ -240,7 +240,7 @@ def refresh_authorization(
     response = requests.post(token_url, data=params)
     if not response.ok:
         raise ValueError(f"Token refresh failed: HTTP {response.status_code}")
-    return OAuthTokens.model_validate(response.json())
+    return OAuthTokens.parse_obj(response.json())
 
 
 def register_client(

api/core/mcp/server/streamable_http.py

@@ -148,7 +148,9 @@ class MCPServerStreamableHTTPRequestHandler:
         if not self.end_user:
             raise ValueError("User not found")
         request = cast(types.CallToolRequest, self.request.root)
-        args = request.params.arguments or {}
+        args = request.params.arguments
+        if not args:
+            raise ValueError("No arguments provided")
         if self.app.mode in {AppMode.WORKFLOW.value}:
             args = {"inputs": args}
         elif self.app.mode in {AppMode.COMPLETION.value}:

api/core/mcp/session/base_session.py

@@ -1,7 +1,7 @@
 import logging
 import queue
 from collections.abc import Callable
-from concurrent.futures import Future, ThreadPoolExecutor, TimeoutError
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import ExitStack
 from datetime import timedelta
 from types import TracebackType
@@ -171,41 +171,23 @@ class BaseSession(
         self._session_read_timeout_seconds = read_timeout_seconds
         self._in_flight = {}
         self._exit_stack = ExitStack()
-        # Initialize executor and future to None for proper cleanup checks
-        self._executor: ThreadPoolExecutor | None = None
-        self._receiver_future: Future | None = None
 
     def __enter__(self) -> Self:
-        # The thread pool is dedicated to running `_receive_loop`. Setting `max_workers` to 1
-        # ensures no unnecessary threads are created.
-        self._executor = ThreadPoolExecutor(max_workers=1)
+        self._executor = ThreadPoolExecutor()
         self._receiver_future = self._executor.submit(self._receive_loop)
         return self
 
     def check_receiver_status(self) -> None:
-        """`check_receiver_status` ensures that any exceptions raised during the
-        execution of `_receive_loop` are retrieved and propagated."""
-        if self._receiver_future and self._receiver_future.done():
+        if self._receiver_future.done():
             self._receiver_future.result()
 
     def __exit__(
         self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
     ) -> None:
+        self._exit_stack.close()
         self._read_stream.put(None)
         self._write_stream.put(None)
 
-        # Wait for the receiver loop to finish
-        if self._receiver_future:
-            try:
-                self._receiver_future.result(timeout=5.0)  # Wait up to 5 seconds
-            except TimeoutError:
-                # If the receiver loop is still running after timeout, we'll force shutdown
-                pass
-
-        # Shutdown the executor
-        if self._executor:
-            self._executor.shutdown(wait=True)
-
     def send_request(
         self,
         request: SendRequestT,

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -284,8 +284,7 @@ class AliyunDataTrace(BaseTraceInstance):
             else:
                 node_span = self.build_workflow_task_span(trace_id, workflow_span_id, trace_info, node_execution)
             return node_span
-        except Exception as e:
-            logging.debug(f"Error occurred in build_workflow_node_span: {e}", exc_info=True)
+        except Exception:
             return None
 
     def get_workflow_node_status(self, node_execution: WorkflowNodeExecution) -> Status:
@@ -307,7 +306,7 @@ class AliyunDataTrace(BaseTraceInstance):
             start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
             end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
             attributes={
-                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id") or "",
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
                 GEN_AI_SPAN_KIND: GenAISpanKind.TASK.value,
                 GEN_AI_FRAMEWORK: "dify",
                 INPUT_VALUE: json.dumps(node_execution.inputs, ensure_ascii=False),
@@ -382,7 +381,7 @@ class AliyunDataTrace(BaseTraceInstance):
             start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
             end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
             attributes={
-                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id") or "",
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
                 GEN_AI_SPAN_KIND: GenAISpanKind.LLM.value,
                 GEN_AI_FRAMEWORK: "dify",
                 GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
@@ -416,7 +415,7 @@ class AliyunDataTrace(BaseTraceInstance):
                 start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
                 end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
                 attributes={
-                    GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id") or "",
+                    GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
                     GEN_AI_USER_ID: str(user_id),
                     GEN_AI_SPAN_KIND: GenAISpanKind.CHAIN.value,
                     GEN_AI_FRAMEWORK: "dify",

api/core/plugin/entities/parameters.py

@@ -5,7 +5,6 @@ from pydantic import BaseModel, Field, field_validator
 
 from core.entities.parameter_entities import CommonParameterType
 from core.tools.entities.common_entities import I18nObject
-from core.workflow.nodes.base.entities import NumberType
 
 
 class PluginParameterOption(BaseModel):
@@ -39,7 +38,6 @@ class PluginParameterType(enum.StrEnum):
     APP_SELECTOR = CommonParameterType.APP_SELECTOR.value
     MODEL_SELECTOR = CommonParameterType.MODEL_SELECTOR.value
     TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR.value
-    ANY = CommonParameterType.ANY.value
     DYNAMIC_SELECT = CommonParameterType.DYNAMIC_SELECT.value
 
     # deprecated, should not use.
@@ -153,10 +151,6 @@ def cast_parameter_value(typ: enum.StrEnum, value: Any, /):
                 if value and not isinstance(value, list):
                     raise ValueError("The tools selector must be a list.")
                 return value
-            case PluginParameterType.ANY:
-                if value and not isinstance(value, str | dict | list | NumberType):
-                    raise ValueError("The var selector must be a string, dictionary, list or number.")
-                return value
             case PluginParameterType.ARRAY:
                 if not isinstance(value, list):
                     # Try to parse JSON string for arrays

api/core/plugin/entities/plugin.py

@@ -135,6 +135,17 @@ class PluginEntity(PluginInstallation):
         return self
 
 
+class GithubPackage(BaseModel):
+    repo: str
+    version: str
+    package: str
+
+
+class GithubVersion(BaseModel):
+    repo: str
+    version: str
+
+
 class GenericProviderID:
     organization: str
     plugin_name: str

api/core/plugin/impl/oauth.py

@@ -56,6 +56,7 @@ class OAuthHandler(BasePluginClient):
         Get credentials from the given request.
         """
 
+
         try:
             # encode request to raw http request
             raw_request_bytes = self._convert_request_to_raw_data(request)

api/core/prompt/advanced_prompt_transform.py

@@ -158,7 +158,7 @@ class AdvancedPromptTransform(PromptTransform):
 
             if prompt_item.edition_type == "basic" or not prompt_item.edition_type:
                 if self.with_variable_tmpl:
-                    vp = VariablePool.empty()
+                    vp = VariablePool()
                     for k, v in inputs.items():
                         if k.startswith("#"):
                             vp.add(k[1:-1].split("."), v)

api/core/rag/cleaner/unstructured/unstructured_extra_whitespace_cleaner.py

@@ -0,0 +1,12 @@
+"""Abstract interface for document clean implementations."""
+
+from core.rag.cleaner.cleaner_base import BaseCleaner
+
+
+class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
+    def clean(self, content) -> str:
+        """clean document content."""
+        from unstructured.cleaners.core import clean_extra_whitespace
+
+        # Returns "ITEM 1A: RISK FACTORS"
+        return clean_extra_whitespace(content)

api/core/rag/cleaner/unstructured/unstructured_group_broken_paragraphs_cleaner.py

@@ -0,0 +1,15 @@
+"""Abstract interface for document clean implementations."""
+
+from core.rag.cleaner.cleaner_base import BaseCleaner
+
+
+class UnstructuredGroupBrokenParagraphsCleaner(BaseCleaner):
+    def clean(self, content) -> str:
+        """clean document content."""
+        import re
+
+        from unstructured.cleaners.core import group_broken_paragraphs
+
+        para_split_re = re.compile(r"(\s*\n\s*){3}")
+
+        return group_broken_paragraphs(content, paragraph_split=para_split_re)

api/core/rag/cleaner/unstructured/unstructured_non_ascii_chars_cleaner.py

@@ -0,0 +1,12 @@
+"""Abstract interface for document clean implementations."""
+
+from core.rag.cleaner.cleaner_base import BaseCleaner
+
+
+class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
+    def clean(self, content) -> str:
+        """clean document content."""
+        from unstructured.cleaners.core import clean_non_ascii_chars
+
+        # Returns "This text contains non-ascii characters!"
+        return clean_non_ascii_chars(content)

api/core/rag/cleaner/unstructured/unstructured_replace_unicode_quotes_cleaner.py

@@ -0,0 +1,12 @@
+"""Abstract interface for document clean implementations."""
+
+from core.rag.cleaner.cleaner_base import BaseCleaner
+
+
+class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
+    def clean(self, content) -> str:
+        """Replaces unicode quote characters, such as the \x91 character in a string."""
+
+        from unstructured.cleaners.core import replace_unicode_quotes
+
+        return replace_unicode_quotes(content)

api/core/rag/cleaner/unstructured/unstructured_translate_text_cleaner.py

@@ -0,0 +1,11 @@
+"""Abstract interface for document clean implementations."""
+
+from core.rag.cleaner.cleaner_base import BaseCleaner
+
+
+class UnstructuredTranslateTextCleaner(BaseCleaner):
+    def clean(self, content) -> str:
+        """clean document content."""
+        from unstructured.cleaners.translate import translate_text
+
+        return translate_text(content)

api/core/rag/datasource/vdb/tidb_on_qdrant/tidb_entities.py

@@ -0,0 +1,17 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class ClusterEntity(BaseModel):
+    """
+    Model Config Entity.
+    """
+
+    name: str
+    cluster_id: str
+    displayName: str
+    region: str
+    spendingLimit: Optional[int] = 1000
+    version: str
+    createdBy: str

api/core/rag/extractor/blob/blob.py

@@ -9,7 +9,8 @@ from __future__ import annotations
 
 import contextlib
 import mimetypes
-from collections.abc import Generator, Mapping
+from abc import ABC, abstractmethod
+from collections.abc import Generator, Iterable, Mapping
 from io import BufferedReader, BytesIO
 from pathlib import Path, PurePath
 from typing import Any, Optional, Union
@@ -142,3 +143,21 @@ class Blob(BaseModel):
         if self.source:
             str_repr += f" {self.source}"
         return str_repr
+
+
+class BlobLoader(ABC):
+    """Abstract interface for blob loaders implementation.
+
+    Implementer should be able to load raw content from a datasource system according
+    to some criteria and return the raw content lazily as a stream of blobs.
+    """
+
+    @abstractmethod
+    def yield_blobs(
+        self,
+    ) -> Iterable[Blob]:
+        """A lazy loader for raw data represented by Blob object.
+
+        Returns:
+            A generator over blobs
+        """

api/core/rag/extractor/unstructured/unstructured_pdf_extractor.py

@@ -0,0 +1,47 @@
+import logging
+
+from core.rag.extractor.extractor_base import BaseExtractor
+from core.rag.models.document import Document
+
+logger = logging.getLogger(__name__)
+
+
+class UnstructuredPDFExtractor(BaseExtractor):
+    """Load pdf files.
+
+
+    Args:
+        file_path: Path to the file to load.
+
+        api_url: Unstructured API URL
+
+        api_key: Unstructured API Key
+    """
+
+    def __init__(self, file_path: str, api_url: str, api_key: str):
+        """Initialize with file path."""
+        self._file_path = file_path
+        self._api_url = api_url
+        self._api_key = api_key
+
+    def extract(self) -> list[Document]:
+        if self._api_url:
+            from unstructured.partition.api import partition_via_api
+
+            elements = partition_via_api(
+                filename=self._file_path, api_url=self._api_url, api_key=self._api_key, strategy="auto"
+            )
+        else:
+            from unstructured.partition.pdf import partition_pdf
+
+            elements = partition_pdf(filename=self._file_path, strategy="auto")
+
+        from unstructured.chunking.title import chunk_by_title
+
+        chunks = chunk_by_title(elements, max_characters=2000, combine_text_under_n_chars=2000)
+        documents = []
+        for chunk in chunks:
+            text = chunk.text.strip()
+            documents.append(Document(page_content=text))
+
+        return documents

api/core/rag/extractor/unstructured/unstructured_text_extractor.py

@@ -0,0 +1,34 @@
+import logging
+
+from core.rag.extractor.extractor_base import BaseExtractor
+from core.rag.models.document import Document
+
+logger = logging.getLogger(__name__)
+
+
+class UnstructuredTextExtractor(BaseExtractor):
+    """Load msg files.
+
+
+    Args:
+        file_path: Path to the file to load.
+    """
+
+    def __init__(self, file_path: str, api_url: str):
+        """Initialize with file path."""
+        self._file_path = file_path
+        self._api_url = api_url
+
+    def extract(self) -> list[Document]:
+        from unstructured.partition.text import partition_text
+
+        elements = partition_text(filename=self._file_path)
+        from unstructured.chunking.title import chunk_by_title
+
+        chunks = chunk_by_title(elements, max_characters=2000, combine_text_under_n_chars=2000)
+        documents = []
+        for chunk in chunks:
+            text = chunk.text.strip()
+            documents.append(Document(page_content=text))
+
+        return documents

api/core/rag/splitter/text_splitter.py

@@ -10,6 +10,7 @@ from typing import (
     Any,
     Literal,
     Optional,
+    TypedDict,
     TypeVar,
     Union,
 )
@@ -167,6 +168,167 @@ class TextSplitter(BaseDocumentTransformer, ABC):
         raise NotImplementedError
 
 
+class CharacterTextSplitter(TextSplitter):
+    """Splitting text that looks at characters."""
+
+    def __init__(self, separator: str = "\n\n", **kwargs: Any) -> None:
+        """Create a new TextSplitter."""
+        super().__init__(**kwargs)
+        self._separator = separator
+
+    def split_text(self, text: str) -> list[str]:
+        """Split incoming text and return chunks."""
+        # First we naively split the large input into a bunch of smaller ones.
+        splits = _split_text_with_regex(text, self._separator, self._keep_separator)
+        _separator = "" if self._keep_separator else self._separator
+        _good_splits_lengths = []  # cache the lengths of the splits
+        if splits:
+            _good_splits_lengths.extend(self._length_function(splits))
+        return self._merge_splits(splits, _separator, _good_splits_lengths)
+
+
+class LineType(TypedDict):
+    """Line type as typed dict."""
+
+    metadata: dict[str, str]
+    content: str
+
+
+class HeaderType(TypedDict):
+    """Header type as typed dict."""
+
+    level: int
+    name: str
+    data: str
+
+
+class MarkdownHeaderTextSplitter:
+    """Splitting markdown files based on specified headers."""
+
+    def __init__(self, headers_to_split_on: list[tuple[str, str]], return_each_line: bool = False):
+        """Create a new MarkdownHeaderTextSplitter.
+
+        Args:
+            headers_to_split_on: Headers we want to track
+            return_each_line: Return each line w/ associated headers
+        """
+        # Output line-by-line or aggregated into chunks w/ common headers
+        self.return_each_line = return_each_line
+        # Given the headers we want to split on,
+        # (e.g., "#, ##, etc") order by length
+        self.headers_to_split_on = sorted(headers_to_split_on, key=lambda split: len(split[0]), reverse=True)
+
+    def aggregate_lines_to_chunks(self, lines: list[LineType]) -> list[Document]:
+        """Combine lines with common metadata into chunks
+        Args:
+            lines: Line of text / associated header metadata
+        """
+        aggregated_chunks: list[LineType] = []
+
+        for line in lines:
+            if aggregated_chunks and aggregated_chunks[-1]["metadata"] == line["metadata"]:
+                # If the last line in the aggregated list
+                # has the same metadata as the current line,
+                # append the current content to the last lines's content
+                aggregated_chunks[-1]["content"] += "  \n" + line["content"]
+            else:
+                # Otherwise, append the current line to the aggregated list
+                aggregated_chunks.append(line)
+
+        return [Document(page_content=chunk["content"], metadata=chunk["metadata"]) for chunk in aggregated_chunks]
+
+    def split_text(self, text: str) -> list[Document]:
+        """Split markdown file
+        Args:
+            text: Markdown file"""
+
+        # Split the input text by newline character ("\n").
+        lines = text.split("\n")
+        # Final output
+        lines_with_metadata: list[LineType] = []
+        # Content and metadata of the chunk currently being processed
+        current_content: list[str] = []
+        current_metadata: dict[str, str] = {}
+        # Keep track of the nested header structure
+        # header_stack: List[Dict[str, Union[int, str]]] = []
+        header_stack: list[HeaderType] = []
+        initial_metadata: dict[str, str] = {}
+
+        for line in lines:
+            stripped_line = line.strip()
+            # Check each line against each of the header types (e.g., #, ##)
+            for sep, name in self.headers_to_split_on:
+                # Check if line starts with a header that we intend to split on
+                if stripped_line.startswith(sep) and (
+                    # Header with no text OR header is followed by space
+                    # Both are valid conditions that sep is being used a header
+                    len(stripped_line) == len(sep) or stripped_line[len(sep)] == " "
+                ):
+                    # Ensure we are tracking the header as metadata
+                    if name is not None:
+                        # Get the current header level
+                        current_header_level = sep.count("#")
+
+                        # Pop out headers of lower or same level from the stack
+                        while header_stack and header_stack[-1]["level"] >= current_header_level:
+                            # We have encountered a new header
+                            # at the same or higher level
+                            popped_header = header_stack.pop()
+                            # Clear the metadata for the
+                            # popped header in initial_metadata
+                            if popped_header["name"] in initial_metadata:
+                                initial_metadata.pop(popped_header["name"])
+
+                        # Push the current header to the stack
+                        header: HeaderType = {
+                            "level": current_header_level,
+                            "name": name,
+                            "data": stripped_line[len(sep) :].strip(),
+                        }
+                        header_stack.append(header)
+                        # Update initial_metadata with the current header
+                        initial_metadata[name] = header["data"]
+
+                    # Add the previous line to the lines_with_metadata
+                    # only if current_content is not empty
+                    if current_content:
+                        lines_with_metadata.append(
+                            {
+                                "content": "\n".join(current_content),
+                                "metadata": current_metadata.copy(),
+                            }
+                        )
+                        current_content.clear()
+
+                    break
+            else:
+                if stripped_line:
+                    current_content.append(stripped_line)
+                elif current_content:
+                    lines_with_metadata.append(
+                        {
+                            "content": "\n".join(current_content),
+                            "metadata": current_metadata.copy(),
+                        }
+                    )
+                    current_content.clear()
+
+            current_metadata = initial_metadata.copy()
+
+        if current_content:
+            lines_with_metadata.append({"content": "\n".join(current_content), "metadata": current_metadata})
+
+        # lines_with_metadata has each line with associated header metadata
+        # aggregate these into chunks based on common metadata
+        if not self.return_each_line:
+            return self.aggregate_lines_to_chunks(lines_with_metadata)
+        else:
+            return [
+                Document(page_content=chunk["content"], metadata=chunk["metadata"]) for chunk in lines_with_metadata
+            ]
+
+
+# should be in newer Python versions (3.10+)
 # @dataclass(frozen=True, kw_only=True, slots=True)
 @dataclass(frozen=True)
 class Tokenizer:

api/core/tools/__base/tool_runtime.py

@@ -17,7 +17,7 @@ class ToolRuntime(BaseModel):
     invoke_from: Optional[InvokeFrom] = None
     tool_invoke_from: Optional[ToolInvokeFrom] = None
     credentials: dict[str, Any] = Field(default_factory=dict)
-    credential_type: CredentialType = Field(default=CredentialType.API_KEY)
+    credential_type: Optional[CredentialType] = CredentialType.API_KEY
     runtime_parameters: dict[str, Any] = Field(default_factory=dict)
 
 

api/core/tools/builtin_tool/provider.py

@@ -125,7 +125,7 @@ class BuiltinToolProviderController(ToolProviderController):
         if credential_type == CredentialType.API_KEY.value:
             return self.entity.credentials_schema.copy() if self.entity.credentials_schema else []
         raise ValueError(f"Invalid credential type: {credential_type}")
-
+    
     def get_oauth_client_schema(self) -> list[ProviderConfig]:
         """
         returns the oauth client schema of the provider

api/core/tools/entities/tool_entities.py

@@ -16,7 +16,6 @@ from core.plugin.entities.parameters import (
     cast_parameter_value,
     init_frontend_parameter,
 )
-from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.constants import TOOL_SELECTOR_MODEL_IDENTITY
 
@@ -180,10 +179,6 @@ class ToolInvokeMessage(BaseModel):
         data: Mapping[str, Any] = Field(..., description="Detailed log data")
         metadata: Optional[Mapping[str, Any]] = Field(default=None, description="The metadata of the log")
 
-    class RetrieverResourceMessage(BaseModel):
-        retriever_resources: list[RetrievalSourceMetadata] = Field(..., description="retriever resources")
-        context: str = Field(..., description="context")
-
     class MessageType(Enum):
         TEXT = "text"
         IMAGE = "image"
@@ -196,22 +191,13 @@ class ToolInvokeMessage(BaseModel):
         FILE = "file"
         LOG = "log"
         BLOB_CHUNK = "blob_chunk"
-        RETRIEVER_RESOURCES = "retriever_resources"
 
     type: MessageType = MessageType.TEXT
     """
         plain text, image url or link url
     """
     message: (
-        JsonMessage
-        | TextMessage
-        | BlobChunkMessage
-        | BlobMessage
-        | LogMessage
-        | FileMessage
-        | None
-        | VariableMessage
-        | RetrieverResourceMessage
+        JsonMessage | TextMessage | BlobChunkMessage | BlobMessage | LogMessage | FileMessage | None | VariableMessage
     )
     meta: dict[str, Any] | None = None
 
@@ -257,7 +243,6 @@ class ToolParameter(PluginParameter):
         FILES = PluginParameterType.FILES.value
         APP_SELECTOR = PluginParameterType.APP_SELECTOR.value
         MODEL_SELECTOR = PluginParameterType.MODEL_SELECTOR.value
-        ANY = PluginParameterType.ANY.value
         DYNAMIC_SELECT = PluginParameterType.DYNAMIC_SELECT.value
 
         # MCP object and array type parameters

api/core/tools/tool_manager.py

@@ -207,7 +207,7 @@ class ToolManager:
                     # if the provider has been deleted, raise an error
                     if builtin_provider is None:
                         raise ToolProviderNotFoundError(f"provider has been deleted: {credential_id}")
-
+                
                 # fallback to the default provider
                 if builtin_provider is None:
                     # use the default provider

api/core/tools/utils/encryption.py

@@ -127,7 +127,6 @@ class ProviderConfigEncrypter:
 def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
     return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache
 
-
 def create_tool_provider_encrypter(tenant_id: str, controller: ToolProviderController):
     cache = SingletonProviderCredentialsCache(
         tenant_id=tenant_id,

api/core/tools/utils/system_oauth_encryption.py

@@ -1,13 +1,12 @@
 import base64
 import hashlib
+import json
 import logging
-from collections.abc import Mapping
 from typing import Any, Optional
 
 from Crypto.Cipher import AES
 from Crypto.Random import get_random_bytes
 from Crypto.Util.Padding import pad, unpad
-from pydantic import TypeAdapter
 
 from configs import dify_config
 
@@ -43,7 +42,7 @@ class SystemOAuthEncrypter:
         # Generate a fixed 256-bit key using SHA-256
         self.key = hashlib.sha256(secret_key.encode()).digest()
 
-    def encrypt_oauth_params(self, oauth_params: Mapping[str, Any]) -> str:
+    def encrypt_oauth_params(self, oauth_params: str) -> str:
         """
         Encrypt OAuth parameters.
 
@@ -58,6 +57,9 @@ class SystemOAuthEncrypter:
             ValueError: If oauth_params is invalid
         """
 
+        if not oauth_params:
+            raise ValueError("oauth_params cannot be empty")
+
         try:
             # Generate random IV (16 bytes)
             iv = get_random_bytes(16)
@@ -66,7 +68,7 @@ class SystemOAuthEncrypter:
             cipher = AES.new(self.key, AES.MODE_CBC, iv)
 
             # Encrypt data
-            padded_data = pad(TypeAdapter(dict).dump_json(dict(oauth_params)), AES.block_size)
+            padded_data = pad(oauth_params.encode("utf-8"), AES.block_size)
             encrypted_data = cipher.encrypt(padded_data)
 
             # Combine IV and encrypted data
@@ -78,7 +80,7 @@ class SystemOAuthEncrypter:
         except Exception as e:
             raise OAuthEncryptionError(f"Encryption failed: {str(e)}") from e
 
-    def decrypt_oauth_params(self, encrypted_data: str) -> Mapping[str, Any]:
+    def decrypt_oauth_params(self, encrypted_data: str) -> dict[str, Any]:
         """
         Decrypt OAuth parameters.
 
@@ -118,13 +120,16 @@ class SystemOAuthEncrypter:
             unpadded_data = unpad(decrypted_data, AES.block_size)
 
             # Parse JSON
-            oauth_params: Mapping[str, Any] = TypeAdapter(Mapping[str, Any]).validate_json(unpadded_data)
+            params_json = unpadded_data.decode("utf-8")
+            oauth_params = json.loads(params_json)
 
             if not isinstance(oauth_params, dict):
                 raise ValueError("Decrypted data is not a valid dictionary")
 
             return oauth_params
 
+        except (ValueError, TypeError) as e:
+            raise OAuthEncryptionError(f"Decryption failed: {str(e)}") from e
         except Exception as e:
             raise OAuthEncryptionError(f"Decryption failed: {str(e)}") from e
 
@@ -161,7 +166,7 @@ def get_system_oauth_encrypter() -> SystemOAuthEncrypter:
 
 
 # Convenience functions for backward compatibility
-def encrypt_system_oauth_params(oauth_params: Mapping[str, Any]) -> str:
+def encrypt_system_oauth_params(oauth_params: str) -> str:
     """
     Encrypt OAuth parameters using the global encrypter.
 
@@ -174,7 +179,7 @@ def encrypt_system_oauth_params(oauth_params: Mapping[str, Any]) -> str:
     return get_system_oauth_encrypter().encrypt_oauth_params(oauth_params)
 
 
-def decrypt_system_oauth_params(encrypted_data: str) -> Mapping[str, Any]:
+def decrypt_system_oauth_params(encrypted_data: str) -> dict[str, Any]:
     """
     Decrypt OAuth parameters using the global encrypter.
 

api/core/tools/utils/uuid_utils.py

@@ -5,6 +5,7 @@ def is_valid_uuid(uuid_str: str | None) -> bool:
     if uuid_str is None or len(uuid_str) == 0:
         return False
     try:
+
         uuid.UUID(uuid_str)
         return True
     except Exception:

api/core/variables/segments.py

@@ -1,9 +1,9 @@
 import json
 import sys
 from collections.abc import Mapping, Sequence
-from typing import Annotated, Any, TypeAlias
+from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Discriminator, Tag, field_validator
+from pydantic import BaseModel, ConfigDict, field_validator
 
 from core.file import File
 
@@ -11,11 +11,6 @@ from .types import SegmentType
 
 
 class Segment(BaseModel):
-    """Segment is runtime type used during the execution of workflow.
-
-    Note: this class is abstract, you should use subclasses of this class instead.
-    """
-
     model_config = ConfigDict(frozen=True)
 
     value_type: SegmentType
@@ -78,7 +73,7 @@ class StringSegment(Segment):
 
 
 class FloatSegment(Segment):
-    value_type: SegmentType = SegmentType.FLOAT
+    value_type: SegmentType = SegmentType.NUMBER
     value: float
     # NOTE(QuantumGhost): seems that the equality for FloatSegment with `NaN` value has some problems.
     # The following tests cannot pass.
@@ -97,7 +92,7 @@ class FloatSegment(Segment):
 
 
 class IntegerSegment(Segment):
-    value_type: SegmentType = SegmentType.INTEGER
+    value_type: SegmentType = SegmentType.NUMBER
     value: int
 
 
@@ -186,46 +181,3 @@ class ArrayFileSegment(ArraySegment):
     @property
     def text(self) -> str:
         return ""
-
-
-def get_segment_discriminator(v: Any) -> SegmentType | None:
-    if isinstance(v, Segment):
-        return v.value_type
-    elif isinstance(v, dict):
-        value_type = v.get("value_type")
-        if value_type is None:
-            return None
-        try:
-            seg_type = SegmentType(value_type)
-        except ValueError:
-            return None
-        return seg_type
-    else:
-        # return None if the discriminator value isn't found
-        return None
-
-
-# The `SegmentUnion`` type is used to enable serialization and deserialization with Pydantic.
-# Use `Segment` for type hinting when serialization is not required.
-#
-# Note:
-# - All variants in `SegmentUnion` must inherit from the `Segment` class.
-# - The union must include all non-abstract subclasses of `Segment`, except:
-#   - `SegmentGroup`, which is not added to the variable pool.
-#   - `Variable` and its subclasses, which are handled by `VariableUnion`.
-SegmentUnion: TypeAlias = Annotated[
-    (
-        Annotated[NoneSegment, Tag(SegmentType.NONE)]
-        | Annotated[StringSegment, Tag(SegmentType.STRING)]
-        | Annotated[FloatSegment, Tag(SegmentType.FLOAT)]
-        | Annotated[IntegerSegment, Tag(SegmentType.INTEGER)]
-        | Annotated[ObjectSegment, Tag(SegmentType.OBJECT)]
-        | Annotated[FileSegment, Tag(SegmentType.FILE)]
-        | Annotated[ArrayAnySegment, Tag(SegmentType.ARRAY_ANY)]
-        | Annotated[ArrayStringSegment, Tag(SegmentType.ARRAY_STRING)]
-        | Annotated[ArrayNumberSegment, Tag(SegmentType.ARRAY_NUMBER)]
-        | Annotated[ArrayObjectSegment, Tag(SegmentType.ARRAY_OBJECT)]
-        | Annotated[ArrayFileSegment, Tag(SegmentType.ARRAY_FILE)]
-    ),
-    Discriminator(get_segment_discriminator),
-]

api/core/variables/types.py

@@ -1,27 +1,8 @@
-from collections.abc import Mapping
 from enum import StrEnum
-from typing import Any, Optional
-
-from core.file.models import File
-
-
-class ArrayValidation(StrEnum):
-    """Strategy for validating array elements"""
-
-    # Skip element validation (only check array container)
-    NONE = "none"
-
-    # Validate the first element (if array is non-empty)
-    FIRST = "first"
-
-    # Validate all elements in the array.
-    ALL = "all"
 
 
 class SegmentType(StrEnum):
     NUMBER = "number"
-    INTEGER = "integer"
-    FLOAT = "float"
     STRING = "string"
     OBJECT = "object"
     SECRET = "secret"
@@ -38,139 +19,16 @@ class SegmentType(StrEnum):
 
     GROUP = "group"
 
-    def is_array_type(self) -> bool:
+    def is_array_type(self):
         return self in _ARRAY_TYPES
 
-    @classmethod
-    def infer_segment_type(cls, value: Any) -> Optional["SegmentType"]:
-        """
-        Attempt to infer the `SegmentType` based on the Python type of the `value` parameter.
-
-        Returns `None` if no appropriate `SegmentType` can be determined for the given `value`.
-        For example, this may occur if the input is a generic Python object of type `object`.
-        """
-
-        if isinstance(value, list):
-            elem_types: set[SegmentType] = set()
-            for i in value:
-                segment_type = cls.infer_segment_type(i)
-                if segment_type is None:
-                    return None
-
-                elem_types.add(segment_type)
-
-            if len(elem_types) != 1:
-                if elem_types.issubset(_NUMERICAL_TYPES):
-                    return SegmentType.ARRAY_NUMBER
-                return SegmentType.ARRAY_ANY
-            elif all(i.is_array_type() for i in elem_types):
-                return SegmentType.ARRAY_ANY
-            match elem_types.pop():
-                case SegmentType.STRING:
-                    return SegmentType.ARRAY_STRING
-                case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
-                    return SegmentType.ARRAY_NUMBER
-                case SegmentType.OBJECT:
-                    return SegmentType.ARRAY_OBJECT
-                case SegmentType.FILE:
-                    return SegmentType.ARRAY_FILE
-                case SegmentType.NONE:
-                    return SegmentType.ARRAY_ANY
-                case _:
-                    # This should be unreachable.
-                    raise ValueError(f"not supported value {value}")
-        if value is None:
-            return SegmentType.NONE
-        elif isinstance(value, int) and not isinstance(value, bool):
-            return SegmentType.INTEGER
-        elif isinstance(value, float):
-            return SegmentType.FLOAT
-        elif isinstance(value, str):
-            return SegmentType.STRING
-        elif isinstance(value, dict):
-            return SegmentType.OBJECT
-        elif isinstance(value, File):
-            return SegmentType.FILE
-        else:
-            return None
-
-    def _validate_array(self, value: Any, array_validation: ArrayValidation) -> bool:
-        if not isinstance(value, list):
-            return False
-        # Skip element validation if array is empty
-        if len(value) == 0:
-            return True
-        if self == SegmentType.ARRAY_ANY:
-            return True
-        element_type = _ARRAY_ELEMENT_TYPES_MAPPING[self]
-
-        if array_validation == ArrayValidation.NONE:
-            return True
-        elif array_validation == ArrayValidation.FIRST:
-            return element_type.is_valid(value[0])
-        else:
-            return all([element_type.is_valid(i, array_validation=ArrayValidation.NONE)] for i in value)
-
-    def is_valid(self, value: Any, array_validation: ArrayValidation = ArrayValidation.FIRST) -> bool:
-        """
-        Check if a value matches the segment type.
-        Users of `SegmentType` should call this method, instead of using
-        `isinstance` manually.
-
-        Args:
-            value: The value to validate
-            array_validation: Validation strategy for array types (ignored for non-array types)
-
-        Returns:
-            True if the value matches the type under the given validation strategy
-        """
-        if self.is_array_type():
-            return self._validate_array(value, array_validation)
-        elif self == SegmentType.NUMBER:
-            return isinstance(value, (int, float))
-        elif self == SegmentType.STRING:
-            return isinstance(value, str)
-        elif self == SegmentType.OBJECT:
-            return isinstance(value, dict)
-        elif self == SegmentType.SECRET:
-            return isinstance(value, str)
-        elif self == SegmentType.FILE:
-            return isinstance(value, File)
-        elif self == SegmentType.NONE:
-            return value is None
-        else:
-            raise AssertionError("this statement should be unreachable.")
-
-    def exposed_type(self) -> "SegmentType":
-        """Returns the type exposed to the frontend.
-
-        The frontend treats `INTEGER` and `FLOAT` as `NUMBER`, so these are returned as `NUMBER` here.
-        """
-        if self in (SegmentType.INTEGER, SegmentType.FLOAT):
-            return SegmentType.NUMBER
-        return self
-
-
-_ARRAY_ELEMENT_TYPES_MAPPING: Mapping[SegmentType, SegmentType] = {
-    # ARRAY_ANY does not have correpond element type.
-    SegmentType.ARRAY_STRING: SegmentType.STRING,
-    SegmentType.ARRAY_NUMBER: SegmentType.NUMBER,
-    SegmentType.ARRAY_OBJECT: SegmentType.OBJECT,
-    SegmentType.ARRAY_FILE: SegmentType.FILE,
-}
 
 _ARRAY_TYPES = frozenset(
-    list(_ARRAY_ELEMENT_TYPES_MAPPING.keys())
-    + [
-        SegmentType.ARRAY_ANY,
-    ]
-)
-
-
-_NUMERICAL_TYPES = frozenset(
     [
-        SegmentType.NUMBER,
-        SegmentType.INTEGER,
-        SegmentType.FLOAT,
+        SegmentType.ARRAY_ANY,
+        SegmentType.ARRAY_STRING,
+        SegmentType.ARRAY_NUMBER,
+        SegmentType.ARRAY_OBJECT,
+        SegmentType.ARRAY_FILE,
     ]
 )

api/core/variables/variables.py

@@ -1,8 +1,8 @@
 from collections.abc import Sequence
-from typing import Annotated, TypeAlias, cast
+from typing import cast
 from uuid import uuid4
 
-from pydantic import Discriminator, Field, Tag
+from pydantic import Field
 
 from core.helper import encrypter
 
@@ -20,7 +20,6 @@ from .segments import (
     ObjectSegment,
     Segment,
     StringSegment,
-    get_segment_discriminator,
 )
 from .types import SegmentType
 
@@ -28,10 +27,6 @@ from .types import SegmentType
 class Variable(Segment):
     """
     A variable is a segment that has a name.
-
-    It is mainly used to store segments and their selector in VariablePool.
-
-    Note: this class is abstract, you should use subclasses of this class instead.
     """
 
     id: str = Field(
@@ -98,28 +93,3 @@ class FileVariable(FileSegment, Variable):
 
 class ArrayFileVariable(ArrayFileSegment, ArrayVariable):
     pass
-
-
-# The `VariableUnion`` type is used to enable serialization and deserialization with Pydantic.
-# Use `Variable` for type hinting when serialization is not required.
-#
-# Note:
-# - All variants in `VariableUnion` must inherit from the `Variable` class.
-# - The union must include all non-abstract subclasses of `Segment`, except:
-VariableUnion: TypeAlias = Annotated[
-    (
-        Annotated[NoneVariable, Tag(SegmentType.NONE)]
-        | Annotated[StringVariable, Tag(SegmentType.STRING)]
-        | Annotated[FloatVariable, Tag(SegmentType.FLOAT)]
-        | Annotated[IntegerVariable, Tag(SegmentType.INTEGER)]
-        | Annotated[ObjectVariable, Tag(SegmentType.OBJECT)]
-        | Annotated[FileVariable, Tag(SegmentType.FILE)]
-        | Annotated[ArrayAnyVariable, Tag(SegmentType.ARRAY_ANY)]
-        | Annotated[ArrayStringVariable, Tag(SegmentType.ARRAY_STRING)]
-        | Annotated[ArrayNumberVariable, Tag(SegmentType.ARRAY_NUMBER)]
-        | Annotated[ArrayObjectVariable, Tag(SegmentType.ARRAY_OBJECT)]
-        | Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
-        | Annotated[SecretVariable, Tag(SegmentType.SECRET)]
-    ),
-    Discriminator(get_segment_discriminator),
-]

api/core/workflow/entities/variable_pool.py

@@ -1,7 +1,7 @@
 import re
 from collections import defaultdict
 from collections.abc import Mapping, Sequence
-from typing import Annotated, Any, Union, cast
+from typing import Any, Union
 
 from pydantic import BaseModel, Field
 
@@ -9,9 +9,8 @@ from core.file import File, FileAttribute, file_manager
 from core.variables import Segment, SegmentGroup, Variable
 from core.variables.consts import MIN_SELECTORS_LENGTH
 from core.variables.segments import FileSegment, NoneSegment
-from core.variables.variables import VariableUnion
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, ENVIRONMENT_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
-from core.workflow.system_variable import SystemVariable
+from core.workflow.enums import SystemVariableKey
 from factories import variable_factory
 
 VariableValue = Union[str, int, float, dict, list, File]
@@ -24,31 +23,31 @@ class VariablePool(BaseModel):
     # The first element of the selector is the node id, it's the first-level key in the dictionary.
     # Other elements of the selector are the keys in the second-level dictionary. To get the key, we hash the
     # elements of the selector except the first one.
-    variable_dictionary: defaultdict[str, Annotated[dict[int, VariableUnion], Field(default_factory=dict)]] = Field(
+    variable_dictionary: dict[str, dict[int, Segment]] = Field(
         description="Variables mapping",
         default=defaultdict(dict),
     )
-
-    # The `user_inputs` is used only when constructing the inputs for the `StartNode`. It's not used elsewhere.
+    # TODO: This user inputs is not used for pool.
     user_inputs: Mapping[str, Any] = Field(
         description="User inputs",
         default_factory=dict,
     )
-    system_variables: SystemVariable = Field(
+    system_variables: Mapping[SystemVariableKey, Any] = Field(
         description="System variables",
+        default_factory=dict,
     )
-    environment_variables: Sequence[VariableUnion] = Field(
+    environment_variables: Sequence[Variable] = Field(
         description="Environment variables.",
         default_factory=list,
     )
-    conversation_variables: Sequence[VariableUnion] = Field(
+    conversation_variables: Sequence[Variable] = Field(
         description="Conversation variables.",
         default_factory=list,
     )
 
     def model_post_init(self, context: Any, /) -> None:
-        # Create a mapping from field names to SystemVariableKey enum values
-        self._add_system_variables(self.system_variables)
+        for key, value in self.system_variables.items():
+            self.add((SYSTEM_VARIABLE_NODE_ID, key.value), value)
         # Add environment variables to the variable pool
         for var in self.environment_variables:
             self.add((ENVIRONMENT_VARIABLE_NODE_ID, var.name), var)
@@ -84,22 +83,8 @@ class VariablePool(BaseModel):
             segment = variable_factory.build_segment(value)
             variable = variable_factory.segment_to_variable(segment=segment, selector=selector)
 
-        key, hash_key = self._selector_to_keys(selector)
-        # Based on the definition of `VariableUnion`,
-        # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-        self.variable_dictionary[key][hash_key] = cast(VariableUnion, variable)
-
-    @classmethod
-    def _selector_to_keys(cls, selector: Sequence[str]) -> tuple[str, int]:
-        return selector[0], hash(tuple(selector[1:]))
-
-    def _has(self, selector: Sequence[str]) -> bool:
-        key, hash_key = self._selector_to_keys(selector)
-        if key not in self.variable_dictionary:
-            return False
-        if hash_key not in self.variable_dictionary[key]:
-            return False
-        return True
+        hash_key = hash(tuple(selector[1:]))
+        self.variable_dictionary[selector[0]][hash_key] = variable
 
     def get(self, selector: Sequence[str], /) -> Segment | None:
         """
@@ -117,8 +102,8 @@ class VariablePool(BaseModel):
         if len(selector) < MIN_SELECTORS_LENGTH:
             return None
 
-        key, hash_key = self._selector_to_keys(selector)
-        value: Segment | None = self.variable_dictionary[key].get(hash_key)
+        hash_key = hash(tuple(selector[1:]))
+        value = self.variable_dictionary[selector[0]].get(hash_key)
 
         if value is None:
             selector, attr = selector[:-1], selector[-1]
@@ -151,8 +136,8 @@ class VariablePool(BaseModel):
         if len(selector) == 1:
             self.variable_dictionary[selector[0]] = {}
             return
-        key, hash_key = self._selector_to_keys(selector)
-        self.variable_dictionary[key].pop(hash_key, None)
+        hash_key = hash(tuple(selector[1:]))
+        self.variable_dictionary[selector[0]].pop(hash_key, None)
 
     def convert_template(self, template: str, /):
         parts = VARIABLE_PATTERN.split(template)
@@ -169,20 +154,3 @@ class VariablePool(BaseModel):
         if isinstance(segment, FileSegment):
             return segment
         return None
-
-    def _add_system_variables(self, system_variable: SystemVariable):
-        sys_var_mapping = system_variable.to_dict()
-        for key, value in sys_var_mapping.items():
-            if value is None:
-                continue
-            selector = (SYSTEM_VARIABLE_NODE_ID, key)
-            # If the system variable already exists, do not add it again.
-            # This ensures that we can keep the id of the system variables intact.
-            if self._has(selector):
-                continue
-            self.add(selector, value)  # type: ignore
-
-    @classmethod
-    def empty(cls) -> "VariablePool":
-        """Create an empty variable pool."""
-        return cls(system_variables=SystemVariable.empty())

api/core/workflow/entities/workflow_entities.py

@@ -0,0 +1,79 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow.nodes.base import BaseIterationState, BaseLoopState, BaseNode
+from models.enums import UserFrom
+from models.workflow import Workflow, WorkflowType
+
+from .node_entities import NodeRunResult
+from .variable_pool import VariablePool
+
+
+class WorkflowNodeAndResult:
+    node: BaseNode
+    result: Optional[NodeRunResult] = None
+
+    def __init__(self, node: BaseNode, result: Optional[NodeRunResult] = None):
+        self.node = node
+        self.result = result
+
+
+class WorkflowRunState:
+    tenant_id: str
+    app_id: str
+    workflow_id: str
+    workflow_type: WorkflowType
+    user_id: str
+    user_from: UserFrom
+    invoke_from: InvokeFrom
+
+    workflow_call_depth: int
+
+    start_at: float
+    variable_pool: VariablePool
+
+    total_tokens: int = 0
+
+    workflow_nodes_and_results: list[WorkflowNodeAndResult]
+
+    class NodeRun(BaseModel):
+        node_id: str
+        iteration_node_id: str
+        loop_node_id: str
+
+    workflow_node_runs: list[NodeRun]
+    workflow_node_steps: int
+
+    current_iteration_state: Optional[BaseIterationState]
+    current_loop_state: Optional[BaseLoopState]
+
+    def __init__(
+        self,
+        workflow: Workflow,
+        start_at: float,
+        variable_pool: VariablePool,
+        user_id: str,
+        user_from: UserFrom,
+        invoke_from: InvokeFrom,
+        workflow_call_depth: int,
+    ):
+        self.workflow_id = workflow.id
+        self.tenant_id = workflow.tenant_id
+        self.app_id = workflow.app_id
+        self.workflow_type = WorkflowType.value_of(workflow.type)
+        self.user_id = user_id
+        self.user_from = user_from
+        self.invoke_from = invoke_from
+        self.workflow_call_depth = workflow_call_depth
+
+        self.start_at = start_at
+        self.variable_pool = variable_pool
+
+        self.total_tokens = 0
+
+        self.workflow_node_steps = 1
+        self.workflow_node_runs = []
+        self.current_iteration_state = None
+        self.current_loop_state = None

api/core/workflow/graph_engine/entities/graph_runtime_state.py

@@ -17,12 +17,8 @@ class GraphRuntimeState(BaseModel):
     """total tokens"""
     llm_usage: LLMUsage = LLMUsage.empty_usage()
     """llm usage info"""
-
-    # The `outputs` field stores the final output values generated by executing workflows or chatflows.
-    #
-    # Note: Since the type of this field is `dict[str, Any]`, its values may not remain consistent
-    # after a serialization and deserialization round trip.
     outputs: dict[str, Any] = {}
+    """outputs"""
 
     node_run_steps: int = 0
     """node run steps"""

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

@@ -1,5 +1,4 @@
 from collections.abc import Mapping, Sequence
-from decimal import Decimal
 from typing import Any, Optional
 
 from configs import dify_config
@@ -115,10 +114,8 @@ class CodeNode(BaseNode[CodeNodeData]):
             )
 
         if isinstance(value, float):
-            decimal_value = Decimal(str(value)).normalize()
-            precision = -decimal_value.as_tuple().exponent if decimal_value.as_tuple().exponent < 0 else 0  # type: ignore[operator]
             # raise error if precision is too high
-            if precision > dify_config.CODE_MAX_PRECISION:
+            if len(str(value).split(".")[1]) > dify_config.CODE_MAX_PRECISION:
                 raise OutputValidationError(
                     f"Output variable `{variable}` has too high precision,"
                     f" it must be less than {dify_config.CODE_MAX_PRECISION} digits."

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

@@ -1,29 +1,11 @@
 from collections.abc import Mapping
-from typing import Annotated, Any, Literal, Optional
+from typing import Any, Literal, Optional
 
-from pydantic import AfterValidator, BaseModel, Field
+from pydantic import BaseModel, Field
 
-from core.variables.types import SegmentType
 from core.workflow.nodes.base import BaseLoopNodeData, BaseLoopState, BaseNodeData
 from core.workflow.utils.condition.entities import Condition
 
-_VALID_VAR_TYPE = frozenset(
-    [
-        SegmentType.STRING,
-        SegmentType.NUMBER,
-        SegmentType.OBJECT,
-        SegmentType.ARRAY_STRING,
-        SegmentType.ARRAY_NUMBER,
-        SegmentType.ARRAY_OBJECT,
-    ]
-)
-
-
-def _is_valid_var_type(seg_type: SegmentType) -> SegmentType:
-    if seg_type not in _VALID_VAR_TYPE:
-        raise ValueError(...)
-    return seg_type
-
 
 class LoopVariableData(BaseModel):
     """
@@ -31,7 +13,7 @@ class LoopVariableData(BaseModel):
     """
 
     label: str
-    var_type: Annotated[SegmentType, AfterValidator(_is_valid_var_type)]
+    var_type: Literal["string", "number", "object", "array[string]", "array[number]", "array[object]"]
     value_type: Literal["variable", "constant"]
     value: Optional[Any | list[str]] = None
 

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

@@ -7,9 +7,14 @@ from typing import TYPE_CHECKING, Any, Literal, cast
 
 from configs import dify_config
 from core.variables import (
+    ArrayNumberSegment,
+    ArrayObjectSegment,
+    ArrayStringSegment,
     IntegerSegment,
+    ObjectSegment,
     Segment,
     SegmentType,
+    StringSegment,
 )
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
@@ -34,7 +39,6 @@ from core.workflow.nodes.enums import NodeType
 from core.workflow.nodes.event import NodeEvent, RunCompletedEvent
 from core.workflow.nodes.loop.entities import LoopNodeData
 from core.workflow.utils.condition.processor import ConditionProcessor
-from factories.variable_factory import TypeMismatchError, build_segment_with_type
 
 if TYPE_CHECKING:
     from core.workflow.entities.variable_pool import VariablePool
@@ -501,21 +505,23 @@ class LoopNode(BaseNode[LoopNodeData]):
         return variable_mapping
 
     @staticmethod
-    def _get_segment_for_constant(var_type: SegmentType, value: Any) -> Segment:
+    def _get_segment_for_constant(var_type: str, value: Any) -> Segment:
         """Get the appropriate segment type for a constant value."""
+        segment_mapping: dict[str, tuple[type[Segment], SegmentType]] = {
+            "string": (StringSegment, SegmentType.STRING),
+            "number": (IntegerSegment, SegmentType.NUMBER),
+            "object": (ObjectSegment, SegmentType.OBJECT),
+            "array[string]": (ArrayStringSegment, SegmentType.ARRAY_STRING),
+            "array[number]": (ArrayNumberSegment, SegmentType.ARRAY_NUMBER),
+            "array[object]": (ArrayObjectSegment, SegmentType.ARRAY_OBJECT),
+        }
         if var_type in ["array[string]", "array[number]", "array[object]"]:
-            if value and isinstance(value, str):
+            if value:
                 value = json.loads(value)
             else:
                 value = []
-        try:
-            return build_segment_with_type(var_type, value)
-        except TypeMismatchError as type_exc:
-            # Attempt to parse the value as a JSON-encoded string, if applicable.
-            if not isinstance(value, str):
-                raise
-            try:
-                value = json.loads(value)
-            except ValueError:
-                raise type_exc
-            return build_segment_with_type(var_type, value)
+        segment_info = segment_mapping.get(var_type)
+        if not segment_info:
+            raise ValueError(f"Invalid variable type: {var_type}")
+        segment_class, value_type = segment_info
+        return segment_class(value=value, value_type=value_type)

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

@@ -16,7 +16,7 @@ class StartNode(BaseNode[StartNodeData]):
 
     def _run(self) -> NodeRunResult:
         node_inputs = dict(self.graph_runtime_state.variable_pool.user_inputs)
-        system_inputs = self.graph_runtime_state.variable_pool.system_variables.to_dict()
+        system_inputs = self.graph_runtime_state.variable_pool.system_variables
 
         # TODO: System variables should be directly accessible, no need for special handling
         # Set system variables as node outputs.

api/core/workflow/nodes/tool/tool_node.py

@@ -22,7 +22,7 @@ from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.event import AgentLogEvent
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
-from core.workflow.nodes.event import RunCompletedEvent, RunRetrieverResourceEvent, RunStreamChunkEvent
+from core.workflow.nodes.event import RunCompletedEvent, RunStreamChunkEvent
 from core.workflow.utils.variable_template_parser import VariableTemplateParser
 from extensions.ext_database import db
 from factories import file_factory
@@ -373,12 +373,6 @@ class ToolNode(BaseNode[ToolNodeData]):
                     agent_logs.append(agent_log)
 
                 yield agent_log
-            elif message.type == ToolInvokeMessage.MessageType.RETRIEVER_RESOURCES:
-                assert isinstance(message.message, ToolInvokeMessage.RetrieverResourceMessage)
-                yield RunRetrieverResourceEvent(
-                    retriever_resources=message.message.retriever_resources,
-                    context=message.message.context,
-                )
 
         # Add agent_logs to outputs['json'] to ensure frontend can access thinking process
         json_output: list[dict[str, Any]] = []

api/core/workflow/nodes/variable_assigner/v1/node.py

@@ -130,7 +130,6 @@ class VariableAssignerNode(BaseNode[VariableAssignerData]):
 
 
 def get_zero_value(t: SegmentType):
-    # TODO(QuantumGhost): this should be a method of `SegmentType`.
     match t:
         case SegmentType.ARRAY_OBJECT | SegmentType.ARRAY_STRING | SegmentType.ARRAY_NUMBER:
             return variable_factory.build_segment([])
@@ -138,10 +137,6 @@ def get_zero_value(t: SegmentType):
             return variable_factory.build_segment({})
         case SegmentType.STRING:
             return variable_factory.build_segment("")
-        case SegmentType.INTEGER:
-            return variable_factory.build_segment(0)
-        case SegmentType.FLOAT:
-            return variable_factory.build_segment(0.0)
         case SegmentType.NUMBER:
             return variable_factory.build_segment(0)
         case _:

api/core/workflow/nodes/variable_assigner/v2/constants.py

@@ -1,6 +1,5 @@
 from core.variables import SegmentType
 
-# Note: This mapping is duplicated with `get_zero_value`. Consider refactoring to avoid redundancy.
 EMPTY_VALUE_MAPPING = {
     SegmentType.STRING: "",
     SegmentType.NUMBER: 0,

api/core/workflow/nodes/variable_assigner/v2/helpers.py

@@ -10,16 +10,10 @@ def is_operation_supported(*, variable_type: SegmentType, operation: Operation):
         case Operation.OVER_WRITE | Operation.CLEAR:
             return True
         case Operation.SET:
-            return variable_type in {
-                SegmentType.OBJECT,
-                SegmentType.STRING,
-                SegmentType.NUMBER,
-                SegmentType.INTEGER,
-                SegmentType.FLOAT,
-            }
+            return variable_type in {SegmentType.OBJECT, SegmentType.STRING, SegmentType.NUMBER}
         case Operation.ADD | Operation.SUBTRACT | Operation.MULTIPLY | Operation.DIVIDE:
             # Only number variable can be added, subtracted, multiplied or divided
-            return variable_type in {SegmentType.NUMBER, SegmentType.INTEGER, SegmentType.FLOAT}
+            return variable_type == SegmentType.NUMBER
         case Operation.APPEND | Operation.EXTEND:
             # Only array variable can be appended or extended
             return variable_type in {
@@ -52,7 +46,7 @@ def is_constant_input_supported(*, variable_type: SegmentType, operation: Operat
     match variable_type:
         case SegmentType.STRING | SegmentType.OBJECT:
             return operation in {Operation.OVER_WRITE, Operation.SET}
-        case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
+        case SegmentType.NUMBER:
             return operation in {
                 Operation.OVER_WRITE,
                 Operation.SET,
@@ -72,7 +66,7 @@ def is_input_value_valid(*, variable_type: SegmentType, operation: Operation, va
         case SegmentType.STRING:
             return isinstance(value, str)
 
-        case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
+        case SegmentType.NUMBER:
             if not isinstance(value, int | float):
                 return False
             if operation == Operation.DIVIDE and value == 0:

api/core/workflow/system_variable.py

@@ -1,89 +0,0 @@
-from collections.abc import Sequence
-from typing import Any
-
-from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
-
-from core.file.models import File
-from core.workflow.enums import SystemVariableKey
-
-
-class SystemVariable(BaseModel):
-    """A model for managing system variables.
-
-    Fields with a value of `None` are treated as absent and will not be included
-    in the variable pool.
-    """
-
-    model_config = ConfigDict(
-        extra="forbid",
-        serialize_by_alias=True,
-        validate_by_alias=True,
-    )
-
-    user_id: str | None = None
-
-    # Ideally, `app_id` and `workflow_id` should be required and not `None`.
-    # However, there are scenarios in the codebase where these fields are not set.
-    # To maintain compatibility, they are marked as optional here.
-    app_id: str | None = None
-    workflow_id: str | None = None
-
-    files: Sequence[File] = Field(default_factory=list)
-
-    # NOTE: The `workflow_execution_id` field was previously named `workflow_run_id`.
-    # To maintain compatibility with existing workflows, it must be serialized
-    # as `workflow_run_id` in dictionaries or JSON objects, and also referenced
-    # as `workflow_run_id` in the variable pool.
-    workflow_execution_id: str | None = Field(
-        validation_alias=AliasChoices("workflow_execution_id", "workflow_run_id"),
-        serialization_alias="workflow_run_id",
-        default=None,
-    )
-    # Chatflow related fields.
-    query: str | None = None
-    conversation_id: str | None = None
-    dialogue_count: int | None = None
-
-    @model_validator(mode="before")
-    @classmethod
-    def validate_json_fields(cls, data):
-        if isinstance(data, dict):
-            # For JSON validation, only allow workflow_run_id
-            if "workflow_execution_id" in data and "workflow_run_id" not in data:
-                # This is likely from direct instantiation, allow it
-                return data
-            elif "workflow_execution_id" in data and "workflow_run_id" in data:
-                # Both present, remove workflow_execution_id
-                data = data.copy()
-                data.pop("workflow_execution_id")
-                return data
-        return data
-
-    @classmethod
-    def empty(cls) -> "SystemVariable":
-        return cls()
-
-    def to_dict(self) -> dict[SystemVariableKey, Any]:
-        # NOTE: This method is provided for compatibility with legacy code.
-        # New code should use the `SystemVariable` object directly instead of converting
-        # it to a dictionary, as this conversion results in the loss of type information
-        # for each key, making static analysis more difficult.
-
-        d: dict[SystemVariableKey, Any] = {
-            SystemVariableKey.FILES: self.files,
-        }
-        if self.user_id is not None:
-            d[SystemVariableKey.USER_ID] = self.user_id
-        if self.app_id is not None:
-            d[SystemVariableKey.APP_ID] = self.app_id
-        if self.workflow_id is not None:
-            d[SystemVariableKey.WORKFLOW_ID] = self.workflow_id
-        if self.workflow_execution_id is not None:
-            d[SystemVariableKey.WORKFLOW_EXECUTION_ID] = self.workflow_execution_id
-        if self.query is not None:
-            d[SystemVariableKey.QUERY] = self.query
-        if self.conversation_id is not None:
-            d[SystemVariableKey.CONVERSATION_ID] = self.conversation_id
-        if self.dialogue_count is not None:
-            d[SystemVariableKey.DIALOGUE_COUNT] = self.dialogue_count
-        return d

api/core/workflow/workflow_cycle_manager.py

@@ -26,7 +26,6 @@ from core.workflow.entities.workflow_node_execution import (
 from core.workflow.enums import SystemVariableKey
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_entry import WorkflowEntry
 from libs.datetime_utils import naive_utc_now
 
@@ -44,7 +43,7 @@ class WorkflowCycleManager:
         self,
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
-        workflow_system_variables: SystemVariable,
+        workflow_system_variables: dict[SystemVariableKey, Any],
         workflow_info: CycleManagerWorkflowInfo,
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
@@ -57,22 +56,17 @@ class WorkflowCycleManager:
 
     def handle_workflow_run_start(self) -> WorkflowExecution:
         inputs = {**self._application_generate_entity.inputs}
-
-        # Iterate over SystemVariable fields using Pydantic's model_fields
-        if self._workflow_system_variables:
-            for field_name, value in self._workflow_system_variables.to_dict().items():
-                if field_name == SystemVariableKey.CONVERSATION_ID:
-                    continue
-                inputs[f"sys.{field_name}"] = value
+        for key, value in (self._workflow_system_variables or {}).items():
+            if key.value == "conversation":
+                continue
+            inputs[f"sys.{key.value}"] = value
 
         # handle special values
         inputs = dict(WorkflowEntry.handle_special_values(inputs) or {})
 
         # init workflow run
         # TODO: This workflow_run_id should always not be None, maybe we can use a more elegant way to handle this
-        execution_id = str(
-            self._workflow_system_variables.workflow_execution_id if self._workflow_system_variables else None
-        ) or str(uuid4())
+        execution_id = str(self._workflow_system_variables.get(SystemVariableKey.WORKFLOW_EXECUTION_ID) or uuid4())
         execution = WorkflowExecution.new(
             id_=execution_id,
             workflow_id=self._workflow_info.workflow_id,

api/core/workflow/workflow_entry.py

@@ -21,7 +21,6 @@ from core.workflow.nodes import NodeType
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.event import NodeEvent
 from core.workflow.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
-from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from factories import file_factory
 from models.enums import UserFrom
@@ -255,7 +254,7 @@ class WorkflowEntry:
 
         # init variable pool
         variable_pool = VariablePool(
-            system_variables=SystemVariable.empty(),
+            system_variables={},
             user_inputs={},
             environment_variables=[],
         )

api/core/workflow/workflow_type_encoder.py

@@ -1,3 +1,4 @@
+import json
 from collections.abc import Mapping
 from typing import Any
 
@@ -7,6 +8,18 @@ from core.file.models import File
 from core.variables import Segment
 
 
+class WorkflowRuntimeTypeEncoder(json.JSONEncoder):
+    def default(self, o: Any):
+        if isinstance(o, Segment):
+            return o.value
+        elif isinstance(o, File):
+            return o.to_dict()
+        elif isinstance(o, BaseModel):
+            return o.model_dump(mode="json")
+        else:
+            return super().default(o)
+
+
 class WorkflowRuntimeTypeConverter:
     def to_json_encodable(self, value: Mapping[str, Any] | None) -> Mapping[str, Any] | None:
         result = self._to_json_encodable_recursive(value)

api/extensions/ext_otel.py

@@ -193,22 +193,13 @@ def init_app(app: DifyApp):
                 insecure=True,
             )
         else:
-            headers = {"Authorization": f"Bearer {dify_config.OTLP_API_KEY}"} if dify_config.OTLP_API_KEY else None
-
-            trace_endpoint = dify_config.OTLP_TRACE_ENDPOINT
-            if not trace_endpoint:
-                trace_endpoint = dify_config.OTLP_BASE_ENDPOINT + "/v1/traces"
             exporter = HTTPSpanExporter(
-                endpoint=trace_endpoint,
-                headers=headers,
+                endpoint=dify_config.OTLP_BASE_ENDPOINT + "/v1/traces",
+                headers={"Authorization": f"Bearer {dify_config.OTLP_API_KEY}"},
             )
-
-            metric_endpoint = dify_config.OTLP_METRIC_ENDPOINT
-            if not metric_endpoint:
-                metric_endpoint = dify_config.OTLP_BASE_ENDPOINT + "/v1/metrics"
             metric_exporter = HTTPMetricExporter(
-                endpoint=metric_endpoint,
-                headers=headers,
+                endpoint=dify_config.OTLP_BASE_ENDPOINT + "/v1/metrics",
+                headers={"Authorization": f"Bearer {dify_config.OTLP_API_KEY}"},
             )
     else:
         exporter = ConsoleSpanExporter()

api/factories/variable_factory.py

@@ -91,13 +91,9 @@ def _build_variable_from_mapping(*, mapping: Mapping[str, Any], selector: Sequen
             result = StringVariable.model_validate(mapping)
         case SegmentType.SECRET:
             result = SecretVariable.model_validate(mapping)
-        case SegmentType.NUMBER | SegmentType.INTEGER if isinstance(value, int):
-            mapping = dict(mapping)
-            mapping["value_type"] = SegmentType.INTEGER
+        case SegmentType.NUMBER if isinstance(value, int):
             result = IntegerVariable.model_validate(mapping)
-        case SegmentType.NUMBER | SegmentType.FLOAT if isinstance(value, float):
-            mapping = dict(mapping)
-            mapping["value_type"] = SegmentType.FLOAT
+        case SegmentType.NUMBER if isinstance(value, float):
             result = FloatVariable.model_validate(mapping)
         case SegmentType.NUMBER if not isinstance(value, float | int):
             raise VariableError(f"invalid number value {value}")
@@ -123,8 +119,6 @@ def infer_segment_type_from_value(value: Any, /) -> SegmentType:
 
 
 def build_segment(value: Any, /) -> Segment:
-    # NOTE: If you have runtime type information available, consider using the `build_segment_with_type`
-    # below
     if value is None:
         return NoneSegment()
     if isinstance(value, str):
@@ -140,17 +134,12 @@ def build_segment(value: Any, /) -> Segment:
     if isinstance(value, list):
         items = [build_segment(item) for item in value]
         types = {item.value_type for item in items}
-        if all(isinstance(item, ArraySegment) for item in items):
+        if len(types) != 1 or all(isinstance(item, ArraySegment) for item in items):
             return ArrayAnySegment(value=value)
-        elif len(types) != 1:
-            if types.issubset({SegmentType.NUMBER, SegmentType.INTEGER, SegmentType.FLOAT}):
-                return ArrayNumberSegment(value=value)
-            return ArrayAnySegment(value=value)
-
         match types.pop():
             case SegmentType.STRING:
                 return ArrayStringSegment(value=value)
-            case SegmentType.NUMBER | SegmentType.INTEGER | SegmentType.FLOAT:
+            case SegmentType.NUMBER:
                 return ArrayNumberSegment(value=value)
             case SegmentType.OBJECT:
                 return ArrayObjectSegment(value=value)
@@ -164,22 +153,6 @@ def build_segment(value: Any, /) -> Segment:
     raise ValueError(f"not supported value {value}")
 
 
-_segment_factory: Mapping[SegmentType, type[Segment]] = {
-    SegmentType.NONE: NoneSegment,
-    SegmentType.STRING: StringSegment,
-    SegmentType.INTEGER: IntegerSegment,
-    SegmentType.FLOAT: FloatSegment,
-    SegmentType.FILE: FileSegment,
-    SegmentType.OBJECT: ObjectSegment,
-    # Array types
-    SegmentType.ARRAY_ANY: ArrayAnySegment,
-    SegmentType.ARRAY_STRING: ArrayStringSegment,
-    SegmentType.ARRAY_NUMBER: ArrayNumberSegment,
-    SegmentType.ARRAY_OBJECT: ArrayObjectSegment,
-    SegmentType.ARRAY_FILE: ArrayFileSegment,
-}
-
-
 def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
     """
     Build a segment with explicit type checking.
@@ -217,7 +190,7 @@ def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
         if segment_type == SegmentType.NONE:
             return NoneSegment()
         else:
-            raise TypeMismatchError(f"Type mismatch: expected {segment_type}, but got None")
+            raise TypeMismatchError(f"Expected {segment_type}, but got None")
 
     # Handle empty list special case for array types
     if isinstance(value, list) and len(value) == 0:
@@ -232,25 +205,21 @@ def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
         elif segment_type == SegmentType.ARRAY_FILE:
             return ArrayFileSegment(value=value)
         else:
-            raise TypeMismatchError(f"Type mismatch: expected {segment_type}, but got empty list")
+            raise TypeMismatchError(f"Expected {segment_type}, but got empty list")
+
+    # Build segment using existing logic to infer actual type
+    inferred_segment = build_segment(value)
+    inferred_type = inferred_segment.value_type
 
-    inferred_type = SegmentType.infer_segment_type(value)
     # Type compatibility checking
-    if inferred_type is None:
-        raise TypeMismatchError(
-            f"Type mismatch: expected {segment_type}, but got python object, type={type(value)}, value={value}"
-        )
     if inferred_type == segment_type:
-        segment_class = _segment_factory[segment_type]
-        return segment_class(value_type=segment_type, value=value)
-    elif segment_type == SegmentType.NUMBER and inferred_type in (
-        SegmentType.INTEGER,
-        SegmentType.FLOAT,
-    ):
-        segment_class = _segment_factory[inferred_type]
-        return segment_class(value_type=inferred_type, value=value)
-    else:
-        raise TypeMismatchError(f"Type mismatch: expected {segment_type}, but got {inferred_type}, value={value}")
+        return inferred_segment
+
+    # Type mismatch - raise error with descriptive message
+    raise TypeMismatchError(
+        f"Type mismatch: expected {segment_type}, but value '{value}' "
+        f"(type: {type(value).__name__}) corresponds to {inferred_type}"
+    )
 
 
 def segment_to_variable(
@@ -278,6 +247,6 @@ def segment_to_variable(
             name=name,
             description=description,
             value=segment.value,
-            selector=list(selector),
+            selector=selector,
         ),
     )

api/fields/_value_type_serializer.py

@@ -1,15 +0,0 @@
-from typing import TypedDict
-
-from core.variables.segments import Segment
-from core.variables.types import SegmentType
-
-
-class _VarTypedDict(TypedDict, total=False):
-    value_type: SegmentType
-
-
-def serialize_value_type(v: _VarTypedDict | Segment) -> str:
-    if isinstance(v, Segment):
-        return v.value_type.exposed_type().value
-    else:
-        return v["value_type"].exposed_type().value

api/fields/app_fields.py

@@ -188,7 +188,6 @@ app_detail_fields_with_site = {
     "site": fields.Nested(site_fields),
     "api_base_url": fields.String,
     "use_icon_as_answer_icon": fields.Boolean,
-    "max_active_requests": fields.Integer,
     "created_by": fields.String,
     "created_at": TimestampField,
     "updated_by": fields.String,

api/fields/conversation_variable_fields.py

@@ -2,12 +2,10 @@ from flask_restful import fields
 
 from libs.helper import TimestampField
 
-from ._value_type_serializer import serialize_value_type
-
 conversation_variable_fields = {
     "id": fields.String,
     "name": fields.String,
-    "value_type": fields.String(attribute=serialize_value_type),
+    "value_type": fields.String(attribute="value_type.value"),
     "value": fields.String,
     "description": fields.String,
     "created_at": TimestampField,

api/fields/workflow_fields.py

@@ -5,8 +5,6 @@ from core.variables import SecretVariable, SegmentType, Variable
 from fields.member_fields import simple_account_fields
 from libs.helper import TimestampField
 
-from ._value_type_serializer import serialize_value_type
-
 ENVIRONMENT_VARIABLE_SUPPORTED_TYPES = (SegmentType.STRING, SegmentType.NUMBER, SegmentType.SECRET)
 
 
@@ -26,16 +24,11 @@ class EnvironmentVariableField(fields.Raw):
                 "id": value.id,
                 "name": value.name,
                 "value": value.value,
-                "value_type": value.value_type.exposed_type().value,
+                "value_type": value.value_type.value,
                 "description": value.description,
             }
         if isinstance(value, dict):
-            value_type_str = value.get("value_type")
-            if not isinstance(value_type_str, str):
-                raise TypeError(
-                    f"unexpected type for value_type field, value={value_type_str}, type={type(value_type_str)}"
-                )
-            value_type = SegmentType(value_type_str).exposed_type()
+            value_type = value.get("value_type")
             if value_type not in ENVIRONMENT_VARIABLE_SUPPORTED_TYPES:
                 raise ValueError(f"Unsupported environment variable value type: {value_type}")
             return value
@@ -44,7 +37,7 @@ class EnvironmentVariableField(fields.Raw):
 conversation_variable_fields = {
     "id": fields.String,
     "name": fields.String,
-    "value_type": fields.String(attribute=serialize_value_type),
+    "value_type": fields.String(attribute="value_type.value"),
     "value": fields.Raw,
     "description": fields.String,
 }

api/libs/helper.py

@@ -148,6 +148,25 @@ class StrLen:
         return value
 
 
+class FloatRange:
+    """Restrict input to an float in a range (inclusive)"""
+
+    def __init__(self, low, high, argument="argument"):
+        self.low = low
+        self.high = high
+        self.argument = argument
+
+    def __call__(self, value):
+        value = _get_float(value)
+        if value < self.low or value > self.high:
+            error = "Invalid {arg}: {val}. {arg} must be within the range {lo} - {hi}".format(
+                arg=self.argument, val=value, lo=self.low, hi=self.high
+            )
+            raise ValueError(error)
+
+        return value
+
+
 class DatetimeString:
     def __init__(self, format, argument="argument"):
         self.format = format

api/libs/jsonutil.py

@@ -0,0 +1,11 @@
+import json
+
+from pydantic import BaseModel
+
+
+class PydanticModelEncoder(json.JSONEncoder):
+    def default(self, o):
+        if isinstance(o, BaseModel):
+            return o.model_dump()
+        else:
+            super().default(o)

api/libs/uuid_utils.py

@@ -1,164 +0,0 @@
-import secrets
-import struct
-import time
-import uuid
-
-# Reference for UUIDv7 specification:
-# RFC 9562, Section 5.7 - https://www.rfc-editor.org/rfc/rfc9562.html#section-5.7
-
-# Define the format for packing the timestamp as an unsigned 64-bit integer (big-endian).
-#
-# For details on the `struct.pack` format, refer to:
-# https://docs.python.org/3/library/struct.html#byte-order-size-and-alignment
-_PACK_TIMESTAMP = ">Q"
-
-# Define the format for packing the 12-bit random data A (as specified in RFC 9562 Section 5.7)
-# into an unsigned 16-bit integer (big-endian).
-_PACK_RAND_A = ">H"
-
-
-def _create_uuidv7_bytes(timestamp_ms: int, random_bytes: bytes) -> bytes:
-    """Create UUIDv7 byte structure with given timestamp and random bytes.
-
-    This is a private helper function that handles the common logic for creating
-    UUIDv7 byte structure according to RFC 9562 specification.
-
-    UUIDv7 Structure:
-    - 48 bits: timestamp (milliseconds since Unix epoch)
-    - 12 bits: random data A (with version bits)
-    - 62 bits: random data B (with variant bits)
-
-    The function performs the following operations:
-    1. Creates a 128-bit (16-byte) UUID structure
-    2. Packs the timestamp into the first 48 bits (6 bytes)
-    3. Sets the version bits to 7 (0111) in the correct position
-    4. Sets the variant bits to 10 (binary) in the correct position
-    5. Fills the remaining bits with the provided random bytes
-
-    Args:
-        timestamp_ms: The timestamp in milliseconds since Unix epoch (48 bits).
-        random_bytes: Random bytes to use for the random portions (must be 10 bytes).
-                     First 2 bytes are used for random data A (12 bits after version).
-                     Last 8 bytes are used for random data B (62 bits after variant).
-
-    Returns:
-        A 16-byte bytes object representing the complete UUIDv7 structure.
-
-    Note:
-        This function assumes the random_bytes parameter is exactly 10 bytes.
-        The caller is responsible for providing appropriate random data.
-    """
-    # Create the 128-bit UUID structure
-    uuid_bytes = bytearray(16)
-
-    # Pack timestamp (48 bits) into first 6 bytes
-    uuid_bytes[0:6] = struct.pack(_PACK_TIMESTAMP, timestamp_ms)[2:8]  # Take last 6 bytes of 8-byte big-endian
-
-    # Next 16 bits: random data A (12 bits) + version (4 bits)
-    # Take first 2 random bytes and set version to 7
-    rand_a = struct.unpack(_PACK_RAND_A, random_bytes[0:2])[0]
-    # Clear the highest 4 bits to make room for the version field
-    # by performing a bitwise AND with 0x0FFF (binary: 0b0000_1111_1111_1111).
-    rand_a = rand_a & 0x0FFF
-    # Set the version field to 7 (binary: 0111) by performing a bitwise OR with 0x7000 (binary: 0b0111_0000_0000_0000).
-    rand_a = rand_a | 0x7000
-    uuid_bytes[6:8] = struct.pack(_PACK_RAND_A, rand_a)
-
-    # Last 64 bits: random data B (62 bits) + variant (2 bits)
-    # Use remaining 8 random bytes and set variant to 10 (binary)
-    uuid_bytes[8:16] = random_bytes[2:10]
-
-    # Set variant bits (first 2 bits of byte 8 should be '10')
-    uuid_bytes[8] = (uuid_bytes[8] & 0x3F) | 0x80  # Set variant to 10xxxxxx
-
-    return bytes(uuid_bytes)
-
-
-def uuidv7(timestamp_ms: int | None = None) -> uuid.UUID:
-    """Generate a UUID version 7 according to RFC 9562 specification.
-
-    UUIDv7 features a time-ordered value field derived from the widely
-    implemented and well known Unix Epoch timestamp source, the number of
-    milliseconds since midnight 1 Jan 1970 UTC, leap seconds excluded.
-
-    Structure:
-    - 48 bits: timestamp (milliseconds since Unix epoch)
-    - 12 bits: random data A (with version bits)
-    - 62 bits: random data B (with variant bits)
-
-    Args:
-        timestamp_ms: The timestamp used when generating UUID, use the current time if unspecified.
-                  Should be an integer representing milliseconds since Unix epoch.
-
-    Returns:
-        A UUID object representing a UUIDv7.
-
-    Example:
-        >>> import time
-        >>> # Generate UUIDv7 with current time
-        >>> uuid_current = uuidv7()
-        >>> # Generate UUIDv7 with specific timestamp
-        >>> uuid_specific = uuidv7(int(time.time() * 1000))
-    """
-    if timestamp_ms is None:
-        timestamp_ms = int(time.time() * 1000)
-
-    # Generate 10 random bytes for the random portions
-    random_bytes = secrets.token_bytes(10)
-
-    # Create UUIDv7 bytes using the helper function
-    uuid_bytes = _create_uuidv7_bytes(timestamp_ms, random_bytes)
-
-    return uuid.UUID(bytes=uuid_bytes)
-
-
-def uuidv7_timestamp(id_: uuid.UUID) -> int:
-    """Extract the timestamp from a UUIDv7.
-
-    UUIDv7 contains a 48-bit timestamp field representing milliseconds since
-    the Unix epoch (1970-01-01 00:00:00 UTC). This function extracts and
-    returns that timestamp as an integer representing milliseconds since the epoch.
-
-    Args:
-        id_: A UUID object that should be a UUIDv7 (version 7).
-
-    Returns:
-        The timestamp as an integer representing milliseconds since Unix epoch.
-
-    Raises:
-        ValueError: If the provided UUID is not version 7.
-
-    Example:
-        >>> uuid_v7 = uuidv7()
-        >>> timestamp = uuidv7_timestamp(uuid_v7)
-        >>> print(f"UUID was created at: {timestamp} ms")
-    """
-    # Verify this is a UUIDv7
-    if id_.version != 7:
-        raise ValueError(f"Expected UUIDv7 (version 7), got version {id_.version}")
-
-    # Extract the UUID bytes
-    uuid_bytes = id_.bytes
-
-    # Extract the first 48 bits (6 bytes) as the timestamp in milliseconds
-    # Pad with 2 zero bytes at the beginning to make it 8 bytes for unpacking as Q (unsigned long long)
-    timestamp_bytes = b"\x00\x00" + uuid_bytes[0:6]
-    ts_in_ms = struct.unpack(_PACK_TIMESTAMP, timestamp_bytes)[0]
-
-    # Return timestamp directly in milliseconds as integer
-    assert isinstance(ts_in_ms, int)
-    return ts_in_ms
-
-
-def uuidv7_boundary(timestamp_ms: int) -> uuid.UUID:
-    """Generate a non-random uuidv7 with the given timestamp (first 48 bits) and
-    all random bits to 0. As the smallest possible uuidv7 for that timestamp,
-    it may be used as a boundary for partitions.
-    """
-    # Use zero bytes for all random portions
-    zero_random_bytes = b"\x00" * 10
-
-    # Create UUIDv7 bytes using the helper function
-    uuid_bytes = _create_uuidv7_bytes(timestamp_ms, zero_random_bytes)
-
-    return uuid.UUID(bytes=uuid_bytes)

api/migrations/versions/2025_07_02_2332-1c9ba48be8e4_add_uuidv7_function_in_sql.py

@@ -1,86 +0,0 @@
-"""add uuidv7 function in SQL
-
-Revision ID: 1c9ba48be8e4
-Revises: 58eb7bdb93fe
-Create Date: 2025-07-02 23:32:38.484499
-
-"""
-
-"""
-The functions in this files comes from https://github.com/dverite/postgres-uuidv7-sql/, with minor modifications.
-
-LICENSE:
-
-# Copyright and License
-
-Copyright (c) 2024, Daniel Vérité
-
-Permission to use, copy, modify, and distribute this software and its documentation for any purpose, without fee, and without a written agreement is hereby granted, provided that the above copyright notice and this paragraph and the following two paragraphs appear in all copies.
-
-In no event shall Daniel Vérité be liable to any party for direct, indirect, special, incidental, or consequential damages, including lost profits, arising out of the use of this software and its documentation, even if Daniel Vérité has been advised of the possibility of such damage.
-
-Daniel Vérité specifically disclaims any warranties, including, but not limited to, the implied warranties of merchantability and fitness for a particular purpose. The software provided hereunder is on an "AS IS" basis, and Daniel Vérité has no obligations to provide maintenance, support, updates, enhancements, or modifications.
-"""
-
-from alembic import op
-import models as models
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '1c9ba48be8e4'
-down_revision = '58eb7bdb93fe'
-branch_labels: None = None
-depends_on: None = None
-
-
-def upgrade():
-    # This implementation differs slightly from the original uuidv7 function in
-    # https://github.com/dverite/postgres-uuidv7-sql/.
-    # The ability to specify source timestamp has been removed because its type signature is incompatible with
-    # PostgreSQL 18's `uuidv7` function. This capability is rarely needed in practice, as IDs can be
-    # generated and controlled within the application layer.
-    op.execute(sa.text(r"""
-/* Main function to generate a uuidv7 value with millisecond precision */
-CREATE FUNCTION uuidv7() RETURNS uuid
-AS
-$$
-    -- Replace the first 48 bits of a uuidv4 with the current
-    -- number of milliseconds since 1970-01-01 UTC
-    -- and set the "ver" field to 7 by setting additional bits
-SELECT encode(
-               set_bit(
-                       set_bit(
-                               overlay(uuid_send(gen_random_uuid()) placing
-                                       substring(int8send((extract(epoch from clock_timestamp()) * 1000)::bigint) from
-                                                 3)
-                                       from 1 for 6),
-                               52, 1),
-                       53, 1), 'hex')::uuid;
-$$ LANGUAGE SQL VOLATILE PARALLEL SAFE;
-
-COMMENT ON FUNCTION uuidv7 IS
-    'Generate a uuid-v7 value with a 48-bit timestamp (millisecond precision) and 74 bits of randomness';
-"""))
-
-    op.execute(sa.text(r"""
-CREATE FUNCTION uuidv7_boundary(timestamptz) RETURNS uuid
-AS
-$$
-    /* uuid fields: version=0b0111, variant=0b10 */
-SELECT encode(
-               overlay('\x00000000000070008000000000000000'::bytea
-                       placing substring(int8send(floor(extract(epoch from $1) * 1000)::bigint) from 3)
-                       from 1 for 6),
-               'hex')::uuid;
-$$ LANGUAGE SQL STABLE STRICT PARALLEL SAFE;
-
-COMMENT ON FUNCTION uuidv7_boundary(timestamptz) IS
-    'Generate a non-random uuidv7 with the given timestamp (first 48 bits) and all random bits to 0. As the smallest possible uuidv7 for that timestamp, it may be used as a boundary for partitions.';
-"""
-))
-
-
-def downgrade():
-    op.execute(sa.text("DROP FUNCTION uuidv7"))
-    op.execute(sa.text("DROP FUNCTION uuidv7_boundary"))

api/migrations/versions/2025_07_04_1705-71f5020c6470_tool_oauth.py

@@ -12,7 +12,7 @@ import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
 revision = '71f5020c6470'
-down_revision = '1c9ba48be8e4'
+down_revision = '58eb7bdb93fe'
 branch_labels = None
 depends_on = None
 

api/models/model.py

@@ -610,6 +610,14 @@ class InstalledApp(Base):
         return tenant
 
 
+class ConversationSource(StrEnum):
+    """This enumeration is designed for use with `Conversation.from_source`."""
+
+    # NOTE(QuantumGhost): The enumeration members may not cover all possible cases.
+    API = "api"
+    CONSOLE = "console"
+
+
 class Conversation(Base):
     __tablename__ = "conversations"
     __table_args__ = (

api/models/tools.py

@@ -331,13 +331,11 @@ class MCPToolProvider(Base):
 
         provider_controller = MCPToolProviderController._from_db(self)
 
-        encrypter, _ = create_provider_encrypter(
+        return create_provider_encrypter(
             tenant_id=self.tenant_id,
             config=[x.to_basic_provider_config() for x in provider_controller.get_credentials_schema()],
             cache=NoOpProviderCredentialCache(),
-        )
-
-        return encrypter.decrypt(self.credentials)  # type: ignore
+        )[0].decrypt(self.credentials)
 
 
 class ToolModelInvoke(Base):

api/models/workflow.py

@@ -12,7 +12,6 @@ from sqlalchemy import orm
 from core.file.constants import maybe_file_object
 from core.file.models import File
 from core.variables import utils as variable_utils
-from core.variables.variables import FloatVariable, IntegerVariable, StringVariable
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from core.workflow.nodes.enums import NodeType
 from factories.variable_factory import TypeMismatchError, build_segment_with_type
@@ -348,7 +347,7 @@ class Workflow(Base):
         )
 
     @property
-    def environment_variables(self) -> Sequence[StringVariable | IntegerVariable | FloatVariable | SecretVariable]:
+    def environment_variables(self) -> Sequence[Variable]:
         # TODO: find some way to init `self._environment_variables` when instance created.
         if self._environment_variables is None:
             self._environment_variables = "{}"
@@ -368,15 +367,11 @@ class Workflow(Base):
         def decrypt_func(var):
             if isinstance(var, SecretVariable):
                 return var.model_copy(update={"value": encrypter.decrypt_token(tenant_id=tenant_id, token=var.value)})
-            elif isinstance(var, (StringVariable, IntegerVariable, FloatVariable)):
-                return var
             else:
-                raise AssertionError("this statement should be unreachable.")
+                return var
 
-        decrypted_results: list[SecretVariable | StringVariable | IntegerVariable | FloatVariable] = list(
-            map(decrypt_func, results)
-        )
-        return decrypted_results
+        results = list(map(decrypt_func, results))
+        return results
 
     @environment_variables.setter
     def environment_variables(self, value: Sequence[Variable]):

api/services/account_service.py

@@ -52,14 +52,8 @@ from services.errors.workspace import WorkSpaceNotAllowedCreateError, Workspaces
 from services.feature_service import FeatureService
 from tasks.delete_account_task import delete_account_task
 from tasks.mail_account_deletion_task import send_account_deletion_verification_code
-from tasks.mail_change_mail_task import send_change_mail_task
 from tasks.mail_email_code_login import send_email_code_login_mail_task
 from tasks.mail_invite_member_task import send_invite_member_mail_task
-from tasks.mail_owner_transfer_task import (
-    send_new_owner_transfer_notify_email_task,
-    send_old_owner_transfer_notify_email_task,
-    send_owner_transfer_confirm_task,
-)
 from tasks.mail_reset_password_task import send_reset_password_mail_task
 
 
@@ -81,13 +75,8 @@ class AccountService:
     email_code_account_deletion_rate_limiter = RateLimiter(
         prefix="email_code_account_deletion_rate_limit", max_attempts=1, time_window=60 * 1
     )
-    change_email_rate_limiter = RateLimiter(prefix="change_email_rate_limit", max_attempts=1, time_window=60 * 1)
-    owner_transfer_rate_limiter = RateLimiter(prefix="owner_transfer_rate_limit", max_attempts=1, time_window=60 * 1)
-
     LOGIN_MAX_ERROR_LIMITS = 5
     FORGOT_PASSWORD_MAX_ERROR_LIMITS = 5
-    CHANGE_EMAIL_MAX_ERROR_LIMITS = 5
-    OWNER_TRANSFER_MAX_ERROR_LIMITS = 5
 
     @staticmethod
     def _get_refresh_token_key(refresh_token: str) -> str:
@@ -430,101 +419,6 @@ class AccountService:
         cls.reset_password_rate_limiter.increment_rate_limit(account_email)
         return token
 
-    @classmethod
-    def send_change_email_email(
-        cls,
-        account: Optional[Account] = None,
-        email: Optional[str] = None,
-        old_email: Optional[str] = None,
-        language: Optional[str] = "en-US",
-        phase: Optional[str] = None,
-    ):
-        account_email = account.email if account else email
-        if account_email is None:
-            raise ValueError("Email must be provided.")
-
-        if cls.change_email_rate_limiter.is_rate_limited(account_email):
-            from controllers.console.auth.error import EmailChangeRateLimitExceededError
-
-            raise EmailChangeRateLimitExceededError()
-
-        code, token = cls.generate_change_email_token(account_email, account, old_email=old_email)
-
-        send_change_mail_task.delay(
-            language=language,
-            to=account_email,
-            code=code,
-            phase=phase,
-        )
-        cls.change_email_rate_limiter.increment_rate_limit(account_email)
-        return token
-
-    @classmethod
-    def send_owner_transfer_email(
-        cls,
-        account: Optional[Account] = None,
-        email: Optional[str] = None,
-        language: Optional[str] = "en-US",
-        workspace_name: Optional[str] = "",
-    ):
-        account_email = account.email if account else email
-        if account_email is None:
-            raise ValueError("Email must be provided.")
-
-        if cls.owner_transfer_rate_limiter.is_rate_limited(account_email):
-            from controllers.console.auth.error import OwnerTransferRateLimitExceededError
-
-            raise OwnerTransferRateLimitExceededError()
-
-        code, token = cls.generate_owner_transfer_token(account_email, account)
-
-        send_owner_transfer_confirm_task.delay(
-            language=language,
-            to=account_email,
-            code=code,
-            workspace=workspace_name,
-        )
-        cls.owner_transfer_rate_limiter.increment_rate_limit(account_email)
-        return token
-
-    @classmethod
-    def send_old_owner_transfer_notify_email(
-        cls,
-        account: Optional[Account] = None,
-        email: Optional[str] = None,
-        language: Optional[str] = "en-US",
-        workspace_name: Optional[str] = "",
-        new_owner_email: Optional[str] = "",
-    ):
-        account_email = account.email if account else email
-        if account_email is None:
-            raise ValueError("Email must be provided.")
-
-        send_old_owner_transfer_notify_email_task.delay(
-            language=language,
-            to=account_email,
-            workspace=workspace_name,
-            new_owner_email=new_owner_email,
-        )
-
-    @classmethod
-    def send_new_owner_transfer_notify_email(
-        cls,
-        account: Optional[Account] = None,
-        email: Optional[str] = None,
-        language: Optional[str] = "en-US",
-        workspace_name: Optional[str] = "",
-    ):
-        account_email = account.email if account else email
-        if account_email is None:
-            raise ValueError("Email must be provided.")
-
-        send_new_owner_transfer_notify_email_task.delay(
-            language=language,
-            to=account_email,
-            workspace=workspace_name,
-        )
-
     @classmethod
     def generate_reset_password_token(
         cls,
@@ -541,64 +435,14 @@ class AccountService:
         )
         return code, token
 
-    @classmethod
-    def generate_change_email_token(
-        cls,
-        email: str,
-        account: Optional[Account] = None,
-        code: Optional[str] = None,
-        old_email: Optional[str] = None,
-        additional_data: dict[str, Any] = {},
-    ):
-        if not code:
-            code = "".join([str(secrets.randbelow(exclusive_upper_bound=10)) for _ in range(6)])
-        additional_data["code"] = code
-        additional_data["old_email"] = old_email
-        token = TokenManager.generate_token(
-            account=account, email=email, token_type="change_email", additional_data=additional_data
-        )
-        return code, token
-
-    @classmethod
-    def generate_owner_transfer_token(
-        cls,
-        email: str,
-        account: Optional[Account] = None,
-        code: Optional[str] = None,
-        additional_data: dict[str, Any] = {},
-    ):
-        if not code:
-            code = "".join([str(secrets.randbelow(exclusive_upper_bound=10)) for _ in range(6)])
-        additional_data["code"] = code
-        token = TokenManager.generate_token(
-            account=account, email=email, token_type="owner_transfer", additional_data=additional_data
-        )
-        return code, token
-
     @classmethod
     def revoke_reset_password_token(cls, token: str):
         TokenManager.revoke_token(token, "reset_password")
 
-    @classmethod
-    def revoke_change_email_token(cls, token: str):
-        TokenManager.revoke_token(token, "change_email")
-
-    @classmethod
-    def revoke_owner_transfer_token(cls, token: str):
-        TokenManager.revoke_token(token, "owner_transfer")
-
     @classmethod
     def get_reset_password_data(cls, token: str) -> Optional[dict[str, Any]]:
         return TokenManager.get_token_data(token, "reset_password")
 
-    @classmethod
-    def get_change_email_data(cls, token: str) -> Optional[dict[str, Any]]:
-        return TokenManager.get_token_data(token, "change_email")
-
-    @classmethod
-    def get_owner_transfer_data(cls, token: str) -> Optional[dict[str, Any]]:
-        return TokenManager.get_token_data(token, "owner_transfer")
-
     @classmethod
     def send_email_code_login_email(
         cls, account: Optional[Account] = None, email: Optional[str] = None, language: Optional[str] = "en-US"
@@ -708,62 +552,6 @@ class AccountService:
         key = f"forgot_password_error_rate_limit:{email}"
         redis_client.delete(key)
 
-    @staticmethod
-    @redis_fallback(default_return=None)
-    def add_change_email_error_rate_limit(email: str) -> None:
-        key = f"change_email_error_rate_limit:{email}"
-        count = redis_client.get(key)
-        if count is None:
-            count = 0
-        count = int(count) + 1
-        redis_client.setex(key, dify_config.CHANGE_EMAIL_LOCKOUT_DURATION, count)
-
-    @staticmethod
-    @redis_fallback(default_return=False)
-    def is_change_email_error_rate_limit(email: str) -> bool:
-        key = f"change_email_error_rate_limit:{email}"
-        count = redis_client.get(key)
-        if count is None:
-            return False
-        count = int(count)
-        if count > AccountService.CHANGE_EMAIL_MAX_ERROR_LIMITS:
-            return True
-        return False
-
-    @staticmethod
-    @redis_fallback(default_return=None)
-    def reset_change_email_error_rate_limit(email: str):
-        key = f"change_email_error_rate_limit:{email}"
-        redis_client.delete(key)
-
-    @staticmethod
-    @redis_fallback(default_return=None)
-    def add_owner_transfer_error_rate_limit(email: str) -> None:
-        key = f"owner_transfer_error_rate_limit:{email}"
-        count = redis_client.get(key)
-        if count is None:
-            count = 0
-        count = int(count) + 1
-        redis_client.setex(key, dify_config.OWNER_TRANSFER_LOCKOUT_DURATION, count)
-
-    @staticmethod
-    @redis_fallback(default_return=False)
-    def is_owner_transfer_error_rate_limit(email: str) -> bool:
-        key = f"owner_transfer_error_rate_limit:{email}"
-        count = redis_client.get(key)
-        if count is None:
-            return False
-        count = int(count)
-        if count > AccountService.OWNER_TRANSFER_MAX_ERROR_LIMITS:
-            return True
-        return False
-
-    @staticmethod
-    @redis_fallback(default_return=None)
-    def reset_owner_transfer_error_rate_limit(email: str):
-        key = f"owner_transfer_error_rate_limit:{email}"
-        redis_client.delete(key)
-
     @staticmethod
     @redis_fallback(default_return=False)
     def is_email_send_ip_limit(ip_address: str):
@@ -805,10 +593,6 @@ class AccountService:
 
         return False
 
-    @staticmethod
-    def check_email_unique(email: str) -> bool:
-        return db.session.query(Account).filter_by(email=email).first() is None
-
 
 class TenantService:
     @staticmethod
@@ -1081,15 +865,6 @@ class TenantService:
 
         return cast(dict, tenant.custom_config_dict)
 
-    @staticmethod
-    def is_owner(account: Account, tenant: Tenant) -> bool:
-        return TenantService.get_user_role(account, tenant) == TenantAccountRole.OWNER
-
-    @staticmethod
-    def is_member(account: Account, tenant: Tenant) -> bool:
-        """Check if the account is a member of the tenant"""
-        return TenantService.get_user_role(account, tenant) is not None
-
 
 class RegisterService:
     @classmethod

api/services/app_service.py

@@ -233,7 +233,6 @@ class AppService:
         app.icon = args.get("icon")
         app.icon_background = args.get("icon_background")
         app.use_icon_as_answer_icon = args.get("use_icon_as_answer_icon", False)
-        app.max_active_requests = args.get("max_active_requests")
         app.updated_by = current_user.id
         app.updated_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()

api/services/entities/knowledge_entities/knowledge_entities.py

@@ -4,6 +4,13 @@ from typing import Literal, Optional
 from pydantic import BaseModel
 
 
+class SegmentUpdateEntity(BaseModel):
+    content: str
+    answer: Optional[str] = None
+    keywords: Optional[list[str]] = None
+    enabled: Optional[bool] = None
+
+
 class ParentMode(StrEnum):
     FULL_DOC = "full-doc"
     PARAGRAPH = "paragraph"
@@ -146,6 +153,10 @@ class MetadataUpdateArgs(BaseModel):
     value: Optional[str | int | float] = None
 
 
+class MetadataValueUpdateArgs(BaseModel):
+    fields: list[MetadataUpdateArgs]
+
+
 class MetadataDetail(BaseModel):
     id: str
     name: str

api/services/feature_service.py

@@ -123,7 +123,7 @@ class FeatureModel(BaseModel):
     dataset_operator_enabled: bool = False
     webapp_copyright_enabled: bool = False
     workspace_members: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
-    is_allow_transfer_workspace: bool = True
+
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())
 
@@ -149,7 +149,6 @@ class SystemFeatureModel(BaseModel):
     branding: BrandingModel = BrandingModel()
     webapp_auth: WebAppAuthModel = WebAppAuthModel()
     plugin_installation_permission: PluginInstallationPermissionModel = PluginInstallationPermissionModel()
-    enable_change_email: bool = True
 
 
 class FeatureService:
@@ -187,7 +186,6 @@ class FeatureService:
         if dify_config.ENTERPRISE_ENABLED:
             system_features.branding.enabled = True
             system_features.webapp_auth.enabled = True
-            system_features.enable_change_email = False
             cls._fulfill_params_from_enterprise(system_features)
 
         if dify_config.MARKETPLACE_ENABLED:
@@ -230,8 +228,6 @@ class FeatureService:
 
         if features.billing.subscription.plan != "sandbox":
             features.webapp_copyright_enabled = True
-        else:
-            features.is_allow_transfer_workspace = False
 
         if "members" in billing_info:
             features.members.size = billing_info["members"]["size"]

api/services/tools/builtin_tools_manage_service.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import re
-from collections.abc import Mapping
 from pathlib import Path
 from typing import Any, Optional
 
@@ -476,7 +475,7 @@ class BuiltinToolManageService:
             return user_client is not None and user_client.enabled
 
     @staticmethod
-    def get_oauth_client(tenant_id: str, provider: str) -> Mapping[str, Any] | None:
+    def get_oauth_client(tenant_id: str, provider: str) -> dict[str, Any] | None:
         """
         get builtin tool provider
         """
@@ -498,7 +497,7 @@ class BuiltinToolManageService:
                 )
                 .first()
             )
-            oauth_params: Mapping[str, Any] | None = None
+            oauth_params: dict[str, Any] | None = None
             if user_client:
                 oauth_params = encrypter.decrypt(user_client.oauth_params)
                 return oauth_params

api/services/workflow_service.py

@@ -3,7 +3,7 @@ import time
 import uuid
 from collections.abc import Callable, Generator, Mapping, Sequence
 from datetime import UTC, datetime
-from typing import Any, Optional, cast
+from typing import Any, Optional
 from uuid import uuid4
 
 from sqlalchemy import select
@@ -15,10 +15,10 @@ from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.file import File
 from core.repositories import DifyCoreRepositoryFactory
 from core.variables import Variable
-from core.variables.variables import VariableUnion
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
+from core.workflow.enums import SystemVariableKey
 from core.workflow.errors import WorkflowNodeRunFailedError
 from core.workflow.graph_engine.entities.event import InNodeEvent
 from core.workflow.nodes import NodeType
@@ -28,7 +28,6 @@ from core.workflow.nodes.event import RunCompletedEvent
 from core.workflow.nodes.event.types import NodeEvent
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from core.workflow.nodes.start.entities import StartNodeData
-from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_entry import WorkflowEntry
 from events.app_event import app_draft_workflow_was_synced, app_published_workflow_was_updated
 from extensions.ext_database import db
@@ -370,7 +369,7 @@ class WorkflowService:
 
         else:
             variable_pool = VariablePool(
-                system_variables=SystemVariable.empty(),
+                system_variables={},
                 user_inputs=user_inputs,
                 environment_variables=draft_workflow.environment_variables,
                 conversation_variables=[],
@@ -686,30 +685,36 @@ def _setup_variable_pool(
 ):
     # Only inject system variables for START node type.
     if node_type == NodeType.START:
-        system_variable = SystemVariable(
-            user_id=user_id,
-            app_id=workflow.app_id,
-            workflow_id=workflow.id,
-            files=files or [],
-            workflow_execution_id=str(uuid.uuid4()),
-        )
+        # Create a variable pool.
+        system_inputs: dict[SystemVariableKey, Any] = {
+            # From inputs:
+            SystemVariableKey.FILES: files,
+            SystemVariableKey.USER_ID: user_id,
+            # From workflow model
+            SystemVariableKey.APP_ID: workflow.app_id,
+            SystemVariableKey.WORKFLOW_ID: workflow.id,
+            # Randomly generated.
+            SystemVariableKey.WORKFLOW_EXECUTION_ID: str(uuid.uuid4()),
+        }
 
         # Only add chatflow-specific variables for non-workflow types
         if workflow.type != WorkflowType.WORKFLOW.value:
-            system_variable.query = query
-            system_variable.conversation_id = conversation_id
-            system_variable.dialogue_count = 0
+            system_inputs.update(
+                {
+                    SystemVariableKey.QUERY: query,
+                    SystemVariableKey.CONVERSATION_ID: conversation_id,
+                    SystemVariableKey.DIALOGUE_COUNT: 0,
+                }
+            )
     else:
-        system_variable = SystemVariable.empty()
+        system_inputs = {}
 
     # init variable pool
     variable_pool = VariablePool(
-        system_variables=system_variable,
+        system_variables=system_inputs,
         user_inputs=user_inputs,
         environment_variables=workflow.environment_variables,
-        # Based on the definition of `VariableUnion`,
-        # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-        conversation_variables=cast(list[VariableUnion], conversation_variables),  #
+        conversation_variables=conversation_variables,
     )
 
     return variable_pool

api/tasks/mail_change_mail_task.py

@@ -1,78 +0,0 @@
-import logging
-import time
-
-import click
-from celery import shared_task  # type: ignore
-from flask import render_template
-
-from extensions.ext_mail import mail
-from services.feature_service import FeatureService
-
-
-@shared_task(queue="mail")
-def send_change_mail_task(language: str, to: str, code: str, phase: str):
-    """
-    Async Send change email mail
-    :param language: Language in which the email should be sent (e.g., 'en', 'zh')
-    :param to: Recipient email address
-    :param code: Change email code
-    :param phase: Change email phase (new_email, old_email)
-    """
-    if not mail.is_inited():
-        return
-
-    logging.info(click.style("Start change email mail to {}".format(to), fg="green"))
-    start_at = time.perf_counter()
-
-    email_config = {
-        "zh-Hans": {
-            "old_email": {
-                "subject": "检测您现在的邮箱",
-                "template_with_brand": "change_mail_confirm_old_template_zh-CN.html",
-                "template_without_brand": "without-brand/change_mail_confirm_old_template_zh-CN.html",
-            },
-            "new_email": {
-                "subject": "确认您的邮箱地址变更",
-                "template_with_brand": "change_mail_confirm_new_template_zh-CN.html",
-                "template_without_brand": "without-brand/change_mail_confirm_new_template_zh-CN.html",
-            },
-        },
-        "en": {
-            "old_email": {
-                "subject": "Check your current email",
-                "template_with_brand": "change_mail_confirm_old_template_en-US.html",
-                "template_without_brand": "without-brand/change_mail_confirm_old_template_en-US.html",
-            },
-            "new_email": {
-                "subject": "Confirm your new email address",
-                "template_with_brand": "change_mail_confirm_new_template_en-US.html",
-                "template_without_brand": "without-brand/change_mail_confirm_new_template_en-US.html",
-            },
-        },
-    }
-
-    # send change email mail using different languages
-    try:
-        system_features = FeatureService.get_system_features()
-        lang_key = "zh-Hans" if language == "zh-Hans" else "en"
-
-        if phase not in ["old_email", "new_email"]:
-            raise ValueError("Invalid phase")
-
-        config = email_config[lang_key][phase]
-        subject = config["subject"]
-
-        if system_features.branding.enabled:
-            template = config["template_without_brand"]
-        else:
-            template = config["template_with_brand"]
-
-        html_content = render_template(template, to=to, code=code)
-        mail.send(to=to, subject=subject, html=html_content)
-
-        end_at = time.perf_counter()
-        logging.info(
-            click.style("Send change email mail to {} succeeded: latency: {}".format(to, end_at - start_at), fg="green")
-        )
-    except Exception:
-        logging.exception("Send change email mail to {} failed".format(to))