perf(web): load integration tools by category

fix(web): defer model status plugin lookup
fix(web): optimize data source plugin loading
2026-07-24 17:40:26 -07:00 · 2026-07-24 16:39:05 -07:00 · 2026-07-24 16:37:53 -07:00 · 2026-07-24 15:08:27 -07:00 · 2026-07-24 15:30:52 +00:00 · 2026-07-24 15:27:35 +00:00
1438 changed files with 90401 additions and 20327 deletions
@@ -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.

@@ -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
@@ -16,7 +16,7 @@ concurrency:
 jobs:
  api-unit:
    name: API Unit Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    env:
      COVERAGE_FILE: coverage-unit
    defaults:
@@ -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 \
@@ -75,7 +72,7 @@ jobs:

  api-integration:
    name: API Integration Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    env:
      COVERAGE_FILE: coverage-integration
      STORAGE_TYPE: opendal
@@ -129,7 +126,7 @@ jobs:

  api-coverage:
    name: API Coverage
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    needs:
      - api-unit
      - api-integration
@@ -173,7 +173,7 @@ jobs:

  create-manifest:
    needs: build
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: github.repository == 'langgenius/dify'
    strategy:
      matrix:
@@ -23,7 +23,7 @@ concurrency:
 jobs:
  validate:
    name: validate manifest + resolve target Dify release
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: github.repository == 'langgenius/dify'
    permissions:
      contents: read
@@ -87,7 +87,7 @@ jobs:
  release:
    name: build + attach standalone binaries (all targets)
    needs: validate
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: write
    defaults:
@@ -9,7 +9,7 @@ concurrency:

 jobs:
  db-migration-test-postgres:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -59,7 +59,7 @@ jobs:
        run: uv run --directory api flask upgrade-db

  db-migration-test-mysql:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/agent'
@@ -10,7 +10,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/dev'
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/enterprise'
@@ -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 }}
@@ -13,7 +13,7 @@ on:

 jobs:
  deploy:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    if: |
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/saas'
@@ -22,7 +22,7 @@ concurrency:
 jobs:
  check-cherry-pick-provenance:
    name: Require cherry-pick provenance
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
        with:
@@ -7,7 +7,7 @@ jobs:
    permissions:
      contents: read
      pull-requests: write
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
        with:
@@ -21,7 +21,7 @@ concurrency:
 jobs:
  pre_job:
    name: Skip Duplicate Checks
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      should_skip: ${{ steps.skip_check.outputs.should_skip || 'false' }}
    steps:
@@ -37,7 +37,7 @@ jobs:
    name: Check Changed Files
    needs: pre_job
    if: needs.pre_job.outputs.should_skip != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      api-changed: ${{ steps.changes.outputs.api }}
      cli-changed: ${{ steps.changes.outputs.cli }}
@@ -81,7 +81,6 @@ jobs:
              - '.npmrc'
              - '.nvmrc'
              - '.github/workflows/cli-tests.yml'
-              - '.github/workflows/cli-docker-build.yml'
              - '.github/actions/setup-web/**'
            web:
              - 'web/**'
@@ -164,7 +163,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.api-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped API tests
        run: echo "No API-related changes detected; skipping API tests."
@@ -177,7 +176,7 @@ jobs:
      - check-changes
      - api-tests-run
      - api-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize API Tests status
        env:
@@ -224,7 +223,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.cli-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped CLI tests
        run: echo "No CLI-related changes detected; skipping CLI tests."
@@ -237,7 +236,7 @@ jobs:
      - check-changes
      - cli-tests-run
      - cli-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize CLI Tests status
        env:
@@ -284,7 +283,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.web-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped web tests
        run: echo "No web-related changes detected; skipping web tests."
@@ -297,7 +296,7 @@ jobs:
      - check-changes
      - web-tests-run
      - web-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Web Tests status
        env:
@@ -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:
@@ -344,7 +345,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped web full-stack e2e
        run: echo "No E2E-related changes detected; skipping web full-stack E2E."
@@ -357,7 +358,7 @@ jobs:
      - check-changes
      - web-e2e-run
      - web-e2e-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Web Full-Stack E2E status
        env:
@@ -411,7 +412,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.vdb-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped VDB tests
        run: echo "No VDB-related changes detected; skipping VDB tests."
@@ -424,7 +425,7 @@ jobs:
      - check-changes
      - vdb-tests-run
      - vdb-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize VDB Tests status
        env:
@@ -470,7 +471,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.migration-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped DB migration tests
        run: echo "No migration-related changes detected; skipping DB migration tests."
@@ -483,7 +484,7 @@ jobs:
      - check-changes
      - db-migration-test-run
      - db-migration-test-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize DB Migration Test status
        env:
@@ -530,7 +531,7 @@ jobs:
      - pre_job
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.sandbox-runtime-changed != 'true'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Report skipped sandbox runtime tests
        run: echo "No sandbox-runtime-related changes detected; skipping sandbox runtime tests."
@@ -543,7 +544,7 @@ jobs:
      - check-changes
      - sandbox-runtime-tests-run
      - sandbox-runtime-tests-skip
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Finalize Sandbox Runtime Tests status
        env:
@@ -14,7 +14,7 @@ concurrency:
 jobs:
  check-changes:
    name: Check Changed Files
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    outputs:
      external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
    steps:
@@ -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/**'
@@ -12,7 +12,7 @@ permissions: {}
 jobs:
  comment:
    name: Comment PR with pyrefly diff
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      actions: read
      contents: read
@@ -10,7 +10,7 @@ permissions:

 jobs:
  pyrefly-diff:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: read
      issues: write
@@ -12,7 +12,7 @@ permissions: {}
 jobs:
  comment:
    name: Comment PR with type coverage
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      actions: read
      contents: read
@@ -10,7 +10,7 @@ permissions:

 jobs:
  pyrefly-type-coverage:
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      contents: read
      issues: write
@@ -13,7 +13,7 @@ concurrency:
 jobs:
  sandbox-runtime-unit:
    name: Sandbox Runtime Unit Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -37,7 +37,7 @@ jobs:

  sandbox-runtime-lint:
    name: Sandbox Runtime Lint
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -64,7 +64,7 @@ jobs:

  sandbox-runtime-integration:
    name: Sandbox Runtime Integration Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        shell: bash
@@ -16,7 +16,7 @@ jobs:
    name: Validate PR title
    permissions:
      pull-requests: read
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    steps:
      - name: Complete merge group check
        if: github.event_name == 'merge_group'
@@ -12,7 +12,7 @@ on:
 jobs:
  stale:

-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    permissions:
      issues: write
      pull-requests: write
@@ -19,7 +19,7 @@ permissions:
 jobs:
  python-style:
    name: Python Style
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -83,7 +83,7 @@ jobs:

  web-style:
    name: Web Style
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    defaults:
      run:
        working-directory: ./web
@@ -182,7 +182,7 @@ jobs:

  superlinter:
    name: SuperLinter
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    steps:
      - name: Checkout code
@@ -17,7 +17,7 @@ concurrency:
 jobs:
  build:
    name: unit test for Node.js SDK
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04

    defaults:
      run:
@@ -35,7 +35,7 @@ concurrency:
 jobs:
  translate:
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    timeout-minutes: 120

    steps:
@@ -16,7 +16,7 @@ concurrency:
 jobs:
  trigger:
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    timeout-minutes: 5

    steps:
@@ -16,7 +16,7 @@ jobs:
  test:
    name: Full VDB Tests
    if: github.repository == 'langgenius/dify'
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    strategy:
      matrix:
        python-version:
@@ -13,7 +13,7 @@ concurrency:
 jobs:
  test:
    name: VDB Smoke Tests
-    runs-on: depot-ubuntu-24.04-4
+    runs-on: depot-ubuntu-24.04
    strategy:
      matrix:
        python-version:
@@ -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
@@ -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:
@@ -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 \
@@ -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
@@ -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:
@@ -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
@@ -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,
@@ -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,
@@ -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")
@@ -40,6 +40,7 @@ from . import (
    init_validate,
    knowledge_fs_proxy,
    notification,
+    onboarding,
    ping,
    setup,
    spec,
@@ -209,6 +210,7 @@ __all__ = [
    "notification",
    "oauth",
    "oauth_server",
+    "onboarding",
    "ops_trace",
    "parameter",
    "ping",
@@ -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
@@ -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
@@ -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,
@@ -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):
@@ -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:
@@ -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,
@@ -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
@@ -359,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
@@ -441,6 +447,7 @@ register_response_schema_models(
    WorkflowOnlineUsersByApp,
    WorkflowOnlineUsersResponse,
    WorkflowPublishResponse,
+    SyncDraftWorkflowResponse,
    WorkflowRestoreResponse,
    DefaultBlockConfigsResponse,
    DefaultBlockConfigResponse,
@@ -556,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")
@@ -618,11 +618,14 @@ class DraftWorkflowApi(Resource):
        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")
@@ -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:
@@ -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()

@@ -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."
@@ -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:
@@ -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:
@@ -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,
@@ -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:
@@ -645,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
@@ -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()
@@ -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
@@ -8,8 +8,9 @@ can be validated explicitly against the pinned KnowledgeFS contract during devel
 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. Upstream 401 responses become 502 so they cannot
-trigger Dify browser-session recovery; resource-level 403 responses remain 403.
+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
@@ -27,9 +28,11 @@ from werkzeug.exceptions import (
    BadGateway,
    Forbidden,
    GatewayTimeout,
+    HTTPException,
    NotFound,
    RequestEntityTooLarge,
    ServiceUnavailable,
+    default_exceptions,
 )

 from configs import dify_config
@@ -41,21 +44,28 @@ from controllers.console.wraps import (
 )
 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,
-    KnowledgeFSMethod,
    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",
@@ -128,27 +138,25 @@ def _translate_proxy_error(exc: Exception, *, tenant_id: str) -> NoReturn:


 def _knowledge_fs_operation_access_required(
-    view: Callable[[KnowledgeFSMethod, str], ResponseReturnValue],
+    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:
-        try:
-            operation = get_knowledge_fs_operation(method, upstream_path)
-        except KnowledgeFSRouteNotAllowedError as exc:
-            raise NotFound() from exc
-
        current_user, tenant_id = current_account_with_tenant()
        try:
-            authorize_knowledge_fs_request(
+            authorization = authorize_knowledge_fs_request(
                account=current_user,
                tenant_id=tenant_id,
-                operation=operation,
+                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(method, upstream_path)
+        return view(authorization)

    return decorated

@@ -190,21 +198,26 @@ def _proxy_response(
    """Expose raw content, status, and allowlisted headers from KnowledgeFS.

    Raises:
-        BadGateway: KnowledgeFS rejects the configured server credential.
-        Forbidden: KnowledgeFS denies the account access to the requested resource.
+        HTTPException: KnowledgeFS returns a status normalized by the operation contract.
    """
    upstream = upstream_result.response
-    if upstream.status_code == HTTPStatus.UNAUTHORIZED:
+    mapped_status = dict(upstream_result.operation.error_status_map).get(upstream.status_code)
+    if mapped_status is not None:
        upstream.close()
-        logger.error(
-            "KnowledgeFS rejected the Dify server credential with HTTP %s for tenant_id=%s",
-            upstream.status_code,
-            tenant_id,
-        )
-        raise BadGateway("KnowledgeFS authentication failed")
-    if upstream.status_code == HTTPStatus.FORBIDDEN:
-        upstream.close()
-        raise Forbidden()
+        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)
@@ -237,26 +250,21 @@ def _proxy_response(
    return Response(content, status=upstream.status_code, headers=headers)


-def _proxy_request(method: KnowledgeFSMethod, upstream_path: str) -> Response:
-    """Forward the current raw request and return its filtered upstream response.
-
-    The call performs one outbound KnowledgeFS request. Integration failures are
-    converted to Console HTTP exceptions for the outer JSON error adapter.
-    """
+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()
-    current_user, tenant_id = current_account_with_tenant()
    try:
-        proxy_result = proxy_knowledge_fs_request(
-            account=current_user,
-            method=method,
-            path=upstream_path,
-            tenant_id=tenant_id,
-            accept=request.headers.get("Accept"),
-            content_type=request.content_type,
-            query=request.query_string or None,
-            body=_request_body() if method != "GET" else None,
-            request_headers=request.headers,
+        proxy_result = forward(
+            request.headers.get("Accept"),
+            request.content_type,
+            request.query_string or None,
+            _request_body() if method != "GET" else None,
        )
    except (
        KnowledgeFSConfigurationError,
@@ -274,20 +282,99 @@ def _proxy_request(method: KnowledgeFSMethod, upstream_path: str) -> Response:
    )


+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(
-    method: KnowledgeFSMethod,
-    upstream_path: str,
+    authorization: KnowledgeFSAuthorization,
 ) -> ResponseReturnValue:
    """Apply knowledge billing checks to one allowlisted non-GET operation."""
-    return _proxy_request(method, upstream_path)
+    return _proxy_authorized_request(authorization)


@bp.route(
    "/knowledge-fs/<path:upstream_path>",
-    methods=["GET", "OPTIONS"],
+    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
@@ -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,
+            ),
+        )
@@ -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
@@ -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:
@@ -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")

@@ -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
@@ -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(
@@ -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:
@@ -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()
@@ -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
@@ -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(
@@ -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

@@ -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(
@@ -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")
@@ -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
@@ -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)
+        )
+    )
@@ -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)
@@ -682,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(
@@ -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,
@@ -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(
@@ -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(
@@ -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()
@@ -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):
@@ -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)
@@ -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 {}
@@ -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
@@ -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__:
@@ -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"

@@ -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(
@@ -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,
@@ -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(
@@ -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,
@@ -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,
@@ -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")

@@ -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])
@@ -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
@@ -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)

@@ -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)
@@ -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 = []
@@ -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)
@@ -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,
@@ -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:
@@ -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"
@@ -58,7 +58,6 @@ class Tool(ABC):
        if self.runtime and self.runtime.runtime_parameters:
            tool_parameters.update(self.runtime.runtime_parameters)

-        # try parse tool parameters into the correct type
        tool_parameters = self._transform_tool_parameters_type(tool_parameters)

        result = self._invoke(
@@ -87,14 +86,14 @@ class Tool(ABC):
                return result

    def _transform_tool_parameters_type(self, tool_parameters: dict[str, Any]) -> dict[str, Any]:
-        """
-        Transform tool parameters type
-        """
-        # Temp fix for the issue that the tool parameters will be converted to empty while validating the credentials
+        """Transform declared tool parameter values without resolving runtime schemas."""
        result = deepcopy(tool_parameters)
        for parameter in self.entity.parameters or []:
            if parameter.name in tool_parameters:
-                result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])
+                if parameter.multiple:
+                    result[parameter.name] = parameter.init_frontend_parameter(result.get(parameter.name))
+                else:
+                    result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])

        return result

@@ -196,17 +195,31 @@ class Tool(ABC):
            }:
                continue

-            parameter_schema: dict[str, Any] = (
-                {
-                    "type": parameter.type.as_normal_type(),
-                    "description": parameter.llm_description or "",
-                }
-                if parameter.input_schema is None
-                else deepcopy(parameter.input_schema)
-            )
+            is_multiple_select = parameter.multiple and parameter.type in {
+                ToolParameter.ToolParameterType.SELECT,
+                ToolParameter.ToolParameterType.DYNAMIC_SELECT,
+            }
+            if is_multiple_select:
+                item_schema: dict[str, Any] = {"type": "string"}
+                if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+                    item_schema["enum"] = [option.value for option in parameter.options]
+                parameter_schema: dict[str, Any] = {"type": "array", "items": item_schema}
+            else:
+                parameter_schema = (
+                    {
+                        "type": parameter.type.as_normal_type(),
+                        "description": parameter.llm_description or "",
+                    }
+                    if parameter.input_schema is None
+                    else deepcopy(parameter.input_schema)
+                )
            parameter_schema.setdefault("description", parameter.llm_description or "")

-            if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+            if (
+                not is_multiple_select
+                and parameter.type == ToolParameter.ToolParameterType.SELECT
+                and parameter.options
+            ):
                parameter_schema["enum"] = [option.value for option in parameter.options]

            schema["properties"][parameter.name] = parameter_schema
--- a/Show More
+++ b/Show More