perf(web): load integration tools by category

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.agents/skills/e2e-cucumber-playwright/SKILL.md

@@ -32,12 +32,11 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
 - `e2e/` uses Cucumber for scenarios and Playwright as the browser layer.
 - `DifyWorld` is the per-scenario context object. Type `this` as `DifyWorld` and use `async function`, not arrow functions.
 - Keep glue organized by capability under `e2e/features/step-definitions/`; use `common/` only for broadly reusable steps.
-- Browser session behavior comes from `features/support/hooks.ts`:
-  - default: authenticated session with shared storage state
-  - `@unauthenticated`: clean browser context
-  - `@authenticated`: readability/selective-run tag only unless implementation changes
-  - `@fresh`: only for `e2e:full*` flows
+- Treat `e2e/AGENTS.md`, `features/support/hooks.ts`, and the Cucumber configuration as the owners of current session and tag semantics. Verify them when behavior depends on session state instead of copying a tag inventory into this skill.
 - Do not import Playwright Test runner patterns that bypass the current Cucumber + `DifyWorld` architecture unless the task is explicitly about changing that architecture.
+- Perform the behavior under test through Playwright. APIs are allowed for setup, seed preparation, persistence polling, and cleanup, but ordinary Console JSON and representable multipart operations must use the scenario- or process-owned generated oRPC client with request and response validation enabled. Keep the setup/cleanup API identity independent from an unauthenticated or logged-out behavior browser.
+- Consume generated operations directly. Do not add one-to-one API wrappers, handwritten endpoint URLs, response DTO casts, duplicate schemas, global mutable clients, or TanStack Query caching in Cucumber. Keep helpers only for real fixture construction, multi-operation orchestration, invariants, polling, derived test views, or protocol adapters.
+- Keep SSE, binary, redirect-only, external-service, and readiness exceptions centralized under their protocol owner. A contract mismatch must fail and be fixed at the backend schema owner followed by regeneration; never weaken validation to make E2E pass.
 
 ## Workflow
 
@@ -66,7 +65,7 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
    - If a product element has real user-facing semantics but no accessible name, prefer fixing that accessible contract over adding a test id.
 5. Validate narrowly.
    - Run the narrowest tagged scenario or flow that exercises the change.
-   - Run `vpr lint --fix --quiet` from the repository root and `pnpm -C e2e type-check`.
+   - Run the package-required static checks documented in `e2e/AGENTS.md`.
    - Broaden verification only when the change affects hooks, tags, setup, or shared step semantics.
 
 ## Review Checklist
@@ -77,6 +76,8 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
 - Are locators user-facing and assertions web-first?
 - Does the change introduce hidden coupling across scenarios, tags, or instance state?
 - Does it document or implement behavior that differs from the real hooks or configuration?
+- Does setup/cleanup use the generated client directly, with any remaining helper owning more than a one-to-one endpoint forward?
+- Is every raw HTTP call a documented protocol or infrastructure exception rather than an ordinary Console operation?
 
 Lead findings with correctness, flake risk, and architecture drift.
 

.agents/skills/frontend-testing/SKILL.md

@@ -9,12 +9,11 @@ Use this skill for Vitest work under `web/` and `packages/dify-ui/`. Do not use
 
 ## Required Source
 
-Before writing, changing, or reviewing frontend tests, read `web/docs/test.md` completely. It is the single source of truth. This skill defines the execution workflow and must not add requirements that conflict with or duplicate that guide.
+Before writing, changing, or reviewing frontend tests, read `web/docs/test.md` completely. It is the single source of truth. This skill provides an execution checklist and must not redefine or extend that policy.
 
 ## Workflow
 
 1. Read the source, its behavior owner, nearby specs, and relevant public dependencies.
-1. Identify whether the contract belongs in `web/`, Dify UI Browser Mode, or a styled Storybook test.
 1. Apply the canonical guide to decide whether a test is needed and choose its boundary.
 1. For a behavior change or bug fix, write or identify the failing scenario first when practical.
 1. Implement one coherent scenario at a time and run the focused spec before expanding scope.
@@ -33,4 +32,4 @@ vp test run path/to/spec-or-directory
 vp test run --project unit src/path/to/spec
 ```
 
-For styled Dify UI behavior, run `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.
+Run Dify UI Storybook tests with `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.

.github/CODEOWNERS

@@ -8,7 +8,6 @@
 
 # Lint bulk suppression baselines.
 /oxlint-suppressions.json
-/eslint-suppressions.json
 
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
@@ -33,31 +32,9 @@
 # Backend (default owner, more specific rules below will override)
 /api/ @QuantumGhost
 
-# Backend - MCP
-/api/core/mcp/ @Nov1c444
-/api/core/entities/mcp_provider.py @Nov1c444
-/api/services/tools/mcp_tools_manage_service.py @Nov1c444
-/api/controllers/mcp/ @Nov1c444
-/api/controllers/console/app/mcp_server.py @Nov1c444
-
 # Backend - Tests
 /api/tests/ @laipz8200 @QuantumGhost
 
-/api/tests/**/*mcp* @Nov1c444
-
-# Backend - Workflow - Engine (Core graph execution engine)
-/api/core/workflow/graph_engine/ @laipz8200 @QuantumGhost
-/api/core/workflow/runtime/ @laipz8200 @QuantumGhost
-/api/core/workflow/graph/ @laipz8200 @QuantumGhost
-/api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
-/api/core/workflow/node_events/ @laipz8200 @QuantumGhost
-
-# Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
-/api/core/workflow/nodes/agent/ @Nov1c444
-/api/core/workflow/nodes/iteration/ @Nov1c444
-/api/core/workflow/nodes/loop/ @Nov1c444
-/api/core/workflow/nodes/llm/ @Nov1c444
-
 # Backend - RAG (Retrieval Augmented Generation)
 /api/core/rag/ @JohnJyong
 /api/services/rag_pipeline/ @JohnJyong
@@ -111,7 +88,6 @@
 /api/core/app/layers/trigger_post_layer.py @CourTeous33
 /api/services/trigger/ @CourTeous33
 /api/models/trigger.py @CourTeous33
-/api/fields/workflow_trigger_fields.py @CourTeous33
 /api/repositories/workflow_trigger_log_repository.py @CourTeous33
 /api/repositories/sqlalchemy_workflow_trigger_log_repository.py @CourTeous33
 /api/libs/schedule_utils.py @CourTeous33
@@ -136,11 +112,11 @@
 /api/controllers/console/billing/ @hj24 @zyssyz123
 
 # Backend - Enterprise
-/api/configs/enterprise/ @GarfieldDai @GareArc
-/api/services/enterprise/ @GarfieldDai @GareArc
-/api/services/feature_service.py @GarfieldDai @GareArc
-/api/controllers/console/feature.py @GarfieldDai @GareArc
-/api/controllers/web/feature.py @GarfieldDai @GareArc
+/api/configs/enterprise/ @GareArc
+/api/services/enterprise/ @GareArc
+/api/services/feature_service.py @GareArc
+/api/controllers/console/feature.py @GareArc
+/api/controllers/web/feature.py @GareArc
 
 # Backend - Database Migrations
 /api/migrations/ @snakevash @laipz8200 @MRZHUH
@@ -153,7 +129,6 @@
 
 # Frontend - Platform and Features
 /web/config/ @lyzno1
-/web/contract/ @lyzno1
 /web/env.ts @lyzno1
 /web/features/ @lyzno1
 /web/hooks/ @lyzno1
@@ -212,7 +187,6 @@
 /web/app/components/rag-pipeline/store/ @iamjoel @zxhlyh
 
 # Frontend - RAG - Documents List
-/web/app/components/datasets/documents/list.tsx @iamjoel @WTW0313
 /web/app/components/datasets/documents/create-from-pipeline/ @iamjoel @WTW0313
 
 # Frontend - RAG - Segments List
@@ -231,22 +205,22 @@
 /web/app/components/plugins/marketplace/ @iamjoel @Yessenia-d
 
 # Frontend - Login and Registration
-/web/app/signin/ @douxc @iamjoel
-/web/app/signup/ @douxc @iamjoel
-/web/app/reset-password/ @douxc @iamjoel
-/web/app/install/ @douxc @iamjoel
-/web/app/init/ @douxc @iamjoel
-/web/app/forgot-password/ @douxc @iamjoel
-/web/app/account/ @douxc @iamjoel
+/web/app/signin/ @iamjoel
+/web/app/signup/ @iamjoel
+/web/app/reset-password/ @iamjoel
+/web/app/install/ @iamjoel
+/web/app/init/ @iamjoel
+/web/app/forgot-password/ @iamjoel
+/web/app/account/ @iamjoel
 
 # Frontend - Service Authentication
-/web/service/base.ts @douxc @iamjoel
+/web/service/base.ts @iamjoel
 
 # Frontend - WebApp Authentication and Access Control
-/web/app/(shareLayout)/components/ @douxc @iamjoel
-/web/app/(shareLayout)/webapp-signin/ @douxc @iamjoel
-/web/app/(shareLayout)/webapp-reset-password/ @douxc @iamjoel
-/web/app/components/app/app-access-control/ @douxc @iamjoel
+/web/app/(shareLayout)/components/ @iamjoel
+/web/app/(shareLayout)/webapp-signin/ @iamjoel
+/web/app/(shareLayout)/webapp-reset-password/ @iamjoel
+/web/app/components/app/app-access-control/ @iamjoel
 
 # Frontend - Explore Page
 /web/app/components/explore/ @CodingOnStar @iamjoel
@@ -265,7 +239,6 @@
 /web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar
 
 # Frontend - Utils and Hooks
-/web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh
 /web/utils/format.ts @iamjoel @zxhlyh
 /web/utils/clipboard.ts @iamjoel @zxhlyh
@@ -277,5 +250,22 @@
 # Frontend - Workspace
 /web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh
 
+# Frontend - App Shell and Console Bootstrap
+/web/app/layout.tsx @iamjoel @lyzno1
+/web/app/error.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/layout.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/providers.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/hydration-boundary.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/profile-bootstrap-gate.tsx @iamjoel @lyzno1
+/web/app/(commonLayout)/error.tsx @iamjoel @lyzno1
+/web/app/account/(commonLayout)/layout.tsx @iamjoel @lyzno1
+/web/app/components/main-nav/* @iamjoel @lyzno1
+/web/app/components/main-nav/components/* @iamjoel @lyzno1
+/web/context/query-client.tsx @iamjoel @lyzno1
+/web/context/query-client-server.ts @iamjoel @lyzno1
+/web/proxy.ts @iamjoel @lyzno1
+/web/app/auth/refresh/route.ts @iamjoel @lyzno1
+/web/service/server.ts @iamjoel @lyzno1
+
 # Docker
 /docker/* @laipz8200

.github/workflows/api-tests.yml

@@ -47,9 +47,6 @@ jobs:
       - name: Install dependencies
         run: uv sync --project api --dev
 
-      - name: Run dify config tests
-        run: uv run --project api pytest api/tests/unit_tests/configs/test_env_consistency.py
-
       - name: Run Unit Tests
         run: |
           uv run --project api pytest \

.github/workflows/deploy-knowledge.yml

@@ -0,0 +1,28 @@
+name: Deploy Knowledge
+
+permissions:
+  contents: read
+
+on:
+  workflow_run:
+    workflows: ["Build and Push API & Web"]
+    branches:
+      - "deploy/konwledge"
+    types:
+      - completed
+
+jobs:
+  deploy:
+    runs-on: depot-ubuntu-24.04
+    if: |
+      github.event.workflow_run.conclusion == 'success' &&
+      github.event.workflow_run.head_branch == 'deploy/konwledge'
+    steps:
+      - name: Deploy to server
+        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
+        with:
+          host: ${{ secrets.SSH_NEW_RAG_HOST }}
+          username: ${{ secrets.SSH_USER }}
+          key: ${{ secrets.SSH_PRIVATE_KEY }}
+          script: |
+            ${{ vars.SSH_SCRIPT || secrets.SSH_SCRIPT }}

.github/workflows/main-ci.yml

@@ -27,7 +27,7 @@ jobs:
     steps:
       - id: skip_check
         continue-on-error: true
-        uses: fkirc/skip-duplicate-actions@f75f66ce1886f00957d99748a42c724f4330bdcf # v5.3.1
+        uses: fkirc/skip-duplicate-actions@b974a9395958c231af965b70070979a577efa578 # v5.3.2
         with:
           cancel_others: 'true'
           concurrent_skipping: same_content_newer
@@ -81,7 +81,6 @@ jobs:
               - '.npmrc'
               - '.nvmrc'
               - '.github/workflows/cli-tests.yml'
-              - '.github/workflows/cli-docker-build.yml'
               - '.github/actions/setup-web/**'
             web:
               - 'web/**'
@@ -336,6 +335,8 @@ jobs:
       - check-changes
     if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
     uses: ./.github/workflows/web-e2e.yml
+    with:
+      run-external-runtime: false
     secrets: inherit
 
   web-e2e-skip:

.github/workflows/post-merge.yml

@@ -26,19 +26,30 @@ jobs:
             external_e2e:
               - 'e2e/features/agent-v2/**'
               - 'e2e/features/step-definitions/agent-v2/**'
+              - 'e2e/features/step-definitions/common/**'
               - 'e2e/features/support/**'
+              - 'e2e/fixtures/auth.ts'
               - 'e2e/fixtures/test-materials/**'
               - 'e2e/scripts/**'
               - 'e2e/support/**'
               - 'e2e/cucumber.config.ts'
               - 'e2e/package.json'
               - 'e2e/test-env.ts'
+              - 'e2e/tsconfig.json'
               - 'e2e/tsx-register.js'
+              - 'package.json'
+              - 'pnpm-lock.yaml'
+              - '.nvmrc'
               - '.github/workflows/post-merge.yml'
               - '.github/workflows/web-e2e.yml'
               - '.github/actions/setup-web/**'
+              - 'docker/docker-compose.middleware.yaml'
+              - 'docker/envs/middleware.env.example'
               - 'dify-agent/**'
               - 'dify-agent-runtime/**'
+              - 'api/pyproject.toml'
+              - 'api/uv.lock'
+              - 'api/tests/integration_tests/.env.example'
               - 'api/clients/agent_backend/**'
               - 'api/core/app/apps/agent_app/**'
               - 'api/core/workflow/nodes/agent_v2/**'
@@ -48,8 +59,13 @@ jobs:
               - 'api/services/plugin/**'
               - 'api/core/tools/**'
               - 'api/services/tools/**'
+              - 'packages/contracts/package.json'
               - 'packages/contracts/generated/api/console/agent/**'
+              - 'packages/contracts/generated/api/console/apps/**'
+              - 'packages/contracts/generated/api/console/datasets/**'
               - 'packages/contracts/generated/api/console/orpc.gen.ts'
+              - 'packages/contracts/generated/api/console/workspaces/**'
+              - 'packages/contracts/generated/api/service/**'
               - 'web/features/agent-v2/**'
               - 'web/app/(commonLayout)/agents/**'
               - 'web/app/(commonLayout)/@detailSidebar/agents/**'

.github/workflows/sandbox-runtime-tests.yml

@@ -27,7 +27,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
         with:
           go-version-file: dify-agent-runtime/go.mod
           cache-dependency-path: dify-agent-runtime/go.sum
@@ -51,13 +51,13 @@ jobs:
           persist-credentials: false
 
       - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
         with:
           go-version-file: dify-agent-runtime/go.mod
           cache-dependency-path: dify-agent-runtime/go.sum
 
       - name: Run golangci-lint
-        uses: golangci/golangci-lint-action@4afd733a84b1f43292c63897423277bb7f4313a9 # v6.5.0
+        uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v6.5.0
         with:
           working-directory: dify-agent-runtime
           version: latest
@@ -78,7 +78,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
         with:
           go-version-file: dify-agent-runtime/go.mod
           cache-dependency-path: dify-agent-runtime/go.sum

.github/workflows/tool-test-sdks.yaml

@@ -29,7 +29,7 @@ jobs:
           persist-credentials: false
 
       - name: Use Node.js
-        uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
+        uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
         with:
           node-version: 22
           cache: ''

.github/workflows/translate-i18n-claude.yml

@@ -158,7 +158,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@e90deca47693f9457b72f2b53c17d7c445a87342 # v1.0.171
+        uses: anthropics/claude-code-action@af0559ee4f514d1ef21826982bed13f7edc3c35e # v1.0.178
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/web-e2e.yml

@@ -4,9 +4,9 @@ on:
   workflow_call:
     inputs:
       run-external-runtime:
-        required: false
+        description: Run only the prepared and external runtime suite instead of the core suites.
+        required: true
         type: boolean
-        default: false
 
 permissions:
   contents: read
@@ -46,6 +46,7 @@ jobs:
         run: uv sync --project api --dev
 
       - name: Run E2E support unit tests
+        if: ${{ !inputs.run-external-runtime }}
         working-directory: ./e2e
         run: vp run test:unit
 
@@ -54,6 +55,7 @@ jobs:
         run: vp run e2e:install
 
       - name: Run isolated source-api and built-web Cucumber E2E tests
+        if: ${{ !inputs.run-external-runtime }}
         working-directory: ./e2e
         env:
           E2E_ADMIN_EMAIL: e2e-admin@example.com
@@ -64,7 +66,7 @@ jobs:
         run: vp run e2e:full
 
       - name: Preserve Chromium E2E report and logs
-        if: ${{ !cancelled() }}
+        if: ${{ !cancelled() && !inputs.run-external-runtime }}
         run: |
           if [[ -d e2e/cucumber-report ]]; then
             mv e2e/cucumber-report e2e/cucumber-report-non-external
@@ -74,6 +76,7 @@ jobs:
           fi
 
       - name: Run WebKit keyboard and browser smoke tests
+        if: ${{ !inputs.run-external-runtime }}
         working-directory: ./e2e
         env:
           E2E_ADMIN_EMAIL: e2e-admin@example.com
@@ -99,7 +102,7 @@ jobs:
           vp run e2e -- --tags '@browser-smoke'
 
       - name: Preserve WebKit E2E report and logs
-        if: ${{ !cancelled() }}
+        if: ${{ !cancelled() && !inputs.run-external-runtime }}
         run: |
           if [[ -d e2e/cucumber-report ]]; then
             mv e2e/cucumber-report e2e/cucumber-report-webkit
@@ -135,16 +138,6 @@ jobs:
             exit 1
           fi
 
-          if [[ -d cucumber-report ]]; then
-            rm -rf cucumber-report-non-external
-            mv cucumber-report cucumber-report-non-external
-          fi
-
-          if [[ -d .logs ]]; then
-            rm -rf .logs-non-external
-            mv .logs .logs-non-external
-          fi
-
           teardown_external_runtime() {
             local run_status=$?
             trap - EXIT

.github/workflows/web-tests.yml

@@ -17,8 +17,6 @@ jobs:
   test:
     name: Web Tests (${{ matrix.shardIndex }}/${{ matrix.shardTotal }})
     runs-on: depot-ubuntu-24.04-4
-    env:
-      VITEST_COVERAGE_SCOPE: app-components
     strategy:
       fail-fast: false
       matrix:

Makefile

@@ -107,6 +107,7 @@ test:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
+		set -e; \
 		echo "Running backend unit tests"; \
 		uv run --project api --dev pytest -p no:benchmark --timeout "$${PYTEST_TIMEOUT:-20}" -n auto \
 			api/tests/unit_tests \
@@ -124,6 +125,7 @@ test-all:
 		echo "Target: $(TARGET_TESTS)"; \
 		uv run --project api --dev pytest $(TARGET_TESTS); \
 	else \
+		set -e; \
 		echo "Running backend unit tests"; \
 		uv run --project api --dev pytest -p no:benchmark --timeout "$${PYTEST_TIMEOUT:-20}" -n auto \
 			api/tests/unit_tests \

README.md

@@ -71,7 +71,7 @@ Dify is an open-source LLM app development platform. Its intuitive interface com
 
 <br/>
 
-The easiest way to start the Dify server is through [Docker Compose](docker/docker-compose.yaml). Before running Dify with the following commands, make sure that [Docker](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/install/) are installed on your machine:
+The easiest way to start the Dify server is through [Docker Compose](docker/docker-compose.yaml). Before running Dify with the following commands, make sure that [Docker](https://docs.docker.com/get-docker/) and Docker Compose v2.24.0 or later are installed on your machine:
 
 ```bash
 cd dify

api/.env.example

@@ -677,6 +677,17 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y
 
 # Dify Agent backend
 AGENT_BACKEND_BASE_URL=http://localhost:5050
+AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS=30
+AGENT_BACKEND_STREAM_MAX_RECONNECTS=3
+AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
+
+# KnowledgeFS (Dataset 2.0)
+KNOWLEDGE_FS_ENABLED=false
+KNOWLEDGE_FS_BASE_URL=
+# Shared with KnowledgeFS; use at least 32 random characters.
+KNOWLEDGE_FS_JWT_SECRET=
+KNOWLEDGE_FS_SSE_READ_TIMEOUT_SECONDS=300
+KNOWLEDGE_FS_TIMEOUT_SECONDS=10
 
 # Marketplace configuration
 MARKETPLACE_ENABLED=true

api/clients/agent_backend/client.py

@@ -8,7 +8,7 @@ creating another wire contract.
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from typing import Protocol
 
 from dify_agent.client import (
@@ -45,7 +45,13 @@ class AgentBackendRunClient(Protocol):
     def cancel_run(self, run_id: str, request: CancelRunRequest | None = None) -> CancelRunResponse:
         """Request explicit cancellation for one Agent backend run."""
 
-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
         """Yield public ``dify-agent`` run events in stream order."""
 
     def wait_run(self, run_id: str, *, timeout_seconds: float | None = None) -> RunStatusResponse:
@@ -61,7 +67,15 @@ class _DifyAgentSyncClient(Protocol):
     def cancel_run_sync(self, run_id: str, request: CancelRunRequest | None = None) -> CancelRunResponse:
         """Cancel one run synchronously."""
 
-    def stream_events_sync(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events_sync(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        max_reconnects: int | None = None,
+        timeout_seconds: float | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
         """Stream run events synchronously."""
 
     def wait_run_sync(self, run_id: str, *, timeout_seconds: float | None = None) -> RunStatusResponse:
@@ -73,8 +87,16 @@ class DifyAgentBackendRunClient:
 
     client: _DifyAgentSyncClient
 
-    def __init__(self, client: _DifyAgentSyncClient) -> None:
+    def __init__(
+        self,
+        client: _DifyAgentSyncClient,
+        *,
+        stream_max_reconnects: int = 3,
+        stream_timeout_seconds: float = 1200,
+    ) -> None:
         self.client = client
+        self._stream_max_reconnects = stream_max_reconnects
+        self._stream_timeout_seconds = stream_timeout_seconds
 
     def create_run(self, request: CreateRunRequest) -> CreateRunResponse:
         """Create one run through ``POST /runs`` and normalize client exceptions."""
@@ -90,10 +112,22 @@ class DifyAgentBackendRunClient:
         except Exception as exc:
             raise _normalize_dify_agent_error(exc) from exc
 
-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
         """Stream run events from ``/events/sse`` with the wrapped client's reconnect policy."""
         try:
-            yield from self.client.stream_events_sync(run_id, after=after)
+            yield from self.client.stream_events_sync(
+                run_id,
+                after=after,
+                max_reconnects=self._stream_max_reconnects,
+                timeout_seconds=self._stream_timeout_seconds,
+                should_stop=should_stop,
+            )
         except Exception as exc:
             raise _normalize_dify_agent_error(exc) from exc
 

api/clients/agent_backend/factory.py

@@ -13,10 +13,17 @@ def create_agent_backend_run_client(
     base_url: str | None = None,
     use_fake: bool = False,
     fake_scenario: str | FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCESS,
+    stream_read_timeout_seconds: float = 30,
+    stream_max_reconnects: int = 3,
+    stream_run_timeout_seconds: float = 1200,
 ) -> AgentBackendRunClient:
     """Create the API-side run client without hiding the ``dify-agent`` protocol."""
     if use_fake:
         return FakeAgentBackendRunClient(scenario=FakeAgentBackendScenario(fake_scenario))
     if base_url is None:
         raise ValueError("base_url is required when creating a real Agent backend client")
-    return DifyAgentBackendRunClient(Client(base_url=base_url))
+    return DifyAgentBackendRunClient(
+        Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds),
+        stream_max_reconnects=stream_max_reconnects,
+        stream_timeout_seconds=stream_run_timeout_seconds,
+    )

api/clients/agent_backend/fake_client.py

@@ -7,7 +7,7 @@ separate ``agent-backend.v1`` event stream.
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from datetime import UTC, datetime
 from enum import StrEnum
 
@@ -69,9 +69,17 @@ class FakeAgentBackendRunClient:
         del request
         return CancelRunResponse(run_id=run_id, status="cancelled")
 
-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
         """Yield the deterministic public ``RunEvent`` sequence for ``run_id``."""
         for event in self._events(run_id):
+            if should_stop is not None and should_stop():
+                return
             if after is not None and event.id is not None and event.id <= after:
                 continue
             yield event

api/commands/retention.py

@@ -1,6 +1,8 @@
 import datetime
 import logging
+import re
 import time
+import uuid
 from collections.abc import Callable
 from typing import TypedDict
 
@@ -21,6 +23,7 @@ from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 logger = logging.getLogger(__name__)
 
 _HEX_PREFIXES = tuple("0123456789abcdef")
+_TARGET_MONTH_PATTERN = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
 
 
 class WorkflowRunArchivePlanRow(TypedDict):
@@ -66,6 +69,7 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
 
 
 def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[str] | None:
+    """Keep an omitted scope unset while rejecting an explicitly empty scope."""
     if raw_ids is None:
         return None
     parsed = sorted({raw_id.strip() for raw_id in raw_ids.split(",") if raw_id.strip()})
@@ -74,6 +78,27 @@ def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[
     return parsed
 
 
+def _parse_archive_target_month(target_month: str) -> tuple[int, int]:
+    """Validate the V2 catalog month selector and return its numeric components."""
+    if not _TARGET_MONTH_PATTERN.fullmatch(target_month):
+        raise click.BadParameter("target-month must use YYYY-MM format", param_hint="--target-month")
+    year_text, month_text = target_month.split("-", maxsplit=1)
+    return int(year_text), int(month_text)
+
+
+def _parse_archive_catalog_cursor(after_catalog_id: str | None) -> str | None:
+    """Normalize the exclusive V2 catalog keyset cursor when one is provided."""
+    if after_catalog_id is None:
+        return None
+    try:
+        return str(uuid.UUID(after_catalog_id))
+    except ValueError as exc:
+        raise click.BadParameter(
+            "after-catalog-id must be a UUID returned by the same V2 operation and scope",
+            param_hint="--after-catalog-id",
+        ) from exc
+
+
 def _get_archive_candidate_tenant_ids_by_prefix(
     session: Session,
     prefix: str,
@@ -810,9 +835,11 @@ def backfill_workflow_run_archive_bundles(
         click.echo(click.style(f"  ... and {len(summary.errors) - 10} more failures", fg="red"))
 
 
-def _echo_bundle_archive_operation_summary(summary) -> None:
+def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> None:
     status = "completed successfully" if summary.bundles_failed == 0 else "completed with failures"
     fg = "green" if summary.bundles_failed == 0 else "red"
+    cursor_label = "preview_next_catalog_id" if dry_run else "next_catalog_id"
+    cursor_value = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
     click.echo(
         click.style(
             f"{summary.operation} {status}. "
@@ -821,10 +848,12 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
             f"archive_bytes={summary.archive_bytes} duration={summary.elapsed_time:.2f}s "
             f"validation_time={summary.validation_time:.2f}s "
             f"runs_per_second={summary.runs_per_second:.2f} rows_per_second={summary.rows_per_second:.2f} "
-            f"bytes_per_second={summary.bytes_per_second:.2f}",
+            f"bytes_per_second={summary.bytes_per_second:.2f} {cursor_label}={cursor_value or 'none'}",
             fg=fg,
         )
     )
+    if dry_run:
+        click.echo(click.style("Dry-run cursor is preview-only; do not persist it for a destructive run.", fg="yellow"))
     click.echo(click.style("table,row_count", fg="white"))
     for table_name in [
         "workflow_runs",
@@ -842,7 +871,8 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
                 click.style(
                     f"  bundle={result.bundle_id} tenant={result.tenant_id} runs={result.run_count} "
                     f"rows={result.row_count} archive_bytes={result.archive_bytes} "
-                    f"time={result.elapsed_time:.2f}s validation={result.validation_time:.2f}s",
+                    f"catalog_id={result.catalog_id} time={result.elapsed_time:.2f}s "
+                    f"validation={result.validation_time:.2f}s",
                     fg="white",
                 )
             )
@@ -850,7 +880,7 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
             click.echo(
                 click.style(
                     f"  failed bundle={result.bundle_id} tenant={result.tenant_id} "
-                    f"object_prefix={result.object_prefix} error={result.error}",
+                    f"catalog_id={result.catalog_id} object_prefix={result.object_prefix} error={result.error}",
                     fg="red",
                 )
             )
@@ -867,25 +897,24 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
 )
 @click.option("--run-id", required=False, help="Workflow run ID to restore.")
 @click.option(
-    "--start-from",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--target-month",
+    metavar="YYYY-MM",
     default=None,
-    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
+    help="V2 catalog month to restore; required unless --run-id is used.",
 )
 @click.option(
-    "--end-before",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--after-catalog-id",
     default=None,
-    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
+    help="Exclusive V2 cursor from the same restore month and tenant scope.",
 )
 @click.option("--workers", default=1, show_default=True, type=int, help="V1 --run-id compatibility only.")
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to restore.")
+@click.option("--limit", type=click.IntRange(min=1), default=100, show_default=True, help="Maximum V2 catalog rows.")
 @click.option("--dry-run", is_flag=True, help="Preview without restoring.")
 def restore_workflow_runs(
     tenant_ids: str | None,
     run_id: str | None,
-    start_from: datetime.datetime | None,
-    end_before: datetime.datetime | None,
+    target_month: str | None,
+    after_catalog_id: str | None,
     workers: int,
     limit: int,
     dry_run: bool,
@@ -905,23 +934,20 @@ def restore_workflow_runs(
     from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
     from services.retention.workflow_run.restore_archived_workflow_run import WorkflowRunRestore
 
-    parsed_tenant_ids = None
-    if tenant_ids:
-        parsed_tenant_ids = [tid.strip() for tid in tenant_ids.split(",") if tid.strip()]
-        if not parsed_tenant_ids:
-            raise click.BadParameter("tenant-ids must not be empty")
+    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
 
-    if (start_from is None) ^ (end_before is None):
-        raise click.UsageError("--start-from and --end-before must be provided together.")
-    if run_id is None and (start_from is None or end_before is None):
-        raise click.UsageError("--start-from and --end-before are required for batch restore.")
     if workers < 1:
         raise click.BadParameter("workers must be at least 1")
+    if run_id is not None and (target_month is not None or after_catalog_id is not None):
+        raise click.UsageError("--target-month and --after-catalog-id are only valid for V2 batch restore.")
+    if run_id is None and target_month is None:
+        raise click.UsageError("--target-month is required for V2 batch restore.")
 
     start_time = datetime.datetime.now(datetime.UTC)
+    target_desc = f"workflow run {run_id}" if run_id else f"workflow archive catalog month {target_month}"
     click.echo(
         click.style(
-            f"Starting restore of workflow run {run_id} at {start_time.isoformat()}.",
+            f"Starting restore of {target_desc} at {start_time.isoformat()}.",
             fg="white",
         )
     )
@@ -955,17 +981,20 @@ def restore_workflow_runs(
         click.echo(
             click.style("--workers is ignored for V2 bundle restore; bundles are processed serially.", fg="yellow")
         )
-    assert start_from is not None
-    assert end_before is not None
+    assert target_month is not None
+    target_year, target_month_number = _parse_archive_target_month(target_month)
+    catalog_cursor = _parse_archive_catalog_cursor(after_catalog_id)
     bundle_restorer = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
     summary = bundle_restorer.restore_batch(
         tenant_ids=parsed_tenant_ids,
-        start_date=start_from,
-        end_date=end_before,
+        target_year=target_year,
+        target_month=target_month_number,
+        after_catalog_id=catalog_cursor,
         limit=limit,
     )
-    _echo_bundle_archive_operation_summary(summary)
-    return
+    _echo_bundle_archive_operation_summary(summary, dry_run=dry_run)
+    if summary.bundles_failed:
+        raise click.exceptions.Exit(1)
 
 
 @click.command(
@@ -979,23 +1008,41 @@ def restore_workflow_runs(
 )
 @click.option("--run-id", required=False, help="Workflow run ID to delete.")
 @click.option(
-    "--start-from",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--target-month",
+    metavar="YYYY-MM",
     default=None,
-    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
+    help="V2 catalog month to delete; required unless --run-id is used.",
 )
 @click.option(
-    "--end-before",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--after-catalog-id",
     default=None,
-    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
+    help="Exclusive V2 cursor from the same delete month and tenant scope.",
+)
+@click.option(
+    "--run-shard-index",
+    default=None,
+    type=click.IntRange(min=0),
+    help="Zero-based archive shard index. Must be paired with --run-shard-total.",
+)
+@click.option(
+    "--run-shard-total",
+    default=None,
+    type=click.IntRange(min=1, max=16),
+    help="Total archive shard count. Must be paired with --run-shard-index.",
+)
+@click.option("--all-pages", is_flag=True, help="Process catalog pages until an empty page is reached.")
+@click.option(
+    "--limit",
+    type=click.IntRange(min=1),
+    default=100,
+    show_default=True,
+    help="Maximum V2 catalog rows per page.",
 )
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to delete.")
 @click.option("--dry-run", is_flag=True, help="Preview without deleting.")
 @click.option(
     "--skip-bad-archives",
     is_flag=True,
-    help="Continue batch deletion when one archive object fails validation.",
+    help="V1 --run-id only: continue when one archive object fails validation.",
 )
 @click.option(
     "--restore-sample-interval",
@@ -1007,8 +1054,11 @@ def restore_workflow_runs(
 def delete_archived_workflow_runs(
     tenant_ids: str | None,
     run_id: str | None,
-    start_from: datetime.datetime | None,
-    end_before: datetime.datetime | None,
+    target_month: str | None,
+    after_catalog_id: str | None,
+    run_shard_index: int | None,
+    run_shard_total: int | None,
+    all_pages: bool,
     limit: int,
     dry_run: bool,
     skip_bad_archives: bool,
@@ -1018,26 +1068,38 @@ def delete_archived_workflow_runs(
     Delete archived workflow runs from the database.
 
     Batch delete uses V2 bundle metadata and validates object existence, manifest schema, object size, checksum, row
-    counts, and source/archive content checksums before deleting source rows. `--run-id` keeps the V1 per-run path.
+    counts, and source/archive content checksums before deleting source rows. Parallel workers may select one exact
+    archive shard; all-pages mode keeps only the current bounded page in memory. `--run-id` keeps the V1 per-run path.
     """
     from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
     from services.retention.workflow_run.delete_archived_workflow_run import ArchivedWorkflowRunDeletion
 
-    parsed_tenant_ids = None
-    if tenant_ids:
-        parsed_tenant_ids = [tid.strip() for tid in tenant_ids.split(",") if tid.strip()]
-        if not parsed_tenant_ids:
-            raise click.BadParameter("tenant-ids must not be empty")
+    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
 
-    if (start_from is None) ^ (end_before is None):
-        raise click.UsageError("--start-from and --end-before must be provided together.")
-    if run_id is None and (start_from is None or end_before is None):
-        raise click.UsageError("--start-from and --end-before are required for batch delete.")
     if restore_sample_interval < 0:
         raise click.BadParameter("restore-sample-interval must be >= 0")
+    if run_id is not None and (
+        target_month is not None
+        or after_catalog_id is not None
+        or run_shard_index is not None
+        or run_shard_total is not None
+        or all_pages
+    ):
+        raise click.UsageError(
+            "--target-month, --after-catalog-id, --run-shard-index, --run-shard-total, and --all-pages "
+            "are only valid for V2 batch delete."
+        )
+    if run_id is None and target_month is None:
+        raise click.UsageError("--target-month is required for V2 batch delete.")
+    if run_id is None and skip_bad_archives:
+        raise click.UsageError("--skip-bad-archives is not supported for V2 catalog batches; they fail fast.")
+    if (run_shard_index is None) ^ (run_shard_total is None):
+        raise click.UsageError("--run-shard-index and --run-shard-total must be provided together.")
+    if run_shard_index is not None and run_shard_total is not None and run_shard_index >= run_shard_total:
+        raise click.UsageError("--run-shard-index must be less than --run-shard-total.")
 
     start_time = datetime.datetime.now(datetime.UTC)
-    target_desc = f"workflow run {run_id}" if run_id else "workflow runs"
+    target_desc = f"workflow run {run_id}" if run_id else f"workflow archive catalog month {target_month}"
     click.echo(
         click.style(
             f"Starting delete of {target_desc} at {start_time.isoformat()}.",
@@ -1110,20 +1172,104 @@ def delete_archived_workflow_runs(
 
     if restore_sample_interval:
         click.echo(click.style("--restore-sample-interval is ignored for V2 bundle delete.", fg="yellow"))
-    assert start_from is not None
-    assert end_before is not None
-    bundle_deleter = WorkflowRunBundleArchiveMaintenance(
-        dry_run=dry_run,
-        strict_content_validation=True,
-        stop_on_error=not skip_bad_archives,
+    assert target_month is not None
+    target_year, target_month_number = _parse_archive_target_month(target_month)
+    catalog_cursor = _parse_archive_catalog_cursor(after_catalog_id)
+    shard = (
+        f"{run_shard_index:02d}-of-{run_shard_total:02d}"
+        if run_shard_index is not None and run_shard_total is not None
+        else None
     )
-    summary = bundle_deleter.delete_batch(
-        tenant_ids=parsed_tenant_ids,
-        start_date=start_from,
-        end_date=end_before,
-        limit=limit,
-    )
-    _echo_bundle_archive_operation_summary(summary)
+    bundle_deleter = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
+    if run_shard_total is not None:
+        try:
+            bundle_deleter.validate_catalog_shards(
+                target_year=target_year,
+                target_month=target_month_number,
+                shard_total=run_shard_total,
+                tenant_ids=parsed_tenant_ids,
+            )
+        except ValueError as exc:
+            logger.exception(
+                "Archive catalog shard preflight failed: target_month=%s shard=%s",
+                target_month,
+                shard,
+            )
+            raise click.ClickException(
+                f"Archive catalog shard preflight failed for target_month={target_month} shard={shard}: {exc}"
+            ) from exc
+
+    initial_catalog_cursor = catalog_cursor
+    pages_processed = 0
+    bundles_succeeded = 0
+    runs_processed = 0
+    rows_processed = 0
+    archive_bytes = 0
+    while True:
+        summary = bundle_deleter.delete_batch(
+            tenant_ids=parsed_tenant_ids,
+            target_year=target_year,
+            target_month=target_month_number,
+            after_catalog_id=catalog_cursor,
+            limit=limit,
+            shard=shard,
+        )
+        _echo_bundle_archive_operation_summary(summary, dry_run=dry_run)
+        if summary.bundles_failed:
+            failed_result = next((result for result in summary.results if not result.success), None)
+            failed_catalog_id = failed_result.catalog_id if failed_result is not None else "unknown"
+            page_resume_cursor = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
+            resume_cursor = page_resume_cursor or catalog_cursor
+            if dry_run:
+                cursor_details = (
+                    f"preview_after_catalog_id={resume_cursor or 'none'} "
+                    f"destructive_retry_after_catalog_id={initial_catalog_cursor or 'none'}"
+                )
+            else:
+                cursor_details = f"resume_after_catalog_id={resume_cursor or 'none'}"
+            click.echo(
+                click.style(
+                    f"Delete stopped: target_month={target_month} shard={shard or 'all'} "
+                    f"failed_catalog_id={failed_catalog_id} "
+                    f"{cursor_details}",
+                    fg="red",
+                )
+            )
+            raise click.exceptions.Exit(1)
+
+        if not all_pages:
+            break
+        if summary.bundles_processed == 0:
+            break
+
+        pages_processed += 1
+        bundles_succeeded += summary.bundles_succeeded
+        runs_processed += summary.runs_processed
+        rows_processed += summary.rows_processed
+        archive_bytes += summary.archive_bytes
+        next_catalog_id = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
+        if next_catalog_id is None or (catalog_cursor is not None and next_catalog_id <= catalog_cursor):
+            click.echo(
+                click.style(
+                    f"Delete cursor did not advance: target_month={target_month} shard={shard or 'all'} "
+                    f"after_catalog_id={catalog_cursor or 'none'} next_catalog_id={next_catalog_id or 'none'}",
+                    fg="red",
+                )
+            )
+            raise click.exceptions.Exit(1)
+        catalog_cursor = next_catalog_id
+
+    if all_pages:
+        final_cursor_label = "preview_final_catalog_id" if dry_run else "final_catalog_id"
+        click.echo(
+            click.style(
+                f"Delete all-pages completed successfully. target_month={target_month} shard={shard or 'all'} "
+                f"pages={pages_processed} bundles_success={bundles_succeeded} runs={runs_processed} "
+                f"rows={rows_processed} archive_bytes={archive_bytes} "
+                f"{final_cursor_label}={catalog_cursor or 'none'}",
+                fg="green",
+            )
+        )
 
 
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:

api/commands/system.py

@@ -6,6 +6,7 @@ from sqlalchemy import delete, select, update
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
+from enums.deployment_edition import DeploymentEdition
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -42,7 +43,7 @@ def reset_encrypt_key_pair():
     After the reset, all LLM credentials will become invalid, requiring re-entry.
     Only support SELF_HOSTED mode.
     """
-    if dify_config.EDITION != "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
         click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
         return
     with sessionmaker(db.engine, expire_on_commit=False).begin() as session:

api/configs/app_config.py

@@ -5,6 +5,7 @@ from typing import Any, override
 from pydantic.fields import FieldInfo
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
 
+from enums.deployment_edition import DeploymentEdition
 from libs.file_utils import search_file_upwards
 
 from .deploy import DeploymentConfig
@@ -116,3 +117,11 @@ class DifyConfig(
                 ),
             ),
         )
+
+    @property
+    def DEPLOYMENT_EDITION(self) -> DeploymentEdition:
+        if self.EDITION == "CLOUD":
+            return DeploymentEdition.CLOUD
+        if self.ENTERPRISE_ENABLED:
+            return DeploymentEdition.ENTERPRISE
+        return DeploymentEdition.COMMUNITY

api/configs/enterprise/__init__.py

@@ -14,6 +14,12 @@ class EnterpriseFeatureConfig(BaseSettings):
         default=False,
     )
 
+    WEBAPP_PUBLIC_ACCESS_ENABLED: bool = Field(
+        description="Whether admins are allowed to set a webapp's access mode to public (anyone with the link, "
+        "no auth). Disable in security-sensitive on-prem deployments.",
+        default=True,
+    )
+
     CAN_REPLACE_LOGO: bool = Field(
         description="Allow customization of the enterprise logo.",
         default=False,

api/configs/extra/__init__.py

@@ -1,5 +1,6 @@
 from configs.extra.agent_backend_config import AgentBackendConfig
 from configs.extra.archive_config import ArchiveStorageConfig
+from configs.extra.knowledge_fs_config import KnowledgeFSConfig
 from configs.extra.notion_config import NotionConfig
 from configs.extra.sentry_config import SentryConfig
 
@@ -8,6 +9,7 @@ class ExtraServiceConfig(
     # place the configs in alphabet order
     AgentBackendConfig,
     ArchiveStorageConfig,
+    KnowledgeFSConfig,
     NotionConfig,
     SentryConfig,
 ):

api/configs/extra/agent_backend_config.py

@@ -1,4 +1,4 @@
-from pydantic import Field, NonNegativeFloat
+from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat
 from pydantic_settings import BaseSettings
 
 
@@ -22,6 +22,21 @@ class AgentBackendConfig(BaseSettings):
         default="success",
     )
 
+    AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS: PositiveFloat = Field(
+        description="Read timeout for one Agent backend SSE connection.",
+        default=30,
+    )
+
+    AGENT_BACKEND_STREAM_MAX_RECONNECTS: NonNegativeInt = Field(
+        description="Maximum Agent backend SSE reconnects before failing the run.",
+        default=3,
+    )
+
+    AGENT_BACKEND_RUN_TIMEOUT_SECONDS: PositiveFloat = Field(
+        description="Total deadline for one Agent backend run event stream.",
+        default=1200,
+    )
+
     AGENT_SHELL_ENABLED: bool = Field(
         description=(
             "Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "

api/configs/extra/knowledge_fs_config.py

@@ -0,0 +1,64 @@
+"""Configuration for the optional KnowledgeFS Console bridge."""
+
+from urllib.parse import urlsplit
+
+from pydantic import Field, PositiveFloat, SecretStr, field_validator, model_validator
+from pydantic_settings import BaseSettings
+
+
+class KnowledgeFSConfig(BaseSettings):
+    """Server-only settings for the KnowledgeFS production connection."""
+
+    KNOWLEDGE_FS_ENABLED: bool = Field(
+        default=False,
+        description="Enable the private KnowledgeFS Console bridge.",
+    )
+    KNOWLEDGE_FS_BASE_URL: str | None = Field(default=None, description="KnowledgeFS gateway base URL.")
+    KNOWLEDGE_FS_JWT_SECRET: SecretStr | None = Field(
+        default=None,
+        min_length=32,
+        description="Shared secret used to sign short-lived KnowledgeFS service JWTs.",
+    )
+    KNOWLEDGE_FS_SSE_READ_TIMEOUT_SECONDS: PositiveFloat = Field(default=300.0, le=3600.0, allow_inf_nan=False)
+    KNOWLEDGE_FS_TIMEOUT_SECONDS: PositiveFloat = Field(default=10.0, le=60.0, allow_inf_nan=False)
+
+    @field_validator(
+        "KNOWLEDGE_FS_BASE_URL",
+        "KNOWLEDGE_FS_JWT_SECRET",
+        mode="before",
+    )
+    @classmethod
+    def normalize_optional_string(cls, value: object) -> object:
+        if isinstance(value, SecretStr):
+            normalized = value.get_secret_value().strip()
+            return SecretStr(normalized) if normalized else None
+        if isinstance(value, str):
+            normalized = value.strip()
+            return normalized or None
+        return value
+
+    @field_validator("KNOWLEDGE_FS_BASE_URL")
+    @classmethod
+    def validate_base_url(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        parsed = urlsplit(value)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must be an absolute HTTP(S) URL")
+        try:
+            _ = parsed.port
+        except ValueError as exc:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must include a valid port") from exc
+        if parsed.username or parsed.password or parsed.query or parsed.fragment:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must not include credentials, query, or fragment")
+        return value.rstrip("/")
+
+    @model_validator(mode="after")
+    def validate_enabled_connection(self) -> "KnowledgeFSConfig":
+        if not self.KNOWLEDGE_FS_ENABLED:
+            return self
+        if bool(self.KNOWLEDGE_FS_BASE_URL) != bool(self.KNOWLEDGE_FS_JWT_SECRET):
+            raise ValueError("KNOWLEDGE_FS_BASE_URL and KNOWLEDGE_FS_JWT_SECRET must be configured together")
+        if not self.KNOWLEDGE_FS_BASE_URL:
+            raise ValueError("KnowledgeFS connection settings are required when the integration is enabled")
+        return self

api/configs/feature/__init__.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 from enum import StrEnum
 from typing import Literal
 
@@ -11,6 +11,7 @@ from pydantic import (
     PositiveFloat,
     PositiveInt,
     computed_field,
+    field_validator,
 )
 from pydantic_settings import BaseSettings
 
@@ -280,6 +281,27 @@ class PluginConfig(BaseSettings):
         default="",
     )
 
+    @field_validator("PLUGIN_REMOTE_INSTALL_PORT", mode="before")
+    @classmethod
+    def _reject_host_port_shaped_plugin_remote_install_port(cls, v):
+        """Reject ``host:port``-shaped values with an actionable hint.
+
+        ``EXPOSE_PLUGIN_DEBUGGING_PORT`` is overloaded: it feeds both the
+        plugin_daemon ``ports:`` mapping (where ``127.0.0.1:5003`` is valid
+        compose syntax) and this integer app setting advertised in the console.
+        Without this guard a loopback bind spec crashloops the api container
+        with an opaque ``int_parsing`` traceback. See issue #39323.
+        """
+        if isinstance(v, str) and ":" in v.strip():
+            raise ValueError(
+                "PLUGIN_REMOTE_INSTALL_PORT must be a bare port number, got "
+                f"{v!r}. A 'host:port' value usually means "
+                "EXPOSE_PLUGIN_DEBUGGING_PORT was set to a compose publish spec "
+                "like '127.0.0.1:5003'; bind loopback via a "
+                "docker-compose.override.yaml instead of overloading this var."
+            )
+        return v
+
     @property
     def NEW_USER_DEFAULT_PLUGIN_ID_LIST(self) -> list[str]:
         return [item.strip() for item in self.NEW_USER_DEFAULT_PLUGIN_IDS.split(",") if item.strip()]
@@ -1138,6 +1160,16 @@ class HomepageConfig(BaseSettings):
         default=True,
     )
 
+    ENABLE_STEP_BY_STEP_TOUR: bool = Field(
+        description="Enable account-level Step-by-step Tour eligibility checks",
+        default=False,
+    )
+
+    STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT: datetime | None = Field(
+        description="UTC timestamp after which newly initialized accounts are eligible for Step-by-step Tour",
+        default=None,
+    )
+
 
 class RagEtlConfig(BaseSettings):
     """
@@ -1465,6 +1497,11 @@ class LoginConfig(BaseSettings):
 
 
 class AccountConfig(BaseSettings):
+    ENABLE_CHANGE_EMAIL: bool = Field(
+        description="whether users can change their email address",
+        default=True,
+    )
+
     ACCOUNT_DELETION_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
         description="Duration in minutes for which a account deletion token remains valid",
         default=5,

api/controllers/common/wraps.py

@@ -10,6 +10,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from models.model import App
+from services.agent.roster_service import AgentRosterService
 from services.enterprise.rbac_service import RBACService
 
 __all__ = ["RBACPermission", "RBACResourceScope", "enforce_rbac_access", "rbac_permission_required"]
@@ -51,7 +52,7 @@ def enforce_rbac_access(
     check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
     resource_id = None
     if resource_required and check_resource_type:
-        resource_id = _extract_resource_id(resource_type, path_args)
+        resource_id = _extract_resource_id(resource_type, tenant_id, path_args)
         if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
             return
     allowed = RBACService.CheckAccess.check(
@@ -131,11 +132,14 @@ def _is_resource_owned_by_current_user(
     return False
 
 
-def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str, object] | None = None) -> str:
+def _extract_resource_id(
+    resource_type: RBACResourceScope, tenant_id: str, path_args: dict[str, object] | None = None
+) -> str:
     """Extract the resource ID from matched path arguments.
 
     Some legacy route classes use neutral names such as ``resource_id`` for
-    app/dataset resources, and Agent App routes use ``agent_id`` as the app id.
+    app/dataset resources, and Agent routes carry ``agent_id``, which is
+    resolved to the App backing that Agent.
     Dataset endpoints behind a rag-pipeline route contain ``pipeline_id``
     instead of ``dataset_id``. In that case we look up the associated
     ``Dataset`` row via ``Dataset.pipeline_id``.
@@ -146,10 +150,19 @@ def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str,
     matched_args = {**view_args, **(path_args or {})}
 
     if resource_type == RBACResourceScope.APP:
-        app_id = matched_args.get("app_id") or matched_args.get("agent_id") or matched_args.get("resource_id")
-        if not app_id:
-            raise ValueError("Missing app_id in request path")
-        return str(app_id)
+        app_id = matched_args.get("app_id")
+        if app_id:
+            return str(app_id)
+
+        agent_id = matched_args.get("agent_id")
+        if agent_id:
+            authz_app_id = AgentRosterService(db.session).peek_authz_app_id(tenant_id=tenant_id, agent_id=str(agent_id))
+            return authz_app_id or str(agent_id)
+
+        resource_id = matched_args.get("resource_id")
+        if resource_id:
+            return str(resource_id)
+        raise ValueError("Missing app_id in request path")
 
     if resource_type == RBACResourceScope.DATASET:
         dataset_id = matched_args.get("dataset_id") or matched_args.get("resource_id")

api/controllers/console/__init__.py

@@ -38,7 +38,9 @@ from . import (
     feature,
     human_input_form,
     init_validate,
+    knowledge_fs_proxy,
     notification,
+    onboarding,
     ping,
     setup,
     spec,
@@ -195,6 +197,7 @@ __all__ = [
     "human_input_form",
     "init_validate",
     "installed_app",
+    "knowledge_fs_proxy",
     "load_balancing_config",
     "login",
     "mcp_server",
@@ -207,6 +210,7 @@ __all__ = [
     "notification",
     "oauth",
     "oauth_server",
+    "onboarding",
     "ops_trace",
     "parameter",
     "ping",

api/controllers/console/agent/composer.py

@@ -230,6 +230,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session
@@ -439,6 +440,7 @@ class SnippetAgentComposerSaveToRosterApi(Resource):
     @rbac_permission_required(
         RBACResourceScope.WORKSPACE, RBACPermission.SNIPPETS_CREATE_AND_MODIFY, resource_required=False
     )
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session
@@ -478,6 +480,7 @@ class AgentComposerApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session

api/controllers/console/agent/roster.py

@@ -62,7 +62,7 @@ from libs.datetime_utils import parse_time_range
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
-from models.agent import Agent, AgentStatus
+from models.agent import Agent, AgentConfigDraftType, AgentStatus
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import ApiTokenType
 from models.model import ApiToken, App, IconType
@@ -266,6 +266,13 @@ class AgentDebugConversationRefreshResponse(BaseModel):
     debug_conversation_message_count: int = 0
 
 
+class AgentDebugConversationRefreshPayload(BaseModel):
+    draft_type: AgentConfigDraftType = Field(
+        default=AgentConfigDraftType.DEBUG_BUILD,
+        description="Agent draft surface whose conversation should be refreshed",
+    )
+
+
 class AgentPublishPayload(BaseModel):
     version_note: str | None = Field(default=None, description="Optional note for this published Agent version")
 
@@ -309,6 +316,7 @@ register_schema_models(
     AgentAppCopyPayload,
     AgentPublishPayload,
     AgentBuildDraftCheckoutPayload,
+    AgentDebugConversationRefreshPayload,
     ComposerSavePayload,
     AgentApiStatusPayload,
     AgentInviteOptionsQuery,
@@ -392,6 +400,7 @@ def _serialize_agent_app_detail(
         tenant_id=app_model.tenant_id,
         agent_id=agent.id,
         account_id=current_user.id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
         commit=False,
     )
     message_count = roster_service.count_agent_app_debug_conversation_messages(
@@ -439,6 +448,7 @@ def _serialize_agent_app_pagination(session: Session, app_pagination, *, tenant_
         tenant_id=tenant_id,
         agents=list(agents_by_app_id.values()),
         account_id=current_user.id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
     )
     payload = AgentAppPagination.model_validate(
         app_pagination,
@@ -529,9 +539,23 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:
 
 
 def _query_values(name: str, alias_name: str | None = None) -> list[str]:
-    values = request.args.getlist(name)
+    def _get_values(field_name: str) -> list[str]:
+        values = request.args.getlist(field_name)
+        indexed_values: list[tuple[int, list[str]]] = []
+        prefix = f"{field_name}["
+        for key in request.args:
+            if not key.startswith(prefix) or not key.endswith("]"):
+                continue
+            index = key[len(prefix) : -1]
+            if index.isdigit():
+                indexed_values.append((int(index), request.args.getlist(key)))
+        for _, items in sorted(indexed_values):
+            values.extend(items)
+        return values
+
+    values = _get_values(name)
     if alias_name:
-        values.extend(request.args.getlist(alias_name))
+        values.extend(_get_values(alias_name))
     return [value.strip() for value in values if value.strip()]
 
 
@@ -542,6 +566,7 @@ class AgentAppListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -579,6 +604,7 @@ class AgentAppListApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -620,6 +646,7 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -645,6 +672,7 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session
     def delete(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -655,6 +683,16 @@ class AgentAppApi(Resource):
 
 @console_ns.route("/agent/<uuid:agent_id>/debug-conversation/refresh")
 class AgentDebugConversationRefreshApi(Resource):
+    @console_ns.expect(console_ns.models[AgentDebugConversationRefreshPayload.__name__])
+    @console_ns.doc(
+        params={
+            "payload": {
+                "in": "body",
+                "required": False,
+                "schema": {"$ref": f"#/components/schemas/{AgentDebugConversationRefreshPayload.__name__}"},
+            }
+        }
+    )
     @console_ns.response(
         200,
         "Agent debug conversation refreshed",
@@ -669,10 +707,12 @@ class AgentDebugConversationRefreshApi(Resource):
     @with_current_tenant_id
     @with_session
     def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
+        args = AgentDebugConversationRefreshPayload.model_validate(request.get_json(silent=True) or {})
         debug_conversation_id = _agent_roster_service(session).refresh_agent_app_debug_conversation_id(
             tenant_id=tenant_id,
             agent_id=str(agent_id),
             account_id=current_user.id,
+            draft_type=args.draft_type,
         )
         return AgentDebugConversationRefreshResponse(
             debug_conversation_id=debug_conversation_id,
@@ -690,6 +730,7 @@ class AgentPublishApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -712,6 +753,7 @@ class AgentBuildDraftCheckoutApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -729,6 +771,7 @@ class AgentBuildDraftCheckoutApi(Resource):
 @console_ns.route("/agent/<uuid:agent_id>/build-draft")
 class AgentBuildDraftApi(Resource):
     @console_ns.response(200, "Agent build draft", console_ns.models[AgentBuildDraftResponse.__name__])
+    @console_ns.response(404, "Agent build draft not found")
     @setup_required
     @login_required
     @account_initialization_required
@@ -787,6 +830,7 @@ class AgentBuildDraftApplyApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -809,6 +853,7 @@ class AgentAppCopyApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -834,6 +879,7 @@ class AgentApiAccessApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -850,6 +896,7 @@ class AgentApiStatusApi(Resource):
     @login_required
     @is_admin_or_owner_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_current_tenant_id
     @with_session
@@ -868,6 +915,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
     token_prefix = "app-"
 
     @console_ns.response(200, "Agent service API keys", console_ns.models[ApiKeyList.__name__])
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID) -> dict[str, object]:
@@ -878,6 +926,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
     @console_ns.response(400, "Maximum keys exceeded")
     @with_current_tenant_id
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_session
     def post(self, session: Session, tenant_id: str, agent_id: UUID) -> tuple[dict[str, object], int]:
@@ -897,6 +946,7 @@ class AgentApiKeyApi(BaseApiKeyResource):
     @console_ns.response(204, "Agent service API key deleted")
     @with_current_user
     @with_current_tenant_id
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_session
     def delete(
@@ -942,6 +992,7 @@ class AgentLogsApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -980,6 +1031,7 @@ class AgentLogMessagesApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1018,6 +1070,7 @@ class AgentLogSourcesApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1038,6 +1091,7 @@ class AgentStatisticsSummaryApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1063,6 +1117,7 @@ class AgentRosterVersionsApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -1078,6 +1133,7 @@ class AgentRosterVersionDetailApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID, version_id: UUID):
@@ -1098,6 +1154,7 @@ class AgentRosterVersionRestoreApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session

api/controllers/console/apikey.py

@@ -12,6 +12,7 @@ from werkzeug.exceptions import Forbidden
 from configs import dify_config
 from controllers.common.schema import register_response_schema_models
 from controllers.common.session import with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
 from libs.login import login_required
@@ -194,6 +195,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc(params={"resource_id": "App ID"})
     @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     @with_current_tenant_id
+    @agent_manage_required_for_agent_app
     @with_session(write=False)
     def get(self, session: Session, current_tenant_id: str, resource_id: UUID) -> dict[str, object]:
         """Get all API keys for an app"""
@@ -210,6 +212,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @with_current_tenant_id
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     def post(self, session: Session, current_tenant_id: str, resource_id: UUID) -> tuple[dict[str, object], int]:
         """Create a new API key for an app"""
@@ -233,6 +236,7 @@ class AppApiKeyResource(BaseApiKeyResource):
     @with_current_user
     @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     def delete(
         self,

api/controllers/console/app/app.py

@@ -9,7 +9,7 @@ from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, ValidationInfo, computed_field, field_validator, model_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, NotFound
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
 from configs import dify_config
 from controllers.common.app_access import resolve_app_access_filter
@@ -23,7 +23,7 @@ from controllers.common.schema import (
     register_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model, with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app, get_app_model, with_session
 from controllers.console.workspace.models import LoadBalancingPayload
 from controllers.console.wraps import (
     RBACPermission,
@@ -75,6 +75,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     WeightModel,
     WeightVectorSetting,
 )
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService
 from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
@@ -246,7 +247,7 @@ class ModelConfigPartial(ResponseModel):
         return to_timestamp(value)
 
 
-class ModelConfig(ResponseModel):
+class AppModelConfigResponse(ResponseModel):
     opening_statement: str | None = None
     suggested_questions: Any | None = Field(
         default=None, validation_alias=AliasChoices("suggested_questions_list", "suggested_questions")
@@ -419,7 +420,7 @@ class AppDetail(AppResponseModel):
     icon_background: str | None = None
     enable_site: bool
     enable_api: bool
-    model_config_: ModelConfig | None = Field(
+    model_config_: AppModelConfigResponse | None = Field(
         default=None,
         validation_alias=AliasChoices("app_model_config", "model_config"),
         alias="model_config",
@@ -525,7 +526,13 @@ def _enrich_app_list_items(session: Session, *, apps: Sequence[App], tenant_id:
 
 register_enum_models(console_ns, RetrievalMethod, WorkflowExecutionStatus, DatasetPermissionEnum)
 register_response_schema_models(
-    console_ns, RedirectUrlResponse, SimpleResultResponse, AppImportResponse, AppTraceResponse
+    console_ns,
+    RedirectUrlResponse,
+    SimpleResultResponse,
+    AppImportResponse,
+    AppTraceResponse,
+    AppModelConfigResponse,
+    AppDetail,
 )
 
 register_schema_models(
@@ -544,10 +551,8 @@ register_schema_models(
     Tag,
     WorkflowPartial,
     ModelConfigPartial,
-    ModelConfig,
     AppDetailSiteResponse,
     DeletedTool,
-    AppDetail,
     AppExportResponse,
     Segmentation,
     PreProcessingRule,
@@ -823,6 +828,7 @@ class AppApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def put(self, session: Session, app_model: App):
@@ -857,6 +863,7 @@ class AppApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_DELETE)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model
     def delete(self, session: Session, app_model: App):
@@ -881,6 +888,7 @@ class AppCopyApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT)
+    @agent_manage_required_for_agent_app
     @with_current_user
     @with_current_tenant_id
     @get_app_model(mode=None)
@@ -892,16 +900,19 @@ class AppCopyApi(Resource):
         with Session(db.engine, expire_on_commit=False) as session:
             import_service = AppDslService(session)
             yaml_content = import_service.export_dsl(app_model=app_model, session=session, include_secret=True)
-            result = import_service.import_app(
-                account=current_user,
-                import_mode=ImportMode.YAML_CONTENT,
-                yaml_content=yaml_content,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-            )
+            try:
+                result = import_service.import_app(
+                    account=current_user,
+                    import_mode=ImportMode.YAML_CONTENT,
+                    yaml_content=yaml_content,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
                 return dump_response(AppImportResponse, result), 400
@@ -955,6 +966,7 @@ class AppExportApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
     @get_app_model
     def get(self, app_model: App):
         """Export app"""
@@ -979,6 +991,7 @@ class AppPublishToCreatorsPlatformApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
     @with_current_user_id
     @get_app_model(mode=None)
     def post(self, current_user_id: str, app_model: App):
@@ -1009,6 +1022,7 @@ class AppNameApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1036,6 +1050,7 @@ class AppIconApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1069,6 +1084,7 @@ class AppSiteStatus(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1096,6 +1112,7 @@ class AppApiStatus(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):

api/controllers/console/app/app_import.py

@@ -1,6 +1,7 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from controllers.common.schema import register_enum_models, register_schema_models
@@ -28,6 +29,7 @@ from services.app_dsl_service import (
 )
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService
 
 from .. import console_ns
@@ -91,18 +93,21 @@ class AppImportApi(Resource):
             import_service = AppDslService(session)
             # Import app
             account = current_user
-            result = import_service.import_app(
-                account=account,
-                import_mode=args.mode,
-                yaml_content=args.yaml_content,
-                yaml_url=args.yaml_url,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-                app_id=args.app_id,
-            )
+            try:
+                result = import_service.import_app(
+                    account=account,
+                    import_mode=args.mode,
+                    yaml_content=args.yaml_content,
+                    yaml_url=args.yaml_url,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                    app_id=args.app_id,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:
@@ -157,7 +162,10 @@ class AppImportConfirmApi(Resource):
             import_service = AppDslService(session)
             # Confirm import
             account = current_user
-            result = import_service.confirm_import(import_id=import_id, account=account)
+            try:
+                result = import_service.confirm_import(import_id=import_id, account=account)
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:

api/controllers/console/app/completion.py

@@ -49,6 +49,7 @@ from libs import helper
 from libs.helper import uuid_value
 from libs.login import login_required
 from models import Account
+from models.agent import AgentConfigDraftType
 from models.model import App, AppMode
 from services.agent.errors import AgentNotFoundError
 from services.agent.roster_service import AgentRosterService
@@ -343,23 +344,40 @@ class AgentChatMessageStopApi(Resource):
 
 
 def _resolve_current_user_agent_debug_conversation_id(
-    *, session: Session, current_tenant_id: str, current_user: Account, app_model: App, agent_id: str | None
+    *,
+    session: Session,
+    current_tenant_id: str,
+    current_user: Account,
+    app_model: App,
+    agent_id: str | None,
+    draft_type: AgentConfigDraftType,
+    start_new: bool = False,
 ) -> str:
-    roster_service = AgentRosterService(session)
-    if agent_id:
-        return roster_service.get_or_create_agent_app_debug_conversation_id(
-            tenant_id=current_tenant_id,
-            agent_id=agent_id,
-            account_id=current_user.id,
-        )
+    """Resolve or rotate the current editor's conversation within one draft surface.
 
-    agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
-    if agent is None:
-        raise AgentNotFoundError()
-    return roster_service.get_or_create_agent_app_debug_conversation_id(
+    ``start_new`` rotates the scoped mapping through ``AgentRosterService`` so
+    the old runtime session is retired before the new conversation is used.
+    Continuations and Build chat keep resolving the existing mapping.
+    """
+
+    roster_service = AgentRosterService(session)
+    resolved_agent_id = agent_id
+    if not resolved_agent_id:
+        agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
+        if agent is None:
+            raise AgentNotFoundError()
+        resolved_agent_id = agent.id
+
+    resolve_conversation = (
+        roster_service.refresh_agent_app_debug_conversation_id
+        if start_new
+        else roster_service.get_or_create_agent_app_debug_conversation_id
+    )
+    return resolve_conversation(
         tenant_id=current_tenant_id,
-        agent_id=agent.id,
+        agent_id=resolved_agent_id,
         account_id=current_user.id,
+        draft_type=draft_type,
     )
 
 
@@ -376,12 +394,17 @@ def _create_chat_message(
     args = args_model.model_dump(exclude_none=True, by_alias=True)
 
     if AppMode.value_of(app_model.mode) == AppMode.AGENT:
+        draft_type = AgentConfigDraftType(args_model.draft_type)
+        # Preview follows the normal chat contract: an omitted/empty conversation ID starts a new
+        # conversation. Build chat keeps its stable mapping so build drafts and finalization stay continuous.
         debug_conversation_id = _resolve_current_user_agent_debug_conversation_id(
             session=session,
             current_tenant_id=current_tenant_id or app_model.tenant_id,
             current_user=current_user,
             app_model=app_model,
             agent_id=agent_id,
+            draft_type=draft_type,
+            start_new=draft_type == AgentConfigDraftType.DRAFT and not args_model.conversation_id,
         )
         if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
             raise NotFound("Conversation Not Exists.")
@@ -418,6 +441,7 @@ def _create_build_chat_finalization_message(
         current_user=current_user,
         app_model=app_model,
         agent_id=agent_id,
+        draft_type=AgentConfigDraftType.DEBUG_BUILD,
     )
     args: dict[str, Any] = {
         "query": _BUILD_CHAT_FINALIZATION_QUERY,

api/controllers/console/app/site.py

@@ -10,7 +10,7 @@ from constants.languages import supported_language
 from controllers.common.schema import register_schema_models
 from controllers.common.session import with_session
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model
+from controllers.console.app.wraps import agent_manage_required_for_agent_app, get_app_model
 from controllers.console.wraps import (
     RBACPermission,
     RBACResourceScope,
@@ -93,6 +93,7 @@ class AppSite(Resource):
     @login_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @account_initialization_required
     @with_current_user
     @with_session
@@ -145,6 +146,7 @@ class AppSiteAccessTokenReset(Resource):
     @login_required
     @is_admin_or_owner_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @account_initialization_required
     @with_current_user
     @with_session

api/controllers/console/app/workflow.py

@@ -105,6 +105,7 @@ class SyncDraftWorkflowPayload(BaseModel):
     graph: dict[str, Any]
     features: dict[str, Any]
     hash: str | None = None
+    is_collaborative: bool = Field(default=False, alias="_is_collaborative")
     environment_variables: list[dict[str, Any]] = Field(
         default_factory=list,
     )
@@ -358,6 +359,12 @@ class WorkflowPublishResponse(ResponseModel):
     created_at: int
 
 
+class SyncDraftWorkflowResponse(ResponseModel):
+    result: str
+    hash: str
+    updated_at: int
+
+
 class WorkflowRestoreResponse(ResponseModel):
     result: str
     hash: str
@@ -440,6 +447,7 @@ register_response_schema_models(
     WorkflowOnlineUsersByApp,
     WorkflowOnlineUsersResponse,
     WorkflowPublishResponse,
+    SyncDraftWorkflowResponse,
     WorkflowRestoreResponse,
     DefaultBlockConfigsResponse,
     DefaultBlockConfigResponse,
@@ -555,14 +563,7 @@ class DraftWorkflowApi(Resource):
     @console_ns.response(
         200,
         "Draft workflow synced successfully",
-        console_ns.model(
-            "SyncDraftWorkflowResponse",
-            {
-                "result": fields.String,
-                "hash": fields.String,
-                "updated_at": fields.String,
-            },
-        ),
+        console_ns.models[SyncDraftWorkflowResponse.__name__],
     )
     @console_ns.response(400, "Invalid workflow configuration")
     @console_ns.response(403, "Permission denied")
@@ -610,17 +611,21 @@ class DraftWorkflowApi(Resource):
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
                 session=db.session(),
+                graph_only=args["is_collaborative"],
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
         except VariableError as e:
             raise InvalidArgumentError(description=str(e))
 
-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
+        return dump_response(
+            SyncDraftWorkflowResponse,
+            {
+                "result": "success",
+                "hash": workflow.unique_hash,
+                "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+            },
+        )
 
 
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")

api/controllers/console/app/wraps.py

@@ -12,14 +12,22 @@ from typing import cast, overload
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from configs import dify_config
 from controllers.common.session import with_session
+from controllers.common.wraps import RBACPermission, RBACResourceScope, enforce_rbac_access
 from controllers.console.app.error import AppNotFoundError
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models import App, AppMode, TrialApp
+from models.agent import AgentScope
 from services.recommended_app_service import RecommendedAppService
 
-__all__ = ["get_app_model", "get_app_model_with_trial", "with_session"]
+__all__ = [
+    "agent_manage_required_for_agent_app",
+    "get_app_model",
+    "get_app_model_with_trial",
+    "with_session",
+]
 
 
 def _load_app_model(session: Session, app_id: str) -> App | None:
@@ -48,6 +56,45 @@ def _load_app_model_with_trial(session: Session, app_id: str) -> App | None:
     return app_model
 
 
+def agent_manage_required_for_agent_app[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+    """Gate generic app management routes that target an Agent App.
+
+    A hidden workflow-only backing App only reuses the App runtime and is not
+    part of the general app management plane, so generic routes reject it
+    outright. Managing a roster Agent App mutates the roster Agent behind it
+    (rename/icon sync, archive, API enablement), so it additionally requires
+    workspace ``agent.manage`` on top of the route's existing App permission
+    checks when RBAC is enabled. A no-op for non-agent Apps. Must be placed
+    above ``get_app_model`` so the ``app_id`` path parameter is still present.
+    """
+
+    @wraps(view)
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+        raw_app_id = kwargs.get("app_id") or kwargs.get("resource_id")
+        if raw_app_id is not None:
+            app_model = _load_app_model_from_scoped_session(str(raw_app_id))
+            binding = (
+                app_model.agent_app_binding_with_session(session=db.session(), include_archived=True)
+                if app_model is not None
+                else None
+            )
+            if binding is not None:
+                if binding.scope == AgentScope.WORKFLOW_ONLY:
+                    raise AppNotFoundError()
+                if dify_config.RBAC_ENABLED:
+                    current_user, current_tenant_id = current_account_with_tenant()
+                    enforce_rbac_access(
+                        tenant_id=current_tenant_id,
+                        account_id=current_user.id,
+                        resource_type=RBACResourceScope.WORKSPACE,
+                        scene=RBACPermission.AGENT_MANAGE,
+                        resource_required=False,
+                    )
+        return view(*args, **kwargs)
+
+    return decorated
+
+
 def _get_injected_session(args: tuple[object, ...]) -> Session | None:
     """Return the request session inserted by `with_session`, if this handler has been migrated."""
     if len(args) < 2:

api/controllers/console/auth/activate.py

@@ -7,10 +7,13 @@ from configs import dify_config
 from constants.languages import supported_language
 from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
+from controllers.console.auth.error import InvitationAccountMismatchError
 from controllers.console.error import AccountInFreezeError, AlreadyActivateError
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, timezone
+from libs.login import current_account_with_tenant
+from libs.token import extract_access_token
 from models import AccountStatus
 from models.account import TenantAccountJoin, TenantAccountRole
 from services.account_service import RegisterService, TenantService
@@ -136,6 +139,12 @@ class ActivateApi(Resource):
     )
     @console_ns.response(400, "Already activated or invalid token")
     def post(self):
+        """Accept an invitation without letting an existing session act for another account.
+
+        Token-only activation remains available for legacy clients. When the request already
+        carries a console session, that session must belong to the account encoded in the
+        invitation before the token is consumed or tenant membership is changed.
+        """
         args = ActivatePayload.model_validate(console_ns.payload)
 
         normalized_request_email = args.email.lower() if args.email else None
@@ -146,6 +155,11 @@ class ActivateApi(Resource):
             raise AlreadyActivateError()
 
         account = invitation["account"]
+        if extract_access_token(request):
+            current_account, _ = current_account_with_tenant()
+            if current_account.id != account.id:
+                raise InvitationAccountMismatchError()
+
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(account.email):
             raise AccountInFreezeError()
 

api/controllers/console/auth/error.py

@@ -13,6 +13,12 @@ class InvalidEmailError(BaseHTTPException):
     code = 400
 
 
+class InvitationAccountMismatchError(BaseHTTPException):
+    error_code = "invitation_account_mismatch"
+    description = "This invitation was sent to another account. Please sign in with the invited account."
+    code = 403
+
+
 class PasswordMismatchError(BaseHTTPException):
     error_code = "password_mismatch"
     description = "The passwords do not match."

api/controllers/console/auth/login.py

@@ -162,9 +162,10 @@ class LoginApi(Resource):
         # SELF_HOSTED only have one workspace
         tenants = TenantService.get_join_tenants(account, session=db.session())
         if len(tenants) == 0:
-            system_features = FeatureService.get_system_features()
-
-            if system_features.is_allow_create_workspace and not system_features.license.workspaces.is_available():
+            if (
+                FeatureService.get_system_features().is_allow_create_workspace
+                and not FeatureService.get_license().workspaces.is_available()
+            ):
                 raise WorkspacesLimitExceeded()
             else:
                 return SimpleResultOptionalDataResponse(
@@ -310,7 +311,7 @@ class EmailCodeLoginApi(Resource):
         if account:
             tenants = TenantService.get_join_tenants(account, session=db.session())
             if not tenants:
-                workspaces = FeatureService.get_system_features().license.workspaces
+                workspaces = FeatureService.get_license().workspaces
                 if not workspaces.is_available():
                     raise WorkspacesLimitExceeded()
                 if not FeatureService.get_system_features().is_allow_create_workspace:

api/controllers/console/auth/oauth.py

@@ -6,6 +6,7 @@ from flask import current_app, redirect, request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Unauthorized
+from werkzeug.wrappers import Response
 
 from configs import dify_config
 from constants.languages import languages
@@ -127,6 +128,20 @@ def _preferred_interface_language(language: str | None = None) -> str:
     return languages[0]
 
 
+def _redirect_with_console_session(account: Account, target_url: str) -> Response:
+    """Create a console session and attach its cookies to a redirect response."""
+    token_pair = AccountService.login(
+        account=account,
+        session=db.session(),
+        ip_address=extract_remote_ip(request),
+    )
+    response = redirect(target_url)
+    set_access_token_to_cookie(request, response, token_pair.access_token)
+    set_refresh_token_to_cookie(request, response, token_pair.refresh_token)
+    set_csrf_token_to_cookie(request, response, token_pair.csrf_token)
+    return response
+
+
 @console_ns.route("/oauth/login/<provider>")
 class OAuthLogin(Resource):
     @console_ns.doc("oauth_login")
@@ -195,16 +210,26 @@ class OAuthCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={urllib.parse.quote(str(e))}")
 
         if invite_token and RegisterService.is_valid_invite_token(invite_token):
-            invitation = RegisterService.get_invitation_by_token(token=invite_token)
-            if invitation:
-                invitation_email = invitation.get("email", None)
-                invitation_email_normalized = (
-                    invitation_email.lower() if isinstance(invitation_email, str) else invitation_email
-                )
-                if invitation_email_normalized != user_info.email.lower():
-                    return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Invalid invitation token.")
+            invitation = RegisterService.get_invitation_if_token_valid(
+                None,
+                None,
+                invite_token,
+                session=db.session(),
+            )
+            if not invitation:
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Invalid invitation token.")
+            if invitation["data"]["email"].lower() != user_info.email.lower():
+                message = "This invitation was sent to another account. Please sign in with the invited account."
+                query = urllib.parse.urlencode({"message": message, "invite_token": invite_token})
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?{query}")
 
-            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}")
+            account = invitation["account"]
+            if account.status == AccountStatus.BANNED:
+                return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account is banned.")
+
+            AccountService.link_account_integrate(provider, user_info.id, account, session=db.session())
+            target_url = f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}"
+            return _redirect_with_console_session(account, target_url)
 
         try:
             account, oauth_new_user = _generate_account(provider, user_info, timezone=timezone, language=language)
@@ -239,21 +264,10 @@ class OAuthCallback(Resource):
                 "?message=Workspace not found, please contact system admin to invite you to join in a workspace."
             )
 
-        token_pair = AccountService.login(
-            account=account,
-            session=db.session(),
-            ip_address=extract_remote_ip(request),
-        )
-
         target_url = _get_redirect_target(redirect_url)
         query_char = "&" if "?" in target_url else "?"
         target_url = f"{target_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
-        response = redirect(target_url)
-
-        set_access_token_to_cookie(request, response, token_pair.access_token)
-        set_refresh_token_to_cookie(request, response, token_pair.refresh_token)
-        set_csrf_token_to_cookie(request, response, token_pair.csrf_token)
-        return response
+        return _redirect_with_console_session(account, target_url)
 
 
 def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) -> Account | None:

api/controllers/console/auth/oauth_server.py

@@ -56,6 +56,7 @@ class OAuthProviderTokenResponse(BaseModel):
 
 
 class OAuthProviderAccountResponse(BaseModel):
+    id: str
     name: str
     email: str
     avatar: str | None = None
@@ -251,6 +252,7 @@ class OAuthServerUserAccountApi(Resource):
     def post(self, oauth_provider_app: OAuthProviderApp, account: Account):
         return jsonable_encoder(
             {
+                "id": account.id,
                 "name": account.name,
                 "email": account.email,
                 "avatar": account.avatar,

api/controllers/console/datasets/datasets.py

@@ -607,6 +607,13 @@ class DatasetListApi(Resource):
                     ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
                 )
                 initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, dataset_id=dataset.id)
+            else:
+                enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist(
+                    current_tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.SPECIFIC),
+                )
 
         permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
             current_tenant_id,
@@ -875,7 +882,7 @@ class DatasetIndexingEstimateApi(Resource):
                 file_details = session.scalars(
                     select(UploadFile).where(UploadFile.tenant_id == current_tenant_id, UploadFile.id.in_(file_ids))
                 ).all()
-                if file_details is None:
+                if not file_details:
                     raise NotFound("File not found.")
 
                 if file_details:

api/controllers/console/explore/trial.py

@@ -47,7 +47,9 @@ from controllers.console.explore.error import (
     NotWorkflowAppError,
 )
 from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.console.wraps import with_current_user
+from controllers.console.files import FILE_UPLOAD_PARAMS, upload_file_from_request
+from controllers.console.remote_files import RemoteFileUploadPayload, upload_remote_file_from_request
+from controllers.console.wraps import cloud_edition_billing_resource_check, with_current_user
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -61,12 +63,13 @@ from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.base import ResponseModel
 from fields.conversation_variable_fields import WorkflowConversationVariableResponse
+from fields.file_fields import FileResponse, FileWithSignedUrl
 from fields.message_fields import SuggestedQuestionsResponse
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import dump_response, to_timestamp, uuid_value
-from models import Account
+from models import Account, App
 from models.account import TenantStatus
 from models.model import AppMode, Site, load_annotation_reply_config
 from models.workflow import Workflow
@@ -428,6 +431,36 @@ register_response_schema_models(
 simple_account_model = console_ns.models[TrialSimpleAccount.__name__]
 
 
+class TrialAppFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.doc(consumes=["multipart/form-data"], params=FILE_UPLOAD_PARAMS)
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a file into the tenant that owns the trial app."""
+        upload_file = upload_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return dump_response(FileResponse, upload_file), 201
+
+
+class TrialAppRemoteFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileWithSignedUrl.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a remote file into the tenant that owns the trial app."""
+        remote_file = upload_remote_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return remote_file.model_dump(mode="json"), 201
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
     @trial_feature_enable
     @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
@@ -612,7 +645,7 @@ class TrialChatAudioApi(TrialAppResource):
     def post(self, current_user: Account, trial_app):
         app_model = trial_app
 
-        file = request.files["file"]
+        file = request.files.get("file")
 
         try:
             # Get IDs before they might be detached from session
@@ -887,6 +920,18 @@ class DatasetListApi(Resource):
 
 console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
 
+console_ns.add_resource(
+    TrialAppFileUploadApi,
+    "/trial-apps/<uuid:app_id>/files/upload",
+    endpoint="trial_app_file_upload",
+)
+
+console_ns.add_resource(
+    TrialAppRemoteFileUploadApi,
+    "/trial-apps/<uuid:app_id>/remote-files/upload",
+    endpoint="trial_app_remote_file_upload",
+)
+
 console_ns.add_resource(
     TrialMessageSuggestedQuestionApi,
     "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",

api/controllers/console/feature.py

@@ -3,10 +3,11 @@ from flask_restx import Resource
 from controllers.common.schema import register_response_schema_models
 from fields.base import ResponseModel
 from libs.helper import dump_response
-from libs.login import current_account_with_tenant_optional, login_required
+from libs.login import login_required
 from services.feature_service import (
     FeatureModel,
     FeatureService,
+    LicenseModel,
     LimitationModel,
     SystemFeatureModel,
 )
@@ -32,6 +33,7 @@ register_response_schema_models(
     console_ns,
     AppDslVersionResponse,
     FeatureModel,
+    LicenseModel,
     LimitationModel,
     SystemFeatureModel,
     TrialModelsResponse,
@@ -135,8 +137,28 @@ class SystemFeatureApi(Resource):
 
         Authentication would create circular dependency (can't login without dashboard loading).
 
-        Only non-sensitive configuration data should be returned by this endpoint.
+        Only non-sensitive configuration data should be returned by this endpoint. Authenticated
+        license detail is served separately by SystemFeatureLicenseApi.
         """
-        current_user, _ = current_account_with_tenant_optional()
-        is_authenticated = current_user is not None
-        return FeatureService.get_system_features(is_authenticated=is_authenticated).model_dump()
+        return FeatureService.get_system_features().model_dump()
+
+
+@console_ns.route("/system-features/license")
+class SystemFeatureLicenseApi(Resource):
+    @console_ns.doc("get_system_license")
+    @console_ns.doc(description="Get license status and usage detail")
+    @console_ns.response(
+        200,
+        "Success",
+        console_ns.models[LicenseModel.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """Get full license detail (status, expiry, workspace/seat usage).
+
+        Authenticated counterpart to the license *status* exposed on the public
+        system-features endpoint.
+        """
+        return FeatureService.get_license().model_dump()

api/controllers/console/files.py

@@ -29,7 +29,7 @@ from extensions.ext_database import db
 from fields.file_fields import FileResponse, UploadConfig
 from libs.helper import dump_response
 from libs.login import login_required
-from models import Account
+from models import Account, UploadFile
 from services.file_service import FileService
 
 from . import console_ns
@@ -39,7 +39,7 @@ register_response_schema_models(console_ns, AllowedExtensionsResponse, TextConte
 
 PREVIEW_WORDS_LIMIT = 3000
 
-_FILE_UPLOAD_PARAMS = {
+FILE_UPLOAD_PARAMS = {
     "file": {
         "description": "File to upload",
         "in": "formData",
@@ -56,6 +56,43 @@ _FILE_UPLOAD_PARAMS = {
 }
 
 
+def upload_file_from_request(*, current_user: Account, resource_tenant_id: str | None = None) -> UploadFile:
+    """Validate the multipart request and persist the file under the requested resource tenant."""
+    source_str = request.form.get("source")
+    source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
+
+    if "file" not in request.files:
+        raise NoFileUploadedError()
+
+    if len(request.files) > 1:
+        raise TooManyFilesError()
+    file = request.files["file"]
+
+    if not file.filename:
+        raise FilenameNotExistsError
+    if source == "datasets" and not current_user.is_dataset_editor:
+        raise Forbidden()
+
+    if source not in ("datasets", None):
+        source = None
+
+    try:
+        return FileService(db.engine).upload_file(
+            filename=file.filename,
+            content=file.stream.read(),
+            mimetype=file.mimetype,
+            user=current_user,
+            tenant_id=resource_tenant_id,
+            source=source,
+        )
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+    except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
+        raise BlockedFileExtensionError(blocked_extension_error.description)
+
+
 @console_ns.route("/files/upload")
 class FileApi(Resource):
     @setup_required
@@ -81,42 +118,11 @@ class FileApi(Resource):
     @login_required
     @account_initialization_required
     @cloud_edition_billing_resource_check("documents")
-    @console_ns.doc(consumes=["multipart/form-data"], params=_FILE_UPLOAD_PARAMS)
+    @console_ns.doc(consumes=["multipart/form-data"], params=FILE_UPLOAD_PARAMS)
     @console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
     @with_current_user
     def post(self, current_user: Account):
-        source_str = request.form.get("source")
-        source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
-
-        if "file" not in request.files:
-            raise NoFileUploadedError()
-
-        if len(request.files) > 1:
-            raise TooManyFilesError()
-        file = request.files["file"]
-
-        if not file.filename:
-            raise FilenameNotExistsError
-        if source == "datasets" and not current_user.is_dataset_editor:
-            raise Forbidden()
-
-        if source not in ("datasets", None):
-            source = None
-
-        try:
-            upload_file = FileService(db.engine).upload_file(
-                filename=file.filename,
-                content=file.stream.read(),
-                mimetype=file.mimetype,
-                user=current_user,
-                source=source,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-        except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
-            raise BlockedFileExtensionError(blocked_extension_error.description)
+        upload_file = upload_file_from_request(current_user=current_user)
 
         return dump_response(FileResponse, upload_file), 201
 

api/controllers/console/init_validate.py

@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from controllers.fastopenapi import console_router
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from models.model import DifySetup
 from services.account_service import TenantService
@@ -63,7 +64,7 @@ def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse
 
 
 def get_init_validate_status() -> bool:
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
         if os.environ.get("INIT_PASSWORD"):
             if session.get("is_init_validated"):
                 return True

api/controllers/console/knowledge_fs_proxy.py

@@ -0,0 +1,419 @@
+"""Authenticated transport adapter for the Console-to-KnowledgeFS proxy.
+
+These raw Blueprint routes deliberately stay outside Dify's OpenAPI surface:
+KnowledgeFS owns the wire contract consumed by the frontend. The catch-all path
+avoids resource-specific Dify controllers, while the forwarding module consumes
+only the operations explicitly enabled by Dify's product registry. The registry
+can be validated explicitly against the pinned KnowledgeFS contract during development.
+Console auth and contract-specific dataset RBAC run before forwarding. Request
+bodies are capped at 64 MiB, JSON and binary responses have separate bounds,
+SSE responses remain streaming with a bounded idle read timeout, and only safe
+response headers are exposed. Operation-specific upstream error mappings are
+applied before Console JSON error handling; the default maps 401 to 502 so it
+cannot trigger browser-session recovery and preserves resource-level 403.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Iterator
+from functools import wraps
+from http import HTTPStatus
+from typing import NoReturn, cast
+
+import httpx
+from flask import Response, request, stream_with_context
+from flask.typing import ResponseReturnValue
+from werkzeug.exceptions import (
+    BadGateway,
+    Forbidden,
+    GatewayTimeout,
+    HTTPException,
+    NotFound,
+    RequestEntityTooLarge,
+    ServiceUnavailable,
+    default_exceptions,
+)
+
+from configs import dify_config
+from controllers.console import api, bp
+from controllers.console.wraps import (
+    account_initialization_required,
+    cloud_edition_billing_rate_limit_check,
+    setup_required,
+)
+from core.helper import ssrf_proxy
+from libs.login import current_account_with_tenant, login_required
+from services.knowledge_fs_operations import KnowledgeFSMethod
+from services.knowledge_fs_proxy import (
+    KnowledgeFSAccessDeniedError,
+    KnowledgeFSAuthorization,
+    KnowledgeFSConfigurationError,
+    KnowledgeFSRouteNotAllowedError,
+    KnowledgeFSTimeoutError,
+    KnowledgeFSTransportError,
+    KnowledgeFSUpstreamResponse,
+    authorize_knowledge_fs_request,
+    get_knowledge_fs_operation,
+    proxy_authorized_knowledge_fs_request,
+    proxy_knowledge_fs_request,
+)
+
+logger = logging.getLogger(__name__)
+
+type _KnowledgeFSRequestForwarder = Callable[
+    [str | None, str | None, bytes | None, bytes | None],
+    KnowledgeFSUpstreamResponse,
+]
+
+_MAX_PROXY_BODY_BYTES = 64 * 1024 * 1024
+_RESPONSE_HEADER_ALLOWLIST = (
+    "Cache-Control",
+    "Content-Disposition",
+    "Content-Type",
+    "Retry-After",
+    "X-Trace-Id",
+)
+_RESPONSE_HEADER_DENYLIST = frozenset(
+    {
+        "authorization",
+        "connection",
+        "cookie",
+        "keep-alive",
+        "proxy-authenticate",
+        "proxy-authorization",
+        "set-cookie",
+        "te",
+        "trailer",
+        "transfer-encoding",
+        "upgrade",
+    }
+)
+
+
+def _console_api_errors[**P](
+    view: Callable[P, ResponseReturnValue],
+) -> Callable[P, ResponseReturnValue]:
+    """Route raw Blueprint exceptions through the Console API JSON handlers."""
+
+    @wraps(view)
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> ResponseReturnValue:
+        try:
+            return view(*args, **kwargs)
+        except Exception as exc:
+            return api.handle_error(exc)
+
+    return decorated
+
+
+def _knowledge_fs_enabled[**P](
+    view: Callable[P, ResponseReturnValue],
+) -> Callable[P, ResponseReturnValue]:
+    """Hide the complete KnowledgeFS route surface while the bridge is disabled."""
+
+    @wraps(view)
+    def decorated(*args: P.args, **kwargs: P.kwargs) -> ResponseReturnValue:
+        if not dify_config.KNOWLEDGE_FS_ENABLED:
+            raise NotFound()
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+def _translate_proxy_error(exc: Exception, *, tenant_id: str) -> NoReturn:
+    """Map forwarding failures to the stable Console HTTP error surface."""
+    if isinstance(exc, KnowledgeFSRouteNotAllowedError):
+        raise NotFound() from exc
+    if isinstance(exc, KnowledgeFSAccessDeniedError):
+        raise Forbidden() from exc
+    if isinstance(exc, KnowledgeFSConfigurationError):
+        logger.error("KnowledgeFS request was blocked by invalid configuration for tenant_id=%s", tenant_id)
+        raise ServiceUnavailable("KnowledgeFS integration is misconfigured") from exc
+    if isinstance(exc, KnowledgeFSTimeoutError):
+        raise GatewayTimeout("KnowledgeFS request timed out") from exc
+    if isinstance(exc, KnowledgeFSTransportError):
+        logger.warning("KnowledgeFS transport request failed for tenant_id=%s", tenant_id)
+        raise BadGateway("KnowledgeFS is unavailable") from exc
+    raise exc
+
+
+def _knowledge_fs_operation_access_required(
+    view: Callable[[KnowledgeFSAuthorization], ResponseReturnValue],
+) -> Callable[[KnowledgeFSMethod, str], ResponseReturnValue]:
+    """Authorize one declared operation before billing and request-body work."""
+
+    @wraps(view)
+    def decorated(method: KnowledgeFSMethod, upstream_path: str) -> ResponseReturnValue:
+        current_user, tenant_id = current_account_with_tenant()
+        try:
+            authorization = authorize_knowledge_fs_request(
+                account=current_user,
+                tenant_id=tenant_id,
+                method=method,
+                path=upstream_path,
+            )
+        except KnowledgeFSRouteNotAllowedError as exc:
+            raise NotFound() from exc
+        except KnowledgeFSAccessDeniedError as exc:
+            _translate_proxy_error(exc, tenant_id=tenant_id)
+        return view(authorization)
+
+    return decorated
+
+
+def _request_body() -> bytes:
+    """Read the raw body up to the proxy limit or raise RequestEntityTooLarge."""
+    body = request.stream.read(_MAX_PROXY_BODY_BYTES + 1)
+    if len(body) > _MAX_PROXY_BODY_BYTES:
+        raise RequestEntityTooLarge("KnowledgeFS proxy request body is too large")
+    return body
+
+
+def _stream_response_body(
+    upstream: httpx.Response,
+    *,
+    tenant_id: str,
+    max_response_bytes: int,
+) -> Iterator[bytes]:
+    """Yield one bounded SSE response and always release its pooled connection."""
+    total_bytes = 0
+    try:
+        for chunk in upstream.iter_bytes():
+            total_bytes += len(chunk)
+            if total_bytes > max_response_bytes:
+                logger.warning("KnowledgeFS stream exceeded the proxy limit for tenant_id=%s", tenant_id)
+                raise ssrf_proxy.ResponseTooLargeError(f"response exceeded {max_response_bytes} bytes")
+            yield chunk
+    finally:
+        upstream.close()
+
+
+def _proxy_response(
+    upstream_result: KnowledgeFSUpstreamResponse,
+    *,
+    tenant_id: str,
+    contract_response_headers: tuple[str, ...],
+    max_response_bytes: int,
+) -> Response:
+    """Expose raw content, status, and allowlisted headers from KnowledgeFS.
+
+    Raises:
+        HTTPException: KnowledgeFS returns a status normalized by the operation contract.
+    """
+    upstream = upstream_result.response
+    mapped_status = dict(upstream_result.operation.error_status_map).get(upstream.status_code)
+    if mapped_status is not None:
+        upstream.close()
+        description = "KnowledgeFS upstream request failed"
+        if upstream.status_code == HTTPStatus.UNAUTHORIZED:
+            description = "KnowledgeFS authentication failed"
+            logger.error(
+                "KnowledgeFS rejected the Dify server credential with HTTP %s for tenant_id=%s",
+                upstream.status_code,
+                tenant_id,
+            )
+        exception_type = default_exceptions.get(mapped_status)
+        if exception_type is None:
+            exception = HTTPException(description)
+            exception.code = mapped_status
+            raise exception
+        raise exception_type(description)
+
+    allowed_header_names = dict.fromkeys(
+        name.lower() for name in (*_RESPONSE_HEADER_ALLOWLIST, *contract_response_headers)
+    )
+    headers = {
+        name: value
+        for name in allowed_header_names
+        if name not in _RESPONSE_HEADER_DENYLIST
+        if (value := upstream.headers.get(name)) is not None
+    }
+    if upstream_result.response_kind == "stream":
+        response = Response(
+            stream_with_context(  # pyrefly: ignore[no-matching-overload]
+                _stream_response_body(
+                    upstream,
+                    tenant_id=tenant_id,
+                    max_response_bytes=max_response_bytes,
+                )
+            ),
+            status=upstream.status_code,
+            headers=headers,
+        )
+        response.call_on_close(upstream.close)
+        return response
+
+    try:
+        content = upstream.content
+    finally:
+        upstream.close()
+    return Response(content, status=upstream.status_code, headers=headers)
+
+
+def _proxy_current_request(
+    *,
+    method: KnowledgeFSMethod,
+    tenant_id: str,
+    forward: _KnowledgeFSRequestForwarder,
+) -> Response:
+    """Forward the current raw request through one preconfigured service entry."""
+    if not dify_config.KNOWLEDGE_FS_ENABLED:
+        raise NotFound()
+    try:
+        proxy_result = forward(
+            request.headers.get("Accept"),
+            request.content_type,
+            request.query_string or None,
+            _request_body() if method != "GET" else None,
+        )
+    except (
+        KnowledgeFSConfigurationError,
+        KnowledgeFSAccessDeniedError,
+        KnowledgeFSRouteNotAllowedError,
+        KnowledgeFSTimeoutError,
+        KnowledgeFSTransportError,
+    ) as exc:
+        _translate_proxy_error(exc, tenant_id=tenant_id)
+    return _proxy_response(
+        proxy_result,
+        tenant_id=tenant_id,
+        contract_response_headers=proxy_result.operation.response_headers,
+        max_response_bytes=proxy_result.operation.max_response_bytes,
+    )
+
+
+def _proxy_request(
+    method: KnowledgeFSMethod,
+    upstream_path: str,
+) -> Response:
+    """Authorize and forward the current request through the combined service use case."""
+    if not dify_config.KNOWLEDGE_FS_ENABLED:
+        raise NotFound()
+    current_user, tenant_id = current_account_with_tenant()
+
+    def forward(
+        accept: str | None,
+        content_type: str | None,
+        query: bytes | None,
+        body: bytes | None,
+    ) -> KnowledgeFSUpstreamResponse:
+        return proxy_knowledge_fs_request(
+            account=current_user,
+            method=method,
+            path=upstream_path,
+            tenant_id=tenant_id,
+            accept=accept,
+            content_type=content_type,
+            query=query,
+            body=body,
+            request_headers=request.headers,
+        )
+
+    return _proxy_current_request(method=method, tenant_id=tenant_id, forward=forward)
+
+
+def _proxy_authorized_request(authorization: KnowledgeFSAuthorization) -> Response:
+    """Forward the current request using one previously authorized operation capability.
+
+    Args:
+        authorization: Request-scoped capability produced before billing and body parsing.
+
+    Returns:
+        The filtered response returned by KnowledgeFS.
+
+    Raises:
+        HTTPException: The integration is disabled or forwarding fails.
+    """
+    operation = authorization.operation
+    tenant_id = authorization.tenant_id
+
+    def forward(
+        accept: str | None,
+        content_type: str | None,
+        query: bytes | None,
+        body: bytes | None,
+    ) -> KnowledgeFSUpstreamResponse:
+        return proxy_authorized_knowledge_fs_request(
+            authorization=authorization,
+            accept=accept,
+            content_type=content_type,
+            query=query,
+            body=body,
+            request_headers=request.headers,
+        )
+
+    return _proxy_current_request(method=operation.method, tenant_id=tenant_id, forward=forward)
+
+
+@_knowledge_fs_enabled
+@_knowledge_fs_operation_access_required
+@cloud_edition_billing_rate_limit_check("knowledge")
+def _proxy_knowledge_fs_non_get(
+    authorization: KnowledgeFSAuthorization,
+) -> ResponseReturnValue:
+    """Apply knowledge billing checks to one allowlisted non-GET operation."""
+    return _proxy_authorized_request(authorization)
+
+
+@bp.route(
+    "/knowledge-fs/<path:upstream_path>",
+    methods=["OPTIONS"],
+    provide_automatic_options=False,
+)
+@_console_api_errors
+@_knowledge_fs_enabled
+def proxy_knowledge_fs_options(upstream_path: str) -> ResponseReturnValue:
+    """Complete a CORS preflight only for an enabled Console operation."""
+    requested_method = cast(KnowledgeFSMethod, request.headers.get("Access-Control-Request-Method", "").upper())
+    try:
+        get_knowledge_fs_operation(requested_method, upstream_path)
+    except KnowledgeFSRouteNotAllowedError as exc:
+        raise NotFound() from exc
+    return Response(status=HTTPStatus.NO_CONTENT)
+
+
+@bp.route(
+    "/knowledge-fs/<path:upstream_path>",
+    methods=["GET"],
+    provide_automatic_options=False,
+)
+@_console_api_errors
+@_knowledge_fs_enabled
+@setup_required
+@login_required
+@account_initialization_required
+def proxy_knowledge_fs_get(upstream_path: str) -> ResponseReturnValue:
+    """Forward one authenticated, dataset-readable GET request.
+
+    Args:
+        upstream_path: Relative KFS path captured after the Console proxy prefix.
+
+    Returns:
+        The filtered raw KnowledgeFS response or a Console JSON error response.
+    """
+    if request.method != "GET":
+        raise NotFound()
+    return _proxy_request("GET", upstream_path)
+
+
+@bp.route(
+    "/knowledge-fs/<path:upstream_path>",
+    methods=["DELETE", "PATCH", "POST", "PUT"],
+    provide_automatic_options=False,
+)
+@_console_api_errors
+@_knowledge_fs_enabled
+@setup_required
+@login_required
+@account_initialization_required
+def proxy_knowledge_fs_write(upstream_path: str) -> ResponseReturnValue:
+    """Forward one authenticated non-GET request under its contract access policy.
+
+    Args:
+        upstream_path: Relative KFS path captured after the Console proxy prefix.
+
+    Returns:
+        The filtered raw KnowledgeFS response or a Console JSON error response.
+    """
+    method = cast(KnowledgeFSMethod, request.method)
+    return _proxy_knowledge_fs_non_get(method, upstream_path)

api/controllers/console/onboarding.py

@@ -0,0 +1,106 @@
+"""Console onboarding APIs.
+
+This module keeps Step-by-step Tour persistence account-scoped. Workspace IDs
+are accepted only as presentation overrides; UI-only state such as minimized
+panels or the currently active task stays on the frontend. PATCH requests are
+action-based so callers do not replace server-side arrays with stale snapshots.
+"""
+
+from datetime import datetime
+from typing import Literal, cast
+
+from flask_restx import Resource
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from controllers.common.schema import register_response_schema_models, register_schema_models
+from extensions.ext_database import db
+from fields.base import ResponseModel
+from libs.helper import dump_response
+from libs.login import login_required
+from models import Account
+from services.step_by_step_tour_service import StepByStepTourPatch, StepByStepTourService
+
+from . import console_ns
+from .wraps import account_initialization_required, setup_required, with_current_tenant_id, with_current_user
+
+StepByStepTourAction = Literal[
+    "skip",
+    "complete_task",
+    "uncomplete_task",
+    "enable_current_workspace",
+    "disable_current_workspace",
+]
+StepByStepTourTaskId = Literal["home", "studio", "knowledge", "integration"]
+
+
+class StepByStepTourStatePatchPayload(BaseModel):
+    action: StepByStepTourAction = Field(description="State update action")
+    task_id: StepByStepTourTaskId | None = Field(default=None, description="Task ID for task actions")
+
+    model_config = ConfigDict(extra="forbid")
+
+    @model_validator(mode="after")
+    def validate_patch_shape(self) -> "StepByStepTourStatePatchPayload":
+        task_actions = {"complete_task", "uncomplete_task"}
+        if self.action in task_actions and self.task_id is None:
+            raise ValueError("task_id is required for task actions")
+        if self.action not in task_actions and self.task_id is not None:
+            raise ValueError("task_id is only supported for task actions")
+
+        return self
+
+
+class StepByStepTourStateResponse(ResponseModel):
+    first_workspace_id: str | None = None
+    skipped: bool = False
+    completed_task_ids: list[StepByStepTourTaskId] = Field(default_factory=list)
+    manually_enabled_workspace_ids: list[str] = Field(default_factory=list)
+    manually_disabled_workspace_ids: list[str] = Field(default_factory=list)
+    updated_at: datetime | None = None
+
+
+register_schema_models(console_ns, StepByStepTourStatePatchPayload)
+register_response_schema_models(console_ns, StepByStepTourStateResponse)
+
+
+@console_ns.route("/onboarding/step-by-step-tour/state")
+class StepByStepTourStateApi(Resource):
+    @console_ns.doc("get_step_by_step_tour_state")
+    @console_ns.doc(description="Get account-level Step-by-step Tour state")
+    @console_ns.response(200, "Success", console_ns.models[StepByStepTourStateResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, current_user: Account):
+        return dump_response(
+            StepByStepTourStateResponse,
+            StepByStepTourService.get_state(
+                account=current_user,
+                current_tenant_id=current_tenant_id,
+                session=db.session,
+            ),
+        )
+
+    @console_ns.doc("patch_step_by_step_tour_state")
+    @console_ns.doc(description="Update account-level Step-by-step Tour state")
+    @console_ns.expect(console_ns.models[StepByStepTourStatePatchPayload.__name__])
+    @console_ns.response(200, "Success", console_ns.models[StepByStepTourStateResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_user
+    @with_current_tenant_id
+    def patch(self, current_tenant_id: str, current_user: Account):
+        payload = StepByStepTourStatePatchPayload.model_validate(console_ns.payload or {})
+        patch = cast(StepByStepTourPatch, payload.model_dump(exclude_unset=True, exclude_none=True))
+        return dump_response(
+            StepByStepTourStateResponse,
+            StepByStepTourService.patch_state(
+                account=current_user,
+                current_tenant_id=current_tenant_id,
+                patch=patch,
+                session=db.session,
+            ),
+        )

api/controllers/console/remote_files.py

@@ -46,6 +46,61 @@ class GetRemoteFileInfo(Resource):
         ).model_dump(mode="json")
 
 
+def upload_remote_file_from_request(
+    *,
+    current_user: Account,
+    resource_tenant_id: str | None = None,
+) -> FileWithSignedUrl:
+    """Validate the JSON request, fetch its remote file, and persist it under the requested tenant."""
+    payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
+    url = payload.url
+
+    # Try to fetch remote file metadata/content first
+    try:
+        resp = remote_fetcher.make_request("HEAD", url=url)
+        if resp.status_code != httpx.codes.OK:
+            resp = remote_fetcher.make_request("GET", url=url, timeout=3, follow_redirects=True)
+        if resp.status_code != httpx.codes.OK:
+            # Normalize into a user-friendly error message expected by tests
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+    except httpx.RequestError as e:
+        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+
+    file_info = helpers.guess_file_info_from_response(resp)
+
+    # Enforce file size limit with 400 (Bad Request) per tests' expectation
+    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+        raise FileTooLargeError()
+
+    # Load content if needed
+    content = resp.content if resp.request.method == "GET" else remote_fetcher.make_request("GET", url).content
+
+    try:
+        upload_file = FileService(db.engine).upload_file(
+            filename=file_info.filename,
+            content=content,
+            mimetype=file_info.mimetype,
+            user=current_user,
+            tenant_id=resource_tenant_id,
+            source_url=url,
+        )
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+
+    return FileWithSignedUrl(
+        id=upload_file.id,
+        name=upload_file.name,
+        size=upload_file.size,
+        extension=upload_file.extension,
+        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+        mime_type=upload_file.mime_type,
+        created_by=upload_file.created_by,
+        created_at=int(upload_file.created_at.timestamp()),
+    )
+
+
 @console_ns.route("/remote-files/upload")
 class RemoteFileUpload(Resource):
     @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
@@ -53,53 +108,8 @@ class RemoteFileUpload(Resource):
     @login_required
     @with_current_user
     def post(self, current_user: Account):
-        payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = payload.url
-
-        # Try to fetch remote file metadata/content first
-        try:
-            resp = remote_fetcher.make_request("HEAD", url=url)
-            if resp.status_code != httpx.codes.OK:
-                resp = remote_fetcher.make_request("GET", url=url, timeout=3, follow_redirects=True)
-            if resp.status_code != httpx.codes.OK:
-                # Normalize into a user-friendly error message expected by tests
-                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-        except httpx.RequestError as e:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
-
-        file_info = helpers.guess_file_info_from_response(resp)
-
-        # Enforce file size limit with 400 (Bad Request) per tests' expectation
-        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError()
-
-        # Load content if needed
-        content = resp.content if resp.request.method == "GET" else remote_fetcher.make_request("GET", url).content
-
-        try:
-            upload_file = FileService(db.engine).upload_file(
-                filename=file_info.filename,
-                content=content,
-                mimetype=file_info.mimetype,
-                user=current_user,
-                source_url=url,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-
-        # Success: return created resource with 201 status
+        remote_file = upload_remote_file_from_request(current_user=current_user)
         return (
-            FileWithSignedUrl(
-                id=upload_file.id,
-                name=upload_file.name,
-                size=upload_file.size,
-                extension=upload_file.extension,
-                url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-                mime_type=upload_file.mime_type,
-                created_by=upload_file.created_by,
-                created_at=int(upload_file.created_at.timestamp()),
-            ).model_dump(mode="json"),
+            remote_file.model_dump(mode="json"),
             201,
         )

api/controllers/console/setup.py

@@ -6,6 +6,7 @@ from sqlalchemy import select
 
 from configs import dify_config
 from controllers.fastopenapi import console_router
+from enums.deployment_edition import DeploymentEdition
 from libs.helper import EmailStr, extract_remote_ip
 from libs.password import valid_password
 from models.model import DifySetup, db
@@ -52,7 +53,7 @@ def get_setup_status_api() -> SetupStatusResponse:
 
     Only bootstrap-safe status information should be returned by this endpoint.
     """
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
         setup_status = get_setup_status()
         if setup_status and not isinstance(setup_status, bool):
             return SetupStatusResponse(step="finished", setup_at=setup_status.setup_at.isoformat())
@@ -102,7 +103,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
 
 
 def get_setup_status() -> DifySetup | bool | None:
-    if dify_config.EDITION == "SELF_HOSTED":
+    if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
         return db.session.scalar(select(DifySetup).limit(1))
 
     return True

api/controllers/console/socketio/workflow.py

@@ -98,6 +98,7 @@ def handle_collaboration_event(sid, data):
     6. workflow_update
     7. comments_update
     8. node_panel_presence
+    9. graph_view_state (session reports tab visibility; drives leader election)
     """
     return collaboration_service.relay_collaboration_event(sid, data)
 

api/controllers/console/workflow_run_archive.py

@@ -4,20 +4,16 @@ from http import HTTPStatus
 from flask import redirect
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from werkzeug.exceptions import Conflict, NotFound
+from werkzeug.exceptions import Conflict, Forbidden, NotFound
 
 from controllers.common.fields import RedirectResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
-    RBACPermission,
-    RBACResourceScope,
     account_initialization_required,
     cloud_edition_billing_enabled,
     cloud_edition_billing_paid_plan_required,
-    is_admin_or_owner_required,
     only_edition_cloud,
-    rbac_permission_required,
     setup_required,
 )
 from extensions.ext_database import db
@@ -25,6 +21,7 @@ from fields.base import ResponseModel
 from libs.archive_storage import get_export_storage
 from libs.helper import dump_response
 from libs.login import current_account_with_tenant, login_required
+from models import TenantAccountRole
 from services.retention.workflow_run.archive_download_preparation import ARCHIVE_DOWNLOAD_MIME_TYPE
 from services.retention.workflow_run.archive_download_task_cache import (
     WorkflowRunArchiveDownloadStatus,
@@ -98,11 +95,13 @@ register_response_schema_models(
 )
 
 
-def _current_ids() -> tuple[str, str]:
-    """Return current `(tenant_id, account_id)` or raise when no workspace is selected."""
+def _current_owner_or_admin_ids() -> tuple[str, str]:
+    """Return current Cloud workspace IDs for an owner or admin, independently of enterprise RBAC."""
     current_user, current_tenant_id = current_account_with_tenant()
     if not current_tenant_id:
         raise NotFound("Current workspace not found")
+    if not TenantAccountRole.is_privileged_role(current_user.current_role):
+        raise Forbidden()
     return current_tenant_id, current_user.id
 
 
@@ -124,12 +123,8 @@ class WorkflowRunArchivesApi(Resource):
     @only_edition_cloud
     @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
     def get(self):
-        tenant_id, _ = _current_ids()
+        tenant_id, _ = _current_owner_or_admin_ids()
         return dump_response(WorkflowRunArchiveListResponse, list_workflow_run_archives(db.session(), tenant_id))
 
 
@@ -149,12 +144,8 @@ class WorkflowRunArchiveDownloadsApi(Resource):
     @only_edition_cloud
     @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
     def post(self):
-        tenant_id, account_id = _current_ids()
+        tenant_id, account_id = _current_owner_or_admin_ids()
         payload = WorkflowRunArchiveDownloadPayload.model_validate(console_ns.payload or {})
         try:
             task = create_workflow_run_archive_download_task(
@@ -180,12 +171,8 @@ class WorkflowRunArchiveDownloadApi(Resource):
     @only_edition_cloud
     @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
     def get(self, download_id: str):
-        tenant_id, _ = _current_ids()
+        tenant_id, _ = _current_owner_or_admin_ids()
         try:
             task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
         except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
@@ -209,12 +196,8 @@ class WorkflowRunArchiveDownloadFileApi(Resource):
     @only_edition_cloud
     @cloud_edition_billing_enabled
     @cloud_edition_billing_paid_plan_required
-    @is_admin_or_owner_required
-    @rbac_permission_required(
-        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
-    )
     def get(self, download_id: str):
-        tenant_id, _ = _current_ids()
+        tenant_id, _ = _current_owner_or_admin_ids()
         try:
             task = get_ready_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
         except WorkflowRunArchiveDownloadTaskNotFoundError as exc:

api/controllers/console/workspace/account.py

@@ -46,6 +46,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from fields.member_fields import AccountResponse
@@ -262,7 +263,7 @@ class AccountInitApi(Resource):
         payload = console_ns.payload or {}
         args = AccountInitPayload.model_validate(payload)
 
-        if dify_config.EDITION == "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
             if not args.invitation_code:
                 raise ValueError("invitation_code is required")
 

api/controllers/console/workspace/members.py

@@ -198,7 +198,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int, new_accou
         if workspace_members.enabled is True and not workspace_members.is_available(new_member_count):
             raise WorkspaceMembersLimitExceeded()
         if new_account_count > 0:
-            seats = FeatureService.get_system_features(is_authenticated=True).license.seats
+            seats = FeatureService.get_license().seats
             if not seats.is_available(new_account_count):
                 raise SeatsLimitExceeded()
         return

api/controllers/console/workspace/plugin.py

@@ -1,5 +1,5 @@
 import io
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any, Literal, TypedDict
 
@@ -43,6 +43,7 @@ from core.plugin.entities.plugin_daemon import PluginDecodeResponse, PluginInsta
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.plugin.plugin_service import PluginService
 from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
+from core.tools.entities.api_entities import ToolProviderApiEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
@@ -67,6 +68,15 @@ from services.plugin.plugin_parameter_service import PluginParameterService
 from services.plugin.plugin_permission_service import PluginPermissionService
 from services.tools.tools_transform_service import ToolTransformService
 
+_PLUGIN_PACKAGE_UPLOAD_PARAMS = {
+    "pkg": {
+        "description": "Plugin package to upload",
+        "in": "formData",
+        "type": "file",
+        "required": True,
+    }
+}
+
 
 class AutoUpgradeSettingsResponse(TypedDict):
     strategy_setting: TenantPluginAutoUpgradeStrategySetting
@@ -84,6 +94,15 @@ class ParserList(BaseModel):
 class PluginCategoryListQuery(BaseModel):
     page: int = Field(default=1, ge=1, description="Page number")
     page_size: int = Field(default=256, ge=1, le=256, description="Page size (1-256)")
+    query: str = Field(default="", max_length=256, description="Case-insensitive search query")
+    tags: list[str] = Field(default_factory=list, max_length=128, description="Match any plugin tag")
+    language: Literal["en_US", "zh_Hans", "ja_JP", "pt_BR"] = Field(
+        default="en_US", description="Language used for localized label and description search"
+    )
+
+
+class PluginInstalledIdsQuery(BaseModel):
+    category: PluginCategory = Field(description="Plugin category to include")
 
 
 class ParserLatest(BaseModel):
@@ -316,6 +335,10 @@ class PluginListResponse(ResponseModel):
     total: int
 
 
+class PluginInstalledIdsResponse(ResponseModel):
+    plugin_ids: list[str]
+
+
 class PluginVersionsResponse(ResponseModel):
     versions: Mapping[str, PluginService.LatestPluginCache | None]
 
@@ -375,6 +398,7 @@ register_schema_models(
     console_ns,
     ParserList,
     PluginCategoryListQuery,
+    PluginInstalledIdsQuery,
     PluginAutoUpgradeSettingsPayload,
     PluginPermissionSettingsPayload,
     ParserLatest,
@@ -411,6 +435,7 @@ register_response_schema_models(
     PluginDebuggingKeyResponse,
     PluginDynamicOptionsResponse,
     PluginInstallationsResponse,
+    PluginInstalledIdsResponse,
     PluginInstallTaskStartResponse,
     PluginListResponse,
     PluginManifestResponse,
@@ -433,12 +458,10 @@ register_enum_models(
 )
 
 
-def _default_auto_upgrade_settings(
-    tenant_id: str,
-    category: TenantPluginAutoUpgradeCategory,
-) -> AutoUpgradeSettingsResponse:
+def _missing_auto_upgrade_settings(tenant_id: str) -> AutoUpgradeSettingsResponse:
+    """Represent a missing persisted strategy as effectively disabled."""
     return {
-        "strategy_setting": PluginAutoUpgradeService.default_strategy_setting_for_category(category),
+        "strategy_setting": TenantPluginAutoUpgradeStrategySetting.DISABLED,
         "upgrade_time_of_day": PluginAutoUpgradeService.default_upgrade_time_of_day(tenant_id),
         "upgrade_mode": TenantPluginAutoUpgradeMode.EXCLUDE,
         "exclude_plugins": [],
@@ -470,7 +493,39 @@ def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
     return content
 
 
-def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any]]:
+def _localized_builtin_tool_text(value: I18nObject, language: str) -> str:
+    return getattr(value, language, None) or value.en_US
+
+
+def _builtin_tool_provider_matches_filters(
+    provider: ToolProviderApiEntity,
+    *,
+    query: str,
+    tags: Sequence[str],
+    language: str,
+) -> bool:
+    if tags and not any(tag in provider.labels for tag in tags):
+        return False
+    if not query:
+        return True
+
+    lower_query = query.lower()
+    candidates = (
+        provider.name,
+        _localized_builtin_tool_text(provider.label, language),
+        _localized_builtin_tool_text(provider.description, language),
+    )
+    return any(lower_query in candidate.lower() for candidate in candidates)
+
+
+def _list_hardcoded_builtin_tool_providers(
+    tenant_id: str,
+    *,
+    query: str = "",
+    tags: Sequence[str] = (),
+    language: str = "en_US",
+) -> list[dict[str, Any]]:
+    """List builtin providers using the same search and tag semantics as category plugins."""
     db_builtin_providers = {
         str(ToolProviderID(provider.provider)): provider
         for provider in ToolManager.list_default_builtin_providers(tenant_id)
@@ -491,6 +546,13 @@ def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any
             db_provider=db_builtin_providers.get(provider.entity.identity.name),
             decrypt_credentials=False,
         )
+        if not _builtin_tool_provider_matches_filters(
+            user_provider,
+            query=query,
+            tags=tags,
+            language=language,
+        ):
+            continue
         ToolTransformService.repack_provider(tenant_id=tenant_id, provider=user_provider)
         builtin_providers.append(user_provider)
 
@@ -545,7 +607,9 @@ class PluginCategoryListApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, tenant_id: str, category: str):
-        args = PluginCategoryListQuery.model_validate(request.args.to_dict(flat=True))
+        args = PluginCategoryListQuery.model_validate(
+            {**request.args.to_dict(flat=True), "tags": request.args.getlist("tags")}
+        )
 
         try:
             plugin_category = PluginCategory(category)
@@ -553,13 +617,26 @@ class PluginCategoryListApi(Resource):
             return {"code": "invalid_param", "message": "invalid plugin category"}, 400
 
         try:
-            plugins = PluginService.list_by_category(tenant_id, plugin_category, args.page, args.page_size)
+            plugins = PluginService.list_by_category(
+                tenant_id,
+                plugin_category,
+                args.page,
+                args.page_size,
+                query=args.query,
+                tags=args.tags,
+                language=args.language,
+            )
         except PluginDaemonClientSideError as e:
             return {"code": "plugin_error", "message": e.description}, 400
 
         builtin_tools = []
         if plugin_category == PluginCategory.Tool:
-            builtin_tools = _list_hardcoded_builtin_tool_providers(tenant_id)
+            builtin_tools = _list_hardcoded_builtin_tool_providers(
+                tenant_id,
+                query=args.query,
+                tags=args.tags,
+                language=args.language,
+            )
 
         return dump_response(
             PluginCategoryListResponse,
@@ -571,6 +648,24 @@ class PluginCategoryListApi(Resource):
         )
 
 
+@console_ns.route("/workspaces/current/plugin/installed-ids")
+class PluginInstalledIdsApi(Resource):
+    @console_ns.doc(params=query_params_from_model(PluginInstalledIdsQuery))
+    @console_ns.response(200, "Success", console_ns.models[PluginInstalledIdsResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @with_current_tenant_id
+    def get(self, tenant_id: str):
+        args = PluginInstalledIdsQuery.model_validate(request.args.to_dict(flat=True))
+        try:
+            plugin_ids = PluginService.list_installed_plugin_ids(tenant_id, args.category)
+        except PluginDaemonClientSideError as e:
+            return {"code": "plugin_error", "message": e.description}, 400
+
+        return dump_response(PluginInstalledIdsResponse, {"plugin_ids": plugin_ids})
+
+
 @console_ns.route("/workspaces/current/plugin/list/latest-versions")
 class PluginListLatestVersionsApi(Resource):
     @console_ns.expect(console_ns.models[ParserLatest.__name__])
@@ -645,6 +740,7 @@ class PluginAssetApi(Resource):
 
 @console_ns.route("/workspaces/current/plugin/upload/pkg")
 class PluginUploadFromPkgApi(Resource):
+    @console_ns.doc(consumes=["multipart/form-data"], params=_PLUGIN_PACKAGE_UPLOAD_PARAMS)
     @console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
     @setup_required
     @login_required
@@ -1125,9 +1221,7 @@ class PluginFetchAutoUpgradeApi(Resource):
         args = ParserAutoUpgradeFetch.model_validate(request.args.to_dict(flat=True))
         auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id, args.category, session=db.session())
         auto_upgrade_dict = (
-            _auto_upgrade_settings_to_dict(auto_upgrade)
-            if auto_upgrade
-            else _default_auto_upgrade_settings(tenant_id, args.category)
+            _auto_upgrade_settings_to_dict(auto_upgrade) if auto_upgrade else _missing_auto_upgrade_settings(tenant_id)
         )
 
         return jsonable_encoder(

api/controllers/console/workspace/workspace.py

@@ -37,6 +37,7 @@ from controllers.console.wraps import (
     with_current_user,
 )
 from enums.cloud_plan import CloudPlan
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
@@ -233,7 +234,7 @@ class TenantListApi(Resource):
         tenants = [tenant for tenant, _ in tenant_rows]
         tenant_dicts = []
         is_enterprise_only = dify_config.ENTERPRISE_ENABLED and not dify_config.BILLING_ENABLED
-        is_saas = dify_config.EDITION == "CLOUD" and dify_config.BILLING_ENABLED
+        is_saas = dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and dify_config.BILLING_ENABLED
         tenant_plans: dict[str, SubscriptionPlan] = {}
 
         if is_saas:

api/controllers/console/wraps.py

@@ -20,6 +20,7 @@ from controllers.common.wraps import (
 from controllers.console.auth.error import AuthenticationFailedError, EmailCodeError
 from controllers.console.workspace.error import AccountNotInitializedError
 from enums.cloud_plan import CloudPlan
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.encryption import FieldEncryption
@@ -129,7 +130,7 @@ def account_initialization_required[R](view: Callable[..., R]) -> Callable[...,
 def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        if dify_config.EDITION != "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
             abort(404)
 
         return view(*args, **kwargs)
@@ -151,7 +152,7 @@ def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
 def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        if dify_config.EDITION != "SELF_HOSTED":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
             abort(404)
 
         return view(*args, **kwargs)
@@ -327,7 +328,7 @@ def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
         # The overloads keep Resource methods method-aware for pyrefly while
         # preserving support for plain functions used in tests and utilities.
         # check setup
-        if dify_config.EDITION == "SELF_HOSTED" and not _is_setup_completed():
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD and not _is_setup_completed():
             if os.environ.get("INIT_PASSWORD"):
                 raise NotInitValidateError()
             raise NotSetupError()

api/controllers/openapi/auth/data.py

@@ -8,6 +8,7 @@ from werkzeug.exceptions import InternalServerError
 
 from configs import dify_config
 from core.rbac import RBACPermission, RBACResourceScope
+from enums.deployment_edition import DeploymentEdition
 from libs.oauth_bearer import Scope, TokenType
 from models.account import Account, Tenant, TenantAccountRole
 from models.model import App, EndUser
@@ -26,7 +27,7 @@ class CallerKind(StrEnum):
 
 
 def current_edition() -> Edition:
-    if dify_config.EDITION == "CLOUD":
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
         return Edition.SAAS
     if dify_config.ENTERPRISE_ENABLED:
         return Edition.EE

api/controllers/service_api/app/audio.py

@@ -101,7 +101,7 @@ class AudioApi(Resource):
 
         Accepts an audio file upload and returns the transcribed text.
         """
-        file = request.files["file"]
+        file = request.files.get("file")
 
         try:
             response = AudioService.transcript_asr(

api/controllers/service_api/dataset/dataset.py

@@ -531,14 +531,22 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        if payload.permission == DatasetPermissionEnum.ALL_TEAM and dify_config.RBAC_ENABLED:
-            RBACService.DatasetAccess.replace_whitelist(
-                tenant_id,
-                current_user.id,
-                dataset.id,
-                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
-            )
-            initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+        if dify_config.RBAC_ENABLED:
+            if payload.permission == DatasetPermissionEnum.ALL_TEAM:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+                )
+                initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+            else:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.SPECIFIC),
+                )
 
         return _dump_service_dataset_detail(dataset, session=session), 200
 

api/controllers/web/audio.py

@@ -76,7 +76,7 @@ class AudioApi(WebApiResource):
     @web_ns.response(200, "Success", web_ns.models[AudioToTextResponse.__name__])
     def post(self, app_model: App, end_user: EndUser):
         """Convert audio to text"""
-        file = request.files["file"]
+        file = request.files.get("file")
 
         try:
             response = AudioService.transcript_asr(

api/controllers/web/site.py

@@ -1,6 +1,6 @@
 from typing import Any, Self
 
-from pydantic import AliasChoices, Field, computed_field
+from pydantic import AliasChoices, Field
 from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
@@ -8,12 +8,15 @@ from configs import dify_config
 from controllers.common.schema import register_response_schema_models
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
+from enums.deployment_edition import DeploymentEdition
 from extensions.ext_database import db
+from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
 from models.account import Tenant, TenantStatus
-from models.model import App, EndUser, Site
+from models.model import App, EndUser, IconType, Site
 from services.feature_service import FeatureModel, FeatureService
+from services.file_service import FileService
 
 
 class WebSiteResponse(ResponseModel):
@@ -32,11 +35,7 @@ class WebSiteResponse(ResponseModel):
     prompt_public: bool | None = None
     show_workflow_steps: bool | None = None
     use_icon_as_answer_icon: bool | None = None
-
-    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
-    @property
-    def icon_url(self) -> str | None:
-        return build_icon_url(self.icon_type, self.icon)
+    icon_url: str | None = None
 
 
 class WebModelConfigResponse(ResponseModel):
@@ -88,6 +87,7 @@ class WebAppSiteResponse(ResponseModel):
         end_user_id: str | None,
         features: FeatureModel,
         can_replace_logo: bool,
+        icon_url: str | None = None,
     ) -> Self:
         custom_config = None
         if can_replace_logo:
@@ -102,6 +102,7 @@ class WebAppSiteResponse(ResponseModel):
             )
 
         site_response = WebSiteResponse.model_validate(site, from_attributes=True)
+        site_response.icon_url = icon_url if icon_url is not None else build_icon_url(site.icon_type, site.icon)
         if features.billing.enabled and not features.webapp_copyright_enabled:
             site_response.copyright = None
             site_response.input_placeholder = None
@@ -123,6 +124,17 @@ register_response_schema_models(
 )
 
 
+def _build_site_icon_url(*, site: Site, tenant_id: str) -> str | None:
+    """Use direct S3 URLs only in Cloud Mode and preserve preview URLs elsewhere."""
+    if site.icon_type != IconType.IMAGE or not site.icon:
+        return None
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and (
+        StorageType(dify_config.STORAGE_TYPE) == StorageType.S3
+    ):
+        return FileService(db.engine).get_file_presigned_url(file_id=site.icon, tenant_id=tenant_id)
+    return build_icon_url(site.icon_type, site.icon)
+
+
 @web_ns.route("/site")
 class AppSiteApi(WebApiResource):
     @web_ns.doc("Get App Site Info")
@@ -159,4 +171,5 @@ class AppSiteApi(WebApiResource):
             end_user_id=end_user.id,
             features=features,
             can_replace_logo=features.can_replace_logo,
+            icon_url=_build_site_icon_url(site=site, tenant_id=tenant.id),
         ).model_dump(mode="json")

api/controllers/web/wraps.py

@@ -11,6 +11,7 @@ from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 from constants import HEADER_NAME_APP_CODE
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
+from core.logging.context import set_identity_context
 from extensions.ext_database import db
 from libs.passport import PassportService
 from libs.token import extract_webapp_passport
@@ -28,6 +29,11 @@ def validate_jwt_token[**P, R](
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
             app_model, end_user = decode_jwt_token()
+            set_identity_context(
+                tenant_id=end_user.tenant_id,
+                user_id=end_user.id,
+                user_type=end_user.type or "end_user",
+            )
             return view(app_model, end_user, *args, **kwargs)
 
         return decorated

api/core/agent/publish_visibility.py

@@ -0,0 +1,93 @@
+"""Publication visibility rules for calling roster Agents from Workflows.
+
+``Agent.active_config_is_published`` describes whether the editable shared
+draft still matches the active snapshot. It is false both before the first
+publish and after a published Agent receives new draft edits, so it must not be
+used as a runtime availability flag. App-backed Agents are callable from a
+Workflow only when the active snapshot has a revision created by a
+publish-visible operation. Direct roster Agents are publish-visible by
+construction and only need an active snapshot.
+"""
+
+from sqlalchemy import and_, or_, select
+from sqlalchemy.orm import Session
+from sqlalchemy.sql.elements import ColumnElement
+
+from models.agent import Agent, AgentConfigRevision, AgentConfigRevisionOperation, AgentScope, AgentSource
+
+PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS = frozenset(
+    {
+        AgentConfigRevisionOperation.PUBLISH_DRAFT,
+        AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
+        AgentConfigRevisionOperation.SAVE_NEW_VERSION,
+        AgentConfigRevisionOperation.SAVE_NEW_AGENT,
+        AgentConfigRevisionOperation.SAVE_TO_ROSTER,
+        AgentConfigRevisionOperation.RESTORE_VERSION,
+    }
+)
+
+
+def workflow_callable_active_snapshot_filter() -> ColumnElement[bool]:
+    """Return the SQL predicate for an Agent with a Workflow-callable active snapshot.
+
+    The caller remains responsible for tenant, roster scope, lifecycle status,
+    and model configuration filters. The correlated revision lookup makes the
+    predicate safe to compose into roster pagination queries.
+    """
+
+    app_backed_agent = or_(
+        Agent.source == AgentSource.AGENT_APP,
+        and_(
+            Agent.source == AgentSource.IMPORTED,
+            Agent.scope == AgentScope.ROSTER,
+            Agent.app_id.is_not(None),
+        ),
+    )
+    publish_visible_revision_exists = (
+        select(AgentConfigRevision.id)
+        .where(
+            AgentConfigRevision.tenant_id == Agent.tenant_id,
+            AgentConfigRevision.agent_id == Agent.id,
+            AgentConfigRevision.current_snapshot_id == Agent.active_config_snapshot_id,
+            AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
+        )
+        .correlate(Agent)
+        .exists()
+    )
+    return and_(
+        Agent.active_config_snapshot_id.is_not(None),
+        or_(
+            ~app_backed_agent,
+            publish_visible_revision_exists,
+        ),
+    )
+
+
+def agent_has_workflow_callable_active_snapshot(*, session: Session, agent: Agent) -> bool:
+    """Return whether ``agent`` has an active snapshot visible to Workflow.
+
+    This object-level form is useful after ownership and lifecycle checks have
+    already loaded an Agent. It intentionally ignores dirty draft state so a
+    previously published snapshot keeps serving while later edits remain
+    unpublished.
+    """
+
+    if not agent.active_config_snapshot_id:
+        return False
+    is_app_backed = agent.source == AgentSource.AGENT_APP or (
+        agent.source == AgentSource.IMPORTED and agent.scope == AgentScope.ROSTER and agent.app_id is not None
+    )
+    if not is_app_backed:
+        return True
+    return bool(
+        session.scalar(
+            select(AgentConfigRevision.id)
+            .where(
+                AgentConfigRevision.tenant_id == agent.tenant_id,
+                AgentConfigRevision.agent_id == agent.id,
+                AgentConfigRevision.current_snapshot_id == agent.active_config_snapshot_id,
+                AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
+            )
+            .limit(1)
+        )
+    )

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

@@ -625,7 +625,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 message=message_snapshot,
                 user=user,
                 stream=stream,
-                draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from, account=user),
+                draft_var_saver_factory=self._get_draft_var_saver_factory(
+                    invoke_from,
+                    account=user,
+                    tenant_id=application_generate_entity.app_config.tenant_id,
+                ),
             )
 
             return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)

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

@@ -540,6 +540,9 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                 base_url=dify_config.AGENT_BACKEND_BASE_URL,
                 use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                 fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
+                stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+                stream_max_reconnects=dify_config.AGENT_BACKEND_STREAM_MAX_RECONNECTS,
+                stream_run_timeout_seconds=dify_config.AGENT_BACKEND_RUN_TIMEOUT_SECONDS,
             ),
             event_adapter=AgentBackendRunEventAdapter(),
             session_store=AgentAppRuntimeSessionStore(),
@@ -679,42 +682,27 @@ class AgentAppGenerator(MessageBasedAppGenerator):
             if draft_type == AgentConfigDraftType.DEBUG_BUILD.value
             else AgentConfigDraftType.DRAFT
         )
+        if effective_draft_type == AgentConfigDraftType.DRAFT:
+            from services.agent.composer_service import AgentComposerService
+
+            return AgentComposerService.get_or_create_normal_agent_draft(
+                session=session,
+                tenant_id=tenant_id,
+                agent=agent,
+                created_by=agent.updated_by or agent.created_by,
+            )
+        if not account_id:
+            raise AgentAppGeneratorError("Build draft requires an account user")
         stmt = select(AgentConfigDraft).where(
             AgentConfigDraft.tenant_id == tenant_id,
             AgentConfigDraft.agent_id == agent.id,
-            AgentConfigDraft.draft_type == effective_draft_type,
+            AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
+            AgentConfigDraft.account_id == account_id,
         )
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            if not account_id:
-                raise AgentAppGeneratorError("Build draft requires an account user")
-            stmt = stmt.where(AgentConfigDraft.account_id == account_id)
-        else:
-            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
         draft = session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))
         if draft is not None:
             return draft
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            raise AgentAppGeneratorError("Agent build draft not found")
-        _, snapshot, agent_soul = AgentAppGenerator._resolve_agent_by_id(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
-            session=session,
-        )
-        draft = AgentConfigDraft(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            draft_type=AgentConfigDraftType.DRAFT,
-            account_id=None,
-            draft_owner_key="",
-            base_snapshot_id=snapshot.id,
-            config_snapshot=agent_soul,
-            created_by=agent.created_by,
-            updated_by=agent.updated_by,
-        )
-        session.add(draft)
-        session.flush()
-        return draft
+        raise AgentAppGeneratorError("Agent build draft not found")
 
     @staticmethod
     def _resolve_agent_by_id(

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

@@ -941,48 +941,64 @@ class AgentAppRunner:
             if pending_text:
                 persist_answer_text(pending_text)
 
-        for public_event in self._agent_backend_client.stream_events(run_id):
-            if queue_manager.is_stopped():
-                flush_pending_agent_message_text()
-                self._cancel_run(run_id)
-                raise GenerateTaskStoppedError()
-            for internal_event in self._event_adapter.adapt(public_event):
+        try:
+            public_events = self._agent_backend_client.stream_events(
+                run_id,
+                should_stop=queue_manager.is_stopped,
+            )
+            for public_event in public_events:
                 if queue_manager.is_stopped():
                     flush_pending_agent_message_text()
                     self._cancel_run(run_id)
                     raise GenerateTaskStoppedError()
-                if internal_event.type in (
-                    AgentBackendInternalEventType.RUN_STARTED,
-                    AgentBackendInternalEventType.STREAM_EVENT,
-                    AgentBackendInternalEventType.AGENT_MESSAGE_DELTA,
-                ):
-                    if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
-                        debounced_delta = text_delta_debouncer.push(internal_event.delta)
-                        if debounced_delta:
-                            persist_answer_text(debounced_delta)
-                        continue
-
-                    if isinstance(internal_event, AgentBackendStreamInternalEvent):
+                for internal_event in self._event_adapter.adapt(public_event):
+                    if queue_manager.is_stopped():
                         flush_pending_agent_message_text()
-                        try:
-                            process_recorder.handle_stream_event(internal_event)
-                        except Exception:
-                            db.session.rollback()
-                            logger.warning(
-                                "Failed to persist Agent App process event: run_id=%s message_id=%s event_kind=%s",
-                                run_id,
-                                message_id,
-                                internal_event.event_kind,
-                                exc_info=True,
-                            )
+                        self._cancel_run(run_id)
+                        raise GenerateTaskStoppedError()
+                    if internal_event.type in (
+                        AgentBackendInternalEventType.RUN_STARTED,
+                        AgentBackendInternalEventType.STREAM_EVENT,
+                        AgentBackendInternalEventType.AGENT_MESSAGE_DELTA,
+                    ):
+                        if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
+                            debounced_delta = text_delta_debouncer.push(internal_event.delta)
+                            if debounced_delta:
+                                persist_answer_text(debounced_delta)
+                            continue
+
+                        if isinstance(internal_event, AgentBackendStreamInternalEvent):
+                            flush_pending_agent_message_text()
+                            try:
+                                process_recorder.handle_stream_event(internal_event)
+                            except Exception:
+                                db.session.rollback()
+                                logger.warning(
+                                    "Failed to persist Agent App process event: run_id=%s message_id=%s event_kind=%s",
+                                    run_id,
+                                    message_id,
+                                    internal_event.event_kind,
+                                    exc_info=True,
+                                )
+                            continue
                         continue
-                    continue
-                flush_pending_agent_message_text()
-                terminal = internal_event
-                break
-            if terminal is not None:
-                break
+                    flush_pending_agent_message_text()
+                    terminal = internal_event
+                    break
+                if terminal is not None:
+                    break
+        except GenerateTaskStoppedError:
+            raise
+        except Exception as error:
+            flush_pending_agent_message_text()
+            self._cancel_run(run_id)
+            if queue_manager.is_stopped():
+                raise GenerateTaskStoppedError() from error
+            raise
         flush_pending_agent_message_text()
+        if queue_manager.is_stopped():
+            self._cancel_run(run_id)
+            raise GenerateTaskStoppedError()
         return terminal, process_recorder
 
     def _cancel_run(self, run_id: str) -> None:

api/core/app/apps/base_app_generator.py

@@ -32,6 +32,7 @@ class _DebuggerDraftVariableSaver:
         self,
         *,
         account: Account,
+        tenant_id: str,
         app_id: str,
         node_id: str,
         node_type: NodeType,
@@ -39,6 +40,7 @@ class _DebuggerDraftVariableSaver:
         enclosing_node_id: str | None = None,
     ) -> None:
         self._account = account
+        self._tenant_id = tenant_id
         self._app_id = app_id
         self._node_id = node_id
         self._node_type = node_type
@@ -49,6 +51,7 @@ class _DebuggerDraftVariableSaver:
         with Session(db.engine) as session, session.begin():
             DraftVariableSaverImpl(
                 session=session,
+                tenant_id=self._tenant_id,
                 app_id=self._app_id,
                 node_id=self._node_id,
                 node_type=self._node_type,
@@ -287,7 +290,12 @@ class BaseAppGenerator:
 
     @final
     @staticmethod
-    def _get_draft_var_saver_factory(invoke_from: InvokeFrom, account: Account | EndUser) -> DraftVariableSaverFactory:
+    def _get_draft_var_saver_factory(
+        invoke_from: InvokeFrom,
+        account: Account | EndUser,
+        *,
+        tenant_id: str,
+    ) -> DraftVariableSaverFactory:
         if invoke_from == InvokeFrom.DEBUGGER:
             assert isinstance(account, Account)
 
@@ -300,6 +308,7 @@ class BaseAppGenerator:
             ) -> DraftVariableSaver:
                 return _DebuggerDraftVariableSaver(
                     account=account,
+                    tenant_id=tenant_id,
                     app_id=app_id,
                     node_id=node_id,
                     node_type=node_type,

api/core/app/apps/base_app_queue_manager.py

@@ -21,6 +21,7 @@ from core.app.entities.queue_entities import (
     WorkflowQueueMessage,
 )
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
 from graphon.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)
@@ -51,6 +52,9 @@ class AppQueueManager(ABC):
         self._graph_runtime_state: GraphRuntimeState | None = None
         self._stopped_cache: TTLCache[tuple, bool] = TTLCache(maxsize=1, ttl=1)
         self._cache_lock = threading.Lock()
+        self._execution_terminal = threading.Event()
+        self._abort_sent = threading.Event()
+        self._lifecycle_lock = threading.Lock()
 
     def listen(self):
         """
@@ -59,7 +63,7 @@ class AppQueueManager(ABC):
         """
         # wait for APP_MAX_EXECUTION_TIME seconds to stop listen
         listen_timeout = dify_config.APP_MAX_EXECUTION_TIME
-        start_time = time.time()
+        start_time = time.monotonic()
         last_ping_time: int | float = 0
         try:
             while True:
@@ -72,8 +76,14 @@ class AppQueueManager(ABC):
                 except queue.Empty:
                     continue
                 finally:
-                    elapsed_time = time.time() - start_time
-                    if elapsed_time >= listen_timeout or self._is_stopped():
+                    elapsed_time = time.monotonic() - start_time
+                    timed_out = elapsed_time >= listen_timeout
+                    manually_stopped = self._is_stopped()
+                    if not self._execution_terminal.is_set() and (timed_out or manually_stopped):
+                        reason = (
+                            f"App execution exceeded {listen_timeout} seconds" if timed_out else "App task was stopped"
+                        )
+                        self._abort_execution(reason)
                         # publish two messages to make sure the client can receive the stop signal
                         # and stop listening after the stop signal processed
                         self.publish(
@@ -84,16 +94,33 @@ class AppQueueManager(ABC):
                         self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
                         last_ping_time = elapsed_time // 10
         finally:
+            if not self._execution_terminal.is_set():
+                self._abort_execution("Client response stream closed before app execution completed")
             self._graph_runtime_state = None  # Release reference once consumers finish or close the generator.
 
-    def stop_listen(self):
+    def stop_listen(self, *, execution_terminal: bool = False):
         """
         Stop listen to queue
         :return:
         """
+        if execution_terminal:
+            self._execution_terminal.set()
         self._clear_task_belong_cache()
         self._q.put(None)
 
+    def _abort_execution(self, reason: str) -> None:
+        """Propagate response timeout/disconnect to legacy and GraphEngine runners."""
+        with self._lifecycle_lock:
+            if self._execution_terminal.is_set() or self._abort_sent.is_set():
+                return
+            self._abort_sent.set()
+
+        try:
+            self.set_stop_flag_no_user_check(self._task_id)
+            GraphEngineManager(redis_client).send_stop_command(self._task_id, reason=reason)
+        except Exception:
+            logger.exception("Failed to abort app execution for task %s", self._task_id)
+
     def _clear_task_belong_cache(self) -> None:
         """
         Remove the task belong cache key once listening is finished.

api/core/app/apps/message_based_app_queue_manager.py

@@ -45,7 +45,7 @@ class MessageBasedAppQueueManager(AppQueueManager):
         if isinstance(
             event, QueueStopEvent | QueueErrorEvent | QueueMessageEndEvent | QueueAdvancedChatMessageEndEvent
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
             if self._app_mode == AppMode.ADVANCED_CHAT.value:

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

@@ -349,6 +349,7 @@ class PipelineGenerator(BaseAppGenerator):
             draft_var_saver_factory = self._get_draft_var_saver_factory(
                 invoke_from,
                 user,
+                tenant_id=pipeline.tenant_id,
             )
             # return response or stream generator
             response = self._handle_response(

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

@@ -42,7 +42,7 @@ class PipelineQueueManager(AppQueueManager):
             | QueueWorkflowFailedEvent
             | QueueWorkflowPartialSuccessEvent,
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
             raise GenerateTaskStoppedError()

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

@@ -399,7 +399,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
 
             worker_thread.start()
 
-            draft_var_saver_factory = self._get_draft_var_saver_factory(invoke_from, user)
+            draft_var_saver_factory = self._get_draft_var_saver_factory(
+                invoke_from,
+                user,
+                tenant_id=app_model.tenant_id,
+            )
 
             # return response or stream generator
             response = self._handle_response(

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

@@ -41,4 +41,4 @@ class WorkflowAppQueueManager(AppQueueManager):
             | QueueWorkflowFailedEvent
             | QueueWorkflowPartialSuccessEvent,
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)

api/core/app/apps/workflow_app_runner.py

@@ -26,6 +26,7 @@ from core.app.entities.queue_entities import (
     QueueNodeSucceededEvent,
     QueueReasoningChunkEvent,
     QueueRetrieverResourcesEvent,
+    QueueStopEvent,
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
     QueueWorkflowPartialSuccessEvent,
@@ -424,7 +425,12 @@ class WorkflowBasedAppRunner:
                     QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
                 )
             case GraphRunAbortedEvent():
-                self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+                self._publish_event(
+                    QueueStopEvent(
+                        stopped_by=QueueStopEvent.StopBy.USER_MANUAL,
+                        reason=event.reason or "Workflow execution aborted",
+                    )
+                )
             case GraphRunPausedEvent():
                 runtime_state = workflow_entry.graph_engine.graph_runtime_state
                 paused_nodes = runtime_state.get_paused_nodes()

api/core/app/entities/queue_entities.py

@@ -500,11 +500,15 @@ class QueueStopEvent(AppQueueEvent):
 
     event: QueueEvent = QueueEvent.STOP
     stopped_by: StopBy
+    reason: str | None = None
 
     def get_stop_reason(self) -> str:
         """
         To stop reason
         """
+        if self.reason:
+            return self.reason
+
         reason_mapping = {
             QueueStopEvent.StopBy.USER_MANUAL: "Stopped by user.",
             QueueStopEvent.StopBy.ANNOTATION_REPLY: "Stopped by annotation reply.",

api/core/helper/ssrf_proxy.py

@@ -47,6 +47,24 @@ class MaxRetriesExceededError(ValueError):
     pass
 
 
+class ResponseLimitError(ValueError):
+    """Base error for responses that cannot be safely bounded."""
+
+    pass
+
+
+class ResponseTooLargeError(ResponseLimitError):
+    """Raised when an identity response exceeds the configured byte limit."""
+
+    pass
+
+
+class UnsupportedResponseEncodingError(ResponseLimitError):
+    """Raised when response encoding prevents safe decoded-size enforcement."""
+
+    pass
+
+
 request_error = httpx.RequestError
 max_retries_exceeded_error = MaxRetriesExceededError
 
@@ -142,7 +160,31 @@ def _inject_trace_headers(headers: Headers | None) -> Headers:
     return headers
 
 
-def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+def make_request(
+    method: str,
+    url: str,
+    max_retries: int = SSRF_DEFAULT_MAX_RETRIES,
+    stream_response: bool = False,
+    **kwargs: Any,
+) -> httpx.Response:
+    """Send one SSRF-protected request with optional streaming.
+
+    Args:
+        method: HTTP method sent through the configured SSRF client.
+        url: Absolute request URL.
+        max_retries: Number of retry attempts after the initial request.
+        stream_response: Return an open streaming response that the caller must close.
+        **kwargs: Additional keyword arguments forwarded to ``httpx.Client``.
+
+    Returns:
+        A buffered response, or an open response when ``stream_response`` is true.
+
+    Raises:
+        ToolSSRFError: The configured SSRF proxy rejects the destination.
+        MaxRetriesExceededError: All configured request attempts fail.
+        httpx.RequestError: A request fails while retries are disabled.
+        ValueError: The SSL verification option or request headers are invalid.
+    """
     # Convert requests-style allow_redirects to httpx-style follow_redirects
     if "allow_redirects" in kwargs:
         allow_redirects = kwargs.pop("allow_redirects")
@@ -175,6 +217,11 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
     # When using a forward proxy, httpx may override the Host header based on the URL.
     # We extract and preserve any explicitly set Host header to support virtual hosting.
     user_provided_host = _get_user_provided_host_header(headers)
+    send_kwargs: dict[str, Any] = {}
+    if "auth" in kwargs:
+        send_kwargs["auth"] = kwargs.pop("auth")
+    if "follow_redirects" in kwargs:
+        send_kwargs["follow_redirects"] = kwargs.pop("follow_redirects")
 
     retries = 0
     while retries <= max_retries:
@@ -185,7 +232,11 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
             if user_provided_host is not None:
                 headers["host"] = user_provided_host
             kwargs["headers"] = headers
-            response = client.request(method=method, url=url, **kwargs)
+            request = client.build_request(method=method, url=url, **kwargs)
+            if stream_response:
+                response = client.send(request, stream=True, **send_kwargs)
+            else:
+                response = client.send(request, **send_kwargs)
 
             # Check for SSRF protection by Squid proxy
             if response.status_code in (401, 403):
@@ -195,6 +246,7 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
 
                 # Squid typically identifies itself in Server or Via headers
                 if "squid" in server_header or "squid" in via_header:
+                    response.close()
                     raise ToolSSRFError(
                         f"Access to '{url}' was blocked by SSRF protection. "
                         f"The URL may point to a private or local network address. "
@@ -208,6 +260,7 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
                     response.status_code,
                     url,
                 )
+                response.close()
 
         except httpx.RequestError as e:
             logger.warning("Request to URL %s failed on attempt %s: %s", url, retries + 1, e)
@@ -220,6 +273,42 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
     raise MaxRetriesExceededError(f"Reached maximum retries ({max_retries}) for URL {url}")
 
 
+def buffer_response(response: httpx.Response, *, max_response_bytes: int) -> httpx.Response:
+    """Consume one open identity response under a decoded byte limit and close its stream."""
+    if max_response_bytes <= 0:
+        raise ValueError("max_response_bytes must be positive")
+
+    try:
+        content_encoding = response.headers.get("content-encoding", "identity").strip().lower()
+        if content_encoding not in {"", "identity"}:
+            raise UnsupportedResponseEncodingError(f"content encoding {content_encoding} cannot be safely bounded")
+        content = bytearray()
+        for chunk in response.iter_bytes():
+            if len(content) + len(chunk) > max_response_bytes:
+                raise ResponseTooLargeError(f"response exceeded {max_response_bytes} bytes")
+            content.extend(chunk)
+        decoded_headers = {
+            name: value
+            for name, value in response.headers.items()
+            if name.lower() not in {"content-encoding", "content-length", "transfer-encoding"}
+        }
+        try:
+            request = response.request
+        except RuntimeError:
+            request = None
+        return httpx.Response(
+            response.status_code,
+            headers=decoded_headers,
+            content=bytes(content),
+            request=request,
+            extensions=response.extensions,
+            history=response.history,
+            default_encoding=response.default_encoding,
+        )
+    finally:
+        response.close()
+
+
 def get(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("GET", url, max_retries=max_retries, **kwargs)
 

api/core/hosting_configuration.py

@@ -6,6 +6,7 @@ from pydantic import BaseModel
 from configs import dify_config
 from core.entities import DEFAULT_PLUGIN_ID
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, RestrictModel
+from enums.deployment_edition import DeploymentEdition
 from graphon.model_runtime.entities.model_entities import ModelType
 
 
@@ -49,7 +50,7 @@ class HostingConfiguration:
         self.moderation_config = None
 
     def init_app(self, app: Flask):
-        if dify_config.EDITION != "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
             return
 
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/azure_openai/azure_openai"] = self.init_azure_openai()

api/core/indexing_runner.py

@@ -21,6 +21,7 @@ from core.model_manager import ModelInstance, ModelManager
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
@@ -113,17 +114,25 @@ class IndexingRunner:
                     current_user=current_user,
                     session=session,
                 )
+                token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
+                total_tokens = sum(token_counts)
                 # save segment
-                self._load_segments(dataset, requeried_document, documents, session)
+                self._load_segments(
+                    session=session,
+                    dataset=dataset,
+                    dataset_document=requeried_document,
+                    documents=documents,
+                    token_counts=token_counts,
+                )
                 session.commit()
 
                 # load
                 self._load(
-                    index_processor=index_processor,
+                    session=session,
                     dataset=dataset,
                     dataset_document=requeried_document,
                     documents=documents,
-                    session=session,
+                    total_tokens=total_tokens,
                 )
             except DocumentIsPausedError:
                 raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -190,17 +199,25 @@ class IndexingRunner:
                 current_user=current_user,
                 session=session,
             )
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
+            total_tokens = sum(token_counts)
             # save segment
-            self._load_segments(dataset, requeried_document, documents, session)
+            self._load_segments(
+                session=session,
+                dataset=dataset,
+                dataset_document=requeried_document,
+                documents=documents,
+                token_counts=token_counts,
+            )
             session.commit()
 
             # load
             self._load(
-                index_processor=index_processor,
+                session=session,
                 dataset=dataset,
                 dataset_document=requeried_document,
                 documents=documents,
-                session=session,
+                total_tokens=total_tokens,
             )
         except DocumentIsPausedError:
             raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -225,7 +242,7 @@ class IndexingRunner:
             if not dataset:
                 raise ValueError("no dataset found")
 
-            # get exist document_segment list and delete
+            # get existing document segments
             document_segments = session.scalars(
                 select(DocumentSegment).where(
                     DocumentSegment.dataset_id == dataset.id,
@@ -264,15 +281,15 @@ class IndexingRunner:
                                     child_documents.append(child_document)
                                 document.children = child_documents
                         documents.append(document)
+            # Preserve the full document total even when only incomplete segments are re-indexed.
+            total_tokens = sum(document_segment.tokens for document_segment in document_segments)
             # build index
-            index_type = requeried_document.doc_form
-            index_processor = IndexProcessorFactory(index_type).init_index_processor()
             self._load(
-                index_processor=index_processor,
+                session=session,
                 dataset=dataset,
                 dataset_document=requeried_document,
                 documents=documents,
-                session=session,
+                total_tokens=total_tokens,
             )
         except DocumentIsPausedError:
             raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -601,28 +618,16 @@ class IndexingRunner:
 
     def _load(
         self,
-        index_processor: BaseIndexProcessor,
+        session: Session,
         dataset: Dataset,
         dataset_document: DatasetDocument,
         documents: list[Document],
-        session: Session,
-    ):
-        """
-        insert index and update document/segment status to completed
-        """
+        total_tokens: int,
+    ) -> None:
+        """Build indexes and mark the document complete using the token total computed before hash sharding."""
 
-        embedding_model_instance = None
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
-            embedding_model_instance = self._get_model_manager(dataset.tenant_id).get_model_instance(
-                tenant_id=dataset.tenant_id,
-                provider=dataset.embedding_model_provider,
-                model_type=ModelType.TEXT_EMBEDDING,
-                model=dataset.embedding_model,
-            )
-
-        # chunk nodes by chunk size
+        # Build indexes using the existing hash-based worker groups.
         indexing_start_at = time.perf_counter()
-        tokens = 0
         create_keyword_thread = None
         if (
             dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
@@ -659,12 +664,11 @@ class IndexingRunner:
                             chunk_documents,
                             dataset.id,
                             dataset_document.id,
-                            embedding_model_instance,
                         )
                     )
 
                 for future in futures:
-                    tokens += future.result()
+                    future.result()
         if (
             dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
             and dataset.indexing_technique == IndexTechniqueType.ECONOMY
@@ -679,7 +683,7 @@ class IndexingRunner:
             document_id=dataset_document.id,
             after_indexing_status=IndexingStatus.COMPLETED,
             extra_update_params={
-                DatasetDocument.tokens: tokens,
+                DatasetDocument.tokens: total_tokens,
                 DatasetDocument.completed_at: naive_utc_now(),
                 DatasetDocument.indexing_latency: indexing_end_at - indexing_start_at,
                 DatasetDocument.error: None,
@@ -720,8 +724,7 @@ class IndexingRunner:
         chunk_documents: list[Document],
         dataset_id: str,
         dataset_document_id: str,
-        embedding_model_instance: ModelInstance | None,
-    ):
+    ) -> None:
         with flask_app.app_context():
             with session_factory.create_session() as session:
                 dataset = session.get(Dataset, dataset_id)
@@ -735,11 +738,6 @@ class IndexingRunner:
                 # check document is paused
                 self._check_document_paused_status(dataset_document.id)
 
-                tokens = 0
-                if embedding_model_instance:
-                    page_content_list = [document.page_content for document in chunk_documents]
-                    tokens += sum(embedding_model_instance.get_text_embedding_num_tokens(page_content_list))
-
                 multimodal_documents = []
                 for document in chunk_documents:
                     if document.attachments and dataset.is_multimodal:
@@ -773,8 +771,6 @@ class IndexingRunner:
 
                 session.commit()
 
-                return tokens
-
     @staticmethod
     def _check_document_paused_status(document_id: str):
         indexing_cache_key = f"document_{document_id}_is_paused"
@@ -864,8 +860,14 @@ class IndexingRunner:
         return documents
 
     def _load_segments(
-        self, dataset: Dataset, dataset_document: DatasetDocument, documents: list[Document], session: Session
-    ):
+        self,
+        session: Session,
+        dataset: Dataset,
+        dataset_document: DatasetDocument,
+        documents: list[Document],
+        token_counts: list[int],
+    ) -> None:
+        """Persist transformed documents and their precomputed token counts before indexing starts."""
         # save node to document segment
         doc_store = DatasetDocumentStore(
             dataset=dataset, user_id=dataset_document.created_by, document_id=dataset_document.id
@@ -873,9 +875,10 @@ class IndexingRunner:
 
         # add document segments
         doc_store.add_documents(
+            session=session,
             docs=documents,
             save_child=dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX,
-            session=session,
+            token_counts=token_counts,
         )
 
         # update document status to indexing
@@ -900,7 +903,6 @@ class IndexingRunner:
                 DocumentSegment.indexing_at: naive_utc_now(),
             },
         )
-        pass
 
 
 class DocumentIsPausedError(Exception):

api/core/logging/context.py

@@ -6,9 +6,21 @@ using Python's contextvars for thread-safe and async-safe storage.
 
 import uuid
 from contextvars import ContextVar
+from typing import NamedTuple
+
+
+class IdentityContext(NamedTuple):
+    """Immutable identity values captured for logging."""
+
+    tenant_id: str
+    user_id: str
+    user_type: str
+
 
 _request_id: ContextVar[str] = ContextVar("log_request_id", default="")
 _trace_id: ContextVar[str] = ContextVar("log_trace_id", default="")
+_EMPTY_IDENTITY_CONTEXT = IdentityContext(tenant_id="", user_id="", user_type="")
+_identity: ContextVar[IdentityContext] = ContextVar("log_identity", default=_EMPTY_IDENTITY_CONTEXT)
 
 
 def get_request_id() -> str:
@@ -21,15 +33,35 @@ def get_trace_id() -> str:
     return _trace_id.get()
 
 
+def get_identity_context() -> IdentityContext:
+    """Get the immutable tenant, user, and user-type snapshot for logging."""
+    return _identity.get()
+
+
+def set_identity_context(
+    *, tenant_id: str | None = None, user_id: str | None = None, user_type: str | None = None
+) -> None:
+    """Set primitive identity values already resolved by an authentication boundary."""
+    _identity.set(
+        IdentityContext(
+            tenant_id=tenant_id or "",
+            user_id=user_id or "",
+            user_type=user_type or "",
+        )
+    )
+
+
 def init_request_context() -> None:
-    """Initialize request context. Call at start of each request."""
+    """Initialize request context and discard identity left by earlier work."""
     req_id = uuid.uuid4().hex[:10]
     trace_id = uuid.uuid5(uuid.NAMESPACE_DNS, req_id).hex
     _request_id.set(req_id)
     _trace_id.set(trace_id)
+    _identity.set(_EMPTY_IDENTITY_CONTEXT)
 
 
 def clear_request_context() -> None:
-    """Clear request context. Call at end of request (optional)."""
+    """Clear request context at a request or task lifecycle boundary."""
     _request_id.set("")
     _trace_id.set("")
+    _identity.set(_EMPTY_IDENTITY_CONTEXT)

api/core/logging/filters.py

@@ -4,10 +4,7 @@ import contextlib
 import logging
 from typing import override
 
-import flask
-
-from core.logging.context import get_request_id, get_trace_id
-from core.logging.structured_formatter import IdentityDict
+from core.logging.context import get_identity_context, get_request_id, get_trace_id
 
 
 class TraceContextFilter(logging.Filter):
@@ -51,49 +48,16 @@ class TraceContextFilter(logging.Filter):
 
 
 class IdentityContextFilter(logging.Filter):
-    """
-    Filter that adds user identity context to log records.
-    Extracts tenant_id, user_id, and user_type from Flask-Login current_user.
+    """Add an identity snapshot without invoking authentication or database work.
+
+    Logging can run while other libraries hold internal locks, so this filter must
+    only read primitive ContextVar values populated by authentication boundaries.
     """
 
     @override
     def filter(self, record: logging.LogRecord) -> bool:
-        identity = self._extract_identity()
-        record.tenant_id = identity.get("tenant_id", "")
-        record.user_id = identity.get("user_id", "")
-        record.user_type = identity.get("user_type", "")
+        identity = get_identity_context()
+        record.tenant_id = identity.tenant_id
+        record.user_id = identity.user_id
+        record.user_type = identity.user_type
         return True
-
-    def _extract_identity(self) -> IdentityDict:
-        """Extract identity from current_user if in request context."""
-        try:
-            if not flask.has_request_context():
-                return {}
-            from flask_login import current_user
-
-            # Check if user is authenticated using the proxy
-            if not current_user.is_authenticated:
-                return {}
-
-            # Access the underlying user object
-            user = current_user
-
-            from models import Account
-            from models.model import EndUser
-
-            identity: IdentityDict = {}
-
-            match user:
-                case Account():
-                    if user.current_tenant_id:
-                        identity["tenant_id"] = user.current_tenant_id
-                    identity["user_id"] = user.id
-                    identity["user_type"] = "account"
-                case EndUser():
-                    identity["tenant_id"] = user.tenant_id
-                    identity["user_id"] = user.id
-                    identity["user_type"] = user.type or "end_user"
-
-            return identity
-        except Exception:
-            return {}

api/core/plugin/entities/plugin_daemon.py

@@ -207,6 +207,10 @@ class PluginListResponse(BaseModel):
     total: int
 
 
+class PluginInstalledIdsDaemonResponse(BaseModel):
+    plugin_ids: list[str]
+
+
 class PluginListWithoutTotalResponse(BaseModel):
     list: list[PluginEntity]
     has_more: bool

api/core/plugin/impl/base.py

@@ -1,7 +1,7 @@
 import inspect
 import json
 import logging
-from collections.abc import Callable, Generator
+from collections.abc import Callable, Generator, Mapping
 from typing import Any, cast
 from urllib.parse import unquote
 
@@ -23,6 +23,7 @@ from core.plugin.impl.exc import (
     PluginLLMPollingUnsupportedError,
     PluginNotFoundError,
     PluginPermissionDeniedError,
+    PluginRuntimeError,
     PluginUniqueIdentifierError,
 )
 from core.trigger.errors import (
@@ -375,6 +376,18 @@ class BasePluginClient:
                     # type `PluginLLMPollingUnsupportedError`.
                     case PluginLLMPollingUnsupportedError.__name__:
                         raise PluginLLMPollingUnsupportedError(description=error_object.get("message"))
+                    case PluginRuntimeError.__name__:
+                        args = error_object.get("args")
+                        lambda_request_id = args.get("request_id") if isinstance(args, Mapping) else None
+                        if not isinstance(lambda_request_id, str):
+                            lambda_request_id = None
+                        runtime_message = error_object.get("message")
+                        if not isinstance(runtime_message, str):
+                            runtime_message = "Plugin runtime request failed"
+                        raise PluginRuntimeError(
+                            description=runtime_message,
+                            lambda_request_id=lambda_request_id,
+                        )
                     case _:
                         raise PluginInvokeError(description=message)
             case PluginDaemonInternalServerError.__name__:

api/core/plugin/impl/exc.py

@@ -49,6 +49,18 @@ class PluginDaemonBadRequestError(PluginDaemonClientSideError):
     description: str = "Bad Request"
 
 
+class PluginRuntimeError(PluginDaemonInternalError):
+    """A plugin runtime failed before it could return a valid plugin response."""
+
+    lambda_request_id: str | None
+
+    def __init__(self, description: str, lambda_request_id: str | None = None) -> None:
+        self.lambda_request_id = lambda_request_id
+        if lambda_request_id:
+            description = description.replace(f"RequestId: {lambda_request_id} Error: ", "", 1)
+        super().__init__(description)
+
+
 class PluginInvokeError(PluginDaemonClientSideError, ValueError):
     description: str = "Invoke Error"
 

api/core/plugin/impl/plugin.py

@@ -14,6 +14,7 @@ from core.plugin.entities.plugin import (
 )
 from core.plugin.entities.plugin_daemon import (
     PluginDecodeResponse,
+    PluginInstalledIdsDaemonResponse,
     PluginInstallTask,
     PluginInstallTaskStartResponse,
     PluginListResponse,
@@ -68,6 +69,16 @@ class PluginInstaller(BasePluginClient):
         )
         return result.list
 
+    def list_installed_plugin_ids(self, tenant_id: str, category: PluginCategory) -> list[str]:
+        """List all currently installed plugin IDs in one category."""
+        result = self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/installation/ids",
+            PluginInstalledIdsDaemonResponse,
+            params={"category": category.value},
+        )
+        return result.plugin_ids
+
     def list_plugins_with_total(self, tenant_id: str, page: int, page_size: int) -> PluginListResponse:
         return self._request_with_plugin_daemon_response(
             "GET",
@@ -77,13 +88,28 @@ class PluginInstaller(BasePluginClient):
         )
 
     def list_plugins_by_category(
-        self, tenant_id: str, category: PluginCategory, page: int, page_size: int
+        self,
+        tenant_id: str,
+        category: PluginCategory,
+        page: int,
+        page_size: int,
+        *,
+        query: str = "",
+        tags: Sequence[str] = (),
+        language: str = "en_US",
     ) -> PluginListWithoutTotalResponse:
         return self._request_with_plugin_daemon_response(
             "GET",
             f"plugin/{tenant_id}/management/{category.value}/list",
             PluginListWithoutTotalResponse,
-            params={"page": page, "page_size": page_size, "response_type": "paged"},
+            params={
+                "page": page,
+                "page_size": page_size,
+                "response_type": "paged",
+                "query": query,
+                "tags": list(tags),
+                "language": language,
+            },
         )
 
     def upload_pkg(

api/core/plugin/plugin_service.py

@@ -656,6 +656,12 @@ class PluginService:
         plugins = manager.list_plugins(tenant_id)
         return plugins
 
+    @staticmethod
+    def list_installed_plugin_ids(tenant_id: str, category: PluginCategory) -> Sequence[str]:
+        """List all currently installed plugin IDs in one category through the daemon's lightweight query."""
+        manager = PluginInstaller()
+        return manager.list_installed_plugin_ids(tenant_id, category)
+
     @staticmethod
     def list_with_total(tenant_id: str, user_id: str, page: int, page_size: int) -> PluginListResponse:
         """List tenant plugins with endpoint counts reconciled from live records.
@@ -672,17 +678,32 @@ class PluginService:
 
     @staticmethod
     def list_by_category(
-        tenant_id: str, category: PluginCategory, page: int, page_size: int
+        tenant_id: str,
+        category: PluginCategory,
+        page: int,
+        page_size: int,
+        *,
+        query: str = "",
+        tags: Sequence[str] = (),
+        language: str = "en_US",
     ) -> PluginListWithoutTotalResponse:
         """
         List plugins in one category with a has-more cursor signal and without calculating total.
 
-        The daemon scans tenant installations in the existing list order and stops once it finds one extra match.
-        This keeps pagination usable before category is persisted on installation rows.
+        The daemon applies category, search, and tag filters before pagination, then stops once it finds one extra
+        match. Filtered model reads are partial views and therefore do not reconcile the full model-provider cache.
         """
         manager = PluginInstaller()
-        plugins = manager.list_plugins_by_category(tenant_id, category, page, page_size)
-        if category == PluginCategory.Model:
+        plugins = manager.list_plugins_by_category(
+            tenant_id,
+            category,
+            page,
+            page_size,
+            query=query,
+            tags=tags,
+            language=language,
+        )
+        if category == PluginCategory.Model and not query and not tags:
             should_invalidate_model_provider_cache = (
                 PluginService._should_invalidate_model_provider_cache_for_remote_model_plugins(
                     tenant_id,

api/core/provider_manager.py

@@ -34,6 +34,7 @@ from core.entities.provider_entities import (
 from core.helper import encrypter
 from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
 from core.helper.position_helper import is_filtered
+from enums.deployment_edition import DeploymentEdition
 from extensions import ext_hosting_provider
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -743,7 +744,7 @@ class ProviderManager:
 
             if preferred_provider_type_record:
                 preferred_provider_type = preferred_provider_type_record.preferred_provider_type
-            elif dify_config.EDITION == "CLOUD" and system_configuration.enabled:
+            elif dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and system_configuration.enabled:
                 preferred_provider_type = ProviderType.SYSTEM
             elif custom_configuration.provider or custom_configuration.models:
                 preferred_provider_type = ProviderType.CUSTOM
@@ -1538,7 +1539,7 @@ class ProviderManager:
                 quota_type_to_provider_records_dict[provider_record.quota_type] = provider_record  # type: ignore[index]
         quota_configurations = []
 
-        if dify_config.EDITION == "CLOUD":
+        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
             from services.credit_pool_service import CreditPoolService
 
             trail_pool = CreditPoolService.get_pool(

api/core/rag/data_post_processor/data_post_processor.py

@@ -10,6 +10,7 @@ from core.rag.rerank.entity.weight import KeywordSetting, VectorSetting, Weights
 from core.rag.rerank.rerank_base import BaseRerankRunner
 from core.rag.rerank.rerank_factory import RerankRunnerFactory
 from core.rag.rerank.rerank_type import RerankMode
+from extensions.otel import trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
 
@@ -52,6 +53,7 @@ class DataPostProcessor:
         )
         self.reorder_runner = self._get_reorder_runner(reorder_enabled)
 
+    @trace_span()
     def invoke(
         self,
         query: str,

api/core/rag/datasource/retrieval_service.py

@@ -1,12 +1,10 @@
 import concurrent.futures
-import functools
 import logging
-from collections.abc import Callable, Sequence
+from collections.abc import Sequence
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any, NotRequired, TypedDict
 
 from flask import Flask, current_app
-from opentelemetry import context as otel_context
 from sqlalchemy import select
 from sqlalchemy.orm import Session, load_only
 
@@ -26,7 +24,7 @@ from core.rag.rerank.rerank_type import RerankMode
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.signature import sign_upload_file_preview_url
 from extensions.ext_database import db
-from extensions.otel import trace_span
+from extensions.otel import propagate_context, trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import (
     ChildChunk,
@@ -92,20 +90,6 @@ default_retrieval_model: DefaultRetrievalModelDict = {
 logger = logging.getLogger(__name__)
 
 
-def _propagate_otel_context[**P, R](func: Callable[P, R]) -> Callable[P, R]:
-    captured_context = otel_context.get_current()
-
-    @functools.wraps(func)
-    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
-        token = otel_context.attach(captured_context)
-        try:
-            return func(*args, **kwargs)
-        finally:
-            otel_context.detach(token)
-
-    return wrapper
-
-
 class RetrievalService:
     # Cache precompiled regular expressions to avoid repeated compilation
     @classmethod
@@ -139,7 +123,7 @@ class RetrievalService:
             if query:
                 futures.append(
                     executor.submit(
-                        _propagate_otel_context(retrieval_service._retrieve),
+                        propagate_context(retrieval_service._retrieve),
                         flask_app=current_app._get_current_object(),  # type: ignore
                         retrieval_method=retrieval_method,
                         dataset=dataset,
@@ -159,7 +143,7 @@ class RetrievalService:
                 for attachment_id in attachment_ids:
                     futures.append(
                         executor.submit(
-                            _propagate_otel_context(retrieval_service._retrieve),
+                            propagate_context(retrieval_service._retrieve),
                             flask_app=current_app._get_current_object(),  # type: ignore
                             retrieval_method=retrieval_method,
                             dataset=dataset,
@@ -820,7 +804,7 @@ class RetrievalService:
                 if retrieval_method == RetrievalMethod.KEYWORD_SEARCH and query:
                     futures.append(
                         executor.submit(
-                            _propagate_otel_context(self.keyword_search),
+                            propagate_context(self.keyword_search),
                             flask_app=current_app._get_current_object(),  # type: ignore
                             dataset_id=dataset.id,
                             query=query,
@@ -834,7 +818,7 @@ class RetrievalService:
                     if query:
                         futures.append(
                             executor.submit(
-                                _propagate_otel_context(self.embedding_search),
+                                propagate_context(self.embedding_search),
                                 flask_app=current_app._get_current_object(),  # type: ignore
                                 dataset_id=dataset.id,
                                 query=query,
@@ -851,7 +835,7 @@ class RetrievalService:
                     if attachment_id:
                         futures.append(
                             executor.submit(
-                                _propagate_otel_context(self.embedding_search),
+                                propagate_context(self.embedding_search),
                                 flask_app=current_app._get_current_object(),  # type: ignore
                                 dataset_id=dataset.id,
                                 query=attachment_id,
@@ -868,7 +852,7 @@ class RetrievalService:
                 if RetrievalMethod.is_support_fulltext_search(retrieval_method) and query:
                     futures.append(
                         executor.submit(
-                            _propagate_otel_context(self.full_text_index_search),
+                            propagate_context(self.full_text_index_search),
                             flask_app=current_app._get_current_object(),  # type: ignore
                             dataset_id=dataset.id,
                             query=query,

api/core/rag/docstore/dataset_docstore.py

@@ -6,10 +6,7 @@ from typing import Any
 from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
-from core.model_manager import ModelManager
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.models.document import AttachmentDocument, Document
-from graphon.model_runtime.entities.model_entities import ModelType
 from models.dataset import ChildChunk, Dataset, DocumentSegment, SegmentAttachmentBinding
 from models.enums import SegmentType
 
@@ -69,34 +66,22 @@ class DatasetDocumentStore:
 
     def add_documents(
         self,
-        docs: Sequence[Document],
         session: Session,
+        docs: Sequence[Document],
+        token_counts: list[int],
         allow_update: bool = True,
         save_child: bool = False,
-    ):
+    ) -> None:
+        document_token_pairs = list(zip(docs, token_counts, strict=True))
+
         max_position = session.scalar(
             select(func.max(DocumentSegment.position)).where(DocumentSegment.document_id == self._document_id)
         )
 
         if max_position is None:
             max_position = 0
-        embedding_model = None
-        if self._dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
-            model_manager = ModelManager.for_tenant(tenant_id=self._dataset.tenant_id)
-            embedding_model = model_manager.get_model_instance(
-                tenant_id=self._dataset.tenant_id,
-                provider=self._dataset.embedding_model_provider,
-                model_type=ModelType.TEXT_EMBEDDING,
-                model=self._dataset.embedding_model,
-            )
 
-        if embedding_model:
-            page_content_list = [doc.page_content for doc in docs]
-            tokens_list = embedding_model.get_text_embedding_num_tokens(page_content_list)
-        else:
-            tokens_list = [0] * len(docs)
-
-        for doc, tokens in zip(docs, tokens_list):
+        for doc, tokens in document_token_pairs:
             if not isinstance(doc, Document):
                 raise ValueError("doc must be a Document")
 

api/core/rag/embedding/token_counter.py

@@ -0,0 +1,25 @@
+"""Token counting for document segments."""
+
+from core.model_manager import ModelManager
+from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from core.rag.models.document import Document
+from graphon.model_runtime.entities.model_entities import ModelType
+from models.dataset import Dataset
+
+
+def calculate_segment_token_counts(dataset: Dataset, documents: list[Document]) -> list[int]:
+    """Return one token count per document, invoking the embedding model only for high-quality indexes."""
+    if not documents:
+        return []
+
+    if dataset.indexing_technique != IndexTechniqueType.HIGH_QUALITY:
+        return [0] * len(documents)
+
+    model_manager = ModelManager.for_tenant(tenant_id=dataset.tenant_id)
+    embedding_model = model_manager.get_model_instance(
+        tenant_id=dataset.tenant_id,
+        provider=dataset.embedding_model_provider,
+        model_type=ModelType.TEXT_EMBEDDING,
+        model=dataset.embedding_model,
+    )
+    return embedding_model.get_text_embedding_num_tokens([document.page_content for document in documents])

api/core/rag/extractor/watercrawl/client.py

@@ -131,7 +131,10 @@ class WaterCrawlAPIClient(BaseAPIClient):
         content_type = response.headers.get("Content-Type", "")
         media_type = content_type.split(";", 1)[0].strip().lower()
         if media_type == "application/json":
-            return response.json() or {}
+            try:
+                return response.json() or {}
+            except ValueError as exc:
+                raise ValueError("Invalid JSON response from WaterCrawl") from exc
 
         if media_type == "application/octet-stream":
             return response.content

api/core/rag/extractor/watercrawl/exceptions.py

@@ -1,5 +1,15 @@
+"""WaterCrawl domain exceptions.
+
+These exceptions are constructed from upstream HTTP responses, which may be
+JSON API errors or plain text/HTML proxy errors. Keep the exception type stable
+even when the body is not JSON so callers can handle WaterCrawl failures by
+domain type instead of low-level parser errors.
+"""
+
 import json
-from typing import override
+from typing import Any, override
+
+from httpx import Response
 
 
 class WaterCrawlError(Exception):
@@ -7,11 +17,16 @@ class WaterCrawlError(Exception):
 
 
 class WaterCrawlBadRequestError(WaterCrawlError):
-    def __init__(self, response):
+    def __init__(self, response: Response):
         self.status_code = response.status_code
         self.response = response
-        data = response.json()
-        self.message = data.get("message", "Unknown error occurred")
+        try:
+            data: Any = response.json()
+        except ValueError:
+            data = {}
+        if not isinstance(data, dict):
+            data = {}
+        self.message = data.get("message") or response.text or "Unknown error occurred"
         self.errors = data.get("errors", {})
         super().__init__(self.message)
 

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -19,6 +19,7 @@ from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -244,10 +245,16 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                         all_multimodal_documents.extend(doc.attachments)
                 documents.append(doc)
         if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
             # save node to document segment
             doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
             # add document segments
-            doc_store.add_documents(docs=documents, save_child=False, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=False,
+            )
             session.commit()
             if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                 vector = Vector(dataset, session=session)

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -15,6 +15,7 @@ from core.model_manager import ModelInstance
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import ParentMode, Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -304,6 +305,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                 doc.attachments = self._get_content_files(doc, current_user=account, session=session)
             documents.append(doc)
         if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
             # update document parent mode
             dataset_process_rule = DatasetProcessRule(
                 dataset_id=dataset.id,
@@ -321,7 +323,12 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             # save node to document segment
             doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
             # add document segments
-            doc_store.add_documents(docs=documents, save_child=True, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=True,
+            )
             session.commit()
             if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                 all_child_documents = []

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -17,6 +17,7 @@ from core.llm_generator.llm_generator import LLMGenerator
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.embedding.token_counter import calculate_segment_token_counts
 from core.rag.entities import Rule
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.extract_processor import ExtractProcessor
@@ -205,9 +206,15 @@ class QAIndexProcessor(BaseIndexProcessor):
             doc = Document(page_content=qa_chunk.question, metadata=metadata)
             documents.append(doc)
         if documents:
+            token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
             # save node to document segment
             doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
-            doc_store.add_documents(docs=documents, save_child=False, session=session)
+            doc_store.add_documents(
+                session=session,
+                docs=documents,
+                token_counts=token_counts,
+                save_child=False,
+            )
             session.commit()
             if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
                 vector = Vector(dataset, session=session)

api/core/rag/rerank/rerank_model.py

@@ -9,6 +9,7 @@ from core.rag.index_processor.constant.query_type import QueryType
 from core.rag.models.document import Document
 from core.rag.rerank.rerank_base import BaseRerankRunner
 from extensions.ext_storage import storage
+from extensions.otel import trace_span
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
 from models.model import UploadFile
@@ -22,6 +23,7 @@ class RerankModelRunner(BaseRerankRunner):
         self._session = session
 
     @override
+    @trace_span()
     def run(
         self,
         query: str,

api/core/rag/retrieval/dataset_retrieval.py

@@ -65,6 +65,7 @@ from core.workflow.nodes.knowledge_retrieval.retrieval import (
 )
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from extensions.otel import propagate_context, trace_span
 from graphon.file import File, FileTransferMethod, FileType
 from graphon.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
@@ -116,6 +117,7 @@ class DatasetRetrieval:
         else:
             self._llm_usage = self._llm_usage.plus(usage)
 
+    @trace_span()
     def knowledge_retrieval(self, session: Session, request: KnowledgeRetrievalRequest) -> list[Source]:
         self._check_knowledge_rate_limit(request.tenant_id)
         available_datasets = self._get_available_datasets(request.tenant_id, request.dataset_ids)
@@ -599,6 +601,7 @@ class DatasetRetrieval:
             return "\n".join([document_context.content for document_context in document_context_list]), context_files
         return "", context_files
 
+    @trace_span()
     def single_retrieve(
         self,
         session: Session,
@@ -724,7 +727,7 @@ class DatasetRetrieval:
 
                 if results:
                     thread = threading.Thread(
-                        target=self._on_retrieval_end,
+                        target=propagate_context(self._on_retrieval_end),
                         kwargs={
                             "flask_app": current_app._get_current_object(),  # type: ignore
                             "documents": results,
@@ -737,6 +740,7 @@ class DatasetRetrieval:
                 return results
         return []
 
+    @trace_span()
     def multiple_retrieve(
         self,
         app_id: str,
@@ -798,7 +802,7 @@ class DatasetRetrieval:
 
             if query:
                 query_thread = threading.Thread(
-                    target=self._multiple_retrieve_thread,
+                    target=propagate_context(self._multiple_retrieve_thread_safely),
                     kwargs={
                         "flask_app": current_app._get_current_object(),  # type: ignore
                         "available_datasets": available_datasets,
@@ -824,7 +828,7 @@ class DatasetRetrieval:
             if attachment_ids:
                 for attachment_id in attachment_ids:
                     attachment_thread = threading.Thread(
-                        target=self._multiple_retrieve_thread,
+                        target=propagate_context(self._multiple_retrieve_thread_safely),
                         kwargs={
                             "flask_app": current_app._get_current_object(),  # type: ignore
                             "available_datasets": available_datasets,
@@ -865,7 +869,7 @@ class DatasetRetrieval:
         if all_documents:
             # add thread to call _on_retrieval_end
             retrieval_end_thread = threading.Thread(
-                target=self._on_retrieval_end,
+                target=propagate_context(self._on_retrieval_end),
                 kwargs={
                     "flask_app": current_app._get_current_object(),  # type: ignore
                     "documents": all_documents,
@@ -1161,6 +1165,7 @@ class DatasetRetrieval:
 
                         all_documents.extend(documents)
 
+    @trace_span()
     def _run_retriever_thread(
         self,
         *,
@@ -1172,27 +1177,51 @@ class DatasetRetrieval:
         document_ids_filter: list[str] | None,
         metadata_condition: MetadataFilteringCondition | None,
         attachment_ids: list[str] | None,
+    ) -> None:
+        with session_factory.create_session() as session:
+            self._retriever(
+                flask_app=flask_app,
+                session=session,
+                dataset_id=dataset_id,
+                query=query or "",
+                top_k=top_k,
+                all_documents=all_documents,
+                document_ids_filter=document_ids_filter,
+                metadata_condition=metadata_condition,
+                attachment_ids=attachment_ids,
+            )
+
+    def _run_retriever_thread_safely(
+        self,
+        *,
+        flask_app: Flask,
+        dataset_id: str,
+        query: str | None,
+        top_k: int,
+        all_documents: list[Document],
+        document_ids_filter: list[str] | None,
+        metadata_condition: MetadataFilteringCondition | None,
+        attachment_ids: list[str] | None,
         cancel_event: threading.Event | None,
         thread_exceptions: list[Exception] | None,
     ) -> None:
+        """Collect errors only after they pass through the traced retrieval method."""
         try:
-            with session_factory.create_session() as session:
-                self._retriever(
-                    flask_app=flask_app,
-                    session=session,
-                    dataset_id=dataset_id,
-                    query=query or "",
-                    top_k=top_k,
-                    all_documents=all_documents,
-                    document_ids_filter=document_ids_filter,
-                    metadata_condition=metadata_condition,
-                    attachment_ids=attachment_ids,
-                )
-        except Exception as e:
+            self._run_retriever_thread(
+                flask_app=flask_app,
+                dataset_id=dataset_id,
+                query=query,
+                top_k=top_k,
+                all_documents=all_documents,
+                document_ids_filter=document_ids_filter,
+                metadata_condition=metadata_condition,
+                attachment_ids=attachment_ids,
+            )
+        except Exception as exc:
             if cancel_event:
                 cancel_event.set()
             if thread_exceptions is not None:
-                thread_exceptions.append(e)
+                thread_exceptions.append(exc)
 
     def to_dataset_retriever_tool(
         self,
@@ -1795,6 +1824,7 @@ class DatasetRetrieval:
 
         return full_text, usage
 
+    @trace_span()
     def _multiple_retrieve_thread(
         self,
         flask_app: Flask,
@@ -1813,11 +1843,11 @@ class DatasetRetrieval:
         attachment_id: str | None,
         dataset_count: int,
         cancel_event: threading.Event | None = None,
-        thread_exceptions: list[Exception] | None = None,
-    ):
+    ) -> None:
         try:
             with flask_app.app_context():
                 threads = []
+                retrieval_thread_exceptions: list[Exception] = []
                 all_documents_item: list[Document] = []
                 index_type = None
                 for dataset in available_datasets:
@@ -1836,7 +1866,7 @@ class DatasetRetrieval:
                             else:
                                 continue
                     retrieval_thread = threading.Thread(
-                        target=self._run_retriever_thread,
+                        target=propagate_context(self._run_retriever_thread_safely),
                         kwargs={
                             "flask_app": flask_app,
                             "dataset_id": dataset.id,
@@ -1847,7 +1877,7 @@ class DatasetRetrieval:
                             "metadata_condition": metadata_condition,
                             "attachment_ids": [attachment_id] if attachment_id else None,
                             "cancel_event": cancel_event,
-                            "thread_exceptions": thread_exceptions,
+                            "thread_exceptions": retrieval_thread_exceptions,
                         },
                     )
                     threads.append(retrieval_thread)
@@ -1862,6 +1892,9 @@ class DatasetRetrieval:
                     if cancel_event and cancel_event.is_set():
                         break
 
+                if retrieval_thread_exceptions:
+                    raise retrieval_thread_exceptions[0]
+
                 # Skip second reranking when there is only one dataset
                 if reranking_enable and dataset_count > 1:
                     # do rerank for searched documents
@@ -1902,11 +1935,55 @@ class DatasetRetrieval:
                         all_documents_item = all_documents_item[:top_k] if top_k else all_documents_item
                 if all_documents_item:
                     all_documents.extend(all_documents_item)
-        except Exception as e:
+        except Exception:
+            raise
+
+    def _multiple_retrieve_thread_safely(
+        self,
+        *,
+        flask_app: Flask,
+        available_datasets: list[Dataset],
+        metadata_condition: MetadataFilteringCondition | None,
+        metadata_filter_document_ids: dict[str, list[str]] | None,
+        all_documents: list[Document],
+        tenant_id: str,
+        reranking_enable: bool,
+        reranking_mode: str,
+        reranking_model: RerankingModelDict | None,
+        weights: WeightsDict | None,
+        top_k: int,
+        score_threshold: float,
+        query: str | None,
+        attachment_id: str | None,
+        dataset_count: int,
+        cancel_event: threading.Event | None = None,
+        thread_exceptions: list[Exception] | None = None,
+    ) -> None:
+        """Collect errors only after they pass through the traced multi-retrieval method."""
+        try:
+            self._multiple_retrieve_thread(
+                flask_app=flask_app,
+                available_datasets=available_datasets,
+                metadata_condition=metadata_condition,
+                metadata_filter_document_ids=metadata_filter_document_ids,
+                all_documents=all_documents,
+                tenant_id=tenant_id,
+                reranking_enable=reranking_enable,
+                reranking_mode=reranking_mode,
+                reranking_model=reranking_model,
+                weights=weights,
+                top_k=top_k,
+                score_threshold=score_threshold,
+                query=query,
+                attachment_id=attachment_id,
+                dataset_count=dataset_count,
+                cancel_event=cancel_event,
+            )
+        except Exception as exc:
             if cancel_event:
                 cancel_event.set()
             if thread_exceptions is not None:
-                thread_exceptions.append(e)
+                thread_exceptions.append(exc)
 
     def _get_available_datasets(self, tenant_id: str, dataset_ids: list[str]) -> list[Dataset]:
         with session_factory.create_session() as session:

api/core/rbac/entities.py

@@ -61,6 +61,7 @@ class RBACPermission(StrEnum):
     WORKSPACE_ROLE_MANAGE = "workspace_role_manage"
     API_EXTENSION_MANAGE = "api_extension_manage"
     CUSTOMIZATION_MANAGE = "customization_manage"
+    AGENT_MANAGE = "agent_manage"
 
     SNIPPETS_CREATE_AND_MODIFY = "snippets_create_and_modify"
     SNIPPETS_MANAGE = "snippets_management"