Merge branch 'feat/queue-based-graph-engine' into feat/rag-2

Signed-off-by: -LAN- <laipz8200@outlook.com>

.github/workflows/api-tests.yml

@@ -42,11 +42,7 @@ jobs:
       - name: Run Unit tests
         run: |
           uv run --project api bash dev/pytest/pytest_unit_tests.sh
-      - name: Run ty check
-        run: |
-          cd api
-          uv add --dev ty
-          uv run ty check || true
+
       - name: Run pyrefly check
         run: |
           cd api
@@ -66,15 +62,6 @@ jobs:
       - name: Run dify config tests
         run: uv run --project api dev/pytest/pytest_config_tests.py
 
-      - name: MyPy Cache
-        uses: actions/cache@v4
-        with:
-          path: api/.mypy_cache
-          key: mypy-${{ matrix.python-version }}-${{ runner.os }}-${{ hashFiles('api/uv.lock') }}
-
-      - name: Run MyPy Checks
-        run: dev/mypy-check
-
       - name: Set up dotenvs
         run: |
           cp docker/.env.example docker/.env

.github/workflows/build-push.yml

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

.github/workflows/deploy-dev.yml

@@ -4,7 +4,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/dev"
+      - "deploy/rag-dev"
     types:
       - completed
 
@@ -12,12 +12,13 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     if: |
-      github.event.workflow_run.conclusion == 'success'
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_branch == 'deploy/rag-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/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

@@ -12,7 +12,6 @@ permissions:
   statuses: write
   contents: read
 
-
 jobs:
   python-style:
     name: Python Style
@@ -44,6 +43,14 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv sync --project api --dev
 
+      - name: Run Basedpyright Checks
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: dev/basedpyright-check
+
+      - name: Run Mypy Type Checks
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --project api dotenv-linter ./api/.env.example ./web/.env.example

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

@@ -67,12 +67,22 @@ jobs:
         working-directory: ./web
         run: pnpm run auto-gen-i18n ${{ env.FILE_ARGS }}
 
+      - name: Generate i18n type definitions
+        if: env.FILES_CHANGED == 'true'
+        working-directory: ./web
+        run: pnpm run gen:i18n-types
+
       - name: Create Pull Request
         if: env.FILES_CHANGED == 'true'
         uses: peter-evans/create-pull-request@v6
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          commit-message: Update i18n files based on en-US changes
-          title: 'chore: translate i18n files'
-          body: This PR was automatically created to update i18n files based on changes in en-US locale.
+          commit-message: Update i18n files and type definitions based on en-US changes
+          title: 'chore: translate i18n files and update type definitions'
+          body: |
+            This PR was automatically created to update i18n files and TypeScript type definitions based on changes in en-US locale.
+            
+            **Changes included:**
+            - Updated translation files for all locales
+            - Regenerated TypeScript type definitions for type safety
           branch: chore/automated-i18n-updates

.github/workflows/web-tests.yml

@@ -47,6 +47,11 @@ jobs:
         working-directory: ./web
         run: pnpm install --frozen-lockfile
 
+      - name: Check i18n types synchronization
+        if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
+        run: pnpm run check:i18n-types
+
       - name: Run tests
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web

.gitignore

@@ -123,10 +123,12 @@ venv.bak/
 # mkdocs documentation
 /site
 
-# mypy
+# type checking
 .mypy_cache/
 .dmypy.json
 dmypy.json
+pyrightconfig.json
+!api/pyrightconfig.json
 
 # Pyre type checker
 .pyre/
@@ -195,8 +197,8 @@ sdks/python-client/dify_client.egg-info
 .vscode/*
 !.vscode/launch.json.template
 !.vscode/README.md
-pyrightconfig.json
 api/.vscode
+web/.vscode
 # vscode Code History Extension
 .history
 
@@ -214,6 +216,13 @@ mise.toml
 # Next.js build output
 .next/
 
+# PWA generated files
+web/public/sw.js
+web/public/sw.js.map
+web/public/workbox-*.js
+web/public/workbox-*.js.map
+web/public/fallback-*.js
+
 # AI Assistant
 .roo/
 api/.env.backup

CLAUDE.md

@@ -32,7 +32,7 @@ uv run --project api pytest tests/integration_tests/  # Integration tests
 ./dev/reformat                    # Run all formatters and linters
 uv run --project api ruff check --fix ./    # Fix linting issues
 uv run --project api ruff format ./         # Format code
-uv run --project api mypy .                 # Type checking
+uv run --directory api basedpyright         # Type checking
 ```
 
 ### Frontend (Web)

Makefile

@@ -4,6 +4,48 @@ WEB_IMAGE=$(DOCKER_REGISTRY)/dify-web
 API_IMAGE=$(DOCKER_REGISTRY)/dify-api
 VERSION=latest
 
+# Backend Development Environment Setup
+.PHONY: dev-setup prepare-docker prepare-web prepare-api
+
+# Default dev setup target
+dev-setup: prepare-docker prepare-web prepare-api
+	@echo "✅ Backend development environment setup complete!"
+
+# Step 1: Prepare Docker middleware
+prepare-docker:
+	@echo "🐳 Setting up Docker middleware..."
+	@cp -n docker/middleware.env.example docker/middleware.env 2>/dev/null || echo "Docker middleware.env already exists"
+	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev up -d
+	@echo "✅ Docker middleware started"
+
+# Step 2: Prepare web environment
+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
+prepare-api:
+	@echo "🔧 Setting up API environment..."
+	@cp -n api/.env.example api/.env 2>/dev/null || echo "API .env already exists"
+	@cd api && uv sync --dev
+	@cd api && uv run flask db upgrade
+	@echo "✅ API environment prepared (not started)"
+
+# Clean dev environment
+dev-clean:
+	@echo "⚠️  Stopping Docker containers..."
+	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev down
+	@echo "🗑️  Removing volumes..."
+	@rm -rf docker/volumes/db
+	@rm -rf docker/volumes/redis
+	@rm -rf docker/volumes/plugin_daemon
+	@rm -rf docker/volumes/weaviate
+	@rm -rf api/storage
+	@echo "✅ Cleanup complete"
+
 # Build Docker images
 build-web:
 	@echo "Building web Docker image: $(WEB_IMAGE):$(VERSION)..."
@@ -39,5 +81,21 @@ build-push-web: build-web push-web
 build-push-all: build-all push-all
 	@echo "All Docker images have been built and pushed."
 
+# Help target
+help:
+	@echo "Development Setup Targets:"
+	@echo "  make dev-setup      - Run all setup steps for backend dev environment"
+	@echo "  make prepare-docker - Set up Docker middleware"
+	@echo "  make prepare-web    - Set up web environment"
+	@echo "  make prepare-api    - Set up API environment"
+	@echo "  make dev-clean      - Stop Docker middleware containers"
+	@echo ""
+	@echo "Docker Build Targets:"
+	@echo "  make build-web      - Build web Docker image"
+	@echo "  make build-api      - Build API Docker image"
+	@echo "  make build-all      - Build all Docker images"
+	@echo "  make push-all       - Push all Docker images"
+	@echo "  make build-push-all - Build and push all Docker images"
+
 # Phony targets
-.PHONY: build-web build-api push-web push-api build-all push-all build-push-all
+.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help

api/.env.example

@@ -75,6 +75,7 @@ DB_PASSWORD=difyai123456
 DB_HOST=localhost
 DB_PORT=5432
 DB_DATABASE=dify
+SQLALCHEMY_POOL_PRE_PING=true
 
 # Storage configuration
 # use for store upload files, private keys...
@@ -460,6 +461,16 @@ WORKFLOW_CALL_MAX_DEPTH=5
 WORKFLOW_PARALLEL_DEPTH_LIMIT=3
 MAX_VARIABLE_SIZE=204800
 
+# GraphEngine Worker Pool Configuration
+# Minimum number of workers per GraphEngine instance (default: 1)
+GRAPH_ENGINE_MIN_WORKERS=1
+# Maximum number of workers per GraphEngine instance (default: 10)
+GRAPH_ENGINE_MAX_WORKERS=10
+# Queue depth threshold that triggers worker scale up (default: 3)
+GRAPH_ENGINE_SCALE_UP_THRESHOLD=3
+# Seconds of idle time before scaling down workers (default: 5.0)
+GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME=5.0
+
 # Workflow storage configuration
 # Options: rdbms, hybrid
 # rdbms: Use only the relational database (default)

api/.importlinter

@@ -0,0 +1,112 @@
+[importlinter]
+root_packages =
+    core
+    configs
+    controllers
+    models
+    tasks
+    services
+
+[importlinter:contract:workflow]
+name = Workflow
+type=layers
+layers =
+    graph_engine
+    graph_events
+    graph
+    nodes
+    node_events
+    entities
+containers =
+    core.workflow
+ignore_imports =
+    core.workflow.nodes.base.node -> core.workflow.graph_events
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_events
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
+    core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine.command_channels
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_events
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph
+    core.workflow.nodes.loop.loop_node -> core.workflow.graph_engine.command_channels
+    core.workflow.nodes.node_factory -> core.workflow.graph
+
+[importlinter:contract:rsc]
+name = RSC
+type = layers
+layers =
+    graph_engine
+    response_coordinator
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:worker]
+name = Worker
+type = layers
+layers =
+    graph_engine
+    worker
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:graph-engine-architecture]
+name = Graph Engine Architecture
+type = layers
+layers =
+    graph_engine
+    orchestration
+    command_processing
+    event_management
+    error_handling
+    graph_traversal
+    state_management
+    worker_management
+    domain
+containers =
+    core.workflow.graph_engine
+
+[importlinter:contract:domain-isolation]
+name = Domain Model Isolation
+type = forbidden
+source_modules =
+    core.workflow.graph_engine.domain
+forbidden_modules =
+    core.workflow.graph_engine.worker_management
+    core.workflow.graph_engine.command_channels
+    core.workflow.graph_engine.layers
+    core.workflow.graph_engine.protocols
+
+[importlinter:contract:worker-management]
+name = Worker Management
+type = forbidden
+source_modules =
+    core.workflow.graph_engine.worker_management
+forbidden_modules =
+    core.workflow.graph_engine.orchestration
+    core.workflow.graph_engine.command_processing
+    core.workflow.graph_engine.event_management
+
+[importlinter:contract:error-handling-strategies]
+name = Error Handling Strategies
+type = independence
+modules =
+    core.workflow.graph_engine.error_handling.abort_strategy
+    core.workflow.graph_engine.error_handling.retry_strategy
+    core.workflow.graph_engine.error_handling.fail_branch_strategy
+    core.workflow.graph_engine.error_handling.default_value_strategy
+
+[importlinter:contract:graph-traversal-components]
+name = Graph Traversal Components
+type = layers
+layers =
+    edge_processor
+    skip_propagator
+containers =
+    core.workflow.graph_engine.graph_traversal
+
+[importlinter:contract:command-channels]
+name = Command Channels Independence
+type = independence
+modules =
+    core.workflow.graph_engine.command_channels.in_memory_channel
+    core.workflow.graph_engine.command_channels.redis_channel

api/README.md

@@ -108,5 +108,5 @@ uv run celery -A app.celery beat
    ../dev/reformat               # Run all formatters and linters
    uv run ruff check --fix ./    # Fix linting issues
    uv run ruff format ./         # Format code
-   uv run mypy .                 # Type checking
+   uv run basedpyright .         # Type checking
    ```

api/app.py

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

api/app_factory.py

@@ -25,6 +25,9 @@ def create_flask_app_with_configs() -> DifyApp:
         # add an unique identifier to each request
         RecyclableContextVar.increment_thread_recycles()
 
+    # Capture the decorator's return value to avoid pyright reportUnusedFunction
+    _ = before_request
+
     return dify_app
 
 

api/celery_entrypoint.py

@@ -0,0 +1,22 @@
+import logging
+
+import psycogreen.gevent as pscycogreen_gevent  # type: ignore
+from grpc.experimental import gevent as grpc_gevent  # type: ignore
+
+_logger = logging.getLogger(__name__)
+
+
+def _log(message: str):
+    print(message, flush=True)
+
+
+# grpc gevent
+grpc_gevent.init_gevent()
+_log("gRPC  patched with gevent.")
+pscycogreen_gevent.patch_psycopg()
+_log("psycopg2 patched with gevent.")
+
+
+from app import app, celery
+
+__all__ = ["app", "celery"]

api/child_class.py

@@ -1,11 +0,0 @@
-from tests.integration_tests.utils.parent_class import ParentClass
-
-
-class ChildClass(ParentClass):
-    """Test child class for module import helper tests"""
-
-    def __init__(self, name):
-        super().__init__(name)
-
-    def get_name(self):
-        return f"Child: {self.name}"

api/commands.py

@@ -13,11 +13,13 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
-from core.plugin.entities.plugin import ToolProviderID
+from core.helper import encrypter
+from core.plugin.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
@@ -30,12 +32,16 @@ from models import Tenant
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
+from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
+from models.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__)
@@ -571,7 +577,7 @@ def old_metadata_migration():
         for document in documents:
             if document.doc_metadata:
                 doc_metadata = document.doc_metadata
-                for key, value in doc_metadata.items():
+                for key in doc_metadata:
                     for field in BuiltInField:
                         if field.value == key:
                             break
@@ -1233,15 +1239,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
@@ -1251,14 +1259,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,
         }
@@ -1287,6 +1302,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:
@@ -1295,10 +1311,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=lambda x: x[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
@@ -1307,7 +1323,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,
         )
 
@@ -1340,3 +1357,231 @@ 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)
+        firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
+        jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
+        oauth_credential_type = CredentialType.OAUTH2
+        api_key_credential_type = CredentialType.API_KEY
+
+        # deal notion credentials
+        deal_notion_count = 0
+        notion_credentials = db.session.query(DataSourceOauthBinding).filter_by(provider="notion").all()
+        if notion_credentials:
+            notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
+            for credential in notion_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in notion_credentials_tenant_mapping:
+                    notion_credentials_tenant_mapping[tenant_id] = []
+                notion_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in notion_credentials_tenant_mapping.items():
+                # check notion plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if notion_plugin_id not in installed_plugins_ids:
+                    if notion_plugin_unique_identifier:
+                        # install notion plugin
+                        PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier])
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential oauth params
+                    access_token = credential.access_token
+                    # notion info
+                    notion_info = credential.source_info
+                    workspace_id = notion_info.get("workspace_id")
+                    workspace_name = notion_info.get("workspace_name")
+                    workspace_icon = notion_info.get("workspace_icon")
+                    new_credentials = {
+                        "integration_secret": encrypter.encrypt_token(tenant_id, access_token),
+                        "workspace_id": workspace_id,
+                        "workspace_name": workspace_name,
+                        "workspace_icon": workspace_icon,
+                    }
+                    datasource_provider = DatasourceProvider(
+                        provider="notion_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
+                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 credential in firecrawl_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in firecrawl_credentials_tenant_mapping:
+                    firecrawl_credentials_tenant_mapping[tenant_id] = []
+                firecrawl_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in firecrawl_credentials_tenant_mapping.items():
+                # check firecrawl plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if firecrawl_plugin_id not in installed_plugins_ids:
+                    if firecrawl_plugin_unique_identifier:
+                        # install firecrawl plugin
+                        PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier])
+
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential api key
+                    credentials_json = json.loads(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
+                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 credential in jina_credentials:
+                tenant_id = credential.tenant_id
+                if tenant_id not in jina_credentials_tenant_mapping:
+                    jina_credentials_tenant_mapping[tenant_id] = []
+                jina_credentials_tenant_mapping[tenant_id].append(credential)
+            for tenant_id, credentials in jina_credentials_tenant_mapping.items():
+                # check jina plugin is installed
+                installed_plugins = installer_manager.list_plugins(tenant_id)
+                installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
+                if jina_plugin_id not in installed_plugins_ids:
+                    if jina_plugin_unique_identifier:
+                        # install jina plugin
+                        print(jina_plugin_unique_identifier)
+                        PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier])
+
+                auth_count = 0
+                for credential in credentials:
+                    auth_count += 1
+                    # get credential api key
+                    credentials_json = json.loads(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
+                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"))

api/configs/feature/__init__.py

@@ -499,6 +499,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(
+        50000,
+        description="maximum length for string to trigger tuncation, measure in number of characters",
+    )
+    WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH: PositiveInt = Field(
+        100,
+        description="maximum length for array to trigger truncation.",
+    )
+
+
 class WorkflowConfig(BaseSettings):
     """
     Configuration for workflow execution
@@ -529,6 +545,28 @@ class WorkflowConfig(BaseSettings):
         default=200 * 1024,
     )
 
+    # GraphEngine Worker Pool Configuration
+    GRAPH_ENGINE_MIN_WORKERS: PositiveInt = Field(
+        description="Minimum number of workers per GraphEngine instance",
+        default=1,
+    )
+
+    GRAPH_ENGINE_MAX_WORKERS: PositiveInt = Field(
+        description="Maximum number of workers per GraphEngine instance",
+        default=10,
+    )
+
+    GRAPH_ENGINE_SCALE_UP_THRESHOLD: PositiveInt = Field(
+        description="Queue depth threshold that triggers worker scale up",
+        default=3,
+    )
+
+    GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME: float = Field(
+        description="Seconds of idle time before scaling down workers",
+        default=5.0,
+        ge=0.1,
+    )
+
 
 class WorkflowNodeExecutionConfig(BaseSettings):
     """
@@ -1025,5 +1063,6 @@ class FeatureConfig(
     CeleryBeatConfig,
     CeleryScheduleTasksConfig,
     WorkflowLogConfig,
+    WorkflowVariableTruncationConfig,
 ):
     pass

api/configs/feature/hosted_service/__init__.py

@@ -222,11 +222,28 @@ class HostedFetchAppTemplateConfig(BaseSettings):
     )
 
 
+class HostedFetchPipelineTemplateConfig(BaseSettings):
+    """
+    Configuration for fetching pipeline templates
+    """
+
+    HOSTED_FETCH_PIPELINE_TEMPLATES_MODE: str = Field(
+        description="Mode for fetching pipeline templates: remote, db, or builtin default to remote,",
+        default="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

@@ -300,8 +300,7 @@ class DatasetQueueMonitorConfig(BaseSettings):
 
 class MiddlewareConfig(
     # place the configs in alphabet order
-    CeleryConfig,
-    DatabaseConfig,
+    CeleryConfig,  # Note: CeleryConfig already inherits from DatabaseConfig
     KeywordStoreConfig,
     RedisConfig,
     # configs of storage and storage providers

api/configs/middleware/vdb/clickzetta_config.py

@@ -1,9 +1,10 @@
 from typing import Optional
 
-from pydantic import BaseModel, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings
 
 
-class ClickzettaConfig(BaseModel):
+class ClickzettaConfig(BaseSettings):
     """
     Clickzetta Lakehouse vector database configuration
     """

api/configs/middleware/vdb/matrixone_config.py

@@ -1,7 +1,8 @@
-from pydantic import BaseModel, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings
 
 
-class MatrixoneConfig(BaseModel):
+class MatrixoneConfig(BaseSettings):
     """Matrixone vector database configuration."""
 
     MATRIXONE_HOST: str = Field(default="localhost", description="Host address of the Matrixone server")

api/configs/packaging/__init__.py

@@ -1,6 +1,6 @@
 from pydantic import Field
 
-from configs.packaging.pyproject import PyProjectConfig, PyProjectTomlConfig
+from configs.packaging.pyproject import PyProjectTomlConfig
 
 
 class PackagingInfo(PyProjectTomlConfig):

api/configs/remote_settings_sources/apollo/client.py

@@ -4,8 +4,9 @@ import logging
 import os
 import threading
 import time
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from pathlib import Path
+from typing import Any
 
 from .python_3x import http_request, makedirs_wrapper
 from .utils import (
@@ -25,13 +26,13 @@ logger = logging.getLogger(__name__)
 class ApolloClient:
     def __init__(
         self,
-        config_url,
-        app_id,
-        cluster="default",
-        secret="",
-        start_hot_update=True,
-        change_listener=None,
-        _notification_map=None,
+        config_url: str,
+        app_id: str,
+        cluster: str = "default",
+        secret: str = "",
+        start_hot_update: bool = True,
+        change_listener: Callable[[str, str, str, Any], None] | None = None,
+        _notification_map: dict[str, int] | None = None,
     ):
         # Core routing parameters
         self.config_url = config_url
@@ -47,17 +48,17 @@ class ApolloClient:
         # Private control variables
         self._cycle_time = 5
         self._stopping = False
-        self._cache = {}
-        self._no_key = {}
-        self._hash = {}
+        self._cache: dict[str, dict[str, Any]] = {}
+        self._no_key: dict[str, str] = {}
+        self._hash: dict[str, str] = {}
         self._pull_timeout = 75
         self._cache_file_path = os.path.expanduser("~") + "/.dify/config/remote-settings/apollo/cache/"
-        self._long_poll_thread = None
+        self._long_poll_thread: threading.Thread | None = None
         self._change_listener = change_listener  # "add" "delete" "update"
         if _notification_map is None:
             _notification_map = {"application": -1}
         self._notification_map = _notification_map
-        self.last_release_key = None
+        self.last_release_key: str | None = None
         # Private startup method
         self._path_checker()
         if start_hot_update:
@@ -68,7 +69,7 @@ class ApolloClient:
         heartbeat.daemon = True
         heartbeat.start()
 
-    def get_json_from_net(self, namespace="application"):
+    def get_json_from_net(self, namespace: str = "application") -> dict[str, Any] | None:
         url = "{}/configs/{}/{}/{}?releaseKey={}&ip={}".format(
             self.config_url, self.app_id, self.cluster, namespace, "", self.ip
         )
@@ -88,7 +89,7 @@ class ApolloClient:
             logger.exception("an error occurred in get_json_from_net")
             return None
 
-    def get_value(self, key, default_val=None, namespace="application"):
+    def get_value(self, key: str, default_val: Any = None, namespace: str = "application") -> Any:
         try:
             # read memory configuration
             namespace_cache = self._cache.get(namespace)
@@ -104,7 +105,8 @@ class ApolloClient:
             namespace_data = self.get_json_from_net(namespace)
             val = get_value_from_dict(namespace_data, key)
             if val is not None:
-                self._update_cache_and_file(namespace_data, namespace)
+                if namespace_data is not None:
+                    self._update_cache_and_file(namespace_data, namespace)
                 return val
 
             # read the file configuration
@@ -126,23 +128,23 @@ class ApolloClient:
     # to ensure the real-time correctness of the function call.
     # If the user does not have the same default val twice
     # and the default val is used here, there may be a problem.
-    def _set_local_cache_none(self, namespace, key):
+    def _set_local_cache_none(self, namespace: str, key: str) -> None:
         no_key = no_key_cache_key(namespace, key)
         self._no_key[no_key] = key
 
-    def _start_hot_update(self):
+    def _start_hot_update(self) -> None:
         self._long_poll_thread = threading.Thread(target=self._listener)
         # When the asynchronous thread is started, the daemon thread will automatically exit
         # when the main thread is launched.
         self._long_poll_thread.daemon = True
         self._long_poll_thread.start()
 
-    def stop(self):
+    def stop(self) -> None:
         self._stopping = True
         logger.info("Stopping listener...")
 
     # Call the set callback function, and if it is abnormal, try it out
-    def _call_listener(self, namespace, old_kv, new_kv):
+    def _call_listener(self, namespace: str, old_kv: dict[str, Any] | None, new_kv: dict[str, Any] | None) -> None:
         if self._change_listener is None:
             return
         if old_kv is None:
@@ -168,12 +170,12 @@ class ApolloClient:
         except BaseException as e:
             logger.warning(str(e))
 
-    def _path_checker(self):
+    def _path_checker(self) -> None:
         if not os.path.isdir(self._cache_file_path):
             makedirs_wrapper(self._cache_file_path)
 
     # update the local cache and file cache
-    def _update_cache_and_file(self, namespace_data, namespace="application"):
+    def _update_cache_and_file(self, namespace_data: dict[str, Any], namespace: str = "application") -> None:
         # update the local cache
         self._cache[namespace] = namespace_data
         # update the file cache
@@ -187,7 +189,7 @@ class ApolloClient:
             self._hash[namespace] = new_hash
 
     # get the configuration from the local file
-    def _get_local_cache(self, namespace="application"):
+    def _get_local_cache(self, namespace: str = "application") -> dict[str, Any]:
         cache_file_path = os.path.join(self._cache_file_path, f"{self.app_id}_configuration_{namespace}.txt")
         if os.path.isfile(cache_file_path):
             with open(cache_file_path) as f:
@@ -195,8 +197,8 @@ class ApolloClient:
             return result
         return {}
 
-    def _long_poll(self):
-        notifications = []
+    def _long_poll(self) -> None:
+        notifications: list[dict[str, Any]] = []
         for key in self._cache:
             namespace_data = self._cache[key]
             notification_id = -1
@@ -236,7 +238,7 @@ class ApolloClient:
         except Exception as e:
             logger.warning(str(e))
 
-    def _get_net_and_set_local(self, namespace, n_id, call_change=False):
+    def _get_net_and_set_local(self, namespace: str, n_id: int, call_change: bool = False) -> None:
         namespace_data = self.get_json_from_net(namespace)
         if not namespace_data:
             return
@@ -248,7 +250,7 @@ class ApolloClient:
             new_kv = namespace_data.get(CONFIGURATIONS)
             self._call_listener(namespace, old_kv, new_kv)
 
-    def _listener(self):
+    def _listener(self) -> None:
         logger.info("start long_poll")
         while not self._stopping:
             self._long_poll()
@@ -266,13 +268,13 @@ class ApolloClient:
         headers["Timestamp"] = time_unix_now
         return headers
 
-    def _heart_beat(self):
+    def _heart_beat(self) -> None:
         while not self._stopping:
             for namespace in self._notification_map:
                 self._do_heart_beat(namespace)
             time.sleep(60 * 10)  # 10 minutes
 
-    def _do_heart_beat(self, namespace):
+    def _do_heart_beat(self, namespace: str) -> None:
         url = f"{self.config_url}/configs/{self.app_id}/{self.cluster}/{namespace}?ip={self.ip}"
         try:
             code, body = http_request(url, timeout=3, headers=self._sign_headers(url))
@@ -292,7 +294,7 @@ class ApolloClient:
             logger.exception("an error occurred in _do_heart_beat")
             return None
 
-    def get_all_dicts(self, namespace):
+    def get_all_dicts(self, namespace: str) -> dict[str, Any] | None:
         namespace_data = self._cache.get(namespace)
         if namespace_data is None:
             net_namespace_data = self.get_json_from_net(namespace)

api/configs/remote_settings_sources/apollo/python_3x.py

@@ -2,6 +2,8 @@ import logging
 import os
 import ssl
 import urllib.request
+from collections.abc import Mapping
+from typing import Any
 from urllib import parse
 from urllib.error import HTTPError
 
@@ -19,9 +21,9 @@ urllib.request.install_opener(opener)
 logger = logging.getLogger(__name__)
 
 
-def http_request(url, timeout, headers={}):
+def http_request(url: str, timeout: int | float, headers: Mapping[str, str] = {}) -> tuple[int, str | None]:
     try:
-        request = urllib.request.Request(url, headers=headers)
+        request = urllib.request.Request(url, headers=dict(headers))
         res = urllib.request.urlopen(request, timeout=timeout)
         body = res.read().decode("utf-8")
         return res.code, body
@@ -33,9 +35,9 @@ def http_request(url, timeout, headers={}):
         raise e
 
 
-def url_encode(params):
+def url_encode(params: dict[str, Any]) -> str:
     return parse.urlencode(params)
 
 
-def makedirs_wrapper(path):
+def makedirs_wrapper(path: str) -> None:
     os.makedirs(path, exist_ok=True)

api/configs/remote_settings_sources/apollo/utils.py

@@ -1,5 +1,6 @@
 import hashlib
 import socket
+from typing import Any
 
 from .python_3x import url_encode
 
@@ -10,7 +11,7 @@ NAMESPACE_NAME = "namespaceName"
 
 
 # add timestamps uris and keys
-def signature(timestamp, uri, secret):
+def signature(timestamp: str, uri: str, secret: str) -> str:
     import base64
     import hmac
 
@@ -19,16 +20,16 @@ def signature(timestamp, uri, secret):
     return base64.b64encode(hmac_code).decode()
 
 
-def url_encode_wrapper(params):
+def url_encode_wrapper(params: dict[str, Any]) -> str:
     return url_encode(params)
 
 
-def no_key_cache_key(namespace, key):
+def no_key_cache_key(namespace: str, key: str) -> str:
     return f"{namespace}{len(namespace)}{key}"
 
 
 # Returns whether the obtained value is obtained, and None if it does not
-def get_value_from_dict(namespace_cache, key):
+def get_value_from_dict(namespace_cache: dict[str, Any] | None, key: str) -> Any | None:
     if namespace_cache:
         kv_data = namespace_cache.get(CONFIGURATIONS)
         if kv_data is None:
@@ -38,7 +39,7 @@ def get_value_from_dict(namespace_cache, key):
     return None
 
 
-def init_ip():
+def init_ip() -> str:
     ip = ""
     s = None
     try:

api/configs/remote_settings_sources/base.py

@@ -11,5 +11,5 @@ class RemoteSettingsSource:
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         raise NotImplementedError
 
-    def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any:
+    def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool):
         return value

api/configs/remote_settings_sources/nacos/__init__.py

@@ -11,16 +11,16 @@ logger = logging.getLogger(__name__)
 
 from configs.remote_settings_sources.base import RemoteSettingsSource
 
-from .utils import _parse_config
+from .utils import parse_config
 
 
 class NacosSettingsSource(RemoteSettingsSource):
     def __init__(self, configs: Mapping[str, Any]):
         self.configs = configs
-        self.remote_configs: dict[str, Any] = {}
+        self.remote_configs: dict[str, str] = {}
         self.async_init()
 
-    def async_init(self):
+    def async_init(self) -> None:
         data_id = os.getenv("DIFY_ENV_NACOS_DATA_ID", "dify-api-env.properties")
         group = os.getenv("DIFY_ENV_NACOS_GROUP", "nacos-dify")
         tenant = os.getenv("DIFY_ENV_NACOS_NAMESPACE", "")
@@ -29,22 +29,19 @@ class NacosSettingsSource(RemoteSettingsSource):
         try:
             content = NacosHttpClient().http_request("/nacos/v1/cs/configs", method="GET", headers={}, params=params)
             self.remote_configs = self._parse_config(content)
-        except Exception as e:
+        except Exception:
             logger.exception("[get-access-token] exception occurred")
             raise
 
-    def _parse_config(self, content: str) -> dict:
+    def _parse_config(self, content: str) -> dict[str, str]:
         if not content:
             return {}
         try:
-            return _parse_config(self, content)
+            return parse_config(content)
         except Exception as e:
             raise RuntimeError(f"Failed to parse config: {e}")
 
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
-        if not isinstance(self.remote_configs, dict):
-            raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}")
-
         field_value = self.remote_configs.get(field_name)
         if field_value is None:
             return None, field_name, False

api/configs/remote_settings_sources/nacos/http_request.py

@@ -17,20 +17,26 @@ class NacosHttpClient:
         self.ak = os.getenv("DIFY_ENV_NACOS_ACCESS_KEY")
         self.sk = os.getenv("DIFY_ENV_NACOS_SECRET_KEY")
         self.server = os.getenv("DIFY_ENV_NACOS_SERVER_ADDR", "localhost:8848")
-        self.token = None
+        self.token: str | None = None
         self.token_ttl = 18000
         self.token_expire_time: float = 0
 
-    def http_request(self, url, method="GET", headers=None, params=None):
+    def http_request(
+        self, url: str, method: str = "GET", headers: dict[str, str] | None = None, params: dict[str, str] | None = None
+    ) -> str:
+        if headers is None:
+            headers = {}
+        if params is None:
+            params = {}
         try:
             self._inject_auth_info(headers, params)
             response = requests.request(method, url="http://" + self.server + url, headers=headers, params=params)
             response.raise_for_status()
             return response.text
-        except requests.exceptions.RequestException as e:
+        except requests.RequestException as e:
             return f"Request to Nacos failed: {e}"
 
-    def _inject_auth_info(self, headers, params, module="config"):
+    def _inject_auth_info(self, headers: dict[str, str], params: dict[str, str], module: str = "config") -> None:
         headers.update({"User-Agent": "Nacos-Http-Client-In-Dify:v0.0.1"})
 
         if module == "login":
@@ -45,16 +51,17 @@ class NacosHttpClient:
             headers["timeStamp"] = ts
         if self.username and self.password:
             self.get_access_token(force_refresh=False)
-            params["accessToken"] = self.token
+            if self.token is not None:
+                params["accessToken"] = self.token
 
-    def __do_sign(self, sign_str, sk):
+    def __do_sign(self, sign_str: str, sk: str) -> str:
         return (
             base64.encodebytes(hmac.new(sk.encode(), sign_str.encode(), digestmod=hashlib.sha1).digest())
             .decode()
             .strip()
         )
 
-    def get_sign_str(self, group, tenant, ts):
+    def get_sign_str(self, group: str, tenant: str, ts: str) -> str:
         sign_str = ""
         if tenant:
             sign_str = tenant + "+"
@@ -63,7 +70,7 @@ class NacosHttpClient:
         sign_str += ts  # Directly concatenate ts without conditional checks, because the nacos auth header forced it.
         return sign_str
 
-    def get_access_token(self, force_refresh=False):
+    def get_access_token(self, force_refresh: bool = False) -> str | None:
         current_time = time.time()
         if self.token and not force_refresh and self.token_expire_time > current_time:
             return self.token
@@ -77,6 +84,7 @@ class NacosHttpClient:
             self.token = response_data.get("accessToken")
             self.token_ttl = response_data.get("tokenTtl", 18000)
             self.token_expire_time = current_time + self.token_ttl - 10
-        except Exception as e:
+            return self.token
+        except Exception:
             logger.exception("[get-access-token] exception occur")
             raise

api/configs/remote_settings_sources/nacos/utils.py

@@ -1,4 +1,4 @@
-def _parse_config(self, content: str) -> dict[str, str]:
+def parse_config(content: str) -> dict[str, str]:
     config: dict[str, str] = {}
     if not content:
         return config

api/constants/languages.py

@@ -19,6 +19,7 @@ language_timezone_mapping = {
     "fa-IR": "Asia/Tehran",
     "sl-SI": "Europe/Ljubljana",
     "th-TH": "Asia/Bangkok",
+    "id-ID": "Asia/Jakarta",
 }
 
 languages = list(language_timezone_mapping.keys())

api/contexts/__init__.py

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

api/controllers/console/__init__.py

@@ -43,7 +43,7 @@ api.add_resource(AppImportConfirmApi, "/apps/imports/<string:import_id>/confirm"
 api.add_resource(AppImportCheckDependenciesApi, "/apps/imports/<string:app_id>/check-dependencies")
 
 # Import other controllers
-from . import admin, apikey, extension, feature, ping, setup, version
+from . import admin, apikey, extension, feature, ping, setup, spec, version
 
 # Import app controllers
 from .app import (
@@ -86,6 +86,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 (

api/controllers/console/admin.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource, reqparse
@@ -6,6 +8,8 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound, Unauthorized
 
+P = ParamSpec("P")
+R = TypeVar("R")
 from configs import dify_config
 from constants.languages import supported_language
 from controllers.console import api
@@ -14,9 +18,9 @@ from extensions.ext_database import db
 from models.model import App, InstalledApp, RecommendedApp
 
 
-def admin_required(view):
+def admin_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ADMIN_API_KEY:
             raise Unauthorized("API key is invalid.")
 
@@ -130,15 +134,19 @@ class InsertExploreAppApi(Resource):
             app.is_public = False
 
         with Session(db.engine) as session:
-            installed_apps = session.execute(
-                select(InstalledApp).where(
-                    InstalledApp.app_id == recommended_app.app_id,
-                    InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
+            installed_apps = (
+                session.execute(
+                    select(InstalledApp).where(
+                        InstalledApp.app_id == recommended_app.app_id,
+                        InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
+                    )
                 )
-            ).all()
+                .scalars()
+                .all()
+            )
 
-        for installed_app in installed_apps:
-            db.session.delete(installed_app)
+            for installed_app in installed_apps:
+                session.delete(installed_app)
 
         db.session.delete(recommended_app)
         db.session.commit()

api/controllers/console/apikey.py

@@ -84,10 +84,10 @@ class BaseApiKeyListResource(Resource):
             flask_restx.abort(
                 400,
                 message=f"Cannot create more than {self.max_keys} API keys for this resource type.",
-                code="max_keys_exceeded",
+                custom="max_keys_exceeded",
             )
 
-        key = ApiToken.generate_api_key(self.token_prefix, 24)
+        key = ApiToken.generate_api_key(self.token_prefix or "", 24)
         api_token = ApiToken()
         setattr(api_token, self.resource_id_field, resource_id)
         api_token.tenant_id = current_user.current_tenant_id

api/controllers/console/app/app.py

@@ -237,9 +237,14 @@ class AppExportApi(Resource):
         # Add include_secret params
         parser = reqparse.RequestParser()
         parser.add_argument("include_secret", type=inputs.boolean, default=False, location="args")
+        parser.add_argument("workflow_id", type=str, location="args")
         args = parser.parse_args()
 
-        return {"data": AppDslService.export_dsl(app_model=app_model, include_secret=args["include_secret"])}
+        return {
+            "data": AppDslService.export_dsl(
+                app_model=app_model, include_secret=args["include_secret"], workflow_id=args.get("workflow_id")
+            )
+        }
 
 
 class AppNameApi(Resource):

api/controllers/console/app/conversation.py

@@ -117,7 +117,7 @@ class CompletionConversationDetailApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
+    @get_app_model(mode=AppMode.COMPLETION)
     def delete(self, app_model, conversation_id):
         if not current_user.is_editor:
             raise Forbidden()

api/controllers/console/app/generator.py

@@ -16,7 +16,10 @@ from core.helper.code_executor.javascript.javascript_code_provider import Javasc
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
+from extensions.ext_database import db
 from libs.login import login_required
+from models import App
+from services.workflow_service import WorkflowService
 
 
 class RuleGenerateApi(Resource):
@@ -135,9 +138,6 @@ class InstructionGenerateApi(Resource):
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":
-                from models import App, db
-                from services.workflow_service import WorkflowService
-
                 app = db.session.query(App).where(App.id == args["flow_id"]).first()
                 if not app:
                     return {"error": f"app {args['flow_id']} not found"}, 400
@@ -207,7 +207,7 @@ class InstructionGenerationTemplateApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def post(self) -> dict:
+    def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("type", type=str, required=True, default=False, location="json")
         args = parser.parse_args()

api/controllers/console/app/workflow.py

@@ -24,6 +24,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
+from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -413,7 +414,12 @@ class WorkflowTaskStopApi(Resource):
         if not current_user.is_editor:
             raise Forbidden()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.DEBUGGER, current_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 
@@ -526,7 +532,7 @@ class PublishedWorkflowApi(Resource):
             )
 
             app_model.workflow_id = workflow.id
-            db.session.commit()
+            db.session.commit()  # NOTE: this is necessary for update app_model.workflow_id
 
             workflow_created_at = TimestampField().format(workflow.created_at)
 

api/controllers/console/app/workflow_app_log.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
-from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs.login import login_required
@@ -27,7 +27,9 @@ class WorkflowAppLogApi(Resource):
         """
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
-        parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument(
+            "status", type=str, choices=["succeeded", "failed", "stopped", "partial-succeeded"], location="args"
+        )
         parser.add_argument(
             "created_at__before", type=str, location="args", help="Filter logs created before this timestamp"
         )

api/controllers/console/app/workflow_draft_variable.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, NoReturn
+from typing import NoReturn
 
 from flask import Response
 from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
@@ -13,14 +13,16 @@ 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
 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 import App, AppMode, db
+from models import App, AppMode
 from models.account import Account
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
@@ -29,7 +31,7 @@ from services.workflow_service import WorkflowService
 logger = logging.getLogger(__name__)
 
 
-def _convert_values_to_json_serializable_object(value: Segment) -> Any:
+def _convert_values_to_json_serializable_object(value: Segment):
     if isinstance(value, FileSegment):
         return value.value.model_dump()
     elif isinstance(value, ArrayFileSegment):
@@ -40,7 +42,7 @@ def _convert_values_to_json_serializable_object(value: Segment) -> Any:
         return value.value
 
 
-def _serialize_var_value(variable: WorkflowDraftVariable) -> Any:
+def _serialize_var_value(variable: WorkflowDraftVariable):
     value = variable.get_value()
     # create a copy of the value to avoid affecting the model cache.
     value = value.model_copy(deep=True)
@@ -74,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()),
@@ -83,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_oauth.py

@@ -81,7 +81,7 @@ class OAuthDataSourceBinding(Resource):
                 return {"error": "Invalid code"}, 400
             try:
                 oauth_provider.get_access_token(code)
-            except requests.exceptions.HTTPError as e:
+            except requests.HTTPError as e:
                 logger.exception(
                     "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
                 )
@@ -104,7 +104,7 @@ class OAuthDataSourceSync(Resource):
             return {"error": "Invalid provider"}, 400
         try:
             oauth_provider.sync_data_source(binding_id)
-        except requests.exceptions.HTTPError as e:
+        except requests.HTTPError as e:
             logger.exception(
                 "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
             )

api/controllers/console/auth/login.py

@@ -130,7 +130,7 @@ class ResetPasswordSendEmailApi(Resource):
             language = "en-US"
         try:
             account = AccountService.get_user_through_email(args["email"])
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
 
         if account is None:
@@ -162,7 +162,7 @@ class EmailCodeLoginSendEmailApi(Resource):
             language = "en-US"
         try:
             account = AccountService.get_user_through_email(args["email"])
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
 
         if account is None:
@@ -200,7 +200,7 @@ class EmailCodeLoginApi(Resource):
         AccountService.revoke_email_code_login_token(args["token"])
         try:
             account = AccountService.get_user_through_email(user_email)
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
         if account:
             tenants = TenantService.get_join_tenants(account)
@@ -223,7 +223,7 @@ class EmailCodeLoginApi(Resource):
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
-            except AccountRegisterError as are:
+            except AccountRegisterError:
                 raise AccountInFreezeError()
             except WorkspacesLimitExceededError:
                 raise WorkspacesLimitExceeded()

api/controllers/console/auth/oauth.py

@@ -80,7 +80,7 @@ class OAuthCallback(Resource):
         try:
             token = oauth_provider.get_access_token(code)
             user_info = oauth_provider.get_user_info(token)
-        except requests.exceptions.RequestException as e:
+        except requests.RequestException as e:
             error_text = e.response.text if e.response else str(e)
             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

@@ -1,8 +1,9 @@
+from collections.abc import Callable
 from functools import wraps
-from typing import cast
+from typing import Concatenate, ParamSpec, TypeVar, cast
 
 import flask_login
-from flask import request
+from flask import jsonify, request
 from flask_restx import Resource, reqparse
 from werkzeug.exceptions import BadRequest, NotFound
 
@@ -15,10 +16,14 @@ from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType,
 
 from .. import api
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
 
-def oauth_server_client_id_required(view):
+
+def oauth_server_client_id_required(view: Callable[Concatenate[T, OAuthProviderApp, P], R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(self: T, *args: P.args, **kwargs: P.kwargs):
         parser = reqparse.RequestParser()
         parser.add_argument("client_id", type=str, required=True, location="json")
         parsed_args = parser.parse_args()
@@ -30,43 +35,53 @@ def oauth_server_client_id_required(view):
         if not oauth_provider_app:
             raise NotFound("client_id is invalid")
 
-        kwargs["oauth_provider_app"] = oauth_provider_app
-
-        return view(*args, **kwargs)
+        return view(self, oauth_provider_app, *args, **kwargs)
 
     return decorated
 
 
-def oauth_server_access_token_required(view):
+def oauth_server_access_token_required(view: Callable[Concatenate[T, OAuthProviderApp, Account, P], R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
-        oauth_provider_app = kwargs.get("oauth_provider_app")
-        if not oauth_provider_app or not isinstance(oauth_provider_app, OAuthProviderApp):
+    def decorated(self: T, oauth_provider_app: OAuthProviderApp, *args: P.args, **kwargs: P.kwargs):
+        if not isinstance(oauth_provider_app, OAuthProviderApp):
             raise BadRequest("Invalid oauth_provider_app")
 
         authorization_header = request.headers.get("Authorization")
         if not authorization_header:
-            raise BadRequest("Authorization header is required")
+            response = jsonify({"error": "Authorization header is required"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
 
-        parts = authorization_header.strip().split(" ")
+        parts = authorization_header.strip().split(None, 1)
         if len(parts) != 2:
-            raise BadRequest("Invalid Authorization header format")
+            response = jsonify({"error": "Invalid Authorization header format"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
 
         token_type = parts[0].strip()
         if token_type.lower() != "bearer":
-            raise BadRequest("token_type is invalid")
+            response = jsonify({"error": "token_type is invalid"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
 
         access_token = parts[1].strip()
         if not access_token:
-            raise BadRequest("access_token is required")
+            response = jsonify({"error": "access_token is required"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
 
         account = OAuthServerService.validate_oauth_access_token(oauth_provider_app.client_id, access_token)
         if not account:
-            raise BadRequest("access_token or client_id is invalid")
+            response = jsonify({"error": "access_token or client_id is invalid"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
 
-        kwargs["account"] = account
-
-        return view(*args, **kwargs)
+        return view(self, oauth_provider_app, account, *args, **kwargs)
 
     return decorated
 

api/controllers/console/billing/billing.py

@@ -1,9 +1,9 @@
-from flask_login import current_user
 from flask_restx import Resource, reqparse
 
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.model import Account
 from services.billing_service import BillingService
 
 
@@ -17,9 +17,10 @@ class Subscription(Resource):
         parser.add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
         parser.add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
         args = parser.parse_args()
+        assert isinstance(current_user, Account)
 
         BillingService.is_tenant_owner_or_admin(current_user)
-
+        assert current_user.current_tenant_id is not None
         return BillingService.get_subscription(
             args["plan"], args["interval"], current_user.email, current_user.current_tenant_id
         )
@@ -31,7 +32,9 @@ class Invoices(Resource):
     @account_initialization_required
     @only_edition_cloud
     def get(self):
+        assert isinstance(current_user, Account)
         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/controllers/console/datasets/data_source.py

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

api/controllers/console/datasets/datasets.py

@@ -19,9 +19,9 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
 from core.model_runtime.entities.model_entities import ModelType
-from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
@@ -31,6 +31,7 @@ from fields.document_fields import document_status_fields
 from libs.login import login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
+from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 
 
@@ -279,6 +280,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()
 
@@ -422,17 +432,21 @@ class DatasetIndexingEstimateApi(Resource):
             if file_details:
                 for file_detail in file_details:
                     extract_setting = ExtractSetting(
-                        datasource_type="upload_file", upload_file=file_detail, document_model=args["doc_form"]
+                        datasource_type=DatasourceType.FILE.value,
+                        upload_file=file_detail,
+                        document_model=args["doc_form"],
                     )
                     extract_settings.append(extract_setting)
         elif args["info_list"]["data_source_type"] == "notion_import":
             notion_info_list = args["info_list"]["notion_info_list"]
             for notion_info in notion_info_list:
                 workspace_id = notion_info["workspace_id"]
+                credential_id = notion_info.get("credential_id")
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
-                        datasource_type="notion_import",
+                        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"],
@@ -445,7 +459,7 @@ class DatasetIndexingEstimateApi(Resource):
             website_info_list = args["info_list"]["website_info_list"]
             for url in website_info_list["urls"]:
                 extract_setting = ExtractSetting(
-                    datasource_type="website_crawl",
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info={
                         "provider": website_info_list["provider"],
                         "job_id": website_info_list["job_id"],

api/controllers/console/datasets/datasets_document.py

@@ -1,3 +1,4 @@
+import json
 import logging
 from argparse import ArgumentTypeError
 from typing import Literal, cast
@@ -40,6 +41,7 @@ from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.plugin.impl.exc import PluginDaemonClientSideError
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from extensions.ext_database import db
 from fields.document_fields import (
@@ -51,6 +53,7 @@ from fields.document_fields import (
 from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Dataset, DatasetProcessRule, Document, DocumentSegment, UploadFile
+from models.dataset import DocumentPipelineExecutionLog
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 
@@ -354,9 +357,6 @@ class DatasetInitApi(Resource):
         parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")
         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()
         knowledge_config = KnowledgeConfig(**args)
         if knowledge_config.indexing_technique == "high_quality":
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
@@ -428,7 +428,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type="upload_file", upload_file=file, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file, document_model=document.doc_form
                 )
 
                 indexing_runner = IndexingRunner()
@@ -477,6 +477,8 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
             data_source_info = document.data_source_info_dict
 
             if document.data_source_type == "upload_file":
+                if not data_source_info:
+                    continue
                 file_id = data_source_info["upload_file_id"]
                 file_detail = (
                     db.session.query(UploadFile)
@@ -488,14 +490,17 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type="upload_file", upload_file=file_detail, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file_detail, document_model=document.doc_form
                 )
                 extract_settings.append(extract_setting)
 
             elif document.data_source_type == "notion_import":
+                if not data_source_info:
+                    continue
                 extract_setting = ExtractSetting(
-                    datasource_type="notion_import",
+                    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"],
@@ -505,8 +510,10 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 )
                 extract_settings.append(extract_setting)
             elif document.data_source_type == "website_crawl":
+                if not data_source_info:
+                    continue
                 extract_setting = ExtractSetting(
-                    datasource_type="website_crawl",
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info={
                         "provider": data_source_info["provider"],
                         "job_id": data_source_info["job_id"],
@@ -649,7 +656,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,
@@ -1012,6 +1019,41 @@ class WebsiteDocumentSyncApi(DocumentResource):
         return {"result": "success"}, 200
 
 
+class DocumentPipelineExecutionLogApi(DocumentResource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        document = DocumentService.get_document(dataset.id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        log = (
+            db.session.query(DocumentPipelineExecutionLog)
+            .filter_by(document_id=document_id)
+            .order_by(DocumentPipelineExecutionLog.created_at.desc())
+            .first()
+        )
+        if not log:
+            return {
+                "datasource_info": None,
+                "datasource_type": None,
+                "input_data": None,
+                "datasource_node_id": None,
+            }, 200
+        return {
+            "datasource_info": json.loads(log.datasource_info),
+            "datasource_type": log.datasource_type,
+            "input_data": log.input_data,
+            "datasource_node_id": log.datasource_node_id,
+        }, 200
+
+
 api.add_resource(GetProcessRuleApi, "/datasets/process-rule")
 api.add_resource(DatasetDocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DatasetInitApi, "/datasets/init")
@@ -1033,3 +1075,6 @@ api.add_resource(DocumentRetryApi, "/datasets/<uuid:dataset_id>/retry")
 api.add_resource(DocumentRenameApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/rename")
 
 api.add_resource(WebsiteDocumentSyncApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/website-sync")
+api.add_resource(
+    DocumentPipelineExecutionLogApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log"
+)

api/controllers/console/datasets/error.py

@@ -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/rag_pipeline/datasource_auth.py

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

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

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

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

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

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

@@ -0,0 +1,114 @@
+from flask_login import current_user  # type: ignore  # type: ignore
+from flask_restx import Resource, marshal, reqparse  # type: ignore
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+import services
+from controllers.console import api
+from controllers.console.datasets.error import DatasetNameDuplicateError
+from controllers.console.wraps import (
+    account_initialization_required,
+    cloud_edition_billing_rate_limit_check,
+    setup_required,
+)
+from 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
+
+
+def _validate_name(name):
+    if not name or len(name) < 1 or len(name) > 40:
+        raise ValueError("Name must be between 1 to 40 characters.")
+    return name
+
+
+def _validate_description_length(description):
+    if len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
+class CreateRagPipelineDatasetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self):
+        parser = reqparse.RequestParser()
+
+        parser.add_argument(
+            "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
+
+
+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.add_resource(CreateRagPipelineDatasetApi, "/rag/pipeline/dataset")
+api.add_resource(CreateEmptyRagPipelineDatasetApi, "/rag/pipeline/empty-dataset")

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

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

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

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

api/controllers/console/datasets/wraps.py

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

api/controllers/console/explore/conversation.py

@@ -61,7 +61,6 @@ class ConversationApi(InstalledAppResource):
             ConversationService.delete(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        WebConversationService.unpin(app_model, conversation_id, current_user)
 
         return {"result": "success"}, 204
 

api/controllers/console/explore/parameter.py

@@ -43,6 +43,8 @@ class ExploreAppMetaApi(InstalledAppResource):
     def get(self, installed_app: InstalledApp):
         """Get app meta"""
         app_model = installed_app.app
+        if not app_model:
+            raise ValueError("App not found")
         return AppService().get_app_meta(app_model)
 
 

api/controllers/console/explore/workflow.py

@@ -20,6 +20,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
+from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
 from libs.login import current_user
 from models.model import AppMode, InstalledApp
@@ -35,6 +36,8 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
         Run workflow
         """
         app_model = installed_app.app
+        if not app_model:
+            raise NotWorkflowAppError()
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
@@ -73,11 +76,18 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
         Stop workflow task
         """
         app_model = installed_app.app
+        if not app_model:
+            raise NotWorkflowAppError()
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
         assert current_user is not None
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.EXPLORE, current_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}

api/controllers/console/explore/wraps.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import Concatenate, Optional, ParamSpec, TypeVar
 
 from flask_login import current_user
 from flask_restx import Resource
@@ -13,19 +15,15 @@ from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
 
-def installed_app_required(view=None):
-    def decorator(view):
+
+def installed_app_required(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
+    def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
-            if not kwargs.get("installed_app_id"):
-                raise ValueError("missing installed_app_id in path parameters")
-
-            installed_app_id = kwargs.get("installed_app_id")
-            installed_app_id = str(installed_app_id)
-
-            del kwargs["installed_app_id"]
-
+        def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
             installed_app = (
                 db.session.query(InstalledApp)
                 .where(
@@ -52,10 +50,10 @@ def installed_app_required(view=None):
     return decorator
 
 
-def user_allowed_to_access_app(view=None):
-    def decorator(view):
+def user_allowed_to_access_app(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
+    def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
-        def decorated(installed_app: InstalledApp, *args, **kwargs):
+        def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):
             feature = FeatureService.get_system_features()
             if feature.webapp_auth.enabled:
                 app_id = installed_app.app_id

api/controllers/console/files.py

@@ -20,6 +20,7 @@ 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 services.file_service import FileService
@@ -68,7 +69,7 @@ class FileApi(Resource):
             source = None
 
         try:
-            upload_file = FileService.upload_file(
+            upload_file = FileService(db.engine).upload_file(
                 filename=file.filename,
                 content=file.read(),
                 mimetype=file.mimetype,
@@ -89,7 +90,7 @@ class FilePreviewApi(Resource):
     @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}
 
 

api/controllers/console/remote_files.py

@@ -14,6 +14,7 @@ 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
@@ -61,7 +62,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,35 @@
+import logging
+
+from flask_restx import Resource
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.schemas.schema_manager import SchemaManager
+from libs.login import login_required
+
+logger = logging.getLogger(__name__)
+
+
+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.add_resource(SpecSchemaDefinitionsApi, "/spec/schema-definitions")

api/controllers/console/workspace/__init__.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask_login import current_user
 from sqlalchemy.orm import Session
@@ -7,14 +9,17 @@ from werkzeug.exceptions import Forbidden
 from extensions.ext_database import db
 from models.account import TenantPluginPermission
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def plugin_permission_required(
     install_required: bool = False,
     debug_required: bool = False,
 ):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             user = current_user
             tenant_id = user.current_tenant_id
 

api/controllers/console/workspace/model_providers.py

@@ -67,7 +67,7 @@ class ModelProviderCredentialApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()
@@ -94,7 +94,7 @@ class ModelProviderCredentialApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("credential_id", type=uuid_value, required=True, nullable=False, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()

api/controllers/console/workspace/models.py

@@ -219,7 +219,11 @@ class ModelProviderModelCredentialApi(Resource):
 
         model_load_balancing_service = ModelLoadBalancingService()
         is_load_balancing_enabled, load_balancing_configs = model_load_balancing_service.get_load_balancing_configs(
-            tenant_id=tenant_id, provider=provider, model=args["model"], model_type=args["model_type"]
+            tenant_id=tenant_id,
+            provider=provider,
+            model=args["model"],
+            model_type=args["model_type"],
+            config_from=args.get("config_from", ""),
         )
 
         if args.get("config_from", "") == "predefined-model":
@@ -263,7 +267,7 @@ class ModelProviderModelCredentialApi(Resource):
             choices=[mt.value for mt in ModelType],
             location="json",
         )
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
@@ -309,7 +313,7 @@ class ModelProviderModelCredentialApi(Resource):
         )
         parser.add_argument("credential_id", type=uuid_value, required=True, nullable=False, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()

api/controllers/console/workspace/tool_providers.py

@@ -21,11 +21,11 @@ from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.plugin.entities.plugin import ToolProviderID
 from core.plugin.impl.oauth import OAuthHandler
 from core.tools.entities.tool_entities import CredentialType
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import login_required
+from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService

api/controllers/console/workspace/workspace.py

@@ -211,7 +211,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

@@ -2,7 +2,9 @@ import contextlib
 import json
 import os
 import time
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import abort, request
 from flask_login import current_user
@@ -19,10 +21,13 @@ from services.operation_service import OperationService
 
 from .error import NotInitValidateError, NotSetupError, UnauthorizedAndForceLogout
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-def account_initialization_required(view):
+
+def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
         account = current_user
 
@@ -34,9 +39,9 @@ def account_initialization_required(view):
     return decorated
 
 
-def only_edition_cloud(view):
+def only_edition_cloud(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "CLOUD":
             abort(404)
 
@@ -45,9 +50,9 @@ def only_edition_cloud(view):
     return decorated
 
 
-def only_edition_enterprise(view):
+def only_edition_enterprise(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ENTERPRISE_ENABLED:
             abort(404)
 
@@ -56,9 +61,9 @@ def only_edition_enterprise(view):
     return decorated
 
 
-def only_edition_self_hosted(view):
+def only_edition_self_hosted(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "SELF_HOSTED":
             abort(404)
 
@@ -67,9 +72,9 @@ def only_edition_self_hosted(view):
     return decorated
 
 
-def cloud_edition_billing_enabled(view):
+def cloud_edition_billing_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_features(current_user.current_tenant_id)
         if not features.billing.enabled:
             abort(403, "Billing feature is not enabled.")
@@ -79,9 +84,9 @@ def cloud_edition_billing_enabled(view):
 
 
 def cloud_edition_billing_resource_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 members = features.members
@@ -120,9 +125,9 @@ def cloud_edition_billing_resource_check(resource: str):
 
 
 def cloud_edition_billing_knowledge_limit_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
@@ -142,9 +147,9 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
 
 
 def cloud_edition_billing_rate_limit_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             if resource == "knowledge":
                 knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_user.current_tenant_id)
                 if knowledge_rate_limit.enabled:
@@ -176,9 +181,9 @@ def cloud_edition_billing_rate_limit_check(resource: str):
     return interceptor
 
 
-def cloud_utm_record(view):
+def cloud_utm_record(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
             features = FeatureService.get_features(current_user.current_tenant_id)
 
@@ -194,9 +199,9 @@ def cloud_utm_record(view):
     return decorated
 
 
-def setup_required(view):
+def setup_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check setup
         if (
             dify_config.EDITION == "SELF_HOSTED"
@@ -212,9 +217,9 @@ def setup_required(view):
     return decorated
 
 
-def enterprise_license_required(view):
+def enterprise_license_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         settings = FeatureService.get_system_features()
         if settings.license.status in [LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST]:
             raise UnauthorizedAndForceLogout("Your license is invalid. Please contact your administrator.")
@@ -224,9 +229,9 @@ def enterprise_license_required(view):
     return decorated
 
 
-def email_password_login_enabled(view):
+def email_password_login_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if features.enable_email_password_login:
             return view(*args, **kwargs)
@@ -237,9 +242,9 @@ def email_password_login_enabled(view):
     return decorated
 
 
-def enable_change_email(view):
+def enable_change_email(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if features.enable_change_email:
             return view(*args, **kwargs)
@@ -250,9 +255,9 @@ def enable_change_email(view):
     return decorated
 
 
-def is_allow_transfer_owner(view):
+def is_allow_transfer_owner(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_features(current_user.current_tenant_id)
         if features.is_allow_transfer_workspace:
             return view(*args, **kwargs)
@@ -261,3 +266,14 @@ def is_allow_transfer_owner(view):
         abort(403)
 
     return decorated
+
+
+def knowledge_pipeline_publish_enabled(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_features(current_user.current_tenant_id)
+        if features.knowledge_pipeline.publish_enabled:
+            return view(*args, **kwargs)
+        abort(403)
+
+    return decorated

api/controllers/files/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/files/tool_files.py

@@ -8,7 +8,7 @@ from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import files_ns
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
-from models import db as global_db
+from extensions.ext_database import db as global_db
 
 
 @files_ns.route("/tools/<uuid:file_id>.<string:extension>")

api/controllers/mcp/mcp.py

@@ -99,7 +99,7 @@ class MCPAppApi(Resource):
 
         return mcp_server, app
 
-    def _validate_server_status(self, mcp_server: AppMCPServer) -> None:
+    def _validate_server_status(self, mcp_server: AppMCPServer):
         """Validate MCP server status"""
         if mcp_server.status != AppMCPServerStatus.ACTIVE:
             raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server is not active")

api/controllers/service_api/app/audio.py

@@ -55,7 +55,7 @@ class AudioApi(Resource):
         file = request.files["file"]
 
         try:
-            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user)
+            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.id)
 
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:

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/app/file_preview.py

@@ -59,7 +59,7 @@ class FilePreviewApi(Resource):
         args = file_preview_parser.parse_args()
 
         # Validate file ownership and get file objects
-        message_file, upload_file = self._validate_file_ownership(file_id, app_model.id)
+        _, upload_file = self._validate_file_ownership(file_id, app_model.id)
 
         # Get file content generator
         try:

api/controllers/service_api/app/workflow.py

@@ -26,7 +26,8 @@ from core.errors.error import (
 )
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus
+from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import build_workflow_app_log_pagination_model
 from libs import helper
@@ -262,7 +263,12 @@ class WorkflowTaskStopApi(Resource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.SERVICE_API, end_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/service_api/dataset/dataset.py

@@ -13,13 +13,13 @@ from controllers.service_api.wraps import (
     validate_dataset_token,
 )
 from core.model_runtime.entities.model_entities import ModelType
-from core.plugin.entities.plugin import ModelProviderID
 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 models.account import Account
 from models.dataset import Dataset, DatasetPermissionEnum
+from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
 from services.tag_service import TagService

api/controllers/service_api/dataset/document.py

@@ -123,7 +123,7 @@ class DocumentAddByTextApi(DatasetApiResource):
                 args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
             )
 
-        upload_file = FileService.upload_text(text=str(text), text_name=str(name))
+        upload_file = FileService(db.engine).upload_text(text=str(text), text_name=str(name))
         data_source = {
             "type": "upload_file",
             "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -133,6 +133,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,
@@ -198,7 +201,7 @@ 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))
+            upload_file = FileService(db.engine).upload_text(text=str(text), text_name=str(name))
             data_source = {
                 "type": "upload_file",
                 "info_list": {"data_source_type": "upload_file", "file_info_list": {"file_ids": [upload_file.id]}},
@@ -298,7 +301,7 @@ class DocumentAddByFileApi(DatasetApiResource):
         if not file.filename:
             raise FilenameNotExistsError
 
-        upload_file = FileService.upload_file(
+        upload_file = FileService(db.engine).upload_file(
             filename=file.filename,
             content=file.read(),
             mimetype=file.mimetype,
@@ -387,7 +390,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
                 raise FilenameNotExistsError
 
             try:
-                upload_file = FileService.upload_file(
+                upload_file = FileService(db.engine).upload_file(
                     filename=file.filename,
                     content=file.read(),
                     mimetype=file.mimetype,
@@ -410,7 +413,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         DocumentService.document_create_args_validate(knowledge_config)
 
         try:
-            documents, batch = DocumentService.save_document_with_dataset_id(
+            documents, _ = DocumentService.save_document_with_dataset_id(
                 dataset=dataset,
                 knowledge_config=knowledge_config,
                 account=dataset.created_by_account,

api/controllers/service_api/dataset/segment.py

@@ -440,7 +440,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Segment not found.")
 
         # validate segment belongs to the specified document
-        if segment.document_id != document_id:
+        if str(segment.document_id) != str(document_id):
             raise NotFound("Document not found.")
 
         # check child chunk
@@ -451,7 +451,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Child chunk not found.")
 
         # validate child chunk belongs to the specified segment
-        if child_chunk.segment_id != segment.id:
+        if str(child_chunk.segment_id) != str(segment.id):
             raise NotFound("Child chunk not found.")
 
         try:
@@ -500,7 +500,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Segment not found.")
 
         # validate segment belongs to the specified document
-        if segment.document_id != document_id:
+        if str(segment.document_id) != str(document_id):
             raise NotFound("Segment not found.")
 
         # get child chunk
@@ -511,7 +511,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Child chunk not found.")
 
         # validate child chunk belongs to the specified segment
-        if child_chunk.segment_id != segment.id:
+        if str(child_chunk.segment_id) != str(segment.id):
             raise NotFound("Child chunk not found.")
 
         # validate args

api/controllers/service_api/wraps.py

@@ -3,7 +3,7 @@ from collections.abc import Callable
 from datetime import timedelta
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Optional
+from typing import Optional, ParamSpec, TypeVar
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -22,6 +22,9 @@ from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App, EndUser
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 class WhereisUserArg(StrEnum):
     """
@@ -60,27 +63,6 @@ def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optio
             if tenant.status == TenantStatus.ARCHIVE:
                 raise Forbidden("The workspace's status is archived.")
 
-            tenant_account_join = (
-                db.session.query(Tenant, TenantAccountJoin)
-                .where(Tenant.id == api_token.tenant_id)
-                .where(TenantAccountJoin.tenant_id == Tenant.id)
-                .where(TenantAccountJoin.role.in_(["owner"]))
-                .where(Tenant.status == TenantStatus.NORMAL)
-                .one_or_none()
-            )  # TODO: only owner information is required, so only one is returned.
-            if tenant_account_join:
-                tenant, ta = tenant_account_join
-                account = db.session.query(Account).where(Account.id == ta.account_id).first()
-                # Login admin
-                if account:
-                    account.current_tenant = tenant
-                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                    user_logged_in.send(current_app._get_current_object(), user=_get_user())  # type: ignore
-                else:
-                    raise Unauthorized("Tenant owner account does not exist.")
-            else:
-                raise Unauthorized("Tenant does not exist.")
-
             kwargs["app_model"] = app_model
 
             if fetch_user_arg:
@@ -118,8 +100,8 @@ def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optio
 
 
 def cloud_edition_billing_resource_check(resource: str, api_token_type: str):
-    def interceptor(view):
-        def decorated(*args, **kwargs):
+    def interceptor(view: Callable[P, R]):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
             features = FeatureService.get_features(api_token.tenant_id)
 
@@ -148,9 +130,9 @@ def cloud_edition_billing_resource_check(resource: str, api_token_type: str):
 
 
 def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
             features = FeatureService.get_features(api_token.tenant_id)
             if features.billing.enabled:
@@ -170,9 +152,9 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
 
 
 def cloud_edition_billing_rate_limit_check(resource: str, api_token_type: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
 
             if resource == "knowledge":
@@ -291,27 +273,28 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: Optional[str]
     if not user_id:
         user_id = "DEFAULT-USER"
 
-    end_user = (
-        db.session.query(EndUser)
-        .where(
-            EndUser.tenant_id == app_model.tenant_id,
-            EndUser.app_id == app_model.id,
-            EndUser.session_id == user_id,
-            EndUser.type == "service_api",
+    with Session(db.engine, expire_on_commit=False) as session:
+        end_user = (
+            session.query(EndUser)
+            .where(
+                EndUser.tenant_id == app_model.tenant_id,
+                EndUser.app_id == app_model.id,
+                EndUser.session_id == user_id,
+                EndUser.type == "service_api",
+            )
+            .first()
         )
-        .first()
-    )
 
-    if end_user is None:
-        end_user = EndUser(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            type="service_api",
-            is_anonymous=user_id == "DEFAULT-USER",
-            session_id=user_id,
-        )
-        db.session.add(end_user)
-        db.session.commit()
+        if end_user is None:
+            end_user = EndUser(
+                tenant_id=app_model.tenant_id,
+                app_id=app_model.id,
+                type="service_api",
+                is_anonymous=user_id == "DEFAULT-USER",
+                session_id=user_id,
+            )
+            session.add(end_user)
+            session.commit()
 
     return end_user
 

api/controllers/web/conversation.py

@@ -73,8 +73,6 @@ class ConversationApi(WebApiResource):
             ConversationService.delete(app_model, conversation_id, end_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        WebConversationService.unpin(app_model, conversation_id, end_user)
-
         return {"result": "success"}, 204
 
 

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/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/controllers/web/workflow.py

@@ -21,6 +21,7 @@ from core.errors.error import (
     QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
+from core.workflow.graph_engine.manager import GraphEngineManager
 from libs import helper
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
@@ -110,7 +111,12 @@ class WorkflowTaskStopApi(WebApiResource):
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
 
-        AppQueueManager.set_stop_flag(task_id, InvokeFrom.WEB_APP, end_user.id)
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
 
         return {"result": "success"}
 

api/controllers/web/wraps.py

@@ -1,9 +1,11 @@
 from datetime import UTC, datetime
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource
 from sqlalchemy import select
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
@@ -14,6 +16,9 @@ from services.enterprise.enterprise_service import EnterpriseService, WebAppSett
 from services.feature_service import FeatureService
 from services.webapp_auth_service import WebAppAuthService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def validate_jwt_token(view=None):
     def decorator(view):
@@ -49,18 +54,19 @@ def decode_jwt_token():
         decoded = PassportService().verify(tk)
         app_code = decoded.get("app_code")
         app_id = decoded.get("app_id")
-        app_model = db.session.scalar(select(App).where(App.id == app_id))
-        site = db.session.scalar(select(Site).where(Site.code == app_code))
-        if not app_model:
-            raise NotFound()
-        if not app_code or not site:
-            raise BadRequest("Site URL is no longer valid.")
-        if app_model.enable_site is False:
-            raise BadRequest("Site is disabled.")
-        end_user_id = decoded.get("end_user_id")
-        end_user = db.session.scalar(select(EndUser).where(EndUser.id == end_user_id))
-        if not end_user:
-            raise NotFound()
+        with Session(db.engine, expire_on_commit=False) as session:
+            app_model = session.scalar(select(App).where(App.id == app_id))
+            site = session.scalar(select(Site).where(Site.code == app_code))
+            if not app_model:
+                raise NotFound()
+            if not app_code or not site:
+                raise BadRequest("Site URL is no longer valid.")
+            if app_model.enable_site is False:
+                raise BadRequest("Site is disabled.")
+            end_user_id = decoded.get("end_user_id")
+            end_user = session.scalar(select(EndUser).where(EndUser.id == end_user_id))
+            if not end_user:
+                raise NotFound()
 
         # for enterprise webapp auth
         app_web_auth_enabled = False

api/core/agent/base_agent_runner.py

@@ -62,7 +62,7 @@ class BaseAgentRunner(AppRunner):
         model_instance: ModelInstance,
         memory: Optional[TokenBufferMemory] = None,
         prompt_messages: Optional[list[PromptMessage]] = None,
-    ) -> None:
+    ):
         self.tenant_id = tenant_id
         self.application_generate_entity = application_generate_entity
         self.conversation = conversation
@@ -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,
@@ -334,7 +336,8 @@ class BaseAgentRunner(AppRunner):
         """
         Save agent thought
         """
-        agent_thought = db.session.query(MessageAgentThought).where(MessageAgentThought.id == agent_thought_id).first()
+        stmt = select(MessageAgentThought).where(MessageAgentThought.id == agent_thought_id)
+        agent_thought = db.session.scalar(stmt)
         if not agent_thought:
             raise ValueError("agent thought not found")
 
@@ -492,7 +495,8 @@ class BaseAgentRunner(AppRunner):
         return result
 
     def organize_agent_user_prompt(self, message: Message) -> UserPromptMessage:
-        files = db.session.query(MessageFile).where(MessageFile.message_id == message.id).all()
+        stmt = select(MessageFile).where(MessageFile.message_id == message.id)
+        files = db.session.scalars(stmt).all()
         if not files:
             return UserPromptMessage(content=message.query)
         if message.app_model_config:

api/core/agent/cot_agent_runner.py

@@ -338,7 +338,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
 
         return instruction
 
-    def _init_react_state(self, query) -> None:
+    def _init_react_state(self, query):
         """
         init agent scratchpad
         """

api/core/agent/entities.py

@@ -41,7 +41,7 @@ class AgentScratchpadUnit(BaseModel):
         action_name: str
         action_input: Union[dict, str]
 
-        def to_dict(self) -> dict:
+        def to_dict(self):
             """
             Convert to dictionary.
             """

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

@@ -158,7 +158,7 @@ class DatasetConfigManager:
         return config, ["agent_mode", "dataset_configs", "dataset_query_variable"]
 
     @classmethod
-    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict) -> dict:
+    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict):
         """
         Extract dataset config for legacy compatibility
 

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

@@ -4,8 +4,8 @@ from typing import Any
 from core.app.app_config.entities import ModelConfigEntity
 from core.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
 from core.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
-from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
+from models.provider_ids import ModelProviderID
 
 
 class ModelConfigManager:
@@ -105,7 +105,7 @@ class ModelConfigManager:
         return dict(config), ["model"]
 
     @classmethod
-    def validate_model_completion_params(cls, cp: dict) -> dict:
+    def validate_model_completion_params(cls, cp: dict):
         # model.completion_params
         if not isinstance(cp, dict):
             raise ValueError("model.completion_params must be of object type")

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

@@ -122,7 +122,7 @@ class PromptTemplateConfigManager:
         return config, ["prompt_type", "pre_prompt", "chat_prompt_config", "completion_prompt_config"]
 
     @classmethod
-    def validate_post_prompt_and_set_defaults(cls, config: dict) -> dict:
+    def validate_post_prompt_and_set_defaults(cls, config: dict):
         """
         Validate post_prompt and set defaults for prompt feature
 

api/core/app/app_config/entities.py

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

api/core/app/app_config/features/more_like_this/manager.py

@@ -26,7 +26,7 @@ class MoreLikeThisConfigManager:
     def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
         try:
             return AppConfigModel.model_validate(config).model_dump(), ["more_like_this"]
-        except ValidationError as e:
+        except ValidationError:
             raise ValueError(
                 "more_like_this must be of dict type and enabled in more_like_this must be of boolean type"
             )

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_config_manager.py

@@ -41,7 +41,7 @@ class AdvancedChatAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False) -> dict:
+    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False):
         """
         Validate for advanced chat app model config
 

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
@@ -450,6 +450,12 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
 
         worker_thread.start()
 
+        # release database connection, because the following new thread operations may take a long time
+        db.session.refresh(workflow)
+        db.session.refresh(message)
+        # db.session.refresh(user)
+        db.session.close()
+
         # return response or stream generator
         response = self._handle_advanced_chat_response(
             application_generate_entity=application_generate_entity,
@@ -461,7 +467,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)
@@ -475,7 +481,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         message_id: str,
         context: contextvars.Context,
         variable_loader: VariableLoader,
-    ) -> None:
+    ):
         """
         Generate worker in a new thread.
         :param flask_app: Flask app

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

@@ -1,11 +1,11 @@
 import logging
+import time
 from collections.abc import Mapping
 from typing import Any, Optional, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from configs import dify_config
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
@@ -23,16 +23,17 @@ from core.app.features.annotation_reply.annotation_reply import AnnotationReplyF
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
-from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
-from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.entities import GraphRuntimeState, VariablePool
+from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
-from models.workflow import ConversationVariable, WorkflowType
+from models.workflow import ConversationVariable
 
 logger = logging.getLogger(__name__)
 
@@ -54,7 +55,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         workflow: Workflow,
         system_user_id: str,
         app: App,
-    ) -> None:
+    ):
         super().__init__(
             queue_manager=queue_manager,
             variable_loader=variable_loader,
@@ -68,31 +69,39 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self.system_user_id = system_user_id
         self._app = app
 
-    def run(self) -> None:
+    def run(self):
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
-        app_record = db.session.query(App).where(App.id == app_config.app_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            app_record = session.scalar(select(App).where(App.id == app_config.app_id))
+
         if not app_record:
             raise ValueError("App not found")
 
-        workflow_callbacks: list[WorkflowCallback] = []
-        if dify_config.DEBUG:
-            workflow_callbacks.append(WorkflowLoggingCallback())
-
         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(
                 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,
             )
         else:
             inputs = self.application_generate_entity.inputs
@@ -144,16 +153,27 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             )
 
             # init graph
-            graph = self._init_graph(graph_config=self._workflow.graph_dict)
+            graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.time())
+            graph = self._init_graph(
+                graph_config=self._workflow.graph_dict,
+                graph_runtime_state=graph_runtime_state,
+                workflow_id=self._workflow.id,
+                tenant_id=self._workflow.tenant_id,
+                user_id=self.application_generate_entity.user_id,
+            )
 
         db.session.close()
 
         # RUN WORKFLOW
+        # Create Redis command channel for this workflow execution
+        task_id = self.application_generate_entity.task_id
+        channel_key = f"workflow:{task_id}:commands"
+        command_channel = RedisChannel(redis_client, channel_key)
+
         workflow_entry = WorkflowEntry(
             tenant_id=self._workflow.tenant_id,
             app_id=self._workflow.app_id,
             workflow_id=self._workflow.id,
-            workflow_type=WorkflowType.value_of(self._workflow.type),
             graph=graph,
             graph_config=self._workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
@@ -165,11 +185,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             invoke_from=self.application_generate_entity.invoke_from,
             call_depth=self.application_generate_entity.call_depth,
             variable_pool=variable_pool,
+            graph_runtime_state=graph_runtime_state,
+            command_channel=command_channel,
         )
 
-        generator = workflow_entry.run(
-            callbacks=workflow_callbacks,
-        )
+        generator = workflow_entry.run()
 
         for event in generator:
             self._handle_event(workflow_entry, event)
@@ -219,7 +239,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         return False
 
-    def _complete_with_stream_output(self, text: str, stopped_by: QueueStopEvent.StopBy) -> None:
+    def _complete_with_stream_output(self, text: str, stopped_by: QueueStopEvent.StopBy):
         """
         Direct output
         """

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

@@ -118,7 +118,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
+                response_chunk.update(sub_stream_response.to_ignore_detail_dict())  # ty: ignore [unresolved-attribute]
             else:
                 response_chunk.update(sub_stream_response.to_dict())