feat: variable preview

Co-authored-by: Copilot <[email protected]>
Co-authored-by: crazywoola <[email protected]>

.devcontainer/post_create_command.sh

@@ -5,7 +5,7 @@ cd web && pnpm install
 pipx install uv
 
 echo 'alias start-api="cd /workspaces/dify/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
-echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
+echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage"' >> ~/.bashrc
 echo 'alias start-web="cd /workspaces/dify/web && pnpm dev"' >> ~/.bashrc
 echo 'alias start-web-prod="cd /workspaces/dify/web && pnpm build && pnpm start"' >> ~/.bashrc
 echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d"' >> ~/.bashrc

.github/ISSUE_TEMPLATE/refactor.yml

@@ -0,0 +1,44 @@
+name: "✨ Refactor"
+description: Refactor existing code for improved readability and maintainability.
+title: "[Chore/Refactor] "
+labels:
+  - refactor
+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 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 refactoring, 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.
+          required: true
+        - label: 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
+          required: true
+        - label: "Please do not modify this template :) and fill in all the required fields."
+          required: true
+  - type: textarea
+    id: description
+    attributes:
+      label: Description
+      placeholder: "Describe the refactor you are proposing."
+    validations:
+      required: true
+  - type: textarea
+    id: motivation
+    attributes:
+      label: Motivation
+      placeholder: "Explain why this refactor is necessary."
+    validations:
+      required: false
+  - type: textarea
+    id: additional-context
+    attributes:
+      label: Additional Context
+      placeholder: "Add any other context or screenshots about the request here."
+    validations:
+      required: false

.github/actions/setup-uv/action.yml

@@ -8,7 +8,7 @@ inputs:
   uv-version:
     description: UV version to set up
     required: true
-    default: '~=0.7.11'
+    default: '0.8.9'
   uv-lockfile:
     description: Path to the UV lockfile to restore cache from
     required: true
@@ -26,7 +26,7 @@ runs:
         python-version: ${{ inputs.python-version }}
 
     - name: Install uv
-      uses: astral-sh/setup-uv@v5
+      uses: astral-sh/setup-uv@v6
       with:
         version: ${{ inputs.uv-version }}
         python-version: ${{ inputs.python-version }}

.github/workflows/api-tests.yml

@@ -99,3 +99,6 @@ jobs:
 
       - name: Run Tool
         run: uv run --project api bash dev/pytest/pytest_tools.sh
+
+      - name: Run TestContainers
+        run: uv run --project api bash dev/pytest/pytest_testcontainers.sh

.github/workflows/autofix.yml

@@ -9,6 +9,7 @@ permissions:
 
 jobs:
   autofix:
+    if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

.github/workflows/build-push.yml

@@ -7,6 +7,8 @@ on:
       - "deploy/dev"
       - "deploy/enterprise"
       - "build/**"
+      - "release/e-*"
+      - "deploy/rag-dev"
     tags:
       - "*"
 

.github/workflows/deploy-dev.yml

@@ -4,7 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/dev"
+      - "deploy/rag-dev"
     types:
       - completed
 
@@ -12,12 +12,13 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     if: |
-      github.event.workflow_run.conclusion == 'success'
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_branch == 'deploy/rag-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/[email protected]
         with:
-          host: ${{ secrets.SSH_HOST }}
+          host: ${{ secrets.RAG_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/style.yml

@@ -82,7 +82,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Setup NodeJS
@@ -95,10 +95,12 @@ jobs:
 
       - name: Web dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm run lint
 
   docker-compose-template:

.github/workflows/translate-i18n-base-on-english.yml

@@ -1,13 +1,18 @@
 name: Check i18n Files and Create PR
 
 on:
-  pull_request:
-    types: [closed]
+  push:
     branches: [main]
+    paths:
+      - 'web/i18n/en-US/*.ts'
+
+permissions:
+  contents: write
+  pull-requests: write
 
 jobs:
   check-and-update:
-    if: github.event.pull_request.merged == true
+    if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     defaults:
       run:
@@ -15,8 +20,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          fetch-depth: 2 # last 2 commits
-          persist-credentials: false
+          fetch-depth: 2
+          token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Check for file changes in i18n/en-US
         id: check_files
@@ -27,6 +32,13 @@ jobs:
           echo "Changed files: $changed_files"
           if [ -n "$changed_files" ]; then
             echo "FILES_CHANGED=true" >> $GITHUB_ENV
+            file_args=""
+            for file in $changed_files; do
+              filename=$(basename "$file" .ts)
+              file_args="$file_args --file=$filename"
+            done
+            echo "FILE_ARGS=$file_args" >> $GITHUB_ENV
+            echo "File arguments: $file_args"
           else
             echo "FILES_CHANGED=false" >> $GITHUB_ENV
           fi
@@ -34,7 +46,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Set up Node.js
@@ -47,16 +59,19 @@ jobs:
 
       - name: Install dependencies
         if: env.FILES_CHANGED == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
-      - name: Run npm script
+      - name: Generate i18n translations
         if: env.FILES_CHANGED == 'true'
-        run: pnpm run auto-gen-i18n
+        working-directory: ./web
+        run: pnpm run auto-gen-i18n ${{ env.FILE_ARGS }}
 
       - name: Create Pull Request
         if: env.FILES_CHANGED == 'true'
         uses: peter-evans/create-pull-request@v6
         with:
+          token: ${{ secrets.GITHUB_TOKEN }}
           commit-message: Update i18n files based on en-US changes
           title: 'chore: translate i18n files'
           body: This PR was automatically created to update i18n files based on changes in en-US locale.

.github/workflows/web-tests.yml

@@ -35,7 +35,7 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Setup Node.js
@@ -48,8 +48,10 @@ jobs:
 
       - name: Install dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm test

.gitignore

@@ -215,3 +215,4 @@ mise.toml
 # AI Assistant
 .roo/
 api/.env.backup
+/clickzetta

CLAUDE.md

@@ -0,0 +1,83 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+Dify is an open-source platform for developing LLM applications with an intuitive interface combining agentic AI workflows, RAG pipelines, agent capabilities, and model management.
+
+The codebase consists of:
+- **Backend API** (`/api`): Python Flask application with Domain-Driven Design architecture
+- **Frontend Web** (`/web`): Next.js 15 application with TypeScript and React 19
+- **Docker deployment** (`/docker`): Containerized deployment configurations
+
+## Development Commands
+
+### Backend (API)
+
+All Python commands must be prefixed with `uv run --project api`:
+
+```bash
+# Start development servers
+./dev/start-api                   # Start API server
+./dev/start-worker                # Start Celery worker
+
+# Run tests
+uv run --project api pytest      # Run all tests
+uv run --project api pytest tests/unit_tests/     # Unit tests only
+uv run --project api pytest tests/integration_tests/  # Integration tests
+
+# Code quality
+./dev/reformat                    # Run all formatters and linters
+uv run --project api ruff check --fix ./    # Fix linting issues
+uv run --project api ruff format ./         # Format code
+uv run --project api mypy .                 # Type checking
+```
+
+### Frontend (Web)
+
+```bash
+cd web
+pnpm lint                         # Run ESLint
+pnpm eslint-fix                   # Fix ESLint issues
+pnpm test                         # Run Jest tests
+```
+
+## Testing Guidelines
+
+### Backend Testing
+- Use `pytest` for all backend tests
+- Write tests first (TDD approach)
+- Test structure: Arrange-Act-Assert
+
+## Code Style Requirements
+
+### Python
+- Use type hints for all functions and class attributes
+- No `Any` types unless absolutely necessary
+- Implement special methods (`__repr__`, `__str__`) appropriately
+
+### TypeScript/JavaScript  
+- Strict TypeScript configuration
+- ESLint with Prettier integration
+- Avoid `any` type
+
+## Important Notes
+
+- **Environment Variables**: Always use UV for Python commands: `uv run --project api <command>`
+- **Comments**: Only write meaningful comments that explain "why", not "what"
+- **File Creation**: Always prefer editing existing files over creating new ones
+- **Documentation**: Don't create documentation files unless explicitly requested
+- **Code Quality**: Always run `./dev/reformat` before committing backend changes
+
+## Common Development Tasks
+
+### Adding a New API Endpoint
+1. Create controller in `/api/controllers/`
+2. Add service logic in `/api/services/`
+3. Update routes in controller's `__init__.py`
+4. Write tests in `/api/tests/`
+
+## Project-Specific Conventions
+
+- All async tasks use Celery with Redis as broker

README.md

@@ -225,7 +225,8 @@ Deploy Dify to AWS with [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Using Alibaba Cloud Computing Nest
 

README_AR.md

@@ -208,7 +208,8 @@ docker compose up -d
 
 ##### AWS
 
-- [AWS CDK بواسطة @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK بواسطة @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK بواسطة @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### استخدام Alibaba Cloud للنشر
  [بسرعة نشر Dify إلى سحابة علي بابا مع عش الحوسبة السحابية علي بابا](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_BN.md

@@ -225,7 +225,8 @@ GitHub-এ ডিফাইকে স্টার দিয়ে রাখুন
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud ব্যবহার করে ডিপ্লয়
 

README_CN.md

@@ -223,7 +223,8 @@ docker compose up -d
 使用 [CDK](https://aws.amazon.com/cdk/) 将 Dify 部署到 AWS
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### 使用 阿里云计算巢 部署
 

README_DE.md

@@ -220,7 +220,8 @@ Stellen Sie Dify mit nur einem Klick mithilfe von [terraform](https://www.terraf
 Bereitstellung von Dify auf AWS mit [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud 
 

README_ES.md

@@ -220,7 +220,8 @@ Despliega Dify en una plataforma en la nube con un solo clic utilizando [terrafo
 Despliegue Dify en AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_FR.md

@@ -218,7 +218,8 @@ Déployez Dify sur une plateforme cloud en un clic en utilisant [terraform](http
 Déployez Dify sur AWS en utilisant [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK par @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK par @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK par @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_JA.md

@@ -219,7 +219,8 @@ docker compose up -d
 [CDK](https://aws.amazon.com/cdk/) を使用して、DifyをAWSにデプロイします
 
 ##### AWS
-- [@KevinZhaoによるAWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [@KevinZhaoによるAWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [@tmokmssによるAWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_KL.md

@@ -218,7 +218,8 @@ wa'logh nIqHom neH ghun deployment toy'wI' [terraform](https://www.terraform.io/
 wa'logh nIqHom neH ghun deployment toy'wI' [CDK](https://aws.amazon.com/cdk/) lo'laH.
 
 ##### AWS 
-- [AWS CDK qachlot @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK qachlot @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK qachlot @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_KR.md

@@ -212,7 +212,8 @@ Dify를 Kubernetes에 배포하고 프리미엄 스케일링 설정을 구성했
 [CDK](https://aws.amazon.com/cdk/)를 사용하여 AWS에 Dify 배포
 
 ##### AWS 
-- [KevinZhao의 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [KevinZhao의 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [tmokmss의 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_PT.md

@@ -217,7 +217,8 @@ Implante o Dify na Plataforma Cloud com um único clique usando [terraform](http
 Implante o Dify na AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_SI.md

@@ -218,7 +218,8 @@ namestite Dify v Cloud Platform z enim klikom z uporabo [terraform](https://www.
 Uvedite Dify v AWS z uporabo [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_TR.md

@@ -211,7 +211,8 @@ Dify'ı bulut platformuna tek tıklamayla dağıtın [terraform](https://www.ter
 [CDK](https://aws.amazon.com/cdk/) kullanarak Dify'ı AWS'ye dağıtın
 
 ##### AWS 
-- [AWS CDK tarafından @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK tarafından @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK tarafından @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_TW.md

@@ -223,7 +223,8 @@ Dify 的所有功能都提供相應的 API，因此您可以輕鬆地將 Dify
 
 ### AWS
 
-- [由 @KevinZhao 提供的 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [由 @KevinZhao 提供的 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [由 @tmokmss 提供的 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### 使用 阿里云计算巢進行部署
 

README_VI.md

@@ -213,7 +213,8 @@ Triển khai Dify lên nền tảng đám mây với một cú nhấp chuột b
 Triển khai Dify trên AWS bằng [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK bởi @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK bởi @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK bởi @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 
 #### Alibaba Cloud

api/.env.example

@@ -42,6 +42,15 @@ REDIS_PORT=6379
 REDIS_USERNAME=
 REDIS_PASSWORD=difyai123456
 REDIS_USE_SSL=false
+# SSL configuration for Redis (when REDIS_USE_SSL=true)
+REDIS_SSL_CERT_REQS=CERT_NONE
+# Options: CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
+REDIS_SSL_CA_CERTS=
+# Path to CA certificate file for SSL verification
+REDIS_SSL_CERTFILE=
+# Path to client certificate file for SSL authentication
+REDIS_SSL_KEYFILE=
+# Path to client private key file for SSL authentication
 REDIS_DB=0
 
 # redis Sentinel configuration.

api/Dockerfile

@@ -4,7 +4,7 @@ FROM python:3.12-slim-bookworm AS base
 WORKDIR /app/api
 
 # Install uv
-ENV UV_VERSION=0.7.11
+ENV UV_VERSION=0.8.9
 
 RUN pip install --no-cache-dir uv==${UV_VERSION}
 
@@ -19,7 +19,7 @@ RUN apt-get update \
 
 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
-RUN uv sync --locked
+RUN uv sync --locked --no-dev
 
 # production stage
 FROM base AS production

api/README.md

@@ -74,7 +74,7 @@
 10. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.
 
    ```bash
-   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin
+   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage
    ```
 
    Addition, if you want to debug the celery scheduled tasks, you can use the following command in another terminal:

api/app.py

@@ -1,4 +1,3 @@
-import os
 import sys
 
 
@@ -17,20 +16,20 @@ else:
     # It seems that JetBrains Python debugger does not work well with gevent,
     # so we need to disable gevent in debug mode.
     # If you are using debugpy and set GEVENT_SUPPORT=True, you can debug with gevent.
-    if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
-        from gevent import monkey
+    # if (flask_debug := os.environ.get("FLASK_DEBUG", "0")) and flask_debug.lower() in {"false", "0", "no"}:
+    # from gevent import monkey
+    #
+    # # gevent
+    # monkey.patch_all()
+    #
+    # from grpc.experimental import gevent as grpc_gevent  # type: ignore
+    #
+    # # grpc gevent
+    # grpc_gevent.init_gevent()
 
-        # gevent
-        monkey.patch_all()
-
-        from grpc.experimental import gevent as grpc_gevent  # type: ignore
-
-        # grpc gevent
-        grpc_gevent.init_gevent()
-
-        import psycogreen.gevent  # type: ignore
-
-        psycogreen.gevent.patch_psycopg()
+    # import psycogreen.gevent  # type: ignore
+    #
+    # psycogreen.gevent.patch_psycopg()
 
     from app_factory import create_app
 

api/app_factory.py

@@ -51,6 +51,7 @@ def initialize_extensions(app: DifyApp):
         ext_login,
         ext_mail,
         ext_migrate,
+        ext_orjson,
         ext_otel,
         ext_proxy_fix,
         ext_redis,
@@ -67,6 +68,7 @@ def initialize_extensions(app: DifyApp):
         ext_logging,
         ext_warnings,
         ext_import_modules,
+        ext_orjson,
         ext_set_secretkey,
         ext_compress,
         ext_code_based_extension,

api/commands.py

@@ -5,18 +5,22 @@ import secrets
 from typing import Any, Optional
 
 import click
+import sqlalchemy as sa
 from flask import current_app
 from pydantic import TypeAdapter
 from sqlalchemy import select
-from werkzeug.exceptions import NotFound
+from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
-from core.plugin.entities.plugin import ToolProviderID
+from core.helper import encrypter
+from core.plugin.entities.plugin import DatasourceProviderID, PluginInstallationSource, ToolProviderID
+from core.plugin.impl.plugin import PluginInstaller
 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.entities.tool_entities import CredentialType
 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
@@ -29,12 +33,15 @@ from models import Tenant
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
+from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
+from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 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
 from services.plugin.plugin_migration import PluginMigration
+from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 
 @click.command("reset-password", help="Reset the account password.")
@@ -180,8 +187,8 @@ def migrate_annotation_vector_database():
             )
             if not apps:
                 break
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
 
         page += 1
         for app in apps:
@@ -307,8 +314,8 @@ def migrate_knowledge_vector_database():
             )
 
             datasets = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
 
         page += 1
         for dataset in datasets:
@@ -457,7 +464,7 @@ def convert_to_agent_apps():
         """
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query))
+            rs = conn.execute(sa.text(sql_query))
 
             apps = []
             for i in rs:
@@ -560,8 +567,8 @@ def old_metadata_migration():
                 .order_by(DatasetDocument.created_at.desc())
             )
             documents = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
         if not documents:
             break
         for document in documents:
@@ -702,7 +709,7 @@ def fix_app_site_missing():
         sql = """select apps.id as id from apps left join sites on sites.app_id=apps.id
 where sites.id is null limit 1000"""
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql))
+            rs = conn.execute(sa.text(sql))
 
             processed_count = 0
             for i in rs:
@@ -916,7 +923,7 @@ def clear_orphaned_file_records(force: bool):
         )
         orphaned_message_files = []
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(query))
+            rs = conn.execute(sa.text(query))
             for i in rs:
                 orphaned_message_files.append({"id": str(i[0]), "message_id": str(i[1])})
 
@@ -937,7 +944,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style("- Deleting orphaned message_files records", fg="white"))
             query = "DELETE FROM message_files WHERE id IN :ids"
             with db.engine.begin() as conn:
-                conn.execute(db.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
+                conn.execute(sa.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
             click.echo(
                 click.style(f"Removed {len(orphaned_message_files)} orphaned message_files records.", fg="green")
             )
@@ -954,7 +961,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style(f"- Listing file records in table {files_table['table']}", fg="white"))
             query = f"SELECT {files_table['id_column']}, {files_table['key_column']} FROM {files_table['table']}"
             with db.engine.begin() as conn:
-                rs = conn.execute(db.text(query))
+                rs = conn.execute(sa.text(query))
             for i in rs:
                 all_files_in_tables.append({"table": files_table["table"], "id": str(i[0]), "key": i[1]})
         click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
@@ -974,7 +981,7 @@ def clear_orphaned_file_records(force: bool):
                     f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
             elif ids_table["type"] == "text":
@@ -989,7 +996,7 @@ def clear_orphaned_file_records(force: bool):
                     f"FROM {ids_table['table']}"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     for j in i[0]:
                         all_ids_in_tables.append({"table": ids_table["table"], "id": j})
@@ -1008,7 +1015,7 @@ def clear_orphaned_file_records(force: bool):
                     f"FROM {ids_table['table']}"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     for j in i[0]:
                         all_ids_in_tables.append({"table": ids_table["table"], "id": j})
@@ -1037,7 +1044,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style(f"- Deleting orphaned file records in table {files_table['table']}", fg="white"))
             query = f"DELETE FROM {files_table['table']} WHERE {files_table['id_column']} IN :ids"
             with db.engine.begin() as conn:
-                conn.execute(db.text(query), {"ids": tuple(orphaned_files)})
+                conn.execute(sa.text(query), {"ids": tuple(orphaned_files)})
     except Exception as e:
         click.echo(click.style(f"Error deleting orphaned file records: {str(e)}", fg="red"))
         return
@@ -1107,7 +1114,7 @@ def remove_orphaned_files_on_storage(force: bool):
             click.echo(click.style(f"- Listing files from table {files_table['table']}", fg="white"))
             query = f"SELECT {files_table['key_column']} FROM {files_table['table']}"
             with db.engine.begin() as conn:
-                rs = conn.execute(db.text(query))
+                rs = conn.execute(sa.text(query))
             for i in rs:
                 all_files_in_tables.append(str(i[0]))
         click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
@@ -1201,3 +1208,363 @@ def setup_system_tool_oauth_client(provider, 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"))
+
+
+def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
+    """
+    Find draft variables that reference non-existent apps.
+
+    Args:
+        batch_size: Maximum number of orphaned app IDs to return
+
+    Returns:
+        List of app IDs that have draft variables but don't exist in the apps table
+    """
+    query = """
+        SELECT DISTINCT wdv.app_id
+        FROM workflow_draft_variables AS wdv
+        WHERE NOT EXISTS(
+            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
+        )
+        LIMIT :batch_size
+    """
+
+    with db.engine.connect() as conn:
+        result = conn.execute(sa.text(query), {"batch_size": batch_size})
+        return [row[0] for row in result]
+
+
+def _count_orphaned_draft_variables() -> dict[str, Any]:
+    """
+    Count orphaned draft variables by app.
+
+    Returns:
+        Dictionary with statistics about orphaned variables
+    """
+    query = """
+        SELECT
+            wdv.app_id,
+            COUNT(*) as variable_count
+        FROM workflow_draft_variables AS wdv
+        WHERE NOT EXISTS(
+            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
+        )
+        GROUP BY wdv.app_id
+        ORDER BY variable_count DESC
+    """
+
+    with db.engine.connect() as conn:
+        result = conn.execute(sa.text(query))
+        orphaned_by_app = {row[0]: row[1] for row in result}
+
+        total_orphaned = sum(orphaned_by_app.values())
+        app_count = len(orphaned_by_app)
+
+        return {
+            "total_orphaned_variables": total_orphaned,
+            "orphaned_app_count": app_count,
+            "orphaned_by_app": orphaned_by_app,
+        }
+
+
+@click.command()
+@click.option("--dry-run", is_flag=True, help="Show what would be deleted without actually deleting")
+@click.option("--batch-size", default=1000, help="Number of records to process per batch (default 1000)")
+@click.option("--max-apps", default=None, type=int, help="Maximum number of apps to process (default: no limit)")
+@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
+def cleanup_orphaned_draft_variables(
+    dry_run: bool,
+    batch_size: int,
+    max_apps: int | None,
+    force: bool = False,
+):
+    """
+    Clean up orphaned draft variables from the database.
+
+    This script finds and removes draft variables that belong to apps
+    that no longer exist in the database.
+    """
+    logger = logging.getLogger(__name__)
+
+    # Get statistics
+    stats = _count_orphaned_draft_variables()
+
+    logger.info("Found %s orphaned draft variables", stats["total_orphaned_variables"])
+    logger.info("Across %s non-existent apps", stats["orphaned_app_count"])
+
+    if stats["total_orphaned_variables"] == 0:
+        logger.info("No orphaned draft variables found. Exiting.")
+        return
+
+    if dry_run:
+        logger.info("DRY RUN: Would delete the following:")
+        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1], reverse=True)[
+            :10
+        ]:  # Show top 10
+            logger.info("  App %s: %s variables", app_id, count)
+        if len(stats["orphaned_by_app"]) > 10:
+            logger.info("  ... and %s more apps", len(stats["orphaned_by_app"]) - 10)
+        return
+
+    # Confirm deletion
+    if not force:
+        click.confirm(
+            f"Are you sure you want to delete {stats['total_orphaned_variables']} "
+            f"orphaned draft variables from {stats['orphaned_app_count']} apps?",
+            abort=True,
+        )
+
+    total_deleted = 0
+    processed_apps = 0
+
+    while True:
+        if max_apps and processed_apps >= max_apps:
+            logger.info("Reached maximum app limit (%s). Stopping.", max_apps)
+            break
+
+        orphaned_app_ids = _find_orphaned_draft_variables(batch_size=10)
+        if not orphaned_app_ids:
+            logger.info("No more orphaned draft variables found.")
+            break
+
+        for app_id in orphaned_app_ids:
+            if max_apps and processed_apps >= max_apps:
+                break
+
+            try:
+                deleted_count = delete_draft_variables_batch(app_id, batch_size)
+                total_deleted += deleted_count
+                processed_apps += 1
+
+                logger.info("Deleted %s variables for app %s", deleted_count, app_id)
+
+            except Exception:
+                logger.exception("Error processing app %s", app_id)
+                continue
+
+    logger.info("Cleanup completed. Total deleted: %s variables across %s apps", total_deleted, processed_apps)
+
+
+@click.command("setup-datasource-oauth-client", help="Setup datasource oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_datasource_oauth_client(provider, client_params):
+    """
+    Setup datasource oauth client
+    """
+    provider_id = DatasourceProviderID(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"))
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+
+    click.echo(click.style(f"Ready to delete existing oauth client params: {provider_name}", fg="yellow"))
+    deleted_count = (
+        db.session.query(DatasourceOauthParamConfig)
+        .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"))
+
+    click.echo(click.style(f"Ready to setup datasource oauth client: {provider_name}", fg="yellow"))
+    oauth_client = DatasourceOauthParamConfig(
+        provider=provider_name,
+        plugin_id=plugin_id,
+        system_credentials=client_params_dict,
+    )
+    db.session.add(oauth_client)
+    db.session.commit()
+    click.echo(click.style(f"provider: {provider_name}", fg="green"))
+    click.echo(click.style(f"plugin_id: {plugin_id}", fg="green"))
+    click.echo(click.style(f"params: {json.dumps(client_params_dict, indent=2, ensure_ascii=False)}", fg="green"))
+    click.echo(click.style(f"Datasource oauth client setup successfully. id: {oauth_client.id}", fg="green"))
+
+
+@click.command("transform-datasource-credentials", help="Transform datasource credentials.")
+def transform_datasource_credentials():
+    """
+    Transform datasource credentials
+    """
+    try:
+        installer_manager = PluginInstaller()
+        plugin_migration = PluginMigration()
+
+        notion_plugin_id = "langgenius/notion_datasource"
+        firecrawl_plugin_id = "langgenius/firecrawl_datasource"
+        jina_plugin_id = "langgenius/jina_datasource"
+        notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
+        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
+        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
+        oauth_credential_type = CredentialType.OAUTH2
+        api_key_credential_type = CredentialType.API_KEY
+
+        # deal notion credentials
+        deal_notion_count = 0
+        notion_credentials = db.session.query(DataSourceOauthBinding).filter_by(provider="notion").all()
+        notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
+        for credential in notion_credentials:
+            tenant_id = credential.tenant_id
+            if tenant_id not in notion_credentials_tenant_mapping:
+                notion_credentials_tenant_mapping[tenant_id] = []
+            notion_credentials_tenant_mapping[tenant_id].append(credential)
+        for tenant_id, credentials in notion_credentials_tenant_mapping.items():
+            # check notion plugin is installed
+            installed_plugins = installer_manager.list_plugins(tenant_id)
+            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+            if notion_plugin_id not in installed_plugins_ids:
+                if notion_plugin_unique_identifier:
+                    # install notion plugin
+                    installer_manager.install_from_identifiers(
+                        tenant_id,
+                        [notion_plugin_unique_identifier],
+                        PluginInstallationSource.Marketplace,
+                        metas=[
+                            {
+                                "plugin_unique_identifier": notion_plugin_unique_identifier,
+                            }
+                        ],
+                    )
+            auth_count = 0
+            for credential in credentials:
+                auth_count += 1
+                # get credential oauth params
+                access_token = credential.access_token
+                # notion info
+                notion_info = credential.source_info
+                workspace_id = notion_info.get("workspace_id")
+                workspace_name = notion_info.get("workspace_name")
+                workspace_icon = notion_info.get("workspace_icon")
+                new_credentials = {
+                    "integration_secret": encrypter.encrypt_token(tenant_id, access_token),
+                    "workspace_id": workspace_id,
+                    "workspace_name": workspace_name,
+                    "workspace_icon": workspace_icon,
+                }
+                datasource_provider = DatasourceProvider(
+                    provider="notion",
+                    tenant_id=tenant_id,
+                    plugin_id=notion_plugin_id,
+                    auth_type=oauth_credential_type.value,
+                    encrypted_credentials=new_credentials,
+                    name=f"Auth {auth_count}",
+                    avatar_url=workspace_icon or "default",
+                    is_default=False,
+                )
+                db.session.add(datasource_provider)
+                deal_notion_count += 1
+            db.session.commit()
+        # deal firecrawl credentials
+        deal_firecrawl_count = 0
+        firecrawl_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="firecrawl").all()
+        firecrawl_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+        for credential in firecrawl_credentials:
+            tenant_id = credential.tenant_id
+            if tenant_id not in firecrawl_credentials_tenant_mapping:
+                firecrawl_credentials_tenant_mapping[tenant_id] = []
+            firecrawl_credentials_tenant_mapping[tenant_id].append(credential)
+        for tenant_id, credentials in firecrawl_credentials_tenant_mapping.items():
+            # check firecrawl plugin is installed
+            installed_plugins = installer_manager.list_plugins(tenant_id)
+            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+            if firecrawl_plugin_id not in installed_plugins_ids:
+                if firecrawl_plugin_unique_identifier:
+                    # install firecrawl plugin
+                    installer_manager.install_from_identifiers(
+                        tenant_id,
+                        [firecrawl_plugin_unique_identifier],
+                        PluginInstallationSource.Marketplace,
+                        metas=[
+                            {
+                                "plugin_unique_identifier": firecrawl_plugin_unique_identifier,
+                            }
+                        ],
+                    )
+            auth_count = 0
+            for credential in credentials:
+                auth_count += 1
+                # get credential api key
+                api_key = credential.credentials.get("config", {}).get("api_key")
+                base_url = credential.credentials.get("config", {}).get("base_url")
+                new_credentials = {
+                    "firecrawl_api_key": api_key,
+                    "base_url": base_url,
+                }
+                datasource_provider = DatasourceProvider(
+                    provider="firecrawl",
+                    tenant_id=tenant_id,
+                    plugin_id=firecrawl_plugin_id,
+                    auth_type=api_key_credential_type.value,
+                    encrypted_credentials=new_credentials,
+                    name=f"Auth {auth_count}",
+                    avatar_url="default",
+                    is_default=False,
+                )
+                db.session.add(datasource_provider)
+                deal_firecrawl_count += 1
+            db.session.commit()
+        # deal jina credentials
+        deal_jina_count = 0
+        jina_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="jina").all()
+        jina_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+        for credential in jina_credentials:
+            tenant_id = credential.tenant_id
+            if tenant_id not in jina_credentials_tenant_mapping:
+                jina_credentials_tenant_mapping[tenant_id] = []
+            jina_credentials_tenant_mapping[tenant_id].append(credential)
+        for tenant_id, credentials in jina_credentials_tenant_mapping.items():
+            # check jina plugin is installed
+            installed_plugins = installer_manager.list_plugins(tenant_id)
+            installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+            if jina_plugin_id not in installed_plugins_ids:
+                if jina_plugin_unique_identifier:
+                    # install jina plugin
+                    installer_manager.install_from_identifiers(
+                        tenant_id,
+                        [jina_plugin_unique_identifier],
+                        PluginInstallationSource.Marketplace,
+                        metas=[
+                            {
+                                "plugin_unique_identifier": jina_plugin_unique_identifier,
+                            }
+                        ],
+                    )
+            auth_count = 0
+            for credential in credentials:
+                auth_count += 1
+                # get credential api key
+                api_key = credential.credentials.get("config", {}).get("api_key")
+                new_credentials = {
+                    "integration_secret": api_key,
+                }
+                datasource_provider = DatasourceProvider(
+                    provider="jina",
+                    tenant_id=tenant_id,
+                    plugin_id=jina_plugin_id,
+                    auth_type=api_key_credential_type.value,
+                    encrypted_credentials=new_credentials,
+                    name=f"Auth {auth_count}",
+                    avatar_url="default",
+                    is_default=False,
+                )
+                db.session.add(datasource_provider)
+                deal_jina_count += 1
+            db.session.commit()
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+    click.echo(click.style(f"Transforming notion successfully. deal_notion_count: {deal_notion_count}", fg="green"))
+    click.echo(
+        click.style(f"Transforming firecrawl successfully. deal_firecrawl_count: {deal_firecrawl_count}", fg="green")
+    )
+    click.echo(click.style(f"Transforming jina successfully. deal_jina_count: {deal_jina_count}", fg="green"))

api/configs/feature/__init__.py

@@ -330,17 +330,17 @@ class HttpConfig(BaseSettings):
     def WEB_API_CORS_ALLOW_ORIGINS(self) -> list[str]:
         return self.inner_WEB_API_CORS_ALLOW_ORIGINS.split(",")
 
-    HTTP_REQUEST_MAX_CONNECT_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=10, description="Maximum connection timeout in seconds for HTTP requests")
-    ] = 10
+    HTTP_REQUEST_MAX_CONNECT_TIMEOUT: int = Field(
+        ge=1, description="Maximum connection timeout in seconds for HTTP requests", default=10
+    )
 
-    HTTP_REQUEST_MAX_READ_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=60, description="Maximum read timeout in seconds for HTTP requests")
-    ] = 60
+    HTTP_REQUEST_MAX_READ_TIMEOUT: int = Field(
+        ge=1, description="Maximum read timeout in seconds for HTTP requests", default=60
+    )
 
-    HTTP_REQUEST_MAX_WRITE_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=10, description="Maximum write timeout in seconds for HTTP requests")
-    ] = 20
+    HTTP_REQUEST_MAX_WRITE_TIMEOUT: int = Field(
+        ge=1, description="Maximum write timeout in seconds for HTTP requests", default=20
+    )
 
     HTTP_REQUEST_NODE_MAX_BINARY_SIZE: PositiveInt = Field(
         description="Maximum allowed size in bytes for binary data in HTTP requests",
@@ -552,12 +552,18 @@ class RepositoryConfig(BaseSettings):
     """
 
     CORE_WORKFLOW_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowExecution. Specify as a module path",
+        description="Repository implementation for WorkflowExecution. Options: "
+        "'core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository' (default), "
+        "'core.repositories.celery_workflow_execution_repository.CeleryWorkflowExecutionRepository'",
         default="core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository",
     )
 
     CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowNodeExecution. Specify as a module path",
+        description="Repository implementation for WorkflowNodeExecution. Options: "
+        "'core.repositories.sqlalchemy_workflow_node_execution_repository."
+        "SQLAlchemyWorkflowNodeExecutionRepository' (default), "
+        "'core.repositories.celery_workflow_node_execution_repository."
+        "CeleryWorkflowNodeExecutionRepository'",
         default="core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository",
     )
 

api/configs/feature/hosted_service/__init__.py

@@ -222,11 +222,28 @@ class HostedFetchAppTemplateConfig(BaseSettings):
     )
 
 
+class HostedFetchPipelineTemplateConfig(BaseSettings):
+    """
+    Configuration for fetching pipeline templates
+    """
+
+    HOSTED_FETCH_PIPELINE_TEMPLATES_MODE: str = Field(
+        description="Mode for fetching pipeline templates: remote, db, or builtin default to remote,",
+        default="database",
+    )
+
+    HOSTED_FETCH_PIPELINE_TEMPLATES_REMOTE_DOMAIN: str = Field(
+        description="Domain for fetching remote pipeline templates",
+        default="https://tmpl.dify.ai",
+    )
+
+
 class HostedServiceConfig(
     # place the configs in alphabet order
     HostedAnthropicConfig,
     HostedAzureOpenAiConfig,
     HostedFetchAppTemplateConfig,
+    HostedFetchPipelineTemplateConfig,
     HostedMinmaxConfig,
     HostedOpenAiConfig,
     HostedSparkConfig,

api/configs/middleware/__init__.py

@@ -10,6 +10,7 @@ from .storage.aliyun_oss_storage_config import AliyunOSSStorageConfig
 from .storage.amazon_s3_storage_config import S3StorageConfig
 from .storage.azure_blob_storage_config import AzureBlobStorageConfig
 from .storage.baidu_obs_storage_config import BaiduOBSStorageConfig
+from .storage.clickzetta_volume_storage_config import ClickZettaVolumeStorageConfig
 from .storage.google_cloud_storage_config import GoogleCloudStorageConfig
 from .storage.huawei_obs_storage_config import HuaweiCloudOBSStorageConfig
 from .storage.oci_storage_config import OCIStorageConfig
@@ -20,6 +21,7 @@ from .storage.volcengine_tos_storage_config import VolcengineTOSStorageConfig
 from .vdb.analyticdb_config import AnalyticdbConfig
 from .vdb.baidu_vector_config import BaiduVectorDBConfig
 from .vdb.chroma_config import ChromaConfig
+from .vdb.clickzetta_config import ClickzettaConfig
 from .vdb.couchbase_config import CouchbaseConfig
 from .vdb.elasticsearch_config import ElasticsearchConfig
 from .vdb.huawei_cloud_config import HuaweiCloudConfig
@@ -52,6 +54,7 @@ class StorageConfig(BaseSettings):
         "aliyun-oss",
         "azure-blob",
         "baidu-obs",
+        "clickzetta-volume",
         "google-storage",
         "huawei-obs",
         "oci-storage",
@@ -61,8 +64,9 @@ class StorageConfig(BaseSettings):
         "local",
     ] = Field(
         description="Type of storage to use."
-        " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', 'google-storage', "
-        "'huawei-obs', 'oci-storage', 'tencent-cos', 'volcengine-tos', 'supabase'. Default is 'opendal'.",
+        " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', "
+        "'clickzetta-volume', 'google-storage', 'huawei-obs', 'oci-storage', 'tencent-cos', "
+        "'volcengine-tos', 'supabase'. Default is 'opendal'.",
         default="opendal",
     )
 
@@ -140,7 +144,8 @@ class DatabaseConfig(BaseSettings):
         default="postgresql",
     )
 
-    @computed_field
+    @computed_field  # type: ignore[misc]
+    @property
     def SQLALCHEMY_DATABASE_URI(self) -> str:
         db_extras = (
             f"{self.DB_EXTRAS}&client_encoding={self.DB_CHARSET}" if self.DB_CHARSET else self.DB_EXTRAS
@@ -303,6 +308,7 @@ class MiddlewareConfig(
     AliyunOSSStorageConfig,
     AzureBlobStorageConfig,
     BaiduOBSStorageConfig,
+    ClickZettaVolumeStorageConfig,
     GoogleCloudStorageConfig,
     HuaweiCloudOBSStorageConfig,
     OCIStorageConfig,
@@ -315,6 +321,7 @@ class MiddlewareConfig(
     VectorStoreConfig,
     AnalyticdbConfig,
     ChromaConfig,
+    ClickzettaConfig,
     HuaweiCloudConfig,
     MilvusConfig,
     MyScaleConfig,

api/configs/middleware/cache/redis_config.py

@@ -39,6 +39,26 @@ class RedisConfig(BaseSettings):
         default=False,
     )
 
+    REDIS_SSL_CERT_REQS: str = Field(
+        description="SSL certificate requirements (CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED)",
+        default="CERT_NONE",
+    )
+
+    REDIS_SSL_CA_CERTS: Optional[str] = Field(
+        description="Path to the CA certificate file for SSL verification",
+        default=None,
+    )
+
+    REDIS_SSL_CERTFILE: Optional[str] = Field(
+        description="Path to the client certificate file for SSL authentication",
+        default=None,
+    )
+
+    REDIS_SSL_KEYFILE: Optional[str] = Field(
+        description="Path to the client private key file for SSL authentication",
+        default=None,
+    )
+
     REDIS_USE_SENTINEL: Optional[bool] = Field(
         description="Enable Redis Sentinel mode for high availability",
         default=False,

api/configs/middleware/storage/clickzetta_volume_storage_config.py

@@ -0,0 +1,65 @@
+"""ClickZetta Volume Storage Configuration"""
+
+from typing import Optional
+
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class ClickZettaVolumeStorageConfig(BaseSettings):
+    """Configuration for ClickZetta Volume storage."""
+
+    CLICKZETTA_VOLUME_USERNAME: Optional[str] = Field(
+        description="Username for ClickZetta Volume authentication",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_PASSWORD: Optional[str] = Field(
+        description="Password for ClickZetta Volume authentication",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_INSTANCE: Optional[str] = Field(
+        description="ClickZetta instance identifier",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_SERVICE: str = Field(
+        description="ClickZetta service endpoint",
+        default="api.clickzetta.com",
+    )
+
+    CLICKZETTA_VOLUME_WORKSPACE: str = Field(
+        description="ClickZetta workspace name",
+        default="quick_start",
+    )
+
+    CLICKZETTA_VOLUME_VCLUSTER: str = Field(
+        description="ClickZetta virtual cluster name",
+        default="default_ap",
+    )
+
+    CLICKZETTA_VOLUME_SCHEMA: str = Field(
+        description="ClickZetta schema name",
+        default="dify",
+    )
+
+    CLICKZETTA_VOLUME_TYPE: str = Field(
+        description="ClickZetta volume type (table|user|external)",
+        default="user",
+    )
+
+    CLICKZETTA_VOLUME_NAME: Optional[str] = Field(
+        description="ClickZetta volume name for external volumes",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_TABLE_PREFIX: str = Field(
+        description="Prefix for ClickZetta volume table names",
+        default="dataset_",
+    )
+
+    CLICKZETTA_VOLUME_DIFY_PREFIX: str = Field(
+        description="Directory prefix for User Volume to organize Dify files",
+        default="dify_km",
+    )

api/configs/middleware/vdb/clickzetta_config.py

@@ -0,0 +1,69 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class ClickzettaConfig(BaseModel):
+    """
+    Clickzetta Lakehouse vector database configuration
+    """
+
+    CLICKZETTA_USERNAME: Optional[str] = Field(
+        description="Username for authenticating with Clickzetta Lakehouse",
+        default=None,
+    )
+
+    CLICKZETTA_PASSWORD: Optional[str] = Field(
+        description="Password for authenticating with Clickzetta Lakehouse",
+        default=None,
+    )
+
+    CLICKZETTA_INSTANCE: Optional[str] = Field(
+        description="Clickzetta Lakehouse instance ID",
+        default=None,
+    )
+
+    CLICKZETTA_SERVICE: Optional[str] = Field(
+        description="Clickzetta API service endpoint (e.g., 'api.clickzetta.com')",
+        default="api.clickzetta.com",
+    )
+
+    CLICKZETTA_WORKSPACE: Optional[str] = Field(
+        description="Clickzetta workspace name",
+        default="default",
+    )
+
+    CLICKZETTA_VCLUSTER: Optional[str] = Field(
+        description="Clickzetta virtual cluster name",
+        default="default_ap",
+    )
+
+    CLICKZETTA_SCHEMA: Optional[str] = Field(
+        description="Database schema name in Clickzetta",
+        default="public",
+    )
+
+    CLICKZETTA_BATCH_SIZE: Optional[int] = Field(
+        description="Batch size for bulk insert operations",
+        default=100,
+    )
+
+    CLICKZETTA_ENABLE_INVERTED_INDEX: Optional[bool] = Field(
+        description="Enable inverted index for full-text search capabilities",
+        default=True,
+    )
+
+    CLICKZETTA_ANALYZER_TYPE: Optional[str] = Field(
+        description="Analyzer type for full-text search: keyword, english, chinese, unicode",
+        default="chinese",
+    )
+
+    CLICKZETTA_ANALYZER_MODE: Optional[str] = Field(
+        description="Analyzer mode for tokenization: max_word (fine-grained) or smart (intelligent)",
+        default="smart",
+    )
+
+    CLICKZETTA_VECTOR_DISTANCE_FUNCTION: Optional[str] = Field(
+        description="Distance function for vector similarity: l2_distance or cosine_distance",
+        default="cosine_distance",
+    )

api/constants/__init__.py

@@ -9,10 +9,10 @@ DEFAULT_FILE_NUMBER_LIMITS = 3
 IMAGE_EXTENSIONS = ["jpg", "jpeg", "png", "webp", "gif", "svg"]
 IMAGE_EXTENSIONS.extend([ext.upper() for ext in IMAGE_EXTENSIONS])
 
-VIDEO_EXTENSIONS = ["mp4", "mov", "mpeg", "mpga"]
+VIDEO_EXTENSIONS = ["mp4", "mov", "mpeg", "webm"]
 VIDEO_EXTENSIONS.extend([ext.upper() for ext in VIDEO_EXTENSIONS])
 
-AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "webm", "amr"]
+AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "amr", "mpga"]
 AUDIO_EXTENSIONS.extend([ext.upper() for ext in AUDIO_EXTENSIONS])
 
 

api/contexts/__init__.py

@@ -3,6 +3,7 @@ from threading import Lock
 from typing import TYPE_CHECKING
 
 from contexts.wrapper import RecyclableContextVar
+from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
 
 if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
@@ -33,3 +34,11 @@ plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(Cont
 plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
     ContextVar("plugin_model_schemas")
 )
+
+datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginProviderController"]] = (
+    RecyclableContextVar(ContextVar("datasource_plugin_providers"))
+)
+
+datasource_plugin_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("datasource_plugin_providers_lock")
+)

api/controllers/common/errors.py

@@ -1,5 +1,7 @@
 from werkzeug.exceptions import HTTPException
 
+from libs.exception import BaseHTTPException
+
 
 class FilenameNotExistsError(HTTPException):
     code = 400
@@ -9,3 +11,27 @@ class FilenameNotExistsError(HTTPException):
 class RemoteFileUploadError(HTTPException):
     code = 400
     description = "Error uploading remote file."
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400

api/controllers/console/__init__.py

@@ -84,8 +84,18 @@ from .datasets import (
     external,
     hit_testing,
     metadata,
+    upload_file,
     website,
 )
+from .datasets.rag_pipeline import (
+    datasource_auth,
+    datasource_content_preview,
+    rag_pipeline,
+    rag_pipeline_datasets,
+    rag_pipeline_draft_variable,
+    rag_pipeline_import,
+    rag_pipeline_workflow,
+)
 
 # Import explore controllers
 from .explore import (

api/controllers/console/app/annotation.py

@@ -3,9 +3,8 @@ from flask_login import current_user
 from flask_restful import Resource, marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
+from controllers.common.errors import NoFileUploadedError, TooManyFilesError
 from controllers.console import api
-from controllers.console.app.error import NoFileUploadedError
-from controllers.console.datasets.error import TooManyFilesError
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
@@ -225,14 +224,15 @@ class AnnotationBatchImportApi(Resource):
             raise Forbidden()
 
         app_id = str(app_id)
-        # get file from request
-        file = request.files["file"]
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
 
         if len(request.files) > 1:
             raise TooManyFilesError()
+
+        # get file from request
+        file = request.files["file"]
         # check file type
         if not file.filename or not file.filename.lower().endswith(".csv"):
             raise ValueError("Invalid file type. Only CSV files are allowed")

api/controllers/console/app/app.py

@@ -28,6 +28,12 @@ from services.feature_service import FeatureService
 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]
 
 
+def _validate_description_length(description):
+    if description and len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
 class AppListApi(Resource):
     @setup_required
     @login_required
@@ -94,7 +100,7 @@ class AppListApi(Resource):
         """Create app"""
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("mode", type=str, choices=ALLOW_CREATE_APP_MODES, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
@@ -146,7 +152,7 @@ class AppApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
         parser.add_argument("icon_background", type=str, location="json")
@@ -189,7 +195,7 @@ class AppCopyApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
         parser.add_argument("icon_background", type=str, location="json")

api/controllers/console/app/completion.py

@@ -1,6 +1,7 @@
 import logging
 
 import flask_login
+from flask import request
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -24,6 +25,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
@@ -115,6 +117,10 @@ class ChatMessageApi(Resource):
         streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         account = flask_login.current_user
 
         try:

api/controllers/console/app/error.py

@@ -79,18 +79,6 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
 class DraftWorkflowNotExist(BaseHTTPException):
     error_code = "draft_workflow_not_exist"
     description = "Draft workflow need to be initialized."

api/controllers/console/app/message.py

@@ -27,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
@@ -124,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 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/statistic.py

@@ -67,7 +67,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "message_count": i.message_count})
 
@@ -176,7 +176,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "terminal_count": i.terminal_count})
 
@@ -234,7 +234,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "token_count": i.token_count, "total_price": i.total_price, "currency": "USD"}
@@ -310,7 +310,7 @@ ORDER BY
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "interactions": float(i.interactions.quantize(Decimal("0.01")))}
@@ -373,7 +373,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {
@@ -435,7 +435,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "latency": round(i.latency * 1000, 4)})
 
@@ -495,7 +495,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "tps": round(i.tokens_per_second, 4)})
 

api/controllers/console/app/workflow.py

@@ -23,6 +23,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
+from core.helper.trace_id_helper import get_external_trace_id
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -185,6 +186,10 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
 
         args = parser.parse_args()
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         try:
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=True
@@ -373,6 +378,10 @@ class DraftWorkflowRunApi(Resource):
         parser.add_argument("files", type=list, required=False, location="json")
         args = parser.parse_args()
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         try:
             response = AppGenerateService.generate(
                 app_model=app_model,

api/controllers/console/app/workflow_draft_variable.py

@@ -163,11 +163,11 @@ class WorkflowVariableCollectionApi(Resource):
             draft_var_srv = WorkflowDraftVariableService(
                 session=session,
             )
-        workflow_vars = draft_var_srv.list_variables_without_values(
-            app_id=app_model.id,
-            page=args.page,
-            limit=args.limit,
-        )
+            workflow_vars = draft_var_srv.list_variables_without_values(
+                app_id=app_model.id,
+                page=args.page,
+                limit=args.limit,
+            )
 
         return workflow_vars
 

api/controllers/console/app/workflow_statistic.py

@@ -2,6 +2,7 @@ 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
@@ -71,7 +72,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "runs": i.runs})
 
@@ -133,7 +134,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "terminal_count": i.terminal_count})
 
@@ -195,7 +196,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {
@@ -277,7 +278,7 @@ GROUP BY
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "interactions": float(i.interactions.quantize(Decimal("0.01")))}

api/controllers/console/datasets/data_source.py

@@ -1,4 +1,6 @@
 import json
+from collections.abc import Generator
+from typing import cast
 
 from flask import request
 from flask_login import current_user
@@ -9,6 +11,8 @@ from werkzeug.exceptions import NotFound
 
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
+from core.datasource.entities.datasource_entities import DatasourceProviderType, OnlineDocumentPagesMessage
+from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
 from core.indexing_runner import IndexingRunner
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.notion_extractor import NotionExtractor
@@ -18,6 +22,7 @@ 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
+from services.datasource_provider_service import DatasourceProviderService
 from tasks.document_indexing_sync_task import document_indexing_sync_task
 
 
@@ -112,6 +117,18 @@ class DataSourceNotionListApi(Resource):
     @marshal_with(integrate_notion_info_list_fields)
     def get(self):
         dataset_id = request.args.get("dataset_id", default=None, type=str)
+        credential_id = request.args.get("credential_id", default=None, type=str)
+        if not credential_id:
+            raise ValueError("Credential id is required.")
+        datasource_provider_service = DatasourceProviderService()
+        credential = datasource_provider_service.get_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            credential_id=credential_id,
+            provider="notion_datasource",
+            plugin_id="langgenius/notion_datasource",
+        )
+        if not credential:
+            raise NotFound("Credential not found.")
         exist_page_ids = []
         with Session(db.engine) as session:
             # import notion in the exist dataset
@@ -135,31 +152,49 @@ class DataSourceNotionListApi(Resource):
                         data_source_info = json.loads(document.data_source_info)
                         exist_page_ids.append(data_source_info["notion_page_id"])
             # get all authorized pages
-            data_source_bindings = session.scalars(
-                select(DataSourceOauthBinding).filter_by(
-                    tenant_id=current_user.current_tenant_id, provider="notion", disabled=False
+            from core.datasource.datasource_manager import DatasourceManager
+
+            datasource_runtime = DatasourceManager.get_datasource_runtime(
+                provider_id="langgenius/notion_datasource/notion_datasource",
+                datasource_name="notion_datasource",
+                tenant_id=current_user.current_tenant_id,
+                datasource_type=DatasourceProviderType.ONLINE_DOCUMENT,
+            )
+            datasource_provider_service = DatasourceProviderService()
+            if credential:
+                datasource_runtime.runtime.credentials = credential
+            datasource_runtime = cast(OnlineDocumentDatasourcePlugin, datasource_runtime)
+            online_document_result: Generator[OnlineDocumentPagesMessage, None, None] = (
+                datasource_runtime.get_online_document_pages(
+                    user_id=current_user.id,
+                    datasource_parameters={},
+                    provider_type=datasource_runtime.datasource_provider_type(),
                 )
-            ).all()
-            if not data_source_bindings:
-                return {"notion_info": []}, 200
-            pre_import_info_list = []
-            for data_source_binding in data_source_bindings:
-                source_info = data_source_binding.source_info
-                pages = source_info["pages"]
-                # Filter out already bound pages
-                for page in pages:
-                    if page["page_id"] in exist_page_ids:
-                        page["is_bound"] = True
-                    else:
-                        page["is_bound"] = False
-                pre_import_info = {
-                    "workspace_name": source_info["workspace_name"],
-                    "workspace_icon": source_info["workspace_icon"],
-                    "workspace_id": source_info["workspace_id"],
-                    "pages": pages,
-                }
-                pre_import_info_list.append(pre_import_info)
-            return {"notion_info": pre_import_info_list}, 200
+            )
+            try:
+                pages = []
+                workspace_info = {}
+                for message in online_document_result:
+                    result = message.result
+                    for info in result:
+                        workspace_info = {
+                            "workspace_id": info.workspace_id,
+                            "workspace_name": info.workspace_name,
+                            "workspace_icon": info.workspace_icon,
+                        }
+                        for page in info.pages:
+                            page_info = {
+                                "page_id": page.page_id,
+                                "page_name": page.page_name,
+                                "type": page.type,
+                                "parent_id": page.parent_id,
+                                "is_bound": page.page_id in exist_page_ids,
+                                "page_icon": page.page_icon,
+                            }
+                            pages.append(page_info)
+            except Exception as e:
+                raise e
+            return {"notion_info": {**workspace_info, "pages": pages}}, 200
 
 
 class DataSourceNotionApi(Resource):
@@ -167,27 +202,25 @@ class DataSourceNotionApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, workspace_id, page_id, page_type):
+        credential_id = request.args.get("credential_id", default=None, type=str)
+        if not credential_id:
+            raise ValueError("Credential id is required.")
+        datasource_provider_service = DatasourceProviderService()
+        credential = datasource_provider_service.get_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            credential_id=credential_id,
+            provider="notion_datasource",
+            plugin_id="langgenius/notion_datasource",
+        )
+
         workspace_id = str(workspace_id)
         page_id = str(page_id)
-        with Session(db.engine) as session:
-            data_source_binding = session.execute(
-                select(DataSourceOauthBinding).where(
-                    db.and_(
-                        DataSourceOauthBinding.tenant_id == current_user.current_tenant_id,
-                        DataSourceOauthBinding.provider == "notion",
-                        DataSourceOauthBinding.disabled == False,
-                        DataSourceOauthBinding.source_info["workspace_id"] == f'"{workspace_id}"',
-                    )
-                )
-            ).scalar_one_or_none()
-        if not data_source_binding:
-            raise NotFound("Data source binding not found.")
 
         extractor = NotionExtractor(
             notion_workspace_id=workspace_id,
             notion_obj_id=page_id,
             notion_page_type=page_type,
-            notion_access_token=data_source_binding.access_token,
+            notion_access_token=credential.get("integration_secret"),
             tenant_id=current_user.current_tenant_id,
         )
 
@@ -212,10 +245,12 @@ class DataSourceNotionApi(Resource):
         extract_settings = []
         for notion_info in notion_info_list:
             workspace_id = notion_info["workspace_id"]
+            credential_id = notion_info.get("credential_id")
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
                     datasource_type="notion_import",
                     notion_info={
+                        "credential_id": credential_id,
                         "notion_workspace_id": workspace_id,
                         "notion_obj_id": page["page_id"],
                         "notion_page_type": page["type"],

api/controllers/console/datasets/datasets.py

@@ -41,7 +41,7 @@ def _validate_name(name):
 
 
 def _validate_description_length(description):
-    if len(description) > 400:
+    if description and len(description) > 400:
         raise ValueError("Description cannot exceed 400 characters.")
     return description
 
@@ -113,7 +113,7 @@ class DatasetListApi(Resource):
         )
         parser.add_argument(
             "description",
-            type=str,
+            type=_validate_description_length,
             nullable=True,
             required=False,
             default="",
@@ -279,6 +279,15 @@ class DatasetApi(Resource):
             location="json",
             help="Invalid external knowledge api id.",
         )
+
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            required=False,
+            nullable=True,
+            location="json",
+            help="Invalid icon info.",
+        )
         args = parser.parse_args()
         data = request.get_json()
 
@@ -429,10 +438,12 @@ class DatasetIndexingEstimateApi(Resource):
             notion_info_list = args["info_list"]["notion_info_list"]
             for notion_info in notion_info_list:
                 workspace_id = notion_info["workspace_id"]
+                credential_id = notion_info.get("credential_id")
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
                         datasource_type="notion_import",
                         notion_info={
+                            "credential_id": credential_id,
                             "notion_workspace_id": workspace_id,
                             "notion_obj_id": page["page_id"],
                             "notion_page_type": page["type"],
@@ -683,6 +694,7 @@ class DatasetRetrievalSettingApi(Resource):
                 | VectorType.HUAWEI_CLOUD
                 | VectorType.TENCENT
                 | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
             ):
                 return {
                     "retrieval_method": [
@@ -731,6 +743,7 @@ class DatasetRetrievalSettingMockApi(Resource):
                 | VectorType.TENCENT
                 | VectorType.HUAWEI_CLOUD
                 | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
             ):
                 return {
                     "retrieval_method": [

api/controllers/console/datasets/datasets_document.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from argparse import ArgumentTypeError
 from typing import cast
@@ -51,6 +52,7 @@ from fields.document_fields import (
 from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Dataset, DatasetProcessRule, Document, DocumentSegment, UploadFile
+from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 
@@ -508,6 +510,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 extract_setting = ExtractSetting(
                     datasource_type="notion_import",
                     notion_info={
+                        "credential_id": data_source_info["credential_id"],
                         "notion_workspace_id": data_source_info["notion_workspace_id"],
                         "notion_obj_id": data_source_info["notion_page_id"],
                         "notion_page_type": data_source_info["type"],
@@ -642,7 +645,7 @@ class DocumentIndexingStatusApi(DocumentResource):
         return marshal(document_dict, document_status_fields)
 
 
-class DocumentDetailApi(DocumentResource):
+class DocumentApi(DocumentResource):
     METADATA_CHOICES = {"all", "only", "without"}
 
     @setup_required
@@ -661,7 +664,7 @@ class DocumentDetailApi(DocumentResource):
             response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
         elif metadata == "without":
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
-            document_process_rules = document.dataset_process_rule.to_dict()
+            document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
             data_source_info = document.data_source_detail_dict
             response = {
                 "id": document.id,
@@ -730,6 +733,28 @@ class DocumentDetailApi(DocumentResource):
 
         return response, 200
 
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def delete(self, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+        dataset = DatasetService.get_dataset(dataset_id)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        # check user's model setting
+        DatasetService.check_dataset_model_setting(dataset)
+
+        document = self.get_document(dataset_id, document_id)
+
+        try:
+            DocumentService.delete_document(document)
+        except services.errors.document.DocumentIndexingError:
+            raise DocumentIndexingError("Cannot delete document during indexing.")
+
+        return {"result": "success"}, 204
+
 
 class DocumentProcessingApi(DocumentResource):
     @setup_required
@@ -768,30 +793,6 @@ class DocumentProcessingApi(DocumentResource):
         return {"result": "success"}, 200
 
 
-class DocumentDeleteApi(DocumentResource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
-    def delete(self, dataset_id, document_id):
-        dataset_id = str(dataset_id)
-        document_id = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id)
-        if dataset is None:
-            raise NotFound("Dataset not found.")
-        # check user's model setting
-        DatasetService.check_dataset_model_setting(dataset)
-
-        document = self.get_document(dataset_id, document_id)
-
-        try:
-            DocumentService.delete_document(document)
-        except services.errors.document.DocumentIndexingError:
-            raise DocumentIndexingError("Cannot delete document during indexing.")
-
-        return {"result": "success"}, 204
-
-
 class DocumentMetadataApi(DocumentResource):
     @setup_required
     @login_required
@@ -1028,6 +1029,41 @@ class WebsiteDocumentSyncApi(DocumentResource):
         return {"result": "success"}, 200
 
 
+class DocumentPipelineExecutionLogApi(DocumentResource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        document = DocumentService.get_document(dataset.id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        log = (
+            db.session.query(DocumentPipelineExecutionLog)
+            .filter_by(document_id=document_id)
+            .order_by(DocumentPipelineExecutionLog.created_at.desc())
+            .first()
+        )
+        if not log:
+            return {
+                "datasource_info": None,
+                "datasource_type": None,
+                "input_data": None,
+                "datasource_node_id": None,
+            }, 200
+        return {
+            "datasource_info": json.loads(log.datasource_info),
+            "datasource_type": log.datasource_type,
+            "input_data": log.input_data,
+            "datasource_node_id": log.datasource_node_id,
+        }, 200
+
+
 api.add_resource(GetProcessRuleApi, "/datasets/process-rule")
 api.add_resource(DatasetDocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DatasetInitApi, "/datasets/init")
@@ -1037,11 +1073,10 @@ api.add_resource(
 api.add_resource(DocumentBatchIndexingEstimateApi, "/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
 api.add_resource(DocumentBatchIndexingStatusApi, "/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-status")
 api.add_resource(DocumentIndexingStatusApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-status")
-api.add_resource(DocumentDetailApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
+api.add_resource(DocumentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(
     DocumentProcessingApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/<string:action>"
 )
-api.add_resource(DocumentDeleteApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(DocumentMetadataApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
 api.add_resource(DocumentStatusApi, "/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
 api.add_resource(DocumentPauseApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")
@@ -1050,3 +1085,6 @@ api.add_resource(DocumentRetryApi, "/datasets/<uuid:dataset_id>/retry")
 api.add_resource(DocumentRenameApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/rename")
 
 api.add_resource(WebsiteDocumentSyncApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/website-sync")
+api.add_resource(
+    DocumentPipelineExecutionLogApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log"
+)

api/controllers/console/datasets/error.py

@@ -1,30 +1,6 @@
 from libs.exception import BaseHTTPException
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."
@@ -95,3 +71,9 @@ class ChildChunkDeleteIndexError(BaseHTTPException):
     error_code = "child_chunk_delete_index_error"
     description = "Delete child chunk index failed: {message}"
     code = 500
+
+
+class PipelineNotFoundError(BaseHTTPException):
+    error_code = "pipeline_not_found"
+    description = "Pipeline not found."
+    code = 404

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

@@ -0,0 +1,365 @@
+from fastapi.encoders import jsonable_encoder
+from flask import make_response, redirect, request
+from flask_login import current_user  # type: ignore
+from flask_restful import (  # type: ignore
+    Resource,  # type: ignore
+    reqparse,
+)
+from werkzeug.exceptions import Forbidden, NotFound
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.model_runtime.errors.validate import CredentialsValidateFailedError
+from core.plugin.entities.plugin import DatasourceProviderID
+from core.plugin.impl.oauth import OAuthHandler
+from libs.helper import StrLen
+from libs.login import login_required
+from services.datasource_provider_service import DatasourceProviderService
+from services.plugin.oauth_service import OAuthProxyService
+
+
+class DatasourcePluginOAuthAuthorizationUrl(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id: str):
+        user = current_user
+        tenant_id = user.current_tenant_id
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        credential_id = request.args.get("credential_id")
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        provider_name = datasource_provider_id.provider_name
+        plugin_id = datasource_provider_id.plugin_id
+        oauth_config = DatasourceProviderService().get_oauth_client(
+            tenant_id=tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        if not oauth_config:
+            raise ValueError(f"No OAuth Client Config for {provider_id}")
+
+        context_id = OAuthProxyService.create_proxy_context(
+            user_id=current_user.id,
+            tenant_id=tenant_id,
+            plugin_id=plugin_id,
+            provider=provider_name,
+            credential_id=credential_id,
+        )
+        oauth_handler = OAuthHandler()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/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_config,
+        )
+        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 DatasourceOAuthCallback(Resource):
+    @setup_required
+    def get(self, provider_id: str):
+        context_id = request.cookies.get("context_id") or request.args.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")
+
+        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        plugin_id = datasource_provider_id.plugin_id
+        datasource_provider_service = DatasourceProviderService()
+        oauth_client_params = datasource_provider_service.get_oauth_client(
+            tenant_id=tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        if not oauth_client_params:
+            raise NotFound()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/callback"
+        oauth_handler = OAuthHandler()
+        oauth_response = oauth_handler.get_credentials(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            plugin_id=plugin_id,
+            provider=datasource_provider_id.provider_name,
+            redirect_uri=redirect_uri,
+            system_credentials=oauth_client_params,
+            request=request,
+        )
+        credential_id = context.get("credential_id")
+        if credential_id:
+            datasource_provider_service.reauthorize_datasource_oauth_provider(
+                tenant_id=tenant_id,
+                provider_id=datasource_provider_id,
+                avatar_url=oauth_response.metadata.get("avatar_url") or None,
+                name=oauth_response.metadata.get("name") or None,
+                expire_at=oauth_response.expires_at,
+                credentials=dict(oauth_response.credentials),
+                credential_id=context.get("credential_id"),
+            )
+        else:
+            datasource_provider_service.add_datasource_oauth_provider(
+                tenant_id=tenant_id,
+                provider_id=datasource_provider_id,
+                avatar_url=oauth_response.metadata.get("avatar_url") or None,
+                name=oauth_response.metadata.get("name") or None,
+                expire_at=oauth_response.expires_at,
+                credentials=dict(oauth_response.credentials),
+            )
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
+class DatasourceAuth(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name", type=StrLen(max_length=100), required=False, nullable=True, location="json", default=None
+        )
+        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+
+        try:
+            datasource_provider_service.add_datasource_api_key_provider(
+                tenant_id=current_user.current_tenant_id,
+                provider_id=datasource_provider_id,
+                credentials=args["credentials"],
+                name=args["name"],
+            )
+        except CredentialsValidateFailedError as ex:
+            raise ValueError(str(ex))
+        return {"result": "success"}, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.list_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            provider=datasource_provider_id.provider_name,
+            plugin_id=datasource_provider_id.plugin_id,
+        )
+        return {"result": datasources}, 200
+
+
+class DatasourceAuthDeleteApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        plugin_id = datasource_provider_id.plugin_id
+        provider_name = datasource_provider_id.provider_name
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.remove_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            auth_id=args["credential_id"],
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceAuthUpdateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        parser = reqparse.RequestParser()
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("name", type=StrLen(max_length=100), required=False, nullable=True, location="json")
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        if not current_user.is_editor:
+            raise Forbidden()
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.update_datasource_credentials(
+            tenant_id=current_user.current_tenant_id,
+            auth_id=args["credential_id"],
+            provider=datasource_provider_id.provider_name,
+            plugin_id=datasource_provider_id.plugin_id,
+            credentials=args.get("credentials", {}),
+            name=args.get("name", None),
+        )
+        return {"result": "success"}, 201
+
+
+class DatasourceAuthListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.get_all_datasource_credentials(
+            tenant_id=current_user.current_tenant_id
+        )
+        return {"result": jsonable_encoder(datasources)}, 200
+
+
+class DatasourceHardCodeAuthListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        datasource_provider_service = DatasourceProviderService()
+        datasources = datasource_provider_service.get_hard_code_datasource_credentials(
+            tenant_id=current_user.current_tenant_id
+        )
+        return {"result": jsonable_encoder(datasources)}, 200
+
+
+class DatasourceAuthOauthCustomClient(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        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()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.setup_oauth_custom_client_params(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            client_params=args.get("client_params", {}),
+            enabled=args.get("enable_oauth_custom_client", False),
+        )
+        return {"result": "success"}, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider_id: str):
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.remove_oauth_custom_client_params(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceAuthDefaultApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.set_default_datasource_provider(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            credential_id=args["id"],
+        )
+        return {"result": "success"}, 200
+
+
+class DatasourceUpdateProviderNameApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider_id: str):
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("name", type=StrLen(max_length=100), required=True, nullable=False, location="json")
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        datasource_provider_id = DatasourceProviderID(provider_id)
+        datasource_provider_service = DatasourceProviderService()
+        datasource_provider_service.update_datasource_provider_name(
+            tenant_id=current_user.current_tenant_id,
+            datasource_provider_id=datasource_provider_id,
+            name=args["name"],
+            credential_id=args["credential_id"],
+        )
+        return {"result": "success"}, 200
+
+
+api.add_resource(
+    DatasourcePluginOAuthAuthorizationUrl,
+    "/oauth/plugin/<path:provider_id>/datasource/get-authorization-url",
+)
+api.add_resource(
+    DatasourceOAuthCallback,
+    "/oauth/plugin/<path:provider_id>/datasource/callback",
+)
+api.add_resource(
+    DatasourceAuth,
+    "/auth/plugin/datasource/<path:provider_id>",
+)
+
+api.add_resource(
+    DatasourceAuthUpdateApi,
+    "/auth/plugin/datasource/<path:provider_id>/update",
+)
+
+api.add_resource(
+    DatasourceAuthDeleteApi,
+    "/auth/plugin/datasource/<path:provider_id>/delete",
+)
+
+api.add_resource(
+    DatasourceAuthListApi,
+    "/auth/plugin/datasource/list",
+)
+
+api.add_resource(
+    DatasourceHardCodeAuthListApi,
+    "/auth/plugin/datasource/default-list",
+)
+
+api.add_resource(
+    DatasourceAuthOauthCustomClient,
+    "/auth/plugin/datasource/<path:provider_id>/custom-client",
+)
+
+api.add_resource(
+    DatasourceAuthDefaultApi,
+    "/auth/plugin/datasource/<path:provider_id>/default",
+)
+
+api.add_resource(
+    DatasourceUpdateProviderNameApi,
+    "/auth/plugin/datasource/<path:provider_id>/update-name",
+)

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

@@ -0,0 +1,57 @@
+from flask_restful import (  # type: ignore
+    Resource,  # type: ignore
+    reqparse,
+)
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import account_initialization_required, setup_required
+from libs.login import current_user, login_required
+from models import Account
+from models.dataset import Pipeline
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+
+class DataSourceContentPreviewApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run datasource content preview
+        """
+        if not isinstance(current_user, Account):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("datasource_type", type=str, required=True, location="json")
+        parser.add_argument("credential_id", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        inputs = args.get("inputs")
+        if inputs is None:
+            raise ValueError("missing inputs")
+        datasource_type = args.get("datasource_type")
+        if datasource_type is None:
+            raise ValueError("missing datasource_type")
+
+        rag_pipeline_service = RagPipelineService()
+        preview_content = rag_pipeline_service.run_datasource_node_preview(
+            pipeline=pipeline,
+            node_id=node_id,
+            user_inputs=inputs,
+            account=current_user,
+            datasource_type=datasource_type,
+            is_published=True,
+            credential_id=args.get("credential_id"),
+        )
+        return preview_content, 200
+
+
+api.add_resource(
+    DataSourceContentPreviewApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview",
+)

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

@@ -0,0 +1,164 @@
+import logging
+
+from flask import request
+from flask_restful import Resource, reqparse
+from sqlalchemy.orm import Session
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    enterprise_license_required,
+    knowledge_pipeline_publish_enabled,
+    setup_required,
+)
+from extensions.ext_database import db
+from libs.login import login_required
+from models.dataset import PipelineCustomizedTemplate
+from services.entities.knowledge_entities.rag_pipeline_entities import PipelineTemplateInfoEntity
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+logger = logging.getLogger(__name__)
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 40:
+        raise ValueError("Name must be between 1 to 40 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class PipelineTemplateListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def get(self):
+        type = request.args.get("type", default="built-in", type=str)
+        language = request.args.get("language", default="en-US", type=str)
+        # get pipeline templates
+        pipeline_templates = RagPipelineService.get_pipeline_templates(type, language)
+        return pipeline_templates, 200
+
+
+class PipelineTemplateDetailApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def get(self, template_id: str):
+        type = request.args.get("type", default="built-in", type=str)
+        rag_pipeline_service = RagPipelineService()
+        pipeline_template = rag_pipeline_service.get_pipeline_template_detail(template_id, type)
+        return pipeline_template, 200
+
+
+class CustomizedPipelineTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def patch(self, template_id: str):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            location="json",
+            nullable=True,
+        )
+        args = parser.parse_args()
+        pipeline_template_info = PipelineTemplateInfoEntity(**args)
+        RagPipelineService.update_customized_pipeline_template(template_id, pipeline_template_info)
+        return 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def delete(self, template_id: str):
+        RagPipelineService.delete_customized_pipeline_template(template_id)
+        return 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    def post(self, template_id: str):
+        with Session(db.engine) as session:
+            template = (
+                session.query(PipelineCustomizedTemplate).filter(PipelineCustomizedTemplate.id == template_id).first()
+            )
+            if not template:
+                raise ValueError("Customized pipeline template not found.")
+
+        return {"data": template.yaml_content}, 200
+
+
+class PublishCustomizedPipelineTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @knowledge_pipeline_publish_enabled
+    def post(self, pipeline_id: str):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            location="json",
+            nullable=True,
+        )
+        args = parser.parse_args()
+        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service.publish_customized_pipeline_template(pipeline_id, args)
+        return {"result": "success"}
+
+
+api.add_resource(
+    PipelineTemplateListApi,
+    "/rag/pipeline/templates",
+)
+api.add_resource(
+    PipelineTemplateDetailApi,
+    "/rag/pipeline/templates/<string:template_id>",
+)
+api.add_resource(
+    CustomizedPipelineTemplateApi,
+    "/rag/pipeline/customized/templates/<string:template_id>",
+)
+api.add_resource(
+    PublishCustomizedPipelineTemplateApi,
+    "/rag/pipelines/<string:pipeline_id>/customized/publish",
+)

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

@@ -0,0 +1,171 @@
+from flask_login import current_user  # type: ignore  # type: ignore
+from flask_restful import Resource, marshal, reqparse  # type: ignore
+from werkzeug.exceptions import Forbidden
+
+import services
+from controllers.console import api
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.wraps import (
+    account_initialization_required,
+    cloud_edition_billing_rate_limit_check,
+    setup_required,
+)
+from fields.dataset_fields import dataset_detail_fields
+from libs.login import login_required
+from models.dataset import DatasetPermissionEnum
+from services.dataset_service import DatasetPermissionService, DatasetService
+from services.entities.knowledge_entities.rag_pipeline_entities import RagPipelineDatasetCreateEntity
+from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 40:
+        raise ValueError("Name must be between 1 to 40 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class CreateRagPipelineDatasetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="type is required. Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            nullable=True,
+            required=False,
+            default={},
+        )
+
+        parser.add_argument(
+            "permission",
+            type=str,
+            choices=(DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.ALL_TEAM, DatasetPermissionEnum.PARTIAL_TEAM),
+            nullable=True,
+            required=False,
+            default=DatasetPermissionEnum.ONLY_ME,
+        )
+
+        parser.add_argument(
+            "partial_member_list",
+            type=list,
+            nullable=True,
+            required=False,
+            default=[],
+        )
+
+        parser.add_argument(
+            "yaml_content",
+            type=str,
+            nullable=False,
+            required=True,
+            help="yaml_content is required.",
+        )
+
+        args = parser.parse_args()
+
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+        rag_pipeline_dataset_create_entity = RagPipelineDatasetCreateEntity(**args)
+        try:
+            import_info = RagPipelineDslService.create_rag_pipeline_dataset(
+                tenant_id=current_user.current_tenant_id,
+                rag_pipeline_dataset_create_entity=rag_pipeline_dataset_create_entity,
+            )
+            if rag_pipeline_dataset_create_entity.permission == "partial_members":
+                DatasetPermissionService.update_partial_member_list(
+                    current_user.current_tenant_id,
+                    import_info["dataset_id"],
+                    rag_pipeline_dataset_create_entity.partial_member_list,
+                )
+        except services.errors.dataset.DatasetNameDuplicateError:
+            raise DatasetNameDuplicateError()
+
+        return import_info, 201
+
+
+class CreateEmptyRagPipelineDatasetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self):
+        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="type is required. Name must be between 1 to 40 characters.",
+            type=_validate_name,
+        )
+        parser.add_argument(
+            "description",
+            type=str,
+            nullable=True,
+            required=False,
+            default="",
+        )
+
+        parser.add_argument(
+            "icon_info",
+            type=dict,
+            nullable=True,
+            required=False,
+            default={},
+        )
+
+        parser.add_argument(
+            "permission",
+            type=str,
+            choices=(DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.ALL_TEAM, DatasetPermissionEnum.PARTIAL_TEAM),
+            nullable=True,
+            required=False,
+            default=DatasetPermissionEnum.ONLY_ME,
+        )
+
+        parser.add_argument(
+            "partial_member_list",
+            type=list,
+            nullable=True,
+            required=False,
+            default=[],
+        )
+
+        args = parser.parse_args()
+        dataset = DatasetService.create_empty_rag_pipeline_dataset(
+            tenant_id=current_user.current_tenant_id,
+            rag_pipeline_dataset_create_entity=RagPipelineDatasetCreateEntity(**args),
+        )
+        return marshal(dataset, dataset_detail_fields), 201
+
+
+api.add_resource(CreateRagPipelineDatasetApi, "/rag/pipeline/dataset")
+api.add_resource(CreateEmptyRagPipelineDatasetApi, "/rag/pipeline/empty-dataset")

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

@@ -0,0 +1,416 @@
+import logging
+from typing import Any, NoReturn
+
+from flask import Response
+from flask_restful import Resource, fields, inputs, marshal, marshal_with, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.app.error import (
+    DraftWorkflowNotExist,
+)
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.segment_group import SegmentGroup
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.variables.types import SegmentType
+from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from libs.login import current_user, login_required
+from models import db
+from models.dataset import Pipeline
+from models.workflow import WorkflowDraftVariable
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+
+logger = logging.getLogger(__name__)
+
+
+def _convert_values_to_json_serializable_object(value: Segment) -> Any:
+    if isinstance(value, FileSegment):
+        return value.value.model_dump()
+    elif isinstance(value, ArrayFileSegment):
+        return [i.model_dump() for i in value.value]
+    elif isinstance(value, SegmentGroup):
+        return [_convert_values_to_json_serializable_object(i) for i in value.value]
+    else:
+        return value.value
+
+
+def _serialize_var_value(variable: WorkflowDraftVariable) -> Any:
+    value = variable.get_value()
+    # create a copy of the value to avoid affecting the model cache.
+    value = value.model_copy(deep=True)
+    # Refresh the url signature before returning it to client.
+    if isinstance(value, FileSegment):
+        file = value.value
+        file.remote_url = file.generate_url()
+    elif isinstance(value, ArrayFileSegment):
+        files = value.value
+        for file in files:
+            file.remote_url = file.generate_url()
+    return _convert_values_to_json_serializable_object(value)
+
+
+def _create_pagination_parser():
+    parser = reqparse.RequestParser()
+    parser.add_argument(
+        "page",
+        type=inputs.int_range(1, 100_000),
+        required=False,
+        default=1,
+        location="args",
+        help="the page of data requested",
+    )
+    parser.add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
+    return parser
+
+
+_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,
+    "edited": fields.Boolean(attribute=lambda model: model.edited),
+    "visible": fields.Boolean,
+}
+
+_WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
+    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
+    value=fields.Raw(attribute=_serialize_var_value),
+)
+
+_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
+    "id": fields.String,
+    "type": fields.String(attribute=lambda _: "env"),
+    "name": fields.String,
+    "description": fields.String,
+    "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
+    "value_type": fields.String,
+    "edited": fields.Boolean(attribute=lambda model: model.edited),
+    "visible": fields.Boolean,
+}
+
+_WORKFLOW_DRAFT_ENV_VARIABLE_LIST_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS)),
+}
+
+
+def _get_items(var_list: WorkflowDraftVariableList) -> list[WorkflowDraftVariable]:
+    return var_list.variables
+
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS), attribute=_get_items),
+    "total": fields.Raw(),
+}
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_FIELDS), attribute=_get_items),
+}
+
+
+def _api_prerequisite(f):
+    """Common prerequisites for all draft workflow variable APIs.
+
+    It ensures the following conditions are satisfied:
+
+    - Dify has been property setup.
+    - The request user has logged in and initialized.
+    - The requested app is a workflow or a chat flow.
+    - The request user has the edit permission for the app.
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def wrapper(*args, **kwargs):
+        if not current_user.is_editor:
+            raise Forbidden()
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+class RagPipelineVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS)
+    def get(self, pipeline: Pipeline):
+        """
+        Get draft workflow
+        """
+        parser = _create_pagination_parser()
+        args = parser.parse_args()
+
+        # fetch draft workflow by app_model
+        rag_pipeline_service = RagPipelineService()
+        workflow_exist = rag_pipeline_service.is_workflow_exist(pipeline=pipeline)
+        if not workflow_exist:
+            raise DraftWorkflowNotExist()
+
+        # fetch draft workflow by app_model
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+        workflow_vars = draft_var_srv.list_variables_without_values(
+            app_id=pipeline.id,
+            page=args.page,
+            limit=args.limit,
+        )
+
+        return workflow_vars
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        draft_var_srv.delete_workflow_variables(pipeline.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+def validate_node_id(node_id: str) -> NoReturn | None:
+    if node_id in [
+        CONVERSATION_VARIABLE_NODE_ID,
+        SYSTEM_VARIABLE_NODE_ID,
+    ]:
+        # NOTE(QuantumGhost): While we store the system and conversation variables as node variables
+        # with specific `node_id` in database, we still want to make the API separated. By disallowing
+        # accessing system and conversation variables in `WorkflowDraftNodeVariableListApi`,
+        # we mitigate the risk that user of the API depending on the implementation detail of the API.
+        #
+        # ref: [Hyrum's Law](https://www.hyrumslaw.com/)
+
+        raise InvalidArgumentError(
+            f"invalid node_id, please use correspond api for conversation and system variables, node_id={node_id}",
+        )
+    return None
+
+
+class RagPipelineNodeVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, pipeline: Pipeline, node_id: str):
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+            node_vars = draft_var_srv.list_node_variables(pipeline.id, node_id)
+
+        return node_vars
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline, node_id: str):
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(pipeline.id, node_id)
+        db.session.commit()
+        return Response("", 204)
+
+
+class RagPipelineVariableApi(Resource):
+    _PATCH_NAME_FIELD = "name"
+    _PATCH_VALUE_FIELD = "value"
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def get(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        return variable
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def patch(self, pipeline: Pipeline, variable_id: str):
+        # Request payload for file types:
+        #
+        # Local File:
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "local_file",
+        #         "url": "",
+        #         "upload_file_id": "daded54f-72c7-4f8e-9d18-9b0abdd9f190"
+        #     }
+        #
+        # Remote File:
+        #
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "remote_url",
+        #         "url": "http://127.0.0.1:5001/files/1602650a-4fe4-423c-85a2-af76c083e3c4/file-preview?timestamp=1750041099&nonce=...&sign=...=",
+        #         "upload_file_id": "1602650a-4fe4-423c-85a2-af76c083e3c4"
+        #     }
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(self._PATCH_NAME_FIELD, type=str, required=False, nullable=True, location="json")
+        # Parse 'value' field as-is to maintain its original data structure
+        parser.add_argument(self._PATCH_VALUE_FIELD, type=lambda x: x, required=False, nullable=True, location="json")
+
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        args = parser.parse_args(strict=True)
+
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        new_name = args.get(self._PATCH_NAME_FIELD, None)
+        raw_value = args.get(self._PATCH_VALUE_FIELD, None)
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(mapping=raw_value, tenant_id=pipeline.tenant_id)
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(mappings=raw_value, tenant_id=pipeline.tenant_id)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @_api_prerequisite
+    def delete(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+class RagPipelineVariableResetApi(Resource):
+    @_api_prerequisite
+    def put(self, pipeline: Pipeline, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+
+        rag_pipeline_service = RagPipelineService()
+        draft_workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, pipeline_id={pipeline.id}",
+            )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != pipeline.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        else:
+            return marshal(resetted, _WORKFLOW_DRAFT_VARIABLE_FIELDS)
+
+
+def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        draft_var_srv = WorkflowDraftVariableService(
+            session=session,
+        )
+        if node_id == CONVERSATION_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id)
+        elif node_id == SYSTEM_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_system_variables(pipeline.id)
+        else:
+            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id)
+    return draft_vars
+
+
+class RagPipelineSystemVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, pipeline: Pipeline):
+        return _get_variable_list(pipeline, SYSTEM_VARIABLE_NODE_ID)
+
+
+class RagPipelineEnvironmentVariableCollectionApi(Resource):
+    @_api_prerequisite
+    def get(self, pipeline: Pipeline):
+        """
+        Get draft workflow
+        """
+        # fetch draft workflow by app_model
+        rag_pipeline_service = RagPipelineService()
+        workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars = workflow.environment_variables
+        env_vars_list = []
+        for v in env_vars:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.value,
+                    "value": v.value,
+                    # Do not track edited for env vars.
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}
+
+
+api.add_resource(
+    RagPipelineVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables",
+)
+api.add_resource(
+    RagPipelineNodeVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/variables",
+)
+api.add_resource(
+    RagPipelineVariableApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>"
+)
+api.add_resource(
+    RagPipelineVariableResetApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>/reset"
+)
+api.add_resource(
+    RagPipelineSystemVariableCollectionApi, "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/system-variables"
+)
+api.add_resource(
+    RagPipelineEnvironmentVariableCollectionApi,
+    "/rag/pipelines/<uuid:pipeline_id>/workflows/draft/environment-variables",
+)

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

@@ -0,0 +1,147 @@
+from typing import cast
+
+from flask_login import current_user  # type: ignore
+from flask_restful import Resource, marshal_with, reqparse  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.rag_pipeline_fields import pipeline_import_check_dependencies_fields, pipeline_import_fields
+from libs.login import login_required
+from models import Account
+from models.dataset import Pipeline
+from services.app_dsl_service import ImportStatus
+from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
+
+
+class RagPipelineImportApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(pipeline_import_fields)
+    def post(self):
+        # Check user role first
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("mode", type=str, required=True, location="json")
+        parser.add_argument("yaml_content", type=str, location="json")
+        parser.add_argument("yaml_url", type=str, location="json")
+        parser.add_argument("name", type=str, location="json")
+        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("icon_type", type=str, location="json")
+        parser.add_argument("icon", type=str, location="json")
+        parser.add_argument("icon_background", type=str, location="json")
+        parser.add_argument("pipeline_id", type=str, location="json")
+        args = parser.parse_args()
+
+        # Create service with session
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            # Import app
+            account = cast(Account, current_user)
+            result = import_service.import_rag_pipeline(
+                account=account,
+                import_mode=args["mode"],
+                yaml_content=args.get("yaml_content"),
+                yaml_url=args.get("yaml_url"),
+                pipeline_id=args.get("pipeline_id"),
+                dataset_name=args.get("name"),
+            )
+            session.commit()
+
+        # Return appropriate status code based on result
+        status = result.status
+        if status == ImportStatus.FAILED.value:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING.value:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineImportConfirmApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(pipeline_import_fields)
+    def post(self, import_id):
+        # Check user role first
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        # Create service with session
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            # Confirm import
+            account = cast(Account, current_user)
+            result = import_service.confirm_import(import_id=import_id, account=account)
+            session.commit()
+
+        # Return appropriate status code based on result
+        if result.status == ImportStatus.FAILED.value:
+            return result.model_dump(mode="json"), 400
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineImportCheckDependenciesApi(Resource):
+    @setup_required
+    @login_required
+    @get_rag_pipeline
+    @account_initialization_required
+    @marshal_with(pipeline_import_check_dependencies_fields)
+    def get(self, pipeline: Pipeline):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+        with Session(db.engine) as session:
+            import_service = RagPipelineDslService(session)
+            result = import_service.check_dependencies(pipeline=pipeline)
+
+        return result.model_dump(mode="json"), 200
+
+
+class RagPipelineExportApi(Resource):
+    @setup_required
+    @login_required
+    @get_rag_pipeline
+    @account_initialization_required
+    def get(self, pipeline: Pipeline):
+        if not current_user.is_editor:
+            raise Forbidden()
+
+            # Add include_secret params
+        parser = reqparse.RequestParser()
+        parser.add_argument("include_secret", type=bool, default=False, location="args")
+        args = parser.parse_args()
+
+        with Session(db.engine) as session:
+            export_service = RagPipelineDslService(session)
+            result = export_service.export_rag_pipeline_dsl(pipeline=pipeline, include_secret=args["include_secret"])
+
+        return {"data": result}, 200
+
+
+# Import Rag Pipeline
+api.add_resource(
+    RagPipelineImportApi,
+    "/rag/pipelines/imports",
+)
+api.add_resource(
+    RagPipelineImportConfirmApi,
+    "/rag/pipelines/imports/<string:import_id>/confirm",
+)
+api.add_resource(
+    RagPipelineImportCheckDependenciesApi,
+    "/rag/pipelines/imports/<string:pipeline_id>/check-dependencies",
+)
+api.add_resource(
+    RagPipelineExportApi,
+    "/rag/pipelines/<string:pipeline_id>/exports",
+)

api/controllers/console/datasets/upload_file.py

@@ -0,0 +1,62 @@
+from flask_login import current_user
+from flask_restful import Resource
+from werkzeug.exceptions import NotFound
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.file import helpers as file_helpers
+from extensions.ext_database import db
+from models.dataset import Dataset
+from models.model import UploadFile
+from services.dataset_service import DocumentService
+
+
+class UploadFileApi(Resource):
+    @setup_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        """Get upload file."""
+        # check dataset
+        dataset_id = str(dataset_id)
+        dataset = (
+            db.session.query(Dataset)
+            .filter(Dataset.tenant_id == current_user.current_tenant_id, Dataset.id == dataset_id)
+            .first()
+        )
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        # check document
+        document_id = str(document_id)
+        document = DocumentService.get_document(dataset.id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        # check upload file
+        if document.data_source_type != "upload_file":
+            raise ValueError(f"Document data source type ({document.data_source_type}) is not upload_file.")
+        data_source_info = document.data_source_info_dict
+        if data_source_info and "upload_file_id" in data_source_info:
+            file_id = data_source_info["upload_file_id"]
+            upload_file = db.session.query(UploadFile).filter(UploadFile.id == file_id).first()
+            if not upload_file:
+                raise NotFound("UploadFile not found.")
+        else:
+            raise ValueError("Upload file id not found in document data source info.")
+
+        url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id)
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "url": url,
+            "download_url": f"{url}&as_attachment=true",
+            "mime_type": upload_file.mime_type,
+            "created_by": upload_file.created_by,
+            "created_at": upload_file.created_at.timestamp(),
+        }, 200
+
+
+api.add_resource(UploadFileApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/upload-file")

api/controllers/console/datasets/wraps.py

@@ -0,0 +1,43 @@
+from collections.abc import Callable
+from functools import wraps
+from typing import Optional
+
+from controllers.console.datasets.error import PipelineNotFoundError
+from extensions.ext_database import db
+from libs.login import current_user
+from models.dataset import Pipeline
+
+
+def get_rag_pipeline(
+    view: Optional[Callable] = None,
+):
+    def decorator(view_func):
+        @wraps(view_func)
+        def decorated_view(*args, **kwargs):
+            if not kwargs.get("pipeline_id"):
+                raise ValueError("missing pipeline_id in path parameters")
+
+            pipeline_id = kwargs.get("pipeline_id")
+            pipeline_id = str(pipeline_id)
+
+            del kwargs["pipeline_id"]
+
+            pipeline = (
+                db.session.query(Pipeline)
+                .filter(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_user.current_tenant_id)
+                .first()
+            )
+
+            if not pipeline:
+                raise PipelineNotFoundError()
+
+            kwargs["pipeline"] = pipeline
+
+            return view_func(*args, **kwargs)
+
+        return decorated_view
+
+    if view is None:
+        return decorator
+    else:
+        return decorator(view)

api/controllers/console/error.py

@@ -76,30 +76,6 @@ class EmailSendIpLimitError(BaseHTTPException):
     code = 429
 
 
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
 class UnauthorizedAndForceLogout(BaseHTTPException):
     error_code = "unauthorized_and_force_logout"
     description = "Unauthorized and force logout."
@@ -127,7 +103,7 @@ class EducationActivateLimitError(BaseHTTPException):
     code = 429
 
 
-class CompilanceRateLimitError(BaseHTTPException):
-    error_code = "compilance_rate_limit"
+class ComplianceRateLimitError(BaseHTTPException):
+    error_code = "compliance_rate_limit"
     description = "Rate limit exceeded for downloading compliance report."
     code = 429

api/controllers/console/explore/installed_app.py

@@ -58,21 +58,38 @@ class InstalledAppsListApi(Resource):
         # filter out apps that user doesn't have access to
         if FeatureService.get_system_features().webapp_auth.enabled:
             user_id = current_user.id
-            res = []
             app_ids = [installed_app["app"].id for installed_app in installed_app_list]
             webapp_settings = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids)
+
+            # Pre-filter out apps without setting or with sso_verified
+            filtered_installed_apps = []
+            app_id_to_app_code = {}
+
             for installed_app in installed_app_list:
-                webapp_setting = webapp_settings.get(installed_app["app"].id)
-                if not webapp_setting:
+                app_id = installed_app["app"].id
+                webapp_setting = webapp_settings.get(app_id)
+                if not webapp_setting or webapp_setting.access_mode == "sso_verified":
                     continue
-                if webapp_setting.access_mode == "sso_verified":
-                    continue
-                app_code = AppService.get_app_code_by_id(str(installed_app["app"].id))
-                if EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
-                    user_id=user_id,
-                    app_code=app_code,
-                ):
+                app_code = AppService.get_app_code_by_id(str(app_id))
+                app_id_to_app_code[app_id] = app_code
+                filtered_installed_apps.append(installed_app)
+
+            app_codes = list(app_id_to_app_code.values())
+
+            # Batch permission check
+            permissions = EnterpriseService.WebAppAuth.batch_is_user_allowed_to_access_webapps(
+                user_id=user_id,
+                app_codes=app_codes,
+            )
+
+            # Keep only allowed apps
+            res = []
+            for installed_app in filtered_installed_apps:
+                app_id = installed_app["app"].id
+                app_code = app_id_to_app_code[app_id]
+                if permissions.get(app_code):
                     res.append(installed_app)
+
             installed_app_list = res
             logger.debug("installed_app_list: %s, user_id: %s", installed_app_list, user_id)
 

api/controllers/console/files.py

@@ -8,7 +8,13 @@ from werkzeug.exceptions import Forbidden
 import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
-from controllers.common.errors import FilenameNotExistsError
+from controllers.common.errors import (
+    FilenameNotExistsError,
+    FileTooLargeError,
+    NoFileUploadedError,
+    TooManyFilesError,
+    UnsupportedFileTypeError,
+)
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
@@ -18,13 +24,6 @@ from fields.file_fields import file_fields, upload_config_fields
 from libs.login import login_required
 from services.file_service import FileService
 
-from .error import (
-    FileTooLargeError,
-    NoFileUploadedError,
-    TooManyFilesError,
-    UnsupportedFileTypeError,
-)
-
 PREVIEW_WORDS_LIMIT = 3000
 
 
@@ -49,7 +48,6 @@ class FileApi(Resource):
     @marshal_with(file_fields)
     @cloud_edition_billing_resource_check("documents")
     def post(self):
-        file = request.files["file"]
         source_str = request.form.get("source")
         source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
 
@@ -58,6 +56,7 @@ class FileApi(Resource):
 
         if len(request.files) > 1:
             raise TooManyFilesError()
+        file = request.files["file"]
 
         if not file.filename:
             raise FilenameNotExistsError

api/controllers/console/remote_files.py

@@ -7,18 +7,17 @@ from flask_restful import Resource, marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import RemoteFileUploadError
+from controllers.common.errors import (
+    FileTooLargeError,
+    RemoteFileUploadError,
+    UnsupportedFileTypeError,
+)
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from models.account import Account
 from services.file_service import FileService
 
-from .error import (
-    FileTooLargeError,
-    UnsupportedFileTypeError,
-)
-
 
 class RemoteFileInfoApi(Resource):
     @marshal_with(remote_file_info_fields)

api/controllers/console/version.py

@@ -32,7 +32,7 @@ class VersionApi(Resource):
             return result
 
         try:
-            response = requests.get(check_update_url, {"current_version": args.get("current_version")})
+            response = requests.get(check_update_url, {"current_version": args.get("current_version")}, timeout=(3, 10))
         except Exception as error:
             logging.warning("Check update version error: %s.", str(error))
             result["version"] = args.get("current_version")

api/controllers/console/workspace/tool_providers.py

@@ -862,6 +862,10 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_type", type=str, required=True, nullable=False, location="json")
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json", default="")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("timeout", type=float, required=False, nullable=False, location="json", default=30)
+        parser.add_argument(
+            "sse_read_timeout", type=float, required=False, nullable=False, location="json", default=300
+        )
         args = parser.parse_args()
         user = current_user
         if not is_valid_url(args["server_url"]):
@@ -876,6 +880,8 @@ class ToolProviderMCPApi(Resource):
                 icon_background=args["icon_background"],
                 user_id=user.id,
                 server_identifier=args["server_identifier"],
+                timeout=args["timeout"],
+                sse_read_timeout=args["sse_read_timeout"],
             )
         )
 
@@ -891,6 +897,8 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json")
         parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("timeout", type=float, required=False, nullable=True, location="json")
+        parser.add_argument("sse_read_timeout", type=float, required=False, nullable=True, location="json")
         args = parser.parse_args()
         if not is_valid_url(args["server_url"]):
             if "[__HIDDEN__]" in args["server_url"]:
@@ -906,6 +914,8 @@ class ToolProviderMCPApi(Resource):
             icon_type=args["icon_type"],
             icon_background=args["icon_background"],
             server_identifier=args["server_identifier"],
+            timeout=args.get("timeout"),
+            sse_read_timeout=args.get("sse_read_timeout"),
         )
         return {"result": "success"}
 

api/controllers/console/workspace/workspace.py

@@ -7,15 +7,15 @@ from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.console import api
-from controllers.console.admin import admin_required
-from controllers.console.datasets.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.console import api
+from controllers.console.admin import admin_required
 from controllers.console.error import AccountNotLinkTenantError
 from controllers.console.wraps import (
     account_initialization_required,
@@ -191,9 +191,6 @@ class WebappLogoWorkspaceApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("workspace_custom")
     def post(self):
-        # get file from request
-        file = request.files["file"]
-
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -201,6 +198,8 @@ class WebappLogoWorkspaceApi(Resource):
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        # get file from request
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 

api/controllers/console/wraps.py

@@ -261,3 +261,14 @@ def is_allow_transfer_owner(view):
         abort(403)
 
     return decorated
+
+
+def knowledge_pipeline_publish_enabled(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_features(current_user.current_tenant_id)
+        if features.knowledge_pipeline.publish_enabled:
+            return view(*args, **kwargs)
+        abort(403)
+
+    return decorated

api/controllers/files/error.py

@@ -1,7 +0,0 @@
-from libs.exception import BaseHTTPException
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415

api/controllers/files/image_preview.py

@@ -5,8 +5,8 @@ from flask_restful import Resource, reqparse
 from werkzeug.exceptions import NotFound
 
 import services
+from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from services.account_service import TenantService
 from services.file_service import FileService
 

api/controllers/files/tool_files.py

@@ -4,8 +4,8 @@ from flask import Response
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
 
+from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from models import db as global_db

api/controllers/files/upload.py

@@ -5,11 +5,13 @@ from flask_restful import Resource, marshal_with
 from werkzeug.exceptions import Forbidden
 
 import services
+from controllers.common.errors import (
+    FileTooLargeError,
+    UnsupportedFileTypeError,
+)
 from controllers.console.wraps import setup_required
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from controllers.inner_api.plugin.wraps import get_user
-from controllers.service_api.app.error import FileTooLargeError
 from core.file.helpers import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from fields.file_fields import file_fields

api/controllers/service_api/__init__.py

@@ -6,6 +6,6 @@ bp = Blueprint("service_api", __name__, url_prefix="/v1")
 api = ExternalApi(bp)
 
 from . import index
-from .app import annotation, app, audio, completion, conversation, file, message, site, workflow
+from .app import annotation, app, audio, completion, conversation, file, file_preview, message, site, workflow
 from .dataset import dataset, document, hit_testing, metadata, segment, upload_file
 from .workspace import models

api/controllers/service_api/app/completion.py

@@ -2,7 +2,7 @@ import logging
 
 from flask import request
 from flask_restful import Resource, reqparse
-from werkzeug.exceptions import InternalServerError, NotFound
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 import services
 from controllers.service_api import api
@@ -30,6 +30,7 @@ from libs import helper
 from libs.helper import uuid_value
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
+from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 
 
@@ -113,7 +114,7 @@ class ChatApi(Resource):
         parser.add_argument("conversation_id", type=uuid_value, location="json")
         parser.add_argument("retriever_from", type=str, required=False, default="dev", location="json")
         parser.add_argument("auto_generate_name", type=bool, required=False, default=True, location="json")
-
+        parser.add_argument("workflow_id", type=str, required=False, location="json")
         args = parser.parse_args()
 
         external_trace_id = get_external_trace_id(request)
@@ -128,6 +129,12 @@ class ChatApi(Resource):
             )
 
             return helper.compact_generate_response(response)
+        except WorkflowNotFoundError as ex:
+            raise NotFound(str(ex))
+        except IsDraftWorkflowError as ex:
+            raise BadRequest(str(ex))
+        except WorkflowIdFormatError as ex:
+            raise BadRequest(str(ex))
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
         except services.errors.conversation.ConversationCompletedError:

api/controllers/service_api/app/conversation.py

@@ -1,5 +1,3 @@
-import json
-
 from flask_restful import Resource, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from sqlalchemy.orm import Session
@@ -136,12 +134,15 @@ class ConversationVariableDetailApi(Resource):
         variable_id = str(variable_id)
 
         parser = reqparse.RequestParser()
-        parser.add_argument("value", required=True, location="json")
+        # using lambda is for passing the already-typed value without modification
+        # if no lambda, it will be converted to string
+        # the string cannot be converted using json.loads
+        parser.add_argument("value", required=True, location="json", type=lambda x: x)
         args = parser.parse_args()
 
         try:
             return ConversationService.update_conversation_variable(
-                app_model, conversation_id, variable_id, end_user, json.loads(args["value"])
+                app_model, conversation_id, variable_id, end_user, args["value"]
             )
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")

api/controllers/service_api/app/error.py

@@ -85,25 +85,13 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
+class FileNotFoundError(BaseHTTPException):
+    error_code = "file_not_found"
+    description = "The requested file was not found."
+    code = 404
 
 
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
+class FileAccessDeniedError(BaseHTTPException):
+    error_code = "file_access_denied"
+    description = "Access to the requested file is denied."
+    code = 403

api/controllers/service_api/app/file.py

@@ -2,14 +2,14 @@ from flask import request
 from flask_restful import Resource, marshal_with
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.service_api import api
-from controllers.service_api.app.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.service_api import api
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from fields.file_fields import file_fields
 from models.model import App, EndUser
@@ -20,18 +20,17 @@ class FileApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.FORM))
     @marshal_with(file_fields)
     def post(self, app_model: App, end_user: EndUser):
-        file = request.files["file"]
-
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
 
-        if not file.mimetype:
-            raise UnsupportedFileTypeError()
-
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        file = request.files["file"]
+        if not file.mimetype:
+            raise UnsupportedFileTypeError()
+
         if not file.filename:
             raise FilenameNotExistsError
 

api/controllers/service_api/app/file_preview.py

@@ -0,0 +1,186 @@
+import logging
+from urllib.parse import quote
+
+from flask import Response
+from flask_restful import Resource, reqparse
+
+from controllers.service_api import api
+from controllers.service_api.app.error import (
+    FileAccessDeniedError,
+    FileNotFoundError,
+)
+from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from models.model import App, EndUser, Message, MessageFile, UploadFile
+
+logger = logging.getLogger(__name__)
+
+
+class FilePreviewApi(Resource):
+    """
+    Service API File Preview endpoint
+
+    Provides secure file preview/download functionality for external API users.
+    Files can only be accessed if they belong to messages within the requesting app's context.
+    """
+
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
+    def get(self, app_model: App, end_user: EndUser, file_id: str):
+        """
+        Preview/Download a file that was uploaded via Service API
+
+        Args:
+            app_model: The authenticated app model
+            end_user: The authenticated end user (optional)
+            file_id: UUID of the file to preview
+
+        Query Parameters:
+            user: Optional user identifier
+            as_attachment: Boolean, whether to download as attachment (default: false)
+
+        Returns:
+            Stream response with file content
+
+        Raises:
+            FileNotFoundError: File does not exist
+            FileAccessDeniedError: File access denied (not owned by app)
+        """
+        file_id = str(file_id)
+
+        # Parse query parameters
+        parser = reqparse.RequestParser()
+        parser.add_argument("as_attachment", type=bool, required=False, default=False, location="args")
+        args = parser.parse_args()
+
+        # Validate file ownership and get file objects
+        message_file, upload_file = self._validate_file_ownership(file_id, app_model.id)
+
+        # Get file content generator
+        try:
+            generator = storage.load(upload_file.key, stream=True)
+        except Exception as e:
+            raise FileNotFoundError(f"Failed to load file content: {str(e)}")
+
+        # Build response with appropriate headers
+        response = self._build_file_response(generator, upload_file, args["as_attachment"])
+
+        return response
+
+    def _validate_file_ownership(self, file_id: str, app_id: str) -> tuple[MessageFile, UploadFile]:
+        """
+        Validate that the file belongs to a message within the requesting app's context
+
+        Security validations performed:
+        1. File exists in MessageFile table (was used in a conversation)
+        2. Message belongs to the requesting app
+        3. UploadFile record exists and is accessible
+        4. File tenant matches app tenant (additional security layer)
+
+        Args:
+            file_id: UUID of the file to validate
+            app_id: UUID of the requesting app
+
+        Returns:
+            Tuple of (MessageFile, UploadFile) if validation passes
+
+        Raises:
+            FileNotFoundError: File or related records not found
+            FileAccessDeniedError: File does not belong to the app's context
+        """
+        try:
+            # Input validation
+            if not file_id or not app_id:
+                raise FileAccessDeniedError("Invalid file or app identifier")
+
+            # First, find the MessageFile that references this upload file
+            message_file = db.session.query(MessageFile).where(MessageFile.upload_file_id == file_id).first()
+
+            if not message_file:
+                raise FileNotFoundError("File not found in message context")
+
+            # Get the message and verify it belongs to the requesting app
+            message = (
+                db.session.query(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).first()
+            )
+
+            if not message:
+                raise FileAccessDeniedError("File access denied: not owned by requesting app")
+
+            # Get the actual upload file record
+            upload_file = db.session.query(UploadFile).where(UploadFile.id == file_id).first()
+
+            if not upload_file:
+                raise FileNotFoundError("Upload file record not found")
+
+            # Additional security: verify tenant isolation
+            app = db.session.query(App).where(App.id == app_id).first()
+            if app and upload_file.tenant_id != app.tenant_id:
+                raise FileAccessDeniedError("File access denied: tenant mismatch")
+
+            return message_file, upload_file
+
+        except (FileNotFoundError, FileAccessDeniedError):
+            # Re-raise our custom exceptions
+            raise
+        except Exception as e:
+            # Log unexpected errors for debugging
+            logger.exception(
+                "Unexpected error during file ownership validation",
+                extra={"file_id": file_id, "app_id": app_id, "error": str(e)},
+            )
+            raise FileAccessDeniedError("File access validation failed")
+
+    def _build_file_response(self, generator, upload_file: UploadFile, as_attachment: bool = False) -> Response:
+        """
+        Build Flask Response object with appropriate headers for file streaming
+
+        Args:
+            generator: File content generator from storage
+            upload_file: UploadFile database record
+            as_attachment: Whether to set Content-Disposition as attachment
+
+        Returns:
+            Flask Response object with streaming file content
+        """
+        response = Response(
+            generator,
+            mimetype=upload_file.mime_type,
+            direct_passthrough=True,
+            headers={},
+        )
+
+        # Add Content-Length if known
+        if upload_file.size and upload_file.size > 0:
+            response.headers["Content-Length"] = str(upload_file.size)
+
+        # Add Accept-Ranges header for audio/video files to support seeking
+        if upload_file.mime_type in [
+            "audio/mpeg",
+            "audio/wav",
+            "audio/mp4",
+            "audio/ogg",
+            "audio/flac",
+            "audio/aac",
+            "video/mp4",
+            "video/webm",
+            "video/quicktime",
+            "audio/x-m4a",
+        ]:
+            response.headers["Accept-Ranges"] = "bytes"
+
+        # Set Content-Disposition for downloads
+        if as_attachment and upload_file.name:
+            encoded_filename = quote(upload_file.name)
+            response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+            # Override content-type for downloads to force download
+            response.headers["Content-Type"] = "application/octet-stream"
+
+        # Add caching headers for performance
+        response.headers["Cache-Control"] = "public, max-age=3600"  # Cache for 1 hour
+
+        return response
+
+
+# Register the API endpoint
+api.add_resource(FilePreviewApi, "/files/<uuid:file_id>/preview")

api/controllers/service_api/app/workflow.py

@@ -5,7 +5,7 @@ from flask import request
 from flask_restful import Resource, fields, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from sqlalchemy.orm import Session, sessionmaker
-from werkzeug.exceptions import InternalServerError
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 from controllers.service_api import api
 from controllers.service_api.app.error import (
@@ -34,6 +34,7 @@ from libs.helper import TimestampField
 from models.model import App, AppMode, EndUser
 from repositories.factory import DifyAPIRepositoryFactory
 from services.app_generate_service import AppGenerateService
+from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_app_service import WorkflowAppService
 
@@ -120,6 +121,59 @@ class WorkflowRunApi(Resource):
             raise InternalServerError()
 
 
+class WorkflowRunByIdApi(Resource):
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
+    def post(self, app_model: App, end_user: EndUser, workflow_id: str):
+        """
+        Run specific workflow by ID
+        """
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode != AppMode.WORKFLOW:
+            raise NotWorkflowAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+        args = parser.parse_args()
+
+        # Add workflow_id to args for AppGenerateService
+        args["workflow_id"] = workflow_id
+
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+        streaming = args.get("response_mode") == "streaming"
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model, user=end_user, args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=streaming
+            )
+
+            return helper.compact_generate_response(response)
+        except WorkflowNotFoundError as ex:
+            raise NotFound(str(ex))
+        except IsDraftWorkflowError as ex:
+            raise BadRequest(str(ex))
+        except WorkflowIdFormatError as ex:
+            raise BadRequest(str(ex))
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logging.exception("internal server error.")
+            raise InternalServerError()
+
+
 class WorkflowTaskStopApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
     def post(self, app_model: App, end_user: EndUser, task_id: str):
@@ -193,5 +247,6 @@ class WorkflowAppLogApi(Resource):
 
 api.add_resource(WorkflowRunApi, "/workflows/run")
 api.add_resource(WorkflowRunDetailApi, "/workflows/run/<string:workflow_run_id>")
+api.add_resource(WorkflowRunByIdApi, "/workflows/<string:workflow_id>/run")
 api.add_resource(WorkflowTaskStopApi, "/workflows/tasks/<string:task_id>/stop")
 api.add_resource(WorkflowAppLogApi, "/workflows/logs")

api/controllers/service_api/dataset/dataset.py

@@ -29,7 +29,7 @@ def _validate_name(name):
 
 
 def _validate_description_length(description):
-    if len(description) > 400:
+    if description and len(description) > 400:
         raise ValueError("Description cannot exceed 400 characters.")
     return description
 
@@ -87,7 +87,7 @@ class DatasetListApi(DatasetApiResource):
         )
         parser.add_argument(
             "description",
-            type=str,
+            type=_validate_description_length,
             nullable=True,
             required=False,
             default="",

api/controllers/service_api/dataset/document.py

@@ -6,15 +6,15 @@ from sqlalchemy import desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.service_api import api
-from controllers.service_api.app.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
-    ProviderNotInitializeError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.service_api import api
+from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.dataset.error import (
     ArchivedDocumentImmutableError,
     DocumentIndexingError,
@@ -234,8 +234,6 @@ class DocumentAddByFileApi(DatasetApiResource):
                 args["retrieval_model"].get("reranking_model").get("reranking_model_name"),
             )
 
-        # save file info
-        file = request.files["file"]
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -243,6 +241,8 @@ class DocumentAddByFileApi(DatasetApiResource):
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        # save file info
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 
@@ -358,39 +358,6 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         return documents_and_batch_fields, 200
 
 
-class DocumentDeleteApi(DatasetApiResource):
-    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def delete(self, tenant_id, dataset_id, document_id):
-        """Delete document."""
-        document_id = str(document_id)
-        dataset_id = str(dataset_id)
-        tenant_id = str(tenant_id)
-
-        # get dataset info
-        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
-
-        if not dataset:
-            raise ValueError("Dataset does not exist.")
-
-        document = DocumentService.get_document(dataset.id, document_id)
-
-        # 404 if document not found
-        if document is None:
-            raise NotFound("Document Not Exists.")
-
-        # 403 if document is archived
-        if DocumentService.check_archived(document):
-            raise ArchivedDocumentImmutableError()
-
-        try:
-            # delete document
-            DocumentService.delete_document(document)
-        except services.errors.document.DocumentIndexingError:
-            raise DocumentIndexingError("Cannot delete document during indexing.")
-
-        return 204
-
-
 class DocumentListApi(DatasetApiResource):
     def get(self, tenant_id, dataset_id):
         dataset_id = str(dataset_id)
@@ -473,7 +440,7 @@ class DocumentIndexingStatusApi(DatasetApiResource):
         return data
 
 
-class DocumentDetailApi(DatasetApiResource):
+class DocumentApi(DatasetApiResource):
     METADATA_CHOICES = {"all", "only", "without"}
 
     def get(self, tenant_id, dataset_id, document_id):
@@ -567,6 +534,37 @@ class DocumentDetailApi(DatasetApiResource):
 
         return response
 
+    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
+    def delete(self, tenant_id, dataset_id, document_id):
+        """Delete document."""
+        document_id = str(document_id)
+        dataset_id = str(dataset_id)
+        tenant_id = str(tenant_id)
+
+        # get dataset info
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
+
+        if not dataset:
+            raise ValueError("Dataset does not exist.")
+
+        document = DocumentService.get_document(dataset.id, document_id)
+
+        # 404 if document not found
+        if document is None:
+            raise NotFound("Document Not Exists.")
+
+        # 403 if document is archived
+        if DocumentService.check_archived(document):
+            raise ArchivedDocumentImmutableError()
+
+        try:
+            # delete document
+            DocumentService.delete_document(document)
+        except services.errors.document.DocumentIndexingError:
+            raise DocumentIndexingError("Cannot delete document during indexing.")
+
+        return 204
+
 
 api.add_resource(
     DocumentAddByTextApi,
@@ -588,7 +586,6 @@ api.add_resource(
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_file",
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-file",
 )
-api.add_resource(DocumentDeleteApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
+api.add_resource(DocumentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(DocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DocumentIndexingStatusApi, "/datasets/<uuid:dataset_id>/documents/<string:batch>/indexing-status")
-api.add_resource(DocumentDetailApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")

api/controllers/service_api/dataset/error.py

@@ -1,30 +1,6 @@
 from libs.exception import BaseHTTPException
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/web/app.py

@@ -1,5 +1,6 @@
 from flask import request
 from flask_restful import Resource, marshal_with, reqparse
+from werkzeug.exceptions import Unauthorized
 
 from controllers.common import fields
 from controllers.web import api
@@ -75,14 +76,14 @@ class AppWebAuthPermission(Resource):
         try:
             auth_header = request.headers.get("Authorization")
             if auth_header is None:
-                raise
+                raise Unauthorized("Authorization header is missing.")
             if " " not in auth_header:
-                raise
+                raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
 
             auth_scheme, tk = auth_header.split(None, 1)
             auth_scheme = auth_scheme.lower()
             if auth_scheme != "bearer":
-                raise
+                raise Unauthorized("Authorization scheme must be 'Bearer'")
 
             decoded = PassportService().verify(tk)
             user_id = decoded.get("user_id", "visitor")

api/controllers/web/error.py

@@ -97,30 +97,6 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 class WebAppAuthRequiredError(BaseHTTPException):
     error_code = "web_sso_auth_required"
     description = "Web app authentication required."

api/controllers/web/files.py

@@ -2,8 +2,13 @@ from flask import request
 from flask_restful import marshal_with
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.web.error import FileTooLargeError, NoFileUploadedError, TooManyFilesError, UnsupportedFileTypeError
+from controllers.common.errors import (
+    FilenameNotExistsError,
+    FileTooLargeError,
+    NoFileUploadedError,
+    TooManyFilesError,
+    UnsupportedFileTypeError,
+)
 from controllers.web.wraps import WebApiResource
 from fields.file_fields import file_fields
 from services.file_service import FileService
@@ -12,18 +17,17 @@ from services.file_service import FileService
 class FileApi(WebApiResource):
     @marshal_with(file_fields)
     def post(self, app_model, end_user):
-        file = request.files["file"]
-        source = request.form.get("source")
-
         if "file" not in request.files:
             raise NoFileUploadedError()
 
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 
+        source = request.form.get("source")
         if source not in ("datasets", None):
             source = None
 

api/controllers/web/remote_files.py

@@ -5,15 +5,17 @@ from flask_restful import marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import RemoteFileUploadError
+from controllers.common.errors import (
+    FileTooLargeError,
+    RemoteFileUploadError,
+    UnsupportedFileTypeError,
+)
 from controllers.web.wraps import WebApiResource
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from services.file_service import FileService
 
-from .error import FileTooLargeError, UnsupportedFileTypeError
-
 
 class RemoteFileInfoApi(WebApiResource):
     @marshal_with(remote_file_info_fields)

api/core/app/app_config/entities.py

@@ -97,6 +97,7 @@ class VariableEntityType(StrEnum):
     EXTERNAL_DATA_TOOL = "external_data_tool"
     FILE = "file"
     FILE_LIST = "file-list"
+    CHECKBOX = "checkbox"
 
 
 class VariableEntity(BaseModel):
@@ -113,9 +114,9 @@ class VariableEntity(BaseModel):
     hide: bool = False
     max_length: Optional[int] = None
     options: Sequence[str] = Field(default_factory=list)
-    allowed_file_types: Sequence[FileType] = Field(default_factory=list)
-    allowed_file_extensions: Sequence[str] = Field(default_factory=list)
-    allowed_file_upload_methods: Sequence[FileTransferMethod] = Field(default_factory=list)
+    allowed_file_types: Optional[Sequence[FileType]] = Field(default_factory=list)
+    allowed_file_extensions: Optional[Sequence[str]] = Field(default_factory=list)
+    allowed_file_upload_methods: Optional[Sequence[FileTransferMethod]] = Field(default_factory=list)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -128,6 +129,16 @@ class VariableEntity(BaseModel):
         return v or []
 
 
+class RagPipelineVariableEntity(VariableEntity):
+    """
+    Rag Pipeline Variable Entity.
+    """
+
+    tooltips: Optional[str] = None
+    placeholder: Optional[str] = None
+    belong_to_node_id: str
+
+
 class ExternalDataVariableEntity(BaseModel):
     """
     External Data Variable Entity.
@@ -287,7 +298,7 @@ class AppConfig(BaseModel):
     tenant_id: str
     app_id: str
     app_mode: AppMode
-    additional_features: AppAdditionalFeatures
+    additional_features: Optional[AppAdditionalFeatures] = None
     variables: list[VariableEntity] = []
     sensitive_word_avoidance: Optional[SensitiveWordAvoidanceEntity] = None
 

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

@@ -1,4 +1,6 @@
-from core.app.app_config.entities import VariableEntity
+import re
+
+from core.app.app_config.entities import RagPipelineVariableEntity, VariableEntity
 from models.workflow import Workflow
 
 
@@ -20,3 +22,44 @@ class WorkflowVariablesConfigManager:
             variables.append(VariableEntity.model_validate(variable))
 
         return variables
+
+    @classmethod
+    def convert_rag_pipeline_variable(cls, workflow: Workflow, start_node_id: str) -> list[RagPipelineVariableEntity]:
+        """
+        Convert workflow start variables to variables
+
+        :param workflow: workflow instance
+        """
+        variables = []
+
+        # get second step node
+        rag_pipeline_variables = workflow.rag_pipeline_variables
+        if not rag_pipeline_variables:
+            return []
+        variables_map = {item["variable"]: item for item in rag_pipeline_variables}
+
+        # get datasource node data
+        datasource_node_data = None
+        datasource_nodes = workflow.graph_dict.get("nodes", [])
+        for datasource_node in datasource_nodes:
+            if datasource_node.get("id") == start_node_id:
+                datasource_node_data = datasource_node.get("data", {})
+                break
+        if datasource_node_data:
+            datasource_parameters = datasource_node_data.get("datasource_parameters", {})
+
+            for key, value in datasource_parameters.items():
+                if value.get("value") and isinstance(value.get("value"), str):
+                    pattern = r"\{\{#([a-zA-Z0-9_]{1,50}(?:\.[a-zA-Z0-9_][a-zA-Z0-9_]{0,29}){1,10})#\}\}"
+                    match = re.match(pattern, value["value"])
+                    if match:
+                        full_path = match.group(1)
+                        last_part = full_path.split(".")[-1]
+                        variables_map.pop(last_part)
+        all_second_step_variables = list(variables_map.values())
+
+        for item in all_second_step_variables:
+            if item.get("belong_to_node_id") == start_node_id or item.get("belong_to_node_id") == "shared":
+                variables.append(RagPipelineVariableEntity.model_validate(item))
+
+        return variables

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

@@ -118,26 +118,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             ):
                 return
 
-            # Init conversation variables
-            stmt = select(ConversationVariable).where(
-                ConversationVariable.app_id == self.conversation.app_id,
-                ConversationVariable.conversation_id == self.conversation.id,
-            )
-            with Session(db.engine) as session:
-                db_conversation_variables = session.scalars(stmt).all()
-                if not db_conversation_variables:
-                    # Create conversation variables if they don't exist.
-                    db_conversation_variables = [
-                        ConversationVariable.from_variable(
-                            app_id=self.conversation.app_id, conversation_id=self.conversation.id, variable=variable
-                        )
-                        for variable in self._workflow.conversation_variables
-                    ]
-                    session.add_all(db_conversation_variables)
-                # Convert database entities to variables.
-                conversation_variables = [item.to_variable() for item in db_conversation_variables]
-
-                session.commit()
+            # Initialize conversation variables
+            conversation_variables = self._initialize_conversation_variables()
 
             # Create a variable pool.
             system_inputs = SystemVariable(
@@ -292,3 +274,100 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             message_id=message_id,
             trace_manager=app_generate_entity.trace_manager,
         )
+
+    def _initialize_conversation_variables(self) -> list[VariableUnion]:
+        """
+        Initialize conversation variables for the current conversation.
+
+        This method:
+        1. Loads existing variables from the database
+        2. Creates new variables if none exist
+        3. Syncs missing variables from the workflow definition
+
+        :return: List of conversation variables ready for use
+        """
+        with Session(db.engine) as session:
+            existing_variables = self._load_existing_conversation_variables(session)
+
+            if not existing_variables:
+                # First time initialization - create all variables
+                existing_variables = self._create_all_conversation_variables(session)
+            else:
+                # Check and add any missing variables from the workflow
+                existing_variables = self._sync_missing_conversation_variables(session, existing_variables)
+
+            # Convert to Variable objects for use in the workflow
+            conversation_variables = [var.to_variable() for var in existing_variables]
+
+            session.commit()
+            return cast(list[VariableUnion], conversation_variables)
+
+    def _load_existing_conversation_variables(self, session: Session) -> list[ConversationVariable]:
+        """
+        Load existing conversation variables from the database.
+
+        :param session: Database session
+        :return: List of existing conversation variables
+        """
+        stmt = select(ConversationVariable).where(
+            ConversationVariable.app_id == self.conversation.app_id,
+            ConversationVariable.conversation_id == self.conversation.id,
+        )
+        return list(session.scalars(stmt).all())
+
+    def _create_all_conversation_variables(self, session: Session) -> list[ConversationVariable]:
+        """
+        Create all conversation variables for a new conversation.
+
+        :param session: Database session
+        :return: List of created conversation variables
+        """
+        new_variables = [
+            ConversationVariable.from_variable(
+                app_id=self.conversation.app_id, conversation_id=self.conversation.id, variable=variable
+            )
+            for variable in self._workflow.conversation_variables
+        ]
+
+        if new_variables:
+            session.add_all(new_variables)
+
+        return new_variables
+
+    def _sync_missing_conversation_variables(
+        self, session: Session, existing_variables: list[ConversationVariable]
+    ) -> list[ConversationVariable]:
+        """
+        Sync missing conversation variables from the workflow definition.
+
+        This handles the case where new variables are added to a workflow
+        after conversations have already been created.
+
+        :param session: Database session
+        :param existing_variables: List of existing conversation variables
+        :return: Updated list including any newly created variables
+        """
+        # Get IDs of existing and workflow variables
+        existing_ids = {var.id for var in existing_variables}
+        workflow_variables = {var.id: var for var in self._workflow.conversation_variables}
+
+        # Find missing variable IDs
+        missing_ids = set(workflow_variables.keys()) - existing_ids
+
+        if not missing_ids:
+            return existing_variables
+
+        # Create missing variables with their default values
+        new_variables = [
+            ConversationVariable.from_variable(
+                app_id=self.conversation.app_id,
+                conversation_id=self.conversation.id,
+                variable=workflow_variables[var_id],
+            )
+            for var_id in missing_ids
+        ]
+
+        session.add_all(new_variables)
+
+        # Return combined list
+        return existing_variables + new_variables

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

@@ -74,6 +74,7 @@ from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from models import Conversation, EndUser, Message, MessageFile
 from models.account import Account
 from models.enums import CreatorUserRole
@@ -568,7 +569,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_partial_success_event(
         self,
@@ -600,7 +601,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_failed_event(
         self,
@@ -845,7 +846,7 @@ class AdvancedChatAppGenerateTaskPipeline:
         # Initialize graph runtime state
         graph_runtime_state: Optional[GraphRuntimeState] = None
 
-        for queue_message in self._base_task_pipeline._queue_manager.listen():
+        for queue_message in self._base_task_pipeline.queue_manager.listen():
             event = queue_message.event
 
             match event:
@@ -896,6 +897,7 @@ class AdvancedChatAppGenerateTaskPipeline:
     def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
         message = self._get_message(session=session)
         message.answer = self._task_state.answer
+        message.updated_at = naive_utc_now()
         message.provider_response_latency = time.perf_counter() - self._base_task_pipeline._start_at
         message.message_metadata = self._task_state.metadata.model_dump_json()
         message_files = [
@@ -959,11 +961,11 @@ class AdvancedChatAppGenerateTaskPipeline:
         if self._base_task_pipeline._output_moderation_handler:
             if self._base_task_pipeline._output_moderation_handler.should_direct_output():
                 self._task_state.answer = self._base_task_pipeline._output_moderation_handler.get_final_output()
-                self._base_task_pipeline._queue_manager.publish(
+                self._base_task_pipeline.queue_manager.publish(
                     QueueTextChunkEvent(text=self._task_state.answer), PublishFrom.TASK_PIPELINE
                 )
 
-                self._base_task_pipeline._queue_manager.publish(
+                self._base_task_pipeline.queue_manager.publish(
                     QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
                 )
                 return True

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

@@ -140,7 +140,9 @@ class ChatAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(app_id=app_model.id)
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
         # init application generate entity
         application_generate_entity = ChatAppGenerateEntity(

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

@@ -43,11 +43,13 @@ from core.app.entities.task_entities import (
     WorkflowStartStreamResponse,
 )
 from core.file import FILE_MODEL_IDENTITY, File
+from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.tool_manager import ToolManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities.workflow_execution import WorkflowExecution
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
 from core.workflow.nodes import NodeType
+from core.workflow.nodes.datasource.entities import DatasourceNodeData
 from core.workflow.nodes.tool.entities import ToolNodeData
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from models import (
@@ -183,6 +185,14 @@ class WorkflowResponseConverter:
                 provider_type=node_data.provider_type,
                 provider_id=node_data.provider_id,
             )
+        elif event.node_type == NodeType.DATASOURCE:
+            node_data = cast(DatasourceNodeData, event.node_data)
+            manager = PluginDatasourceManager()
+            provider_entity = manager.fetch_datasource_provider(
+                self._application_generate_entity.app_config.tenant_id,
+                f"{node_data.plugin_id}/{node_data.provider_name}",
+            )
+            response.data.extras["icon"] = provider_entity.declaration.identity.icon
 
         return response
 

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

@@ -124,7 +124,9 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(app_model.id)
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
         # init application generate entity
         application_generate_entity = CompletionAppGenerateEntity(

api/core/app/apps/message_based_app_queue_manager.py

@@ -6,7 +6,6 @@ from core.app.entities.queue_entities import (
     MessageQueueMessage,
     QueueAdvancedChatMessageEndEvent,
     QueueErrorEvent,
-    QueueMessage,
     QueueMessageEndEvent,
     QueueStopEvent,
 )
@@ -22,15 +21,6 @@ class MessageBasedAppQueueManager(AppQueueManager):
         self._app_mode = app_mode
         self._message_id = str(message_id)
 
-    def construct_queue_message(self, event: AppQueueEvent) -> QueueMessage:
-        return MessageQueueMessage(
-            task_id=self._task_id,
-            message_id=self._message_id,
-            conversation_id=self._conversation_id,
-            app_mode=self._app_mode,
-            event=event,
-        )
-
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
         """
         Publish event to queue