fix: update OAuth button logic to improve configuration handling

.devcontainer/post_create_command.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-npm add -g pnpm@10.13.1
+npm add -g pnpm@10.11.1
 cd web && pnpm install
 pipx install uv
 
@@ -12,4 +12,3 @@ echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f do
 echo 'alias stop-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down"' >> ~/.bashrc
 
 source /home/vscode/.bashrc
-

.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:
       - "*"
 

.github/workflows/style.yml

@@ -28,7 +28,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@v45
         with:
           files: |
             api/**
@@ -75,7 +75,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@v45
         with:
           files: web/**
 
@@ -113,7 +113,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@v45
         with:
           files: |
             docker/generate_docker_compose
@@ -144,7 +144,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@v45
         with:
           files: |
             **.sh
@@ -152,15 +152,13 @@ jobs:
             **.yml
             **Dockerfile
             dev/**
-            .editorconfig
 
       - name: Super-linter
-        uses: super-linter/super-linter/slim@v8
+        uses: super-linter/super-linter/slim@v7
         if: steps.changed-files.outputs.any_changed == 'true'
         env:
           BASH_SEVERITY: warning
-          DEFAULT_BRANCH: origin/main
-          EDITORCONFIG_FILE_NAME: editorconfig-checker.json
+          DEFAULT_BRANCH: main
           FILTER_REGEX_INCLUDE: pnpm-lock.yaml
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           IGNORE_GENERATED_FILES: true
@@ -170,6 +168,16 @@ jobs:
           # FIXME: temporarily disabled until api-docker.yaml's run script is fixed for shellcheck
           # VALIDATE_GITHUB_ACTIONS: true
           VALIDATE_DOCKERFILE_HADOLINT: true
-          VALIDATE_EDITORCONFIG: true
           VALIDATE_XML: true
           VALIDATE_YAML: true
+
+      - name: EditorConfig checks
+        uses: super-linter/super-linter/slim@v7
+        env:
+          DEFAULT_BRANCH: main
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          IGNORE_GENERATED_FILES: true
+          IGNORE_GITIGNORED_FILES: true
+          # EditorConfig validation
+          VALIDATE_EDITORCONFIG: true
+          EDITORCONFIG_FILE_NAME: editorconfig-checker.json

.github/workflows/web-tests.yml

@@ -27,7 +27,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v46
+        uses: tj-actions/changed-files@v45
         with:
           files: web/**
 

api/.env.example

@@ -5,17 +5,17 @@
 SECRET_KEY=
 
 # Console API base URL
-CONSOLE_API_URL=http://localhost:5001
-CONSOLE_WEB_URL=http://localhost:3000
+CONSOLE_API_URL=http://127.0.0.1:5001
+CONSOLE_WEB_URL=http://127.0.0.1:3000
 
 # Service API base URL
-SERVICE_API_URL=http://localhost:5001
+SERVICE_API_URL=http://127.0.0.1:5001
 
 # Web APP base URL
-APP_WEB_URL=http://localhost:3000
+APP_WEB_URL=http://127.0.0.1:3000
 
 # Files URL
-FILES_URL=http://localhost:5001
+FILES_URL=http://127.0.0.1:5001
 
 # INTERNAL_FILES_URL is used for plugin daemon communication within Docker network.
 # Set this to the internal Docker service URL for proper plugin file access.
@@ -54,7 +54,7 @@ REDIS_CLUSTERS_PASSWORD=
 
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
-CELERY_BACKEND=redis
+
 # PostgreSQL database configuration
 DB_USERNAME=postgres
 DB_PASSWORD=difyai123456
@@ -138,14 +138,12 @@ SUPABASE_API_KEY=your-access-key
 SUPABASE_URL=your-server-url
 
 # CORS configuration
-WEB_API_CORS_ALLOW_ORIGINS=http://localhost:3000,*
-CONSOLE_CORS_ALLOW_ORIGINS=http://localhost:3000,*
+WEB_API_CORS_ALLOW_ORIGINS=http://127.0.0.1:3000,*
+CONSOLE_CORS_ALLOW_ORIGINS=http://127.0.0.1:3000,*
 
 # Vector database configuration
-# Supported values are `weaviate`, `qdrant`, `milvus`, `myscale`, `relyt`, `pgvector`, `pgvecto-rs`, `chroma`, `opensearch`, `oracle`, `tencent`, `elasticsearch`, `elasticsearch-ja`, `analyticdb`, `couchbase`, `vikingdb`, `oceanbase`, `opengauss`, `tablestore`,`vastbase`,`tidb`,`tidb_on_qdrant`,`baidu`,`lindorm`,`huawei_cloud`,`upstash`, `matrixone`.
+# support: weaviate, qdrant, milvus, myscale, relyt, pgvecto_rs, pgvector, pgvector, chroma, opensearch, tidb_vector, couchbase, vikingdb, upstash, lindorm, oceanbase, opengauss, tablestore, matrixone
 VECTOR_STORE=weaviate
-# Prefix used to create collection name in vector database
-VECTOR_INDEX_NAME_PREFIX=Vector_index
 
 # Weaviate configuration
 WEAVIATE_ENDPOINT=http://localhost:8080
@@ -497,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
 
@@ -509,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/Dockerfile

@@ -47,8 +47,6 @@ RUN \
         curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
         # For Security
         expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
-        # install fonts to support the use of tools like pypdfium2
-        fonts-noto-cjk \
         # install a package to improve the accuracy of guessing mime type and file extension
         media-types \
         # install libmagic to support the use of python-magic guess MIMETYPE

api/commands.py

@@ -2,22 +2,19 @@ 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
 
 from configs import dify_config
 from constants.languages import languages
-from core.plugin.entities.plugin import ToolProviderID
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import Document
-from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -30,7 +27,6 @@ from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, D
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
 from models.provider import Provider, ProviderModel
-from models.tools import ToolOAuthSystemClient
 from services.account_service import AccountService, RegisterService, TenantService
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
 from services.plugin.data_migration import PluginDataMigration
@@ -1159,49 +1155,3 @@ def remove_orphaned_files_on_storage(force: bool):
         click.echo(click.style(f"Removed {removed_files} orphaned files without errors.", fg="green"))
     else:
         click.echo(click.style(f"Removed {removed_files} orphaned files, with {error_files} errors.", fg="yellow"))
-
-
-@click.command("setup-system-tool-oauth-client", help="Setup system tool oauth client.")
-@click.option("--provider", prompt=True, help="Provider name")
-@click.option("--client-params", prompt=True, help="Client Params")
-def setup_system_tool_oauth_client(provider, client_params):
-    """
-    Setup system tool oauth client
-    """
-    provider_id = ToolProviderID(provider)
-    provider_name = provider_id.provider_name
-    plugin_id = provider_id.plugin_id
-
-    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)
-        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)
-        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"))
-        return
-
-    deleted_count = (
-        db.session.query(ToolOAuthSystemClient)
-        .filter_by(
-            provider=provider_name,
-            plugin_id=plugin_id,
-        )
-        .delete()
-    )
-    if deleted_count > 0:
-        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
-
-    oauth_client = ToolOAuthSystemClient(
-        provider=provider_name,
-        plugin_id=plugin_id,
-        encrypted_oauth_params=oauth_client_params,
-    )
-    db.session.add(oauth_client)
-    db.session.commit()
-    click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))

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

@@ -85,11 +85,6 @@ class VectorStoreConfig(BaseSettings):
         default=False,
     )
 
-    VECTOR_INDEX_NAME_PREFIX: Optional[str] = Field(
-        description="Prefix used to create collection name in vector database",
-        default="Vector_index",
-    )
-
 
 class KeywordStoreConfig(BaseSettings):
     KEYWORD_STORE: str = Field(
@@ -167,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,
@@ -209,14 +199,13 @@ 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,
         }
 
 
 class CeleryConfig(DatabaseConfig):
     CELERY_BACKEND: str = Field(
         description="Backend for Celery task results. Options: 'database', 'redis'.",
-        default="redis",
+        default="database",
     )
 
     CELERY_BROKER_URL: Optional[str] = Field(

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/constants/__init__.py

@@ -1,7 +1,6 @@
 from configs import dify_config
 
 HIDDEN_VALUE = "[__HIDDEN__]"
-UNKNOWN_VALUE = "[__UNKNOWN__]"
 UUID_NIL = "00000000-0000-0000-0000-000000000000"
 
 DEFAULT_FILE_NUMBER_LIMITS = 3

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/conversation.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import UTC, datetime
 
 import pytz  # pip install pytz
 from flask_login import current_user
@@ -19,7 +19,6 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
-from libs.datetime_utils import naive_utc_now
 from libs.helper import DatetimeString
 from libs.login import login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -316,7 +315,7 @@ def _get_conversation(app_model, conversation_id):
         raise NotFound("Conversation Not Exists.")
 
     if not conversation.read_at:
-        conversation.read_at = naive_utc_now()
+        conversation.read_at = datetime.now(UTC).replace(tzinfo=None)
         conversation.read_account_id = current_user.id
         db.session.commit()
 

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/message.py

@@ -5,7 +5,6 @@ from flask_restful import Resource, fields, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
-import services
 from controllers.console import api
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -28,7 +27,7 @@ from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import login_required
-from models.model import AppMode, Conversation, Message, MessageAnnotation
+from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
@@ -125,17 +124,34 @@ class MessageFeedbackApi(Resource):
         parser.add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
         args = parser.parse_args()
 
-        try:
-            MessageService.create_feedback(
-                app_model=app_model,
-                message_id=str(args["message_id"]),
-                user=current_user,
-                rating=args.get("rating"),
-                content=None,
-            )
-        except services.errors.message.MessageNotExistsError:
+        message_id = str(args["message_id"])
+
+        message = db.session.query(Message).filter(Message.id == message_id, Message.app_id == app_model.id).first()
+
+        if not message:
             raise NotFound("Message Not Exists.")
 
+        feedback = message.admin_feedback
+
+        if not args["rating"] and feedback:
+            db.session.delete(feedback)
+        elif args["rating"] and feedback:
+            feedback.rating = args["rating"]
+        elif not args["rating"] and not feedback:
+            raise ValueError("rating cannot be None when feedback not exists")
+        else:
+            feedback = MessageFeedback(
+                app_id=app_model.id,
+                conversation_id=message.conversation_id,
+                message_id=message.id,
+                rating=args["rating"],
+                from_source="admin",
+                from_account_id=current_user.id,
+            )
+            db.session.add(feedback)
+
+        db.session.commit()
+
         return {"result": "success"}
 
 

api/controllers/console/app/site.py

@@ -1,3 +1,5 @@
+from datetime import UTC, datetime
+
 from flask_login import current_user
 from flask_restful import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
@@ -8,7 +10,6 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
 from fields.app_fields import app_site_fields
-from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Site
 
@@ -76,7 +77,7 @@ class AppSite(Resource):
                 setattr(site, attr_name, value)
 
         site.updated_by = current_user.id
-        site.updated_at = naive_utc_now()
+        site.updated_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()
 
         return site
@@ -100,7 +101,7 @@ class AppSiteAccessTokenReset(Resource):
 
         site.code = Site.generate_code(16)
         site.updated_by = current_user.id
-        site.updated_at = naive_utc_now()
+        site.updated_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()
 
         return site

api/controllers/console/app/statistic.py

@@ -2,7 +2,6 @@ from datetime import datetime
 from decimal import Decimal
 
 import pytz
-import sqlalchemy as sa
 from flask import jsonify
 from flask_login import current_user
 from flask_restful import Resource, reqparse
@@ -10,11 +9,10 @@ from flask_restful import Resource, reqparse
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from libs.helper import DatetimeString
 from libs.login import login_required
-from models import AppMode, Message
+from models.model import AppMode
 
 
 class DailyMessageStatistic(Resource):
@@ -87,41 +85,46 @@ class DailyConversationStatistic(Resource):
         parser.add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         args = parser.parse_args()
 
+        sql_query = """SELECT
+    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+    COUNT(DISTINCT messages.conversation_id) AS conversation_count
+FROM
+    messages
+WHERE
+    app_id = :app_id"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
 
-        stmt = (
-            sa.select(
-                sa.func.date(
-                    sa.func.date_trunc("day", sa.text("created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz"))
-                ).label("date"),
-                sa.func.count(sa.distinct(Message.conversation_id)).label("conversation_count"),
-            )
-            .select_from(Message)
-            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER.value)
-        )
-
         if args["start"]:
             start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M")
             start_datetime = start_datetime.replace(second=0)
+
             start_datetime_timezone = timezone.localize(start_datetime)
             start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone)
-            stmt = stmt.where(Message.created_at >= start_datetime_utc)
+
+            sql_query += " AND created_at >= :start"
+            arg_dict["start"] = start_datetime_utc
 
         if args["end"]:
             end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M")
             end_datetime = end_datetime.replace(second=0)
+
             end_datetime_timezone = timezone.localize(end_datetime)
             end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone)
-            stmt = stmt.where(Message.created_at < end_datetime_utc)
 
-        stmt = stmt.group_by("date").order_by("date")
+            sql_query += " AND created_at < :end"
+            arg_dict["end"] = end_datetime_utc
+
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
+
         with db.engine.begin() as conn:
-            rs = conn.execute(stmt, {"tz": account.timezone})
-            for row in rs:
-                response_data.append({"date": str(row.date), "conversation_count": row.conversation_count})
+            rs = conn.execute(db.text(sql_query), arg_dict)
+            for i in rs:
+                response_data.append({"date": str(i.date), "conversation_count": i.conversation_count})
 
         return jsonify({"data": response_data})
 

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/activate.py

@@ -1,3 +1,5 @@
+import datetime
+
 from flask import request
 from flask_restful import Resource, reqparse
 
@@ -5,7 +7,6 @@ from constants.languages import supported_language
 from controllers.console import api
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from libs.helper import StrLen, email, extract_remote_ip, timezone
 from models.account import AccountStatus
 from services.account_service import AccountService, RegisterService
@@ -64,7 +65,7 @@ class ActivateApi(Resource):
         account.timezone = args["timezone"]
         account.interface_theme = "light"
         account.status = AccountStatus.ACTIVE.value
-        account.initialized_at = naive_utc_now()
+        account.initialized_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
         db.session.commit()
 
         token_pair = AccountService.login(account, ip_address=extract_remote_ip(request))

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/auth/oauth.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import UTC, datetime
 from typing import Optional
 
 import requests
@@ -12,7 +13,6 @@ from configs import dify_config
 from constants.languages import languages
 from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_remote_ip
 from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo
 from models import Account
@@ -110,7 +110,7 @@ class OAuthCallback(Resource):
 
         if account.status == AccountStatus.PENDING.value:
             account.status = AccountStatus.ACTIVE.value
-            account.initialized_at = naive_utc_now()
+            account.initialized_at = datetime.now(UTC).replace(tzinfo=None)
             db.session.commit()
 
         try:

api/controllers/console/datasets/data_source.py

@@ -1,3 +1,4 @@
+import datetime
 import json
 
 from flask import request
@@ -14,7 +15,6 @@ from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.notion_extractor import NotionExtractor
 from extensions.ext_database import db
 from fields.data_source_fields import integrate_list_fields, integrate_notion_info_list_fields
-from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import DataSourceOauthBinding, Document
 from services.dataset_service import DatasetService, DocumentService
@@ -88,7 +88,7 @@ class DataSourceApi(Resource):
         if action == "enable":
             if data_source_binding.disabled:
                 data_source_binding.disabled = False
-                data_source_binding.updated_at = naive_utc_now()
+                data_source_binding.updated_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
                 db.session.add(data_source_binding)
                 db.session.commit()
             else:
@@ -97,7 +97,7 @@ class DataSourceApi(Resource):
         if action == "disable":
             if not data_source_binding.disabled:
                 data_source_binding.disabled = True
-                data_source_binding.updated_at = naive_utc_now()
+                data_source_binding.updated_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
                 db.session.add(data_source_binding)
                 db.session.commit()
             else:

api/controllers/console/datasets/datasets.py

@@ -211,6 +211,10 @@ class DatasetApi(Resource):
         else:
             data["embedding_available"] = True
 
+        if data.get("permission") == "partial_members":
+            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
+            data.update({"partial_member_list": part_users_list})
+
         return data, 200
 
     @setup_required

api/controllers/console/datasets/datasets_document.py

@@ -1,5 +1,6 @@
 import logging
 from argparse import ArgumentTypeError
+from datetime import UTC, datetime
 from typing import cast
 
 from flask import request
@@ -48,7 +49,6 @@ from fields.document_fields import (
     document_status_fields,
     document_with_segments_fields,
 )
-from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Dataset, DatasetProcessRule, Document, DocumentSegment, UploadFile
 from services.dataset_service import DatasetService, DocumentService
@@ -750,7 +750,7 @@ class DocumentProcessingApi(DocumentResource):
                 raise InvalidActionError("Document not in indexing state.")
 
             document.paused_by = current_user.id
-            document.paused_at = naive_utc_now()
+            document.paused_at = datetime.now(UTC).replace(tzinfo=None)
             document.is_paused = True
             db.session.commit()
 
@@ -830,7 +830,7 @@ class DocumentMetadataApi(DocumentResource):
                     document.doc_metadata[key] = value
 
         document.doc_type = doc_type
-        document.updated_at = naive_utc_now()
+        document.updated_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()
 
         return {"result": "success", "message": "Document metadata updated."}, 200

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/datasets/website.py

@@ -4,7 +4,7 @@ from controllers.console import api
 from controllers.console.datasets.error import WebsiteCrawlError
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
-from services.website_service import WebsiteCrawlApiRequest, WebsiteCrawlStatusApiRequest, WebsiteService
+from services.website_service import WebsiteService
 
 
 class WebsiteCrawlApi(Resource):
@@ -24,16 +24,10 @@ class WebsiteCrawlApi(Resource):
         parser.add_argument("url", type=str, required=True, nullable=True, location="json")
         parser.add_argument("options", type=dict, required=True, nullable=True, location="json")
         args = parser.parse_args()
-
-        # Create typed request and validate
+        WebsiteService.document_create_args_validate(args)
+        # crawl url
         try:
-            api_request = WebsiteCrawlApiRequest.from_args(args)
-        except ValueError as e:
-            raise WebsiteCrawlError(str(e))
-
-        # Crawl URL using typed request
-        try:
-            result = WebsiteService.crawl_url(api_request)
+            result = WebsiteService.crawl_url(args)
         except Exception as e:
             raise WebsiteCrawlError(str(e))
         return result, 200
@@ -49,16 +43,9 @@ class WebsiteCrawlStatusApi(Resource):
             "provider", type=str, choices=["firecrawl", "watercrawl", "jinareader"], required=True, location="args"
         )
         args = parser.parse_args()
-
-        # Create typed request and validate
+        # get crawl status
         try:
-            api_request = WebsiteCrawlStatusApiRequest.from_args(args, job_id)
-        except ValueError as e:
-            raise WebsiteCrawlError(str(e))
-
-        # Get crawl status using typed request
-        try:
-            result = WebsiteService.get_crawl_status_typed(api_request)
+            result = WebsiteService.get_crawl_status(job_id, args["provider"])
         except Exception as e:
             raise WebsiteCrawlError(str(e))
         return result, 200

api/controllers/console/explore/completion.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import UTC, datetime
 
 from flask_login import current_user
 from flask_restful import reqparse
@@ -26,7 +27,6 @@ from core.errors.error import (
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from libs import helper
-from libs.datetime_utils import naive_utc_now
 from libs.helper import uuid_value
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
@@ -51,7 +51,7 @@ class CompletionApi(InstalledAppResource):
         streaming = args["response_mode"] == "streaming"
         args["auto_generate_name"] = False
 
-        installed_app.last_used_at = naive_utc_now()
+        installed_app.last_used_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()
 
         try:
@@ -111,7 +111,7 @@ class ChatApi(InstalledAppResource):
 
         args["auto_generate_name"] = False
 
-        installed_app.last_used_at = naive_utc_now()
+        installed_app.last_used_at = datetime.now(UTC).replace(tzinfo=None)
         db.session.commit()
 
         try:

api/controllers/console/explore/installed_app.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import UTC, datetime
 from typing import Any
 
 from flask import request
@@ -12,7 +13,6 @@ from controllers.console.explore.wraps import InstalledAppResource
 from controllers.console.wraps import account_initialization_required, cloud_edition_billing_resource_check
 from extensions.ext_database import db
 from fields.installed_app_fields import installed_app_list_fields
-from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import App, InstalledApp, RecommendedApp
 from services.account_service import TenantService
@@ -122,7 +122,7 @@ class InstalledAppsListApi(Resource):
                 tenant_id=current_tenant_id,
                 app_owner_tenant_id=app.tenant_id,
                 is_pinned=False,
-                last_used_at=naive_utc_now(),
+                last_used_at=datetime.now(UTC).replace(tzinfo=None),
             )
             db.session.add(new_installed_app)
             db.session.commit()

api/controllers/console/workspace/account.py

@@ -1,21 +1,13 @@
+import datetime
+
 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,
@@ -26,18 +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.datetime_utils import naive_utc_now
-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
@@ -79,7 +68,7 @@ class AccountInitApi(Resource):
                 raise InvalidInvitationCodeError()
 
             invitation_code.status = "used"
-            invitation_code.used_at = naive_utc_now()
+            invitation_code.used_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
             invitation_code.used_by_tenant_id = account.current_tenant_id
             invitation_code.used_by_account_id = account.id
 
@@ -87,7 +76,7 @@ class AccountInitApi(Resource):
         account.timezone = args["timezone"]
         account.interface_theme = "light"
         account.status = "active"
-        account.initialized_at = naive_utc_now()
+        account.initialized_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
         db.session.commit()
 
         return {"result": "success"}
@@ -380,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")
@@ -524,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

@@ -1,35 +1,26 @@
 import io
 from urllib.parse import urlparse
 
-from flask import make_response, redirect, request, send_file
+from flask import redirect, send_file
 from flask_login import current_user
-from flask_restful import (
-    Resource,
-    reqparse,
-)
+from flask_restful import Resource, reqparse
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from controllers.console import api
-from controllers.console.wraps import (
-    account_initialization_required,
-    enterprise_license_required,
-    setup_required,
-)
+from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
 from core.mcp.auth.auth_flow import auth, handle_callback
 from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin import ToolProviderID
-from core.plugin.impl.oauth import OAuthHandler
-from core.tools.entities.tool_entities import CredentialType
-from libs.helper import StrLen, alphanumeric, uuid_value
+from extensions.ext_database import db
+from libs.helper import alphanumeric, uuid_value
 from libs.login import login_required
-from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
-from services.tools.mcp_tools_manage_service import MCPToolManageService
+from services.tools.mcp_tools_mange_service import MCPToolManageService
 from services.tools.tool_labels_service import ToolLabelsService
 from services.tools.tools_manage_service import ToolCommonService
 from services.tools.tools_transform_service import ToolTransformService
@@ -98,7 +89,7 @@ class ToolBuiltinProviderInfoApi(Resource):
         user_id = user.id
         tenant_id = user.current_tenant_id
 
-        return jsonable_encoder(BuiltinToolManageService.get_builtin_tool_provider_info(tenant_id, provider))
+        return jsonable_encoder(BuiltinToolManageService.get_builtin_tool_provider_info(user_id, tenant_id, provider))
 
 
 class ToolBuiltinProviderDeleteApi(Resource):
@@ -107,47 +98,17 @@ class ToolBuiltinProviderDeleteApi(Resource):
     @account_initialization_required
     def post(self, provider):
         user = current_user
+
         if not user.is_admin_or_owner:
             raise Forbidden()
 
-        tenant_id = user.current_tenant_id
-        req = reqparse.RequestParser()
-        req.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
-        args = req.parse_args()
-
-        return BuiltinToolManageService.delete_builtin_tool_provider(
-            tenant_id,
-            provider,
-            args["credential_id"],
-        )
-
-
-class ToolBuiltinProviderAddApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider):
-        user = current_user
-
         user_id = user.id
         tenant_id = user.current_tenant_id
 
-        parser = reqparse.RequestParser()
-        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=False, nullable=False, location="json")
-        parser.add_argument("type", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-
-        if args["type"] not in CredentialType.values():
-            raise ValueError(f"Invalid credential type: {args['type']}")
-
-        return BuiltinToolManageService.add_builtin_tool_provider(
-            user_id=user_id,
-            tenant_id=tenant_id,
-            provider=provider,
-            credentials=args["credentials"],
-            name=args["name"],
-            api_type=CredentialType.of(args["type"]),
+        return BuiltinToolManageService.delete_builtin_tool_provider(
+            user_id,
+            tenant_id,
+            provider,
         )
 
 
@@ -165,20 +126,19 @@ class ToolBuiltinProviderUpdateApi(Resource):
         tenant_id = user.current_tenant_id
 
         parser = reqparse.RequestParser()
-        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
+        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
 
         args = parser.parse_args()
 
-        result = BuiltinToolManageService.update_builtin_tool_provider(
-            user_id=user_id,
-            tenant_id=tenant_id,
-            provider=provider,
-            credential_id=args["credential_id"],
-            credentials=args.get("credentials", None),
-            name=args.get("name", ""),
-        )
+        with Session(db.engine) as session:
+            result = BuiltinToolManageService.update_builtin_tool_provider(
+                session=session,
+                user_id=user_id,
+                tenant_id=tenant_id,
+                provider_name=provider,
+                credentials=args["credentials"],
+            )
+            session.commit()
         return result
 
 
@@ -189,11 +149,9 @@ class ToolBuiltinProviderGetCredentialsApi(Resource):
     def get(self, provider):
         tenant_id = current_user.current_tenant_id
 
-        return jsonable_encoder(
-            BuiltinToolManageService.get_builtin_tool_provider_credentials(
-                tenant_id=tenant_id,
-                provider_name=provider,
-            )
+        return BuiltinToolManageService.get_builtin_tool_provider_credentials(
+            tenant_id=tenant_id,
+            provider_name=provider,
         )
 
 
@@ -386,15 +344,12 @@ class ToolBuiltinProviderCredentialsSchemaApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, provider, credential_type):
+    def get(self, provider):
         user = current_user
+
         tenant_id = user.current_tenant_id
 
-        return jsonable_encoder(
-            BuiltinToolManageService.list_builtin_provider_credentials_schema(
-                provider, CredentialType.of(credential_type), tenant_id
-            )
-        )
+        return BuiltinToolManageService.list_builtin_provider_credentials_schema(provider, tenant_id)
 
 
 class ToolApiProviderSchemaApi(Resource):
@@ -631,12 +586,15 @@ class ToolApiListApi(Resource):
     @account_initialization_required
     def get(self):
         user = current_user
+
+        user_id = user.id
         tenant_id = user.current_tenant_id
 
         return jsonable_encoder(
             [
                 provider.to_dict()
                 for provider in ApiToolManageService.list_api_tools(
+                    user_id,
                     tenant_id,
                 )
             ]
@@ -673,183 +631,6 @@ class ToolLabelsApi(Resource):
         return jsonable_encoder(ToolLabelsService.list_tool_labels())
 
 
-class ToolPluginOAuthApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        tool_provider = ToolProviderID(provider)
-        plugin_id = tool_provider.plugin_id
-        provider_name = tool_provider.provider_name
-
-        # todo check permission
-        user = current_user
-
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        tenant_id = user.current_tenant_id
-        oauth_client_params = BuiltinToolManageService.get_oauth_client(tenant_id=tenant_id, provider=provider)
-        if oauth_client_params is None:
-            raise Forbidden("no oauth available client config found for this tool provider")
-
-        oauth_handler = OAuthHandler()
-        context_id = OAuthProxyService.create_proxy_context(
-            user_id=current_user.id, tenant_id=tenant_id, plugin_id=plugin_id, provider=provider_name
-        )
-        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
-        authorization_url_response = oauth_handler.get_authorization_url(
-            tenant_id=tenant_id,
-            user_id=user.id,
-            plugin_id=plugin_id,
-            provider=provider_name,
-            redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
-        )
-        response = make_response(jsonable_encoder(authorization_url_response))
-        response.set_cookie(
-            "context_id",
-            context_id,
-            httponly=True,
-            samesite="Lax",
-            max_age=OAuthProxyService.__MAX_AGE__,
-        )
-        return response
-
-
-class ToolOAuthCallback(Resource):
-    @setup_required
-    def get(self, provider):
-        context_id = request.cookies.get("context_id")
-        if not context_id:
-            raise Forbidden("context_id not found")
-
-        context = OAuthProxyService.use_proxy_context(context_id)
-        if context is None:
-            raise Forbidden("Invalid context_id")
-
-        tool_provider = ToolProviderID(provider)
-        plugin_id = tool_provider.plugin_id
-        provider_name = tool_provider.provider_name
-        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
-
-        oauth_handler = OAuthHandler()
-        oauth_client_params = BuiltinToolManageService.get_oauth_client(tenant_id, provider)
-        if oauth_client_params is None:
-            raise Forbidden("no oauth available client config found for this tool provider")
-
-        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
-        credentials_response = oauth_handler.get_credentials(
-            tenant_id=tenant_id,
-            user_id=user_id,
-            plugin_id=plugin_id,
-            provider=provider_name,
-            redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
-            request=request,
-        )
-
-        credentials = credentials_response.credentials
-        expires_at = credentials_response.expires_at
-
-        if not credentials:
-            raise Exception("the plugin credentials failed")
-
-        # add credentials to database
-        BuiltinToolManageService.add_builtin_tool_provider(
-            user_id=user_id,
-            tenant_id=tenant_id,
-            provider=provider,
-            credentials=dict(credentials),
-            expires_at=expires_at,
-            api_type=CredentialType.OAUTH2,
-        )
-        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
-
-
-class ToolBuiltinProviderSetDefaultApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True, nullable=False, location="json")
-        args = parser.parse_args()
-        return BuiltinToolManageService.set_default_provider(
-            tenant_id=current_user.current_tenant_id, user_id=current_user.id, provider=provider, id=args["id"]
-        )
-
-
-class ToolOAuthCustomClient(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def post(self, provider):
-        parser = reqparse.RequestParser()
-        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
-        parser.add_argument("enable_oauth_custom_client", type=bool, required=False, nullable=True, location="json")
-        args = parser.parse_args()
-
-        user = current_user
-
-        if not user.is_admin_or_owner:
-            raise Forbidden()
-
-        return BuiltinToolManageService.save_custom_oauth_client_params(
-            tenant_id=user.current_tenant_id,
-            provider=provider,
-            client_params=args.get("client_params", {}),
-            enable_oauth_custom_client=args.get("enable_oauth_custom_client", True),
-        )
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        return jsonable_encoder(
-            BuiltinToolManageService.get_custom_oauth_client_params(
-                tenant_id=current_user.current_tenant_id, provider=provider
-            )
-        )
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def delete(self, provider):
-        return jsonable_encoder(
-            BuiltinToolManageService.delete_custom_oauth_client_params(
-                tenant_id=current_user.current_tenant_id, provider=provider
-            )
-        )
-
-
-class ToolBuiltinProviderGetOauthClientSchemaApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        return jsonable_encoder(
-            BuiltinToolManageService.get_builtin_tool_provider_oauth_client_schema(
-                tenant_id=current_user.current_tenant_id, provider_name=provider
-            )
-        )
-
-
-class ToolBuiltinProviderGetCredentialInfoApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self, provider):
-        tenant_id = current_user.current_tenant_id
-
-        return jsonable_encoder(
-            BuiltinToolManageService.get_builtin_tool_provider_credential_info(
-                tenant_id=tenant_id,
-                provider=provider,
-            )
-        )
-
-
 class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
@@ -1013,33 +794,17 @@ class ToolMCPCallbackApi(Resource):
 # tool provider
 api.add_resource(ToolProviderListApi, "/workspaces/current/tool-providers")
 
-# tool oauth
-api.add_resource(ToolPluginOAuthApi, "/oauth/plugin/<path:provider>/tool/authorization-url")
-api.add_resource(ToolOAuthCallback, "/oauth/plugin/<path:provider>/tool/callback")
-api.add_resource(ToolOAuthCustomClient, "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/custom-client")
-
 # builtin tool provider
 api.add_resource(ToolBuiltinProviderListToolsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/tools")
 api.add_resource(ToolBuiltinProviderInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/info")
-api.add_resource(ToolBuiltinProviderAddApi, "/workspaces/current/tool-provider/builtin/<path:provider>/add")
 api.add_resource(ToolBuiltinProviderDeleteApi, "/workspaces/current/tool-provider/builtin/<path:provider>/delete")
 api.add_resource(ToolBuiltinProviderUpdateApi, "/workspaces/current/tool-provider/builtin/<path:provider>/update")
-api.add_resource(
-    ToolBuiltinProviderSetDefaultApi, "/workspaces/current/tool-provider/builtin/<path:provider>/default-credential"
-)
-api.add_resource(
-    ToolBuiltinProviderGetCredentialInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credential/info"
-)
 api.add_resource(
     ToolBuiltinProviderGetCredentialsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credentials"
 )
 api.add_resource(
     ToolBuiltinProviderCredentialsSchemaApi,
-    "/workspaces/current/tool-provider/builtin/<path:provider>/credential/schema/<path:credential_type>",
-)
-api.add_resource(
-    ToolBuiltinProviderGetOauthClientSchemaApi,
-    "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/client-schema",
+    "/workspaces/current/tool-provider/builtin/<path:provider>/credentials_schema",
 )
 api.add_resource(ToolBuiltinProviderIconApi, "/workspaces/current/tool-provider/builtin/<path:provider>/icon")
 

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,6 @@ class PluginInvokeToolApi(Resource):
                     provider=payload.provider,
                     tool_name=payload.tool,
                     tool_parameters=payload.tool_parameters,
-                    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/controllers/service_api/wraps.py

@@ -1,6 +1,6 @@
 import time
 from collections.abc import Callable
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 from enum import Enum
 from functools import wraps
 from typing import Optional
@@ -15,7 +15,6 @@ from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from libs.datetime_utils import naive_utc_now
 from libs.login import _get_user
 from models.account import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
@@ -257,7 +256,7 @@ def validate_and_get_api_token(scope: str | None = None):
     if auth_scheme != "bearer":
         raise Unauthorized("Authorization scheme must be 'Bearer'")
 
-    current_time = naive_utc_now()
+    current_time = datetime.now(UTC).replace(tzinfo=None)
     cutoff_time = current_time - timedelta(minutes=1)
     with Session(db.engine, expire_on_commit=False) as session:
         update_stmt = (

api/core/agent/entities.py

@@ -16,7 +16,6 @@ class AgentToolEntity(BaseModel):
     tool_name: str
     tool_parameters: dict[str, Any] = Field(default_factory=dict)
     plugin_unique_identifier: str | None = None
-    credential_id: str | None = None
 
 
 class AgentPromptEntity(BaseModel):

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/base.py

@@ -4,7 +4,6 @@ from typing import Any, Optional
 
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyParameter
-from core.plugin.entities.request import InvokeCredentials
 
 
 class BaseAgentStrategy(ABC):
@@ -19,12 +18,11 @@ class BaseAgentStrategy(ABC):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
-        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent strategy.
         """
-        yield from self._invoke(params, user_id, conversation_id, app_id, message_id, credentials)
+        yield from self._invoke(params, user_id, conversation_id, app_id, message_id)
 
     def get_parameters(self) -> Sequence[AgentStrategyParameter]:
         """
@@ -40,6 +38,5 @@ class BaseAgentStrategy(ABC):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
-        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         pass

api/core/agent/strategy/plugin.py

@@ -4,7 +4,6 @@ from typing import Any, Optional
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyEntity, AgentStrategyParameter
 from core.agent.strategy.base import BaseAgentStrategy
-from core.plugin.entities.request import InvokeCredentials, PluginInvokeContext
 from core.plugin.impl.agent import PluginAgentClient
 from core.plugin.utils.converter import convert_parameters_to_plugin_format
 
@@ -41,7 +40,6 @@ class PluginAgentStrategy(BaseAgentStrategy):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
-        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent strategy.
@@ -60,5 +58,4 @@ class PluginAgentStrategy(BaseAgentStrategy):
             conversation_id=conversation_id,
             app_id=app_id,
             message_id=message_id,
-            context=PluginInvokeContext(credentials=credentials or InvokeCredentials()),
         )

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

@@ -39,7 +39,6 @@ class AgentConfigManager:
                         "provider_id": tool["provider_id"],
                         "tool_name": tool["tool_name"],
                         "tool_parameters": tool.get("tool_parameters", {}),
-                        "credential_id": tool.get("credential_id", None),
                     }
 
                     agent_tools.append(AgentToolEntity(**agent_tool_properties))

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

@@ -17,8 +17,7 @@ from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
 from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter
 from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom

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/agent_chat/app_generator.py

@@ -15,8 +15,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfigManager
 from core.app.apps.agent_chat.app_runner import AgentChatAppRunner
 from core.app.apps.agent_chat.generate_response_converter import AgentChatAppGenerateResponseConverter
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, InvokeFrom

api/core/app/apps/base_app_queue_manager.py

@@ -169,3 +169,7 @@ class AppQueueManager:
                 raise TypeError(
                     "Critical Error: Passing SQLAlchemy Model instances that cause thread safety issues is not allowed."
                 )
+
+
+class GenerateTaskStoppedError(Exception):
+    pass

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]
     ):
@@ -118,7 +181,7 @@ class AppRunner:
         else:
             memory_config = MemoryConfig(window=MemoryConfig.WindowConfig(enabled=False))
 
-            model_mode = ModelMode(model_config.mode)
+            model_mode = ModelMode.value_of(model_config.mode)
             prompt_template: Union[CompletionModelPromptTemplate, list[ChatModelMessage]]
             if model_mode == ModelMode.COMPLETION:
                 advanced_completion_prompt_template = prompt_template_entity.advanced_completion_prompt_template

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

@@ -11,11 +11,10 @@ from configs import dify_config
 from constants import UUID_NIL
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.chat.app_config_manager import ChatAppConfigManager
 from core.app.apps.chat.app_runner import ChatAppRunner
 from core.app.apps.chat.generate_response_converter import ChatAppGenerateResponseConverter
-from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity, InvokeFrom

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

@@ -10,11 +10,10 @@ from pydantic import ValidationError
 from configs import dify_config
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
 from core.app.apps.completion.app_runner import CompletionAppRunner
 from core.app.apps.completion.generate_response_converter import CompletionAppGenerateResponseConverter
-from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, InvokeFrom

api/core/app/apps/exc.py

@@ -1,2 +0,0 @@
-class GenerateTaskStoppedError(Exception):
-    pass

api/core/app/apps/message_based_app_generator.py

@@ -1,12 +1,12 @@
 import json
 import logging
 from collections.abc import Generator
+from datetime import UTC, datetime
 from typing import Optional, Union, cast
 
 from core.app.app_config.entities import EasyUIBasedAppConfig, EasyUIBasedAppModelConfigFrom
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     AgentChatAppGenerateEntity,
@@ -24,7 +24,6 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
 from models import Account
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, AppModelConfig, Conversation, EndUser, Message, MessageFile
@@ -184,7 +183,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             db.session.commit()
             db.session.refresh(conversation)
         else:
-            conversation.updated_at = naive_utc_now()
+            conversation.updated_at = datetime.now(UTC).replace(tzinfo=None)
             db.session.commit()
 
         message = Message(

api/core/app/apps/message_based_app_queue_manager.py

@@ -1,5 +1,4 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,

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

@@ -13,8 +13,7 @@ import contexts
 from configs import dify_config
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.apps.workflow.app_runner import WorkflowAppRunner

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

@@ -1,5 +1,4 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
-from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,

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

@@ -1,8 +1,7 @@
 import logging
 import time
-from collections.abc import Callable, Generator
-from contextlib import contextmanager
-from typing import Any, Optional, Union
+from collections.abc import Generator
+from typing import Optional, Union
 
 from sqlalchemy.orm import Session
 
@@ -14,7 +13,6 @@ from core.app.entities.app_invoke_entities import (
     WorkflowAppGenerateEntity,
 )
 from core.app.entities.queue_entities import (
-    MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
     QueueIterationCompletedEvent,
@@ -40,13 +38,11 @@ from core.app.entities.queue_entities import (
     QueueWorkflowPartialSuccessEvent,
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
-    WorkflowQueueMessage,
 )
 from core.app.entities.task_entities import (
     ErrorStreamResponse,
     MessageAudioEndStreamResponse,
     MessageAudioStreamResponse,
-    PingStreamResponse,
     StreamResponse,
     TextChunkStreamResponse,
     WorkflowAppBlockingResponse,
@@ -58,11 +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.graph_engine.entities.graph_runtime_state import GraphRuntimeState
+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
@@ -112,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),
@@ -251,492 +246,315 @@ class WorkflowAppGenerateTaskPipeline:
         if tts_publisher:
             yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
 
-    @contextmanager
-    def _database_session(self):
-        """Context manager for database sessions."""
-        with Session(db.engine, expire_on_commit=False) as session:
-            try:
-                yield session
-                session.commit()
-            except Exception:
-                session.rollback()
-                raise
-
-    def _ensure_workflow_initialized(self) -> None:
-        """Fluent validation for workflow state."""
-        if not self._workflow_run_id:
-            raise ValueError("workflow run not initialized.")
-
-    def _ensure_graph_runtime_initialized(self, graph_runtime_state: Optional[GraphRuntimeState]) -> GraphRuntimeState:
-        """Fluent validation for graph runtime state."""
-        if not graph_runtime_state:
-            raise ValueError("graph runtime state not initialized.")
-        return graph_runtime_state
-
-    def _handle_ping_event(self, event: QueuePingEvent, **kwargs) -> Generator[PingStreamResponse, None, None]:
-        """Handle ping events."""
-        yield self._base_task_pipeline._ping_stream_response()
-
-    def _handle_error_event(self, event: QueueErrorEvent, **kwargs) -> Generator[ErrorStreamResponse, None, None]:
-        """Handle error events."""
-        err = self._base_task_pipeline._handle_error(event=event)
-        yield self._base_task_pipeline._error_to_stream_response(err)
-
-    def _handle_workflow_started_event(
-        self, event: QueueWorkflowStartedEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle workflow started events."""
-        # init workflow run
-        workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
-        self._workflow_run_id = workflow_execution.id_
-        start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution=workflow_execution,
-        )
-        yield start_resp
-
-    def _handle_node_retry_event(self, event: QueueNodeRetryEvent, **kwargs) -> Generator[StreamResponse, None, None]:
-        """Handle node retry events."""
-        self._ensure_workflow_initialized()
-
-        with self._database_session() as session:
-            workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
-                workflow_execution_id=self._workflow_run_id,
-                event=event,
-            )
-            response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
-                event=event,
-                task_id=self._application_generate_entity.task_id,
-                workflow_node_execution=workflow_node_execution,
-            )
-
-        if response:
-            yield response
-
-    def _handle_node_started_event(
-        self, event: QueueNodeStartedEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle node started events."""
-        self._ensure_workflow_initialized()
-
-        workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
-            workflow_execution_id=self._workflow_run_id, event=event
-        )
-        node_start_response = self._workflow_response_converter.workflow_node_start_to_stream_response(
-            event=event,
-            task_id=self._application_generate_entity.task_id,
-            workflow_node_execution=workflow_node_execution,
-        )
-
-        if node_start_response:
-            yield node_start_response
-
-    def _handle_node_succeeded_event(
-        self, event: QueueNodeSucceededEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle node succeeded events."""
-        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(event=event)
-        node_success_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
-            event=event,
-            task_id=self._application_generate_entity.task_id,
-            workflow_node_execution=workflow_node_execution,
-        )
-
-        self._save_output_for_event(event, workflow_node_execution.id)
-
-        if node_success_response:
-            yield node_success_response
-
-    def _handle_node_failed_events(
-        self,
-        event: Union[
-            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
-        ],
-        **kwargs,
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle various node failure events."""
-        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_failed(
-            event=event,
-        )
-        node_failed_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
-            event=event,
-            task_id=self._application_generate_entity.task_id,
-            workflow_node_execution=workflow_node_execution,
-        )
-
-        if isinstance(event, QueueNodeExceptionEvent):
-            self._save_output_for_event(event, workflow_node_execution.id)
-
-        if node_failed_response:
-            yield node_failed_response
-
-    def _handle_parallel_branch_started_event(
-        self, event: QueueParallelBranchRunStartedEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle parallel branch started events."""
-        self._ensure_workflow_initialized()
-
-        parallel_start_resp = self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield parallel_start_resp
-
-    def _handle_parallel_branch_finished_events(
-        self, event: Union[QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent], **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle parallel branch finished events."""
-        self._ensure_workflow_initialized()
-
-        parallel_finish_resp = self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield parallel_finish_resp
-
-    def _handle_iteration_start_event(
-        self, event: QueueIterationStartEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle iteration start events."""
-        self._ensure_workflow_initialized()
-
-        iter_start_resp = self._workflow_response_converter.workflow_iteration_start_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield iter_start_resp
-
-    def _handle_iteration_next_event(
-        self, event: QueueIterationNextEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle iteration next events."""
-        self._ensure_workflow_initialized()
-
-        iter_next_resp = self._workflow_response_converter.workflow_iteration_next_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield iter_next_resp
-
-    def _handle_iteration_completed_event(
-        self, event: QueueIterationCompletedEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle iteration completed events."""
-        self._ensure_workflow_initialized()
-
-        iter_finish_resp = self._workflow_response_converter.workflow_iteration_completed_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield iter_finish_resp
-
-    def _handle_loop_start_event(self, event: QueueLoopStartEvent, **kwargs) -> Generator[StreamResponse, None, None]:
-        """Handle loop start events."""
-        self._ensure_workflow_initialized()
-
-        loop_start_resp = self._workflow_response_converter.workflow_loop_start_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield loop_start_resp
-
-    def _handle_loop_next_event(self, event: QueueLoopNextEvent, **kwargs) -> Generator[StreamResponse, None, None]:
-        """Handle loop next events."""
-        self._ensure_workflow_initialized()
-
-        loop_next_resp = self._workflow_response_converter.workflow_loop_next_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield loop_next_resp
-
-    def _handle_loop_completed_event(
-        self, event: QueueLoopCompletedEvent, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle loop completed events."""
-        self._ensure_workflow_initialized()
-
-        loop_finish_resp = self._workflow_response_converter.workflow_loop_completed_to_stream_response(
-            task_id=self._application_generate_entity.task_id,
-            workflow_execution_id=self._workflow_run_id,
-            event=event,
-        )
-        yield loop_finish_resp
-
-    def _handle_workflow_succeeded_event(
-        self,
-        event: QueueWorkflowSucceededEvent,
-        *,
-        graph_runtime_state: Optional[GraphRuntimeState] = None,
-        trace_manager: Optional[TraceQueueManager] = None,
-        **kwargs,
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle workflow succeeded events."""
-        self._ensure_workflow_initialized()
-        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
-
-        with self._database_session() as session:
-            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_success(
-                workflow_run_id=self._workflow_run_id,
-                total_tokens=validated_state.total_tokens,
-                total_steps=validated_state.node_run_steps,
-                outputs=event.outputs,
-                conversation_id=None,
-                trace_manager=trace_manager,
-            )
-
-            # save workflow app log
-            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
-
-            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                session=session,
-                task_id=self._application_generate_entity.task_id,
-                workflow_execution=workflow_execution,
-            )
-
-        yield workflow_finish_resp
-
-    def _handle_workflow_partial_success_event(
-        self,
-        event: QueueWorkflowPartialSuccessEvent,
-        *,
-        graph_runtime_state: Optional[GraphRuntimeState] = None,
-        trace_manager: Optional[TraceQueueManager] = None,
-        **kwargs,
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle workflow partial success events."""
-        self._ensure_workflow_initialized()
-        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
-
-        with self._database_session() as session:
-            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_partial_success(
-                workflow_run_id=self._workflow_run_id,
-                total_tokens=validated_state.total_tokens,
-                total_steps=validated_state.node_run_steps,
-                outputs=event.outputs,
-                exceptions_count=event.exceptions_count,
-                conversation_id=None,
-                trace_manager=trace_manager,
-            )
-
-            # save workflow app log
-            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
-
-            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                session=session,
-                task_id=self._application_generate_entity.task_id,
-                workflow_execution=workflow_execution,
-            )
-
-        yield workflow_finish_resp
-
-    def _handle_workflow_failed_and_stop_events(
-        self,
-        event: Union[QueueWorkflowFailedEvent, QueueStopEvent],
-        *,
-        graph_runtime_state: Optional[GraphRuntimeState] = None,
-        trace_manager: Optional[TraceQueueManager] = None,
-        **kwargs,
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle workflow failed and stop events."""
-        self._ensure_workflow_initialized()
-        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
-
-        with self._database_session() as session:
-            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
-                workflow_run_id=self._workflow_run_id,
-                total_tokens=validated_state.total_tokens,
-                total_steps=validated_state.node_run_steps,
-                status=WorkflowExecutionStatus.FAILED
-                if isinstance(event, QueueWorkflowFailedEvent)
-                else WorkflowExecutionStatus.STOPPED,
-                error_message=event.error if isinstance(event, QueueWorkflowFailedEvent) else event.get_stop_reason(),
-                conversation_id=None,
-                trace_manager=trace_manager,
-                exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
-            )
-
-            # save workflow app log
-            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
-
-            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                session=session,
-                task_id=self._application_generate_entity.task_id,
-                workflow_execution=workflow_execution,
-            )
-
-        yield workflow_finish_resp
-
-    def _handle_text_chunk_event(
-        self,
-        event: QueueTextChunkEvent,
-        *,
-        tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
-        queue_message: Optional[Union[WorkflowQueueMessage, MessageQueueMessage]] = None,
-        **kwargs,
-    ) -> Generator[StreamResponse, None, None]:
-        """Handle text chunk events."""
-        delta_text = event.text
-        if delta_text is None:
-            return
-
-        # only publish tts message at text chunk streaming
-        if tts_publisher and queue_message:
-            tts_publisher.publish(queue_message)
-
-        yield self._text_chunk_to_stream_response(delta_text, from_variable_selector=event.from_variable_selector)
-
-    def _handle_agent_log_event(self, event: QueueAgentLogEvent, **kwargs) -> Generator[StreamResponse, None, None]:
-        """Handle agent log events."""
-        yield self._workflow_response_converter.handle_agent_log(
-            task_id=self._application_generate_entity.task_id, event=event
-        )
-
-    def _get_event_handlers(self) -> dict[type, Callable]:
-        """Get mapping of event types to their handlers using fluent pattern."""
-        return {
-            # Basic events
-            QueuePingEvent: self._handle_ping_event,
-            QueueErrorEvent: self._handle_error_event,
-            QueueTextChunkEvent: self._handle_text_chunk_event,
-            # Workflow events
-            QueueWorkflowStartedEvent: self._handle_workflow_started_event,
-            QueueWorkflowSucceededEvent: self._handle_workflow_succeeded_event,
-            QueueWorkflowPartialSuccessEvent: self._handle_workflow_partial_success_event,
-            # Node events
-            QueueNodeRetryEvent: self._handle_node_retry_event,
-            QueueNodeStartedEvent: self._handle_node_started_event,
-            QueueNodeSucceededEvent: self._handle_node_succeeded_event,
-            # Parallel branch events
-            QueueParallelBranchRunStartedEvent: self._handle_parallel_branch_started_event,
-            # Iteration events
-            QueueIterationStartEvent: self._handle_iteration_start_event,
-            QueueIterationNextEvent: self._handle_iteration_next_event,
-            QueueIterationCompletedEvent: self._handle_iteration_completed_event,
-            # Loop events
-            QueueLoopStartEvent: self._handle_loop_start_event,
-            QueueLoopNextEvent: self._handle_loop_next_event,
-            QueueLoopCompletedEvent: self._handle_loop_completed_event,
-            # Agent events
-            QueueAgentLogEvent: self._handle_agent_log_event,
-        }
-
-    def _dispatch_event(
-        self,
-        event: Any,
-        *,
-        graph_runtime_state: Optional[GraphRuntimeState] = None,
-        tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
-        trace_manager: Optional[TraceQueueManager] = None,
-        queue_message: Optional[Union[WorkflowQueueMessage, MessageQueueMessage]] = None,
-    ) -> Generator[StreamResponse, None, None]:
-        """Dispatch events using elegant pattern matching."""
-        handlers = self._get_event_handlers()
-        event_type = type(event)
-
-        # Direct handler lookup
-        if handler := handlers.get(event_type):
-            yield from handler(
-                event,
-                graph_runtime_state=graph_runtime_state,
-                tts_publisher=tts_publisher,
-                trace_manager=trace_manager,
-                queue_message=queue_message,
-            )
-            return
-
-        # Handle node failure events with isinstance check
-        if isinstance(
-            event,
-            (
-                QueueNodeFailedEvent,
-                QueueNodeInIterationFailedEvent,
-                QueueNodeInLoopFailedEvent,
-                QueueNodeExceptionEvent,
-            ),
-        ):
-            yield from self._handle_node_failed_events(
-                event,
-                graph_runtime_state=graph_runtime_state,
-                tts_publisher=tts_publisher,
-                trace_manager=trace_manager,
-                queue_message=queue_message,
-            )
-            return
-
-        # Handle parallel branch finished events with isinstance check
-        if isinstance(event, (QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent)):
-            yield from self._handle_parallel_branch_finished_events(
-                event,
-                graph_runtime_state=graph_runtime_state,
-                tts_publisher=tts_publisher,
-                trace_manager=trace_manager,
-                queue_message=queue_message,
-            )
-            return
-
-        # Handle workflow failed and stop events with isinstance check
-        if isinstance(event, (QueueWorkflowFailedEvent, QueueStopEvent)):
-            yield from self._handle_workflow_failed_and_stop_events(
-                event,
-                graph_runtime_state=graph_runtime_state,
-                tts_publisher=tts_publisher,
-                trace_manager=trace_manager,
-                queue_message=queue_message,
-            )
-            return
-
-        # For unhandled events, we continue (original behavior)
-        return
-
     def _process_stream_response(
         self,
         tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
         trace_manager: Optional[TraceQueueManager] = None,
     ) -> Generator[StreamResponse, None, None]:
         """
-        Process stream response using elegant Fluent Python patterns.
-        Maintains exact same functionality as original 44-if-statement version.
+        Process stream response.
+        :return:
         """
-        # Initialize graph runtime state
         graph_runtime_state = None
 
         for queue_message in self._base_task_pipeline._queue_manager.listen():
             event = queue_message.event
 
-            match event:
-                case QueueWorkflowStartedEvent():
-                    graph_runtime_state = event.graph_runtime_state
-                    yield from self._handle_workflow_started_event(event)
+            if isinstance(event, QueuePingEvent):
+                yield self._base_task_pipeline._ping_stream_response()
+            elif isinstance(event, QueueErrorEvent):
+                err = self._base_task_pipeline._handle_error(event=event)
+                yield self._base_task_pipeline._error_to_stream_response(err)
+                break
+            elif isinstance(event, QueueWorkflowStartedEvent):
+                # override graph runtime state
+                graph_runtime_state = event.graph_runtime_state
 
-                case QueueTextChunkEvent():
-                    yield from self._handle_text_chunk_event(
-                        event, tts_publisher=tts_publisher, queue_message=queue_message
+                # init workflow run
+                workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
+                self._workflow_run_id = workflow_execution.id_
+                start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution=workflow_execution,
+                )
+
+                yield start_resp
+            elif isinstance(
+                event,
+                QueueNodeRetryEvent,
+            ):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+                with Session(db.engine, expire_on_commit=False) as session:
+                    workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
+                        workflow_execution_id=self._workflow_run_id,
+                        event=event,
+                    )
+                    response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
+                        event=event,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_node_execution=workflow_node_execution,
+                    )
+                    session.commit()
+
+                if response:
+                    yield response
+            elif isinstance(event, QueueNodeStartedEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
+                    workflow_execution_id=self._workflow_run_id, event=event
+                )
+                node_start_response = self._workflow_response_converter.workflow_node_start_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
+
+                if node_start_response:
+                    yield node_start_response
+            elif isinstance(event, QueueNodeSucceededEvent):
+                workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(
+                    event=event
+                )
+                node_success_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
+
+                self._save_output_for_event(event, workflow_node_execution.id)
+
+                if node_success_response:
+                    yield node_success_response
+            elif isinstance(
+                event,
+                QueueNodeFailedEvent
+                | QueueNodeInIterationFailedEvent
+                | QueueNodeInLoopFailedEvent
+                | QueueNodeExceptionEvent,
+            ):
+                workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_failed(
+                    event=event,
+                )
+                node_failed_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )
+                if isinstance(event, QueueNodeExceptionEvent):
+                    self._save_output_for_event(event, workflow_node_execution.id)
+
+                if node_failed_response:
+                    yield node_failed_response
+
+            elif isinstance(event, QueueParallelBranchRunStartedEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                parallel_start_resp = (
+                    self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution_id=self._workflow_run_id,
+                        event=event,
+                    )
+                )
+
+                yield parallel_start_resp
+
+            elif isinstance(event, QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                parallel_finish_resp = (
+                    self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution_id=self._workflow_run_id,
+                        event=event,
+                    )
+                )
+
+                yield parallel_finish_resp
+
+            elif isinstance(event, QueueIterationStartEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                iter_start_resp = self._workflow_response_converter.workflow_iteration_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield iter_start_resp
+
+            elif isinstance(event, QueueIterationNextEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                iter_next_resp = self._workflow_response_converter.workflow_iteration_next_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield iter_next_resp
+
+            elif isinstance(event, QueueIterationCompletedEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                iter_finish_resp = self._workflow_response_converter.workflow_iteration_completed_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield iter_finish_resp
+
+            elif isinstance(event, QueueLoopStartEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                loop_start_resp = self._workflow_response_converter.workflow_loop_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield loop_start_resp
+
+            elif isinstance(event, QueueLoopNextEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                loop_next_resp = self._workflow_response_converter.workflow_loop_next_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield loop_next_resp
+
+            elif isinstance(event, QueueLoopCompletedEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+
+                loop_finish_resp = self._workflow_response_converter.workflow_loop_completed_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )
+
+                yield loop_finish_resp
+
+            elif isinstance(event, QueueWorkflowSucceededEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+                if not graph_runtime_state:
+                    raise ValueError("graph runtime state not initialized.")
+
+                with Session(db.engine, expire_on_commit=False) as session:
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_success(
+                        workflow_run_id=self._workflow_run_id,
+                        total_tokens=graph_runtime_state.total_tokens,
+                        total_steps=graph_runtime_state.node_run_steps,
+                        outputs=event.outputs,
+                        conversation_id=None,
+                        trace_manager=trace_manager,
                     )
 
-                case QueueErrorEvent():
-                    yield from self._handle_error_event(event)
-                    break
+                    # save workflow app log
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
 
-                # Handle all other events through elegant dispatch
-                case _:
-                    if responses := list(
-                        self._dispatch_event(
-                            event,
-                            graph_runtime_state=graph_runtime_state,
-                            tts_publisher=tts_publisher,
-                            trace_manager=trace_manager,
-                            queue_message=queue_message,
-                        )
-                    ):
-                        yield from responses
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
+                    )
+                    session.commit()
+
+                yield workflow_finish_resp
+            elif isinstance(event, QueueWorkflowPartialSuccessEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+                if not graph_runtime_state:
+                    raise ValueError("graph runtime state not initialized.")
+
+                with Session(db.engine, expire_on_commit=False) as session:
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_partial_success(
+                        workflow_run_id=self._workflow_run_id,
+                        total_tokens=graph_runtime_state.total_tokens,
+                        total_steps=graph_runtime_state.node_run_steps,
+                        outputs=event.outputs,
+                        exceptions_count=event.exceptions_count,
+                        conversation_id=None,
+                        trace_manager=trace_manager,
+                    )
+
+                    # save workflow app log
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
+                    )
+                    session.commit()
+
+                yield workflow_finish_resp
+            elif isinstance(event, QueueWorkflowFailedEvent | QueueStopEvent):
+                if not self._workflow_run_id:
+                    raise ValueError("workflow run not initialized.")
+                if not graph_runtime_state:
+                    raise ValueError("graph runtime state not initialized.")
+
+                with Session(db.engine, expire_on_commit=False) as session:
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
+                        workflow_run_id=self._workflow_run_id,
+                        total_tokens=graph_runtime_state.total_tokens,
+                        total_steps=graph_runtime_state.node_run_steps,
+                        status=WorkflowExecutionStatus.FAILED
+                        if isinstance(event, QueueWorkflowFailedEvent)
+                        else WorkflowExecutionStatus.STOPPED,
+                        error_message=event.error
+                        if isinstance(event, QueueWorkflowFailedEvent)
+                        else event.get_stop_reason(),
+                        conversation_id=None,
+                        trace_manager=trace_manager,
+                        exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
+                    )
+
+                    # save workflow app log
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
+                    )
+                    session.commit()
+
+                yield workflow_finish_resp
+            elif isinstance(event, QueueTextChunkEvent):
+                delta_text = event.text
+                if delta_text is None:
+                    continue
+
+                # only publish tts message at text chunk streaming
+                if tts_publisher:
+                    tts_publisher.publish(queue_message)
+
+                yield self._text_chunk_to_stream_response(
+                    delta_text, from_variable_selector=event.from_variable_selector
+                )
+            elif isinstance(event, QueueAgentLogEvent):
+                yield self._workflow_response_converter.handle_agent_log(
+                    task_id=self._application_generate_entity.task_id, event=event
+                )
+            else:
+                continue
 
         if tts_publisher:
             tts_publisher.publish(None)

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/file_manager.py

@@ -7,7 +7,6 @@ from core.model_runtime.entities import (
     AudioPromptMessageContent,
     DocumentPromptMessageContent,
     ImagePromptMessageContent,
-    TextPromptMessageContent,
     VideoPromptMessageContent,
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
@@ -45,44 +44,11 @@ def to_prompt_message_content(
     *,
     image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
 ) -> PromptMessageContentUnionTypes:
-    """
-    Convert a file to prompt message content.
-
-    This function converts files to their appropriate prompt message content types.
-    For supported file types (IMAGE, AUDIO, VIDEO, DOCUMENT), it creates the
-    corresponding message content with proper encoding/URL.
-
-    For unsupported file types, instead of raising an error, it returns a
-    TextPromptMessageContent with a descriptive message about the file.
-
-    Args:
-        f: The file to convert
-        image_detail_config: Optional detail configuration for image files
-
-    Returns:
-        PromptMessageContentUnionTypes: The appropriate message content type
-
-    Raises:
-        ValueError: If file extension or mime_type is missing
-    """
     if f.extension is None:
         raise ValueError("Missing file extension")
     if f.mime_type is None:
         raise ValueError("Missing file mime_type")
 
-    prompt_class_map: Mapping[FileType, type[PromptMessageContentUnionTypes]] = {
-        FileType.IMAGE: ImagePromptMessageContent,
-        FileType.AUDIO: AudioPromptMessageContent,
-        FileType.VIDEO: VideoPromptMessageContent,
-        FileType.DOCUMENT: DocumentPromptMessageContent,
-    }
-
-    # Check if file type is supported
-    if f.type not in prompt_class_map:
-        # For unsupported file types, return a text description
-        return TextPromptMessageContent(data=f"[Unsupported file type: {f.filename} ({f.type.value})]")
-
-    # Process supported file types
     params = {
         "base64_data": _get_encoded_string(f) if dify_config.MULTIMODAL_SEND_FORMAT == "base64" else "",
         "url": _to_url(f) if dify_config.MULTIMODAL_SEND_FORMAT == "url" else "",
@@ -92,7 +58,17 @@ def to_prompt_message_content(
     if f.type == FileType.IMAGE:
         params["detail"] = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
-    return prompt_class_map[f.type].model_validate(params)
+    prompt_class_map: Mapping[FileType, type[PromptMessageContentUnionTypes]] = {
+        FileType.IMAGE: ImagePromptMessageContent,
+        FileType.AUDIO: AudioPromptMessageContent,
+        FileType.VIDEO: VideoPromptMessageContent,
+        FileType.DOCUMENT: DocumentPromptMessageContent,
+    }
+
+    try:
+        return prompt_class_map[f.type].model_validate(params)
+    except KeyError:
+        raise ValueError(f"file type {f.type} is not supported")
 
 
 def download(f: File, /):

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/encrypter.py

@@ -21,7 +21,7 @@ def encrypt_token(tenant_id: str, token: str):
     return base64.b64encode(encrypted_token).decode()
 
 
-def decrypt_token(tenant_id: str, token: str) -> str:
+def decrypt_token(tenant_id: str, token: str):
     return rsa.decrypt(base64.b64decode(token), tenant_id)
 
 

api/core/helper/provider_cache.py

@@ -1,84 +0,0 @@
-import json
-from abc import ABC, abstractmethod
-from json import JSONDecodeError
-from typing import Any, Optional
-
-from extensions.ext_redis import redis_client
-
-
-class ProviderCredentialsCache(ABC):
-    """Base class for provider credentials cache"""
-
-    def __init__(self, **kwargs):
-        self.cache_key = self._generate_cache_key(**kwargs)
-
-    @abstractmethod
-    def _generate_cache_key(self, **kwargs) -> str:
-        """Generate cache key based on subclass implementation"""
-        pass
-
-    def get(self) -> Optional[dict]:
-        """Get cached provider credentials"""
-        cached_credentials = redis_client.get(self.cache_key)
-        if cached_credentials:
-            try:
-                cached_credentials = cached_credentials.decode("utf-8")
-                return dict(json.loads(cached_credentials))
-            except JSONDecodeError:
-                return None
-        return None
-
-    def set(self, config: dict[str, Any]) -> None:
-        """Cache provider credentials"""
-        redis_client.setex(self.cache_key, 86400, json.dumps(config))
-
-    def delete(self) -> None:
-        """Delete cached provider credentials"""
-        redis_client.delete(self.cache_key)
-
-
-class SingletonProviderCredentialsCache(ProviderCredentialsCache):
-    """Cache for tool single provider credentials"""
-
-    def __init__(self, tenant_id: str, provider_type: str, provider_identity: str):
-        super().__init__(
-            tenant_id=tenant_id,
-            provider_type=provider_type,
-            provider_identity=provider_identity,
-        )
-
-    def _generate_cache_key(self, **kwargs) -> str:
-        tenant_id = kwargs["tenant_id"]
-        provider_type = kwargs["provider_type"]
-        identity_name = kwargs["provider_identity"]
-        identity_id = f"{provider_type}.{identity_name}"
-        return f"{provider_type}_credentials:tenant_id:{tenant_id}:id:{identity_id}"
-
-
-class ToolProviderCredentialsCache(ProviderCredentialsCache):
-    """Cache for tool provider credentials"""
-
-    def __init__(self, tenant_id: str, provider: str, credential_id: str):
-        super().__init__(tenant_id=tenant_id, provider=provider, credential_id=credential_id)
-
-    def _generate_cache_key(self, **kwargs) -> str:
-        tenant_id = kwargs["tenant_id"]
-        provider = kwargs["provider"]
-        credential_id = kwargs["credential_id"]
-        return f"tool_credentials:tenant_id:{tenant_id}:provider:{provider}:credential_id:{credential_id}"
-
-
-class NoOpProviderCredentialCache:
-    """No-op provider credential cache"""
-
-    def get(self) -> Optional[dict]:
-        """Get cached provider credentials"""
-        return None
-
-    def set(self, config: dict[str, Any]) -> None:
-        """Cache provider credentials"""
-        pass
-
-    def delete(self) -> None:
-        """Delete cached provider credentials"""
-        pass

api/core/helper/tool_provider_cache.py

@@ -0,0 +1,51 @@
+import json
+from enum import Enum
+from json import JSONDecodeError
+from typing import Optional
+
+from extensions.ext_redis import redis_client
+
+
+class ToolProviderCredentialsCacheType(Enum):
+    PROVIDER = "tool_provider"
+    ENDPOINT = "endpoint"
+
+
+class ToolProviderCredentialsCache:
+    def __init__(self, tenant_id: str, identity_id: str, cache_type: ToolProviderCredentialsCacheType):
+        self.cache_key = f"{cache_type.value}_credentials:tenant_id:{tenant_id}:id:{identity_id}"
+
+    def get(self) -> Optional[dict]:
+        """
+        Get cached model provider credentials.
+
+        :return:
+        """
+        cached_provider_credentials = redis_client.get(self.cache_key)
+        if cached_provider_credentials:
+            try:
+                cached_provider_credentials = cached_provider_credentials.decode("utf-8")
+                cached_provider_credentials = json.loads(cached_provider_credentials)
+            except JSONDecodeError:
+                return None
+
+            return dict(cached_provider_credentials)
+        else:
+            return None
+
+    def set(self, credentials: dict) -> None:
+        """
+        Cache model provider credentials.
+
+        :param credentials: provider credentials
+        :return:
+        """
+        redis_client.setex(self.cache_key, 86400, json.dumps(credentials))
+
+    def delete(self) -> None:
+        """
+        Delete cached model provider credentials.
+
+        :return:
+        """
+        redis_client.delete(self.cache_key)

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/auth/auth_provider.py

@@ -8,7 +8,7 @@ from core.mcp.types import (
     OAuthTokens,
 )
 from models.tools import MCPToolProvider
-from services.tools.mcp_tools_manage_service import MCPToolManageService
+from services.tools.mcp_tools_mange_service import MCPToolManageService
 
 LATEST_PROTOCOL_VERSION = "1.0"
 

api/core/mcp/mcp_client.py

@@ -68,17 +68,15 @@ class MCPClient:
         }
 
         parsed_url = urlparse(self.server_url)
-        path = parsed_url.path or ""
+        path = parsed_url.path
         method_name = path.rstrip("/").split("/")[-1] if path else ""
-        if method_name in connection_methods:
+        try:
             client_factory = connection_methods[method_name]
             self.connect_server(client_factory, method_name)
-        else:
+        except KeyError:
             try:
-                logger.debug(f"Not supported method {method_name} found in URL path, trying default 'mcp' method.")
                 self.connect_server(sse_client, "sse")
             except MCPConnectionError:
-                logger.debug("MCP connection failed with 'sse', falling back to 'mcp' method.")
                 self.connect_server(streamablehttp_client, "mcp")
 
     def connect_server(
@@ -93,7 +91,7 @@ class MCPClient:
                 else {}
             )
             self._streams_context = client_factory(url=self.server_url, headers=headers)
-            if not self._streams_context:
+            if self._streams_context is None:
                 raise MCPConnectionError("Failed to create connection context")
 
             # Use exit_stack to manage context managers properly
@@ -143,11 +141,10 @@ class MCPClient:
         try:
             # ExitStack will handle proper cleanup of all managed context managers
             self.exit_stack.close()
-        except Exception as e:
-            logging.exception("Error during cleanup")
-            raise ValueError(f"Error during cleanup: {e}")
-        finally:
             self._session = None
             self._session_context = None
             self._streams_context = None
             self._initialized = False
+        except Exception as e:
+            logging.exception("Error during cleanup")
+            raise ValueError(f"Error during cleanup: {e}")

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/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -3,7 +3,7 @@ import json
 import logging
 import os
 from datetime import datetime, timedelta
-from typing import Any, Optional, Union, cast
+from typing import Optional, Union, cast
 
 from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes
 from opentelemetry import trace
@@ -142,8 +142,11 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             raise
 
     def workflow_trace(self, trace_info: WorkflowTraceInfo):
+        if trace_info.message_data is None:
+            return
+
         workflow_metadata = {
-            "workflow_run_id": trace_info.workflow_run_id or "",
+            "workflow_id": trace_info.workflow_run_id or "",
             "message_id": trace_info.message_id or "",
             "workflow_app_log_id": trace_info.workflow_app_log_id or "",
             "status": trace_info.workflow_run_status or "",
@@ -153,7 +156,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         workflow_metadata.update(trace_info.metadata)
 
-        trace_id = uuid_to_trace_id(trace_info.workflow_run_id)
+        trace_id = uuid_to_trace_id(trace_info.message_id)
         span_id = RandomIdGenerator().generate_span_id()
         context = SpanContext(
             trace_id=trace_id,
@@ -210,7 +213,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                     if model:
                         node_metadata["ls_model_name"] = model
 
-                    outputs = json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
+                    outputs = json.loads(node_execution.outputs).get("usage", {})
                     usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                     if usage_data:
                         node_metadata["total_tokens"] = usage_data.get("total_tokens", 0)
@@ -233,34 +236,31 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                         SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
                     },
                     start_time=datetime_to_nanos(created_at),
-                    context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
                 )
 
                 try:
                     if node_execution.node_type == "llm":
-                        llm_attributes: dict[str, Any] = {
-                            SpanAttributes.INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
-                        }
                         provider = process_data.get("model_provider")
                         model = process_data.get("model_name")
                         if provider:
-                            llm_attributes[SpanAttributes.LLM_PROVIDER] = provider
+                            node_span.set_attribute(SpanAttributes.LLM_PROVIDER, provider)
                         if model:
-                            llm_attributes[SpanAttributes.LLM_MODEL_NAME] = model
-                        outputs = (
-                            json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
-                        )
+                            node_span.set_attribute(SpanAttributes.LLM_MODEL_NAME, model)
+
+                        outputs = json.loads(node_execution.outputs).get("usage", {})
                         usage_data = (
                             process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                         )
                         if usage_data:
-                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_TOTAL] = usage_data.get("total_tokens", 0)
-                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_PROMPT] = usage_data.get("prompt_tokens", 0)
-                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_COMPLETION] = usage_data.get(
-                                "completion_tokens", 0
+                            node_span.set_attribute(
+                                SpanAttributes.LLM_TOKEN_COUNT_TOTAL, usage_data.get("total_tokens", 0)
+                            )
+                            node_span.set_attribute(
+                                SpanAttributes.LLM_TOKEN_COUNT_PROMPT, usage_data.get("prompt_tokens", 0)
+                            )
+                            node_span.set_attribute(
+                                SpanAttributes.LLM_TOKEN_COUNT_COMPLETION, usage_data.get("completion_tokens", 0)
                             )
-                        llm_attributes.update(self._construct_llm_attributes(process_data.get("prompts", [])))
-                        node_span.set_attributes(llm_attributes)
                 finally:
                     node_span.end(end_time=datetime_to_nanos(finished_at))
         finally:
@@ -352,7 +352,25 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.METADATA: json.dumps(message_metadata, ensure_ascii=False),
                 SpanAttributes.SESSION_ID: trace_info.message_data.conversation_id,
             }
-            llm_attributes.update(self._construct_llm_attributes(trace_info.inputs))
+
+            if isinstance(trace_info.inputs, list):
+                for i, msg in enumerate(trace_info.inputs):
+                    if isinstance(msg, dict):
+                        llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.content"] = msg.get("text", "")
+                        llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.role"] = msg.get(
+                            "role", "user"
+                        )
+                        # todo: handle assistant and tool role messages, as they don't always
+                        # have a text field, but may have a tool_calls field instead
+                        # e.g. 'tool_calls': [{'id': '98af3a29-b066-45a5-b4b1-46c74ddafc58',
+                        # 'type': 'function', 'function': {'name': 'current_time', 'arguments': '{}'}}]}
+            elif isinstance(trace_info.inputs, dict):
+                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = json.dumps(trace_info.inputs)
+                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
+            elif isinstance(trace_info.inputs, str):
+                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = trace_info.inputs
+                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
+
             if trace_info.total_tokens is not None and trace_info.total_tokens > 0:
                 llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_TOTAL] = trace_info.total_tokens
             if trace_info.message_tokens is not None and trace_info.message_tokens > 0:
@@ -706,24 +724,3 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             .all()
         )
         return workflow_nodes
-
-    def _construct_llm_attributes(self, prompts: dict | list | str | None) -> dict[str, str]:
-        """Helper method to construct LLM attributes with passed prompts."""
-        attributes = {}
-        if isinstance(prompts, list):
-            for i, msg in enumerate(prompts):
-                if isinstance(msg, dict):
-                    attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.content"] = msg.get("text", "")
-                    attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.role"] = msg.get("role", "user")
-                    # todo: handle assistant and tool role messages, as they don't always
-                    # have a text field, but may have a tool_calls field instead
-                    # e.g. 'tool_calls': [{'id': '98af3a29-b066-45a5-b4b1-46c74ddafc58',
-                    # 'type': 'function', 'function': {'name': 'current_time', 'arguments': '{}'}}]}
-        elif isinstance(prompts, dict):
-            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = json.dumps(prompts)
-            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
-        elif isinstance(prompts, str):
-            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = prompts
-            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
-
-        return attributes

api/core/plugin/backwards_invocation/encrypt.py

@@ -1,20 +1,16 @@
-from core.helper.provider_cache import SingletonProviderCredentialsCache
 from core.plugin.entities.request import RequestInvokeEncrypt
-from core.tools.utils.encryption import create_provider_encrypter
+from core.tools.utils.configuration import ProviderConfigEncrypter
 from models.account import Tenant
 
 
 class PluginEncrypter:
     @classmethod
     def invoke_encrypt(cls, tenant: Tenant, payload: RequestInvokeEncrypt) -> dict:
-        encrypter, cache = create_provider_encrypter(
+        encrypter = ProviderConfigEncrypter(
             tenant_id=tenant.id,
             config=payload.config,
-            cache=SingletonProviderCredentialsCache(
-                tenant_id=tenant.id,
-                provider_type=payload.namespace,
-                provider_identity=payload.identity,
-            ),
+            provider_type=payload.namespace,
+            provider_identity=payload.identity,
         )
 
         if payload.opt == "encrypt":
@@ -26,7 +22,7 @@ class PluginEncrypter:
                 "data": encrypter.decrypt(payload.data),
             }
         elif payload.opt == "clear":
-            cache.delete()
+            encrypter.delete_tool_credentials_cache()
             return {
                 "data": {},
             }

api/core/plugin/backwards_invocation/tool.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any, Optional
+from typing import Any
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
@@ -23,7 +23,6 @@ class PluginToolBackwardsInvocation(BaseBackwardsInvocation):
         provider: str,
         tool_name: str,
         tool_parameters: dict[str, Any],
-        credential_id: Optional[str] = None,
     ) -> Generator[ToolInvokeMessage, None, None]:
         """
         invoke tool
@@ -31,7 +30,7 @@ class PluginToolBackwardsInvocation(BaseBackwardsInvocation):
         # get tool runtime
         try:
             tool_runtime = ToolManager.get_tool_runtime_from_plugin(
-                tool_type, tenant_id, provider, tool_name, tool_parameters, credential_id
+                tool_type, tenant_id, provider, tool_name, tool_parameters
             )
             response = ToolEngine.generic_invoke(
                 tool_runtime, tool_parameters, user_id, DifyWorkflowCallbackHandler(), workflow_call_depth=1

api/core/plugin/entities/marketplace.py

@@ -32,13 +32,6 @@ class MarketplacePluginDeclaration(BaseModel):
     latest_package_identifier: str = Field(
         ..., description="Unique identifier for the latest package release of the plugin"
     )
-    status: str = Field(..., description="Indicate the status of marketplace plugin, enum from `active` `deleted`")
-    deprecated_reason: str = Field(
-        ..., description="Not empty when status='deleted', indicates the reason why this plugin is deleted(deprecated)"
-    )
-    alternative_plugin_id: str = Field(
-        ..., description="Optional, indicates the alternative plugin for user to switch to"
-    )
 
     @model_validator(mode="before")
     @classmethod

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/entities/plugin_daemon.py

@@ -182,10 +182,6 @@ class PluginOAuthAuthorizationUrlResponse(BaseModel):
 
 
 class PluginOAuthCredentialsResponse(BaseModel):
-    metadata: Mapping[str, Any] = Field(
-        default_factory=dict, description="The metadata of the OAuth, like avatar url, name, etc."
-    )
-    expires_at: int = Field(default=-1, description="The expires at time of the credentials. UTC timestamp.")
     credentials: Mapping[str, Any] = Field(description="The credentials of the OAuth.")
 
 

api/core/plugin/entities/request.py

@@ -27,20 +27,6 @@ from core.workflow.nodes.question_classifier.entities import (
 )
 
 
-class InvokeCredentials(BaseModel):
-    tool_credentials: dict[str, str] = Field(
-        default_factory=dict,
-        description="Map of tool provider to credential id, used to store the credential id for the tool provider.",
-    )
-
-
-class PluginInvokeContext(BaseModel):
-    credentials: Optional[InvokeCredentials] = Field(
-        default_factory=InvokeCredentials,
-        description="Credentials context for the plugin invocation or backward invocation.",
-    )
-
-
 class RequestInvokeTool(BaseModel):
     """
     Request to invoke a tool
@@ -50,7 +36,6 @@ class RequestInvokeTool(BaseModel):
     provider: str
     tool: str
     tool_parameters: dict
-    credential_id: Optional[str] = None
 
 
 class BaseRequestInvokeModel(BaseModel):

api/core/plugin/impl/agent.py

@@ -6,7 +6,6 @@ from core.plugin.entities.plugin import GenericProviderID
 from core.plugin.entities.plugin_daemon import (
     PluginAgentProviderEntity,
 )
-from core.plugin.entities.request import PluginInvokeContext
 from core.plugin.impl.base import BasePluginClient
 
 
@@ -84,7 +83,6 @@ class PluginAgentClient(BasePluginClient):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
-        context: Optional[PluginInvokeContext] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent with the given tenant, user, plugin, provider, name and parameters.
@@ -101,7 +99,6 @@ class PluginAgentClient(BasePluginClient):
                 "conversation_id": conversation_id,
                 "app_id": app_id,
                 "message_id": message_id,
-                "context": context.model_dump() if context else {},
                 "data": {
                     "agent_strategy_provider": agent_provider_id.provider_name,
                     "agent_strategy": agent_strategy,

api/core/plugin/impl/oauth.py

@@ -15,32 +15,27 @@ class OAuthHandler(BasePluginClient):
         user_id: str,
         plugin_id: str,
         provider: str,
-        redirect_uri: str,
         system_credentials: Mapping[str, Any],
     ) -> PluginOAuthAuthorizationUrlResponse:
-        try:
-            response = self._request_with_plugin_daemon_response_stream(
-                "POST",
-                f"plugin/{tenant_id}/dispatch/oauth/get_authorization_url",
-                PluginOAuthAuthorizationUrlResponse,
-                data={
-                    "user_id": user_id,
-                    "data": {
-                        "provider": provider,
-                        "redirect_uri": redirect_uri,
-                        "system_credentials": system_credentials,
-                    },
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/oauth/get_authorization_url",
+            PluginOAuthAuthorizationUrlResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider,
+                    "system_credentials": system_credentials,
                 },
-                headers={
-                    "X-Plugin-ID": plugin_id,
-                    "Content-Type": "application/json",
-                },
-            )
-            for resp in response:
-                return resp
-            raise ValueError("No response received from plugin daemon for authorization URL request.")
-        except Exception as e:
-            raise ValueError(f"Error getting authorization URL: {e}")
+            },
+            headers={
+                "X-Plugin-ID": plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+        for resp in response:
+            return resp
+        raise ValueError("No response received from plugin daemon for authorization URL request.")
 
     def get_credentials(
         self,
@@ -48,7 +43,6 @@ class OAuthHandler(BasePluginClient):
         user_id: str,
         plugin_id: str,
         provider: str,
-        redirect_uri: str,
         system_credentials: Mapping[str, Any],
         request: Request,
     ) -> PluginOAuthCredentialsResponse:
@@ -56,68 +50,30 @@ 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)
-            response = self._request_with_plugin_daemon_response_stream(
-                "POST",
-                f"plugin/{tenant_id}/dispatch/oauth/get_credentials",
-                PluginOAuthCredentialsResponse,
-                data={
-                    "user_id": user_id,
-                    "data": {
-                        "provider": provider,
-                        "redirect_uri": redirect_uri,
-                        "system_credentials": system_credentials,
-                        # for json serialization
-                        "raw_http_request": binascii.hexlify(raw_request_bytes).decode(),
-                    },
-                },
-                headers={
-                    "X-Plugin-ID": plugin_id,
-                    "Content-Type": "application/json",
-                },
-            )
-            for resp in response:
-                return resp
-            raise ValueError("No response received from plugin daemon for authorization URL request.")
-        except Exception as e:
-            raise ValueError(f"Error getting credentials: {e}")
+        # encode request to raw http request
+        raw_request_bytes = self._convert_request_to_raw_data(request)
 
-    def refresh_credentials(
-        self,
-        tenant_id: str,
-        user_id: str,
-        plugin_id: str,
-        provider: str,
-        redirect_uri: str,
-        system_credentials: Mapping[str, Any],
-        credentials: Mapping[str, Any],
-    ) -> PluginOAuthCredentialsResponse:
-        try:
-            response = self._request_with_plugin_daemon_response_stream(
-                "POST",
-                f"plugin/{tenant_id}/dispatch/oauth/refresh_credentials",
-                PluginOAuthCredentialsResponse,
-                data={
-                    "user_id": user_id,
-                    "data": {
-                        "provider": provider,
-                        "redirect_uri": redirect_uri,
-                        "system_credentials": system_credentials,
-                        "credentials": credentials,
-                    },
+        response = self._request_with_plugin_daemon_response_stream(
+            "POST",
+            f"plugin/{tenant_id}/dispatch/oauth/get_credentials",
+            PluginOAuthCredentialsResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider,
+                    "system_credentials": system_credentials,
+                    # for json serialization
+                    "raw_http_request": binascii.hexlify(raw_request_bytes).decode(),
                 },
-                headers={
-                    "X-Plugin-ID": plugin_id,
-                    "Content-Type": "application/json",
-                },
-            )
-            for resp in response:
-                return resp
-            raise ValueError("No response received from plugin daemon for refresh credentials request.")
-        except Exception as e:
-            raise ValueError(f"Error refreshing credentials: {e}")
+            },
+            headers={
+                "X-Plugin-ID": plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+        for resp in response:
+            return resp
+        raise ValueError("No response received from plugin daemon for authorization URL request.")
 
     def _convert_request_to_raw_data(self, request: Request) -> bytes:
         """

api/core/plugin/impl/tool.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
 from core.plugin.entities.plugin_daemon import PluginBasicBooleanResponse, PluginToolProviderEntity
 from core.plugin.impl.base import BasePluginClient
-from core.tools.entities.tool_entities import CredentialType, ToolInvokeMessage, ToolParameter
+from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 
 
 class PluginToolManager(BasePluginClient):
@@ -78,7 +78,6 @@ class PluginToolManager(BasePluginClient):
         tool_provider: str,
         tool_name: str,
         credentials: dict[str, Any],
-        credential_type: CredentialType,
         tool_parameters: dict[str, Any],
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
@@ -103,7 +102,6 @@ class PluginToolManager(BasePluginClient):
                     "provider": tool_provider_id.provider_name,
                     "tool": tool_name,
                     "credentials": credentials,
-                    "credential_type": credential_type,
                     "tool_parameters": tool_parameters,
                 },
             },

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/prompt/simple_prompt_transform.py

@@ -29,6 +29,19 @@ class ModelMode(enum.StrEnum):
     COMPLETION = "completion"
     CHAT = "chat"
 
+    @classmethod
+    def value_of(cls, value: str) -> "ModelMode":
+        """
+        Get value of given mode.
+
+        :param value: mode value
+        :return: mode
+        """
+        for mode in cls:
+            if mode.value == value:
+                return mode
+        raise ValueError(f"invalid mode value {value}")
+
 
 prompt_file_contents: dict[str, Any] = {}
 
@@ -52,7 +65,7 @@ class SimplePromptTransform(PromptTransform):
     ) -> tuple[list[PromptMessage], Optional[list[str]]]:
         inputs = {key: str(value) for key, value in inputs.items()}
 
-        model_mode = ModelMode(model_config.mode)
+        model_mode = ModelMode.value_of(model_config.mode)
         if model_mode == ModelMode.CHAT:
             prompt_messages, stops = self._get_chat_model_prompt_messages(
                 app_mode=app_mode,

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/analyticdb/analyticdb_vector_openapi.py

@@ -233,12 +233,6 @@ class AnalyticdbVectorOpenAPI:
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
         from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
 
-        document_ids_filter = kwargs.get("document_ids_filter")
-        where_clause = ""
-        if document_ids_filter:
-            document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
-            where_clause += f"metadata_->>'document_id' IN ({document_ids})"
-
         score_threshold = kwargs.get("score_threshold") or 0.0
         request = gpdb_20160503_models.QueryCollectionDataRequest(
             dbinstance_id=self.config.instance_id,
@@ -251,7 +245,7 @@ class AnalyticdbVectorOpenAPI:
             vector=query_vector,
             content=None,
             top_k=kwargs.get("top_k", 4),
-            filter=where_clause,
+            filter=None,
         )
         response = self._client.query_collection_data(request)
         documents = []
@@ -271,11 +265,6 @@ class AnalyticdbVectorOpenAPI:
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
 
-        document_ids_filter = kwargs.get("document_ids_filter")
-        where_clause = ""
-        if document_ids_filter:
-            document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
-            where_clause += f"metadata_->>'document_id' IN ({document_ids})"
         score_threshold = float(kwargs.get("score_threshold") or 0.0)
         request = gpdb_20160503_models.QueryCollectionDataRequest(
             dbinstance_id=self.config.instance_id,
@@ -288,7 +277,7 @@ class AnalyticdbVectorOpenAPI:
             vector=None,
             content=query,
             top_k=kwargs.get("top_k", 4),
-            filter=where_clause,
+            filter=None,
         )
         response = self._client.query_collection_data(request)
         documents = []

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py

@@ -147,17 +147,10 @@ class ElasticSearchVector(BaseVector):
         return docs
 
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        query_str: dict[str, Any] = {"match": {Field.CONTENT_KEY.value: query}}
+        query_str = {"match": {Field.CONTENT_KEY.value: query}}
         document_ids_filter = kwargs.get("document_ids_filter")
-
         if document_ids_filter:
-            query_str = {
-                "bool": {
-                    "must": {"match": {Field.CONTENT_KEY.value: query}},
-                    "filter": {"terms": {"metadata.document_id": document_ids_filter}},
-                }
-            }
-
+            query_str["filter"] = {"terms": {"metadata.document_id": document_ids_filter}}  # type: ignore
         results = self._client.search(index=self._collection_name, query=query_str, size=kwargs.get("top_k", 4))
         docs = []
         for hit in results["hits"]["hits"]:

api/core/rag/datasource/vdb/tencent/tencent_vector.py

@@ -206,19 +206,9 @@ class TencentVector(BaseVector):
     def delete_by_ids(self, ids: list[str]) -> None:
         if not ids:
             return
-
-        total_count = len(ids)
-        batch_size = self._client_config.max_upsert_batch_size
-        batch = math.ceil(total_count / batch_size)
-
-        for j in range(batch):
-            start_idx = j * batch_size
-            end_idx = min(total_count, (j + 1) * batch_size)
-            batch_ids = ids[start_idx:end_idx]
-
-            self._client.delete(
-                database_name=self._client_config.database, collection_name=self.collection_name, document_ids=batch_ids
-            )
+        self._client.delete(
+            database_name=self._client_config.database, collection_name=self.collection_name, document_ids=ids
+        )
 
     def delete_by_metadata_field(self, key: str, value: str) -> None:
         self._client.delete(

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
+        """