chore: workflow performance

Signed-off-by: yihong0618 <zouzou0208@gmail.com>

.devcontainer/Dockerfile

@@ -1,4 +1,4 @@
-FROM mcr.microsoft.com/devcontainers/python:3.12
+FROM mcr.microsoft.com/devcontainers/python:3.12-bookworm
 
 RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
      && apt-get -y install libgmp-dev libmpfr-dev libmpc-dev

.devcontainer/post_create_command.sh

@@ -1,15 +1,16 @@
 #!/bin/bash
+WORKSPACE_ROOT=$(pwd)
 
 corepack enable
 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,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
-echo 'alias stop-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down"' >> ~/.bashrc
+echo "alias start-api=\"cd $WORKSPACE_ROOT/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug\"" >> ~/.bashrc
+echo "alias start-worker=\"cd $WORKSPACE_ROOT/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 $WORKSPACE_ROOT/web && pnpm dev\"" >> ~/.bashrc
+echo "alias start-web-prod=\"cd $WORKSPACE_ROOT/web && pnpm build && pnpm start\"" >> ~/.bashrc
+echo "alias start-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d\"" >> ~/.bashrc
+echo "alias stop-containers=\"cd $WORKSPACE_ROOT/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down\"" >> ~/.bashrc
 
 source /home/vscode/.bashrc
 

.github/ISSUE_TEMPLATE/config.yml

@@ -1,5 +1,8 @@
 blank_issues_enabled: false
 contact_links:
+  - name: "\U0001F510 Security Vulnerabilities"
+    url: "https://github.com/langgenius/dify/security/advisories/new"
+    about: Report security vulnerabilities through GitHub Security Advisories to ensure responsible disclosure. 💡 Please do not report security vulnerabilities in public issues.
   - name: "\U0001F4A1 Model Providers & Plugins"
     url: "https://github.com/langgenius/dify-official-plugins/issues/new/choose"
     about: Report issues with official plugins or model providers, you will need to provide the plugin version and other relevant details.

.github/workflows/autofix.yml

@@ -15,10 +15,12 @@ jobs:
       # Use uv to ensure we have the same ruff version in CI and locally.
       - uses: astral-sh/setup-uv@v6
         with:
-          python-version: "3.12"
+          python-version: "3.11"
       - run: |
           cd api
           uv sync --dev
+          # fmt first to avoid line too long
+          uv run ruff format ..
           # Fix lint errors
           uv run ruff check --fix .
           # Format code

.github/workflows/build-push.yml

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

.github/workflows/deploy-dev.yml

@@ -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/dev'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.SSH_HOST }}
+          host: ${{ secrets.RAG_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/style.yml

@@ -102,7 +102,6 @@ jobs:
         working-directory: ./web
         run: |
           pnpm run lint
-          pnpm run eslint
 
   docker-compose-template:
     name: Docker Compose Template

.gitignore

@@ -230,4 +230,8 @@ api/.env.backup
 
 # Benchmark
 scripts/stress-test/setup/config/
-scripts/stress-test/reports/
+scripts/stress-test/reports/
+
+# mcp
+.playwright-mcp/
+.serena/

AGENTS.md

@@ -4,84 +4,51 @@
 
 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:
+The codebase is split into:
 
-- **Backend API** (`/api`): Python Flask application with Domain-Driven Design architecture
-- **Frontend Web** (`/web`): Next.js 15 application with TypeScript and React 19
+- **Backend API** (`/api`): Python Flask application organized with Domain-Driven Design
+- **Frontend Web** (`/web`): Next.js 15 application using TypeScript and React 19
 - **Docker deployment** (`/docker`): Containerized deployment configurations
 
-## Development Commands
+## Backend Workflow
 
-### Backend (API)
+- Run backend CLI commands through `uv run --project api <command>`.
 
-All Python commands must be prefixed with `uv run --project api`:
+- Backend QA gate requires passing `make lint`, `make type-check`, and `uv run --project api --dev dev/pytest/pytest_unit_tests.sh` before review.
 
-```bash
-# Start development servers
-./dev/start-api                   # Start API server
-./dev/start-worker                # Start Celery worker
+- Use Makefile targets for linting and formatting; `make lint` and `make type-check` cover the required checks.
 
-# 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
+- Integration tests are CI-only and are not expected to run in the local environment.
 
-# 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 --directory api basedpyright         # Type checking
-```
-
-### Frontend (Web)
+## Frontend Workflow
 
 ```bash
 cd web
-pnpm lint                         # Run ESLint
-pnpm eslint-fix                   # Fix ESLint issues
-pnpm test                         # Run Jest tests
+pnpm lint
+pnpm lint:fix
+pnpm test
 ```
 
-## Testing Guidelines
+## Testing & Quality Practices
 
-### Backend Testing
+- Follow TDD: red → green → refactor.
+- Use `pytest` for backend tests with Arrange-Act-Assert structure.
+- Enforce strong typing; avoid `Any` and prefer explicit type annotations.
+- Write self-documenting code; only add comments that explain intent.
 
-- Use `pytest` for all backend tests
-- Write tests first (TDD approach)
-- Test structure: Arrange-Act-Assert
+## Language Style
 
-## Code Style Requirements
+- **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`).
+- **TypeScript**: Use the strict config, lean on ESLint + Prettier workflows, and avoid `any` types.
 
-### Python
+## General Practices
 
-- Use type hints for all functions and class attributes
-- No `Any` types unless absolutely necessary
-- Implement special methods (`__repr__`, `__str__`) appropriately
+- Prefer editing existing files; add new documentation only when requested.
+- Inject dependencies through constructors and preserve clean architecture boundaries.
+- Handle errors with domain-specific exceptions at the correct layer.
 
-### TypeScript/JavaScript
+## Project Conventions
 
-- 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/`
-1. Add service logic in `/api/services/`
-1. Update routes in controller's `__init__.py`
-1. Write tests in `/api/tests/`
-
-## Project-Specific Conventions
-
-- All async tasks use Celery with Redis as broker
-- **Internationalization**: Frontend supports multiple languages with English (`web/i18n/en-US/`) as the source. All user-facing text must use i18n keys, no hardcoded strings. Edit corresponding module files in `en-US/` directory for translations.
+- Backend architecture adheres to DDD and Clean Architecture principles.
+- Async work runs through Celery with Redis as the broker.
+- Frontend user-facing strings must use `web/i18n/en-US/`; avoid hardcoded text.

Makefile

@@ -26,7 +26,6 @@ prepare-web:
 	@echo "🌐 Setting up web environment..."
 	@cp -n web/.env.example web/.env 2>/dev/null || echo "Web .env already exists"
 	@cd web && pnpm install
-	@cd web && pnpm build
 	@echo "✅ Web environment prepared (not started)"
 
 # Step 3: Prepare API environment
@@ -61,8 +60,9 @@ check:
 	@echo "✅ Code check complete"
 
 lint:
-	@echo "🔧 Running ruff format and check with fixes..."
-	@uv run --directory api --dev sh -c 'ruff format ./api && ruff check --fix ./api'
+	@echo "🔧 Running ruff format, check with fixes, and import linter..."
+	@uv run --project api --dev sh -c 'ruff format ./api && ruff check --fix ./api'
+	@uv run --directory api --dev lint-imports
 	@echo "✅ Linting complete"
 
 type-check:

README.md

@@ -40,18 +40,18 @@
 
 <p align="center">
   <a href="./README.md"><img alt="README in English" src="https://img.shields.io/badge/English-d9d9d9"></a>
-  <a href="./README_TW.md"><img alt="繁體中文文件" src="https://img.shields.io/badge/繁體中文-d9d9d9"></a>
-  <a href="./README_CN.md"><img alt="简体中文版自述文件" src="https://img.shields.io/badge/简体中文-d9d9d9"></a>
-  <a href="./README_JA.md"><img alt="日本語のREADME" src="https://img.shields.io/badge/日本語-d9d9d9"></a>
-  <a href="./README_ES.md"><img alt="README en Español" src="https://img.shields.io/badge/Español-d9d9d9"></a>
-  <a href="./README_FR.md"><img alt="README en Français" src="https://img.shields.io/badge/Français-d9d9d9"></a>
-  <a href="./README_KL.md"><img alt="README tlhIngan Hol" src="https://img.shields.io/badge/Klingon-d9d9d9"></a>
-  <a href="./README_KR.md"><img alt="README in Korean" src="https://img.shields.io/badge/한국어-d9d9d9"></a>
-  <a href="./README_AR.md"><img alt="README بالعربية" src="https://img.shields.io/badge/العربية-d9d9d9"></a>
-  <a href="./README_TR.md"><img alt="Türkçe README" src="https://img.shields.io/badge/Türkçe-d9d9d9"></a>
-  <a href="./README_VI.md"><img alt="README Tiếng Việt" src="https://img.shields.io/badge/Ti%E1%BA%BFng%20Vi%E1%BB%87t-d9d9d9"></a>
-  <a href="./README_DE.md"><img alt="README in Deutsch" src="https://img.shields.io/badge/German-d9d9d9"></a>
-  <a href="./README_BN.md"><img alt="README in বাংলা" src="https://img.shields.io/badge/বাংলা-d9d9d9"></a>
+  <a href="./docs/zh-TW/README.md"><img alt="繁體中文文件" src="https://img.shields.io/badge/繁體中文-d9d9d9"></a>
+  <a href="./docs/zh-CN/README.md"><img alt="简体中文文件" src="https://img.shields.io/badge/简体中文-d9d9d9"></a>
+  <a href="./docs/ja-JP/README.md"><img alt="日本語のREADME" src="https://img.shields.io/badge/日本語-d9d9d9"></a>
+  <a href="./docs/es-ES/README.md"><img alt="README en Español" src="https://img.shields.io/badge/Español-d9d9d9"></a>
+  <a href="./docs/fr-FR/README.md"><img alt="README en Français" src="https://img.shields.io/badge/Français-d9d9d9"></a>
+  <a href="./docs/tlh/README.md"><img alt="README tlhIngan Hol" src="https://img.shields.io/badge/Klingon-d9d9d9"></a>
+  <a href="./docs/ko-KR/README.md"><img alt="README in Korean" src="https://img.shields.io/badge/한국어-d9d9d9"></a>
+  <a href="./docs/ar-SA/README.md"><img alt="README بالعربية" src="https://img.shields.io/badge/العربية-d9d9d9"></a>
+  <a href="./docs/tr-TR/README.md"><img alt="Türkçe README" src="https://img.shields.io/badge/Türkçe-d9d9d9"></a>
+  <a href="./docs/vi-VN/README.md"><img alt="README Tiếng Việt" src="https://img.shields.io/badge/Ti%E1%BA%BFng%20Vi%E1%BB%87t-d9d9d9"></a>
+  <a href="./docs/de-DE/README.md"><img alt="README in Deutsch" src="https://img.shields.io/badge/German-d9d9d9"></a>
+  <a href="./docs/bn-BD/README.md"><img alt="README in বাংলা" src="https://img.shields.io/badge/বাংলা-d9d9d9"></a>
 </p>
 
 Dify is an open-source platform for developing LLM applications. Its intuitive interface combines agentic AI workflows, RAG pipelines, agent capabilities, model management, observability features, and more—allowing you to quickly move from prototype to production.

api/.env.example

@@ -76,6 +76,7 @@ DB_HOST=localhost
 DB_PORT=5432
 DB_DATABASE=dify
 SQLALCHEMY_POOL_PRE_PING=true
+SQLALCHEMY_POOL_TIMEOUT=30
 
 # Storage configuration
 # use for store upload files, private keys...
@@ -303,6 +304,8 @@ BAIDU_VECTOR_DB_API_KEY=dify
 BAIDU_VECTOR_DB_DATABASE=dify
 BAIDU_VECTOR_DB_SHARD=1
 BAIDU_VECTOR_DB_REPLICAS=3
+BAIDU_VECTOR_DB_INVERTED_INDEX_ANALYZER=DEFAULT_ANALYZER
+BAIDU_VECTOR_DB_INVERTED_INDEX_PARSER_MODE=COARSE_MODE
 
 # Upstash configuration
 UPSTASH_VECTOR_URL=your-server-url
@@ -405,6 +408,9 @@ SSRF_DEFAULT_TIME_OUT=5
 SSRF_DEFAULT_CONNECT_TIME_OUT=5
 SSRF_DEFAULT_READ_TIME_OUT=5
 SSRF_DEFAULT_WRITE_TIME_OUT=5
+SSRF_POOL_MAX_CONNECTIONS=100
+SSRF_POOL_MAX_KEEPALIVE_CONNECTIONS=20
+SSRF_POOL_KEEPALIVE_EXPIRY=5.0
 
 BATCH_UPLOAD_LIMIT=10
 KEYWORD_DATA_SOURCE_TYPE=database
@@ -415,10 +421,14 @@ WORKFLOW_FILE_UPLOAD_LIMIT=10
 # CODE EXECUTION CONFIGURATION
 CODE_EXECUTION_ENDPOINT=http://127.0.0.1:8194
 CODE_EXECUTION_API_KEY=dify-sandbox
+CODE_EXECUTION_SSL_VERIFY=True
+CODE_EXECUTION_POOL_MAX_CONNECTIONS=100
+CODE_EXECUTION_POOL_MAX_KEEPALIVE_CONNECTIONS=20
+CODE_EXECUTION_POOL_KEEPALIVE_EXPIRY=5.0
 CODE_MAX_NUMBER=9223372036854775807
 CODE_MIN_NUMBER=-9223372036854775808
-CODE_MAX_STRING_LENGTH=80000
-TEMPLATE_TRANSFORM_MAX_LENGTH=80000
+CODE_MAX_STRING_LENGTH=400000
+TEMPLATE_TRANSFORM_MAX_LENGTH=400000
 CODE_MAX_STRING_ARRAY_LENGTH=30
 CODE_MAX_OBJECT_ARRAY_LENGTH=30
 CODE_MAX_NUMBER_ARRAY_LENGTH=1000
@@ -458,7 +468,6 @@ INDEXING_MAX_SEGMENTATION_TOKENS_LENGTH=4000
 WORKFLOW_MAX_EXECUTION_STEPS=500
 WORKFLOW_MAX_EXECUTION_TIME=1200
 WORKFLOW_CALL_MAX_DEPTH=5
-WORKFLOW_PARALLEL_DEPTH_LIMIT=3
 MAX_VARIABLE_SIZE=204800
 
 # GraphEngine Worker Pool Configuration

api/.ruff.toml

@@ -30,6 +30,7 @@ select = [
     "RUF022",  # unsorted-dunder-all
     "S506",    # unsafe-yaml-load
     "SIM",     # flake8-simplify rules
+    "T201",    # print-found
     "TRY400",  # error-instead-of-exception
     "TRY401",  # verbose-log-message
     "UP",      # pyupgrade rules
@@ -91,11 +92,18 @@ ignore = [
 "configs/*" = [
     "N802", # invalid-function-name
 ]
+"core/model_runtime/callbacks/base_callback.py" = [
+    "T201",
+]
+"core/workflow/callbacks/workflow_logging_callback.py" = [
+    "T201",
+]
 "libs/gmpy2_pkcs10aep_cipher.py" = [
     "N803", # invalid-argument-name
 ]
 "tests/*" = [
     "F811", # redefined-while-unused
+    "T201", # allow print in tests
 ]
 
 [lint.pyflakes]

api/README.md

@@ -80,10 +80,10 @@
 1. 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,workflow_storage,conversation
+uv run celery -A app.celery worker -P gevent -c 2 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation
 ```
 
-Addition, if you want to debug the celery scheduled tasks, you can use the following command in another terminal:
+Additionally, if you want to debug the celery scheduled tasks, you can run the following command in another terminal to start the beat service:
 
 ```bash
 uv run celery -A app.celery beat

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

@@ -0,0 +1,13 @@
+import psycogreen.gevent as pscycogreen_gevent  # type: ignore
+from grpc.experimental import gevent as grpc_gevent  # type: ignore
+
+# grpc gevent
+grpc_gevent.init_gevent()
+print("gRPC patched with gevent.", flush=True)  # noqa: T201
+pscycogreen_gevent.patch_psycopg()
+print("psycopg2 patched with gevent.", flush=True)  # noqa: T201
+
+
+from app import app, celery
+
+__all__ = ["app", "celery"]

api/commands.py

@@ -1,7 +1,6 @@
 import base64
 import json
 import logging
-import operator
 import secrets
 from typing import Any
 
@@ -11,32 +10,41 @@ from flask import current_app
 from pydantic import TypeAdapter
 from sqlalchemy import select
 from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
 from constants.languages import languages
+from core.helper import encrypter
+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
 from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
+from extensions.storage.opendal_storage import OpenDALStorage
+from extensions.storage.storage_type import StorageType
 from libs.helper import email as email_validate
 from libs.password import hash_password, password_pattern, valid_password
 from libs.rsa import generate_key_pair
 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.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation, UploadFile
+from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
-from models.provider_ids import ToolProviderID
+from models.provider_ids import DatasourceProviderID, ToolProviderID
+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 services.plugin.plugin_service import PluginService
 from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 logger = logging.getLogger(__name__)
@@ -54,31 +62,30 @@ def reset_password(email, new_password, password_confirm):
     if str(new_password).strip() != str(password_confirm).strip():
         click.echo(click.style("Passwords do not match.", fg="red"))
         return
+    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+        account = session.query(Account).where(Account.email == email).one_or_none()
 
-    account = db.session.query(Account).where(Account.email == email).one_or_none()
+        if not account:
+            click.echo(click.style(f"Account not found for email: {email}", fg="red"))
+            return
 
-    if not account:
-        click.echo(click.style(f"Account not found for email: {email}", fg="red"))
-        return
+        try:
+            valid_password(new_password)
+        except:
+            click.echo(click.style(f"Invalid password. Must match {password_pattern}", fg="red"))
+            return
 
-    try:
-        valid_password(new_password)
-    except:
-        click.echo(click.style(f"Invalid password. Must match {password_pattern}", fg="red"))
-        return
+        # generate password salt
+        salt = secrets.token_bytes(16)
+        base64_salt = base64.b64encode(salt).decode()
 
-    # generate password salt
-    salt = secrets.token_bytes(16)
-    base64_salt = base64.b64encode(salt).decode()
-
-    # encrypt password with salt
-    password_hashed = hash_password(new_password, salt)
-    base64_password_hashed = base64.b64encode(password_hashed).decode()
-    account.password = base64_password_hashed
-    account.password_salt = base64_salt
-    db.session.commit()
-    AccountService.reset_login_error_rate_limit(email)
-    click.echo(click.style("Password reset successfully.", fg="green"))
+        # encrypt password with salt
+        password_hashed = hash_password(new_password, salt)
+        base64_password_hashed = base64.b64encode(password_hashed).decode()
+        account.password = base64_password_hashed
+        account.password_salt = base64_salt
+        AccountService.reset_login_error_rate_limit(email)
+        click.echo(click.style("Password reset successfully.", fg="green"))
 
 
 @click.command("reset-email", help="Reset the account email.")
@@ -93,22 +100,21 @@ def reset_email(email, new_email, email_confirm):
     if str(new_email).strip() != str(email_confirm).strip():
         click.echo(click.style("New emails do not match.", fg="red"))
         return
+    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+        account = session.query(Account).where(Account.email == email).one_or_none()
 
-    account = db.session.query(Account).where(Account.email == email).one_or_none()
+        if not account:
+            click.echo(click.style(f"Account not found for email: {email}", fg="red"))
+            return
 
-    if not account:
-        click.echo(click.style(f"Account not found for email: {email}", fg="red"))
-        return
+        try:
+            email_validate(new_email)
+        except:
+            click.echo(click.style(f"Invalid email: {new_email}", fg="red"))
+            return
 
-    try:
-        email_validate(new_email)
-    except:
-        click.echo(click.style(f"Invalid email: {new_email}", fg="red"))
-        return
-
-    account.email = new_email
-    db.session.commit()
-    click.echo(click.style("Email updated successfully.", fg="green"))
+        account.email = new_email
+        click.echo(click.style("Email updated successfully.", fg="green"))
 
 
 @click.command(
@@ -132,25 +138,24 @@ def reset_encrypt_key_pair():
     if dify_config.EDITION != "SELF_HOSTED":
         click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
         return
+    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+        tenants = session.query(Tenant).all()
+        for tenant in tenants:
+            if not tenant:
+                click.echo(click.style("No workspaces found. Run /install first.", fg="red"))
+                return
 
-    tenants = db.session.query(Tenant).all()
-    for tenant in tenants:
-        if not tenant:
-            click.echo(click.style("No workspaces found. Run /install first.", fg="red"))
-            return
+            tenant.encrypt_public_key = generate_key_pair(tenant.id)
 
-        tenant.encrypt_public_key = generate_key_pair(tenant.id)
+            session.query(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id).delete()
+            session.query(ProviderModel).where(ProviderModel.tenant_id == tenant.id).delete()
 
-        db.session.query(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id).delete()
-        db.session.query(ProviderModel).where(ProviderModel.tenant_id == tenant.id).delete()
-        db.session.commit()
-
-        click.echo(
-            click.style(
-                f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",
-                fg="green",
+            click.echo(
+                click.style(
+                    f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",
+                    fg="green",
+                )
             )
-        )
 
 
 @click.command("vdb-migrate", help="Migrate vector db.")
@@ -175,14 +180,15 @@ def migrate_annotation_vector_database():
         try:
             # get apps info
             per_page = 50
-            apps = (
-                db.session.query(App)
-                .where(App.status == "normal")
-                .order_by(App.created_at.desc())
-                .limit(per_page)
-                .offset((page - 1) * per_page)
-                .all()
-            )
+            with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+                apps = (
+                    session.query(App)
+                    .where(App.status == "normal")
+                    .order_by(App.created_at.desc())
+                    .limit(per_page)
+                    .offset((page - 1) * per_page)
+                    .all()
+                )
             if not apps:
                 break
         except SQLAlchemyError:
@@ -196,26 +202,27 @@ def migrate_annotation_vector_database():
             )
             try:
                 click.echo(f"Creating app annotation index: {app.id}")
-                app_annotation_setting = (
-                    db.session.query(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app.id).first()
-                )
+                with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+                    app_annotation_setting = (
+                        session.query(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app.id).first()
+                    )
 
-                if not app_annotation_setting:
-                    skipped_count = skipped_count + 1
-                    click.echo(f"App annotation setting disabled: {app.id}")
-                    continue
-                # get dataset_collection_binding info
-                dataset_collection_binding = (
-                    db.session.query(DatasetCollectionBinding)
-                    .where(DatasetCollectionBinding.id == app_annotation_setting.collection_binding_id)
-                    .first()
-                )
-                if not dataset_collection_binding:
-                    click.echo(f"App annotation collection binding not found: {app.id}")
-                    continue
-                annotations = db.session.scalars(
-                    select(MessageAnnotation).where(MessageAnnotation.app_id == app.id)
-                ).all()
+                    if not app_annotation_setting:
+                        skipped_count = skipped_count + 1
+                        click.echo(f"App annotation setting disabled: {app.id}")
+                        continue
+                    # get dataset_collection_binding info
+                    dataset_collection_binding = (
+                        session.query(DatasetCollectionBinding)
+                        .where(DatasetCollectionBinding.id == app_annotation_setting.collection_binding_id)
+                        .first()
+                    )
+                    if not dataset_collection_binding:
+                        click.echo(f"App annotation collection binding not found: {app.id}")
+                        continue
+                    annotations = session.scalars(
+                        select(MessageAnnotation).where(MessageAnnotation.app_id == app.id)
+                    ).all()
                 dataset = Dataset(
                     id=app.id,
                     tenant_id=app.tenant_id,
@@ -732,18 +739,18 @@ where sites.id is null limit 1000"""
                 try:
                     app = db.session.query(App).where(App.id == app_id).first()
                     if not app:
-                        print(f"App {app_id} not found")
+                        logger.info("App %s not found", app_id)
                         continue
 
                     tenant = app.tenant
                     if tenant:
                         accounts = tenant.get_accounts()
                         if not accounts:
-                            print(f"Fix failed for app {app.id}")
+                            logger.info("Fix failed for app %s", app.id)
                             continue
 
                         account = accounts[0]
-                        print(f"Fixing missing site for app {app.id}")
+                        logger.info("Fixing missing site for app %s", app.id)
                         app_was_created.send(app, account=account)
                 except Exception:
                     failed_app_ids.append(app_id)
@@ -1246,15 +1253,17 @@ def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
 
 def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
-    Count orphaned draft variables by app.
+    Count orphaned draft variables by app, including associated file counts.
 
     Returns:
-        Dictionary with statistics about orphaned variables
+        Dictionary with statistics about orphaned variables and files
     """
-    query = """
+    # Count orphaned variables by app
+    variables_query = """
         SELECT
             wdv.app_id,
-            COUNT(*) as variable_count
+            COUNT(*) as variable_count,
+            COUNT(wdv.file_id) as file_count
         FROM workflow_draft_variables AS wdv
         WHERE NOT EXISTS(
             SELECT 1 FROM apps WHERE apps.id = wdv.app_id
@@ -1264,14 +1273,21 @@ def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
 
     with db.engine.connect() as conn:
-        result = conn.execute(sa.text(query))
-        orphaned_by_app = {row[0]: row[1] for row in result}
+        result = conn.execute(sa.text(variables_query))
+        orphaned_by_app = {}
+        total_files = 0
 
-        total_orphaned = sum(orphaned_by_app.values())
+        for row in result:
+            app_id, variable_count, file_count = row
+            orphaned_by_app[app_id] = {"variables": variable_count, "files": file_count}
+            total_files += file_count
+
+        total_orphaned = sum(app_data["variables"] for app_data in orphaned_by_app.values())
         app_count = len(orphaned_by_app)
 
         return {
             "total_orphaned_variables": total_orphaned,
+            "total_orphaned_files": total_files,
             "orphaned_app_count": app_count,
             "orphaned_by_app": orphaned_by_app,
         }
@@ -1300,6 +1316,7 @@ def cleanup_orphaned_draft_variables(
     stats = _count_orphaned_draft_variables()
 
     logger.info("Found %s orphaned draft variables", stats["total_orphaned_variables"])
+    logger.info("Found %s associated offload files", stats["total_orphaned_files"])
     logger.info("Across %s non-existent apps", stats["orphaned_app_count"])
 
     if stats["total_orphaned_variables"] == 0:
@@ -1308,10 +1325,10 @@ def cleanup_orphaned_draft_variables(
 
     if dry_run:
         logger.info("DRY RUN: Would delete the following:")
-        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=operator.itemgetter(1), reverse=True)[
+        for app_id, data in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1]["variables"], reverse=True)[
             :10
         ]:  # Show top 10
-            logger.info("  App %s: %s variables", app_id, count)
+            logger.info("  App %s: %s variables, %s files", app_id, data["variables"], data["files"])
         if len(stats["orphaned_by_app"]) > 10:
             logger.info("  ... and %s more apps", len(stats["orphaned_by_app"]) - 10)
         return
@@ -1320,7 +1337,8 @@ def cleanup_orphaned_draft_variables(
     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?",
+            f"orphaned draft variables and {stats['total_orphaned_files']} associated files "
+            f"from {stats['orphaned_app_count']} apps?",
             abort=True,
         )
 
@@ -1353,3 +1371,456 @@ def cleanup_orphaned_draft_variables(
                 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)  # pyright: ignore[reportPrivateUsage]
+        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)  # pyright: ignore[reportPrivateUsage]
+        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)  # pyright: ignore[reportPrivateUsage]
+        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()
+        if notion_credentials:
+            notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
+            for notion_credential in notion_credentials:
+                tenant_id = notion_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(notion_credential)
+            for tenant_id, notion_tenant_credentials in notion_credentials_tenant_mapping.items():
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
+                if not tenant:
+                    continue
+                try:
+                    # 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
+                            PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier])
+                    auth_count = 0
+                    for notion_tenant_credential in notion_tenant_credentials:
+                        auth_count += 1
+                        # get credential oauth params
+                        access_token = notion_tenant_credential.access_token
+                        # notion info
+                        notion_info = notion_tenant_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_datasource",
+                            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
+                except Exception as e:
+                    click.echo(
+                        click.style(
+                            f"Error transforming notion credentials: {str(e)}, tenant_id: {tenant_id}", fg="red"
+                        )
+                    )
+                    continue
+                db.session.commit()
+        # deal firecrawl credentials
+        deal_firecrawl_count = 0
+        firecrawl_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="firecrawl").all()
+        if firecrawl_credentials:
+            firecrawl_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+            for firecrawl_credential in firecrawl_credentials:
+                tenant_id = firecrawl_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(firecrawl_credential)
+            for tenant_id, firecrawl_tenant_credentials in firecrawl_credentials_tenant_mapping.items():
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
+                if not tenant:
+                    continue
+                try:
+                    # 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
+                            PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier])
+
+                    auth_count = 0
+                    for firecrawl_tenant_credential in firecrawl_tenant_credentials:
+                        auth_count += 1
+                        # get credential api key
+                        credentials_json = json.loads(firecrawl_tenant_credential.credentials)
+                        api_key = credentials_json.get("config", {}).get("api_key")
+                        base_url = credentials_json.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
+                except Exception as e:
+                    click.echo(
+                        click.style(
+                            f"Error transforming firecrawl credentials: {str(e)}, tenant_id: {tenant_id}", fg="red"
+                        )
+                    )
+                    continue
+                db.session.commit()
+        # deal jina credentials
+        deal_jina_count = 0
+        jina_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="jinareader").all()
+        if jina_credentials:
+            jina_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
+            for jina_credential in jina_credentials:
+                tenant_id = jina_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(jina_credential)
+            for tenant_id, jina_tenant_credentials in jina_credentials_tenant_mapping.items():
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
+                if not tenant:
+                    continue
+                try:
+                    # 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
+                            logger.debug("Installing Jina plugin %s", jina_plugin_unique_identifier)
+                            PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier])
+
+                    auth_count = 0
+                    for jina_tenant_credential in jina_tenant_credentials:
+                        auth_count += 1
+                        # get credential api key
+                        credentials_json = json.loads(jina_tenant_credential.credentials)
+                        api_key = credentials_json.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
+                except Exception as e:
+                    click.echo(
+                        click.style(f"Error transforming jina credentials: {str(e)}, tenant_id: {tenant_id}", fg="red")
+                    )
+                    continue
+                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"))
+
+
+@click.command("install-rag-pipeline-plugins", help="Install rag pipeline plugins.")
+@click.option(
+    "--input_file", prompt=True, help="The file to store the extracted unique identifiers.", default="plugins.jsonl"
+)
+@click.option(
+    "--output_file", prompt=True, help="The file to store the installed plugins.", default="installed_plugins.jsonl"
+)
+@click.option("--workers", prompt=True, help="The number of workers to install plugins.", default=100)
+def install_rag_pipeline_plugins(input_file, output_file, workers):
+    """
+    Install rag pipeline plugins
+    """
+    click.echo(click.style("Installing rag pipeline plugins", fg="yellow"))
+    plugin_migration = PluginMigration()
+    plugin_migration.install_rag_pipeline_plugins(
+        input_file,
+        output_file,
+        workers,
+    )
+    click.echo(click.style("Installing rag pipeline plugins successfully", fg="green"))
+
+
+@click.command(
+    "migrate-oss",
+    help="Migrate files from Local or OpenDAL source to a cloud OSS storage (destination must NOT be local/opendal).",
+)
+@click.option(
+    "--path",
+    "paths",
+    multiple=True,
+    help="Storage path prefixes to migrate (repeatable). Defaults: privkeys, upload_files, image_files,"
+    " tools, website_files, keyword_files, ops_trace",
+)
+@click.option(
+    "--source",
+    type=click.Choice(["local", "opendal"], case_sensitive=False),
+    default="opendal",
+    show_default=True,
+    help="Source storage type to read from",
+)
+@click.option("--overwrite", is_flag=True, default=False, help="Overwrite destination if file already exists")
+@click.option("--dry-run", is_flag=True, default=False, help="Show what would be migrated without uploading")
+@click.option("-f", "--force", is_flag=True, help="Skip confirmation and run without prompts")
+@click.option(
+    "--update-db/--no-update-db",
+    default=True,
+    help="Update upload_files.storage_type from source type to current storage after migration",
+)
+def migrate_oss(
+    paths: tuple[str, ...],
+    source: str,
+    overwrite: bool,
+    dry_run: bool,
+    force: bool,
+    update_db: bool,
+):
+    """
+    Copy all files under selected prefixes from a source storage
+    (Local filesystem or OpenDAL-backed) into the currently configured
+    destination storage backend, then optionally update DB records.
+
+    Expected usage: set STORAGE_TYPE (and its credentials) to your target backend.
+    """
+    # Ensure target storage is not local/opendal
+    if dify_config.STORAGE_TYPE in (StorageType.LOCAL, StorageType.OPENDAL):
+        click.echo(
+            click.style(
+                "Target STORAGE_TYPE must be a cloud OSS (not 'local' or 'opendal').\n"
+                "Please set STORAGE_TYPE to one of: s3, aliyun-oss, azure-blob, google-storage, tencent-cos, \n"
+                "volcengine-tos, supabase, oci-storage, huawei-obs, baidu-obs, clickzetta-volume.",
+                fg="red",
+            )
+        )
+        return
+
+    # Default paths if none specified
+    default_paths = ("privkeys", "upload_files", "image_files", "tools", "website_files", "keyword_files", "ops_trace")
+    path_list = list(paths) if paths else list(default_paths)
+    is_source_local = source.lower() == "local"
+
+    click.echo(click.style("Preparing migration to target storage.", fg="yellow"))
+    click.echo(click.style(f"Target storage type: {dify_config.STORAGE_TYPE}", fg="white"))
+    if is_source_local:
+        src_root = dify_config.STORAGE_LOCAL_PATH
+        click.echo(click.style(f"Source: local fs, root: {src_root}", fg="white"))
+    else:
+        click.echo(click.style(f"Source: opendal scheme={dify_config.OPENDAL_SCHEME}", fg="white"))
+    click.echo(click.style(f"Paths to migrate: {', '.join(path_list)}", fg="white"))
+    click.echo("")
+
+    if not force:
+        click.confirm("Proceed with migration?", abort=True)
+
+    # Instantiate source storage
+    try:
+        if is_source_local:
+            src_root = dify_config.STORAGE_LOCAL_PATH
+            source_storage = OpenDALStorage(scheme="fs", root=src_root)
+        else:
+            source_storage = OpenDALStorage(scheme=dify_config.OPENDAL_SCHEME)
+    except Exception as e:
+        click.echo(click.style(f"Failed to initialize source storage: {str(e)}", fg="red"))
+        return
+
+    total_files = 0
+    copied_files = 0
+    skipped_files = 0
+    errored_files = 0
+    copied_upload_file_keys: list[str] = []
+
+    for prefix in path_list:
+        click.echo(click.style(f"Scanning source path: {prefix}", fg="white"))
+        try:
+            keys = source_storage.scan(path=prefix, files=True, directories=False)
+        except FileNotFoundError:
+            click.echo(click.style(f"  -> Skipping missing path: {prefix}", fg="yellow"))
+            continue
+        except NotImplementedError:
+            click.echo(click.style("  -> Source storage does not support scanning.", fg="red"))
+            return
+        except Exception as e:
+            click.echo(click.style(f"  -> Error scanning '{prefix}': {str(e)}", fg="red"))
+            continue
+
+        click.echo(click.style(f"Found {len(keys)} files under {prefix}", fg="white"))
+
+        for key in keys:
+            total_files += 1
+
+            # check destination existence
+            if not overwrite:
+                try:
+                    if storage.exists(key):
+                        skipped_files += 1
+                        continue
+                except Exception as e:
+                    # existence check failures should not block migration attempt
+                    # but should be surfaced to user as a warning for visibility
+                    click.echo(
+                        click.style(
+                            f"  -> Warning: failed target existence check for {key}: {str(e)}",
+                            fg="yellow",
+                        )
+                    )
+
+            if dry_run:
+                copied_files += 1
+                continue
+
+            # read from source and write to destination
+            try:
+                data = source_storage.load_once(key)
+            except FileNotFoundError:
+                errored_files += 1
+                click.echo(click.style(f"  -> Missing on source: {key}", fg="yellow"))
+                continue
+            except Exception as e:
+                errored_files += 1
+                click.echo(click.style(f"  -> Error reading {key}: {str(e)}", fg="red"))
+                continue
+
+            try:
+                storage.save(key, data)
+                copied_files += 1
+                if prefix == "upload_files":
+                    copied_upload_file_keys.append(key)
+            except Exception as e:
+                errored_files += 1
+                click.echo(click.style(f"  -> Error writing {key} to target: {str(e)}", fg="red"))
+                continue
+
+    click.echo("")
+    click.echo(click.style("Migration summary:", fg="yellow"))
+    click.echo(click.style(f"  Total:   {total_files}", fg="white"))
+    click.echo(click.style(f"  Copied:  {copied_files}", fg="green"))
+    click.echo(click.style(f"  Skipped: {skipped_files}", fg="white"))
+    if errored_files:
+        click.echo(click.style(f"  Errors:  {errored_files}", fg="red"))
+
+    if dry_run:
+        click.echo(click.style("Dry-run complete. No changes were made.", fg="green"))
+        return
+
+    if errored_files:
+        click.echo(
+            click.style(
+                "Some files failed to migrate. Review errors above before updating DB records.",
+                fg="yellow",
+            )
+        )
+        if update_db and not force:
+            if not click.confirm("Proceed to update DB storage_type despite errors?", default=False):
+                update_db = False
+
+    # Optionally update DB records for upload_files.storage_type (only for successfully copied upload_files)
+    if update_db:
+        if not copied_upload_file_keys:
+            click.echo(click.style("No upload_files copied. Skipping DB storage_type update.", fg="yellow"))
+        else:
+            try:
+                source_storage_type = StorageType.LOCAL if is_source_local else StorageType.OPENDAL
+                updated = (
+                    db.session.query(UploadFile)
+                    .where(
+                        UploadFile.storage_type == source_storage_type,
+                        UploadFile.key.in_(copied_upload_file_keys),
+                    )
+                    .update({UploadFile.storage_type: dify_config.STORAGE_TYPE}, synchronize_session=False)
+                )
+                db.session.commit()
+                click.echo(click.style(f"Updated storage_type for {updated} upload_files records.", fg="green"))
+            except Exception as e:
+                db.session.rollback()
+                click.echo(click.style(f"Failed to update DB storage_type: {str(e)}", fg="red"))

api/configs/__init__.py

@@ -1,3 +1,3 @@
 from .app_config import DifyConfig
 
-dify_config = DifyConfig()
+dify_config = DifyConfig()  # type: ignore

api/configs/feature/__init__.py

@@ -1,3 +1,4 @@
+from enum import StrEnum
 from typing import Literal
 
 from pydantic import (
@@ -112,6 +113,21 @@ class CodeExecutionSandboxConfig(BaseSettings):
         default=10.0,
     )
 
+    CODE_EXECUTION_POOL_MAX_CONNECTIONS: PositiveInt = Field(
+        description="Maximum number of concurrent connections for the code execution HTTP client",
+        default=100,
+    )
+
+    CODE_EXECUTION_POOL_MAX_KEEPALIVE_CONNECTIONS: PositiveInt = Field(
+        description="Maximum number of persistent keep-alive connections for the code execution HTTP client",
+        default=20,
+    )
+
+    CODE_EXECUTION_POOL_KEEPALIVE_EXPIRY: PositiveFloat | None = Field(
+        description="Keep-alive expiry in seconds for idle connections (set to None to disable)",
+        default=5.0,
+    )
+
     CODE_MAX_NUMBER: PositiveInt = Field(
         description="Maximum allowed numeric value in code execution",
         default=9223372036854775807,
@@ -134,7 +150,7 @@ class CodeExecutionSandboxConfig(BaseSettings):
 
     CODE_MAX_STRING_LENGTH: PositiveInt = Field(
         description="Maximum allowed length for strings in code execution",
-        default=80000,
+        default=400_000,
     )
 
     CODE_MAX_STRING_ARRAY_LENGTH: PositiveInt = Field(
@@ -152,6 +168,11 @@ class CodeExecutionSandboxConfig(BaseSettings):
         default=1000,
     )
 
+    CODE_EXECUTION_SSL_VERIFY: bool = Field(
+        description="Enable or disable SSL verification for code execution requests",
+        default=True,
+    )
+
 
 class PluginConfig(BaseSettings):
     """
@@ -403,6 +424,21 @@ class HttpConfig(BaseSettings):
         default=5,
     )
 
+    SSRF_POOL_MAX_CONNECTIONS: PositiveInt = Field(
+        description="Maximum number of concurrent connections for the SSRF HTTP client",
+        default=100,
+    )
+
+    SSRF_POOL_MAX_KEEPALIVE_CONNECTIONS: PositiveInt = Field(
+        description="Maximum number of persistent keep-alive connections for the SSRF HTTP client",
+        default=20,
+    )
+
+    SSRF_POOL_KEEPALIVE_EXPIRY: PositiveFloat | None = Field(
+        description="Keep-alive expiry in seconds for idle SSRF connections (set to None to disable)",
+        default=5.0,
+    )
+
     RESPECT_XFORWARD_HEADERS_ENABLED: bool = Field(
         description="Enable handling of X-Forwarded-For, X-Forwarded-Proto, and X-Forwarded-Port headers"
         " when the app is behind a single trusted reverse proxy.",
@@ -505,6 +541,22 @@ class UpdateConfig(BaseSettings):
     )
 
 
+class WorkflowVariableTruncationConfig(BaseSettings):
+    WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
+        # 100KB
+        1024_000,
+        description="Maximum size for variable to trigger final truncation.",
+    )
+    WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH: PositiveInt = Field(
+        100000,
+        description="maximum length for string to trigger tuncation, measure in number of characters",
+    )
+    WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH: PositiveInt = Field(
+        1000,
+        description="maximum length for array to trigger truncation.",
+    )
+
+
 class WorkflowConfig(BaseSettings):
     """
     Configuration for workflow execution
@@ -525,16 +577,16 @@ class WorkflowConfig(BaseSettings):
         default=5,
     )
 
-    WORKFLOW_PARALLEL_DEPTH_LIMIT: PositiveInt = Field(
-        description="Maximum allowed depth for nested parallel executions",
-        default=3,
-    )
-
     MAX_VARIABLE_SIZE: PositiveInt = Field(
         description="Maximum size in bytes for a single variable in workflows. Default to 200 KB.",
         default=200 * 1024,
     )
 
+    TEMPLATE_TRANSFORM_MAX_LENGTH: PositiveInt = Field(
+        description="Maximum number of characters allowed in Template Transform node output",
+        default=400_000,
+    )
+
     # GraphEngine Worker Pool Configuration
     GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
         description="Minimum number of workers per GraphEngine instance",
@@ -695,11 +747,35 @@ class ToolConfig(BaseSettings):
     )
 
 
+class TemplateMode(StrEnum):
+    # unsafe mode allows flexible operations in templates, but may cause security vulnerabilities
+    UNSAFE = "unsafe"
+
+    # sandbox mode restricts some unsafe operations like accessing __class__.
+    # however, it is still not 100% safe, for example, cpu exploitation can happen.
+    SANDBOX = "sandbox"
+
+    # templating is disabled
+    DISABLED = "disabled"
+
+
 class MailConfig(BaseSettings):
     """
     Configuration for email services
     """
 
+    MAIL_TEMPLATING_MODE: TemplateMode = Field(
+        description="Template mode for email services",
+        default=TemplateMode.SANDBOX,
+    )
+
+    MAIL_TEMPLATING_TIMEOUT: int = Field(
+        description="""
+        Timeout for email templating in seconds. Used to prevent infinite loops in malicious templates. 
+        Only available in sandbox mode.""",
+        default=3,
+    )
+
     MAIL_TYPE: str | None = Field(
         description="Email service provider type ('smtp' or 'resend' or 'sendGrid), default to None.",
         default=None,
@@ -1063,5 +1139,6 @@ class FeatureConfig(
     CeleryBeatConfig,
     CeleryScheduleTasksConfig,
     WorkflowLogConfig,
+    WorkflowVariableTruncationConfig,
 ):
     pass

api/configs/feature/hosted_service/__init__.py

@@ -220,11 +220,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="remote",
+    )
+
+    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

@@ -187,6 +187,11 @@ class DatabaseConfig(BaseSettings):
         default=False,
     )
 
+    SQLALCHEMY_POOL_TIMEOUT: NonNegativeInt = Field(
+        description="Number of seconds to wait for a connection from the pool before raising a timeout error.",
+        default=30,
+    )
+
     RETRIEVAL_SERVICE_EXECUTORS: NonNegativeInt = Field(
         description="Number of processes for the retrieval service, default to CPU cores.",
         default=os.cpu_count() or 1,
@@ -216,6 +221,7 @@ class DatabaseConfig(BaseSettings):
             "connect_args": connect_args,
             "pool_use_lifo": self.SQLALCHEMY_POOL_USE_LIFO,
             "pool_reset_on_return": None,
+            "pool_timeout": self.SQLALCHEMY_POOL_TIMEOUT,
         }
 
 

api/configs/middleware/vdb/baidu_vector_config.py

@@ -41,3 +41,13 @@ class BaiduVectorDBConfig(BaseSettings):
         description="Number of replicas for the Baidu Vector Database (default is 3)",
         default=3,
     )
+
+    BAIDU_VECTOR_DB_INVERTED_INDEX_ANALYZER: str = Field(
+        description="Analyzer type for inverted index in Baidu Vector Database (default is DEFAULT_ANALYZER)",
+        default="DEFAULT_ANALYZER",
+    )
+
+    BAIDU_VECTOR_DB_INVERTED_INDEX_PARSER_MODE: str = Field(
+        description="Parser mode for inverted index in Baidu Vector Database (default is COARSE_MODE)",
+        default="COARSE_MODE",
+    )

api/configs/middleware/vdb/oceanbase_config.py

@@ -37,3 +37,15 @@ class OceanBaseVectorConfig(BaseSettings):
         "with older versions",
         default=False,
     )
+
+    OCEANBASE_FULLTEXT_PARSER: str | None = Field(
+        description=(
+            "Fulltext parser to use for text indexing. "
+            "Built-in options: 'ngram' (N-gram tokenizer for English/numbers), "
+            "'beng' (Basic English tokenizer), 'space' (Space-based tokenizer), "
+            "'ngram2' (Improved N-gram tokenizer), 'ik' (Chinese tokenizer). "
+            "External plugins (require installation): 'japanese_ftparser' (Japanese tokenizer), "
+            "'thai_ftparser' (Thai tokenizer). Default is 'ik'"
+        ),
+        default="ik",
+    )

api/configs/remote_settings_sources/nacos/http_request.py

@@ -5,7 +5,7 @@ import logging
 import os
 import time
 
-import requests
+import httpx
 
 logger = logging.getLogger(__name__)
 
@@ -30,10 +30,10 @@ class NacosHttpClient:
             params = {}
         try:
             self._inject_auth_info(headers, params)
-            response = requests.request(method, url="http://" + self.server + url, headers=headers, params=params)
+            response = httpx.request(method, url="http://" + self.server + url, headers=headers, params=params)
             response.raise_for_status()
             return response.text
-        except requests.RequestException as e:
+        except httpx.RequestError as e:
             return f"Request to Nacos failed: {e}"
 
     def _inject_auth_info(self, headers: dict[str, str], params: dict[str, str], module: str = "config") -> None:
@@ -78,7 +78,7 @@ class NacosHttpClient:
         params = {"username": self.username, "password": self.password}
         url = "http://" + self.server + "/nacos/v1/auth/login"
         try:
-            resp = requests.request("POST", url, headers=None, params=params)
+            resp = httpx.request("POST", url, headers=None, params=params)
             resp.raise_for_status()
             response_data = resp.json()
             self.token = response_data.get("accessToken")

api/constants/__init__.py

@@ -1,4 +1,5 @@
 from configs import dify_config
+from libs.collection_utils import convert_to_lower_and_upper_set
 
 HIDDEN_VALUE = "[__HIDDEN__]"
 UNKNOWN_VALUE = "[__UNKNOWN__]"
@@ -6,24 +7,39 @@ UUID_NIL = "00000000-0000-0000-0000-000000000000"
 
 DEFAULT_FILE_NUMBER_LIMITS = 3
 
-IMAGE_EXTENSIONS = ["jpg", "jpeg", "png", "webp", "gif", "svg"]
-IMAGE_EXTENSIONS.extend([ext.upper() for ext in IMAGE_EXTENSIONS])
+IMAGE_EXTENSIONS = convert_to_lower_and_upper_set({"jpg", "jpeg", "png", "webp", "gif", "svg"})
 
-VIDEO_EXTENSIONS = ["mp4", "mov", "mpeg", "webm"]
-VIDEO_EXTENSIONS.extend([ext.upper() for ext in VIDEO_EXTENSIONS])
+VIDEO_EXTENSIONS = convert_to_lower_and_upper_set({"mp4", "mov", "mpeg", "webm"})
 
-AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "amr", "mpga"]
-AUDIO_EXTENSIONS.extend([ext.upper() for ext in AUDIO_EXTENSIONS])
+AUDIO_EXTENSIONS = convert_to_lower_and_upper_set({"mp3", "m4a", "wav", "amr", "mpga"})
 
-
-_doc_extensions: list[str]
+_doc_extensions: set[str]
 if dify_config.ETL_TYPE == "Unstructured":
-    _doc_extensions = ["txt", "markdown", "md", "mdx", "pdf", "html", "htm", "xlsx", "xls", "vtt", "properties"]
-    _doc_extensions.extend(("doc", "docx", "csv", "eml", "msg", "pptx", "xml", "epub"))
+    _doc_extensions = {
+        "txt",
+        "markdown",
+        "md",
+        "mdx",
+        "pdf",
+        "html",
+        "htm",
+        "xlsx",
+        "xls",
+        "vtt",
+        "properties",
+        "doc",
+        "docx",
+        "csv",
+        "eml",
+        "msg",
+        "pptx",
+        "xml",
+        "epub",
+    }
     if dify_config.UNSTRUCTURED_API_URL:
-        _doc_extensions.append("ppt")
+        _doc_extensions.add("ppt")
 else:
-    _doc_extensions = [
+    _doc_extensions = {
         "txt",
         "markdown",
         "md",
@@ -37,5 +53,5 @@ else:
         "csv",
         "vtt",
         "properties",
-    ]
-DOCUMENT_EXTENSIONS = _doc_extensions + [ext.upper() for ext in _doc_extensions]
+    }
+DOCUMENT_EXTENSIONS: set[str] = convert_to_lower_and_upper_set(_doc_extensions)

api/contexts/__init__.py

@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING
 from contexts.wrapper import RecyclableContextVar
 
 if TYPE_CHECKING:
+    from core.datasource.__base.datasource_provider import DatasourcePluginProviderController
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
@@ -32,3 +33,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/console/__init__.py

@@ -1,31 +1,10 @@
+from importlib import import_module
+
 from flask import Blueprint
 from flask_restx import Namespace
 
 from libs.external_api import ExternalApi
 
-from .app.app_import import AppImportApi, AppImportCheckDependenciesApi, AppImportConfirmApi
-from .explore.audio import ChatAudioApi, ChatTextApi
-from .explore.completion import ChatApi, ChatStopApi, CompletionApi, CompletionStopApi
-from .explore.conversation import (
-    ConversationApi,
-    ConversationListApi,
-    ConversationPinApi,
-    ConversationRenameApi,
-    ConversationUnPinApi,
-)
-from .explore.message import (
-    MessageFeedbackApi,
-    MessageListApi,
-    MessageMoreLikeThisApi,
-    MessageSuggestedQuestionApi,
-)
-from .explore.workflow import (
-    InstalledAppWorkflowRunApi,
-    InstalledAppWorkflowTaskStopApi,
-)
-from .files import FileApi, FilePreviewApi, FileSupportTypeApi
-from .remote_files import RemoteFileInfoApi, RemoteFileUploadApi
-
 bp = Blueprint("console", __name__, url_prefix="/console/api")
 
 api = ExternalApi(
@@ -35,23 +14,23 @@ api = ExternalApi(
     description="Console management APIs for app configuration, monitoring, and administration",
 )
 
-# Create namespace
 console_ns = Namespace("console", description="Console management API operations", path="/")
 
-# File
-api.add_resource(FileApi, "/files/upload")
-api.add_resource(FilePreviewApi, "/files/<uuid:file_id>/preview")
-api.add_resource(FileSupportTypeApi, "/files/support-type")
+RESOURCE_MODULES = (
+    "controllers.console.app.app_import",
+    "controllers.console.explore.audio",
+    "controllers.console.explore.completion",
+    "controllers.console.explore.conversation",
+    "controllers.console.explore.message",
+    "controllers.console.explore.workflow",
+    "controllers.console.files",
+    "controllers.console.remote_files",
+)
 
-# Remote files
-api.add_resource(RemoteFileInfoApi, "/remote-files/<path:url>")
-api.add_resource(RemoteFileUploadApi, "/remote-files/upload")
-
-# Import App
-api.add_resource(AppImportApi, "/apps/imports")
-api.add_resource(AppImportConfirmApi, "/apps/imports/<string:import_id>/confirm")
-api.add_resource(AppImportCheckDependenciesApi, "/apps/imports/<string:app_id>/check-dependencies")
+for module_name in RESOURCE_MODULES:
+    import_module(module_name)
 
+# Ensure resource modules are imported so route decorators are evaluated.
 # Import other controllers
 from . import (
     admin,
@@ -61,6 +40,7 @@ from . import (
     init_validate,
     ping,
     setup,
+    spec,
     version,
 )
 
@@ -114,6 +94,15 @@ from .datasets import (
     metadata,
     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 (
@@ -140,77 +129,6 @@ from .workspace import (
     workspace,
 )
 
-# Explore Audio
-api.add_resource(ChatAudioApi, "/installed-apps/<uuid:installed_app_id>/audio-to-text", endpoint="installed_app_audio")
-api.add_resource(ChatTextApi, "/installed-apps/<uuid:installed_app_id>/text-to-audio", endpoint="installed_app_text")
-
-# Explore Completion
-api.add_resource(
-    CompletionApi, "/installed-apps/<uuid:installed_app_id>/completion-messages", endpoint="installed_app_completion"
-)
-api.add_resource(
-    CompletionStopApi,
-    "/installed-apps/<uuid:installed_app_id>/completion-messages/<string:task_id>/stop",
-    endpoint="installed_app_stop_completion",
-)
-api.add_resource(
-    ChatApi, "/installed-apps/<uuid:installed_app_id>/chat-messages", endpoint="installed_app_chat_completion"
-)
-api.add_resource(
-    ChatStopApi,
-    "/installed-apps/<uuid:installed_app_id>/chat-messages/<string:task_id>/stop",
-    endpoint="installed_app_stop_chat_completion",
-)
-
-# Explore Conversation
-api.add_resource(
-    ConversationRenameApi,
-    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/name",
-    endpoint="installed_app_conversation_rename",
-)
-api.add_resource(
-    ConversationListApi, "/installed-apps/<uuid:installed_app_id>/conversations", endpoint="installed_app_conversations"
-)
-api.add_resource(
-    ConversationApi,
-    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>",
-    endpoint="installed_app_conversation",
-)
-api.add_resource(
-    ConversationPinApi,
-    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/pin",
-    endpoint="installed_app_conversation_pin",
-)
-api.add_resource(
-    ConversationUnPinApi,
-    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/unpin",
-    endpoint="installed_app_conversation_unpin",
-)
-
-
-# Explore Message
-api.add_resource(MessageListApi, "/installed-apps/<uuid:installed_app_id>/messages", endpoint="installed_app_messages")
-api.add_resource(
-    MessageFeedbackApi,
-    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/feedbacks",
-    endpoint="installed_app_message_feedback",
-)
-api.add_resource(
-    MessageMoreLikeThisApi,
-    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/more-like-this",
-    endpoint="installed_app_more_like_this",
-)
-api.add_resource(
-    MessageSuggestedQuestionApi,
-    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/suggested-questions",
-    endpoint="installed_app_suggested_question",
-)
-# Explore Workflow
-api.add_resource(InstalledAppWorkflowRunApi, "/installed-apps/<uuid:installed_app_id>/workflows/run")
-api.add_resource(
-    InstalledAppWorkflowTaskStopApi, "/installed-apps/<uuid:installed_app_id>/workflows/tasks/<string:task_id>/stop"
-)
-
 api.add_namespace(console_ns)
 
 __all__ = [
@@ -238,6 +156,8 @@ __all__ = [
     "datasets",
     "datasets_document",
     "datasets_segments",
+    "datasource_auth",
+    "datasource_content_preview",
     "email_register",
     "endpoint",
     "extension",
@@ -263,10 +183,16 @@ __all__ = [
     "parameter",
     "ping",
     "plugin",
+    "rag_pipeline",
+    "rag_pipeline_datasets",
+    "rag_pipeline_draft_variable",
+    "rag_pipeline_import",
+    "rag_pipeline_workflow",
     "recommended_app",
     "saved_message",
     "setup",
     "site",
+    "spec",
     "statistic",
     "tags",
     "tool_providers",

api/controllers/console/app/app.py

@@ -19,6 +19,7 @@ from core.ops.ops_trace_manager import OpsTraceManager
 from extensions.ext_database import db
 from fields.app_fields import app_detail_fields, app_detail_fields_with_site, app_pagination_fields
 from libs.login import login_required
+from libs.validators import validate_description_length
 from models import Account, App
 from services.app_dsl_service import AppDslService, ImportMode
 from services.app_service import AppService
@@ -28,12 +29,6 @@ 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
-
-
 @console_ns.route("/apps")
 class AppListApi(Resource):
     @api.doc("list_apps")
@@ -138,7 +133,7 @@ class AppListApi(Resource):
         """Create app"""
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
-        parser.add_argument("description", type=_validate_description_length, 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")
@@ -219,7 +214,7 @@ class AppApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=_validate_description_length, 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")
@@ -297,7 +292,7 @@ class AppCopyApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, location="json")
-        parser.add_argument("description", type=_validate_description_length, 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/app_import.py

@@ -20,7 +20,10 @@ from services.app_dsl_service import AppDslService, ImportStatus
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
+from .. import console_ns
 
+
+@console_ns.route("/apps/imports")
 class AppImportApi(Resource):
     @setup_required
     @login_required
@@ -74,6 +77,7 @@ class AppImportApi(Resource):
         return result.model_dump(mode="json"), 200
 
 
+@console_ns.route("/apps/imports/<string:import_id>/confirm")
 class AppImportConfirmApi(Resource):
     @setup_required
     @login_required
@@ -98,6 +102,7 @@ class AppImportConfirmApi(Resource):
         return result.model_dump(mode="json"), 200
 
 
+@console_ns.route("/apps/imports/<string:app_id>/check-dependencies")
 class AppImportCheckDependenciesApi(Resource):
     @setup_required
     @login_required

api/controllers/console/app/conversation.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 
 import pytz  # pip install pytz
+import sqlalchemy as sa
 from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from flask_restx.inputs import int_range
@@ -70,7 +71,7 @@ class CompletionConversationApi(Resource):
         parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
         args = parser.parse_args()
 
-        query = db.select(Conversation).where(
+        query = sa.select(Conversation).where(
             Conversation.app_id == app_model.id, Conversation.mode == "completion", Conversation.is_deleted.is_(False)
         )
 
@@ -236,7 +237,7 @@ class ChatConversationApi(Resource):
             .subquery()
         )
 
-        query = db.select(Conversation).where(Conversation.app_id == app_model.id, Conversation.is_deleted.is_(False))
+        query = sa.select(Conversation).where(Conversation.app_id == app_model.id, Conversation.is_deleted.is_(False))
 
         if args["keyword"]:
             keyword_filter = f"%{args['keyword']}%"

api/controllers/console/app/generator.py

@@ -261,6 +261,7 @@ class InstructionGenerateApi(Resource):
                     instruction=args["instruction"],
                     model_config=args["model_config"],
                     ideal_output=args["ideal_output"],
+                    workflow_service=WorkflowService(),
                 )
             return {"error": "incompatible parameters"}, 400
         except ProviderTokenNotInitError as ex:

api/controllers/console/app/message.py

@@ -62,6 +62,9 @@ class ChatMessageListApi(Resource):
     @account_initialization_required
     @marshal_with(message_infinite_scroll_pagination_fields)
     def get(self, app_model):
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
         parser = reqparse.RequestParser()
         parser.add_argument("conversation_id", required=True, type=uuid_value, location="args")
         parser.add_argument("first_id", type=uuid_value, location="args")

api/controllers/console/app/statistic.py

@@ -50,8 +50,9 @@ class DailyMessageStatistic(Resource):
 FROM
     messages
 WHERE
-    app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -187,8 +188,9 @@ class DailyTerminalsStatistic(Resource):
 FROM
     messages
 WHERE
-    app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -259,8 +261,9 @@ class DailyTokenCostStatistic(Resource):
 FROM
     messages
 WHERE
-    app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -340,8 +343,9 @@ FROM
             messages m
             ON c.id = m.conversation_id
         WHERE
-            c.app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+            c.app_id = :app_id
+            AND m.invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -426,8 +430,9 @@ LEFT JOIN
     message_feedbacks mf
     ON mf.message_id=m.id AND mf.rating='like'
 WHERE
-    m.app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    m.app_id = :app_id
+    AND m.invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -502,8 +507,9 @@ class AverageResponseTimeStatistic(Resource):
 FROM
     messages
 WHERE
-    app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -576,8 +582,9 @@ class TokensPerSecondStatistic(Resource):
 FROM
     messages
 WHERE
-    app_id = :app_id"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id}
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc

api/controllers/console/app/workflow.py

@@ -9,7 +9,6 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
-from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -797,24 +796,6 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
-class WorkflowConfigApi(Resource):
-    """Resource for workflow configuration."""
-
-    @api.doc("get_workflow_config")
-    @api.doc(description="Get workflow configuration")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Workflow configuration retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        return {
-            "parallel_depth_limit": dify_config.WORKFLOW_PARALLEL_DEPTH_LIMIT,
-        }
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @api.doc("get_all_published_workflows")

api/controllers/console/app/workflow_draft_variable.py

@@ -13,6 +13,7 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.file import helpers as file_helpers
 from core.variables.segment_group import SegmentGroup
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
@@ -75,6 +76,22 @@ def _serialize_variable_type(workflow_draft_var: WorkflowDraftVariable) -> str:
     return value_type.exposed_type().value
 
 
+def _serialize_full_content(variable: WorkflowDraftVariable) -> dict | None:
+    """Serialize full_content information for large variables."""
+    if not variable.is_truncated():
+        return None
+
+    variable_file = variable.variable_file
+    assert variable_file is not None
+
+    return {
+        "size_bytes": variable_file.size,
+        "value_type": variable_file.value_type.exposed_type().value,
+        "length": variable_file.length,
+        "download_url": file_helpers.get_signed_file_url(variable_file.upload_file_id, as_attachment=True),
+    }
+
+
 _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "id": fields.String,
     "type": fields.String(attribute=lambda model: model.get_variable_type()),
@@ -84,11 +101,13 @@ _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "value_type": fields.String(attribute=_serialize_variable_type),
     "edited": fields.Boolean(attribute=lambda model: model.edited),
     "visible": fields.Boolean,
+    "is_truncated": fields.Boolean(attribute=lambda model: model.file_id is not None),
 }
 
 _WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
     _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
     value=fields.Raw(attribute=_serialize_var_value),
+    full_content=fields.Raw(attribute=_serialize_full_content),
 )
 
 _WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {

api/controllers/console/auth/data_source_bearer_auth.py

@@ -2,7 +2,7 @@ from flask_login import current_user
 from flask_restx import Resource, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.auth.error import ApiKeyAuthFailedError
 from libs.login import login_required
 from services.auth.api_key_auth_service import ApiKeyAuthService
@@ -10,6 +10,7 @@ from services.auth.api_key_auth_service import ApiKeyAuthService
 from ..wraps import account_initialization_required, setup_required
 
 
+@console_ns.route("/api-key-auth/data-source")
 class ApiKeyAuthDataSource(Resource):
     @setup_required
     @login_required
@@ -33,6 +34,7 @@ class ApiKeyAuthDataSource(Resource):
         return {"sources": []}
 
 
+@console_ns.route("/api-key-auth/data-source/binding")
 class ApiKeyAuthDataSourceBinding(Resource):
     @setup_required
     @login_required
@@ -54,6 +56,7 @@ class ApiKeyAuthDataSourceBinding(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/api-key-auth/data-source/<uuid:binding_id>")
 class ApiKeyAuthDataSourceBindingDelete(Resource):
     @setup_required
     @login_required
@@ -66,8 +69,3 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
         ApiKeyAuthService.delete_provider_auth(current_user.current_tenant_id, binding_id)
 
         return {"result": "success"}, 204
-
-
-api.add_resource(ApiKeyAuthDataSource, "/api-key-auth/data-source")
-api.add_resource(ApiKeyAuthDataSourceBinding, "/api-key-auth/data-source/binding")
-api.add_resource(ApiKeyAuthDataSourceBindingDelete, "/api-key-auth/data-source/<uuid:binding_id>")

api/controllers/console/auth/data_source_oauth.py

@@ -1,6 +1,6 @@
 import logging
 
-import requests
+import httpx
 from flask import current_app, redirect, request
 from flask_login import current_user
 from flask_restx import Resource, fields
@@ -119,7 +119,7 @@ class OAuthDataSourceBinding(Resource):
                 return {"error": "Invalid code"}, 400
             try:
                 oauth_provider.get_access_token(code)
-            except requests.HTTPError as e:
+            except httpx.HTTPStatusError as e:
                 logger.exception(
                     "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
                 )
@@ -152,7 +152,7 @@ class OAuthDataSourceSync(Resource):
             return {"error": "Invalid provider"}, 400
         try:
             oauth_provider.sync_data_source(binding_id)
-        except requests.HTTPError as e:
+        except httpx.HTTPStatusError as e:
             logger.exception(
                 "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
             )

api/controllers/console/auth/email_register.py

@@ -5,7 +5,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import languages
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.auth.error import (
     EmailAlreadyInUseError,
     EmailCodeError,
@@ -25,6 +25,7 @@ from services.billing_service import BillingService
 from services.errors.account import AccountNotFoundError, AccountRegisterError
 
 
+@console_ns.route("/email-register/send-email")
 class EmailRegisterSendEmailApi(Resource):
     @setup_required
     @email_password_login_enabled
@@ -52,6 +53,7 @@ class EmailRegisterSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/email-register/validity")
 class EmailRegisterCheckApi(Resource):
     @setup_required
     @email_password_login_enabled
@@ -92,6 +94,7 @@ class EmailRegisterCheckApi(Resource):
         return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
+@console_ns.route("/email-register")
 class EmailRegisterResetApi(Resource):
     @setup_required
     @email_password_login_enabled
@@ -148,8 +151,3 @@ class EmailRegisterResetApi(Resource):
             raise AccountInFreezeError()
 
         return account
-
-
-api.add_resource(EmailRegisterSendEmailApi, "/email-register/send-email")
-api.add_resource(EmailRegisterCheckApi, "/email-register/validity")
-api.add_resource(EmailRegisterResetApi, "/email-register")

api/controllers/console/auth/forgot_password.py

@@ -221,8 +221,3 @@ class ForgotPasswordResetApi(Resource):
             TenantService.create_tenant_member(tenant, account, role="owner")
             account.current_tenant = tenant
             tenant_was_created.send(tenant)
-
-
-api.add_resource(ForgotPasswordSendEmailApi, "/forgot-password")
-api.add_resource(ForgotPasswordCheckApi, "/forgot-password/validity")
-api.add_resource(ForgotPasswordResetApi, "/forgot-password/resets")

api/controllers/console/auth/login.py

@@ -7,7 +7,7 @@ from flask_restx import Resource, reqparse
 import services
 from configs import dify_config
 from constants.languages import languages
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.auth.error import (
     AuthenticationFailedError,
     EmailCodeError,
@@ -34,6 +34,7 @@ from services.errors.workspace import WorkSpaceNotAllowedCreateError, Workspaces
 from services.feature_service import FeatureService
 
 
+@console_ns.route("/login")
 class LoginApi(Resource):
     """Resource for user login."""
 
@@ -91,6 +92,7 @@ class LoginApi(Resource):
         return {"result": "success", "data": token_pair.model_dump()}
 
 
+@console_ns.route("/logout")
 class LogoutApi(Resource):
     @setup_required
     def get(self):
@@ -102,6 +104,7 @@ class LogoutApi(Resource):
         return {"result": "success"}
 
 
+@console_ns.route("/reset-password")
 class ResetPasswordSendEmailApi(Resource):
     @setup_required
     @email_password_login_enabled
@@ -130,6 +133,7 @@ class ResetPasswordSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/email-code-login")
 class EmailCodeLoginSendEmailApi(Resource):
     @setup_required
     def post(self):
@@ -162,6 +166,7 @@ class EmailCodeLoginSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@console_ns.route("/email-code-login/validity")
 class EmailCodeLoginApi(Resource):
     @setup_required
     def post(self):
@@ -218,6 +223,7 @@ class EmailCodeLoginApi(Resource):
         return {"result": "success", "data": token_pair.model_dump()}
 
 
+@console_ns.route("/refresh-token")
 class RefreshTokenApi(Resource):
     def post(self):
         parser = reqparse.RequestParser()
@@ -229,11 +235,3 @@ class RefreshTokenApi(Resource):
             return {"result": "success", "data": new_token_pair.model_dump()}
         except Exception as e:
             return {"result": "fail", "data": str(e)}, 401
-
-
-api.add_resource(LoginApi, "/login")
-api.add_resource(LogoutApi, "/logout")
-api.add_resource(EmailCodeLoginSendEmailApi, "/email-code-login")
-api.add_resource(EmailCodeLoginApi, "/email-code-login/validity")
-api.add_resource(ResetPasswordSendEmailApi, "/reset-password")
-api.add_resource(RefreshTokenApi, "/refresh-token")

api/controllers/console/auth/oauth.py

@@ -1,6 +1,6 @@
 import logging
 
-import requests
+import httpx
 from flask import current_app, redirect, request
 from flask_restx import Resource
 from sqlalchemy import select
@@ -101,8 +101,10 @@ class OAuthCallback(Resource):
         try:
             token = oauth_provider.get_access_token(code)
             user_info = oauth_provider.get_user_info(token)
-        except requests.RequestException as e:
-            error_text = e.response.text if e.response else str(e)
+        except httpx.RequestError as e:
+            error_text = str(e)
+            if isinstance(e, httpx.HTTPStatusError):
+                error_text = e.response.text
             logger.exception("An error occurred during the OAuth process with %s: %s", provider, error_text)
             return {"error": "OAuth process failed"}, 400
 

api/controllers/console/auth/oauth_server.py

@@ -14,7 +14,7 @@ from models.account import Account
 from models.model import OAuthProviderApp
 from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType, OAuthServerService
 
-from .. import api
+from .. import console_ns
 
 P = ParamSpec("P")
 R = TypeVar("R")
@@ -86,6 +86,7 @@ def oauth_server_access_token_required(view: Callable[Concatenate[T, OAuthProvid
     return decorated
 
 
+@console_ns.route("/oauth/provider")
 class OAuthServerAppApi(Resource):
     @setup_required
     @oauth_server_client_id_required
@@ -108,6 +109,7 @@ class OAuthServerAppApi(Resource):
         )
 
 
+@console_ns.route("/oauth/provider/authorize")
 class OAuthServerUserAuthorizeApi(Resource):
     @setup_required
     @login_required
@@ -125,6 +127,7 @@ class OAuthServerUserAuthorizeApi(Resource):
         )
 
 
+@console_ns.route("/oauth/provider/token")
 class OAuthServerUserTokenApi(Resource):
     @setup_required
     @oauth_server_client_id_required
@@ -180,6 +183,7 @@ class OAuthServerUserTokenApi(Resource):
             )
 
 
+@console_ns.route("/oauth/provider/account")
 class OAuthServerUserAccountApi(Resource):
     @setup_required
     @oauth_server_client_id_required
@@ -194,9 +198,3 @@ class OAuthServerUserAccountApi(Resource):
                 "timezone": account.timezone,
             }
         )
-
-
-api.add_resource(OAuthServerAppApi, "/oauth/provider")
-api.add_resource(OAuthServerUserAuthorizeApi, "/oauth/provider/authorize")
-api.add_resource(OAuthServerUserTokenApi, "/oauth/provider/token")
-api.add_resource(OAuthServerUserAccountApi, "/oauth/provider/account")

api/controllers/console/billing/billing.py

@@ -1,12 +1,13 @@
 from flask_restx import Resource, reqparse
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
 from libs.login import current_user, login_required
 from models.model import Account
 from services.billing_service import BillingService
 
 
+@console_ns.route("/billing/subscription")
 class Subscription(Resource):
     @setup_required
     @login_required
@@ -26,6 +27,7 @@ class Subscription(Resource):
         )
 
 
+@console_ns.route("/billing/invoices")
 class Invoices(Resource):
     @setup_required
     @login_required
@@ -36,7 +38,3 @@ class Invoices(Resource):
         BillingService.is_tenant_owner_or_admin(current_user)
         assert current_user.current_tenant_id is not None
         return BillingService.get_invoices(current_user.email, current_user.current_tenant_id)
-
-
-api.add_resource(Subscription, "/billing/subscription")
-api.add_resource(Invoices, "/billing/invoices")

api/controllers/console/billing/compliance.py

@@ -6,10 +6,11 @@ from libs.helper import extract_remote_ip
 from libs.login import login_required
 from services.billing_service import BillingService
 
-from .. import api
+from .. import console_ns
 from ..wraps import account_initialization_required, only_edition_cloud, setup_required
 
 
+@console_ns.route("/compliance/download")
 class ComplianceApi(Resource):
     @setup_required
     @login_required
@@ -30,6 +31,3 @@ class ComplianceApi(Resource):
             ip=ip_address,
             device_info=device_info,
         )
-
-
-api.add_resource(ComplianceApi, "/compliance/download")

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
@@ -7,8 +9,10 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound
 
-from controllers.console import api
+from controllers.console import console_ns
 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.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
@@ -19,9 +23,14 @@ 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
 
 
+@console_ns.route(
+    "/data-source/integrates",
+    "/data-source/integrates/<uuid:binding_id>/<string:action>",
+)
 class DataSourceApi(Resource):
     @setup_required
     @login_required
@@ -104,6 +113,7 @@ class DataSourceApi(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/notion/pre-import/pages")
 class DataSourceNotionListApi(Resource):
     @setup_required
     @login_required
@@ -111,6 +121,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
@@ -134,59 +156,79 @@ 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
 
 
+@console_ns.route(
+    "/notion/workspaces/<uuid:workspace_id>/pages/<uuid:page_id>/<string:page_type>/preview",
+    "/datasets/notion-indexing-estimate",
+)
 class DataSourceNotionApi(Resource):
     @setup_required
     @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,
         )
 
@@ -211,10 +253,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=DatasourceType.NOTION.value,
                     notion_info={
+                        "credential_id": credential_id,
                         "notion_workspace_id": workspace_id,
                         "notion_obj_id": page["page_id"],
                         "notion_page_type": page["type"],
@@ -234,6 +278,7 @@ class DataSourceNotionApi(Resource):
         return response.model_dump(), 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/notion/sync")
 class DataSourceNotionDatasetSyncApi(Resource):
     @setup_required
     @login_required
@@ -250,6 +295,7 @@ class DataSourceNotionDatasetSyncApi(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/notion/sync")
 class DataSourceNotionDocumentSyncApi(Resource):
     @setup_required
     @login_required
@@ -266,16 +312,3 @@ class DataSourceNotionDocumentSyncApi(Resource):
             raise NotFound("Document not found.")
         document_indexing_sync_task.delay(dataset_id_str, document_id_str)
         return {"result": "success"}, 200
-
-
-api.add_resource(DataSourceApi, "/data-source/integrates", "/data-source/integrates/<uuid:binding_id>/<string:action>")
-api.add_resource(DataSourceNotionListApi, "/notion/pre-import/pages")
-api.add_resource(
-    DataSourceNotionApi,
-    "/notion/workspaces/<uuid:workspace_id>/pages/<uuid:page_id>/<string:page_type>/preview",
-    "/datasets/notion-indexing-estimate",
-)
-api.add_resource(DataSourceNotionDatasetSyncApi, "/datasets/<uuid:dataset_id>/notion/sync")
-api.add_resource(
-    DataSourceNotionDocumentSyncApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/notion/sync"
-)

api/controllers/console/datasets/datasets.py

@@ -1,4 +1,5 @@
-import flask_restx
+from typing import Any, cast
+
 from flask import request
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal, marshal_with, reqparse
@@ -30,24 +31,20 @@ from fields.app_fields import related_app_list
 from fields.dataset_fields import dataset_detail_fields, dataset_query_detail_fields
 from fields.document_fields import document_status_fields
 from libs.login import login_required
+from libs.validators import validate_description_length
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
+from models.account import Account
 from models.dataset import DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 
-def _validate_name(name):
+def _validate_name(name: str) -> str:
     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 description and len(description) > 400:
-        raise ValueError("Description cannot exceed 400 characters.")
-    return description
-
-
 @console_ns.route("/datasets")
 class DatasetListApi(Resource):
     @api.doc("get_datasets")
@@ -92,7 +89,7 @@ class DatasetListApi(Resource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        data = marshal(datasets, dataset_detail_fields)
+        data = cast(list[dict[str, Any]], marshal(datasets, dataset_detail_fields))
         for item in data:
             # convert embedding_model_provider to plugin standard format
             if item["indexing_technique"] == "high_quality" and item["embedding_model_provider"]:
@@ -147,7 +144,7 @@ class DatasetListApi(Resource):
         )
         parser.add_argument(
             "description",
-            type=_validate_description_length,
+            type=validate_description_length,
             nullable=True,
             required=False,
             default="",
@@ -192,7 +189,7 @@ class DatasetListApi(Resource):
                 name=args["name"],
                 description=args["description"],
                 indexing_technique=args["indexing_technique"],
-                account=current_user,
+                account=cast(Account, current_user),
                 permission=DatasetPermissionEnum.ONLY_ME,
                 provider=args["provider"],
                 external_knowledge_api_id=args["external_knowledge_api_id"],
@@ -224,7 +221,7 @@ class DatasetApi(Resource):
             DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        data = marshal(dataset, dataset_detail_fields)
+        data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         if dataset.indexing_technique == "high_quality":
             if dataset.embedding_model_provider:
                 provider_id = ModelProviderID(dataset.embedding_model_provider)
@@ -288,7 +285,7 @@ class DatasetApi(Resource):
             help="type is required. Name must be between 1 to 40 characters.",
             type=_validate_name,
         )
-        parser.add_argument("description", location="json", store_missing=False, type=_validate_description_length)
+        parser.add_argument("description", location="json", store_missing=False, type=validate_description_length)
         parser.add_argument(
             "indexing_technique",
             type=str,
@@ -337,6 +334,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()
 
@@ -360,7 +366,7 @@ class DatasetApi(Resource):
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        result_data = marshal(dataset, dataset_detail_fields)
+        result_data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         tenant_id = current_user.current_tenant_id
 
         if data.get("partial_member_list") and data.get("permission") == "partial_members":
@@ -387,7 +393,7 @@ class DatasetApi(Resource):
         dataset_id_str = str(dataset_id)
 
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor or current_user.is_dataset_operator:
+        if not (current_user.is_editor or current_user.is_dataset_operator):
             raise Forbidden()
 
         try:
@@ -503,10 +509,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=DatasourceType.NOTION.value,
                         notion_info={
+                            "credential_id": credential_id,
                             "notion_workspace_id": workspace_id,
                             "notion_obj_id": page["page_id"],
                             "notion_page_type": page["type"],
@@ -677,7 +685,7 @@ class DatasetApiKeyApi(Resource):
         )
 
         if current_key_count >= self.max_keys:
-            flask_restx.abort(
+            api.abort(
                 400,
                 message=f"Cannot create more than {self.max_keys} API keys for this resource type.",
                 code="max_keys_exceeded",
@@ -722,7 +730,7 @@ class DatasetApiDeleteApi(Resource):
         )
 
         if key is None:
-            flask_restx.abort(404, message="API key not found")
+            api.abort(404, message="API key not found")
 
         db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
@@ -730,6 +738,19 @@ class DatasetApiDeleteApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/api-keys/<string:status>")
+class DatasetEnableApiApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id, status):
+        dataset_id_str = str(dataset_id)
+
+        DatasetService.update_dataset_api_status(dataset_id_str, status == "enable")
+
+        return {"result": "success"}, 200
+
+
 @console_ns.route("/datasets/api-base-info")
 class DatasetApiBaseUrlApi(Resource):
     @api.doc("get_dataset_api_base_info")
@@ -758,7 +779,6 @@ class DatasetRetrievalSettingApi(Resource):
                 | VectorType.TIDB_VECTOR
                 | VectorType.CHROMA
                 | VectorType.PGVECTO_RS
-                | VectorType.BAIDU
                 | VectorType.VIKINGDB
                 | VectorType.UPSTASH
             ):
@@ -785,6 +805,7 @@ class DatasetRetrievalSettingApi(Resource):
                 | VectorType.TENCENT
                 | VectorType.MATRIXONE
                 | VectorType.CLICKZETTA
+                | VectorType.BAIDU
             ):
                 return {
                     "retrieval_method": [
@@ -814,7 +835,6 @@ class DatasetRetrievalSettingMockApi(Resource):
                 | VectorType.TIDB_VECTOR
                 | VectorType.CHROMA
                 | VectorType.PGVECTO_RS
-                | VectorType.BAIDU
                 | VectorType.VIKINGDB
                 | VectorType.UPSTASH
             ):
@@ -839,6 +859,7 @@ class DatasetRetrievalSettingMockApi(Resource):
                 | VectorType.HUAWEI_CLOUD
                 | VectorType.MATRIXONE
                 | VectorType.CLICKZETTA
+                | VectorType.BAIDU
             ):
                 return {
                     "retrieval_method": [

api/controllers/console/datasets/datasets_document.py

@@ -1,8 +1,10 @@
+import json
 import logging
 from argparse import ArgumentTypeError
 from collections.abc import Sequence
 from typing import Literal, cast
 
+import sqlalchemy as sa
 from flask import request
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal, marshal_with, reqparse
@@ -53,6 +55,8 @@ 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.account import Account
+from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 
@@ -209,13 +213,13 @@ class DatasetDocumentListApi(Resource):
 
         if sort == "hit_count":
             sub_query = (
-                db.select(DocumentSegment.document_id, db.func.sum(DocumentSegment.hit_count).label("total_hit_count"))
+                sa.select(DocumentSegment.document_id, sa.func.sum(DocumentSegment.hit_count).label("total_hit_count"))
                 .group_by(DocumentSegment.document_id)
                 .subquery()
             )
 
             query = query.outerjoin(sub_query, sub_query.c.document_id == Document.id).order_by(
-                sort_logic(db.func.coalesce(sub_query.c.total_hit_count, 0)),
+                sort_logic(sa.func.coalesce(sub_query.c.total_hit_count, 0)),
                 sort_logic(Document.position),
             )
         elif sort == "created_at":
@@ -415,7 +419,9 @@ class DatasetInitApi(Resource):
 
         try:
             dataset, documents, batch = DocumentService.save_document_without_dataset_id(
-                tenant_id=current_user.current_tenant_id, knowledge_config=knowledge_config, account=current_user
+                tenant_id=current_user.current_tenant_id,
+                knowledge_config=knowledge_config,
+                account=cast(Account, current_user),
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -449,7 +455,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
             raise DocumentAlreadyFinishedError()
 
         data_process_rule = document.dataset_process_rule
-        data_process_rule_dict = data_process_rule.to_dict()
+        data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
 
         response = {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}
 
@@ -511,7 +517,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
         if not documents:
             return {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}, 200
         data_process_rule = documents[0].dataset_process_rule
-        data_process_rule_dict = data_process_rule.to_dict()
+        data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
         extract_settings = []
         for document in documents:
             if document.indexing_status in {"completed", "error"}:
@@ -542,6 +548,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 extract_setting = ExtractSetting(
                     datasource_type=DatasourceType.NOTION.value,
                     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"],
@@ -716,7 +723,7 @@ class DocumentApi(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,
@@ -749,7 +756,7 @@ class DocumentApi(DocumentResource):
             }
         else:
             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,
@@ -1069,7 +1076,9 @@ class DocumentRenameApi(DocumentResource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         dataset = DatasetService.get_dataset(dataset_id)
-        DatasetService.check_dataset_operator_permission(current_user, dataset)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        DatasetService.check_dataset_operator_permission(cast(Account, current_user), dataset)
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
@@ -1108,3 +1117,39 @@ class WebsiteDocumentSyncApi(DocumentResource):
         DocumentService.sync_website_document(dataset_id, document)
 
         return {"result": "success"}, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log")
+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/controllers/console/datasets/datasets_segments.py

@@ -7,7 +7,7 @@ from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.app.error import ProviderNotInitializeError
 from controllers.console.datasets.error import (
     ChildChunkDeleteIndexError,
@@ -37,6 +37,7 @@ from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingS
 from tasks.batch_create_segment_to_index_task import batch_create_segment_to_index_task
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments")
 class DatasetDocumentSegmentListApi(Resource):
     @setup_required
     @login_required
@@ -139,6 +140,7 @@ class DatasetDocumentSegmentListApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment/<string:action>")
 class DatasetDocumentSegmentApi(Resource):
     @setup_required
     @login_required
@@ -193,6 +195,7 @@ class DatasetDocumentSegmentApi(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment")
 class DatasetDocumentSegmentAddApi(Resource):
     @setup_required
     @login_required
@@ -244,6 +247,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>")
 class DatasetDocumentSegmentUpdateApi(Resource):
     @setup_required
     @login_required
@@ -345,6 +349,10 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route(
+    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/batch_import",
+    "/datasets/batch_import_status/<uuid:job_id>",
+)
 class DatasetDocumentSegmentBatchImportApi(Resource):
     @setup_required
     @login_required
@@ -384,7 +392,12 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
             # send batch add segments task
             redis_client.setnx(indexing_cache_key, "waiting")
             batch_create_segment_to_index_task.delay(
-                str(job_id), upload_file_id, dataset_id, document_id, current_user.current_tenant_id, current_user.id
+                str(job_id),
+                upload_file_id,
+                dataset_id,
+                document_id,
+                current_user.current_tenant_id,
+                current_user.id,
             )
         except Exception as e:
             return {"error": str(e)}, 500
@@ -393,7 +406,9 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, job_id):
+    def get(self, job_id=None, dataset_id=None, document_id=None):
+        if job_id is None:
+            raise NotFound("The job does not exist.")
         job_id = str(job_id)
         indexing_cache_key = f"segment_batch_import_{job_id}"
         cache_result = redis_client.get(indexing_cache_key)
@@ -403,6 +418,7 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
         return {"job_id": job_id, "job_status": cache_result.decode()}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>/child_chunks")
 class ChildChunkAddApi(Resource):
     @setup_required
     @login_required
@@ -457,7 +473,8 @@ class ChildChunkAddApi(Resource):
         parser.add_argument("content", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
         try:
-            child_chunk = SegmentService.create_child_chunk(args.get("content"), segment, document, dataset)
+            content = args["content"]
+            child_chunk = SegmentService.create_child_chunk(content, segment, document, dataset)
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))
         return {"data": marshal(child_chunk, child_chunk_fields)}, 200
@@ -546,13 +563,17 @@ class ChildChunkAddApi(Resource):
         parser.add_argument("chunks", type=list, required=True, nullable=False, location="json")
         args = parser.parse_args()
         try:
-            chunks = [ChildChunkUpdateArgs(**chunk) for chunk in args.get("chunks")]
+            chunks_data = args["chunks"]
+            chunks = [ChildChunkUpdateArgs(**chunk) for chunk in chunks_data]
             child_chunks = SegmentService.update_child_chunks(chunks, segment, document, dataset)
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))
         return {"data": marshal(child_chunks, child_chunk_fields)}, 200
 
 
+@console_ns.route(
+    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>/child_chunks/<uuid:child_chunk_id>"
+)
 class ChildChunkUpdateApi(Resource):
     @setup_required
     @login_required
@@ -660,33 +681,8 @@ class ChildChunkUpdateApi(Resource):
         parser.add_argument("content", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
         try:
-            child_chunk = SegmentService.update_child_chunk(
-                args.get("content"), child_chunk, segment, document, dataset
-            )
+            content = args["content"]
+            child_chunk = SegmentService.update_child_chunk(content, child_chunk, segment, document, dataset)
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))
         return {"data": marshal(child_chunk, child_chunk_fields)}, 200
-
-
-api.add_resource(DatasetDocumentSegmentListApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments")
-api.add_resource(
-    DatasetDocumentSegmentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment/<string:action>"
-)
-api.add_resource(DatasetDocumentSegmentAddApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment")
-api.add_resource(
-    DatasetDocumentSegmentUpdateApi,
-    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>",
-)
-api.add_resource(
-    DatasetDocumentSegmentBatchImportApi,
-    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/batch_import",
-    "/datasets/batch_import_status/<uuid:job_id>",
-)
-api.add_resource(
-    ChildChunkAddApi,
-    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>/child_chunks",
-)
-api.add_resource(
-    ChildChunkUpdateApi,
-    "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>/child_chunks/<uuid:child_chunk_id>",
-)

api/controllers/console/datasets/error.py

@@ -71,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/external.py

@@ -1,3 +1,5 @@
+from typing import cast
+
 from flask import request
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal, reqparse
@@ -9,13 +11,14 @@ from controllers.console.datasets.error import DatasetNameDuplicateError
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.dataset_fields import dataset_detail_fields
 from libs.login import login_required
+from models.account import Account
 from services.dataset_service import DatasetService
 from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
 from services.knowledge_service import ExternalDatasetTestService
 
 
-def _validate_name(name):
+def _validate_name(name: str) -> str:
     if not name or len(name) < 1 or len(name) > 100:
         raise ValueError("Name must be between 1 to 100 characters.")
     return name
@@ -148,7 +151,7 @@ class ExternalApiTemplateApi(Resource):
         external_knowledge_api_id = str(external_knowledge_api_id)
 
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor or current_user.is_dataset_operator:
+        if not (current_user.is_editor or current_user.is_dataset_operator):
             raise Forbidden()
 
         ExternalDatasetService.delete_external_knowledge_api(current_user.current_tenant_id, external_knowledge_api_id)
@@ -274,7 +277,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
             response = HitTestingService.external_retrieve(
                 dataset=dataset,
                 query=args["query"],
-                account=current_user,
+                account=cast(Account, current_user),
                 external_retrieval_model=args["external_retrieval_model"],
                 metadata_filtering_conditions=args["metadata_filtering_conditions"],
             )

api/controllers/console/datasets/hit_testing_base.py

@@ -1,10 +1,11 @@
 import logging
+from typing import cast
 
 from flask_login import current_user
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
-import services.dataset_service
+import services
 from controllers.console.app.error import (
     CompletionRequestError,
     ProviderModelCurrentlyNotSupportError,
@@ -20,6 +21,7 @@ from core.errors.error import (
 )
 from core.model_runtime.errors.invoke import InvokeError
 from fields.hit_testing_fields import hit_testing_record_fields
+from models.account import Account
 from services.dataset_service import DatasetService
 from services.hit_testing_service import HitTestingService
 
@@ -59,7 +61,7 @@ class DatasetsHitTestingBase:
             response = HitTestingService.retrieve(
                 dataset=dataset,
                 query=args["query"],
-                account=current_user,
+                account=cast(Account, current_user),
                 retrieval_model=args["retrieval_model"],
                 external_retrieval_model=args["external_retrieval_model"],
                 limit=10,

api/controllers/console/datasets/metadata.py

@@ -4,7 +4,7 @@ from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from werkzeug.exceptions import NotFound
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
 from fields.dataset_fields import dataset_metadata_fields
 from libs.login import login_required
@@ -16,6 +16,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
 from services.metadata_service import MetadataService
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/metadata")
 class DatasetMetadataCreateApi(Resource):
     @setup_required
     @login_required
@@ -50,6 +51,7 @@ class DatasetMetadataCreateApi(Resource):
         return MetadataService.get_dataset_metadatas(dataset), 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/metadata/<uuid:metadata_id>")
 class DatasetMetadataApi(Resource):
     @setup_required
     @login_required
@@ -60,6 +62,7 @@ class DatasetMetadataApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
         args = parser.parse_args()
+        name = args["name"]
 
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
@@ -68,7 +71,7 @@ class DatasetMetadataApi(Resource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, args.get("name"))
+        metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, name)
         return metadata, 200
 
     @setup_required
@@ -87,6 +90,7 @@ class DatasetMetadataApi(Resource):
         return {"result": "success"}, 204
 
 
+@console_ns.route("/datasets/metadata/built-in")
 class DatasetMetadataBuiltInFieldApi(Resource):
     @setup_required
     @login_required
@@ -97,6 +101,7 @@ class DatasetMetadataBuiltInFieldApi(Resource):
         return {"fields": built_in_fields}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in/<string:action>")
 class DatasetMetadataBuiltInFieldActionApi(Resource):
     @setup_required
     @login_required
@@ -116,6 +121,7 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/metadata")
 class DocumentMetadataEditApi(Resource):
     @setup_required
     @login_required
@@ -136,10 +142,3 @@ class DocumentMetadataEditApi(Resource):
         MetadataService.update_documents_metadata(dataset, metadata_args)
 
         return {"result": "success"}, 200
-
-
-api.add_resource(DatasetMetadataCreateApi, "/datasets/<uuid:dataset_id>/metadata")
-api.add_resource(DatasetMetadataApi, "/datasets/<uuid:dataset_id>/metadata/<uuid:metadata_id>")
-api.add_resource(DatasetMetadataBuiltInFieldApi, "/datasets/metadata/built-in")
-api.add_resource(DatasetMetadataBuiltInFieldActionApi, "/datasets/<uuid:dataset_id>/metadata/built-in/<string:action>")
-api.add_resource(DocumentMetadataEditApi, "/datasets/<uuid:dataset_id>/documents/metadata")

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

@@ -0,0 +1,323 @@
+from flask import make_response, redirect, request
+from flask_login import current_user
+from flask_restx import Resource, reqparse
+from werkzeug.exceptions import Forbidden, NotFound
+
+from configs import dify_config
+from controllers.console import console_ns
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.model_runtime.errors.validate import CredentialsValidateFailedError
+from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.impl.oauth import OAuthHandler
+from libs.helper import StrLen
+from libs.login import login_required
+from models.provider_ids import DatasourceProviderID
+from services.datasource_provider_service import DatasourceProviderService
+from services.plugin.oauth_service import OAuthProxyService
+
+
+@console_ns.route("/oauth/plugin/<path:provider_id>/datasource/get-authorization-url")
+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
+
+
+@console_ns.route("/oauth/plugin/<path:provider_id>/datasource/callback")
+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")
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>/delete")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>/update")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/list")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/default-list")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>/custom-client")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>/default")
+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
+
+
+@console_ns.route("/auth/plugin/datasource/<path:provider_id>/update-name")
+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/controllers/console/datasets/rag_pipeline/datasource_content_preview.py

@@ -0,0 +1,52 @@
+from flask_restx import (  # type: ignore
+    Resource,  # type: ignore
+    reqparse,
+)
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import console_ns
+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
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview")
+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/controllers/console/datasets/rag_pipeline/rag_pipeline.py

@@ -0,0 +1,150 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, reqparse
+from sqlalchemy.orm import Session
+
+from controllers.console import console_ns
+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: str) -> str:
+    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: str) -> str:
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+@console_ns.route("/rag/pipeline/templates")
+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
+
+
+@console_ns.route("/rag/pipeline/templates/<string:template_id>")
+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
+
+
+@console_ns.route("/rag/pipeline/customized/templates/<string:template_id>")
+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=_validate_description_length,
+            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).where(PipelineCustomizedTemplate.id == template_id).first()
+            )
+            if not template:
+                raise ValueError("Customized pipeline template not found.")
+
+        return {"data": template.yaml_content}, 200
+
+
+@console_ns.route("/rag/pipelines/<string:pipeline_id>/customized/publish")
+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=_validate_description_length,
+            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/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py

@@ -0,0 +1,100 @@
+from flask_login import current_user
+from flask_restx import Resource, marshal, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+import services
+from controllers.console import console_ns
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.wraps import (
+    account_initialization_required,
+    cloud_edition_billing_rate_limit_check,
+    setup_required,
+)
+from extensions.ext_database import db
+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 IconInfo, RagPipelineDatasetCreateEntity
+from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
+
+
+@console_ns.route("/rag/pipeline/dataset")
+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(
+            "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(
+            name="",
+            description="",
+            icon_info=IconInfo(
+                icon="📙",
+                icon_background="#FFF4ED",
+                icon_type="emoji",
+            ),
+            permission=DatasetPermissionEnum.ONLY_ME,
+            partial_member_list=None,
+            yaml_content=args["yaml_content"],
+        )
+        try:
+            with Session(db.engine) as session:
+                rag_pipeline_dsl_service = RagPipelineDslService(session)
+                import_info = rag_pipeline_dsl_service.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
+
+
+@console_ns.route("/rag/pipeline/empty-dataset")
+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()
+        dataset = DatasetService.create_empty_rag_pipeline_dataset(
+            tenant_id=current_user.current_tenant_id,
+            rag_pipeline_dataset_create_entity=RagPipelineDatasetCreateEntity(
+                name="",
+                description="",
+                icon_info=IconInfo(
+                    icon="📙",
+                    icon_background="#FFF4ED",
+                    icon_type="emoji",
+                ),
+                permission=DatasetPermissionEnum.ONLY_ME,
+                partial_member_list=None,
+            ),
+        )
+        return marshal(dataset, dataset_detail_fields), 201

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

@@ -0,0 +1,344 @@
+import logging
+from typing import NoReturn
+
+from flask import Response
+from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import console_ns
+from controllers.console.app.error import (
+    DraftWorkflowNotExist,
+)
+from controllers.console.app.workflow_draft_variable import (
+    _WORKFLOW_DRAFT_VARIABLE_FIELDS,  # type: ignore[private-usage]
+    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,  # type: ignore[private-usage]
+)
+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.types import SegmentType
+from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from extensions.ext_database import db
+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.account import Account
+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 _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
+
+
+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 isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables")
+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
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/variables")
+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)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>")
+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)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+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
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/system-variables")
+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)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/environment-variables")
+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/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py

@@ -0,0 +1,134 @@
+from typing import cast
+
+from flask_login import current_user  # type: ignore
+from flask_restx import Resource, marshal_with, reqparse  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import console_ns
+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
+
+
+@console_ns.route("/rag/pipelines/imports")
+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
+
+
+@console_ns.route("/rag/pipelines/imports/<string:import_id>/confirm")
+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
+
+
+@console_ns.route("/rag/pipelines/imports/<string:pipeline_id>/check-dependencies")
+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
+
+
+@console_ns.route("/rag/pipelines/<string:pipeline_id>/exports")
+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=str, 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"] == "true"
+            )
+
+        return {"data": result}, 200

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

@@ -0,0 +1,994 @@
+import json
+import logging
+from typing import cast
+
+from flask import abort, request
+from flask_restx import Resource, inputs, marshal_with, reqparse  # type: ignore  # type: ignore
+from flask_restx.inputs import int_range  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
+
+import services
+from controllers.console import console_ns
+from controllers.console.app.error import (
+    ConversationCompletedError,
+    DraftWorkflowNotExist,
+    DraftWorkflowNotSync,
+)
+from controllers.console.datasets.wraps import get_rag_pipeline
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.model_runtime.utils.encoders import jsonable_encoder
+from extensions.ext_database import db
+from factories import variable_factory
+from fields.workflow_fields import workflow_fields, workflow_pagination_fields
+from fields.workflow_run_fields import (
+    workflow_run_detail_fields,
+    workflow_run_node_execution_fields,
+    workflow_run_node_execution_list_fields,
+    workflow_run_pagination_fields,
+)
+from libs import helper
+from libs.helper import TimestampField, uuid_value
+from libs.login import current_user, login_required
+from models.account import Account
+from models.dataset import Pipeline
+from models.model import EndUser
+from services.errors.app import WorkflowHashNotEqualError
+from services.errors.llm import InvokeRateLimitError
+from services.rag_pipeline.pipeline_generate_service import PipelineGenerateService
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+from services.rag_pipeline.rag_pipeline_manage_service import RagPipelineManageService
+from services.rag_pipeline.rag_pipeline_transform_service import RagPipelineTransformService
+
+logger = logging.getLogger(__name__)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft")
+class DraftRagPipelineApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_fields)
+    def get(self, pipeline: Pipeline):
+        """
+        Get draft rag pipeline's workflow
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        # fetch draft workflow by app_model
+        rag_pipeline_service = RagPipelineService()
+        workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        # return workflow, if not found, return None (initiate graph by frontend)
+        return workflow
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline):
+        """
+        Sync draft workflow
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        content_type = request.headers.get("Content-Type", "")
+
+        if "application/json" in content_type:
+            parser = reqparse.RequestParser()
+            parser.add_argument("graph", type=dict, required=True, nullable=False, location="json")
+            parser.add_argument("hash", type=str, required=False, location="json")
+            parser.add_argument("environment_variables", type=list, required=False, location="json")
+            parser.add_argument("conversation_variables", type=list, required=False, location="json")
+            parser.add_argument("rag_pipeline_variables", type=list, required=False, location="json")
+            args = parser.parse_args()
+        elif "text/plain" in content_type:
+            try:
+                data = json.loads(request.data.decode("utf-8"))
+                if "graph" not in data or "features" not in data:
+                    raise ValueError("graph or features not found in data")
+
+                if not isinstance(data.get("graph"), dict):
+                    raise ValueError("graph is not a dict")
+
+                args = {
+                    "graph": data.get("graph"),
+                    "features": data.get("features"),
+                    "hash": data.get("hash"),
+                    "environment_variables": data.get("environment_variables"),
+                    "conversation_variables": data.get("conversation_variables"),
+                    "rag_pipeline_variables": data.get("rag_pipeline_variables"),
+                }
+            except json.JSONDecodeError:
+                return {"message": "Invalid JSON data"}, 400
+        else:
+            abort(415)
+
+        try:
+            environment_variables_list = args.get("environment_variables") or []
+            environment_variables = [
+                variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+            ]
+            conversation_variables_list = args.get("conversation_variables") or []
+            conversation_variables = [
+                variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+            ]
+            rag_pipeline_service = RagPipelineService()
+            workflow = rag_pipeline_service.sync_draft_workflow(
+                pipeline=pipeline,
+                graph=args["graph"],
+                unique_hash=args.get("hash"),
+                account=current_user,
+                environment_variables=environment_variables,
+                conversation_variables=conversation_variables,
+                rag_pipeline_variables=args.get("rag_pipeline_variables") or [],
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/iteration/nodes/<string:node_id>/run")
+class RagPipelineDraftRunIterationNodeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run draft workflow iteration node
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
+
+        try:
+            response = PipelineGenerateService.generate_single_iteration(
+                pipeline=pipeline, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except ValueError as e:
+            raise e
+        except Exception:
+            logging.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/loop/nodes/<string:node_id>/run")
+class RagPipelineDraftRunLoopNodeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run draft workflow loop node
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
+
+        try:
+            response = PipelineGenerateService.generate_single_loop(
+                pipeline=pipeline, user=current_user, node_id=node_id, args=args, streaming=True
+            )
+
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except ValueError as e:
+            raise e
+        except Exception:
+            logging.exception("internal server error.")
+            raise InternalServerError()
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/run")
+class DraftRagPipelineRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline):
+        """
+        Run draft workflow
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            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("datasource_info_list", type=list, required=True, location="json")
+        parser.add_argument("start_node_id", type=str, required=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            response = PipelineGenerateService.generate(
+                pipeline=pipeline,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+            )
+
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/run")
+class PublishedRagPipelineRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline):
+        """
+        Run published workflow
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            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("datasource_info_list", type=list, required=True, location="json")
+        parser.add_argument("start_node_id", type=str, required=True, location="json")
+        parser.add_argument("is_preview", type=bool, required=True, location="json", default=False)
+        parser.add_argument("response_mode", type=str, required=True, location="json", default="streaming")
+        parser.add_argument("original_document_id", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        streaming = args["response_mode"] == "streaming"
+
+        try:
+            response = PipelineGenerateService.generate(
+                pipeline=pipeline,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.DEBUGGER if args.get("is_preview") else InvokeFrom.PUBLISHED,
+                streaming=streaming,
+            )
+
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+
+
+# class RagPipelinePublishedDatasourceNodeRunStatusApi(Resource):
+#     @setup_required
+#     @login_required
+#     @account_initialization_required
+#     @get_rag_pipeline
+#     def post(self, pipeline: Pipeline, node_id: str):
+#         """
+#         Run rag pipeline datasource
+#         """
+#         # The role of the current user in the ta table must be admin, owner, or editor
+#         if not current_user.is_editor:
+#             raise Forbidden()
+#
+#         if not isinstance(current_user, Account):
+#             raise Forbidden()
+#
+#         parser = reqparse.RequestParser()
+#         parser.add_argument("job_id", type=str, required=True, nullable=False, location="json")
+#         parser.add_argument("datasource_type", type=str, required=True, location="json")
+#         args = parser.parse_args()
+#
+#         job_id = args.get("job_id")
+#         if job_id == None:
+#             raise ValueError("missing job_id")
+#         datasource_type = args.get("datasource_type")
+#         if datasource_type == None:
+#             raise ValueError("missing datasource_type")
+#
+#         rag_pipeline_service = RagPipelineService()
+#         result = rag_pipeline_service.run_datasource_workflow_node_status(
+#             pipeline=pipeline,
+#             node_id=node_id,
+#             job_id=job_id,
+#             account=current_user,
+#             datasource_type=datasource_type,
+#             is_published=True
+#         )
+#
+#         return result
+
+
+# class RagPipelineDraftDatasourceNodeRunStatusApi(Resource):
+#     @setup_required
+#     @login_required
+#     @account_initialization_required
+#     @get_rag_pipeline
+#     def post(self, pipeline: Pipeline, node_id: str):
+#         """
+#         Run rag pipeline datasource
+#         """
+#         # The role of the current user in the ta table must be admin, owner, or editor
+#         if not current_user.is_editor:
+#             raise Forbidden()
+#
+#         if not isinstance(current_user, Account):
+#             raise Forbidden()
+#
+#         parser = reqparse.RequestParser()
+#         parser.add_argument("job_id", type=str, required=True, nullable=False, location="json")
+#         parser.add_argument("datasource_type", type=str, required=True, location="json")
+#         args = parser.parse_args()
+#
+#         job_id = args.get("job_id")
+#         if job_id == None:
+#             raise ValueError("missing job_id")
+#         datasource_type = args.get("datasource_type")
+#         if datasource_type == None:
+#             raise ValueError("missing datasource_type")
+#
+#         rag_pipeline_service = RagPipelineService()
+#         result = rag_pipeline_service.run_datasource_workflow_node_status(
+#             pipeline=pipeline,
+#             node_id=node_id,
+#             job_id=job_id,
+#             account=current_user,
+#             datasource_type=datasource_type,
+#             is_published=False
+#         )
+#
+#         return result
+#
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/run")
+class RagPipelinePublishedDatasourceNodeRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run rag pipeline datasource
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            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()
+        return helper.compact_generate_response(
+            PipelineGenerator.convert_to_event_stream(
+                rag_pipeline_service.run_datasource_workflow_node(
+                    pipeline=pipeline,
+                    node_id=node_id,
+                    user_inputs=inputs,
+                    account=current_user,
+                    datasource_type=datasource_type,
+                    is_published=False,
+                    credential_id=args.get("credential_id"),
+                )
+            )
+        )
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/nodes/<string:node_id>/run")
+class RagPipelineDraftDatasourceNodeRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run rag pipeline datasource
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            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()
+        return helper.compact_generate_response(
+            PipelineGenerator.convert_to_event_stream(
+                rag_pipeline_service.run_datasource_workflow_node(
+                    pipeline=pipeline,
+                    node_id=node_id,
+                    user_inputs=inputs,
+                    account=current_user,
+                    datasource_type=datasource_type,
+                    is_published=False,
+                    credential_id=args.get("credential_id"),
+                )
+            )
+        )
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/run")
+class RagPipelineDraftNodeRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_node_execution_fields)
+    def post(self, pipeline: Pipeline, node_id: str):
+        """
+        Run draft workflow node
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        inputs = args.get("inputs")
+        if inputs == None:
+            raise ValueError("missing inputs")
+
+        rag_pipeline_service = RagPipelineService()
+        workflow_node_execution = rag_pipeline_service.run_draft_workflow_node(
+            pipeline=pipeline, node_id=node_id, user_inputs=inputs, account=current_user
+        )
+
+        if workflow_node_execution is None:
+            raise ValueError("Workflow node execution not found")
+
+        return workflow_node_execution
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/tasks/<string:task_id>/stop")
+class RagPipelineTaskStopApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline, task_id: str):
+        """
+        Stop workflow task
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+
+        return {"result": "success"}
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/publish")
+class PublishedRagPipelineApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_fields)
+    def get(self, pipeline: Pipeline):
+        """
+        Get published pipeline
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        if not pipeline.is_published:
+            return None
+        # fetch published workflow by pipeline
+        rag_pipeline_service = RagPipelineService()
+        workflow = rag_pipeline_service.get_published_workflow(pipeline=pipeline)
+
+        # return workflow, if not found, return None
+        return workflow
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def post(self, pipeline: Pipeline):
+        """
+        Publish workflow
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        rag_pipeline_service = RagPipelineService()
+        with Session(db.engine) as session:
+            pipeline = session.merge(pipeline)
+            workflow = rag_pipeline_service.publish_workflow(
+                session=session,
+                pipeline=pipeline,
+                account=current_user,
+            )
+            pipeline.is_published = True
+            pipeline.workflow_id = workflow.id
+            session.add(pipeline)
+            workflow_created_at = TimestampField().format(workflow.created_at)
+
+            session.commit()
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/default-workflow-block-configs")
+class DefaultRagPipelineBlockConfigsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline):
+        """
+        Get default block config
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        # Get default block configs
+        rag_pipeline_service = RagPipelineService()
+        return rag_pipeline_service.get_default_block_configs()
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/default-workflow-block-configs/<string:block_type>")
+class DefaultRagPipelineBlockConfigApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline, block_type: str):
+        """
+        Get default block config
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("q", type=str, location="args")
+        args = parser.parse_args()
+
+        q = args.get("q")
+
+        filters = None
+        if q:
+            try:
+                filters = json.loads(args.get("q", ""))
+            except json.JSONDecodeError:
+                raise ValueError("Invalid filters")
+
+        # Get default block configs
+        rag_pipeline_service = RagPipelineService()
+        return rag_pipeline_service.get_default_block_config(node_type=block_type, filters=filters)
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows")
+class PublishedAllRagPipelineApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_pagination_fields)
+    def get(self, pipeline: Pipeline):
+        """
+        Get published workflows
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args")
+        parser.add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
+        parser.add_argument("user_id", type=str, required=False, location="args")
+        parser.add_argument("named_only", type=inputs.boolean, required=False, default=False, location="args")
+        args = parser.parse_args()
+        page = int(args.get("page", 1))
+        limit = int(args.get("limit", 10))
+        user_id = args.get("user_id")
+        named_only = args.get("named_only", False)
+
+        if user_id:
+            if user_id != current_user.id:
+                raise Forbidden()
+            user_id = cast(str, user_id)
+
+        rag_pipeline_service = RagPipelineService()
+        with Session(db.engine) as session:
+            workflows, has_more = rag_pipeline_service.get_all_published_workflow(
+                session=session,
+                pipeline=pipeline,
+                page=page,
+                limit=limit,
+                user_id=user_id,
+                named_only=named_only,
+            )
+
+            return {
+                "items": workflows,
+                "page": page,
+                "limit": limit,
+                "has_more": has_more,
+            }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>")
+class RagPipelineByIdApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_fields)
+    def patch(self, pipeline: Pipeline, workflow_id: str):
+        """
+        Update workflow attributes
+        """
+        # Check permission
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("marked_name", type=str, required=False, location="json")
+        parser.add_argument("marked_comment", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        # Validate name and comment length
+        if args.marked_name and len(args.marked_name) > 20:
+            raise ValueError("Marked name cannot exceed 20 characters")
+        if args.marked_comment and len(args.marked_comment) > 100:
+            raise ValueError("Marked comment cannot exceed 100 characters")
+        args = parser.parse_args()
+
+        # Prepare update data
+        update_data = {}
+        if args.get("marked_name") is not None:
+            update_data["marked_name"] = args["marked_name"]
+        if args.get("marked_comment") is not None:
+            update_data["marked_comment"] = args["marked_comment"]
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        rag_pipeline_service = RagPipelineService()
+
+        # Create a session and manage the transaction
+        with Session(db.engine, expire_on_commit=False) as session:
+            workflow = rag_pipeline_service.update_workflow(
+                session=session,
+                workflow_id=workflow_id,
+                tenant_id=pipeline.tenant_id,
+                account_id=current_user.id,
+                data=update_data,
+            )
+
+            if not workflow:
+                raise NotFound("Workflow not found")
+
+            # Commit the transaction in the controller
+            session.commit()
+
+        return workflow
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/processing/parameters")
+class PublishedRagPipelineSecondStepApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline):
+        """
+        Get second step parameters of rag pipeline
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="args")
+        args = parser.parse_args()
+        node_id = args.get("node_id")
+        if not node_id:
+            raise ValueError("Node ID is required")
+        rag_pipeline_service = RagPipelineService()
+        variables = rag_pipeline_service.get_second_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=False)
+        return {
+            "variables": variables,
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/pre-processing/parameters")
+class PublishedRagPipelineFirstStepApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline):
+        """
+        Get first step parameters of rag pipeline
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="args")
+        args = parser.parse_args()
+        node_id = args.get("node_id")
+        if not node_id:
+            raise ValueError("Node ID is required")
+        rag_pipeline_service = RagPipelineService()
+        variables = rag_pipeline_service.get_first_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=False)
+        return {
+            "variables": variables,
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/pre-processing/parameters")
+class DraftRagPipelineFirstStepApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline):
+        """
+        Get first step parameters of rag pipeline
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="args")
+        args = parser.parse_args()
+        node_id = args.get("node_id")
+        if not node_id:
+            raise ValueError("Node ID is required")
+        rag_pipeline_service = RagPipelineService()
+        variables = rag_pipeline_service.get_first_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=True)
+        return {
+            "variables": variables,
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/processing/parameters")
+class DraftRagPipelineSecondStepApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    def get(self, pipeline: Pipeline):
+        """
+        Get second step parameters of rag pipeline
+        """
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="args")
+        args = parser.parse_args()
+        node_id = args.get("node_id")
+        if not node_id:
+            raise ValueError("Node ID is required")
+
+        rag_pipeline_service = RagPipelineService()
+        variables = rag_pipeline_service.get_second_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=True)
+        return {
+            "variables": variables,
+        }
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs")
+class RagPipelineWorkflowRunListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_pagination_fields)
+    def get(self, pipeline: Pipeline):
+        """
+        Get workflow run list
+        """
+        parser = reqparse.RequestParser()
+        parser.add_argument("last_id", type=uuid_value, location="args")
+        parser.add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
+        args = parser.parse_args()
+
+        rag_pipeline_service = RagPipelineService()
+        result = rag_pipeline_service.get_rag_pipeline_paginate_workflow_runs(pipeline=pipeline, args=args)
+
+        return result
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>")
+class RagPipelineWorkflowRunDetailApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_detail_fields)
+    def get(self, pipeline: Pipeline, run_id):
+        """
+        Get workflow run detail
+        """
+        run_id = str(run_id)
+
+        rag_pipeline_service = RagPipelineService()
+        workflow_run = rag_pipeline_service.get_rag_pipeline_workflow_run(pipeline=pipeline, run_id=run_id)
+
+        return workflow_run
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>/node-executions")
+class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_node_execution_list_fields)
+    def get(self, pipeline: Pipeline, run_id):
+        """
+        Get workflow run node execution list
+        """
+        run_id = str(run_id)
+
+        rag_pipeline_service = RagPipelineService()
+        user = cast("Account | EndUser", current_user)
+        node_executions = rag_pipeline_service.get_rag_pipeline_workflow_run_node_executions(
+            pipeline=pipeline,
+            run_id=run_id,
+            user=user,
+        )
+
+        return {"data": node_executions}
+
+
+@console_ns.route("/rag/pipelines/datasource-plugins")
+class DatasourceListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        user = current_user
+        if not isinstance(user, Account):
+            raise Forbidden()
+        tenant_id = user.current_tenant_id
+        if not tenant_id:
+            raise Forbidden()
+
+        return jsonable_encoder(RagPipelineManageService.list_rag_pipeline_datasources(tenant_id))
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/last-run")
+class RagPipelineWorkflowLastRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_node_execution_fields)
+    def get(self, pipeline: Pipeline, node_id: str):
+        rag_pipeline_service = RagPipelineService()
+        workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
+        if not workflow:
+            raise NotFound("Workflow not found")
+        node_exec = rag_pipeline_service.get_node_last_run(
+            pipeline=pipeline,
+            workflow=workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("last run not found")
+        return node_exec
+
+
+@console_ns.route("/rag/pipelines/transform/datasets/<uuid:dataset_id>")
+class RagPipelineTransformApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, dataset_id):
+        if not isinstance(current_user, Account):
+            raise Forbidden()
+
+        if not (current_user.has_edit_permission or current_user.is_dataset_operator):
+            raise Forbidden()
+
+        dataset_id = str(dataset_id)
+        rag_pipeline_transform_service = RagPipelineTransformService()
+        result = rag_pipeline_transform_service.transform_dataset(dataset_id)
+        return result
+
+
+@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/variables-inspect")
+class RagPipelineDatasourceVariableApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_rag_pipeline
+    @marshal_with(workflow_run_node_execution_fields)
+    def post(self, pipeline: Pipeline):
+        """
+        Set datasource variables
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("datasource_type", type=str, required=True, location="json")
+        parser.add_argument("datasource_info", type=dict, required=True, location="json")
+        parser.add_argument("start_node_id", type=str, required=True, location="json")
+        parser.add_argument("start_node_title", type=str, required=True, location="json")
+        args = parser.parse_args()
+
+        rag_pipeline_service = RagPipelineService()
+        workflow_node_execution = rag_pipeline_service.set_datasource_variables(
+            pipeline=pipeline,
+            args=args,
+            current_user=current_user,
+        )
+        return workflow_node_execution
+
+
+@console_ns.route("/rag/pipelines/recommended-plugins")
+class RagPipelineRecommendedPluginApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        rag_pipeline_service = RagPipelineService()
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins()
+        return recommended_plugins

api/controllers/console/datasets/wraps.py

@@ -0,0 +1,46 @@
+from collections.abc import Callable
+from functools import wraps
+
+from controllers.console.datasets.error import PipelineNotFoundError
+from extensions.ext_database import db
+from libs.login import current_user
+from models.account import Account
+from models.dataset import Pipeline
+
+
+def get_rag_pipeline(
+    view: Callable | None = 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")
+
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user is not an account")
+
+            pipeline_id = kwargs.get("pipeline_id")
+            pipeline_id = str(pipeline_id)
+
+            del kwargs["pipeline_id"]
+
+            pipeline = (
+                db.session.query(Pipeline)
+                .where(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/explore/audio.py

@@ -26,9 +26,15 @@ from services.errors.audio import (
     UnsupportedAudioTypeServiceError,
 )
 
+from .. import console_ns
+
 logger = logging.getLogger(__name__)
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/audio-to-text",
+    endpoint="installed_app_audio",
+)
 class ChatAudioApi(InstalledAppResource):
     def post(self, installed_app):
         app_model = installed_app.app
@@ -65,6 +71,10 @@ class ChatAudioApi(InstalledAppResource):
             raise InternalServerError()
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/text-to-audio",
+    endpoint="installed_app_text",
+)
 class ChatTextApi(InstalledAppResource):
     def post(self, installed_app):
         from flask_restx import reqparse

api/controllers/console/explore/completion.py

@@ -33,10 +33,16 @@ from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
 
+from .. import console_ns
+
 logger = logging.getLogger(__name__)
 
 
 # define completion api for user
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/completion-messages",
+    endpoint="installed_app_completion",
+)
 class CompletionApi(InstalledAppResource):
     def post(self, installed_app):
         app_model = installed_app.app
@@ -87,6 +93,10 @@ class CompletionApi(InstalledAppResource):
             raise InternalServerError()
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/completion-messages/<string:task_id>/stop",
+    endpoint="installed_app_stop_completion",
+)
 class CompletionStopApi(InstalledAppResource):
     def post(self, installed_app, task_id):
         app_model = installed_app.app
@@ -100,6 +110,10 @@ class CompletionStopApi(InstalledAppResource):
         return {"result": "success"}, 200
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/chat-messages",
+    endpoint="installed_app_chat_completion",
+)
 class ChatApi(InstalledAppResource):
     def post(self, installed_app):
         app_model = installed_app.app
@@ -153,6 +167,10 @@ class ChatApi(InstalledAppResource):
             raise InternalServerError()
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/chat-messages/<string:task_id>/stop",
+    endpoint="installed_app_stop_chat_completion",
+)
 class ChatStopApi(InstalledAppResource):
     def post(self, installed_app, task_id):
         app_model = installed_app.app

api/controllers/console/explore/conversation.py

@@ -16,7 +16,13 @@ from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError
 from services.web_conversation_service import WebConversationService
 
+from .. import console_ns
 
+
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/conversations",
+    endpoint="installed_app_conversations",
+)
 class ConversationListApi(InstalledAppResource):
     @marshal_with(conversation_infinite_scroll_pagination_fields)
     def get(self, installed_app):
@@ -52,6 +58,10 @@ class ConversationListApi(InstalledAppResource):
             raise NotFound("Last Conversation Not Exists.")
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>",
+    endpoint="installed_app_conversation",
+)
 class ConversationApi(InstalledAppResource):
     def delete(self, installed_app, c_id):
         app_model = installed_app.app
@@ -70,6 +80,10 @@ class ConversationApi(InstalledAppResource):
         return {"result": "success"}, 204
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/name",
+    endpoint="installed_app_conversation_rename",
+)
 class ConversationRenameApi(InstalledAppResource):
     @marshal_with(simple_conversation_fields)
     def post(self, installed_app, c_id):
@@ -95,6 +109,10 @@ class ConversationRenameApi(InstalledAppResource):
             raise NotFound("Conversation Not Exists.")
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/pin",
+    endpoint="installed_app_conversation_pin",
+)
 class ConversationPinApi(InstalledAppResource):
     def patch(self, installed_app, c_id):
         app_model = installed_app.app
@@ -114,6 +132,10 @@ class ConversationPinApi(InstalledAppResource):
         return {"result": "success"}
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/conversations/<uuid:c_id>/unpin",
+    endpoint="installed_app_conversation_unpin",
+)
 class ConversationUnPinApi(InstalledAppResource):
     def patch(self, installed_app, c_id):
         app_model = installed_app.app

api/controllers/console/explore/message.py

@@ -36,9 +36,15 @@ from services.errors.message import (
 )
 from services.message_service import MessageService
 
+from .. import console_ns
+
 logger = logging.getLogger(__name__)
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/messages",
+    endpoint="installed_app_messages",
+)
 class MessageListApi(InstalledAppResource):
     @marshal_with(message_infinite_scroll_pagination_fields)
     def get(self, installed_app):
@@ -66,6 +72,10 @@ class MessageListApi(InstalledAppResource):
             raise NotFound("First Message Not Exists.")
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/feedbacks",
+    endpoint="installed_app_message_feedback",
+)
 class MessageFeedbackApi(InstalledAppResource):
     def post(self, installed_app, message_id):
         app_model = installed_app.app
@@ -93,6 +103,10 @@ class MessageFeedbackApi(InstalledAppResource):
         return {"result": "success"}
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/more-like-this",
+    endpoint="installed_app_more_like_this",
+)
 class MessageMoreLikeThisApi(InstalledAppResource):
     def get(self, installed_app, message_id):
         app_model = installed_app.app
@@ -139,6 +153,10 @@ class MessageMoreLikeThisApi(InstalledAppResource):
             raise InternalServerError()
 
 
+@console_ns.route(
+    "/installed-apps/<uuid:installed_app_id>/messages/<uuid:message_id>/suggested-questions",
+    endpoint="installed_app_suggested_question",
+)
 class MessageSuggestedQuestionApi(InstalledAppResource):
     def get(self, installed_app, message_id):
         app_model = installed_app.app

api/controllers/console/explore/workflow.py

@@ -27,9 +27,12 @@ from models.model import AppMode, InstalledApp
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
 
+from .. import console_ns
+
 logger = logging.getLogger(__name__)
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>/workflows/run")
 class InstalledAppWorkflowRunApi(InstalledAppResource):
     def post(self, installed_app: InstalledApp):
         """
@@ -70,6 +73,7 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
             raise InternalServerError()
 
 
+@console_ns.route("/installed-apps/<uuid:installed_app_id>/workflows/tasks/<string:task_id>/stop")
 class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
     def post(self, installed_app: InstalledApp, task_id: str):
         """

api/controllers/console/files.py

@@ -20,14 +20,18 @@ from controllers.console.wraps import (
     cloud_edition_billing_resource_check,
     setup_required,
 )
+from extensions.ext_database import db
 from fields.file_fields import file_fields, upload_config_fields
 from libs.login import login_required
 from models import Account
 from services.file_service import FileService
 
+from . import console_ns
+
 PREVIEW_WORDS_LIMIT = 3000
 
 
+@console_ns.route("/files/upload")
 class FileApi(Resource):
     @setup_required
     @login_required
@@ -68,10 +72,11 @@ class FileApi(Resource):
         if source not in ("datasets", None):
             source = None
 
+        if not isinstance(current_user, Account):
+            raise ValueError("Invalid user account")
+
         try:
-            if not isinstance(current_user, Account):
-                raise ValueError("Invalid user account")
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,
@@ -86,16 +91,18 @@ class FileApi(Resource):
         return upload_file, 201
 
 
+@console_ns.route("/files/<uuid:file_id>/preview")
 class FilePreviewApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
     def get(self, file_id):
         file_id = str(file_id)
-        text = FileService.get_file_preview(file_id)
+        text = FileService(db.engine).get_file_preview(file_id)
         return {"content": text}
 
 
+@console_ns.route("/files/support-type")
 class FileSupportTypeApi(Resource):
     @setup_required
     @login_required

api/controllers/console/remote_files.py

@@ -14,11 +14,15 @@ from controllers.common.errors import (
 )
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
+from extensions.ext_database import db
 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 . import console_ns
 
+
+@console_ns.route("/remote-files/<path:url>")
 class RemoteFileInfoApi(Resource):
     @marshal_with(remote_file_info_fields)
     def get(self, url):
@@ -34,6 +38,7 @@ class RemoteFileInfoApi(Resource):
         }
 
 
+@console_ns.route("/remote-files/upload")
 class RemoteFileUploadApi(Resource):
     @marshal_with(file_fields_with_signed_url)
     def post(self):
@@ -61,7 +66,7 @@ class RemoteFileUploadApi(Resource):
 
         try:
             user = cast(Account, current_user)
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,
                 mimetype=file_info.mimetype,

api/controllers/console/spec.py

@@ -0,0 +1,34 @@
+import logging
+
+from flask_restx import Resource
+
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.schemas.schema_manager import SchemaManager
+from libs.login import login_required
+
+from . import console_ns
+
+logger = logging.getLogger(__name__)
+
+
+@console_ns.route("/spec/schema-definitions")
+class SpecSchemaDefinitionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """
+        Get system JSON Schema definitions specification
+        Used for frontend component type mapping
+        """
+        try:
+            schema_manager = SchemaManager()
+            schema_definitions = schema_manager.get_all_schema_definitions()
+            return schema_definitions, 200
+        except Exception:
+            logger.exception("Failed to get schema definitions from local registry")
+            # Return empty array as fallback
+            return [], 200

api/controllers/console/tag/tags.py

@@ -3,7 +3,7 @@ from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
-from controllers.console import api
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.tag_fields import dataset_tag_fields
 from libs.login import login_required
@@ -17,6 +17,7 @@ def _validate_name(name):
     return name
 
 
+@console_ns.route("/tags")
 class TagListApi(Resource):
     @setup_required
     @login_required
@@ -52,6 +53,7 @@ class TagListApi(Resource):
         return response, 200
 
 
+@console_ns.route("/tags/<uuid:tag_id>")
 class TagUpdateDeleteApi(Resource):
     @setup_required
     @login_required
@@ -89,6 +91,7 @@ class TagUpdateDeleteApi(Resource):
         return 204
 
 
+@console_ns.route("/tag-bindings/create")
 class TagBindingCreateApi(Resource):
     @setup_required
     @login_required
@@ -114,6 +117,7 @@ class TagBindingCreateApi(Resource):
         return {"result": "success"}, 200
 
 
+@console_ns.route("/tag-bindings/remove")
 class TagBindingDeleteApi(Resource):
     @setup_required
     @login_required
@@ -133,9 +137,3 @@ class TagBindingDeleteApi(Resource):
         TagService.delete_tag_binding(args)
 
         return {"result": "success"}, 200
-
-
-api.add_resource(TagListApi, "/tags")
-api.add_resource(TagUpdateDeleteApi, "/tags/<uuid:tag_id>")
-api.add_resource(TagBindingCreateApi, "/tag-bindings/create")
-api.add_resource(TagBindingDeleteApi, "/tag-bindings/remove")

api/controllers/console/version.py

@@ -1,7 +1,7 @@
 import json
 import logging
 
-import requests
+import httpx
 from flask_restx import Resource, fields, reqparse
 from packaging import version
 
@@ -57,7 +57,11 @@ class VersionApi(Resource):
             return result
 
         try:
-            response = requests.get(check_update_url, {"current_version": args["current_version"]}, timeout=(3, 10))
+            response = httpx.get(
+                check_update_url,
+                params={"current_version": args["current_version"]},
+                timeout=httpx.Timeout(connect=3, read=10),
+            )
         except Exception as error:
             logger.warning("Check update version error: %s.", str(error))
             result["version"] = args["current_version"]

api/controllers/console/workspace/workspace.py

@@ -227,7 +227,7 @@ class WebappLogoWorkspaceApi(Resource):
             raise UnsupportedFileTypeError()
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/console/wraps.py

@@ -279,3 +279,14 @@ def is_allow_transfer_owner(view: Callable[P, R]):
         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/image_preview.py

@@ -7,6 +7,7 @@ from werkzeug.exceptions import NotFound
 import services
 from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import files_ns
+from extensions.ext_database import db
 from services.account_service import TenantService
 from services.file_service import FileService
 
@@ -28,7 +29,7 @@ class ImagePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, mimetype = FileService.get_image_preview(
+            generator, mimetype = FileService(db.engine).get_image_preview(
                 file_id=file_id,
                 timestamp=timestamp,
                 nonce=nonce,
@@ -57,7 +58,7 @@ class FilePreviewApi(Resource):
             return {"content": "Invalid request."}, 400
 
         try:
-            generator, upload_file = FileService.get_file_generator_by_file_id(
+            generator, upload_file = FileService(db.engine).get_file_generator_by_file_id(
                 file_id=file_id,
                 timestamp=args["timestamp"],
                 nonce=args["nonce"],
@@ -108,7 +109,7 @@ class WorkspaceWebappLogoApi(Resource):
             raise NotFound("webapp logo is not found")
 
         try:
-            generator, mimetype = FileService.get_public_image_preview(
+            generator, mimetype = FileService(db.engine).get_public_image_preview(
                 webapp_logo_file_id,
             )
         except services.errors.file.UnsupportedFileTypeError:

api/controllers/inner_api/plugin/plugin.py

@@ -420,7 +420,12 @@ class PluginUploadFileRequestApi(Resource):
     )
     def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestRequestUploadFile):
         # generate signed url
-        url = get_signed_file_url_for_plugin(payload.filename, payload.mimetype, tenant_model.id, user_model.id)
+        url = get_signed_file_url_for_plugin(
+            filename=payload.filename,
+            mimetype=payload.mimetype,
+            tenant_id=tenant_model.id,
+            user_id=user_model.id,
+        )
         return BaseBackwardsInvocationResponse(data={"url": url}).model_dump()
 
 

api/controllers/inner_api/plugin/wraps.py

@@ -24,24 +24,37 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     NOTE: user_id is not trusted, it could be maliciously set to any value.
     As a result, it could only be considered as an end user id.
     """
+    if not user_id:
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
     try:
         with Session(db.engine) as session:
-            if not user_id:
-                user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+            user_model = None
 
-            user_model = (
-                session.query(EndUser)
-                .where(
-                    EndUser.session_id == user_id,
-                    EndUser.tenant_id == tenant_id,
+            if is_anonymous:
+                user_model = (
+                    session.query(EndUser)
+                    .where(
+                        EndUser.session_id == user_id,
+                        EndUser.tenant_id == tenant_id,
+                    )
+                    .first()
                 )
-                .first()
-            )
+            else:
+                user_model = (
+                    session.query(EndUser)
+                    .where(
+                        EndUser.id == user_id,
+                        EndUser.tenant_id == tenant_id,
+                    )
+                    .first()
+                )
+
             if not user_model:
                 user_model = EndUser(
                     tenant_id=tenant_id,
                     type="service_api",
-                    is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value,
+                    is_anonymous=is_anonymous,
                     session_id=user_id,
                 )
                 session.add(user_model)

api/controllers/service_api/app/file.py

@@ -12,8 +12,9 @@ from controllers.common.errors import (
 )
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from extensions.ext_database import db
 from fields.file_fields import build_file_model
-from models.model import App, EndUser
+from models import App, EndUser
 from services.file_service import FileService
 
 
@@ -52,7 +53,7 @@ class FileApi(Resource):
             raise FilenameNotExistsError
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/service_api/dataset/dataset.py

@@ -1,10 +1,10 @@
-from typing import Literal
+from typing import Any, Literal, cast
 
 from flask import request
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
 
-import services.dataset_service
+import services
 from controllers.service_api import service_api_ns
 from controllers.service_api.dataset.error import DatasetInUseError, DatasetNameDuplicateError, InvalidActionError
 from controllers.service_api.wraps import (
@@ -17,6 +17,7 @@ from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import build_dataset_tag_fields
 from libs.login import current_user
+from libs.validators import validate_description_length
 from models.account import Account
 from models.dataset import Dataset, DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
@@ -31,12 +32,6 @@ def _validate_name(name):
     return name
 
 
-def _validate_description_length(description):
-    if description and len(description) > 400:
-        raise ValueError("Description cannot exceed 400 characters.")
-    return description
-
-
 # Define parsers for dataset operations
 dataset_create_parser = reqparse.RequestParser()
 dataset_create_parser.add_argument(
@@ -48,7 +43,7 @@ dataset_create_parser.add_argument(
 )
 dataset_create_parser.add_argument(
     "description",
-    type=_validate_description_length,
+    type=validate_description_length,
     nullable=True,
     required=False,
     default="",
@@ -101,7 +96,7 @@ dataset_update_parser.add_argument(
     type=_validate_name,
 )
 dataset_update_parser.add_argument(
-    "description", location="json", store_missing=False, type=_validate_description_length
+    "description", location="json", store_missing=False, type=validate_description_length
 )
 dataset_update_parser.add_argument(
     "indexing_technique",
@@ -254,19 +249,21 @@ class DatasetListApi(DatasetApiResource):
         """Resource for creating datasets."""
         args = dataset_create_parser.parse_args()
 
-        if args.get("embedding_model_provider"):
-            DatasetService.check_embedding_model_setting(
-                tenant_id, args.get("embedding_model_provider"), args.get("embedding_model")
-            )
+        embedding_model_provider = args.get("embedding_model_provider")
+        embedding_model = args.get("embedding_model")
+        if embedding_model_provider and embedding_model:
+            DatasetService.check_embedding_model_setting(tenant_id, embedding_model_provider, embedding_model)
+
+        retrieval_model = args.get("retrieval_model")
         if (
-            args.get("retrieval_model")
-            and args.get("retrieval_model").get("reranking_model")
-            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+            retrieval_model
+            and retrieval_model.get("reranking_model")
+            and retrieval_model.get("reranking_model").get("reranking_provider_name")
         ):
             DatasetService.check_reranking_model_setting(
                 tenant_id,
-                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
-                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+                retrieval_model.get("reranking_model").get("reranking_provider_name"),
+                retrieval_model.get("reranking_model").get("reranking_model_name"),
             )
 
         try:
@@ -317,7 +314,7 @@ class DatasetApi(DatasetApiResource):
             DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        data = marshal(dataset, dataset_detail_fields)
+        data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         # check embedding setting
         provider_manager = ProviderManager()
         assert isinstance(current_user, Account)
@@ -331,8 +328,8 @@ class DatasetApi(DatasetApiResource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        if data["indexing_technique"] == "high_quality":
-            item_model = f"{data['embedding_model']}:{data['embedding_model_provider']}"
+        if data.get("indexing_technique") == "high_quality":
+            item_model = f"{data.get('embedding_model')}:{data.get('embedding_model_provider')}"
             if item_model in model_names:
                 data["embedding_available"] = True
             else:
@@ -341,7 +338,9 @@ class DatasetApi(DatasetApiResource):
             data["embedding_available"] = True
 
             # force update search method to keyword_search if indexing_technique is economic
-            data["retrieval_model_dict"]["search_method"] = "keyword_search"
+            retrieval_model_dict = data.get("retrieval_model_dict")
+            if retrieval_model_dict:
+                retrieval_model_dict["search_method"] = "keyword_search"
 
         if data.get("permission") == "partial_members":
             part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
@@ -372,19 +371,24 @@ class DatasetApi(DatasetApiResource):
         data = request.get_json()
 
         # check embedding model setting
-        if data.get("indexing_technique") == "high_quality" or data.get("embedding_model_provider"):
-            DatasetService.check_embedding_model_setting(
-                dataset.tenant_id, data.get("embedding_model_provider"), data.get("embedding_model")
-            )
+        embedding_model_provider = data.get("embedding_model_provider")
+        embedding_model = data.get("embedding_model")
+        if data.get("indexing_technique") == "high_quality" or embedding_model_provider:
+            if embedding_model_provider and embedding_model:
+                DatasetService.check_embedding_model_setting(
+                    dataset.tenant_id, embedding_model_provider, embedding_model
+                )
+
+        retrieval_model = data.get("retrieval_model")
         if (
-            data.get("retrieval_model")
-            and data.get("retrieval_model").get("reranking_model")
-            and data.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+            retrieval_model
+            and retrieval_model.get("reranking_model")
+            and retrieval_model.get("reranking_model").get("reranking_provider_name")
         ):
             DatasetService.check_reranking_model_setting(
                 dataset.tenant_id,
-                data.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
-                data.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+                retrieval_model.get("reranking_model").get("reranking_provider_name"),
+                retrieval_model.get("reranking_model").get("reranking_model_name"),
             )
 
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
@@ -397,7 +401,7 @@ class DatasetApi(DatasetApiResource):
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        result_data = marshal(dataset, dataset_detail_fields)
+        result_data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
         assert isinstance(current_user, Account)
         tenant_id = current_user.current_tenant_id
 
@@ -591,9 +595,10 @@ class DatasetTagsApi(DatasetApiResource):
 
         args = tag_update_parser.parse_args()
         args["type"] = "knowledge"
-        tag = TagService.update_tags(args, args.get("tag_id"))
+        tag_id = args["tag_id"]
+        tag = TagService.update_tags(args, tag_id)
 
-        binding_count = TagService.get_tag_binding_count(args.get("tag_id"))
+        binding_count = TagService.get_tag_binding_count(tag_id)
 
         response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
 
@@ -616,7 +621,7 @@ class DatasetTagsApi(DatasetApiResource):
         if not current_user.has_edit_permission:
             raise Forbidden()
         args = tag_delete_parser.parse_args()
-        TagService.delete_tag(args.get("tag_id"))
+        TagService.delete_tag(args["tag_id"])
 
         return 204
 

api/controllers/service_api/dataset/document.py

@@ -30,7 +30,6 @@ from extensions.ext_database import db
 from fields.document_fields import document_fields, document_status_fields
 from libs.login import current_user
 from models.dataset import Dataset, Document, DocumentSegment
-from models.model import EndUser
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 from services.file_service import FileService
@@ -109,22 +108,29 @@ class DocumentAddByTextApi(DatasetApiResource):
         if text is None or name is None:
             raise ValueError("Both 'text' and 'name' must be non-null values.")
 
-        if args.get("embedding_model_provider"):
-            DatasetService.check_embedding_model_setting(
-                tenant_id, args.get("embedding_model_provider"), args.get("embedding_model")
-            )
+        embedding_model_provider = args.get("embedding_model_provider")
+        embedding_model = args.get("embedding_model")
+        if embedding_model_provider and embedding_model:
+            DatasetService.check_embedding_model_setting(tenant_id, embedding_model_provider, embedding_model)
+
+        retrieval_model = args.get("retrieval_model")
         if (
-            args.get("retrieval_model")
-            and args.get("retrieval_model").get("reranking_model")
-            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+            retrieval_model
+            and retrieval_model.get("reranking_model")
+            and retrieval_model.get("reranking_model").get("reranking_provider_name")
         ):
             DatasetService.check_reranking_model_setting(
                 tenant_id,
-                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
-                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+                retrieval_model.get("reranking_model").get("reranking_provider_name"),
+                retrieval_model.get("reranking_model").get("reranking_model_name"),
             )
 
-        upload_file = FileService.upload_text(text=str(text), text_name=str(name))
+        if not current_user:
+            raise ValueError("current_user is required")
+
+        upload_file = FileService(db.engine).upload_text(
+            text=str(text), text_name=str(name), user_id=current_user.id, tenant_id=tenant_id
+        )
         data_source = {
             "type": "upload_file",
             "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -134,6 +140,9 @@ class DocumentAddByTextApi(DatasetApiResource):
         # validate args
         DocumentService.document_create_args_validate(knowledge_config)
 
+        if not current_user:
+            raise ValueError("current_user is required")
+
         try:
             documents, batch = DocumentService.save_document_with_dataset_id(
                 dataset=dataset,
@@ -180,15 +189,16 @@ class DocumentUpdateByTextApi(DatasetApiResource):
         if not dataset:
             raise ValueError("Dataset does not exist.")
 
+        retrieval_model = args.get("retrieval_model")
         if (
-            args.get("retrieval_model")
-            and args.get("retrieval_model").get("reranking_model")
-            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+            retrieval_model
+            and retrieval_model.get("reranking_model")
+            and retrieval_model.get("reranking_model").get("reranking_provider_name")
         ):
             DatasetService.check_reranking_model_setting(
                 tenant_id,
-                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
-                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+                retrieval_model.get("reranking_model").get("reranking_provider_name"),
+                retrieval_model.get("reranking_model").get("reranking_model_name"),
             )
 
         # indexing_technique is already set in dataset since this is an update
@@ -199,7 +209,11 @@ class DocumentUpdateByTextApi(DatasetApiResource):
             name = args.get("name")
             if text is None or name is None:
                 raise ValueError("Both text and name must be strings.")
-            upload_file = FileService.upload_text(text=str(text), text_name=str(name))
+            if not current_user:
+                raise ValueError("current_user is required")
+            upload_file = FileService(db.engine).upload_text(
+                text=str(text), text_name=str(name), user_id=current_user.id, tenant_id=tenant_id
+            )
             data_source = {
                 "type": "upload_file",
                 "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -299,10 +313,9 @@ class DocumentAddByFileApi(DatasetApiResource):
         if not file.filename:
             raise FilenameNotExistsError
 
-        if not isinstance(current_user, EndUser):
-            raise ValueError("Invalid user account")
-
-        upload_file = FileService.upload_file(
+        if not current_user:
+            raise ValueError("current_user is required")
+        upload_file = FileService(db.engine).upload_file(
             filename=file.filename,
             content=file.read(),
             mimetype=file.mimetype,
@@ -390,10 +403,11 @@ class DocumentUpdateByFileApi(DatasetApiResource):
             if not file.filename:
                 raise FilenameNotExistsError
 
+            if not current_user:
+                raise ValueError("current_user is required")
+
             try:
-                if not isinstance(current_user, EndUser):
-                    raise ValueError("Invalid user account")
-                upload_file = FileService.upload_file(
+                upload_file = FileService(db.engine).upload_file(
                     filename=file.filename,
                     content=file.read(),
                     mimetype=file.mimetype,
@@ -571,7 +585,7 @@ class DocumentApi(DatasetApiResource):
             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,
@@ -604,7 +618,7 @@ class DocumentApi(DatasetApiResource):
             }
         else:
             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,

api/controllers/service_api/dataset/error.py

@@ -47,3 +47,9 @@ class DatasetInUseError(BaseHTTPException):
     error_code = "dataset_in_use"
     description = "The dataset is being used by some apps. Please remove the dataset from the apps before deleting it."
     code = 409
+
+
+class PipelineRunError(BaseHTTPException):
+    error_code = "pipeline_run_error"
+    description = "An error occurred while running the pipeline."
+    code = 500

api/controllers/service_api/dataset/metadata.py

@@ -106,7 +106,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
             raise NotFound("Dataset not found.")
         DatasetService.check_dataset_permission(dataset, current_user)
 
-        metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, args.get("name"))
+        metadata = MetadataService.update_metadata_name(dataset_id_str, metadata_id_str, args["name"])
         return marshal(metadata, dataset_metadata_fields), 200
 
     @service_api_ns.doc("delete_dataset_metadata")
@@ -133,7 +133,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
         return 204
 
 
-@service_api_ns.route("/datasets/metadata/built-in")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in")
 class DatasetMetadataBuiltInFieldServiceApi(DatasetApiResource):
     @service_api_ns.doc("get_built_in_fields")
     @service_api_ns.doc(description="Get all built-in metadata fields")
@@ -143,7 +143,7 @@ class DatasetMetadataBuiltInFieldServiceApi(DatasetApiResource):
             401: "Unauthorized - invalid API token",
         }
     )
-    def get(self, tenant_id):
+    def get(self, tenant_id, dataset_id):
         """Get all built-in metadata fields."""
         built_in_fields = MetadataService.get_built_in_fields()
         return {"fields": built_in_fields}, 200

api/controllers/service_api/dataset/rag_pipeline/rag_pipeline_workflow.py

@@ -0,0 +1,242 @@
+import string
+import uuid
+from collections.abc import Generator
+from typing import Any
+
+from flask import request
+from flask_restx import reqparse
+from flask_restx.reqparse import ParseResult, RequestParser
+from werkzeug.exceptions import Forbidden
+
+import services
+from controllers.common.errors import FilenameNotExistsError, NoFileUploadedError, TooManyFilesError
+from controllers.service_api import service_api_ns
+from controllers.service_api.dataset.error import PipelineRunError
+from controllers.service_api.wraps import DatasetApiResource
+from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
+from core.app.entities.app_invoke_entities import InvokeFrom
+from libs import helper
+from libs.login import current_user
+from models.account import Account
+from models.dataset import Pipeline
+from models.engine import db
+from services.errors.file import FileTooLargeError, UnsupportedFileTypeError
+from services.file_service import FileService
+from services.rag_pipeline.entity.pipeline_service_api_entities import DatasourceNodeRunApiEntity
+from services.rag_pipeline.pipeline_generate_service import PipelineGenerateService
+from services.rag_pipeline.rag_pipeline import RagPipelineService
+
+
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource-plugins")
+class DatasourcePluginsApi(DatasetApiResource):
+    """Resource for datasource plugins."""
+
+    @service_api_ns.doc(shortcut="list_rag_pipeline_datasource_plugins")
+    @service_api_ns.doc(description="List all datasource plugins for a rag pipeline")
+    @service_api_ns.doc(
+        path={
+            "dataset_id": "Dataset ID",
+        }
+    )
+    @service_api_ns.doc(
+        params={
+            "is_published": "Whether to get published or draft datasource plugins "
+            "(true for published, false for draft, default: true)"
+        }
+    )
+    @service_api_ns.doc(
+        responses={
+            200: "Datasource plugins retrieved successfully",
+            401: "Unauthorized - invalid API token",
+        }
+    )
+    def get(self, tenant_id: str, dataset_id: str):
+        """Resource for getting datasource plugins."""
+        # Get query parameter to determine published or draft
+        is_published: bool = request.args.get("is_published", default=True, type=bool)
+
+        rag_pipeline_service: RagPipelineService = RagPipelineService()
+        datasource_plugins: list[dict[Any, Any]] = rag_pipeline_service.get_datasource_plugins(
+            tenant_id=tenant_id, dataset_id=dataset_id, is_published=is_published
+        )
+        return datasource_plugins, 200
+
+
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource/nodes/{string:node_id}/run")
+class DatasourceNodeRunApi(DatasetApiResource):
+    """Resource for datasource node run."""
+
+    @service_api_ns.doc(shortcut="pipeline_datasource_node_run")
+    @service_api_ns.doc(description="Run a datasource node for a rag pipeline")
+    @service_api_ns.doc(
+        path={
+            "dataset_id": "Dataset ID",
+        }
+    )
+    @service_api_ns.doc(
+        body={
+            "inputs": "User input variables",
+            "datasource_type": "Datasource type, e.g. online_document",
+            "credential_id": "Credential ID",
+            "is_published": "Whether to get published or draft datasource plugins "
+            "(true for published, false for draft, default: true)",
+        }
+    )
+    @service_api_ns.doc(
+        responses={
+            200: "Datasource node run successfully",
+            401: "Unauthorized - invalid API token",
+        }
+    )
+    def post(self, tenant_id: str, dataset_id: str, node_id: str):
+        """Resource for getting datasource plugins."""
+        # Get query parameter to determine published or draft
+        parser: RequestParser = 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")
+        parser.add_argument("is_published", type=bool, required=True, location="json")
+        args: ParseResult = parser.parse_args()
+
+        datasource_node_run_api_entity: DatasourceNodeRunApiEntity = DatasourceNodeRunApiEntity(**args)
+        assert isinstance(current_user, Account)
+        rag_pipeline_service: RagPipelineService = RagPipelineService()
+        pipeline: Pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id)
+        return helper.compact_generate_response(
+            PipelineGenerator.convert_to_event_stream(
+                rag_pipeline_service.run_datasource_workflow_node(
+                    pipeline=pipeline,
+                    node_id=node_id,
+                    user_inputs=datasource_node_run_api_entity.inputs,
+                    account=current_user,
+                    datasource_type=datasource_node_run_api_entity.datasource_type,
+                    is_published=datasource_node_run_api_entity.is_published,
+                    credential_id=datasource_node_run_api_entity.credential_id,
+                )
+            )
+        )
+
+
+@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/run")
+class PipelineRunApi(DatasetApiResource):
+    """Resource for datasource node run."""
+
+    @service_api_ns.doc(shortcut="pipeline_datasource_node_run")
+    @service_api_ns.doc(description="Run a datasource node for a rag pipeline")
+    @service_api_ns.doc(
+        path={
+            "dataset_id": "Dataset ID",
+        }
+    )
+    @service_api_ns.doc(
+        body={
+            "inputs": "User input variables",
+            "datasource_type": "Datasource type, e.g. online_document",
+            "datasource_info_list": "Datasource info list",
+            "start_node_id": "Start node ID",
+            "is_published": "Whether to get published or draft datasource plugins "
+            "(true for published, false for draft, default: true)",
+            "streaming": "Whether to stream the response(streaming or blocking), default: streaming",
+        }
+    )
+    @service_api_ns.doc(
+        responses={
+            200: "Pipeline run successfully",
+            401: "Unauthorized - invalid API token",
+        }
+    )
+    def post(self, tenant_id: str, dataset_id: str):
+        """Resource for running a rag pipeline."""
+        parser: RequestParser = 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("datasource_info_list", type=list, required=True, location="json")
+        parser.add_argument("start_node_id", type=str, required=True, location="json")
+        parser.add_argument("is_published", type=bool, required=True, default=True, location="json")
+        parser.add_argument(
+            "response_mode",
+            type=str,
+            required=True,
+            choices=["streaming", "blocking"],
+            default="blocking",
+            location="json",
+        )
+        args: ParseResult = parser.parse_args()
+
+        if not isinstance(current_user, Account):
+            raise Forbidden()
+
+        rag_pipeline_service: RagPipelineService = RagPipelineService()
+        pipeline: Pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id)
+        try:
+            response: dict[Any, Any] | Generator[str, Any, None] = PipelineGenerateService.generate(
+                pipeline=pipeline,
+                user=current_user,
+                args=args,
+                invoke_from=InvokeFrom.PUBLISHED if args.get("is_published") else InvokeFrom.DEBUGGER,
+                streaming=args.get("response_mode") == "streaming",
+            )
+
+            return helper.compact_generate_response(response)
+        except Exception as ex:
+            raise PipelineRunError(description=str(ex))
+
+
+@service_api_ns.route("/datasets/pipeline/file-upload")
+class KnowledgebasePipelineFileUploadApi(DatasetApiResource):
+    """Resource for uploading a file to a knowledgebase pipeline."""
+
+    @service_api_ns.doc(shortcut="knowledgebase_pipeline_file_upload")
+    @service_api_ns.doc(description="Upload a file to a knowledgebase pipeline")
+    @service_api_ns.doc(
+        responses={
+            201: "File uploaded successfully",
+            400: "Bad request - no file or invalid file",
+            401: "Unauthorized - invalid API token",
+            413: "File too large",
+            415: "Unsupported file type",
+        }
+    )
+    def post(self, tenant_id: str):
+        """Upload a file for use in conversations.
+
+        Accepts a single file upload via multipart/form-data.
+        """
+        # check file
+        if "file" not in request.files:
+            raise NoFileUploadedError()
+
+        if len(request.files) > 1:
+            raise TooManyFilesError()
+
+        file = request.files["file"]
+        if not file.mimetype:
+            raise UnsupportedFileTypeError()
+
+        if not file.filename:
+            raise FilenameNotExistsError
+
+        if not current_user:
+            raise ValueError("Invalid user account")
+
+        try:
+            upload_file = FileService(db.engine).upload_file(
+                filename=file.filename,
+                content=file.read(),
+                mimetype=file.mimetype,
+                user=current_user,
+            )
+        except services.errors.file.FileTooLargeError as file_too_large_error:
+            raise FileTooLargeError(file_too_large_error.description)
+        except services.errors.file.UnsupportedFileTypeError:
+            raise UnsupportedFileTypeError()
+
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "mime_type": upload_file.mime_type,
+            "created_by": upload_file.created_by,
+            "created_at": upload_file.created_at,
+        }, 201

api/controllers/service_api/wraps.py

@@ -193,6 +193,47 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
     def decorator(view: Callable[Concatenate[T, P], R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
+            # get url path dataset_id from positional args or kwargs
+            # Flask passes URL path parameters as positional arguments
+            dataset_id = None
+
+            # First try to get from kwargs (explicit parameter)
+            dataset_id = kwargs.get("dataset_id")
+
+            # If not in kwargs, try to extract from positional args
+            if not dataset_id and args:
+                # For class methods: args[0] is self, args[1] is dataset_id (if exists)
+                # Check if first arg is likely a class instance (has __dict__ or __class__)
+                if len(args) > 1 and hasattr(args[0], "__dict__"):
+                    # This is a class method, dataset_id should be in args[1]
+                    potential_id = args[1]
+                    # Validate it's a string-like UUID, not another object
+                    try:
+                        # Try to convert to string and check if it's a valid UUID format
+                        str_id = str(potential_id)
+                        # Basic check: UUIDs are 36 chars with hyphens
+                        if len(str_id) == 36 and str_id.count("-") == 4:
+                            dataset_id = str_id
+                    except:
+                        pass
+                elif len(args) > 0:
+                    # Not a class method, check if args[0] looks like a UUID
+                    potential_id = args[0]
+                    try:
+                        str_id = str(potential_id)
+                        if len(str_id) == 36 and str_id.count("-") == 4:
+                            dataset_id = str_id
+                    except:
+                        pass
+
+            # Validate dataset if dataset_id is provided
+            if dataset_id:
+                dataset_id = str(dataset_id)
+                dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
+                if not dataset:
+                    raise NotFound("Dataset not found.")
+                if not dataset.enable_api:
+                    raise Forbidden("Dataset api access is not enabled.")
             api_token = validate_and_get_api_token("dataset")
             tenant_account_join = (
                 db.session.query(Tenant, TenantAccountJoin)

api/controllers/web/files.py

@@ -11,6 +11,7 @@ from controllers.common.errors import (
 )
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
+from extensions.ext_database import db
 from fields.file_fields import build_file_model
 from services.file_service import FileService
 
@@ -68,7 +69,7 @@ class FileApi(WebApiResource):
             source = None
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,

api/controllers/web/message.py

@@ -261,6 +261,8 @@ class MessageSuggestedQuestionApi(WebApiResource):
             questions = MessageService.get_suggested_questions_after_answer(
                 app_model=app_model, user=end_user, message_id=message_id, invoke_from=InvokeFrom.WEB_APP
             )
+            # questions is a list of strings, not a list of Message objects
+            # so we can directly return it
         except MessageNotExistsError:
             raise NotFound("Message not found")
         except ConversationNotExistsError:

api/controllers/web/remote_files.py

@@ -14,6 +14,7 @@ from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
+from extensions.ext_database import db
 from fields.file_fields import build_file_with_signed_url_model, build_remote_file_info_model
 from services.file_service import FileService
 
@@ -119,7 +120,7 @@ class RemoteFileUploadApi(WebApiResource):
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,
                 mimetype=file_info.mimetype,

api/core/agent/base_agent_runner.py

@@ -90,7 +90,9 @@ class BaseAgentRunner(AppRunner):
             tenant_id=tenant_id,
             dataset_ids=app_config.dataset.dataset_ids if app_config.dataset else [],
             retrieve_config=app_config.dataset.retrieve_config if app_config.dataset else None,
-            return_resource=app_config.additional_features.show_retrieve_source,
+            return_resource=(
+                app_config.additional_features.show_retrieve_source if app_config.additional_features else False
+            ),
             invoke_from=application_generate_entity.invoke_from,
             hit_callback=hit_callback,
             user_id=user_id,

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

@@ -1,4 +1,5 @@
 import uuid
+from typing import Literal, cast
 
 from core.app.app_config.entities import (
     DatasetEntity,
@@ -74,6 +75,9 @@ class DatasetConfigManager:
             return None
         query_variable = config.get("dataset_query_variable")
 
+        metadata_model_config_dict = dataset_configs.get("metadata_model_config")
+        metadata_filtering_conditions_dict = dataset_configs.get("metadata_filtering_conditions")
+
         if dataset_configs["retrieval_model"] == "single":
             return DatasetEntity(
                 dataset_ids=dataset_ids,
@@ -82,18 +86,23 @@ class DatasetConfigManager:
                     retrieve_strategy=DatasetRetrieveConfigEntity.RetrieveStrategy.value_of(
                         dataset_configs["retrieval_model"]
                     ),
-                    metadata_filtering_mode=dataset_configs.get("metadata_filtering_mode", "disabled"),
-                    metadata_model_config=ModelConfig(**dataset_configs.get("metadata_model_config"))
-                    if dataset_configs.get("metadata_model_config")
+                    metadata_filtering_mode=cast(
+                        Literal["disabled", "automatic", "manual"],
+                        dataset_configs.get("metadata_filtering_mode", "disabled"),
+                    ),
+                    metadata_model_config=ModelConfig(**metadata_model_config_dict)
+                    if isinstance(metadata_model_config_dict, dict)
                     else None,
-                    metadata_filtering_conditions=MetadataFilteringCondition(
-                        **dataset_configs.get("metadata_filtering_conditions", {})
-                    )
-                    if dataset_configs.get("metadata_filtering_conditions")
+                    metadata_filtering_conditions=MetadataFilteringCondition(**metadata_filtering_conditions_dict)
+                    if isinstance(metadata_filtering_conditions_dict, dict)
                     else None,
                 ),
             )
         else:
+            score_threshold_val = dataset_configs.get("score_threshold")
+            reranking_model_val = dataset_configs.get("reranking_model")
+            weights_val = dataset_configs.get("weights")
+
             return DatasetEntity(
                 dataset_ids=dataset_ids,
                 retrieve_config=DatasetRetrieveConfigEntity(
@@ -101,22 +110,23 @@ class DatasetConfigManager:
                     retrieve_strategy=DatasetRetrieveConfigEntity.RetrieveStrategy.value_of(
                         dataset_configs["retrieval_model"]
                     ),
-                    top_k=dataset_configs.get("top_k", 4),
-                    score_threshold=dataset_configs.get("score_threshold")
-                    if dataset_configs.get("score_threshold_enabled", False)
+                    top_k=int(dataset_configs.get("top_k", 4)),
+                    score_threshold=float(score_threshold_val)
+                    if dataset_configs.get("score_threshold_enabled", False) and score_threshold_val is not None
                     else None,
-                    reranking_model=dataset_configs.get("reranking_model"),
-                    weights=dataset_configs.get("weights"),
-                    reranking_enabled=dataset_configs.get("reranking_enabled", True),
+                    reranking_model=reranking_model_val if isinstance(reranking_model_val, dict) else None,
+                    weights=weights_val if isinstance(weights_val, dict) else None,
+                    reranking_enabled=bool(dataset_configs.get("reranking_enabled", True)),
                     rerank_mode=dataset_configs.get("reranking_mode", "reranking_model"),
-                    metadata_filtering_mode=dataset_configs.get("metadata_filtering_mode", "disabled"),
-                    metadata_model_config=ModelConfig(**dataset_configs.get("metadata_model_config"))
-                    if dataset_configs.get("metadata_model_config")
+                    metadata_filtering_mode=cast(
+                        Literal["disabled", "automatic", "manual"],
+                        dataset_configs.get("metadata_filtering_mode", "disabled"),
+                    ),
+                    metadata_model_config=ModelConfig(**metadata_model_config_dict)
+                    if isinstance(metadata_model_config_dict, dict)
                     else None,
-                    metadata_filtering_conditions=MetadataFilteringCondition(
-                        **dataset_configs.get("metadata_filtering_conditions", {})
-                    )
-                    if dataset_configs.get("metadata_filtering_conditions")
+                    metadata_filtering_conditions=MetadataFilteringCondition(**metadata_filtering_conditions_dict)
+                    if isinstance(metadata_filtering_conditions_dict, dict)
                     else None,
                 ),
             )
@@ -134,18 +144,17 @@ class DatasetConfigManager:
         config = cls.extract_dataset_config_for_legacy_compatibility(tenant_id, app_mode, config)
 
         # dataset_configs
-        if not config.get("dataset_configs"):
-            config["dataset_configs"] = {"retrieval_model": "single"}
+        if "dataset_configs" not in config or not config.get("dataset_configs"):
+            config["dataset_configs"] = {}
+        config["dataset_configs"]["retrieval_model"] = config["dataset_configs"].get("retrieval_model", "single")
 
         if not isinstance(config["dataset_configs"], dict):
             raise ValueError("dataset_configs must be of object type")
 
-        if not config["dataset_configs"].get("datasets"):
+        if "datasets" not in config["dataset_configs"] or not config["dataset_configs"].get("datasets"):
             config["dataset_configs"]["datasets"] = {"strategy": "router", "datasets": []}
 
-        need_manual_query_datasets = config.get("dataset_configs") and config["dataset_configs"].get(
-            "datasets", {}
-        ).get("datasets")
+        need_manual_query_datasets = config.get("dataset_configs", {}).get("datasets", {}).get("datasets")
 
         if need_manual_query_datasets and app_mode == AppMode.COMPLETION:
             # Only check when mode is completion
@@ -166,8 +175,8 @@ class DatasetConfigManager:
         :param config: app model config args
         """
         # Extract dataset config for legacy compatibility
-        if not config.get("agent_mode"):
-            config["agent_mode"] = {"enabled": False, "tools": []}
+        if "agent_mode" not in config or not config.get("agent_mode"):
+            config["agent_mode"] = {}
 
         if not isinstance(config["agent_mode"], dict):
             raise ValueError("agent_mode must be of object type")
@@ -180,19 +189,22 @@ class DatasetConfigManager:
             raise ValueError("enabled in agent_mode must be of boolean type")
 
         # tools
-        if not config["agent_mode"].get("tools"):
+        if "tools" not in config["agent_mode"] or not config["agent_mode"].get("tools"):
             config["agent_mode"]["tools"] = []
 
         if not isinstance(config["agent_mode"]["tools"], list):
             raise ValueError("tools in agent_mode must be a list of objects")
 
         # strategy
-        if not config["agent_mode"].get("strategy"):
+        if "strategy" not in config["agent_mode"] or not config["agent_mode"].get("strategy"):
             config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER.value
 
         has_datasets = False
-        if config["agent_mode"]["strategy"] in {PlanningStrategy.ROUTER.value, PlanningStrategy.REACT_ROUTER.value}:
-            for tool in config["agent_mode"]["tools"]:
+        if config.get("agent_mode", {}).get("strategy") in {
+            PlanningStrategy.ROUTER.value,
+            PlanningStrategy.REACT_ROUTER.value,
+        }:
+            for tool in config.get("agent_mode", {}).get("tools", []):
                 key = list(tool.keys())[0]
                 if key == "dataset":
                     # old style, use tool name as key
@@ -217,7 +229,7 @@ class DatasetConfigManager:
 
                     has_datasets = True
 
-        need_manual_query_datasets = has_datasets and config["agent_mode"]["enabled"]
+        need_manual_query_datasets = has_datasets and config.get("agent_mode", {}).get("enabled")
 
         if need_manual_query_datasets and app_mode == AppMode.COMPLETION:
             # Only check when mode is completion

api/core/app/app_config/entities.py

@@ -114,9 +114,9 @@ class VariableEntity(BaseModel):
     hide: bool = False
     max_length: int | None = 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: Sequence[FileType] | None = Field(default_factory=list)
+    allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
+    allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -129,6 +129,16 @@ class VariableEntity(BaseModel):
         return v or []
 
 
+class RagPipelineVariableEntity(VariableEntity):
+    """
+    Rag Pipeline Variable Entity.
+    """
+
+    tooltips: str | None = None
+    placeholder: str | None = None
+    belong_to_node_id: str
+
+
 class ExternalDataVariableEntity(BaseModel):
     """
     External Data Variable Entity.
@@ -288,7 +298,7 @@ class AppConfig(BaseModel):
     tenant_id: str
     app_id: str
     app_mode: AppMode
-    additional_features: AppAdditionalFeatures
+    additional_features: AppAdditionalFeatures | None = None
     variables: list[VariableEntity] = []
     sensitive_word_avoidance: SensitiveWordAvoidanceEntity | None = 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,48 @@ 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 _, 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, None)
+                if value.get("value") and isinstance(value.get("value"), list):
+                    last_part = value.get("value")[-1]
+                    variables_map.pop(last_part, None)
+
+        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_generator.py

@@ -154,7 +154,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
 
         if invoke_from == InvokeFrom.DEBUGGER:
             # always enable retriever resource in debugger mode
-            app_config.additional_features.show_retrieve_source = True
+            app_config.additional_features.show_retrieve_source = True  # type: ignore
 
         workflow_run_id = str(uuid.uuid4())
         # init application generate entity
@@ -420,7 +420,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             db.session.refresh(conversation)
 
         # get conversation dialogue count
-        self._dialogue_count = get_thread_messages_length(conversation.id)
+        # NOTE: dialogue_count should not start from 0,
+        # because during the first conversation, dialogue_count should be 1.
+        self._dialogue_count = get_thread_messages_length(conversation.id) + 1
 
         # init queue manager
         queue_manager = MessageBasedAppQueueManager(
@@ -467,7 +469,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             workflow_execution_repository=workflow_execution_repository,
             workflow_node_execution_repository=workflow_node_execution_repository,
             stream=stream,
-            draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from),
+            draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
         )
 
         return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)

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

@@ -79,29 +79,12 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         if not app_record:
             raise ValueError("App not found")
 
-        if self.application_generate_entity.single_iteration_run:
-            # if only single iteration run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
-            graph, variable_pool = self._get_graph_and_variable_pool_of_single_iteration(
+        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+            # Handle single iteration or single loop run
+            graph, variable_pool, graph_runtime_state = self._prepare_single_node_execution(
                 workflow=self._workflow,
-                node_id=self.application_generate_entity.single_iteration_run.node_id,
-                user_inputs=dict(self.application_generate_entity.single_iteration_run.inputs),
-                graph_runtime_state=graph_runtime_state,
-            )
-        elif self.application_generate_entity.single_loop_run:
-            # if only single loop run is requested
-            graph_runtime_state = GraphRuntimeState(
-                variable_pool=VariablePool.empty(),
-                start_at=time.time(),
-            )
-            graph, variable_pool = self._get_graph_and_variable_pool_of_single_loop(
-                workflow=self._workflow,
-                node_id=self.application_generate_entity.single_loop_run.node_id,
-                user_inputs=dict(self.application_generate_entity.single_loop_run.inputs),
-                graph_runtime_state=graph_runtime_state,
+                single_iteration_run=self.application_generate_entity.single_iteration_run,
+                single_loop_run=self.application_generate_entity.single_loop_run,
             )
         else:
             inputs = self.application_generate_entity.inputs

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

@@ -551,7 +551,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                 total_steps=validated_state.node_run_steps,
                 outputs=event.outputs,
                 exceptions_count=event.exceptions_count,
-                conversation_id=None,
+                conversation_id=self._conversation_id,
                 trace_manager=trace_manager,
                 external_trace_id=self._application_generate_entity.extras.get("external_trace_id"),
             )

api/core/app/apps/base_app_generator.py

@@ -14,6 +14,7 @@ from core.workflow.repositories.draft_variable_repository import (
 )
 from factories import file_factory
 from libs.orjson import orjson_dumps
+from models import Account, EndUser
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
@@ -44,9 +45,9 @@ class BaseAppGenerator:
                 mapping=v,
                 tenant_id=tenant_id,
                 config=FileUploadConfig(
-                    allowed_file_types=entity_dictionary[k].allowed_file_types,
-                    allowed_file_extensions=entity_dictionary[k].allowed_file_extensions,
-                    allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods,
+                    allowed_file_types=entity_dictionary[k].allowed_file_types or [],
+                    allowed_file_extensions=entity_dictionary[k].allowed_file_extensions or [],
+                    allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
                 strict_type_validation=strict_type_validation,
             )
@@ -59,9 +60,9 @@ class BaseAppGenerator:
                 mappings=v,
                 tenant_id=tenant_id,
                 config=FileUploadConfig(
-                    allowed_file_types=entity_dictionary[k].allowed_file_types,
-                    allowed_file_extensions=entity_dictionary[k].allowed_file_extensions,
-                    allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods,
+                    allowed_file_types=entity_dictionary[k].allowed_file_types or [],
+                    allowed_file_extensions=entity_dictionary[k].allowed_file_extensions or [],
+                    allowed_file_upload_methods=entity_dictionary[k].allowed_file_upload_methods or [],
                 ),
             )
             for k, v in user_inputs.items()
@@ -182,8 +183,9 @@ class BaseAppGenerator:
 
     @final
     @staticmethod
-    def _get_draft_var_saver_factory(invoke_from: InvokeFrom) -> DraftVariableSaverFactory:
+    def _get_draft_var_saver_factory(invoke_from: InvokeFrom, account: Account | EndUser) -> DraftVariableSaverFactory:
         if invoke_from == InvokeFrom.DEBUGGER:
+            assert isinstance(account, Account)
 
             def draft_var_saver_factory(
                 session: Session,
@@ -200,6 +202,7 @@ class BaseAppGenerator:
                     node_type=node_type,
                     node_execution_id=node_execution_id,
                     enclosing_node_id=enclosing_node_id,
+                    user=account,
                 )
         else:
 

api/core/app/apps/base_app_queue_manager.py

@@ -1,9 +1,11 @@
+import logging
 import queue
 import time
 from abc import abstractmethod
 from enum import IntEnum, auto
 from typing import Any
 
+from redis.exceptions import RedisError
 from sqlalchemy.orm import DeclarativeMeta
 
 from configs import dify_config
@@ -18,6 +20,8 @@ from core.app.entities.queue_entities import (
 )
 from extensions.ext_redis import redis_client
 
+logger = logging.getLogger(__name__)
+
 
 class PublishFrom(IntEnum):
     APPLICATION_MANAGER = auto()
@@ -35,9 +39,8 @@ class AppQueueManager:
         self.invoke_from = invoke_from  # Public accessor for invoke_from
 
         user_prefix = "account" if self._invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER} else "end-user"
-        redis_client.setex(
-            AppQueueManager._generate_task_belong_cache_key(self._task_id), 1800, f"{user_prefix}-{self._user_id}"
-        )
+        self._task_belong_cache_key = AppQueueManager._generate_task_belong_cache_key(self._task_id)
+        redis_client.setex(self._task_belong_cache_key, 1800, f"{user_prefix}-{self._user_id}")
 
         q: queue.Queue[WorkflowQueueMessage | MessageQueueMessage | None] = queue.Queue()
 
@@ -79,9 +82,21 @@ class AppQueueManager:
         Stop listen to queue
         :return:
         """
+        self._clear_task_belong_cache()
         self._q.put(None)
 
-    def publish_error(self, e, pub_from: PublishFrom):
+    def _clear_task_belong_cache(self) -> None:
+        """
+        Remove the task belong cache key once listening is finished.
+        """
+        try:
+            redis_client.delete(self._task_belong_cache_key)
+        except RedisError:
+            logger.exception(
+                "Failed to clear task belong cache for task %s (key: %s)", self._task_id, self._task_belong_cache_key
+            )
+
+    def publish_error(self, e, pub_from: PublishFrom) -> None:
         """
         Publish error
         :param e: error

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

@@ -164,7 +164,9 @@ class ChatAppRunner(AppRunner):
                 config=app_config.dataset,
                 query=query,
                 invoke_from=application_generate_entity.invoke_from,
-                show_retrieve_source=app_config.additional_features.show_retrieve_source,
+                show_retrieve_source=(
+                    app_config.additional_features.show_retrieve_source if app_config.additional_features else False
+                ),
                 hit_callback=hit_callback,
                 memory=memory,
                 message_id=message.id,

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

@@ -35,18 +35,19 @@ 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.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities import WorkflowExecution, WorkflowNodeExecution
-from core.workflow.enums import WorkflowNodeExecutionStatus
-from core.workflow.nodes import NodeType
+from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import (
     Account,
     EndUser,
 )
+from services.variable_truncator import VariableTruncator
 
 
 class WorkflowResponseConverter:
@@ -58,6 +59,7 @@ class WorkflowResponseConverter:
     ):
         self._application_generate_entity = application_generate_entity
         self._user = user
+        self._truncator = VariableTruncator.default()
 
     def workflow_start_to_stream_response(
         self,
@@ -149,7 +151,8 @@ class WorkflowResponseConverter:
                 title=workflow_node_execution.title,
                 index=workflow_node_execution.index,
                 predecessor_node_id=workflow_node_execution.predecessor_node_id,
-                inputs=workflow_node_execution.inputs,
+                inputs=workflow_node_execution.get_response_inputs(),
+                inputs_truncated=workflow_node_execution.inputs_truncated,
                 created_at=int(workflow_node_execution.created_at.timestamp()),
                 parallel_id=event.parallel_id,
                 parallel_start_node_id=event.parallel_start_node_id,
@@ -169,6 +172,15 @@ class WorkflowResponseConverter:
                 provider_type=ToolProviderType(event.provider_type),
                 provider_id=event.provider_id,
             )
+        elif event.node_type == NodeType.DATASOURCE:
+            manager = PluginDatasourceManager()
+            provider_entity = manager.fetch_datasource_provider(
+                self._application_generate_entity.app_config.tenant_id,
+                event.provider_id,
+            )
+            response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url(
+                self._application_generate_entity.app_config.tenant_id
+            )
 
         return response
 
@@ -198,9 +210,12 @@ class WorkflowResponseConverter:
                 index=workflow_node_execution.index,
                 title=workflow_node_execution.title,
                 predecessor_node_id=workflow_node_execution.predecessor_node_id,
-                inputs=workflow_node_execution.inputs,
-                process_data=workflow_node_execution.process_data,
-                outputs=json_converter.to_json_encodable(workflow_node_execution.outputs),
+                inputs=workflow_node_execution.get_response_inputs(),
+                inputs_truncated=workflow_node_execution.inputs_truncated,
+                process_data=workflow_node_execution.get_response_process_data(),
+                process_data_truncated=workflow_node_execution.process_data_truncated,
+                outputs=json_converter.to_json_encodable(workflow_node_execution.get_response_outputs()),
+                outputs_truncated=workflow_node_execution.outputs_truncated,
                 status=workflow_node_execution.status,
                 error=workflow_node_execution.error,
                 elapsed_time=workflow_node_execution.elapsed_time,
@@ -240,9 +255,12 @@ class WorkflowResponseConverter:
                 index=workflow_node_execution.index,
                 title=workflow_node_execution.title,
                 predecessor_node_id=workflow_node_execution.predecessor_node_id,
-                inputs=workflow_node_execution.inputs,
-                process_data=workflow_node_execution.process_data,
-                outputs=json_converter.to_json_encodable(workflow_node_execution.outputs),
+                inputs=workflow_node_execution.get_response_inputs(),
+                inputs_truncated=workflow_node_execution.inputs_truncated,
+                process_data=workflow_node_execution.get_response_process_data(),
+                process_data_truncated=workflow_node_execution.process_data_truncated,
+                outputs=json_converter.to_json_encodable(workflow_node_execution.get_response_outputs()),
+                outputs_truncated=workflow_node_execution.outputs_truncated,
                 status=workflow_node_execution.status,
                 error=workflow_node_execution.error,
                 elapsed_time=workflow_node_execution.elapsed_time,
@@ -267,6 +285,7 @@ class WorkflowResponseConverter:
         workflow_execution_id: str,
         event: QueueIterationStartEvent,
     ) -> IterationNodeStartStreamResponse:
+        new_inputs, truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
         return IterationNodeStartStreamResponse(
             task_id=task_id,
             workflow_run_id=workflow_execution_id,
@@ -277,7 +296,8 @@ class WorkflowResponseConverter:
                 title=event.node_title,
                 created_at=int(time.time()),
                 extras={},
-                inputs=event.inputs or {},
+                inputs=new_inputs,
+                inputs_truncated=truncated,
                 metadata=event.metadata or {},
             ),
         )
@@ -311,6 +331,11 @@ class WorkflowResponseConverter:
         event: QueueIterationCompletedEvent,
     ) -> IterationNodeCompletedStreamResponse:
         json_converter = WorkflowRuntimeTypeConverter()
+
+        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
+        new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping(
+            json_converter.to_json_encodable(event.outputs) or {}
+        )
         return IterationNodeCompletedStreamResponse(
             task_id=task_id,
             workflow_run_id=workflow_execution_id,
@@ -319,10 +344,12 @@ class WorkflowResponseConverter:
                 node_id=event.node_id,
                 node_type=event.node_type.value,
                 title=event.node_title,
-                outputs=json_converter.to_json_encodable(event.outputs) or {},
+                outputs=new_outputs,
+                outputs_truncated=outputs_truncated,
                 created_at=int(time.time()),
                 extras={},
-                inputs=event.inputs or {},
+                inputs=new_inputs,
+                inputs_truncated=inputs_truncated,
                 status=WorkflowNodeExecutionStatus.SUCCEEDED
                 if event.error is None
                 else WorkflowNodeExecutionStatus.FAILED,
@@ -338,6 +365,7 @@ class WorkflowResponseConverter:
     def workflow_loop_start_to_stream_response(
         self, *, task_id: str, workflow_execution_id: str, event: QueueLoopStartEvent
     ) -> LoopNodeStartStreamResponse:
+        new_inputs, truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
         return LoopNodeStartStreamResponse(
             task_id=task_id,
             workflow_run_id=workflow_execution_id,
@@ -348,7 +376,8 @@ class WorkflowResponseConverter:
                 title=event.node_title,
                 created_at=int(time.time()),
                 extras={},
-                inputs=event.inputs or {},
+                inputs=new_inputs,
+                inputs_truncated=truncated,
                 metadata=event.metadata or {},
                 parallel_id=event.parallel_id,
                 parallel_start_node_id=event.parallel_start_node_id,
@@ -371,7 +400,9 @@ class WorkflowResponseConverter:
                 node_type=event.node_type.value,
                 title=event.node_title,
                 index=event.index,
-                pre_loop_output=event.output,
+                # The `pre_loop_output` field is not utilized by the frontend.
+                # Previously, it was assigned the value of `event.output`.
+                pre_loop_output={},
                 created_at=int(time.time()),
                 extras={},
                 parallel_id=event.parallel_id,
@@ -387,6 +418,11 @@ class WorkflowResponseConverter:
         workflow_execution_id: str,
         event: QueueLoopCompletedEvent,
     ) -> LoopNodeCompletedStreamResponse:
+        json_converter = WorkflowRuntimeTypeConverter()
+        new_inputs, inputs_truncated = self._truncator.truncate_variable_mapping(event.inputs or {})
+        new_outputs, outputs_truncated = self._truncator.truncate_variable_mapping(
+            json_converter.to_json_encodable(event.outputs) or {}
+        )
         return LoopNodeCompletedStreamResponse(
             task_id=task_id,
             workflow_run_id=workflow_execution_id,
@@ -395,10 +431,12 @@ class WorkflowResponseConverter:
                 node_id=event.node_id,
                 node_type=event.node_type.value,
                 title=event.node_title,
-                outputs=WorkflowRuntimeTypeConverter().to_json_encodable(event.outputs) or {},
+                outputs=new_outputs,
+                outputs_truncated=outputs_truncated,
                 created_at=int(time.time()),
                 extras={},
-                inputs=event.inputs or {},
+                inputs=new_inputs,
+                inputs_truncated=inputs_truncated,
                 status=WorkflowNodeExecutionStatus.SUCCEEDED
                 if event.error is None
                 else WorkflowNodeExecutionStatus.FAILED,

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

@@ -124,7 +124,9 @@ class CompletionAppRunner(AppRunner):
                 config=dataset_config,
                 query=query or "",
                 invoke_from=application_generate_entity.invoke_from,
-                show_retrieve_source=app_config.additional_features.show_retrieve_source,
+                show_retrieve_source=app_config.additional_features.show_retrieve_source
+                if app_config.additional_features
+                else False,
                 hit_callback=hit_callback,
                 message_id=message.id,
                 inputs=inputs,

api/core/app/apps/pipeline/generate_response_converter.py

@@ -0,0 +1,95 @@
+from collections.abc import Generator
+from typing import cast
+
+from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
+from core.app.entities.task_entities import (
+    AppStreamResponse,
+    ErrorStreamResponse,
+    NodeFinishStreamResponse,
+    NodeStartStreamResponse,
+    PingStreamResponse,
+    WorkflowAppBlockingResponse,
+    WorkflowAppStreamResponse,
+)
+
+
+class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
+    _blocking_response_type = WorkflowAppBlockingResponse
+
+    @classmethod
+    def convert_blocking_full_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
+        """
+        Convert blocking full response.
+        :param blocking_response: blocking response
+        :return:
+        """
+        return dict(blocking_response.model_dump())
+
+    @classmethod
+    def convert_blocking_simple_response(cls, blocking_response: WorkflowAppBlockingResponse) -> dict:  # type: ignore[override]
+        """
+        Convert blocking simple response.
+        :param blocking_response: blocking response
+        :return:
+        """
+        return cls.convert_blocking_full_response(blocking_response)
+
+    @classmethod
+    def convert_stream_full_response(
+        cls, stream_response: Generator[AppStreamResponse, None, None]
+    ) -> Generator[dict | str, None, None]:
+        """
+        Convert stream full response.
+        :param stream_response: stream response
+        :return:
+        """
+        for chunk in stream_response:
+            chunk = cast(WorkflowAppStreamResponse, chunk)
+            sub_stream_response = chunk.stream_response
+
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
+
+            response_chunk = {
+                "event": sub_stream_response.event.value,
+                "workflow_run_id": chunk.workflow_run_id,
+            }
+
+            if isinstance(sub_stream_response, ErrorStreamResponse):
+                data = cls._error_to_stream_response(sub_stream_response.err)
+                response_chunk.update(cast(dict, data))
+            else:
+                response_chunk.update(sub_stream_response.model_dump())
+            yield response_chunk
+
+    @classmethod
+    def convert_stream_simple_response(
+        cls, stream_response: Generator[AppStreamResponse, None, None]
+    ) -> Generator[dict | str, None, None]:
+        """
+        Convert stream simple response.
+        :param stream_response: stream response
+        :return:
+        """
+        for chunk in stream_response:
+            chunk = cast(WorkflowAppStreamResponse, chunk)
+            sub_stream_response = chunk.stream_response
+
+            if isinstance(sub_stream_response, PingStreamResponse):
+                yield "ping"
+                continue
+
+            response_chunk = {
+                "event": sub_stream_response.event.value,
+                "workflow_run_id": chunk.workflow_run_id,
+            }
+
+            if isinstance(sub_stream_response, ErrorStreamResponse):
+                data = cls._error_to_stream_response(sub_stream_response.err)
+                response_chunk.update(cast(dict, data))
+            elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
+                response_chunk.update(cast(dict, sub_stream_response.to_ignore_detail_dict()))
+            else:
+                response_chunk.update(sub_stream_response.model_dump())
+            yield response_chunk