feat(tools): support tool-generated A2UI parts

Stream, validate, persist, and render constrained UI parts from tools, including the current-time component. Preserve markdown following think blocks as normal answer content.
feat(web): locate workflow nodes by node_id in the workflow editor (#38187 )
2026-07-24 10:46:12 +08:00 · 2026-07-24 02:16:41 +00:00 · 2026-07-24 02:10:39 +00:00 · 2026-07-24 02:09:48 +00:00 · 2026-07-24 01:59:57 +00:00 · 2026-07-23 13:54:12 +00:00
1075 changed files with 64366 additions and 16223 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
@@ -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'
@@ -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
@@ -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):
    """
@@ -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."
@@ -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
@@ -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,
+            ),
+        )
@@ -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:
@@ -67,6 +67,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
@@ -645,6 +654,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
@@ -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

@@ -9,11 +9,13 @@ from controllers.common.schema import register_response_schema_models
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
 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 +34,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 +86,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 +101,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 +123,15 @@ 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.EDITION == "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 +168,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
@@ -10,12 +10,13 @@ from sqlalchemy.orm import Session
 from core.agent.entities import AgentEntity, AgentToolEntity
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfig
-from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.base_app_runner import AppRunner
 from core.app.entities.app_invoke_entities import (
    AgentChatAppGenerateEntity,
    ModelConfigWithCredentialsEntity,
 )
+from core.app.entities.queue_entities import QueueUIPartEvent
 from core.app.file_access import DatabaseFileAccessController
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
@@ -23,6 +24,12 @@ from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.tools.__base.tool import Tool
+from core.tools.entities.ui_entities import (
+    MessageUIPart,
+    ToolUIMessage,
+    build_ui_part_id,
+    validate_tool_ui_message_batch,
+)
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
 from extensions.ext_database import db
@@ -352,6 +359,34 @@ class BaseAgentRunner(AppRunner):
        db.session.commit()
        db.session.close()

+    def publish_ui_messages(self, *, namespace: str, ui_messages: list[ToolUIMessage]) -> None:
+        """Publish validated tool UI messages without adding them to agent observations."""
+        try:
+            validate_tool_ui_message_batch(ui_messages)
+        except ValueError:
+            logger.warning(
+                "Ignored tool UI batch that exceeds queue publication limits",
+                extra={"message_id": self.message.id, "namespace": namespace},
+                exc_info=True,
+            )
+            return
+
+        sequences: dict[str, int] = {}
+        for ui_message in ui_messages:
+            part_id = build_ui_part_id(namespace, ui_message.surface_id)
+            sequence = sequences.get(part_id, 0) + 1
+            sequences[part_id] = sequence
+            self.queue_manager.publish(
+                QueueUIPartEvent(
+                    part=MessageUIPart.from_tool_ui_message(
+                        part_id=part_id,
+                        sequence=sequence,
+                        ui_message=ui_message,
+                    )
+                ),
+                PublishFrom.APPLICATION_MANAGER,
+            )
+
    def organize_agent_history(self, prompt_messages: list[PromptMessage], *, session: Session) -> list[PromptMessage]:
        """
        Organize agent history
@@ -235,6 +235,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                    # action is tool call, invoke tool
                    tool_invoke_response, tool_invoke_meta = self._handle_invoke_action(
                        session=session,
+                        agent_thought_id=agent_thought_id,
                        action=scratchpad.action,
                        tool_instances=tool_instances,
                        message_file_ids=message_file_ids,
@@ -306,9 +307,11 @@ class CotAgentRunner(BaseAgentRunner, ABC):
        tool_instances: Mapping[str, Tool],
        message_file_ids: list[str],
        trace_manager: TraceQueueManager | None = None,
+        agent_thought_id: str | None = None,
    ) -> tuple[str, ToolInvokeMeta]:
        """
        handle invoke action
+        :param agent_thought_id: namespace for UI parts emitted by this action
        :param action: action
        :param tool_instances: tool instances
        :param message_file_ids: message file ids
@@ -331,7 +334,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                pass

        # invoke tool
-        tool_invoke_response, message_files, tool_invoke_meta = ToolEngine.agent_invoke(
+        invoke_result = ToolEngine.agent_invoke(
            session=session,
            tool=tool_instance,
            tool_parameters=tool_call_args,
@@ -342,6 +345,13 @@ class CotAgentRunner(BaseAgentRunner, ABC):
            agent_tool_callback=self.agent_callback,
            trace_manager=trace_manager,
        )
+        tool_invoke_response = invoke_result.observation
+        message_files = invoke_result.message_files
+        tool_invoke_meta = invoke_result.meta
+        self.publish_ui_messages(
+            namespace=agent_thought_id or self.message.id,
+            ui_messages=invoke_result.ui_messages,
+        )
        session.commit()
        session.close()

@@ -251,7 +251,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                    }
                else:
                    # invoke tool
-                    tool_invoke_response, message_files, tool_invoke_meta = ToolEngine.agent_invoke(
+                    invoke_result = ToolEngine.agent_invoke(
                        session=session,
                        tool=tool_instance,
                        tool_parameters=tool_call_args,
@@ -265,6 +265,13 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                        message_id=self.message.id,
                        conversation_id=self.conversation.id,
                    )
+                    tool_invoke_response = invoke_result.observation
+                    message_files = invoke_result.message_files
+                    tool_invoke_meta = invoke_result.meta
+                    self.publish_ui_messages(
+                        namespace=tool_call_id,
+                        ui_messages=invoke_result.ui_messages,
+                    )
                    session.commit()
                    session.close()
                    # publish files
@@ -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(
@@ -52,8 +52,16 @@ from core.app.entities.queue_entities import (
    QueueAgentThoughtEvent,
    QueueLLMChunkEvent,
    QueueMessageEndEvent,
+    QueueUIPartEvent,
 )
 from core.repositories.human_input_repository import HumanInputFormRepository, HumanInputFormRepositoryImpl
+from core.tools.entities.ui_entities import (
+    MessageUIPart,
+    ToolUIMessage,
+    build_ui_part_id,
+    parse_tool_ui_messages,
+    validate_tool_ui_message_batch,
+)
 from core.workflow.nodes.agent_v2.ask_human_hitl import AskHumanFormBuildError, create_ask_human_form
 from core.workflow.nodes.agent_v2.ask_human_resume import build_deferred_tool_results, resolve_ask_human_form
 from extensions.ext_database import db
@@ -435,9 +443,29 @@ class _AgentProcessRecorder:
        content = part.get("content")
        if content is None:
            content = part
-        self._record_tool_observation(tool_call_id=tool_call_id, tool_name=tool_name, observation=content)
+        thought_id = self._record_tool_observation(
+            tool_call_id=tool_call_id,
+            tool_name=tool_name,
+            observation=content,
+        )

-    def _record_tool_observation(self, *, tool_call_id: str | None, tool_name: str | None, observation: Any) -> None:
+        metadata = part.get("metadata")
+        if not isinstance(metadata, Mapping) or "dify_ui_messages" not in metadata:
+            return
+        try:
+            ui_messages = parse_tool_ui_messages(metadata["dify_ui_messages"])
+        except (TypeError, ValueError):
+            logger.warning("Ignored invalid dify_ui_messages metadata from Agent backend", exc_info=True)
+            return
+        self._publish_ui_messages(namespace=tool_call_id or thought_id, ui_messages=ui_messages)
+
+    def _record_tool_observation(
+        self,
+        *,
+        tool_call_id: str | None,
+        tool_name: str | None,
+        observation: Any,
+    ) -> str:
        self._close_thinking_segments()
        thought_id = self._lookup_observation_thought(tool_call_id=tool_call_id, tool_name=tool_name)
        if thought_id is None:
@@ -445,6 +473,34 @@ class _AgentProcessRecorder:
        else:
            self._mark_tool_observed(thought_id)
        self._update_thought(thought_id, observation=_json_or_text(observation))
+        return thought_id
+
+    def _publish_ui_messages(self, *, namespace: str, ui_messages: list[ToolUIMessage]) -> None:
+        try:
+            validate_tool_ui_message_batch(ui_messages)
+        except ValueError:
+            logger.warning(
+                "Ignored Agent backend UI batch that exceeds queue publication limits",
+                extra={"message_id": self._message_id, "namespace": namespace},
+                exc_info=True,
+            )
+            return
+
+        sequences: dict[str, int] = {}
+        for ui_message in ui_messages:
+            part_id = build_ui_part_id(namespace, ui_message.surface_id)
+            sequence = sequences.get(part_id, 0) + 1
+            sequences[part_id] = sequence
+            self._queue_manager.publish(
+                QueueUIPartEvent(
+                    part=MessageUIPart.from_tool_ui_message(
+                        part_id=part_id,
+                        sequence=sequence,
+                        ui_message=ui_message,
+                    )
+                ),
+                PublishFrom.APPLICATION_MANAGER,
+            )

    def _lookup_tool_thought(self, *, index: int, tool_call_id: str | None) -> str | None:
        if tool_call_id and tool_call_id in self._tool_by_call_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(
@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field

 from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities import RetrievalSourceMetadata
+from core.tools.entities.ui_entities import MessageUIPart
 from core.workflow.nodes.human_input.pause_reason import PauseReason
 from graphon.entities import WorkflowStartReason
 from graphon.enums import NodeType, WorkflowNodeExecutionMetadataKey
@@ -43,6 +44,7 @@ class QueueEvent(StrEnum):
    REASONING_CHUNK = "reasoning_chunk"
    ANNOTATION_REPLY = "annotation_reply"
    AGENT_THOUGHT = "agent_thought"
+    UI_PART = "ui_part"
    MESSAGE_FILE = "message_file"
    AGENT_LOG = "agent_log"
    ERROR = "error"
@@ -457,6 +459,13 @@ class QueueAgentThoughtEvent(AppQueueEvent):
    agent_thought_id: str


+class QueueUIPartEvent(AppQueueEvent):
+    """A validated tool UI surface revision."""
+
+    event: QueueEvent = QueueEvent.UI_PART
+    part: MessageUIPart
+
+
 class QueueMessageFileEvent(AppQueueEvent):
    """
    QueueAgentThoughtEvent entity
@@ -2,10 +2,11 @@ from collections.abc import Mapping, Sequence
 from enum import StrEnum
 from typing import Any, Literal

-from pydantic import BaseModel, ConfigDict, Field, JsonValue
+from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator

 from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities import RetrievalSourceMetadata
+from core.tools.entities.ui_entities import MessageUIPart, validate_ui_part_batch
 from core.workflow.nodes.human_input.entities import FormInputConfig, UserActionConfig
 from core.workflow.nodes.human_input.pause_reason import DifyHITLEventType
 from graphon.entities import WorkflowStartReason
@@ -30,6 +31,14 @@ class TaskStateMetadata(BaseModel):
    reasoning: dict[str, str] = Field(default_factory=dict)
    """reasoning_content per LLM node id (separated mode), accumulated across iteration/loop
    passes for that node; persisted to message_metadata"""
+    ui_parts: list[MessageUIPart] = Field(default_factory=list)
+    """Latest validated revision of each tool UI part; persisted to message metadata."""
+
+    @field_validator("ui_parts")
+    @classmethod
+    def _validate_ui_parts(cls, value: list[MessageUIPart]) -> list[MessageUIPart]:
+        validate_ui_part_batch(value)
+        return value


 class TaskState(BaseModel):
@@ -73,6 +82,7 @@ class StreamEvent(StrEnum):
    MESSAGE_FILE = "message_file"
    MESSAGE_REPLACE = "message_replace"
    AGENT_THOUGHT = "agent_thought"
+    UI_PART = "ui_part"
    AGENT_MESSAGE = "agent_message"
    WORKFLOW_STARTED = "workflow_started"
    WORKFLOW_PAUSED = "workflow_paused"
@@ -192,6 +202,14 @@ class AgentThoughtStreamResponse(StreamResponse):
    message_files: list[str] | None = None


+class UIPartStreamResponse(StreamResponse):
+    """A tool-owned UI surface revision."""
+
+    event: StreamEvent = StreamEvent.UI_PART
+    id: str
+    part: MessageUIPart
+
+
 class AgentMessageStreamResponse(StreamResponse):
    """
    AgentMessageStreamResponse entity
@@ -26,6 +26,7 @@ from core.app.entities.queue_entities import (
    QueuePingEvent,
    QueueRetrieverResourcesEvent,
    QueueStopEvent,
+    QueueUIPartEvent,
 )
 from core.app.entities.task_entities import (
    AgentMessageStreamResponse,
@@ -41,6 +42,7 @@ from core.app.entities.task_entities import (
    MessageEndStreamResponse,
    StreamEvent,
    StreamResponse,
+    UIPartStreamResponse,
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
@@ -52,6 +54,7 @@ from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.tools.entities.ui_entities import upsert_ui_part
 from events.message_event import message_was_created
 from graphon.file import FileTransferMethod
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
@@ -311,6 +314,10 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
                    agent_thought_response = self._agent_thought_to_stream_response(event)
                    if agent_thought_response is not None:
                        yield agent_thought_response
+                case QueueUIPartEvent():
+                    ui_part_response = self._ui_part_to_stream_response(event)
+                    if ui_part_response is not None:
+                        yield ui_part_response
                case QueueMessageFileEvent():
                    response = self._message_cycle_manager.message_file_to_stream_response(event)
                    if response:
@@ -526,6 +533,27 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
            task_id=self._application_generate_entity.task_id, id=message_id, answer=answer
        )

+    def _ui_part_to_stream_response(self, event: QueueUIPartEvent) -> UIPartStreamResponse | None:
+        """Upsert a UI part revision into persistent metadata and stream it."""
+        try:
+            updated_parts = upsert_ui_part(self._task_state.metadata.ui_parts, event.part)
+        except ValueError:
+            logger.warning(
+                "Ignored UI part that exceeds assistant message limits",
+                extra={"message_id": self._message_id, "part_id": event.part.part_id},
+                exc_info=True,
+            )
+            return None
+        if updated_parts is None:
+            return None
+        self._task_state.metadata.ui_parts = updated_parts
+
+        return UIPartStreamResponse(
+            task_id=self._application_generate_entity.task_id,
+            id=self._message_id,
+            part=event.part,
+        )
+
    def _agent_thought_to_stream_response(self, event: QueueAgentThoughtEvent) -> AgentThoughtStreamResponse | None:
        """
        Agent thought to stream response.
@@ -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 {}
@@ -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"

@@ -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)
@@ -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"
@@ -17,6 +17,7 @@ from core.tools.entities.tool_entities import (
    ToolParameter,
    ToolProviderType,
 )
+from core.tools.entities.ui_entities import ToolUIMessage


 class Tool(ABC):
@@ -58,7 +59,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 +87,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 +196,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
@@ -282,6 +296,13 @@ class Tool(ABC):
            message=ToolInvokeMessage.JsonMessage(json_object=object, suppress_output=suppress_output),
        )

+    def create_ui_message(self, ui_message: ToolUIMessage | dict[str, Any]) -> ToolInvokeMessage:
+        """Create a validated, model-invisible UI message for chat clients."""
+        return ToolInvokeMessage(
+            type=ToolInvokeMessage.MessageType.UI,
+            message=ToolUIMessage.model_validate(ui_message),
+        )
+
    def create_variable_message(
        self, variable_name: str, variable_value: Any, stream: bool = False
    ) -> ToolInvokeMessage:
@@ -7,6 +7,9 @@ from sqlalchemy.orm import Session

 from core.tools.builtin_tool.tool import BuiltinTool
 from core.tools.entities.tool_entities import ToolInvokeMessage
+from core.tools.entities.ui_entities import A2UI_CATALOG_ID, A2UI_PROTOCOL_VERSION
+
+_CURRENT_TIME_SURFACE_ID = "current-time"


 class CurrentTimeTool(BuiltinTool):
@@ -20,19 +23,60 @@ class CurrentTimeTool(BuiltinTool):
        app_id: str | None = None,
        message_id: str | None = None,
    ) -> Generator[ToolInvokeMessage, None, None]:
-        """
-        invoke tools
-        """
-        # get timezone
-        tz = tool_parameters.get("timezone", "UTC")
-        fm = tool_parameters.get("format") or "%Y-%m-%d %H:%M:%S %Z"
-        if tz == "UTC":
-            yield self.create_text_message(f"{datetime.now(UTC).strftime(fm)}")
-            return
+        """Return the formatted time for the model and a standard time card for chat clients.

-        try:
-            tz = pytz_timezone(tz)
-        except Exception:
-            yield self.create_text_message(f"Invalid timezone: {tz}")
-            return
-        yield self.create_text_message(f"{datetime.now(tz).strftime(fm)}")
+        Invalid timezone values retain the existing text-only error response.
+        """
+        timezone_name = tool_parameters.get("timezone", "UTC")
+        fm = tool_parameters.get("format") or "%Y-%m-%d %H:%M:%S %Z"
+        if timezone_name == "UTC":
+            current_time = datetime.now(UTC)
+        else:
+            try:
+                timezone_info = pytz_timezone(timezone_name)
+            except Exception:
+                yield self.create_text_message(f"Invalid timezone: {timezone_name}")
+                return
+            current_time = datetime.now(timezone_info)
+
+        formatted_time = current_time.strftime(fm)
+        yield self.create_text_message(formatted_time)
+        yield self.create_ui_message(
+            {
+                "messages": [
+                    {
+                        "version": A2UI_PROTOCOL_VERSION,
+                        "createSurface": {
+                            "surfaceId": _CURRENT_TIME_SURFACE_ID,
+                            "catalogId": A2UI_CATALOG_ID,
+                        },
+                    },
+                    {
+                        "version": A2UI_PROTOCOL_VERSION,
+                        "updateDataModel": {
+                            "surfaceId": _CURRENT_TIME_SURFACE_ID,
+                            "value": {"currentTime": current_time.isoformat()},
+                        },
+                    },
+                    {
+                        "version": A2UI_PROTOCOL_VERSION,
+                        "updateComponents": {
+                            "surfaceId": _CURRENT_TIME_SURFACE_ID,
+                            "components": [
+                                {
+                                    "id": "root",
+                                    "component": "Card",
+                                    "children": ["time"],
+                                },
+                                {
+                                    "id": "time",
+                                    "component": "DateTime",
+                                    "value": {"path": "/currentTime"},
+                                    "format": "datetime",
+                                },
+                            ],
+                        },
+                    },
+                ],
+            }
+        )
@@ -32,6 +32,7 @@ from core.plugin.entities.parameters import (
 from core.rag.entities import RetrievalSourceMetadata
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.constants import TOOL_SELECTOR_MODEL_IDENTITY
+from core.tools.entities.ui_entities import ToolUIMessage


 class EmojiIconDict(TypedDict):
@@ -240,13 +241,15 @@ class ToolInvokeMessage(BaseModel):
        LOG = auto()
        BLOB_CHUNK = auto()
        RETRIEVER_RESOURCES = auto()
+        UI = auto()

    type: MessageType = MessageType.TEXT
    """
        plain text, image url or link url
    """
    message: (
-        JsonMessage
+        ToolUIMessage
+        | JsonMessage
        | TextMessage
        | BlobChunkMessage
        | BlobMessage
@@ -275,6 +278,8 @@ class ToolInvokeMessage(BaseModel):
                    v = {"json_object": v}
            elif msg_type == cls.MessageType.FILE:
                v = {"file_marker": "file_marker"}
+            elif msg_type == cls.MessageType.UI:
+                v = ToolUIMessage.model_validate(v)

        return v

@@ -292,9 +297,7 @@ class ToolInvokeMessageBinary(BaseModel):


 class ToolParameter(PluginParameter):
-    """
-    Overrides type
-    """
+    """Tool-specific parameter declaration and invocation-value normalization."""

    class ToolParameterType(StrEnum):
        """
@@ -333,12 +336,28 @@ class ToolParameter(PluginParameter):
        LLM = auto()  # will be set by LLM

    type: ToolParameterType = Field(..., description="The type of the parameter")
+    multiple: bool = Field(
+        default=False,
+        description="Whether the parameter is multiple select, only valid for select or dynamic-select type",
+    )
    human_description: I18nObject | None = Field(default=None, description="The description presented to the user")
    form: ToolParameterForm = Field(..., description="The form of the parameter, schema/form/llm")
    llm_description: str | None = None
    # MCP object and array type parameters use this field to store the schema
    input_schema: dict[str, Any] | None = None

+    @model_validator(mode="after")
+    def validate_multiple(self) -> ToolParameter:
+        supports_multiple = self.type in {
+            self.ToolParameterType.SELECT,
+            self.ToolParameterType.DYNAMIC_SELECT,
+        }
+        if self.multiple and not supports_multiple:
+            raise ValueError("multiple is only valid for select and dynamic-select parameters")
+        if supports_multiple and self.default is not None and (isinstance(self.default, list) != self.multiple):
+            raise ValueError("default must be a list exactly when multiple is true")
+        return self
+
    @classmethod
    def get_simple_instance(
        cls,
@@ -378,8 +397,25 @@ class ToolParameter(PluginParameter):
            options=option_objs,
        )

-    def init_frontend_parameter(self, value: Any):
-        return init_frontend_parameter(self, self.type, value)
+    def init_frontend_parameter(self, value: Any) -> Any:
+        """Normalize a value against this tool parameter's full declaration."""
+        if not self.multiple:
+            return init_frontend_parameter(self, self.type, value)
+
+        parameter_value = self.default if value is None else value
+        if parameter_value is None:
+            parameter_value = []
+        if not isinstance(parameter_value, list):
+            raise ValueError(f"tool parameter {self.name} must be a list when multiple is true")
+        if not all(isinstance(item, str) for item in parameter_value):
+            raise ValueError(f"tool parameter {self.name} must contain only strings")
+        if self.required and not parameter_value:
+            raise ValueError(f"tool parameter {self.name} not found in tool config")
+        if self.type == self.ToolParameterType.SELECT:
+            options = [option.value for option in self.options]
+            if any(item not in options for item in parameter_value):
+                raise ValueError(f"tool parameter {self.name} value {parameter_value} not in options {options}")
+        return parameter_value


 class ToolProviderIdentity(BaseModel):
@@ -0,0 +1,675 @@
+"""Validated, read-only UI messages emitted by tools.
+
+The public wire format is the A2UI v0.9.1 flat message shape, narrowed to a
+Dify-owned catalog. Tool authors can bind display values to the surface data
+model, but cannot supply executable actions, HTML, styles, themes, or custom
+components. A complete ``ToolUIMessage`` describes exactly one surface and is
+validated as a bounded, self-contained component graph before it crosses into
+chat streaming. Sequential data-model patches are materialized with the same
+object/array upsert semantics as the web renderer so cumulative limits cannot
+be bypassed with individually small updates.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from collections.abc import Sequence
+from enum import StrEnum
+from typing import Annotated, Literal
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    JsonValue,
+    SerializerFunctionWrapHandler,
+    field_validator,
+    model_serializer,
+    model_validator,
+)
+
+A2UI_PROTOCOL = "a2ui"
+A2UI_PROTOCOL_VERSION = "v0.9.1"
+A2UI_CATALOG_ID = "https://dify.ai/a2ui/catalog/v1"
+DIFY_UI_JSON_ENVELOPE_KEY = "__dify_ui__"
+
+MAX_UI_MESSAGES = 64
+MAX_UI_COMPONENTS = 100
+MAX_UI_STRING_LENGTH = 4096
+MAX_UI_PAYLOAD_BYTES = 128 * 1024
+MAX_DATA_MODEL_DEPTH = 16
+MAX_DATA_MODEL_NODES = 2000
+MAX_DATA_MODEL_ARRAY_INDEX = 1000
+MAX_JSON_POINTER_SEGMENTS = 16
+MAX_UI_PARTS_PER_MESSAGE = 16
+MAX_UI_PARTS_PAYLOAD_BYTES = 512 * 1024
+
+_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
+_ARRAY_INDEX_PATTERN = re.compile(r"^(?:0|[1-9]\d*)$")
+_DANGEROUS_POINTER_SEGMENTS = {"__proto__", "constructor", "prototype"}
+_DANGEROUS_DATA_KEYS = _DANGEROUS_POINTER_SEGMENTS
+_MAX_HISTORY_UI_PART_CANDIDATES = MAX_UI_PARTS_PER_MESSAGE * 4
+_UI_PART_ID_NAMESPACE = uuid.uuid5(uuid.NAMESPACE_URL, "https://dify.ai/a2ui/part-id/v1")
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+        serialize_by_alias=True,
+        allow_inf_nan=False,
+    )
+
+
+class A2UIDataBinding(_StrictModel):
+    """A JSON Pointer into the surface data model."""
+
+    path: str
+
+    @field_validator("path")
+    @classmethod
+    def _validate_path(cls, value: str) -> str:
+        return _validate_json_pointer(value, label="data binding path")
+
+
+type DynamicString = str | A2UIDataBinding
+type DynamicNumber = int | float | A2UIDataBinding
+type DynamicScalar = str | int | float | bool | None | A2UIDataBinding
+
+
+class A2UIComponentType(StrEnum):
+    CARD = "Card"
+    ROW = "Row"
+    COLUMN = "Column"
+    TEXT = "Text"
+    ICON = "Icon"
+    DIVIDER = "Divider"
+    BADGE = "Badge"
+    METRIC = "Metric"
+    DATE_TIME = "DateTime"
+    PROGRESS = "Progress"
+    KEY_VALUE = "KeyValue"
+
+
+class A2UIComponent(_StrictModel):
+    """One flat component entry from the Dify catalog."""
+
+    id: str
+    component: A2UIComponentType
+    children: list[str] | None = None
+    title: DynamicString | None = None
+    gap: Literal["small", "medium", "large"] | None = None
+    align: Literal["start", "center", "end"] | None = None
+    text: DynamicString | None = None
+    variant: Literal["body", "caption"] | None = None
+    name: (
+        Literal[
+            "clock",
+            "cloud",
+            "sun",
+            "rain",
+            "snow",
+            "wind",
+            "thermometer",
+            "calendar",
+            "location",
+        ]
+        | None
+    ) = None
+    tone: Literal["neutral", "info", "success", "warning", "critical"] | None = None
+    label: DynamicString | None = None
+    value: DynamicScalar | None = None
+    unit: DynamicString | None = None
+    format: Literal["date", "time", "datetime"] | None = None
+    max: DynamicNumber | None = None
+
+    @field_validator("id")
+    @classmethod
+    def _validate_id(cls, value: str) -> str:
+        if not _ID_PATTERN.fullmatch(value):
+            raise ValueError("component id contains unsupported characters or is too long")
+        return value
+
+    @field_validator("children")
+    @classmethod
+    def _validate_children(cls, value: list[str] | None) -> list[str] | None:
+        if value is None:
+            return None
+        if len(value) > MAX_UI_COMPONENTS:
+            raise ValueError("component has too many children")
+        if len(value) != len(set(value)):
+            raise ValueError("component children must not contain duplicate ids")
+        for child_id in value:
+            if not _ID_PATTERN.fullmatch(child_id):
+                raise ValueError("child component id contains unsupported characters or is too long")
+        return value
+
+    @model_validator(mode="after")
+    def _validate_component_props(self) -> A2UIComponent:
+        present = self.model_fields_set - {"id", "component"}
+        allowed: dict[A2UIComponentType, set[str]] = {
+            A2UIComponentType.CARD: {"children", "title"},
+            A2UIComponentType.ROW: {"children", "gap", "align"},
+            A2UIComponentType.COLUMN: {"children", "gap"},
+            A2UIComponentType.TEXT: {"text", "variant"},
+            A2UIComponentType.ICON: {"name"},
+            A2UIComponentType.DIVIDER: set(),
+            A2UIComponentType.BADGE: {"text", "tone"},
+            A2UIComponentType.METRIC: {"label", "value", "unit"},
+            A2UIComponentType.DATE_TIME: {"value", "format"},
+            A2UIComponentType.PROGRESS: {"value", "max", "label"},
+            A2UIComponentType.KEY_VALUE: {"label", "value"},
+        }
+        required: dict[A2UIComponentType, set[str]] = {
+            A2UIComponentType.CARD: {"children"},
+            A2UIComponentType.ROW: {"children"},
+            A2UIComponentType.COLUMN: {"children"},
+            A2UIComponentType.TEXT: {"text"},
+            A2UIComponentType.ICON: {"name"},
+            A2UIComponentType.DIVIDER: set(),
+            A2UIComponentType.BADGE: {"text"},
+            A2UIComponentType.METRIC: {"label", "value"},
+            A2UIComponentType.DATE_TIME: {"value"},
+            A2UIComponentType.PROGRESS: {"value", "label"},
+            A2UIComponentType.KEY_VALUE: {"label", "value"},
+        }
+        unexpected = present - allowed[self.component]
+        missing = required[self.component] - present
+        if unexpected:
+            raise ValueError(f"{self.component} does not support properties: {sorted(unexpected)}")
+        if missing:
+            raise ValueError(f"{self.component} requires properties: {sorted(missing)}")
+        nullable_props = (
+            {"value"} if self.component in {A2UIComponentType.METRIC, A2UIComponentType.KEY_VALUE} else set()
+        )
+        null_props = {prop for prop in present if getattr(self, prop) is None and prop not in nullable_props}
+        if null_props:
+            raise ValueError(f"{self.component} properties cannot be null: {sorted(null_props)}")
+        if self.component == A2UIComponentType.DATE_TIME and not isinstance(self.value, str | A2UIDataBinding):
+            raise ValueError("DateTime value must be a string or data binding")
+        if self.component == A2UIComponentType.PROGRESS and (
+            isinstance(self.value, bool) or not isinstance(self.value, int | float | A2UIDataBinding)
+        ):
+            raise ValueError("Progress value must be a number or data binding")
+        if (
+            self.component == A2UIComponentType.PROGRESS
+            and "max" in present
+            and (isinstance(self.max, bool) or not isinstance(self.max, int | float | A2UIDataBinding))
+        ):
+            raise ValueError("Progress max must be a number or data binding")
+        return self
+
+    @model_serializer(mode="wrap")
+    def _serialize_component(self, handler: SerializerFunctionWrapHandler) -> dict[str, JsonValue]:
+        serialized = handler(self)
+        return {
+            key: value
+            for key, value in serialized.items()
+            if key in {"id", "component"} or key in self.model_fields_set
+        }
+
+
+class A2UICreateSurface(_StrictModel):
+    surface_id: str = Field(alias="surfaceId")
+    catalog_id: Literal["https://dify.ai/a2ui/catalog/v1"] = Field(alias="catalogId")
+
+    @field_validator("surface_id")
+    @classmethod
+    def _validate_surface_id(cls, value: str) -> str:
+        if not _ID_PATTERN.fullmatch(value):
+            raise ValueError("surface id contains unsupported characters or is too long")
+        return value
+
+
+class A2UIUpdateComponents(_StrictModel):
+    surface_id: str = Field(alias="surfaceId")
+    components: Annotated[list[A2UIComponent], Field(min_length=1, max_length=MAX_UI_COMPONENTS)]
+
+    @field_validator("surface_id")
+    @classmethod
+    def _validate_surface_id(cls, value: str) -> str:
+        if not _ID_PATTERN.fullmatch(value):
+            raise ValueError("surface id contains unsupported characters or is too long")
+        return value
+
+    @field_validator("components")
+    @classmethod
+    def _validate_unique_component_ids(cls, value: list[A2UIComponent]) -> list[A2UIComponent]:
+        ids = [component.id for component in value]
+        if len(ids) != len(set(ids)):
+            raise ValueError("updateComponents contains duplicate component ids")
+        return value
+
+
+class A2UIUpdateDataModel(_StrictModel):
+    surface_id: str = Field(alias="surfaceId")
+    value: JsonValue
+    path: str | None = None
+
+    @field_validator("surface_id")
+    @classmethod
+    def _validate_surface_id(cls, value: str) -> str:
+        if not _ID_PATTERN.fullmatch(value):
+            raise ValueError("surface id contains unsupported characters or is too long")
+        return value
+
+    @field_validator("value")
+    @classmethod
+    def _validate_value(cls, value: JsonValue) -> JsonValue:
+        _validate_data_model_value(value)
+        return value
+
+    @field_validator("path")
+    @classmethod
+    def _validate_path(cls, value: str | None) -> str | None:
+        if value is None:
+            return value
+        return _validate_json_pointer(value, label="data model path", enforce_array_index_limit=True)
+
+    @model_serializer(mode="wrap")
+    def _serialize_update(self, handler: SerializerFunctionWrapHandler) -> dict[str, JsonValue]:
+        serialized = handler(self)
+        if "path" not in self.model_fields_set:
+            serialized.pop("path", None)
+        return serialized
+
+
+class A2UIDeleteSurface(_StrictModel):
+    surface_id: str = Field(alias="surfaceId")
+
+    @field_validator("surface_id")
+    @classmethod
+    def _validate_surface_id(cls, value: str) -> str:
+        if not _ID_PATTERN.fullmatch(value):
+            raise ValueError("surface id contains unsupported characters or is too long")
+        return value
+
+
+class A2UIMessage(_StrictModel):
+    """A single A2UI v0.9.1 server message."""
+
+    version: Literal["v0.9.1"]
+    create_surface: A2UICreateSurface | None = Field(default=None, alias="createSurface")
+    update_components: A2UIUpdateComponents | None = Field(default=None, alias="updateComponents")
+    update_data_model: A2UIUpdateDataModel | None = Field(default=None, alias="updateDataModel")
+    delete_surface: A2UIDeleteSurface | None = Field(default=None, alias="deleteSurface")
+
+    @model_validator(mode="after")
+    def _validate_exactly_one_operation(self) -> A2UIMessage:
+        operations = (
+            self.create_surface,
+            self.update_components,
+            self.update_data_model,
+            self.delete_surface,
+        )
+        if sum(operation is not None for operation in operations) != 1:
+            raise ValueError("A2UI message must contain exactly one operation")
+        return self
+
+    @model_serializer(mode="wrap")
+    def _serialize_message(self, handler: SerializerFunctionWrapHandler) -> dict[str, JsonValue]:
+        return {key: value for key, value in handler(self).items() if value is not None}
+
+    @property
+    def surface_id(self) -> str:
+        operation = self.create_surface or self.update_components or self.update_data_model or self.delete_surface
+        assert operation is not None
+        return operation.surface_id
+
+
+class ToolUIMessage(_StrictModel):
+    """A complete, validated UI surface emitted by one tool result.
+
+    Validation covers both the component graph and every materialized
+    data-model revision, because clients render the message sequence in order.
+    """
+
+    protocol: Literal["a2ui"] = "a2ui"
+    protocol_version: Literal["v0.9.1"] = "v0.9.1"
+    messages: Annotated[list[A2UIMessage], Field(min_length=1, max_length=MAX_UI_MESSAGES)]
+    fallback: str | None = None
+
+    @field_validator("fallback")
+    @classmethod
+    def _validate_fallback(cls, value: str | None) -> str | None:
+        if value is not None and len(value) > MAX_UI_STRING_LENGTH:
+            raise ValueError("UI fallback is too long")
+        return value
+
+    @model_validator(mode="after")
+    def _validate_surface(self) -> ToolUIMessage:
+        first = self.messages[0]
+        if first.create_surface is None:
+            raise ValueError("first A2UI message must be createSurface")
+        if any(message.create_surface is not None for message in self.messages[1:]):
+            raise ValueError("createSurface may only appear as the first message")
+        delete_indexes = [index for index, message in enumerate(self.messages) if message.delete_surface]
+        if delete_indexes and delete_indexes != [len(self.messages) - 1]:
+            raise ValueError("deleteSurface may only appear once as the final message")
+
+        surface_id = first.surface_id
+        if any(message.surface_id != surface_id for message in self.messages):
+            raise ValueError("all A2UI messages in a tool UI message must target the same surface")
+
+        components: dict[str, A2UIComponent] = {}
+        data_model: JsonValue = {}
+        root_seen_before_delete = False
+        for message in self.messages[1:]:
+            if message.update_components is not None:
+                for component in message.update_components.components:
+                    components[component.id] = component
+                if len(components) > MAX_UI_COMPONENTS:
+                    raise ValueError("UI surface has too many components")
+            if message.update_data_model is not None:
+                data_model = _apply_data_model_update(data_model, message.update_data_model)
+                _validate_data_model_value(data_model)
+            if message.delete_surface is not None:
+                root_seen_before_delete = "root" in components
+
+        if "root" not in components and not root_seen_before_delete:
+            raise ValueError("UI surface must define a component with id 'root'")
+
+        _validate_component_graph(components)
+        payload = self.model_dump(mode="json", by_alias=True, exclude_none=True)
+        _validate_bounded_json(payload)
+        return self
+
+    @model_serializer(mode="wrap")
+    def _serialize_ui_message(self, handler: SerializerFunctionWrapHandler) -> dict[str, JsonValue]:
+        return {key: value for key, value in handler(self).items() if value is not None}
+
+    @property
+    def surface_id(self) -> str:
+        return self.messages[0].surface_id
+
+
+class MessageUIPart(_StrictModel):
+    """SSE/history representation of one tool-owned UI surface revision."""
+
+    part_id: Annotated[str, Field(min_length=1, max_length=512)]
+    sequence: Annotated[int, Field(ge=1)]
+    protocol: Literal["a2ui"] = "a2ui"
+    protocol_version: Literal["v0.9.1"] = "v0.9.1"
+    messages: Annotated[list[A2UIMessage], Field(min_length=1, max_length=MAX_UI_MESSAGES)]
+    fallback: str | None = None
+
+    @model_validator(mode="after")
+    def _validate_messages(self) -> MessageUIPart:
+        ToolUIMessage(
+            protocol=self.protocol,
+            protocol_version=self.protocol_version,
+            messages=self.messages,
+            fallback=self.fallback,
+        )
+        return self
+
+    @model_serializer(mode="wrap")
+    def _serialize_part(self, handler: SerializerFunctionWrapHandler) -> dict[str, JsonValue]:
+        return {key: value for key, value in handler(self).items() if value is not None}
+
+    @classmethod
+    def from_tool_ui_message(
+        cls,
+        *,
+        part_id: str,
+        sequence: int,
+        ui_message: ToolUIMessage,
+    ) -> MessageUIPart:
+        return cls(
+            part_id=part_id,
+            sequence=sequence,
+            protocol=ui_message.protocol,
+            protocol_version=ui_message.protocol_version,
+            messages=ui_message.messages,
+            fallback=ui_message.fallback,
+        )
+
+
+def extract_ui_message_from_json(value: JsonValue) -> ToolUIMessage | None:
+    """Recognize the reserved compatibility envelope used by older SDKs."""
+
+    if not isinstance(value, dict) or set(value) != {DIFY_UI_JSON_ENVELOPE_KEY}:
+        return None
+    return ToolUIMessage.model_validate(value[DIFY_UI_JSON_ENVELOPE_KEY])
+
+
+def parse_tool_ui_messages(value: object) -> list[ToolUIMessage]:
+    """Validate a list received through the Agent backend metadata channel."""
+
+    if not isinstance(value, list):
+        raise ValueError("dify_ui_messages metadata must be a list")
+    if len(value) > MAX_UI_PARTS_PER_MESSAGE:
+        raise ValueError(f"tool UI batch cannot contain more than {MAX_UI_PARTS_PER_MESSAGE} messages")
+    messages = [ToolUIMessage.model_validate(item) for item in value]
+    validate_tool_ui_message_batch(messages)
+    return messages
+
+
+def build_ui_part_id(namespace: str, surface_id: str) -> str:
+    """Build a stable, bounded ID from an unambiguous tool/surface tuple."""
+    if not isinstance(namespace, str) or not isinstance(surface_id, str):
+        raise TypeError("UI part namespace and surface id must be strings")
+    identity = json.dumps((namespace, surface_id), ensure_ascii=True, separators=(",", ":"))
+    return f"ui-{uuid.uuid5(_UI_PART_ID_NAMESPACE, identity)}"
+
+
+def validate_tool_ui_message_batch(messages: Sequence[ToolUIMessage]) -> None:
+    """Enforce the per-tool-call UI batch budget before queue publication."""
+    if len(messages) > MAX_UI_PARTS_PER_MESSAGE:
+        raise ValueError(f"tool UI batch cannot contain more than {MAX_UI_PARTS_PER_MESSAGE} messages")
+    if _serialized_model_list_size(messages) > MAX_UI_PARTS_PAYLOAD_BYTES:
+        raise ValueError(f"tool UI batch payload cannot exceed {MAX_UI_PARTS_PAYLOAD_BYTES} bytes")
+
+
+def validate_ui_part_batch(parts: Sequence[MessageUIPart]) -> None:
+    """Enforce persisted/current assistant-message UI budgets."""
+    if len(parts) > MAX_UI_PARTS_PER_MESSAGE:
+        raise ValueError(f"assistant message cannot contain more than {MAX_UI_PARTS_PER_MESSAGE} UI parts")
+    part_ids = [part.part_id for part in parts]
+    if len(part_ids) != len(set(part_ids)):
+        raise ValueError("assistant message UI parts must have distinct part ids")
+    if _serialized_model_list_size(parts) > MAX_UI_PARTS_PAYLOAD_BYTES:
+        raise ValueError(f"assistant message UI parts cannot exceed {MAX_UI_PARTS_PAYLOAD_BYTES} bytes")
+
+
+def upsert_ui_part(parts: Sequence[MessageUIPart], part: MessageUIPart) -> list[MessageUIPart] | None:
+    """Return a bounded new part list, or ``None`` for a stale revision."""
+    candidate = list(parts)
+    for index, existing_part in enumerate(candidate):
+        if existing_part.part_id != part.part_id:
+            continue
+        if existing_part.sequence >= part.sequence:
+            return None
+        candidate[index] = part
+        break
+    else:
+        candidate.append(part)
+
+    validate_ui_part_batch(candidate)
+    return candidate
+
+
+def parse_history_ui_parts(value: object) -> list[MessageUIPart]:
+    """Best-effort recovery of bounded UI parts from untrusted historical metadata."""
+    if not isinstance(value, list):
+        return []
+
+    parts: list[MessageUIPart] = []
+    for item in value[:_MAX_HISTORY_UI_PART_CANDIDATES]:
+        try:
+            part = MessageUIPart.model_validate(item)
+            updated_parts = upsert_ui_part(parts, part)
+        except (TypeError, ValueError):
+            continue
+        if updated_parts is not None:
+            parts = updated_parts
+    return parts
+
+
+def _validate_component_graph(components: dict[str, A2UIComponent]) -> None:
+    for component in components.values():
+        for child_id in component.children or []:
+            if child_id not in components:
+                raise ValueError(f"component {component.id!r} references unknown child {child_id!r}")
+
+    visiting: set[str] = set()
+    visited: set[str] = set()
+
+    def visit(component_id: str) -> None:
+        if component_id in visiting:
+            raise ValueError("UI component graph contains a cycle")
+        if component_id in visited:
+            return
+        visiting.add(component_id)
+        for child_id in components[component_id].children or []:
+            visit(child_id)
+        visiting.remove(component_id)
+        visited.add(component_id)
+
+    for component_id in components:
+        visit(component_id)
+
+    parent_by_child: dict[str, str] = {}
+    for component in components.values():
+        for child_id in component.children or []:
+            if child_id == "root":
+                raise ValueError("root component cannot be referenced as a child")
+            existing_parent = parent_by_child.get(child_id)
+            if existing_parent is not None:
+                raise ValueError(
+                    f"component {child_id!r} has multiple parents: {existing_parent!r} and {component.id!r}"
+                )
+            parent_by_child[child_id] = component.id
+
+    reachable = {"root"}
+    pending = ["root"]
+    while pending:
+        component_id = pending.pop()
+        for child_id in components[component_id].children or []:
+            if child_id in reachable:
+                continue
+            reachable.add(child_id)
+            pending.append(child_id)
+
+    unreachable = set(components) - reachable
+    if unreachable:
+        raise ValueError(f"UI component graph contains components unreachable from root: {sorted(unreachable)}")
+
+
+def _serialized_model_list_size(values: Sequence[BaseModel]) -> int:
+    payload = [value.model_dump(mode="json", by_alias=True, exclude_none=True) for value in values]
+    return len(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode())
+
+
+def _validate_json_pointer(
+    value: str,
+    *,
+    label: str,
+    enforce_array_index_limit: bool = False,
+) -> str:
+    if not value.startswith("/"):
+        raise ValueError(f"{label} must be a JSON Pointer beginning with '/'")
+    if len(value) > 256:
+        raise ValueError(f"{label} is too long")
+    if re.search(r"~(?![01])", value):
+        raise ValueError(f"{label} contains an invalid JSON Pointer escape")
+    decoded_segments = _decode_json_pointer(value)
+    if len(decoded_segments) > MAX_JSON_POINTER_SEGMENTS:
+        raise ValueError(f"{label} contains too many segments")
+    if set(decoded_segments).intersection(_DANGEROUS_POINTER_SEGMENTS):
+        raise ValueError(f"{label} contains a forbidden segment")
+    if enforce_array_index_limit and any(
+        segment.isdecimal() and int(segment) > MAX_DATA_MODEL_ARRAY_INDEX for segment in decoded_segments
+    ):
+        raise ValueError(f"{label} contains an array index larger than {MAX_DATA_MODEL_ARRAY_INDEX}")
+    return value
+
+
+def _decode_json_pointer(value: str) -> list[str]:
+    if value == "/":
+        return []
+    return [segment.replace("~1", "/").replace("~0", "~") for segment in value.split("/")[1:]]
+
+
+def _apply_data_model_update(current: JsonValue, update: A2UIUpdateDataModel) -> JsonValue:
+    path = update.path
+    if path is None or path == "/":
+        return update.value
+
+    segments = _decode_json_pointer(path)
+
+    def apply_at(node: JsonValue | None, segment_index: int) -> JsonValue:
+        if segment_index == len(segments):
+            return update.value
+
+        segment = segments[segment_index]
+        is_array_segment = _ARRAY_INDEX_PATTERN.fullmatch(segment) is not None
+        if isinstance(node, list) or (not isinstance(node, dict) and is_array_segment):
+            if not is_array_segment:
+                raise ValueError("data model array paths must use canonical non-negative indexes")
+            index = int(segment)
+            if index > MAX_DATA_MODEL_ARRAY_INDEX:
+                raise ValueError(f"data model array index cannot exceed {MAX_DATA_MODEL_ARRAY_INDEX}")
+
+            next_node = list(node) if isinstance(node, list) else []
+            if index > len(next_node):
+                raise ValueError("data model array index cannot create a gap")
+            child = next_node[index] if index < len(next_node) else None
+            updated_child = apply_at(child, segment_index + 1)
+            if index == len(next_node):
+                next_node.append(updated_child)
+            else:
+                next_node[index] = updated_child
+            return next_node
+
+        next_node = dict(node) if isinstance(node, dict) else {}
+        next_node[segment] = apply_at(next_node.get(segment), segment_index + 1)
+        return next_node
+
+    return apply_at(current, 0)
+
+
+def _validate_data_model_value(value: JsonValue) -> None:
+    nodes = 0
+
+    def walk(node: JsonValue, *, depth: int) -> None:
+        nonlocal nodes
+        nodes += 1
+        if nodes > MAX_DATA_MODEL_NODES:
+            raise ValueError("data model value has too many nodes")
+        if depth > MAX_DATA_MODEL_DEPTH:
+            raise ValueError("data model value is nested too deeply")
+        if isinstance(node, list):
+            for item in node:
+                walk(item, depth=depth + 1)
+            return
+        if isinstance(node, dict):
+            unsafe_keys = set(node).intersection(_DANGEROUS_DATA_KEYS)
+            if unsafe_keys:
+                raise ValueError(f"data model value contains forbidden keys: {sorted(unsafe_keys)}")
+            for item in node.values():
+                walk(item, depth=depth + 1)
+
+    walk(value, depth=0)
+
+
+def _validate_bounded_json(value: JsonValue) -> None:
+    def walk(node: JsonValue) -> None:
+        if isinstance(node, str):
+            if len(node) > MAX_UI_STRING_LENGTH:
+                raise ValueError("UI payload contains a string that is too long")
+            return
+        if isinstance(node, list):
+            for item in node:
+                walk(item)
+            return
+        if isinstance(node, dict):
+            for item in node.values():
+                walk(item)
+
+    walk(value)
+    encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
+    if len(encoded) > MAX_UI_PAYLOAD_BYTES:
+        raise ValueError("UI payload is too large")
@@ -3,6 +3,7 @@ import json
 import logging
 from collections.abc import Generator, Iterable
 from copy import deepcopy
+from dataclasses import dataclass
 from datetime import UTC, datetime
 from mimetypes import guess_type
 from typing import Any, Union, cast
@@ -21,6 +22,12 @@ from core.tools.entities.tool_entities import (
    ToolInvokeMeta,
    ToolParameter,
 )
+from core.tools.entities.ui_entities import (
+    DIFY_UI_JSON_ENVELOPE_KEY,
+    ToolUIMessage,
+    extract_ui_message_from_json,
+    validate_tool_ui_message_batch,
+)
 from core.tools.errors import (
    ToolEngineInvokeError,
    ToolInvokeError,
@@ -40,6 +47,16 @@ from models.model import Message, MessageFile
 logger = logging.getLogger(__name__)


+@dataclass(frozen=True)
+class ToolAgentInvokeResult:
+    """Agent-facing tool result with UI kept outside the model observation."""
+
+    observation: str
+    message_files: list[str]
+    ui_messages: list[ToolUIMessage]
+    meta: ToolInvokeMeta
+
+
 class ToolEngine:
    """
    Tool runtime engine take care of the tool executions.
@@ -59,9 +76,13 @@ class ToolEngine:
        conversation_id: str | None = None,
        app_id: str | None = None,
        message_id: str | None = None,
-    ) -> tuple[str, list[str], ToolInvokeMeta]:
+    ) -> ToolAgentInvokeResult:
        """
-        Agent invokes the tool with the given arguments.
+        Invoke a tool for an agent.
+
+        UI messages are validated and returned on a dedicated channel. They are
+        intentionally excluded from both the LLM observation and tracing
+        callback output.
        """
        # check if arguments is a string
        if isinstance(tool_parameters, str):
@@ -103,7 +124,7 @@ class ToolEngine:
                conversation_id=message.conversation_id,
            )

-            message_list = list(messages)
+            message_list, ui_messages = ToolEngine.collect_agent_messages(messages)

            # extract binary data from tool invoke message
            binary_files = ToolEngine._extract_tool_response_binary_and_text(message_list)
@@ -126,7 +147,12 @@ class ToolEngine:
            )

            # transform tool invoke message to get LLM friendly message
-            return plain_text, message_files, meta
+            return ToolAgentInvokeResult(
+                observation=plain_text,
+                message_files=message_files,
+                ui_messages=ui_messages,
+                meta=meta,
+            )
        except ToolProviderCredentialValidationError as e:
            logger.error(e, exc_info=True)
            error_response = "Please check your tool provider credentials"
@@ -148,13 +174,23 @@ class ToolEngine:
            error_response = f"tool invoke error: {meta.error}"
            agent_tool_callback.on_tool_error(e)
            logger.error(e, exc_info=True)
-            return error_response, [], meta
+            return ToolAgentInvokeResult(
+                observation=error_response,
+                message_files=[],
+                ui_messages=[],
+                meta=meta,
+            )
        except Exception as e:
            error_response = f"unknown error: {e}"
            agent_tool_callback.on_tool_error(e)
            logger.error(e, exc_info=True)

-        return error_response, [], ToolInvokeMeta.error_instance(error_response)
+        return ToolAgentInvokeResult(
+            observation=error_response,
+            message_files=[],
+            ui_messages=[],
+            meta=ToolInvokeMeta.error_instance(error_response),
+        )

    @staticmethod
    def generic_invoke(
@@ -197,7 +233,7 @@ class ToolEngine:
                tool_outputs=response,
            )

-            return response
+            return ToolEngine.normalize_ui_messages(response)
        except Exception as e:
            workflow_tool_callback.on_tool_error(e)
            raise e
@@ -266,7 +302,10 @@ class ToolEngine:
                        ensure_ascii=False,
                    )
                )
-            elif response.type == ToolInvokeMessage.MessageType.VARIABLE:
+            elif response.type in {
+                ToolInvokeMessage.MessageType.VARIABLE,
+                ToolInvokeMessage.MessageType.UI,
+            }:
                continue
            else:
                parts.append(str(response.message))
@@ -278,6 +317,124 @@ class ToolEngine:

        return "".join(parts)

+    @staticmethod
+    def normalize_ui_messages(messages: Iterable[ToolInvokeMessage]) -> Generator[ToolInvokeMessage, None, None]:
+        """Convert reserved variable/JSON compatibility envelopes into native UI."""
+        for message in messages:
+            if message.type == ToolInvokeMessage.MessageType.VARIABLE:
+                variable_message = cast(ToolInvokeMessage.VariableMessage, message.message)
+                if variable_message.variable_name != DIFY_UI_JSON_ENVELOPE_KEY:
+                    yield message
+                    continue
+                yield ToolInvokeMessage(
+                    type=ToolInvokeMessage.MessageType.UI,
+                    message=ToolUIMessage.model_validate(variable_message.variable_value),
+                    meta=message.meta,
+                )
+                continue
+
+            if message.type != ToolInvokeMessage.MessageType.JSON:
+                yield message
+                continue
+
+            json_message = cast(ToolInvokeMessage.JsonMessage, message.message)
+            ui_message = extract_ui_message_from_json(json_message.json_object)
+            if ui_message is None:
+                yield message
+                continue
+            yield ToolInvokeMessage(type=ToolInvokeMessage.MessageType.UI, message=ui_message, meta=message.meta)
+
+    @staticmethod
+    def collect_agent_messages(
+        messages: Iterable[ToolInvokeMessage],
+    ) -> tuple[list[ToolInvokeMessage], list[ToolUIMessage]]:
+        """Collect a bounded UI batch while preserving all non-UI observations.
+
+        A malformed or over-budget UI message invalidates the complete UI
+        batch. Remaining UI is skipped without parsing, but the input iterable
+        is still consumed so later text and file messages are retained.
+        """
+        normalized_messages: list[ToolInvokeMessage] = []
+        ui_messages: list[ToolUIMessage] = []
+        ui_batch_rejected = False
+
+        for message in messages:
+            is_ui_transport = ToolEngine._is_ui_transport_message(message)
+            if ui_batch_rejected and is_ui_transport:
+                continue
+
+            try:
+                normalized = ToolEngine.normalize_ui_messages([message])
+            except (TypeError, ValueError):
+                if not is_ui_transport:
+                    raise
+                # Reserved UI envelopes must never fall back to JSON/variable
+                # observations, even when malformed.
+                ui_batch_rejected = True
+                ui_messages.clear()
+                normalized_messages = [
+                    normalized_message
+                    for normalized_message in normalized_messages
+                    if normalized_message.type != ToolInvokeMessage.MessageType.UI
+                ]
+                logger.warning("Ignored invalid reserved tool UI message", exc_info=True)
+                continue
+
+            try:
+                for normalized_message in normalized:
+                    if normalized_message.type != ToolInvokeMessage.MessageType.UI:
+                        normalized_messages.append(normalized_message)
+                        continue
+                    if ui_batch_rejected:
+                        continue
+
+                    ui_message = cast(ToolUIMessage, normalized_message.message)
+                    candidate = [*ui_messages, ui_message]
+                    try:
+                        validate_tool_ui_message_batch(candidate)
+                    except ValueError:
+                        ui_batch_rejected = True
+                        ui_messages.clear()
+                        normalized_messages = [
+                            collected_message
+                            for collected_message in normalized_messages
+                            if collected_message.type != ToolInvokeMessage.MessageType.UI
+                        ]
+                        logger.warning(
+                            "Ignored tool UI batch that exceeds agent invocation limits",
+                            exc_info=True,
+                        )
+                        continue
+
+                    ui_messages = candidate
+                    normalized_messages.append(normalized_message)
+            except (TypeError, ValueError):
+                if not is_ui_transport:
+                    raise
+                ui_batch_rejected = True
+                ui_messages.clear()
+                normalized_messages = [
+                    normalized_message
+                    for normalized_message in normalized_messages
+                    if normalized_message.type != ToolInvokeMessage.MessageType.UI
+                ]
+                logger.warning("Ignored invalid reserved tool UI message", exc_info=True)
+
+        return normalized_messages, ui_messages
+
+    @staticmethod
+    def _is_ui_transport_message(message: ToolInvokeMessage) -> bool:
+        if message.type == ToolInvokeMessage.MessageType.UI:
+            return True
+        if message.type == ToolInvokeMessage.MessageType.VARIABLE:
+            variable_message = cast(ToolInvokeMessage.VariableMessage, message.message)
+            return variable_message.variable_name == DIFY_UI_JSON_ENVELOPE_KEY
+        if message.type == ToolInvokeMessage.MessageType.JSON:
+            json_message = cast(ToolInvokeMessage.JsonMessage, message.message)
+            value = json_message.json_object
+            return isinstance(value, dict) and set(value) == {DIFY_UI_JSON_ENVELOPE_KEY}
+        return False
+
    @staticmethod
    def _extract_tool_response_binary_and_text(
        tool_response: list[ToolInvokeMessage],
@@ -90,6 +90,7 @@ from .system_variables import SystemVariableKey, get_system_text
 if TYPE_CHECKING:
    from core.tools.__base.tool import Tool
    from core.tools.entities.tool_entities import ToolInvokeMessage as CoreToolInvokeMessage
+    from core.tools.entities.ui_entities import ToolUIMessage
    from graphon.nodes.llm.file_saver import LLMFileSaver
    from graphon.nodes.tool.entities import ToolNodeData

@@ -666,6 +667,11 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
    ) -> Generator[ToolRuntimeMessage, None, None]:
        try:
            for message in messages:
+                # Tool Gen-UI is currently a chat presentation channel. Graphon
+                # does not define an equivalent runtime message, so workflow
+                # execution safely ignores it instead of coercing it to JSON.
+                if message.type.value == "ui":
+                    continue
                yield self._convert_message(message)
        except Exception as exc:
            raise self._map_invocation_exception(exc, provider_name=provider_name) from exc
@@ -686,6 +692,7 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
        | CoreToolInvokeMessage.FileMessage
        | CoreToolInvokeMessage.VariableMessage
        | CoreToolInvokeMessage.RetrieverResourceMessage
+        | ToolUIMessage
        | None,
    ) -> (
        ToolRuntimeMessage.TextMessage
@@ -2,7 +2,7 @@ from __future__ import annotations

 from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Literal, Protocol, cast
+from typing import Any, Final, Literal, Protocol

 from dify_agent.layers.dify_core_tools import DifyCoreToolConfig, DifyCoreToolProviderType, DifyCoreToolsLayerConfig
 from dify_agent.layers.dify_plugin import (
@@ -29,6 +29,13 @@ from models.provider_ids import ToolProviderID
 from models.tools import WorkflowToolProvider
 from services.tools.mcp_tools_manage_service import MCPToolManageService

+_CORE_TOOL_PROVIDER_TYPES: Final[dict[ToolProviderType, DifyCoreToolProviderType]] = {
+    ToolProviderType.BUILT_IN: "builtin",
+    ToolProviderType.API: "api",
+    ToolProviderType.WORKFLOW: "workflow",
+    ToolProviderType.MCP: "mcp",
+}
+

 class WorkflowAgentDifyToolsBuildError(ValueError):
    """Raised when Agent Soul tools cannot be prepared for Agent backend."""
@@ -235,7 +242,7 @@ class WorkflowAgentDifyToolsBuilder:
            if tool_config.tool_name is not None:
                expanded.append(tool_config)
                continue
-            provider_type = ToolProviderType.value_of(tool_config.provider_type)
+            provider_type = tool_config.provider_type
            provider_id = self._provider_id(tool_config)
            try:
                tool_names = self._provider_declared_tool_names(
@@ -279,7 +286,7 @@ class WorkflowAgentDifyToolsBuilder:
        tenant_id: str,
        tool_config: AgentSoulDifyToolConfig,
    ) -> AgentSoulDifyToolConfig:
-        if tool_config.provider_type != ToolProviderType.MCP.value:
+        if tool_config.provider_type is not ToolProviderType.MCP:
            return tool_config
        provider_id = self._mcp_provider_id_resolver(tenant_id=tenant_id, provider_id=self._provider_id(tool_config))
        return tool_config.model_copy(update={"provider_id": provider_id, "plugin_id": None, "provider": None})
@@ -326,7 +333,7 @@ class WorkflowAgentDifyToolsBuilder:
    def _to_agent_tool_entity(tool_config: AgentSoulDifyToolConfig) -> AgentToolEntity:
        assert tool_config.tool_name is not None
        return AgentToolEntity(
-            provider_type=ToolProviderType.value_of(tool_config.provider_type),
+            provider_type=tool_config.provider_type,
            provider_id=WorkflowAgentDifyToolsBuilder._provider_id(tool_config),
            tool_name=tool_config.tool_name,
            tool_parameters=dict(tool_config.runtime_parameters),
@@ -343,24 +350,16 @@ class WorkflowAgentDifyToolsBuilder:

    @staticmethod
    def _provider_key(tool_config: AgentSoulDifyToolConfig) -> tuple[ToolProviderType, str]:
-        return (
-            ToolProviderType.value_of(tool_config.provider_type),
-            WorkflowAgentDifyToolsBuilder._provider_id(tool_config),
-        )
+        return (tool_config.provider_type, WorkflowAgentDifyToolsBuilder._provider_id(tool_config))

    @staticmethod
    def _tool_layer_destination(tool_config: AgentSoulDifyToolConfig) -> Literal["plugin", "core"]:
-        provider_type = ToolProviderType.value_of(tool_config.provider_type)
+        provider_type = tool_config.provider_type
        if provider_type is ToolProviderType.PLUGIN or (
            provider_type is ToolProviderType.BUILT_IN and _is_plugin_provider_id(tool_config.provider_id)
        ):
            return "plugin"
-        if provider_type in {
-            ToolProviderType.BUILT_IN,
-            ToolProviderType.API,
-            ToolProviderType.WORKFLOW,
-            ToolProviderType.MCP,
-        }:
+        if provider_type in _CORE_TOOL_PROVIDER_TYPES:
            return "core"
        if provider_type is ToolProviderType.DATASET_RETRIEVAL:
            raise WorkflowAgentDifyToolsBuildError(
@@ -417,7 +416,7 @@ class WorkflowAgentDifyToolsBuilder:
    ) -> DifyCoreToolConfig:
        parameters = self._prepared_parameters(tool_runtime)
        return DifyCoreToolConfig(
-            provider_type=cast(DifyCoreToolProviderType, tool_config.provider_type),
+            provider_type=_CORE_TOOL_PROVIDER_TYPES[tool_config.provider_type],
            provider_id=self._provider_id(tool_config),
            tool_name=tool_config.tool_name or exposed_name,
            credential_id=tool_config.credential_ref.id if tool_config.credential_ref else None,
@@ -12,6 +12,7 @@ import json
 import subprocess
 import sys
 import tempfile
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, Literal, TypedDict

@@ -24,6 +25,16 @@ LOCK_PATH = API_ROOT / "knowledge-fs-contract.lock.json"
 DEFAULT_REPOSITORY = WORKSPACE_ROOT.parent / "knowledge-fs"
 OPENAPI_METHODS = ("delete", "get", "head", "options", "patch", "post", "put", "trace")
 PROXY_METHODS = frozenset({"delete", "get", "patch", "post", "put"})
+CONSOLE_PROXY_ERROR_SCHEMA_NAME = "ConsoleProxyError"
+CONSOLE_PROXY_ERROR_SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "required": ["code", "message", "status"],
+    "properties": {
+        "code": {"type": "string"},
+        "message": {"type": "string"},
+        "status": {"type": "integer"},
+    },
+}


 class ContractDeclaration(TypedDict):
@@ -38,6 +49,7 @@ class ContractDeclaration(TypedDict):
    request_headers: tuple[str, ...]
    response_headers: tuple[str, ...]
    response_media_types: tuple[str, ...]
+    error_status_map: tuple[tuple[int, int], ...]


 type DeclarationField = Literal[
@@ -70,6 +82,7 @@ def main() -> None:
    mode.add_argument("--check", action="store_true")
    mode.add_argument("--update-lock", action="store_true")
    parser.add_argument("--repository", type=Path, default=DEFAULT_REPOSITORY)
+    parser.add_argument("--output-openapi", type=Path)
    args = parser.parse_args()

    repository = args.repository.resolve()
@@ -101,7 +114,15 @@ def main() -> None:
        )

    document: dict[str, Any] = json.loads(openapi_content)
-    validate_declarations(document, console_contract_declarations())
+    declarations = console_contract_declarations()
+    validate_declarations(document, declarations)
+
+    if args.output_openapi:
+        filtered_document = filter_openapi_document(document, declarations)
+        filtered_document["x-dify-source-openapi-sha256"] = openapi_sha256
+        filtered_document["x-dify-console-declarations-sha256"] = contract_declarations_sha256(declarations)
+        args.output_openapi.parent.mkdir(parents=True, exist_ok=True)
+        args.output_openapi.write_text(json.dumps(filtered_document, indent=2) + "\n")

    if args.update_lock:
        LOCK_PATH.write_text(
@@ -147,8 +168,7 @@ def validate_declarations(document: dict[str, Any], declarations: tuple[Contract
            raise ValueError(f"KnowledgeFS OpenAPI path must be absolute: {path}")
        if method not in PROXY_METHODS:
            raise ValueError(f"KnowledgeFS proxy does not support {method.upper()} {path}")
-        expected: ContractDeclaration = {
-            "operation_id": operation_id,
+        expected: dict[DeclarationField, object] = {
            "method": method.upper(),
            "path": path[1:],
            "required_scope": required_scope(operation),
@@ -166,11 +186,114 @@ def validate_declarations(document: dict[str, Any], declarations: tuple[Contract
                    f"KnowledgeFS operation {operation_id} field {field} drifted: "
                    f"expected {expected_value!r}, received {received_value!r}"
                )
+        validate_error_status_map(operation_id, declaration["error_status_map"])
+
+
+def filter_openapi_document(
+    document: dict[str, Any],
+    declarations: tuple[ContractDeclaration, ...],
+) -> dict[str, Any]:
+    """Return a code-generation document containing only Console-allowlisted operations."""
+    filtered_document: dict[str, Any] = {
+        key: value for key, value in document.items() if key not in {"components", "paths"}
+    }
+    source_paths = document.get("paths", {})
+    filtered_paths: dict[str, Any] = {}
+
+    for declaration in declarations:
+        path = f"/{declaration['path']}"
+        method = declaration["method"].lower()
+        source_path_item = source_paths[path]
+        path_metadata = {key: value for key, value in source_path_item.items() if key not in OPENAPI_METHODS}
+        filtered_path_item = filtered_paths.setdefault(path, path_metadata)
+        filtered_operation = deepcopy(source_path_item[method])
+        _rewrite_proxy_error_responses(filtered_operation, declaration["error_status_map"])
+        filtered_path_item[method] = filtered_operation
+
+    filtered_document["paths"] = filtered_paths
+
+    source_components = document.get("components", {})
+    filtered_components = {key: value for key, value in source_components.items() if key != "schemas"}
+    source_schemas = source_components.get("schemas", {})
+    available_schemas = {**source_schemas, CONSOLE_PROXY_ERROR_SCHEMA_NAME: CONSOLE_PROXY_ERROR_SCHEMA}
+    schema_names = _referenced_schema_names(filtered_paths, available_schemas)
+    filtered_components["schemas"] = {
+        name: schema for name, schema in available_schemas.items() if name in schema_names
+    }
+    filtered_document["components"] = filtered_components
+    return filtered_document
+
+
+def validate_error_status_map(operation_id: str, error_status_map: tuple[tuple[int, int], ...]) -> None:
+    """Validate the status normalization advertised by one Console operation."""
+    upstream_statuses: set[int] = set()
+    for upstream_status, console_status in error_status_map:
+        if upstream_status in upstream_statuses:
+            raise ValueError(f"KnowledgeFS operation {operation_id} has duplicate error status: {upstream_status}")
+        if not 400 <= upstream_status <= 599 or not 400 <= console_status <= 599:
+            raise ValueError(f"KnowledgeFS operation {operation_id} has invalid error status mapping")
+        upstream_statuses.add(upstream_status)
+
+
+def _rewrite_proxy_error_responses(
+    operation: dict[str, Any],
+    error_status_map: tuple[tuple[int, int], ...],
+) -> None:
+    responses = operation.setdefault("responses", {})
+    proxy_error_response = {
+        "description": "Error normalized by the Dify Console KnowledgeFS proxy.",
+        "content": {
+            "application/json": {"schema": {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"}}
+        },
+    }
+    for upstream_status, console_status in error_status_map:
+        existing_target = responses.get(str(console_status)) if upstream_status != console_status else None
+        responses.pop(str(upstream_status), None)
+        normalized_response: dict[str, Any] = deepcopy(proxy_error_response)
+        existing_schema = (
+            existing_target.get("content", {}).get("application/json", {}).get("schema")
+            if isinstance(existing_target, dict)
+            else None
+        )
+        if existing_schema is not None:
+            normalized_response["content"]["application/json"]["schema"] = {
+                "oneOf": [
+                    deepcopy(existing_schema),
+                    {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"},
+                ]
+            }
+        responses[str(console_status)] = normalized_response
+
+
+def _referenced_schema_names(value: Any, schemas: dict[str, Any]) -> set[str]:
+    reference_prefix = "#/components/schemas/"
+    selected: set[str] = set()
+    pending: list[Any] = [value]
+
+    while pending:
+        current = pending.pop()
+        if isinstance(current, list):
+            pending.extend(current)
+            continue
+        if not isinstance(current, dict):
+            continue
+
+        reference = current.get("$ref")
+        if isinstance(reference, str) and reference.startswith(reference_prefix):
+            name = reference.removeprefix(reference_prefix)
+            if name not in selected:
+                if name not in schemas:
+                    raise ValueError(f"KnowledgeFS OpenAPI references missing schema: {name}")
+                selected.add(name)
+                pending.append(schemas[name])
+        pending.extend(current.values())
+
+    return selected


 def console_contract_declarations() -> tuple[ContractDeclaration, ...]:
    """Return transport declarations from the runtime Console operation registry."""
-    from services.knowledge_fs_proxy import KNOWLEDGE_FS_CONSOLE_OPERATIONS
+    from services.knowledge_fs_operations import KNOWLEDGE_FS_CONSOLE_OPERATIONS

    return tuple(
        {
@@ -183,11 +306,18 @@ def console_contract_declarations() -> tuple[ContractDeclaration, ...]:
            "request_headers": operation.request_headers,
            "response_headers": operation.response_headers,
            "response_media_types": operation.response_media_types,
+            "error_status_map": operation.error_status_map,
        }
        for operation in KNOWLEDGE_FS_CONSOLE_OPERATIONS
    )


+def contract_declarations_sha256(declarations: tuple[ContractDeclaration, ...]) -> str:
+    """Return a stable digest for the runtime Console operation declarations."""
+    content = json.dumps(declarations, separators=(",", ":"), sort_keys=True).encode()
+    return sha256(content)
+
+
 def response_kind(operation: dict[str, Any]) -> str:
    media_types = response_media_types(operation)
    if "text/event-stream" in media_types:
@@ -157,6 +157,7 @@ def init_app(app: DifyApp) -> Celery:
        "tasks.regenerate_summary_index_task",  # summary index regeneration
        "tasks.initialize_created_app_rbac_access_task",  # app access initialization
        "tasks.install_default_plugins_task",  # tenant default plugin installation
+        "tasks.refresh_billing_vector_space_task",  # billing vector-space cache refresh
        "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
        "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
    ]
@@ -1,5 +1,6 @@
 import json
-from typing import cast, override
+import logging
+from typing import assert_never, cast, override

 import flask_login
 from flask import Request, Response, request
@@ -11,6 +12,7 @@ from werkzeug.exceptions import NotFound, Unauthorized
 from configs import dify_config
 from constants import HEADER_NAME_APP_CODE
 from core.db.session_factory import session_factory
+from core.logging.context import set_identity_context
 from dify_app import DifyApp
 from libs.passport import PassportService
 from libs.token import extract_access_token, extract_console_cookie_token, extract_webapp_passport
@@ -19,6 +21,8 @@ from models.enums import EndUserType
 from models.model import AppMCPServer, EndUser
 from services.account_service import AccountService

+logger = logging.getLogger(__name__)
+
 type LoginUser = Account | EndUser


@@ -156,13 +160,24 @@ def _load_user_from_request(request_from_flask_login: Request, session: Session)
@user_logged_in.connect
@user_loaded_from_request.connect
 def on_user_logged_in(_sender: object, user: LoginUser) -> None:
-    """Called when a user logged in.
+    """Snapshot authenticated identity into the side-effect-free logging context.

    Note: AccountService.load_logged_in_account will populate user.current_tenant_id
    through the load_user method, which calls account.set_tenant_id_with_session().
    """
-    # tenant_id context variable removed - using current_user.current_tenant_id directly
-    pass
+    set_identity_context()
+    try:
+        match user:
+            case Account():
+                set_identity_context(tenant_id=user.current_tenant_id, user_id=user.id, user_type="account")
+            case EndUser():
+                set_identity_context(tenant_id=user.tenant_id, user_id=user.id, user_type=user.type or "end_user")
+            case _ as unreachable:
+                assert_never(unreachable)
+    except Exception:
+        # Logging enrichment must never make authentication fail.
+        logger.exception("Failed to set logging identity context")
+        return


@login_manager.unauthorized_handler
@@ -119,6 +119,19 @@ class Storage:
    def delete(self, filename: str):
        return self.storage_runner.delete(filename)

+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        return self.storage_runner.generate_presigned_url(
+            filename,
+            expires_in=expires_in,
+            content_type=content_type,
+        )
+
    def scan(self, path: str, files: bool = True, directories: bool = False) -> list[str]:
        return self.storage_runner.scan(path, files=files, directories=directories)

@@ -69,12 +69,17 @@ def on_user_loaded(_sender, user: Union["Account", "EndUser"]):
        if user:
            try:
                current_span = get_current_span()
+                if not current_span.is_recording():
+                    return
                tenant_id = extract_tenant_id(user)
                if not tenant_id:
                    return
-                if current_span:
-                    current_span.set_attribute(DifySpanAttributes.TENANT_ID, tenant_id)
-                    current_span.set_attribute(GenAIAttributes.USER_ID, user.id)
+                current_span.set_attributes(
+                    {
+                        DifySpanAttributes.TENANT_ID: tenant_id,
+                        GenAIAttributes.USER_ID: user.id,
+                    }
+                )
            except Exception:
                logger.exception("Error setting tenant and user attributes")
                pass
@@ -92,3 +92,21 @@ class AwsS3Storage(BaseStorage):
    @override
    def delete(self, filename: str):
        self.client.delete_object(Bucket=self.bucket_name, Key=filename)
+
+    @override
+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        params = {"Bucket": self.bucket_name, "Key": filename}
+        if content_type:
+            params["ResponseContentType"] = content_type
+
+        return self.client.generate_presigned_url(
+            "get_object",
+            Params=params,
+            ExpiresIn=expires_in,
+        )
@@ -31,6 +31,16 @@ class BaseStorage(ABC):
    def delete(self, filename: str):
        raise NotImplementedError

+    def generate_presigned_url(
+        self,
+        filename: str,
+        *,
+        expires_in: int,
+        content_type: str | None = None,
+    ) -> str:
+        """Generate a temporary direct-download URL when the backend supports it."""
+        raise NotImplementedError("This storage backend doesn't support presigned URLs")
+
    def scan(self, path, files=True, directories=False) -> list[str]:
        """
        Scan files and directories in the given path.
@@ -7,6 +7,7 @@ from uuid import uuid4
 from pydantic import Field, computed_field, field_validator

 from core.entities.execution_extra_content import ExecutionExtraContentDomainModel
+from core.tools.entities.ui_entities import MessageUIPart, parse_history_ui_parts
 from fields.base import ResponseModel
 from fields.conversation_fields import AgentThought, JSONValue, MessageFile
 from graphon.file import File
@@ -64,6 +65,7 @@ class MessageListItem(ResponseModel):
    status: str
    error: str | None = None
    extra_contents: list[ExecutionExtraContentDomainModel]
+    ui_parts: list[MessageUIPart] = Field(default_factory=list, validation_alias="message_metadata_dict")

    @computed_field
    def total_tokens(self) -> int:
@@ -79,6 +81,12 @@ class MessageListItem(ResponseModel):
    def _normalize_created_at(cls, value: datetime | int | None) -> int | None:
        return to_timestamp(value)

+    @field_validator("ui_parts", mode="before")
+    @classmethod
+    def _normalize_ui_parts(cls, value: object) -> list[MessageUIPart]:
+        raw_parts = value.get("ui_parts", []) if isinstance(value, dict) else value
+        return parse_history_ui_parts(raw_parts)
+

 class WebMessageListItem(MessageListItem):
    metadata: JSONValueType | None = Field(
@@ -1,5 +1,5 @@
 {
-  "commit": "4310e2d582d25e7de58183f27720afab01e123cf",
-  "openapiSha256": "5827ca930ce38462bfd1b2bef387efbf37eb7ffcaedde4558af2fbaeccbfbc4b",
+  "commit": "a0f50470612cc0b3656f89e4f2435aaf412e6e3b",
+  "openapiSha256": "f18910e9c45a64f0855e0643a7a626fb2889021b4f943458de86c6bd2469facb",
  "repository": "https://github.com/langgenius/knowledge-fs"
 }
--- a/Show More
+++ b/Show More