fix: knip

test(web): seed system features in component tests
test(api): define change email config in integration tests
2026-07-23 17:07:04 +08:00 · 2026-07-23 16:51:44 +08:00 · 2026-07-23 16:38:46 +08:00 · 2026-07-23 16:07:22 +08:00 · 2026-07-23 16:04:41 +08:00 · 2026-07-23 16:04:41 +08:00
2453 changed files with 73020 additions and 129915 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.

@@ -9,12 +9,11 @@ Use this skill for Vitest work under `web/` and `packages/dify-ui/`. Do not use

 ## Required Source

-Before writing, changing, or reviewing frontend tests, read `web/docs/test.md` completely. It is the single source of truth. This skill defines the execution workflow and must not add requirements that conflict with or duplicate that guide.
+Before writing, changing, or reviewing frontend tests, read `web/docs/test.md` completely. It is the single source of truth. This skill provides an execution checklist and must not redefine or extend that policy.

 ## Workflow

 1. Read the source, its behavior owner, nearby specs, and relevant public dependencies.
-1. Identify whether the contract belongs in `web/`, Dify UI Browser Mode, or a styled Storybook test.
 1. Apply the canonical guide to decide whether a test is needed and choose its boundary.
 1. For a behavior change or bug fix, write or identify the failing scenario first when practical.
 1. Implement one coherent scenario at a time and run the focused spec before expanding scope.
@@ -33,4 +32,4 @@ vp test run path/to/spec-or-directory
 vp test run --project unit src/path/to/spec
 ```

-For styled Dify UI behavior, run `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.
+Run Dify UI Storybook tests with `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.
@@ -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
@@ -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 \
@@ -27,7 +27,7 @@ jobs:
    steps:
      - id: skip_check
        continue-on-error: true
-        uses: fkirc/skip-duplicate-actions@f75f66ce1886f00957d99748a42c724f4330bdcf # v5.3.1
+        uses: fkirc/skip-duplicate-actions@b974a9395958c231af965b70070979a577efa578 # v5.3.2
        with:
          cancel_others: 'true'
          concurrent_skipping: same_content_newer
@@ -81,7 +81,6 @@ jobs:
              - '.npmrc'
              - '.nvmrc'
              - '.github/workflows/cli-tests.yml'
-              - '.github/workflows/cli-docker-build.yml'
              - '.github/actions/setup-web/**'
            web:
              - 'web/**'
@@ -336,6 +335,8 @@ jobs:
      - check-changes
    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
    uses: ./.github/workflows/web-e2e.yml
+    with:
+      run-external-runtime: false
    secrets: inherit

  web-e2e-skip:
@@ -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/**'
@@ -27,7 +27,7 @@ jobs:
          persist-credentials: false

      - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
        with:
          go-version-file: dify-agent-runtime/go.mod
          cache-dependency-path: dify-agent-runtime/go.sum
@@ -51,13 +51,13 @@ jobs:
          persist-credentials: false

      - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
        with:
          go-version-file: dify-agent-runtime/go.mod
          cache-dependency-path: dify-agent-runtime/go.sum

      - name: Run golangci-lint
-        uses: golangci/golangci-lint-action@4afd733a84b1f43292c63897423277bb7f4313a9 # v6.5.0
+        uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v6.5.0
        with:
          working-directory: dify-agent-runtime
          version: latest
@@ -78,7 +78,7 @@ jobs:
          persist-credentials: false

      - name: Setup Go
-        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
        with:
          go-version-file: dify-agent-runtime/go.mod
          cache-dependency-path: dify-agent-runtime/go.sum
@@ -29,7 +29,7 @@ jobs:
          persist-credentials: false

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

      - name: Run Claude Code for Translation Sync
        if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@e90deca47693f9457b72f2b53c17d7c445a87342 # v1.0.171
+        uses: anthropics/claude-code-action@af0559ee4f514d1ef21826982bed13f7edc3c35e # v1.0.178
        with:
          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
          github_token: ${{ secrets.GITHUB_TOKEN }}
@@ -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
@@ -677,6 +677,17 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y

 # Dify Agent backend
 AGENT_BACKEND_BASE_URL=http://localhost:5050
+AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS=30
+AGENT_BACKEND_STREAM_MAX_RECONNECTS=3
+AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
+
+# KnowledgeFS (Dataset 2.0)
+KNOWLEDGE_FS_ENABLED=false
+KNOWLEDGE_FS_BASE_URL=
+# Shared with KnowledgeFS; use at least 32 random characters.
+KNOWLEDGE_FS_JWT_SECRET=
+KNOWLEDGE_FS_SSE_READ_TIMEOUT_SECONDS=300
+KNOWLEDGE_FS_TIMEOUT_SECONDS=10

 # Marketplace configuration
 MARKETPLACE_ENABLED=true
@@ -8,7 +8,7 @@ creating another wire contract.

 from __future__ import annotations

-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from typing import Protocol

 from dify_agent.client import (
@@ -45,7 +45,13 @@ class AgentBackendRunClient(Protocol):
    def cancel_run(self, run_id: str, request: CancelRunRequest | None = None) -> CancelRunResponse:
        """Request explicit cancellation for one Agent backend run."""

-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
        """Yield public ``dify-agent`` run events in stream order."""

    def wait_run(self, run_id: str, *, timeout_seconds: float | None = None) -> RunStatusResponse:
@@ -61,7 +67,15 @@ class _DifyAgentSyncClient(Protocol):
    def cancel_run_sync(self, run_id: str, request: CancelRunRequest | None = None) -> CancelRunResponse:
        """Cancel one run synchronously."""

-    def stream_events_sync(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events_sync(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        max_reconnects: int | None = None,
+        timeout_seconds: float | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
        """Stream run events synchronously."""

    def wait_run_sync(self, run_id: str, *, timeout_seconds: float | None = None) -> RunStatusResponse:
@@ -73,8 +87,16 @@ class DifyAgentBackendRunClient:

    client: _DifyAgentSyncClient

-    def __init__(self, client: _DifyAgentSyncClient) -> None:
+    def __init__(
+        self,
+        client: _DifyAgentSyncClient,
+        *,
+        stream_max_reconnects: int = 3,
+        stream_timeout_seconds: float = 1200,
+    ) -> None:
        self.client = client
+        self._stream_max_reconnects = stream_max_reconnects
+        self._stream_timeout_seconds = stream_timeout_seconds

    def create_run(self, request: CreateRunRequest) -> CreateRunResponse:
        """Create one run through ``POST /runs`` and normalize client exceptions."""
@@ -90,10 +112,22 @@ class DifyAgentBackendRunClient:
        except Exception as exc:
            raise _normalize_dify_agent_error(exc) from exc

-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
        """Stream run events from ``/events/sse`` with the wrapped client's reconnect policy."""
        try:
-            yield from self.client.stream_events_sync(run_id, after=after)
+            yield from self.client.stream_events_sync(
+                run_id,
+                after=after,
+                max_reconnects=self._stream_max_reconnects,
+                timeout_seconds=self._stream_timeout_seconds,
+                should_stop=should_stop,
+            )
        except Exception as exc:
            raise _normalize_dify_agent_error(exc) from exc

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

 from __future__ import annotations

-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from datetime import UTC, datetime
 from enum import StrEnum

@@ -69,9 +69,17 @@ class FakeAgentBackendRunClient:
        del request
        return CancelRunResponse(run_id=run_id, status="cancelled")

-    def stream_events(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
+    def stream_events(
+        self,
+        run_id: str,
+        *,
+        after: str | None = None,
+        should_stop: Callable[[], bool] | None = None,
+    ) -> Iterator[RunEvent]:
        """Yield the deterministic public ``RunEvent`` sequence for ``run_id``."""
        for event in self._events(run_id):
+            if should_stop is not None and should_stop():
+                return
            if after is not None and event.id is not None and event.id <= after:
                continue
            yield event
@@ -1,6 +1,8 @@
 import datetime
 import logging
+import re
 import time
+import uuid
 from collections.abc import Callable
 from typing import TypedDict

@@ -21,6 +23,7 @@ from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 logger = logging.getLogger(__name__)

 _HEX_PREFIXES = tuple("0123456789abcdef")
+_TARGET_MONTH_PATTERN = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")


 class WorkflowRunArchivePlanRow(TypedDict):
@@ -66,6 +69,7 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:


 def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[str] | None:
+    """Keep an omitted scope unset while rejecting an explicitly empty scope."""
    if raw_ids is None:
        return None
    parsed = sorted({raw_id.strip() for raw_id in raw_ids.split(",") if raw_id.strip()})
@@ -74,6 +78,27 @@ def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[
    return parsed


+def _parse_archive_target_month(target_month: str) -> tuple[int, int]:
+    """Validate the V2 catalog month selector and return its numeric components."""
+    if not _TARGET_MONTH_PATTERN.fullmatch(target_month):
+        raise click.BadParameter("target-month must use YYYY-MM format", param_hint="--target-month")
+    year_text, month_text = target_month.split("-", maxsplit=1)
+    return int(year_text), int(month_text)
+
+
+def _parse_archive_catalog_cursor(after_catalog_id: str | None) -> str | None:
+    """Normalize the exclusive V2 catalog keyset cursor when one is provided."""
+    if after_catalog_id is None:
+        return None
+    try:
+        return str(uuid.UUID(after_catalog_id))
+    except ValueError as exc:
+        raise click.BadParameter(
+            "after-catalog-id must be a UUID returned by the same V2 operation and scope",
+            param_hint="--after-catalog-id",
+        ) from exc
+
+
 def _get_archive_candidate_tenant_ids_by_prefix(
    session: Session,
    prefix: str,
@@ -810,9 +835,11 @@ def backfill_workflow_run_archive_bundles(
        click.echo(click.style(f"  ... and {len(summary.errors) - 10} more failures", fg="red"))


-def _echo_bundle_archive_operation_summary(summary) -> None:
+def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> None:
    status = "completed successfully" if summary.bundles_failed == 0 else "completed with failures"
    fg = "green" if summary.bundles_failed == 0 else "red"
+    cursor_label = "preview_next_catalog_id" if dry_run else "next_catalog_id"
+    cursor_value = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
    click.echo(
        click.style(
            f"{summary.operation} {status}. "
@@ -821,10 +848,12 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
            f"archive_bytes={summary.archive_bytes} duration={summary.elapsed_time:.2f}s "
            f"validation_time={summary.validation_time:.2f}s "
            f"runs_per_second={summary.runs_per_second:.2f} rows_per_second={summary.rows_per_second:.2f} "
-            f"bytes_per_second={summary.bytes_per_second:.2f}",
+            f"bytes_per_second={summary.bytes_per_second:.2f} {cursor_label}={cursor_value or 'none'}",
            fg=fg,
        )
    )
+    if dry_run:
+        click.echo(click.style("Dry-run cursor is preview-only; do not persist it for a destructive run.", fg="yellow"))
    click.echo(click.style("table,row_count", fg="white"))
    for table_name in [
        "workflow_runs",
@@ -842,7 +871,8 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
                click.style(
                    f"  bundle={result.bundle_id} tenant={result.tenant_id} runs={result.run_count} "
                    f"rows={result.row_count} archive_bytes={result.archive_bytes} "
-                    f"time={result.elapsed_time:.2f}s validation={result.validation_time:.2f}s",
+                    f"catalog_id={result.catalog_id} time={result.elapsed_time:.2f}s "
+                    f"validation={result.validation_time:.2f}s",
                    fg="white",
                )
            )
@@ -850,7 +880,7 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
            click.echo(
                click.style(
                    f"  failed bundle={result.bundle_id} tenant={result.tenant_id} "
-                    f"object_prefix={result.object_prefix} error={result.error}",
+                    f"catalog_id={result.catalog_id} object_prefix={result.object_prefix} error={result.error}",
                    fg="red",
                )
            )
@@ -867,25 +897,24 @@ def _echo_bundle_archive_operation_summary(summary) -> None:
 )
@click.option("--run-id", required=False, help="Workflow run ID to restore.")
@click.option(
-    "--start-from",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--target-month",
+    metavar="YYYY-MM",
    default=None,
-    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
+    help="V2 catalog month to restore; required unless --run-id is used.",
 )
@click.option(
-    "--end-before",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--after-catalog-id",
    default=None,
-    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
+    help="Exclusive V2 cursor from the same restore month and tenant scope.",
 )
@click.option("--workers", default=1, show_default=True, type=int, help="V1 --run-id compatibility only.")
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to restore.")
+@click.option("--limit", type=click.IntRange(min=1), default=100, show_default=True, help="Maximum V2 catalog rows.")
@click.option("--dry-run", is_flag=True, help="Preview without restoring.")
 def restore_workflow_runs(
    tenant_ids: str | None,
    run_id: str | None,
-    start_from: datetime.datetime | None,
-    end_before: datetime.datetime | None,
+    target_month: str | None,
+    after_catalog_id: str | None,
    workers: int,
    limit: int,
    dry_run: bool,
@@ -905,23 +934,20 @@ def restore_workflow_runs(
    from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
    from services.retention.workflow_run.restore_archived_workflow_run import WorkflowRunRestore

-    parsed_tenant_ids = None
-    if tenant_ids:
-        parsed_tenant_ids = [tid.strip() for tid in tenant_ids.split(",") if tid.strip()]
-        if not parsed_tenant_ids:
-            raise click.BadParameter("tenant-ids must not be empty")
+    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")

-    if (start_from is None) ^ (end_before is None):
-        raise click.UsageError("--start-from and --end-before must be provided together.")
-    if run_id is None and (start_from is None or end_before is None):
-        raise click.UsageError("--start-from and --end-before are required for batch restore.")
    if workers < 1:
        raise click.BadParameter("workers must be at least 1")
+    if run_id is not None and (target_month is not None or after_catalog_id is not None):
+        raise click.UsageError("--target-month and --after-catalog-id are only valid for V2 batch restore.")
+    if run_id is None and target_month is None:
+        raise click.UsageError("--target-month is required for V2 batch restore.")

    start_time = datetime.datetime.now(datetime.UTC)
+    target_desc = f"workflow run {run_id}" if run_id else f"workflow archive catalog month {target_month}"
    click.echo(
        click.style(
-            f"Starting restore of workflow run {run_id} at {start_time.isoformat()}.",
+            f"Starting restore of {target_desc} at {start_time.isoformat()}.",
            fg="white",
        )
    )
@@ -955,17 +981,20 @@ def restore_workflow_runs(
        click.echo(
            click.style("--workers is ignored for V2 bundle restore; bundles are processed serially.", fg="yellow")
        )
-    assert start_from is not None
-    assert end_before is not None
+    assert target_month is not None
+    target_year, target_month_number = _parse_archive_target_month(target_month)
+    catalog_cursor = _parse_archive_catalog_cursor(after_catalog_id)
    bundle_restorer = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
    summary = bundle_restorer.restore_batch(
        tenant_ids=parsed_tenant_ids,
-        start_date=start_from,
-        end_date=end_before,
+        target_year=target_year,
+        target_month=target_month_number,
+        after_catalog_id=catalog_cursor,
        limit=limit,
    )
-    _echo_bundle_archive_operation_summary(summary)
-    return
+    _echo_bundle_archive_operation_summary(summary, dry_run=dry_run)
+    if summary.bundles_failed:
+        raise click.exceptions.Exit(1)


@click.command(
@@ -979,23 +1008,41 @@ def restore_workflow_runs(
 )
@click.option("--run-id", required=False, help="Workflow run ID to delete.")
@click.option(
-    "--start-from",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--target-month",
+    metavar="YYYY-MM",
    default=None,
-    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
+    help="V2 catalog month to delete; required unless --run-id is used.",
 )
@click.option(
-    "--end-before",
-    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    "--after-catalog-id",
    default=None,
-    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
+    help="Exclusive V2 cursor from the same delete month and tenant scope.",
+)
+@click.option(
+    "--run-shard-index",
+    default=None,
+    type=click.IntRange(min=0),
+    help="Zero-based archive shard index. Must be paired with --run-shard-total.",
+)
+@click.option(
+    "--run-shard-total",
+    default=None,
+    type=click.IntRange(min=1, max=16),
+    help="Total archive shard count. Must be paired with --run-shard-index.",
+)
+@click.option("--all-pages", is_flag=True, help="Process catalog pages until an empty page is reached.")
+@click.option(
+    "--limit",
+    type=click.IntRange(min=1),
+    default=100,
+    show_default=True,
+    help="Maximum V2 catalog rows per page.",
 )
-@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to delete.")
@click.option("--dry-run", is_flag=True, help="Preview without deleting.")
@click.option(
    "--skip-bad-archives",
    is_flag=True,
-    help="Continue batch deletion when one archive object fails validation.",
+    help="V1 --run-id only: continue when one archive object fails validation.",
 )
@click.option(
    "--restore-sample-interval",
@@ -1007,8 +1054,11 @@ def restore_workflow_runs(
 def delete_archived_workflow_runs(
    tenant_ids: str | None,
    run_id: str | None,
-    start_from: datetime.datetime | None,
-    end_before: datetime.datetime | None,
+    target_month: str | None,
+    after_catalog_id: str | None,
+    run_shard_index: int | None,
+    run_shard_total: int | None,
+    all_pages: bool,
    limit: int,
    dry_run: bool,
    skip_bad_archives: bool,
@@ -1018,26 +1068,38 @@ def delete_archived_workflow_runs(
    Delete archived workflow runs from the database.

    Batch delete uses V2 bundle metadata and validates object existence, manifest schema, object size, checksum, row
-    counts, and source/archive content checksums before deleting source rows. `--run-id` keeps the V1 per-run path.
+    counts, and source/archive content checksums before deleting source rows. Parallel workers may select one exact
+    archive shard; all-pages mode keeps only the current bounded page in memory. `--run-id` keeps the V1 per-run path.
    """
    from services.retention.workflow_run.bundle_archive_maintenance import WorkflowRunBundleArchiveMaintenance
    from services.retention.workflow_run.delete_archived_workflow_run import ArchivedWorkflowRunDeletion

-    parsed_tenant_ids = None
-    if tenant_ids:
-        parsed_tenant_ids = [tid.strip() for tid in tenant_ids.split(",") if tid.strip()]
-        if not parsed_tenant_ids:
-            raise click.BadParameter("tenant-ids must not be empty")
+    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")

-    if (start_from is None) ^ (end_before is None):
-        raise click.UsageError("--start-from and --end-before must be provided together.")
-    if run_id is None and (start_from is None or end_before is None):
-        raise click.UsageError("--start-from and --end-before are required for batch delete.")
    if restore_sample_interval < 0:
        raise click.BadParameter("restore-sample-interval must be >= 0")
+    if run_id is not None and (
+        target_month is not None
+        or after_catalog_id is not None
+        or run_shard_index is not None
+        or run_shard_total is not None
+        or all_pages
+    ):
+        raise click.UsageError(
+            "--target-month, --after-catalog-id, --run-shard-index, --run-shard-total, and --all-pages "
+            "are only valid for V2 batch delete."
+        )
+    if run_id is None and target_month is None:
+        raise click.UsageError("--target-month is required for V2 batch delete.")
+    if run_id is None and skip_bad_archives:
+        raise click.UsageError("--skip-bad-archives is not supported for V2 catalog batches; they fail fast.")
+    if (run_shard_index is None) ^ (run_shard_total is None):
+        raise click.UsageError("--run-shard-index and --run-shard-total must be provided together.")
+    if run_shard_index is not None and run_shard_total is not None and run_shard_index >= run_shard_total:
+        raise click.UsageError("--run-shard-index must be less than --run-shard-total.")

    start_time = datetime.datetime.now(datetime.UTC)
-    target_desc = f"workflow run {run_id}" if run_id else "workflow runs"
+    target_desc = f"workflow run {run_id}" if run_id else f"workflow archive catalog month {target_month}"
    click.echo(
        click.style(
            f"Starting delete of {target_desc} at {start_time.isoformat()}.",
@@ -1110,20 +1172,104 @@ def delete_archived_workflow_runs(

    if restore_sample_interval:
        click.echo(click.style("--restore-sample-interval is ignored for V2 bundle delete.", fg="yellow"))
-    assert start_from is not None
-    assert end_before is not None
-    bundle_deleter = WorkflowRunBundleArchiveMaintenance(
-        dry_run=dry_run,
-        strict_content_validation=True,
-        stop_on_error=not skip_bad_archives,
+    assert target_month is not None
+    target_year, target_month_number = _parse_archive_target_month(target_month)
+    catalog_cursor = _parse_archive_catalog_cursor(after_catalog_id)
+    shard = (
+        f"{run_shard_index:02d}-of-{run_shard_total:02d}"
+        if run_shard_index is not None and run_shard_total is not None
+        else None
    )
-    summary = bundle_deleter.delete_batch(
-        tenant_ids=parsed_tenant_ids,
-        start_date=start_from,
-        end_date=end_before,
-        limit=limit,
-    )
-    _echo_bundle_archive_operation_summary(summary)
+    bundle_deleter = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
+    if run_shard_total is not None:
+        try:
+            bundle_deleter.validate_catalog_shards(
+                target_year=target_year,
+                target_month=target_month_number,
+                shard_total=run_shard_total,
+                tenant_ids=parsed_tenant_ids,
+            )
+        except ValueError as exc:
+            logger.exception(
+                "Archive catalog shard preflight failed: target_month=%s shard=%s",
+                target_month,
+                shard,
+            )
+            raise click.ClickException(
+                f"Archive catalog shard preflight failed for target_month={target_month} shard={shard}: {exc}"
+            ) from exc
+
+    initial_catalog_cursor = catalog_cursor
+    pages_processed = 0
+    bundles_succeeded = 0
+    runs_processed = 0
+    rows_processed = 0
+    archive_bytes = 0
+    while True:
+        summary = bundle_deleter.delete_batch(
+            tenant_ids=parsed_tenant_ids,
+            target_year=target_year,
+            target_month=target_month_number,
+            after_catalog_id=catalog_cursor,
+            limit=limit,
+            shard=shard,
+        )
+        _echo_bundle_archive_operation_summary(summary, dry_run=dry_run)
+        if summary.bundles_failed:
+            failed_result = next((result for result in summary.results if not result.success), None)
+            failed_catalog_id = failed_result.catalog_id if failed_result is not None else "unknown"
+            page_resume_cursor = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
+            resume_cursor = page_resume_cursor or catalog_cursor
+            if dry_run:
+                cursor_details = (
+                    f"preview_after_catalog_id={resume_cursor or 'none'} "
+                    f"destructive_retry_after_catalog_id={initial_catalog_cursor or 'none'}"
+                )
+            else:
+                cursor_details = f"resume_after_catalog_id={resume_cursor or 'none'}"
+            click.echo(
+                click.style(
+                    f"Delete stopped: target_month={target_month} shard={shard or 'all'} "
+                    f"failed_catalog_id={failed_catalog_id} "
+                    f"{cursor_details}",
+                    fg="red",
+                )
+            )
+            raise click.exceptions.Exit(1)
+
+        if not all_pages:
+            break
+        if summary.bundles_processed == 0:
+            break
+
+        pages_processed += 1
+        bundles_succeeded += summary.bundles_succeeded
+        runs_processed += summary.runs_processed
+        rows_processed += summary.rows_processed
+        archive_bytes += summary.archive_bytes
+        next_catalog_id = summary.preview_next_catalog_id if dry_run else summary.next_catalog_id
+        if next_catalog_id is None or (catalog_cursor is not None and next_catalog_id <= catalog_cursor):
+            click.echo(
+                click.style(
+                    f"Delete cursor did not advance: target_month={target_month} shard={shard or 'all'} "
+                    f"after_catalog_id={catalog_cursor or 'none'} next_catalog_id={next_catalog_id or 'none'}",
+                    fg="red",
+                )
+            )
+            raise click.exceptions.Exit(1)
+        catalog_cursor = next_catalog_id
+
+    if all_pages:
+        final_cursor_label = "preview_final_catalog_id" if dry_run else "final_catalog_id"
+        click.echo(
+            click.style(
+                f"Delete all-pages completed successfully. target_month={target_month} shard={shard or 'all'} "
+                f"pages={pages_processed} bundles_success={bundles_succeeded} runs={runs_processed} "
+                f"rows={rows_processed} archive_bytes={archive_bytes} "
+                f"{final_cursor_label}={catalog_cursor or 'none'}",
+                fg="green",
+            )
+        )


 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
@@ -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,5 +1,6 @@
 from configs.extra.agent_backend_config import AgentBackendConfig
 from configs.extra.archive_config import ArchiveStorageConfig
+from configs.extra.knowledge_fs_config import KnowledgeFSConfig
 from configs.extra.notion_config import NotionConfig
 from configs.extra.sentry_config import SentryConfig

@@ -8,6 +9,7 @@ class ExtraServiceConfig(
    # place the configs in alphabet order
    AgentBackendConfig,
    ArchiveStorageConfig,
+    KnowledgeFSConfig,
    NotionConfig,
    SentryConfig,
 ):
@@ -1,4 +1,4 @@
-from pydantic import Field, NonNegativeFloat
+from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat
 from pydantic_settings import BaseSettings


@@ -22,6 +22,21 @@ class AgentBackendConfig(BaseSettings):
        default="success",
    )

+    AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS: PositiveFloat = Field(
+        description="Read timeout for one Agent backend SSE connection.",
+        default=30,
+    )
+
+    AGENT_BACKEND_STREAM_MAX_RECONNECTS: NonNegativeInt = Field(
+        description="Maximum Agent backend SSE reconnects before failing the run.",
+        default=3,
+    )
+
+    AGENT_BACKEND_RUN_TIMEOUT_SECONDS: PositiveFloat = Field(
+        description="Total deadline for one Agent backend run event stream.",
+        default=1200,
+    )
+
    AGENT_SHELL_ENABLED: bool = Field(
        description=(
            "Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "
@@ -0,0 +1,64 @@
+"""Configuration for the optional KnowledgeFS Console bridge."""
+
+from urllib.parse import urlsplit
+
+from pydantic import Field, PositiveFloat, SecretStr, field_validator, model_validator
+from pydantic_settings import BaseSettings
+
+
+class KnowledgeFSConfig(BaseSettings):
+    """Server-only settings for the KnowledgeFS production connection."""
+
+    KNOWLEDGE_FS_ENABLED: bool = Field(
+        default=False,
+        description="Enable the private KnowledgeFS Console bridge.",
+    )
+    KNOWLEDGE_FS_BASE_URL: str | None = Field(default=None, description="KnowledgeFS gateway base URL.")
+    KNOWLEDGE_FS_JWT_SECRET: SecretStr | None = Field(
+        default=None,
+        min_length=32,
+        description="Shared secret used to sign short-lived KnowledgeFS service JWTs.",
+    )
+    KNOWLEDGE_FS_SSE_READ_TIMEOUT_SECONDS: PositiveFloat = Field(default=300.0, le=3600.0, allow_inf_nan=False)
+    KNOWLEDGE_FS_TIMEOUT_SECONDS: PositiveFloat = Field(default=10.0, le=60.0, allow_inf_nan=False)
+
+    @field_validator(
+        "KNOWLEDGE_FS_BASE_URL",
+        "KNOWLEDGE_FS_JWT_SECRET",
+        mode="before",
+    )
+    @classmethod
+    def normalize_optional_string(cls, value: object) -> object:
+        if isinstance(value, SecretStr):
+            normalized = value.get_secret_value().strip()
+            return SecretStr(normalized) if normalized else None
+        if isinstance(value, str):
+            normalized = value.strip()
+            return normalized or None
+        return value
+
+    @field_validator("KNOWLEDGE_FS_BASE_URL")
+    @classmethod
+    def validate_base_url(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        parsed = urlsplit(value)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must be an absolute HTTP(S) URL")
+        try:
+            _ = parsed.port
+        except ValueError as exc:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must include a valid port") from exc
+        if parsed.username or parsed.password or parsed.query or parsed.fragment:
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must not include credentials, query, or fragment")
+        return value.rstrip("/")
+
+    @model_validator(mode="after")
+    def validate_enabled_connection(self) -> "KnowledgeFSConfig":
+        if not self.KNOWLEDGE_FS_ENABLED:
+            return self
+        if bool(self.KNOWLEDGE_FS_BASE_URL) != bool(self.KNOWLEDGE_FS_JWT_SECRET):
+            raise ValueError("KNOWLEDGE_FS_BASE_URL and KNOWLEDGE_FS_JWT_SECRET must be configured together")
+        if not self.KNOWLEDGE_FS_BASE_URL:
+            raise ValueError("KnowledgeFS connection settings are required when the integration is enabled")
+        return self
@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 from enum import StrEnum
 from typing import Literal

@@ -11,6 +11,7 @@ from pydantic import (
    PositiveFloat,
    PositiveInt,
    computed_field,
+    field_validator,
 )
 from pydantic_settings import BaseSettings

@@ -280,6 +281,27 @@ class PluginConfig(BaseSettings):
        default="",
    )

+    @field_validator("PLUGIN_REMOTE_INSTALL_PORT", mode="before")
+    @classmethod
+    def _reject_host_port_shaped_plugin_remote_install_port(cls, v):
+        """Reject ``host:port``-shaped values with an actionable hint.
+
+        ``EXPOSE_PLUGIN_DEBUGGING_PORT`` is overloaded: it feeds both the
+        plugin_daemon ``ports:`` mapping (where ``127.0.0.1:5003`` is valid
+        compose syntax) and this integer app setting advertised in the console.
+        Without this guard a loopback bind spec crashloops the api container
+        with an opaque ``int_parsing`` traceback. See issue #39323.
+        """
+        if isinstance(v, str) and ":" in v.strip():
+            raise ValueError(
+                "PLUGIN_REMOTE_INSTALL_PORT must be a bare port number, got "
+                f"{v!r}. A 'host:port' value usually means "
+                "EXPOSE_PLUGIN_DEBUGGING_PORT was set to a compose publish spec "
+                "like '127.0.0.1:5003'; bind loopback via a "
+                "docker-compose.override.yaml instead of overloading this var."
+            )
+        return v
+
    @property
    def NEW_USER_DEFAULT_PLUGIN_ID_LIST(self) -> list[str]:
        return [item.strip() for item in self.NEW_USER_DEFAULT_PLUGIN_IDS.split(",") if item.strip()]
@@ -1138,6 +1160,16 @@ class HomepageConfig(BaseSettings):
        default=True,
    )

+    ENABLE_STEP_BY_STEP_TOUR: bool = Field(
+        description="Enable account-level Step-by-step Tour eligibility checks",
+        default=False,
+    )
+
+    STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT: datetime | None = Field(
+        description="UTC timestamp after which newly initialized accounts are eligible for Step-by-step Tour",
+        default=None,
+    )
+

 class RagEtlConfig(BaseSettings):
    """
@@ -1465,6 +1497,11 @@ class LoginConfig(BaseSettings):


 class AccountConfig(BaseSettings):
+    ENABLE_CHANGE_EMAIL: bool = Field(
+        description="whether users can change their email address",
+        default=True,
+    )
+
    ACCOUNT_DELETION_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
        description="Duration in minutes for which a account deletion token remains valid",
        default=5,
@@ -10,6 +10,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from models.model import App
+from services.agent.roster_service import AgentRosterService
 from services.enterprise.rbac_service import RBACService

 __all__ = ["RBACPermission", "RBACResourceScope", "enforce_rbac_access", "rbac_permission_required"]
@@ -51,7 +52,7 @@ def enforce_rbac_access(
    check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
    resource_id = None
    if resource_required and check_resource_type:
-        resource_id = _extract_resource_id(resource_type, path_args)
+        resource_id = _extract_resource_id(resource_type, tenant_id, path_args)
        if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
            return
    allowed = RBACService.CheckAccess.check(
@@ -131,11 +132,14 @@ def _is_resource_owned_by_current_user(
    return False


-def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str, object] | None = None) -> str:
+def _extract_resource_id(
+    resource_type: RBACResourceScope, tenant_id: str, path_args: dict[str, object] | None = None
+) -> str:
    """Extract the resource ID from matched path arguments.

    Some legacy route classes use neutral names such as ``resource_id`` for
-    app/dataset resources, and Agent App routes use ``agent_id`` as the app id.
+    app/dataset resources, and Agent routes carry ``agent_id``, which is
+    resolved to the App backing that Agent.
    Dataset endpoints behind a rag-pipeline route contain ``pipeline_id``
    instead of ``dataset_id``. In that case we look up the associated
    ``Dataset`` row via ``Dataset.pipeline_id``.
@@ -146,10 +150,19 @@ def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str,
    matched_args = {**view_args, **(path_args or {})}

    if resource_type == RBACResourceScope.APP:
-        app_id = matched_args.get("app_id") or matched_args.get("agent_id") or matched_args.get("resource_id")
-        if not app_id:
-            raise ValueError("Missing app_id in request path")
-        return str(app_id)
+        app_id = matched_args.get("app_id")
+        if app_id:
+            return str(app_id)
+
+        agent_id = matched_args.get("agent_id")
+        if agent_id:
+            authz_app_id = AgentRosterService(db.session).peek_authz_app_id(tenant_id=tenant_id, agent_id=str(agent_id))
+            return authz_app_id or str(agent_id)
+
+        resource_id = matched_args.get("resource_id")
+        if resource_id:
+            return str(resource_id)
+        raise ValueError("Missing app_id in request path")

    if resource_type == RBACResourceScope.DATASET:
        dataset_id = matched_args.get("dataset_id") or matched_args.get("resource_id")
@@ -38,7 +38,9 @@ from . import (
    feature,
    human_input_form,
    init_validate,
+    knowledge_fs_proxy,
    notification,
+    onboarding,
    ping,
    setup,
    spec,
@@ -195,6 +197,7 @@ __all__ = [
    "human_input_form",
    "init_validate",
    "installed_app",
+    "knowledge_fs_proxy",
    "load_balancing_config",
    "login",
    "mcp_server",
@@ -207,6 +210,7 @@ __all__ = [
    "notification",
    "oauth",
    "oauth_server",
+    "onboarding",
    "ops_trace",
    "parameter",
    "ping",
@@ -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,
@@ -542,6 +552,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 +590,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 +632,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 +658,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 +669,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 +693,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 +716,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 +739,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 +757,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 +816,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 +839,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 +865,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 +882,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 +901,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 +912,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 +932,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 +978,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 +1017,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 +1056,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 +1077,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 +1103,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 +1119,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 +1140,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,14 +344,23 @@ 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,
 ) -> str:
+    """Resolve the current editor's conversation without crossing draft surfaces."""
+
    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,
+            draft_type=draft_type,
        )

    agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
@@ -360,6 +370,7 @@ def _resolve_current_user_agent_debug_conversation_id(
        tenant_id=current_tenant_id,
        agent_id=agent.id,
        account_id=current_user.id,
+        draft_type=draft_type,
    )


@@ -382,6 +393,7 @@ def _create_chat_message(
            current_user=current_user,
            app_model=app_model,
            agent_id=agent_id,
+            draft_type=AgentConfigDraftType(args_model.draft_type),
        )
        if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
            raise NotFound("Conversation Not Exists.")
@@ -418,6 +430,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
@@ -105,6 +105,7 @@ class SyncDraftWorkflowPayload(BaseModel):
    graph: dict[str, Any]
    features: dict[str, Any]
    hash: str | None = None
+    is_collaborative: bool = Field(default=False, alias="_is_collaborative")
    environment_variables: list[dict[str, Any]] = Field(
        default_factory=list,
    )
@@ -358,6 +359,12 @@ class WorkflowPublishResponse(ResponseModel):
    created_at: int


+class SyncDraftWorkflowResponse(ResponseModel):
+    result: str
+    hash: str
+    updated_at: int
+
+
 class WorkflowRestoreResponse(ResponseModel):
    result: str
    hash: str
@@ -440,6 +447,7 @@ register_response_schema_models(
    WorkflowOnlineUsersByApp,
    WorkflowOnlineUsersResponse,
    WorkflowPublishResponse,
+    SyncDraftWorkflowResponse,
    WorkflowRestoreResponse,
    DefaultBlockConfigsResponse,
    DefaultBlockConfigResponse,
@@ -555,14 +563,7 @@ class DraftWorkflowApi(Resource):
    @console_ns.response(
        200,
        "Draft workflow synced successfully",
-        console_ns.model(
-            "SyncDraftWorkflowResponse",
-            {
-                "result": fields.String,
-                "hash": fields.String,
-                "updated_at": fields.String,
-            },
-        ),
+        console_ns.models[SyncDraftWorkflowResponse.__name__],
    )
    @console_ns.response(400, "Invalid workflow configuration")
    @console_ns.response(403, "Permission denied")
@@ -610,17 +611,21 @@ class DraftWorkflowApi(Resource):
                environment_variables=environment_variables,
                conversation_variables=conversation_variables,
                session=db.session(),
+                graph_only=args["is_collaborative"],
            )
        except WorkflowHashNotEqualError:
            raise DraftWorkflowNotSync()
        except VariableError as e:
            raise InvalidArgumentError(description=str(e))

-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
+        return dump_response(
+            SyncDraftWorkflowResponse,
+            {
+                "result": "success",
+                "hash": workflow.unique_hash,
+                "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+            },
+        )


@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")
@@ -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:
@@ -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:
@@ -47,7 +47,9 @@ from controllers.console.explore.error import (
    NotWorkflowAppError,
 )
 from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.console.wraps import with_current_user
+from controllers.console.files import FILE_UPLOAD_PARAMS, upload_file_from_request
+from controllers.console.remote_files import RemoteFileUploadPayload, upload_remote_file_from_request
+from controllers.console.wraps import cloud_edition_billing_resource_check, with_current_user
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -61,12 +63,13 @@ from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.base import ResponseModel
 from fields.conversation_variable_fields import WorkflowConversationVariableResponse
+from fields.file_fields import FileResponse, FileWithSignedUrl
 from fields.message_fields import SuggestedQuestionsResponse
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import dump_response, to_timestamp, uuid_value
-from models import Account
+from models import Account, App
 from models.account import TenantStatus
 from models.model import AppMode, Site, load_annotation_reply_config
 from models.workflow import Workflow
@@ -428,6 +431,36 @@ register_response_schema_models(
 simple_account_model = console_ns.models[TrialSimpleAccount.__name__]


+class TrialAppFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.doc(consumes=["multipart/form-data"], params=FILE_UPLOAD_PARAMS)
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a file into the tenant that owns the trial app."""
+        upload_file = upload_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return dump_response(FileResponse, upload_file), 201
+
+
+class TrialAppRemoteFileUploadApi(TrialAppResource):
+    @trial_feature_enable
+    @cloud_edition_billing_resource_check("documents")
+    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
+    @console_ns.response(201, "File uploaded successfully", console_ns.models[FileWithSignedUrl.__name__])
+    @with_current_user
+    def post(self, current_user: Account, app_model: App):
+        """Upload a remote file into the tenant that owns the trial app."""
+        remote_file = upload_remote_file_from_request(
+            current_user=current_user,
+            resource_tenant_id=app_model.tenant_id,
+        )
+        return remote_file.model_dump(mode="json"), 201
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
    @trial_feature_enable
    @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
@@ -612,7 +645,7 @@ class TrialChatAudioApi(TrialAppResource):
    def post(self, current_user: Account, trial_app):
        app_model = trial_app

-        file = request.files["file"]
+        file = request.files.get("file")

        try:
            # Get IDs before they might be detached from session
@@ -887,6 +920,18 @@ class DatasetListApi(Resource):

 console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")

+console_ns.add_resource(
+    TrialAppFileUploadApi,
+    "/trial-apps/<uuid:app_id>/files/upload",
+    endpoint="trial_app_file_upload",
+)
+
+console_ns.add_resource(
+    TrialAppRemoteFileUploadApi,
+    "/trial-apps/<uuid:app_id>/remote-files/upload",
+    endpoint="trial_app_remote_file_upload",
+)
+
 console_ns.add_resource(
    TrialMessageSuggestedQuestionApi,
    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
@@ -29,7 +29,7 @@ from extensions.ext_database import db
 from fields.file_fields import FileResponse, UploadConfig
 from libs.helper import dump_response
 from libs.login import login_required
-from models import Account
+from models import Account, UploadFile
 from services.file_service import FileService

 from . import console_ns
@@ -39,7 +39,7 @@ register_response_schema_models(console_ns, AllowedExtensionsResponse, TextConte

 PREVIEW_WORDS_LIMIT = 3000

-_FILE_UPLOAD_PARAMS = {
+FILE_UPLOAD_PARAMS = {
    "file": {
        "description": "File to upload",
        "in": "formData",
@@ -56,6 +56,43 @@ _FILE_UPLOAD_PARAMS = {
 }


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

        return dump_response(FileResponse, upload_file), 201

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


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

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

    @staticmethod
    def _resolve_agent_by_id(
@@ -941,48 +941,64 @@ class AgentAppRunner:
            if pending_text:
                persist_answer_text(pending_text)

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

    def _cancel_run(self, run_id: str) -> None:
@@ -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,
@@ -21,6 +21,7 @@ from core.app.entities.queue_entities import (
    WorkflowQueueMessage,
 )
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
 from graphon.runtime import GraphRuntimeState

 logger = logging.getLogger(__name__)
@@ -51,6 +52,9 @@ class AppQueueManager(ABC):
        self._graph_runtime_state: GraphRuntimeState | None = None
        self._stopped_cache: TTLCache[tuple, bool] = TTLCache(maxsize=1, ttl=1)
        self._cache_lock = threading.Lock()
+        self._execution_terminal = threading.Event()
+        self._abort_sent = threading.Event()
+        self._lifecycle_lock = threading.Lock()

    def listen(self):
        """
@@ -59,7 +63,7 @@ class AppQueueManager(ABC):
        """
        # wait for APP_MAX_EXECUTION_TIME seconds to stop listen
        listen_timeout = dify_config.APP_MAX_EXECUTION_TIME
-        start_time = time.time()
+        start_time = time.monotonic()
        last_ping_time: int | float = 0
        try:
            while True:
@@ -72,8 +76,14 @@ class AppQueueManager(ABC):
                except queue.Empty:
                    continue
                finally:
-                    elapsed_time = time.time() - start_time
-                    if elapsed_time >= listen_timeout or self._is_stopped():
+                    elapsed_time = time.monotonic() - start_time
+                    timed_out = elapsed_time >= listen_timeout
+                    manually_stopped = self._is_stopped()
+                    if not self._execution_terminal.is_set() and (timed_out or manually_stopped):
+                        reason = (
+                            f"App execution exceeded {listen_timeout} seconds" if timed_out else "App task was stopped"
+                        )
+                        self._abort_execution(reason)
                        # publish two messages to make sure the client can receive the stop signal
                        # and stop listening after the stop signal processed
                        self.publish(
@@ -84,16 +94,33 @@ class AppQueueManager(ABC):
                        self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
                        last_ping_time = elapsed_time // 10
        finally:
+            if not self._execution_terminal.is_set():
+                self._abort_execution("Client response stream closed before app execution completed")
            self._graph_runtime_state = None  # Release reference once consumers finish or close the generator.

-    def stop_listen(self):
+    def stop_listen(self, *, execution_terminal: bool = False):
        """
        Stop listen to queue
        :return:
        """
+        if execution_terminal:
+            self._execution_terminal.set()
        self._clear_task_belong_cache()
        self._q.put(None)

+    def _abort_execution(self, reason: str) -> None:
+        """Propagate response timeout/disconnect to legacy and GraphEngine runners."""
+        with self._lifecycle_lock:
+            if self._execution_terminal.is_set() or self._abort_sent.is_set():
+                return
+            self._abort_sent.set()
+
+        try:
+            self.set_stop_flag_no_user_check(self._task_id)
+            GraphEngineManager(redis_client).send_stop_command(self._task_id, reason=reason)
+        except Exception:
+            logger.exception("Failed to abort app execution for task %s", self._task_id)
+
    def _clear_task_belong_cache(self) -> None:
        """
        Remove the task belong cache key once listening is finished.
@@ -45,7 +45,7 @@ class MessageBasedAppQueueManager(AppQueueManager):
        if isinstance(
            event, QueueStopEvent | QueueErrorEvent | QueueMessageEndEvent | QueueAdvancedChatMessageEndEvent
        ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)

        if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
            if self._app_mode == AppMode.ADVANCED_CHAT.value:
@@ -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(
@@ -42,7 +42,7 @@ class PipelineQueueManager(AppQueueManager):
            | QueueWorkflowFailedEvent
            | QueueWorkflowPartialSuccessEvent,
        ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)

        if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
            raise GenerateTaskStoppedError()
@@ -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(
@@ -41,4 +41,4 @@ class WorkflowAppQueueManager(AppQueueManager):
            | QueueWorkflowFailedEvent
            | QueueWorkflowPartialSuccessEvent,
        ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)
@@ -26,6 +26,7 @@ from core.app.entities.queue_entities import (
    QueueNodeSucceededEvent,
    QueueReasoningChunkEvent,
    QueueRetrieverResourcesEvent,
+    QueueStopEvent,
    QueueTextChunkEvent,
    QueueWorkflowFailedEvent,
    QueueWorkflowPartialSuccessEvent,
@@ -424,7 +425,12 @@ class WorkflowBasedAppRunner:
                    QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
                )
            case GraphRunAbortedEvent():
-                self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+                self._publish_event(
+                    QueueStopEvent(
+                        stopped_by=QueueStopEvent.StopBy.USER_MANUAL,
+                        reason=event.reason or "Workflow execution aborted",
+                    )
+                )
            case GraphRunPausedEvent():
                runtime_state = workflow_entry.graph_engine.graph_runtime_state
                paused_nodes = runtime_state.get_paused_nodes()
@@ -500,11 +500,15 @@ class QueueStopEvent(AppQueueEvent):

    event: QueueEvent = QueueEvent.STOP
    stopped_by: StopBy
+    reason: str | None = None

    def get_stop_reason(self) -> str:
        """
        To stop reason
        """
+        if self.reason:
+            return self.reason
+
        reason_mapping = {
            QueueStopEvent.StopBy.USER_MANUAL: "Stopped by user.",
            QueueStopEvent.StopBy.ANNOTATION_REPLY: "Stopped by annotation reply.",
@@ -47,6 +47,24 @@ class MaxRetriesExceededError(ValueError):
    pass


+class ResponseLimitError(ValueError):
+    """Base error for responses that cannot be safely bounded."""
+
+    pass
+
+
+class ResponseTooLargeError(ResponseLimitError):
+    """Raised when an identity response exceeds the configured byte limit."""
+
+    pass
+
+
+class UnsupportedResponseEncodingError(ResponseLimitError):
+    """Raised when response encoding prevents safe decoded-size enforcement."""
+
+    pass
+
+
 request_error = httpx.RequestError
 max_retries_exceeded_error = MaxRetriesExceededError

@@ -142,7 +160,31 @@ def _inject_trace_headers(headers: Headers | None) -> Headers:
    return headers


-def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+def make_request(
+    method: str,
+    url: str,
+    max_retries: int = SSRF_DEFAULT_MAX_RETRIES,
+    stream_response: bool = False,
+    **kwargs: Any,
+) -> httpx.Response:
+    """Send one SSRF-protected request with optional streaming.
+
+    Args:
+        method: HTTP method sent through the configured SSRF client.
+        url: Absolute request URL.
+        max_retries: Number of retry attempts after the initial request.
+        stream_response: Return an open streaming response that the caller must close.
+        **kwargs: Additional keyword arguments forwarded to ``httpx.Client``.
+
+    Returns:
+        A buffered response, or an open response when ``stream_response`` is true.
+
+    Raises:
+        ToolSSRFError: The configured SSRF proxy rejects the destination.
+        MaxRetriesExceededError: All configured request attempts fail.
+        httpx.RequestError: A request fails while retries are disabled.
+        ValueError: The SSL verification option or request headers are invalid.
+    """
    # Convert requests-style allow_redirects to httpx-style follow_redirects
    if "allow_redirects" in kwargs:
        allow_redirects = kwargs.pop("allow_redirects")
@@ -175,6 +217,11 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
    # When using a forward proxy, httpx may override the Host header based on the URL.
    # We extract and preserve any explicitly set Host header to support virtual hosting.
    user_provided_host = _get_user_provided_host_header(headers)
+    send_kwargs: dict[str, Any] = {}
+    if "auth" in kwargs:
+        send_kwargs["auth"] = kwargs.pop("auth")
+    if "follow_redirects" in kwargs:
+        send_kwargs["follow_redirects"] = kwargs.pop("follow_redirects")

    retries = 0
    while retries <= max_retries:
@@ -185,7 +232,11 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
            if user_provided_host is not None:
                headers["host"] = user_provided_host
            kwargs["headers"] = headers
-            response = client.request(method=method, url=url, **kwargs)
+            request = client.build_request(method=method, url=url, **kwargs)
+            if stream_response:
+                response = client.send(request, stream=True, **send_kwargs)
+            else:
+                response = client.send(request, **send_kwargs)

            # Check for SSRF protection by Squid proxy
            if response.status_code in (401, 403):
@@ -195,6 +246,7 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR

                # Squid typically identifies itself in Server or Via headers
                if "squid" in server_header or "squid" in via_header:
+                    response.close()
                    raise ToolSSRFError(
                        f"Access to '{url}' was blocked by SSRF protection. "
                        f"The URL may point to a private or local network address. "
@@ -208,6 +260,7 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
                    response.status_code,
                    url,
                )
+                response.close()

        except httpx.RequestError as e:
            logger.warning("Request to URL %s failed on attempt %s: %s", url, retries + 1, e)
@@ -220,6 +273,42 @@ def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETR
    raise MaxRetriesExceededError(f"Reached maximum retries ({max_retries}) for URL {url}")


+def buffer_response(response: httpx.Response, *, max_response_bytes: int) -> httpx.Response:
+    """Consume one open identity response under a decoded byte limit and close its stream."""
+    if max_response_bytes <= 0:
+        raise ValueError("max_response_bytes must be positive")
+
+    try:
+        content_encoding = response.headers.get("content-encoding", "identity").strip().lower()
+        if content_encoding not in {"", "identity"}:
+            raise UnsupportedResponseEncodingError(f"content encoding {content_encoding} cannot be safely bounded")
+        content = bytearray()
+        for chunk in response.iter_bytes():
+            if len(content) + len(chunk) > max_response_bytes:
+                raise ResponseTooLargeError(f"response exceeded {max_response_bytes} bytes")
+            content.extend(chunk)
+        decoded_headers = {
+            name: value
+            for name, value in response.headers.items()
+            if name.lower() not in {"content-encoding", "content-length", "transfer-encoding"}
+        }
+        try:
+            request = response.request
+        except RuntimeError:
+            request = None
+        return httpx.Response(
+            response.status_code,
+            headers=decoded_headers,
+            content=bytes(content),
+            request=request,
+            extensions=response.extensions,
+            history=response.history,
+            default_encoding=response.default_encoding,
+        )
+    finally:
+        response.close()
+
+
 def get(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
    return make_request("GET", url, max_retries=max_retries, **kwargs)

@@ -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"

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

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

    SNIPPETS_CREATE_AND_MODIFY = "snippets_create_and_modify"
    SNIPPETS_MANAGE = "snippets_management"
@@ -58,7 +58,6 @@ class Tool(ABC):
        if self.runtime and self.runtime.runtime_parameters:
            tool_parameters.update(self.runtime.runtime_parameters)

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

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

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

        return result

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

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

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

            schema["properties"][parameter.name] = parameter_schema
@@ -292,9 +292,7 @@ class ToolInvokeMessageBinary(BaseModel):


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

    class ToolParameterType(StrEnum):
        """
@@ -333,12 +331,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 +392,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):
@@ -499,6 +499,9 @@ class DifyNodeFactory(NodeFactory):
                    base_url=dify_config.AGENT_BACKEND_BASE_URL,
                    use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                    fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
+                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+                    stream_max_reconnects=dify_config.AGENT_BACKEND_STREAM_MAX_RECONNECTS,
+                    stream_run_timeout_seconds=dify_config.AGENT_BACKEND_RUN_TIMEOUT_SECONDS,
                ),
                "event_adapter": AgentBackendRunEventAdapter(),
                # Agent Files §4.6: reback file outputs from the ToolFile row so
@@ -5,6 +5,7 @@ from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, override

 from agenton.compositor import CompositorSessionSnapshot
+from dify_agent.protocol import CancelRunRequest

 from clients.agent_backend import (
    AgentBackendAgentMessageDeltaInternalEvent,
@@ -473,7 +474,10 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        """
        stream_event_count = 0
        try:
-            for public_event in self._agent_backend_client.stream_events(run_id):
+            for public_event in self._agent_backend_client.stream_events(
+                run_id,
+                should_stop=self._is_graph_aborted,
+            ):
                stream_event_count += 1
                for internal_event in self._event_adapter.adapt(public_event):
                    if internal_event.type == AgentBackendInternalEventType.RUN_STARTED:
@@ -501,6 +505,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                        | AgentBackendDeferredToolCallInternalEvent,
                    ):
                        return internal_event, None
+                    self._cancel_backend_run(run_id, reason="unexpected_event")
                    return None, self._failure_event(
                        inputs={},
                        process_data={},
@@ -509,6 +514,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                        error_type="agent_backend_stream_error",
                    )
        except AgentBackendError as error:
+            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
            return None, self._failure_event(
                inputs={},
                process_data={},
@@ -517,6 +523,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                error_type=self._agent_backend_error_type(error),
            )
        except Exception as error:
+            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
            return None, self._failure_event(
                inputs={},
                process_data={},
@@ -525,8 +532,28 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                error_type="agent_backend_stream_error",
            )

+        self._cancel_backend_run(run_id, reason="stream_ended_without_terminal_event")
        return None, None

+    def _is_graph_aborted(self) -> bool:
+        """Let Agent SSE consumption observe GraphEngine's cooperative abort state."""
+        try:
+            return self.graph_runtime_state.graph_execution.aborted
+        except (AttributeError, RuntimeError):
+            return False
+
+    def _stream_stop_reason(self) -> str:
+        return "workflow_graph_aborted" if self._is_graph_aborted() else "event_stream_failed"
+
+    def _cancel_backend_run(self, run_id: str, *, reason: str) -> None:
+        try:
+            self._agent_backend_client.cancel_run(
+                run_id,
+                CancelRunRequest(reason=reason, message="Workflow Agent event consumption stopped"),
+            )
+        except Exception:
+            logger.warning("Failed to cancel Workflow Agent backend run: run_id=%s", run_id, exc_info=True)
+
    @staticmethod
    def _record_type_check_metadata(metadata: dict[str, Any], outcome: OutputTypeCheckOutcome) -> None:
        # Surface enough detail in metadata for Inspector / debug logs without
@@ -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,
@@ -61,16 +61,33 @@ class WorkflowAgentNodeValidator:

    @classmethod
    def validate_draft_workflow(cls, *, session: Session, workflow: Workflow) -> None:
-        cls._validate_workflow(session=session, workflow=workflow, require_binding=False)
+        cls._validate_workflow(
+            session=session,
+            workflow=workflow,
+            require_binding=False,
+            validate_previous_node_topology=False,
+        )

    @classmethod
    def validate_published_workflow(cls, *, session: Session, workflow: Workflow) -> None:
-        cls._validate_workflow(session=session, workflow=workflow, require_binding=True)
+        cls._validate_workflow(
+            session=session,
+            workflow=workflow,
+            require_binding=True,
+            validate_previous_node_topology=True,
+        )

    @classmethod
-    def _validate_workflow(cls, *, session: Session, workflow: Workflow, require_binding: bool) -> None:
+    def _validate_workflow(
+        cls,
+        *,
+        session: Session,
+        workflow: Workflow,
+        require_binding: bool,
+        validate_previous_node_topology: bool,
+    ) -> None:
        graph = workflow.graph_dict
-        topology = _WorkflowGraphTopology.from_graph(graph)
+        topology = _WorkflowGraphTopology.from_graph(graph) if validate_previous_node_topology else None
        for node_id, node_data in cls.iter_agent_v2_nodes(graph):
            cls._validate_node_schema(node_id=node_id, node_data=node_data)
            binding = cls._find_binding(
@@ -185,12 +202,12 @@ class WorkflowAgentNodeValidator:
                raise WorkflowAgentNodeValidationError(
                    f"Workflow Agent node {binding.node_id} has invalid previous node output ref."
                )
-            if topology is None:
-                continue
            if len(selector) < 2:
                raise WorkflowAgentNodeValidationError(
                    f"Workflow Agent node {binding.node_id} has incomplete previous node output ref."
                )
+            if topology is None:
+                continue
            source_node_id = selector[0]
            if not topology.has_node(source_node_id):
                raise WorkflowAgentNodeValidationError(
@@ -0,0 +1,385 @@
+"""Validate Dify Console KnowledgeFS declarations against a pinned OpenAPI document.
+
+The OpenAPI document is exported only during explicit development validation. Runtime declarations live with Dify
+product policy; this module validates their transport metadata without generating a complete operation catalog.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import subprocess
+import sys
+import tempfile
+from copy import deepcopy
+from pathlib import Path
+from typing import Any, Literal, TypedDict
+
+API_ROOT = Path(__file__).resolve().parents[1]
+if str(API_ROOT) not in sys.path:
+    sys.path.insert(0, str(API_ROOT))
+
+WORKSPACE_ROOT = API_ROOT.parent
+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):
+    """KnowledgeFS transport contract declared by one Dify Console registry entry."""
+
+    operation_id: str
+    method: str
+    path: str
+    required_scope: str | None
+    response_kind: str
+    max_response_bytes: int
+    request_headers: tuple[str, ...]
+    response_headers: tuple[str, ...]
+    response_media_types: tuple[str, ...]
+    error_status_map: tuple[tuple[int, int], ...]
+
+
+type DeclarationField = Literal[
+    "method",
+    "path",
+    "required_scope",
+    "response_kind",
+    "max_response_bytes",
+    "request_headers",
+    "response_headers",
+    "response_media_types",
+]
+
+DECLARATION_FIELDS: tuple[DeclarationField, ...] = (
+    "method",
+    "path",
+    "required_scope",
+    "response_kind",
+    "max_response_bytes",
+    "request_headers",
+    "response_headers",
+    "response_media_types",
+)
+
+
+def main() -> None:
+    """Update or verify the pin and validate Console declarations against its OpenAPI document."""
+    parser = argparse.ArgumentParser()
+    mode = parser.add_mutually_exclusive_group()
+    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()
+    lock = json.loads(LOCK_PATH.read_text())
+    tracked_changes = run("git", "status", "--porcelain", "--untracked-files=no", cwd=repository).strip()
+    if tracked_changes:
+        raise RuntimeError("KnowledgeFS checkout must not contain tracked changes during contract export")
+
+    commit = run("git", "rev-parse", "HEAD", cwd=repository).strip()
+    if not args.update_lock and commit != lock["commit"]:
+        raise RuntimeError(
+            f"KnowledgeFS checkout mismatch: expected {lock['commit']}, received {commit}. "
+            "Use the pinned commit or pass --update-lock intentionally."
+        )
+
+    with tempfile.TemporaryDirectory(prefix="dify-knowledge-fs-contract-") as directory:
+        openapi_path = Path(directory) / "knowledge-fs.openapi.json"
+        subprocess.run(
+            ["pnpm", "openapi:export", "--", "--output", str(openapi_path)],
+            cwd=repository,
+            check=True,
+        )
+        openapi_content = openapi_path.read_bytes()
+
+    openapi_sha256 = sha256(openapi_content)
+    if not args.update_lock and openapi_sha256 != lock["openapiSha256"]:
+        raise RuntimeError(
+            f"KnowledgeFS OpenAPI hash mismatch: expected {lock['openapiSha256']}, received {openapi_sha256}"
+        )
+
+    document: dict[str, Any] = json.loads(openapi_content)
+    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(
+            json.dumps(
+                {
+                    "commit": commit,
+                    "openapiSha256": openapi_sha256,
+                    "repository": lock["repository"],
+                },
+                indent=2,
+            )
+            + "\n"
+        )
+
+
+def validate_declarations(document: dict[str, Any], declarations: tuple[ContractDeclaration, ...]) -> None:
+    """Validate Dify Console declarations against matching pinned OpenAPI operations."""
+    operations_by_id: dict[str, list[tuple[str, str, dict[str, Any], dict[str, Any]]]] = {}
+    for path, path_item in document.get("paths", {}).items():
+        for method in OPENAPI_METHODS:
+            operation = path_item.get(method)
+            if operation is None:
+                continue
+            operation_id = operation.get("operationId")
+            if isinstance(operation_id, str) and operation_id:
+                operations_by_id.setdefault(operation_id, []).append((method, path, path_item, operation))
+
+    declared_ids: set[str] = set()
+    for declaration in declarations:
+        operation_id = declaration["operation_id"]
+        if operation_id in declared_ids:
+            raise ValueError(f"Dify Console registry has duplicate operationId: {operation_id}")
+        declared_ids.add(operation_id)
+
+        matches = operations_by_id.get(operation_id, [])
+        if not matches:
+            raise ValueError(f"KnowledgeFS OpenAPI has no operationId: {operation_id}")
+        if len(matches) > 1:
+            raise ValueError(f"KnowledgeFS OpenAPI has duplicate operationId: {operation_id}")
+
+        method, path, path_item, operation = matches[0]
+        if not path.startswith("/"):
+            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: dict[DeclarationField, object] = {
+            "method": method.upper(),
+            "path": path[1:],
+            "required_scope": required_scope(operation),
+            "response_kind": response_kind(operation),
+            "max_response_bytes": required_max_response_bytes(operation),
+            "request_headers": request_header_names(path_item, operation),
+            "response_headers": response_header_names(operation),
+            "response_media_types": response_media_types(operation),
+        }
+        for field in DECLARATION_FIELDS:
+            expected_value = expected[field]
+            received_value = declaration[field]
+            if received_value != expected_value:
+                raise ValueError(
+                    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_operations import KNOWLEDGE_FS_CONSOLE_OPERATIONS
+
+    return tuple(
+        {
+            "operation_id": operation.operation_id,
+            "method": operation.method,
+            "path": operation.path,
+            "required_scope": operation.required_scope,
+            "response_kind": operation.response_kind,
+            "max_response_bytes": operation.max_response_bytes,
+            "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:
+        return "stream"
+    if "application/octet-stream" in media_types:
+        return "binary"
+    return "buffered"
+
+
+def response_media_types(operation: dict[str, Any]) -> tuple[str, ...]:
+    media_types: set[str] = set()
+    for status, response in operation.get("responses", {}).items():
+        if status == "2XX" or (len(status) == 3 and status.startswith("2") and status.isdigit()):
+            media_types.update(response.get("content", {}))
+    return tuple(sorted(media_types))
+
+
+def required_scope(operation: dict[str, Any]) -> str | None:
+    scope = operation.get("x-knowledge-fs-required-scope")
+    if scope in ("knowledge-spaces:read", "knowledge-spaces:write"):
+        return scope
+    if operation.get("security") == []:
+        return None
+    raise ValueError(f"KnowledgeFS operation has no supported required scope: {scope}")
+
+
+def required_max_response_bytes(operation: dict[str, Any]) -> int:
+    value = operation.get("x-knowledge-fs-max-response-bytes")
+    if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
+        raise ValueError(f"KnowledgeFS operation has no valid response byte limit: {value}")
+    return value
+
+
+def request_header_names(path_item: dict[str, Any], operation: dict[str, Any]) -> tuple[str, ...]:
+    names: set[str] = set()
+    for parameter in [*path_item.get("parameters", []), *operation.get("parameters", [])]:
+        if "$ref" in parameter:
+            raise ValueError(f"KnowledgeFS request header references are not supported: {parameter['$ref']}")
+        if parameter.get("in") == "header":
+            names.add(parameter["name"].lower())
+    return tuple(sorted(names))
+
+
+def response_header_names(operation: dict[str, Any]) -> tuple[str, ...]:
+    return tuple(
+        sorted(
+            {
+                name.lower()
+                for response in operation.get("responses", {}).values()
+                for name in response.get("headers", {})
+            }
+        )
+    )
+
+
+def sha256(content: bytes) -> str:
+    return hashlib.sha256(content).hexdigest()
+
+
+def run(*command: str, cwd: Path) -> str:
+    return subprocess.run(command, cwd=cwd, check=True, capture_output=True, text=True).stdout
+
+
+if __name__ == "__main__":
+    main()
@@ -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
    ]
@@ -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)

@@ -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.
@@ -0,0 +1,5 @@
+{
+  "commit": "a0f50470612cc0b3656f89e4f2435aaf412e6e3b",
+  "openapiSha256": "f18910e9c45a64f0855e0643a7a626fb2889021b4f943458de86c6bd2469facb",
+  "repository": "https://github.com/langgenius/knowledge-fs"
+}
@@ -16,12 +16,19 @@ from urllib.parse import quote
 import boto3
 import orjson
 from botocore.client import Config
-from botocore.exceptions import ClientError
+from botocore.exceptions import BotoCoreError, ClientError

 from configs import dify_config

 logger = logging.getLogger(__name__)

+_OBJECT_NOT_FOUND_ERROR_CODES = frozenset({"404", "NoSuchKey", "NotFound"})
+
+
+def _is_object_not_found_error(error: ClientError) -> bool:
+    error_code = str(error.response.get("Error", {}).get("Code", ""))
+    return error_code in _OBJECT_NOT_FOUND_ERROR_CODES
+

 class ArchiveStorageError(Exception):
    """Base exception for archive storage operations."""
@@ -138,10 +145,11 @@ class ArchiveStorage:
            response = self.client.get_object(Bucket=self.bucket, Key=key)
            return response["Body"].read()
        except ClientError as e:
-            error_code = e.response.get("Error", {}).get("Code")
-            if error_code == "NoSuchKey":
-                raise FileNotFoundError(f"Archive object not found: {key}")
-            raise ArchiveStorageError(f"Failed to download object '{key}': {e}")
+            if _is_object_not_found_error(e):
+                raise FileNotFoundError(f"Archive object not found: {key}") from e
+            raise ArchiveStorageError(f"Failed to download object '{key}': {e}") from e
+        except BotoCoreError as e:
+            raise ArchiveStorageError(f"Failed to download object '{key}': {e}") from e

    def get_object_stream(self, key: str) -> Generator[bytes, None, None]:
        """
@@ -161,10 +169,11 @@ class ArchiveStorage:
            response = self.client.get_object(Bucket=self.bucket, Key=key)
            yield from response["Body"].iter_chunks()
        except ClientError as e:
-            error_code = e.response.get("Error", {}).get("Code")
-            if error_code == "NoSuchKey":
-                raise FileNotFoundError(f"Archive object not found: {key}")
-            raise ArchiveStorageError(f"Failed to stream object '{key}': {e}")
+            if _is_object_not_found_error(e):
+                raise FileNotFoundError(f"Archive object not found: {key}") from e
+            raise ArchiveStorageError(f"Failed to stream object '{key}': {e}") from e
+        except BotoCoreError as e:
+            raise ArchiveStorageError(f"Failed to stream object '{key}': {e}") from e

    def object_exists(self, key: str) -> bool:
        """
@@ -175,12 +184,19 @@ class ArchiveStorage:

        Returns:
            True if object exists, False otherwise
+
+        Raises:
+            ArchiveStorageError: If storage cannot authoritatively determine object existence
        """
        try:
            self.client.head_object(Bucket=self.bucket, Key=key)
            return True
-        except ClientError:
-            return False
+        except ClientError as e:
+            if _is_object_not_found_error(e):
+                return False
+            raise ArchiveStorageError(f"Failed to check archive object '{key}': {e}") from e
+        except BotoCoreError as e:
+            raise ArchiveStorageError(f"Failed to check archive object '{key}': {e}") from e

    def delete_object(self, key: str) -> None:
        """
@@ -196,7 +212,12 @@ class ArchiveStorage:
            self.client.delete_object(Bucket=self.bucket, Key=key)
            logger.debug("Deleted object: %s", key)
        except ClientError as e:
-            raise ArchiveStorageError(f"Failed to delete object '{key}': {e}")
+            if _is_object_not_found_error(e):
+                logger.debug("Archive object was already absent: %s", key)
+                return
+            raise ArchiveStorageError(f"Failed to delete object '{key}': {e}") from e
+        except BotoCoreError as e:
+            raise ArchiveStorageError(f"Failed to delete object '{key}': {e}") from e

    def generate_presigned_url(
        self,
@@ -9,6 +9,8 @@ from werkzeug.http import HTTP_STATUS_CODES

 from configs import dify_config
 from core.errors.error import AppInvokeQuotaExceededError
+from core.plugin.impl.exc import PluginRuntimeError
+from extensions.ext_logging import get_request_id
 from libs.flask_restx_compat import install_swagger_compatibility
 from libs.token import build_force_logout_cookie_headers

@@ -100,6 +102,20 @@ def register_external_error_handlers(api: Api, body_formatter: ErrorBodyFormatte
        data = {"code": "too_many_requests", "message": str(e), "status": status_code}
        return _finalize(e, data, status_code), status_code

+    def handle_plugin_runtime_error(e: PluginRuntimeError):
+        got_request_exception.send(current_app, exception=e)
+        status_code = 502
+        details = {"request_id": get_request_id()}
+        if e.lambda_request_id:
+            details["lambda_request_id"] = e.lambda_request_id
+        data = {
+            "code": "plugin_runtime_error",
+            "message": e.description,
+            "details": details,
+            "status": status_code,
+        }
+        return _finalize(e, data, status_code), status_code
+
    def handle_general_exception(e: Exception):
        got_request_exception.send(current_app, exception=e)

@@ -121,6 +137,7 @@ def register_external_error_handlers(api: Api, body_formatter: ErrorBodyFormatte
    api.errorhandler(HTTPException)(handle_http_exception)
    api.errorhandler(ValueError)(handle_value_error)
    api.errorhandler(AppInvokeQuotaExceededError)(handle_quota_exceeded)
+    api.errorhandler(PluginRuntimeError)(handle_plugin_runtime_error)
    api.errorhandler(Exception)(handle_general_exception)


@@ -221,8 +221,11 @@ def current_timestamp() -> int:
 def email(email):
    # Define a regex pattern for email addresses
    pattern = r"^[\w\.!#$%&'*+\-/=?^_`{|}~]+@([\w-]+\.)+[\w-]{2,}$"
-    # Check if the email matches the pattern
-    if re.match(pattern, email) is not None:
+    # Use re.fullmatch instead of re.match to reject trailing newlines.
+    # In Python, '$' matches at end-of-string OR just before a trailing newline,
+    # so re.match accepts "user@example.com\n". re.fullmatch requires the entire
+    # string to match, closing the mail header-injection vector. (#39234)
+    if re.fullmatch(pattern, email) is not None:
        return email

    error = f"{email} is not a valid email."
@@ -23,9 +23,9 @@ class PaginatedResult[T]:

    @property
    def pages(self) -> int:
-        if self.per_page == 0:
+        if self.total == 0 or self.per_page == 0:
            return 0
-        return max(1, math.ceil(self.total / self.per_page))
+        return math.ceil(self.total / self.per_page)

    @property
    def has_next(self) -> bool:
@@ -0,0 +1,39 @@
+"""add step by step tour state
+
+Revision ID: b8c9d0e1f2a3
+Revises: 3c9f8e2a1d7b
+Create Date: 2026-06-29 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+import models
+
+# revision identifiers, used by Alembic.
+revision = "b8c9d0e1f2a3"
+down_revision = "3c9f8e2a1d7b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "account_step_by_step_tour_states",
+        sa.Column("id", models.types.StringUUID(), nullable=False),
+        sa.Column("account_id", models.types.StringUUID(), nullable=False),
+        sa.Column("first_workspace_id", models.types.StringUUID(), nullable=True),
+        sa.Column("skipped", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+        sa.Column("completed_task_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("manually_enabled_workspace_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("manually_disabled_workspace_ids", models.types.AdjustedJSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name="account_step_by_step_tour_state_pkey"),
+        sa.UniqueConstraint("account_id", name="account_step_by_step_tour_state_account_id_key"),
+    )
+
+
+def downgrade():
+    op.drop_table("account_step_by_step_tour_states")
@@ -0,0 +1,45 @@
+"""add workflow-run archive bundle cursor indexes
+
+Revision ID: 3c9f8e2a1d7b
+Revises: 7a1c2d9e4b60
+Create Date: 2026-07-15 16:00:00.000000
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "3c9f8e2a1d7b"
+down_revision = "7a1c2d9e4b60"
+branch_labels = None
+depends_on = None
+
+_TABLE_NAME = "workflow_run_archive_bundles"
+_INDEXES = (
+    ("workflow_run_archive_bundle_month_id_idx", ("year", "month", "id")),
+    ("workflow_run_archive_bundle_month_shard_id_idx", ("year", "month", "shard", "id")),
+)
+
+
+def _is_postgresql() -> bool:
+    return op.get_bind().dialect.name == "postgresql"
+
+
+def upgrade() -> None:
+    if _is_postgresql():
+        with op.get_context().autocommit_block():
+            for index_name, columns in _INDEXES:
+                op.create_index(index_name, _TABLE_NAME, columns, postgresql_concurrently=True)
+        return
+    for index_name, columns in _INDEXES:
+        op.create_index(index_name, _TABLE_NAME, columns)
+
+
+def downgrade() -> None:
+    if _is_postgresql():
+        with op.get_context().autocommit_block():
+            for index_name, _ in reversed(_INDEXES):
+                op.drop_index(index_name, table_name=_TABLE_NAME, postgresql_concurrently=True)
+        return
+    for index_name, _ in reversed(_INDEXES):
+        op.drop_index(index_name, table_name=_TABLE_NAME)
@@ -0,0 +1,77 @@
+"""scope agent debug conversations by draft type
+
+Revision ID: d2825e7b9c10
+Revises: b8c9d0e1f2a3
+Create Date: 2026-07-22 15:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+import models
+
+# revision identifiers, used by Alembic.
+revision = "d2825e7b9c10"
+down_revision = "b8c9d0e1f2a3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Existing pointers have always represented Build chat because the Agent
+    # detail API exposes them as ``debug_conversation_id`` for that surface.
+    op.add_column(
+        "agent_debug_conversations",
+        sa.Column(
+            "draft_type",
+            sa.String(length=32),
+            nullable=False,
+            server_default=sa.text("'debug_build'"),
+        ),
+    )
+    op.drop_constraint(
+        "agent_debug_conversation_agent_account_unique",
+        "agent_debug_conversations",
+        type_="unique",
+    )
+    op.create_unique_constraint(
+        "agent_debug_conversation_agent_account_draft_type_unique",
+        "agent_debug_conversations",
+        ["tenant_id", "agent_id", "account_id", "draft_type"],
+    )
+
+
+def downgrade():
+    debug_conversations = sa.table(
+        "agent_debug_conversations",
+        sa.column("tenant_id", models.types.StringUUID()),
+        sa.column("agent_id", models.types.StringUUID()),
+        sa.column("account_id", models.types.StringUUID()),
+        sa.column("draft_type", sa.String(length=32)),
+    )
+    build_conversations = debug_conversations.alias("build_conversations")
+    op.get_bind().execute(
+        sa.delete(debug_conversations).where(
+            debug_conversations.c.draft_type == "draft",
+            sa.exists(
+                sa.select(sa.literal(1)).where(
+                    build_conversations.c.tenant_id == debug_conversations.c.tenant_id,
+                    build_conversations.c.agent_id == debug_conversations.c.agent_id,
+                    build_conversations.c.account_id == debug_conversations.c.account_id,
+                    build_conversations.c.draft_type == "debug_build",
+                )
+            ),
+        )
+    )
+    op.drop_constraint(
+        "agent_debug_conversation_agent_account_draft_type_unique",
+        "agent_debug_conversations",
+        type_="unique",
+    )
+    op.create_unique_constraint(
+        "agent_debug_conversation_agent_account_unique",
+        "agent_debug_conversations",
+        ["tenant_id", "agent_id", "account_id"],
+    )
+    op.drop_column("agent_debug_conversations", "draft_type")
@@ -101,6 +101,7 @@ from .model import (
    UploadFile,
 )
 from .oauth import DatasourceOauthParamConfig, DatasourceProvider, OAuthAccessToken
+from .onboarding import AccountStepByStepTourState
 from .provider import (
    LoadBalancingModelConfig,
    Provider,
@@ -155,6 +156,7 @@ __all__ = [
    "Account",
    "AccountIntegrate",
    "AccountStatus",
+    "AccountStepByStepTourState",
    "AccountTrialAppRecord",
    "Agent",
    "AgentConfigDraft",
@@ -222,11 +222,13 @@ class Agent(DefaultFieldsMixin, Base):


 class AgentDebugConversation(DefaultFieldsMixin, Base):
-    """Per-account console debug conversation for an Agent App.
+    """Per-account, per-draft console debug conversation for an Agent App.

    Agent App preview state must be isolated by editor account. The Agent row is
    shared by everyone in the workspace, so this table owns the user-specific
-    conversation pointer used by console debug chat.
+    conversation pointers used by console debug chat. ``draft`` is the Preview
+    conversation and ``debug_build`` is the Build conversation; they must never
+    share persisted messages or runtime sessions.
    """

    __tablename__ = "agent_debug_conversations"
@@ -236,7 +238,8 @@ class AgentDebugConversation(DefaultFieldsMixin, Base):
            "tenant_id",
            "agent_id",
            "account_id",
-            name="agent_debug_conversation_agent_account_unique",
+            "draft_type",
+            name="agent_debug_conversation_agent_account_draft_type_unique",
        ),
        Index("agent_debug_conversation_conversation_idx", "conversation_id"),
        Index("agent_debug_conversation_account_idx", "tenant_id", "account_id"),
@@ -246,6 +249,12 @@ class AgentDebugConversation(DefaultFieldsMixin, Base):
    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    draft_type: Mapped[AgentConfigDraftType] = mapped_column(
+        EnumText(AgentConfigDraftType, length=32),
+        nullable=False,
+        default=AgentConfigDraftType.DEBUG_BUILD,
+        server_default=sa.text("'debug_build'"),
+    )
    conversation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)


@@ -7,6 +7,7 @@ from typing import Annotated, Any, Final, Literal, Self
 from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, field_validator, model_validator

 from core.rag.entities.metadata_entities import ConditionValue, SupportedComparisonOperator
+from core.tools.entities.tool_entities import ToolProviderType
 from core.workflow.file_reference import is_canonical_file_reference
 from graphon.file import FileTransferMethod, FileType

@@ -43,18 +44,28 @@ _DECLARED_OUTPUT_CHILDREN_JSON_SCHEMA = {
            },
            "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
            "required": {"type": "boolean"},
-            "file": {"type": "object", "additionalProperties": True},
+            "file": {
+                "anyOf": [
+                    {"type": "object", "additionalProperties": True},
+                    {"type": "null"},
+                ]
+            },
            "array_item": {
-                "type": "object",
-                "additionalProperties": True,
-                "properties": {
-                    "type": {
-                        "type": "string",
-                        "enum": [item.value for item in DeclaredOutputType],
+                "anyOf": [
+                    {
+                        "type": "object",
+                        "additionalProperties": True,
+                        "properties": {
+                            "type": {
+                                "type": "string",
+                                "enum": [item.value for item in DeclaredOutputType],
+                            },
+                            "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
+                            "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
+                        },
                    },
-                    "description": {"anyOf": [{"type": "string"}, {"type": "null"}]},
-                    "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
-                },
+                    {"type": "null"},
+                ]
            },
            "children": {"type": "array", "items": {"type": "object", "additionalProperties": True}},
        },
@@ -653,11 +664,7 @@ class AgentSoulDifyToolConfig(BaseModel):
    model_config = ConfigDict(extra="ignore")

    enabled: bool = True
-    # ``plugin`` remains the default for legacy Agent Soul payloads. The runtime
-    # now also accepts ``builtin`` / ``api`` / ``workflow`` / ``mcp`` here and
-    # routes them through ``dify.core.tools``; keeping the default narrow still
-    # makes a missing field resolve against the plugin provider table.
-    provider_type: str = "plugin"
+    provider_type: ToolProviderType
    provider_id: str | None = Field(default=None, max_length=255)
    plugin_id: str | None = Field(default=None, max_length=255)
    provider: str | None = Field(default=None, max_length=255)
@@ -56,6 +56,7 @@ from .provider_ids import GenericProviderID
 from .types import EnumText, LongText, StringUUID

 if TYPE_CHECKING:
+    from .agent import Agent
    from .workflow import Workflow


@@ -501,25 +502,42 @@ class App(Base):
        Resolved via ``Agent.app_id`` so the console can open the Composer in
        roster-detail mode from the app id. ``None`` for non-agent apps.
        """
+        agent = self.agent_app_binding_with_session(session=session)
+        return agent.id if agent else None
+
+    def agent_app_binding_with_session(self, *, session: Session, include_archived: bool = False) -> Agent | None:
+        """For an Agent App (mode=agent), the Agent bound to it.
+
+        A roster Agent is bound through ``Agent.app_id``; a workflow-only Agent
+        is bound to its hidden runtime backing App through
+        ``Agent.backing_app_id``. Callers branch on ``Agent.scope`` to tell the
+        public roster Agent App apart from the hidden backing App. Archived
+        Agents are excluded unless ``include_archived`` is set (authorization
+        gates must keep covering an Agent App after its Agent is archived).
+        ``None`` for non-agent apps and unbound agent apps.
+        """
        if self.mode != AppMode.AGENT:
            return None
        from .agent import APP_BACKED_AGENT_SOURCES, Agent, AgentScope, AgentStatus

-        agent = session.scalar(
-            select(Agent).where(
-                Agent.tenant_id == self.tenant_id,
-                sa.or_(
-                    sa.and_(
-                        Agent.app_id == self.id,
-                        Agent.scope == AgentScope.ROSTER,
-                        Agent.source.in_(APP_BACKED_AGENT_SOURCES),
-                    ),
-                    Agent.backing_app_id == self.id,
+        conditions = [
+            Agent.tenant_id == self.tenant_id,
+            sa.or_(
+                sa.and_(
+                    Agent.app_id == self.id,
+                    Agent.scope == AgentScope.ROSTER,
+                    Agent.source.in_(APP_BACKED_AGENT_SOURCES),
                ),
-                Agent.status == AgentStatus.ACTIVE,
-            )
-        )
-        return agent.id if agent else None
+                sa.and_(
+                    Agent.backing_app_id == self.id,
+                    Agent.scope == AgentScope.WORKFLOW_ONLY,
+                ),
+            ),
+        ]
+        if not include_archived:
+            conditions.append(Agent.status == AgentStatus.ACTIVE)
+
+        return session.scalar(select(Agent).where(*conditions).limit(1))

    @property
    def api_base_url(self) -> str:
@@ -0,0 +1,59 @@
+"""Account-level onboarding state models."""
+
+from datetime import datetime
+
+import sqlalchemy as sa
+from sqlalchemy import DateTime, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import TypeBase, gen_uuidv7_string
+from .types import AdjustedJSON, StringUUID
+
+
+class AccountStepByStepTourState(TypeBase):
+    """Persistent account-level Step-by-step Tour state.
+
+    The tour is account-owned, with workspace IDs stored only as presentation
+    overrides. The first workspace is the workspace context where an eligible
+    account first asks for tour state; subsequent workspaces are opt-in only.
+    """
+
+    __tablename__ = "account_step_by_step_tour_states"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="account_step_by_step_tour_state_pkey"),
+        sa.UniqueConstraint("account_id", name="account_step_by_step_tour_state_account_id_key"),
+    )
+
+    id: Mapped[str] = mapped_column(
+        StringUUID,
+        insert_default=gen_uuidv7_string,
+        default_factory=gen_uuidv7_string,
+        init=False,
+    )
+    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    first_workspace_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
+    skipped: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"), default=False)
+    completed_task_ids: Mapped[list[str]] = mapped_column(AdjustedJSON, nullable=False, default_factory=list)
+    manually_enabled_workspace_ids: Mapped[list[str]] = mapped_column(
+        AdjustedJSON,
+        nullable=False,
+        default_factory=list,
+    )
+    manually_disabled_workspace_ids: Mapped[list[str]] = mapped_column(
+        AdjustedJSON,
+        nullable=False,
+        default_factory=list,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        nullable=False,
+        init=False,
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        nullable=False,
+        init=False,
+        onupdate=func.current_timestamp(),
+    )
@@ -1476,6 +1476,8 @@ class WorkflowRunArchiveBundle(DefaultFieldsDCMixin, TypeBase):
            name="workflow_run_archive_bundle_identity_uq",
        ),
        sa.Index("workflow_run_archive_bundle_tenant_month_idx", "tenant_id", "year", "month"),
+        sa.Index("workflow_run_archive_bundle_month_id_idx", "year", "month", "id"),
+        sa.Index("workflow_run_archive_bundle_month_shard_id_idx", "year", "month", "shard", "id"),
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
@@ -260,7 +260,12 @@ Get account avatar url
 | 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |

 ### [POST] /activate
+**Accept an invitation without letting an existing session act for another account**
+
 Activate account with invitation token
+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.

 #### Request Body

@@ -531,6 +536,7 @@ Run a build-draft Agent App turn that asks the agent to push config updates
 | Code | Description | Schema |
 | ---- | ----------- | ------ |
 | 200 | Agent build draft | **application/json**: [AgentBuildDraftResponse](#agentbuilddraftresponse)<br> |
+| 404 | Agent build draft not found |  |

 ### [PUT] /agent/{agent_id}/build-draft
 #### Parameters
@@ -958,6 +964,12 @@ Stop a running Agent App chat message generation
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path |  | Yes | string (uuid) |

+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  No | **application/json**: [AgentDebugConversationRefreshPayload](#agentdebugconversationrefreshpayload)<br> |
+
 #### Responses

 | Code | Description | Schema |
@@ -7787,6 +7799,30 @@ Initiate OAuth login process
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [OAuthProviderTokenResponse](#oauthprovidertokenresponse)<br> |

+### [GET] /onboarding/step-by-step-tour/state
+Get account-level Step-by-step Tour state
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [StepByStepTourStateResponse](#stepbysteptourstateresponse)<br> |
+
+### [PATCH] /onboarding/step-by-step-tour/state
+Update account-level Step-by-step Tour state
+
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **application/json**: [StepByStepTourStatePatchPayload](#stepbysteptourstatepatchpayload)<br> |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [StepByStepTourStateResponse](#stepbysteptourstateresponse)<br> |
+
 ### [DELETE] /rag/pipeline/customized/templates/{template_id}
 #### Parameters

@@ -9639,6 +9675,27 @@ Bedrock retrieval test (internal use only)
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [TrialDatasetListResponse](#trialdatasetlistresponse)<br> |

+### [POST] /trial-apps/{app_id}/files/upload
+**Upload a file into the tenant that owns the trial app**
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| app_id | path |  | Yes | string (uuid) |
+
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **multipart/form-data**: { **"file"**: binary, **"source"**: string, <br>**Available values:** "datasets" }<br> |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 201 | File uploaded successfully | **application/json**: [FileResponse](#fileresponse)<br> |
+
 ### [GET] /trial-apps/{app_id}/messages/{message_id}/suggested-questions
 #### Parameters

@@ -9668,6 +9725,27 @@ Bedrock retrieval test (internal use only)
 | ---- | ----------- | ------ |
 | 200 | Success | **application/json**: [Parameters](#parameters)<br> |

+### [POST] /trial-apps/{app_id}/remote-files/upload
+**Upload a remote file into the tenant that owns the trial app**
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| app_id | path |  | Yes | string (uuid) |
+
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **application/json**: [RemoteFileUploadPayload](#remotefileuploadpayload)<br> |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 201 | File uploaded successfully | **application/json**: [FileWithSignedUrl](#filewithsignedurl)<br> |
+
 ### [GET] /trial-apps/{app_id}/site
 **Retrieve app site info**

@@ -11266,6 +11344,12 @@ Returns permission flags that control workspace features like member invitations
 | 200 | Success | **application/json**: [PluginDecodeResponse](#plugindecoderesponse)<br> |

 ### [POST] /workspaces/current/plugin/upload/pkg
+#### Request Body
+
+| Required | Schema |
+| -------- | ------ |
+|  Yes | **multipart/form-data**: { **"pkg"**: binary }<br> |
+
 #### Responses

 | Code | Description | Schema |
@@ -13212,7 +13296,7 @@ Model class for AI model.
 | maintainer | string |  | No |
 | max_active_requests | integer |  | No |
 | mode | string |  | Yes |
-| model_config | [ModelConfig](#modelconfig) |  | No |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | role | string |  | No |
@@ -13833,6 +13917,12 @@ Stable Agent Soul reference to one normalized skill archive.
 | date | string |  | Yes |
 | message_count | integer |  | Yes |

+#### AgentDebugConversationRefreshPayload
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| draft_type | [AgentConfigDraftType](#agentconfigdrafttype) | Agent draft surface whose conversation should be refreshed | No |
+
 #### AgentDebugConversationRefreshResponse

 | Name | Type | Description | Required |
@@ -14671,7 +14761,7 @@ should send ``plugin_id`` + ``provider`` when available.
 | plugin_id | string |  | No |
 | provider | string |  | No |
 | provider_id | string |  | No |
-| provider_type | string, <br>**Default:** plugin |  | No |
+| provider_type | [ToolProviderType](#toolprovidertype) |  | Yes |
 | runtime_parameters | object |  | No |
 | tool_name | string |  | No |

@@ -15282,7 +15372,6 @@ This class is used to store the schema information of an api based tool.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | access_mode | string |  | No |
-| app_model_config | [ModelConfig](#modelconfig) |  | No |
 | created_at | integer |  | No |
 | created_by | string |  | No |
 | description | string |  | No |
@@ -15292,7 +15381,8 @@ This class is used to store the schema information of an api based tool.
 | icon_background | string |  | No |
 | id | string |  | Yes |
 | maintainer | string |  | No |
-| mode_compatible_with_agent | string |  | Yes |
+| mode | string |  | Yes |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | tags | [ [Tag](#tag) ] |  | No |
@@ -15354,7 +15444,7 @@ This class is used to store the schema information of an api based tool.
 | maintainer | string |  | No |
 | max_active_requests | integer |  | No |
 | mode | string |  | Yes |
-| model_config | [ModelConfig](#modelconfig) |  | No |
+| model_config | [AppModelConfigResponse](#appmodelconfigresponse) |  | No |
 | name | string |  | Yes |
 | permission_keys | [ string ] |  | No |
 | site | [AppDetailSiteResponse](#appdetailsiteresponse) |  | No |
@@ -15453,6 +15543,35 @@ AppMCPServer Status Enum
 | ---- | ---- | ----------- | -------- |
 | AppMCPServerStatus | string | AppMCPServer Status Enum |  |

+#### AppModelConfigResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| agent_mode |  |  | No |
+| annotation_reply |  |  | No |
+| chat_prompt_config |  |  | No |
+| completion_prompt_config |  |  | No |
+| created_at | integer |  | No |
+| created_by | string |  | No |
+| dataset_configs |  |  | No |
+| dataset_query_variable | string |  | No |
+| external_data_tools |  |  | No |
+| file_upload |  |  | No |
+| model |  |  | No |
+| more_like_this |  |  | No |
+| opening_statement | string |  | No |
+| pre_prompt | string |  | No |
+| prompt_type | string |  | No |
+| retriever_resource |  |  | No |
+| sensitive_word_avoidance |  |  | No |
+| speech_to_text |  |  | No |
+| suggested_questions |  |  | No |
+| suggested_questions_after_answer |  |  | No |
+| text_to_speech |  |  | No |
+| updated_at | integer |  | No |
+| updated_by | string |  | No |
+| user_input_form |  |  | No |
+
 #### AppNamePayload

 | Name | Type | Description | Required |
@@ -17081,7 +17200,7 @@ about. Stage 4 §4.2.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| children | [ { **"array_item"**: { **"children"**: [ object ], **"description"**: , **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" }, **"children"**: [ object ], **"description"**: , **"file"**: object, **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
+| children | [ { **"array_item"**: , **"children"**: [ object ], **"description"**: , **"file"**: , **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
 | description | string |  | No |
 | type | [DeclaredOutputType](#declaredoutputtype) |  | Yes |

@@ -17110,7 +17229,7 @@ code can call ``output.failure_strategy.on_failure`` without None-guards.
 | ---- | ---- | ----------- | -------- |
 | array_item | [DeclaredArrayItem](#declaredarrayitem) |  | No |
 | check | [DeclaredOutputCheckConfig](#declaredoutputcheckconfig) |  | No |
-| children | [ { **"array_item"**: { **"children"**: [ object ], **"description"**: , **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" }, **"children"**: [ object ], **"description"**: , **"file"**: object, **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
+| children | [ { **"array_item"**: , **"children"**: [ object ], **"description"**: , **"file"**: , **"name"**: string, **"required"**: boolean, **"type"**: string, <br>**Available values:** "array", "boolean", "file", "number", "object", "string" } ] |  | No |
 | description | string |  | No |
 | failure_strategy | [DeclaredOutputFailureStrategy](#declaredoutputfailurestrategy) |  | No |
 | file | [DeclaredOutputFileConfig](#declaredoutputfileconfig) |  | No |
@@ -17199,6 +17318,12 @@ Default model entity.
 | tool_name | string |  | Yes |
 | type | string |  | Yes |

+#### DeploymentEdition
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| DeploymentEdition | string |  |  |
+
 #### DismissNotificationPayload

 | Name | Type | Description | Required |
@@ -21746,6 +21871,24 @@ Query parameters for listing snippet published workflows.
 | paused | integer |  | Yes |
 | success | integer |  | Yes |

+#### StepByStepTourStatePatchPayload
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| action | string, <br>**Available values:** "complete_task", "disable_current_workspace", "enable_current_workspace", "skip", "uncomplete_task" | State update action<br>*Enum:* `"complete_task"`, `"disable_current_workspace"`, `"enable_current_workspace"`, `"skip"`, `"uncomplete_task"` | Yes |
+| task_id | string | Task ID for task actions | No |
+
+#### StepByStepTourStateResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| completed_task_ids | [ string, <br>**Available values:** "home", "integration", "knowledge", "studio" ] |  | No |
+| first_workspace_id | string |  | No |
+| manually_disabled_workspace_ids | [ string ] |  | No |
+| manually_enabled_workspace_ids | [ string ] |  | No |
+| skipped | boolean |  | No |
+| updated_at | string |  | No |
+
 #### Storage

 | Name | Type | Description | Required |
@@ -21859,6 +22002,7 @@ The subscription constructor of the trigger provider

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
+| _is_collaborative | boolean |  | No |
 | conversation_variables | [ object ] |  | No |
 | environment_variables | [ object ] |  | No |
 | features | object |  | Yes |
@@ -21869,9 +22013,9 @@ The subscription constructor of the trigger provider

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| hash | string |  | No |
-| result | string |  | No |
-| updated_at | string |  | No |
+| hash | string |  | Yes |
+| result | string |  | Yes |
+| updated_at | integer |  | Yes |

 #### SystemConfigurationResponse

@@ -21888,6 +22032,7 @@ Model class for provider system configuration response.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | branding | [BrandingModel](#brandingmodel) |  | Yes |
+| deployment_edition | [DeploymentEdition](#deploymentedition) |  | Yes |
 | enable_app_deploy | boolean |  | Yes |
 | enable_change_email | boolean, <br>**Default:** true |  | Yes |
 | enable_collaboration_mode | boolean, <br>**Default:** true |  | Yes |
@@ -21898,10 +22043,12 @@ Model class for provider system configuration response.
 | enable_learn_app | boolean, <br>**Default:** true |  | Yes |
 | enable_marketplace | boolean |  | Yes |
 | enable_social_oauth_login | boolean |  | Yes |
+| enable_step_by_step_tour | boolean |  | Yes |
 | enable_trial_app | boolean |  | Yes |
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
 | is_email_setup | boolean |  | Yes |
+| knowledge_fs_enabled | boolean |  | Yes |
 | license | [LicenseModel](#licensemodel) |  | Yes |
 | max_plugin_package_size | integer, <br>**Default:** 15728640 |  | Yes |
 | plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) |  | Yes |
@@ -22195,7 +22342,7 @@ Tool label

 #### ToolParameter

-Overrides type
+Tool-specific parameter declaration and invocation-value normalization.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
@@ -22208,6 +22355,7 @@ Overrides type
 | llm_description | string |  | No |
 | max | number<br>integer |  | No |
 | min | number<br>integer |  | No |
+| multiple | boolean | Whether the parameter is multiple select, only valid for select or dynamic-select type | No |
 | name | string | The name of the parameter | Yes |
 | options | [ [PluginParameterOption](#pluginparameteroption) ] |  | No |
 | placeholder | [I18nObject](#i18nobject) | The placeholder presented to the user | No |
@@ -22854,6 +23002,7 @@ in form definiton, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | allow_email_code_login | boolean |  | Yes |
 | allow_email_password_login | boolean |  | Yes |
+| allow_public_access | boolean, <br>**Default:** true |  | Yes |
 | allow_sso | boolean |  | Yes |
 | enabled | boolean |  | Yes |
 | sso_config | [WebAppAuthSSOModel](#webappauthssomodel) |  | Yes |
@@ -1058,6 +1058,12 @@ Button styles for user actions.
 | auto_generate | boolean | Automatically generate the conversation name. When `true`, the `name` field is ignored. | No |
 | name | string | Conversation name. Required when `auto_generate` is `false`. | No |

+#### DeploymentEdition
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| DeploymentEdition | string |  |  |
+
 #### EmailCodeLoginSendPayload

 | Name | Type | Description | Required |
@@ -1567,6 +1573,7 @@ Default configuration for form inputs.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | branding | [BrandingModel](#brandingmodel) |  | Yes |
+| deployment_edition | [DeploymentEdition](#deploymentedition) |  | Yes |
 | enable_app_deploy | boolean |  | Yes |
 | enable_change_email | boolean, <br>**Default:** true |  | Yes |
 | enable_collaboration_mode | boolean, <br>**Default:** true |  | Yes |
@@ -1577,10 +1584,12 @@ Default configuration for form inputs.
 | enable_learn_app | boolean, <br>**Default:** true |  | Yes |
 | enable_marketplace | boolean |  | Yes |
 | enable_social_oauth_login | boolean |  | Yes |
+| enable_step_by_step_tour | boolean |  | Yes |
 | enable_trial_app | boolean |  | Yes |
 | is_allow_create_workspace | boolean |  | Yes |
 | is_allow_register | boolean |  | Yes |
 | is_email_setup | boolean |  | Yes |
+| knowledge_fs_enabled | boolean |  | Yes |
 | license | [LicenseModel](#licensemodel) |  | Yes |
 | max_plugin_package_size | integer, <br>**Default:** 15728640 |  | Yes |
 | plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) |  | Yes |
@@ -1642,6 +1651,7 @@ in form definiton, or a variable while the workflow is running.
 | ---- | ---- | ----------- | -------- |
 | allow_email_code_login | boolean |  | Yes |
 | allow_email_password_login | boolean |  | Yes |
+| allow_public_access | boolean, <br>**Default:** true |  | Yes |
 | allow_sso | boolean |  | Yes |
 | enabled | boolean |  | Yes |
 | sso_config | [WebAppAuthSSOModel](#webappauthssomodel) |  | Yes |
@@ -1731,7 +1741,7 @@ in form definiton, or a variable while the workflow is running.
 | icon | string |  | No |
 | icon_background | string |  | No |
 | icon_type | string |  | No |
-| icon_url | string |  | Yes |
+| icon_url | string |  | No |
 | input_placeholder | string |  | No |
 | privacy_policy | string |  | No |
 | prompt_public | boolean |  | No |
@@ -0,0 +1,35 @@
+"""Shared database fixtures for provider tests.
+
+Provider tests live outside ``tests/unit_tests`` and therefore cannot use that
+suite's SQLite fixtures. Keep this fixture scoped to ``providers`` so each
+provider package can exercise real SQLAlchemy queries without a service
+database or mocked sessions.
+"""
+
+from collections.abc import Iterator
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from models.base import TypeBase
+
+
+@pytest.fixture
+def sqlite3_session(request: pytest.FixtureRequest) -> Iterator[Session]:
+    """Yield an isolated SQLite session with the parametrized model tables.
+
+    Pass the required ORM classes through indirect parametrization. The engine
+    is per-test so committed rows and identity-map state cannot leak between
+    provider packages.
+    """
+
+    models: tuple[type[TypeBase], ...] = request.param
+    engine = create_engine("sqlite:///:memory:")
+    tables = [model.metadata.tables[model.__tablename__] for model in models]
+    TypeBase.metadata.create_all(engine, tables=tables)
+    try:
+        with Session(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        engine.dispose()
@@ -1,6 +1,6 @@
 [project]
 name = "dify-api"
-version = "1.16.0-rc1"
+version = "1.16.0"
 requires-python = "~=3.12.0"

 dependencies = [
@@ -3,7 +3,10 @@ from __future__ import annotations
 import json
 from typing import NotRequired, TypedDict, override

+from redis.lock import Lock
+
 from extensions.ext_redis import redis_client
+from extensions.redis_names import serialize_redis_name

 SESSION_STATE_TTL_SECONDS = 3600
 SERVER_HEARTBEAT_TTL_SECONDS = 90
@@ -12,6 +15,31 @@ WORKFLOW_LEADER_PREFIX = "workflow_leader:"
 WS_SID_MAP_PREFIX = "ws_sid_map:"
 WS_SERVER_HEARTBEAT_PREFIX = "ws_server_heartbeat:"
 WS_SERVER_SESSIONS_PREFIX = "ws_server_sessions:"
+GRAPH_VIEW_STATE_LOCK_PREFIX = "workflow_graph_view_state_lock:"
+GRAPH_VIEW_STATE_LOCK_TIMEOUT_SECONDS = 15
+
+_UPDATE_SESSION_GRAPH_ACTIVE_LUA = """
+local raw = redis.call('HGET', KEYS[1], ARGV[1])
+if not raw then
+    return 0
+end
+
+local decoded_ok, session_info = pcall(cjson.decode, raw)
+if not decoded_ok or type(session_info) ~= 'table' then
+    return 0
+end
+
+local incoming_sequence = tonumber(ARGV[3])
+local current_sequence = tonumber(session_info.graph_active_sequence)
+if current_sequence and incoming_sequence <= current_sequence then
+    return 0
+end
+
+session_info.graph_active = ARGV[2] == '1'
+session_info.graph_active_sequence = incoming_sequence
+redis.call('HSET', KEYS[1], ARGV[1], cjson.encode(session_info))
+return 1
+"""


 class WorkflowSessionInfo(TypedDict):
@@ -156,6 +184,55 @@ class WorkflowCollaborationRepository:

        return users

+    def get_session_info(self, workflow_id: str, sid: str) -> WorkflowSessionInfo | None:
+        raw = self._redis.hget(self.workflow_key(workflow_id), sid)
+        value = self._decode(raw)
+        if not value:
+            return None
+        try:
+            session_info = json.loads(value)
+        except (TypeError, json.JSONDecodeError):
+            return None
+
+        if not isinstance(session_info, dict):
+            return None
+        if "user_id" not in session_info or "username" not in session_info or "sid" not in session_info:
+            return None
+
+        user: WorkflowSessionInfo = {
+            "user_id": str(session_info["user_id"]),
+            "username": str(session_info["username"]),
+            "avatar": session_info.get("avatar"),
+            "sid": str(session_info["sid"]),
+            "connected_at": int(session_info.get("connected_at") or 0),
+        }
+        if isinstance(session_info.get("server_id"), str):
+            user["server_id"] = session_info["server_id"]
+        if isinstance(session_info.get("graph_active"), bool):
+            user["graph_active"] = session_info["graph_active"]
+        return user
+
+    def update_session_graph_active(self, workflow_id: str, sid: str, active: bool, sequence: int) -> bool:
+        """Atomically apply a graph visibility update when its client sequence is newer."""
+        # RedisClientWrapper prefixes regular hash calls, but eval is delegated to the raw client.
+        workflow_key = serialize_redis_name(self.workflow_key(workflow_id))
+        result = self._redis.eval(
+            _UPDATE_SESSION_GRAPH_ACTIVE_LUA,
+            1,
+            workflow_key,
+            sid,
+            "1" if active else "0",
+            sequence,
+        )
+        return bool(result)
+
+    def graph_view_state_lock(self, workflow_id: str) -> Lock:
+        """Serialize visibility state changes and their leader-election side effects."""
+        return self._redis.lock(
+            f"{GRAPH_VIEW_STATE_LOCK_PREFIX}{workflow_id}",
+            timeout=GRAPH_VIEW_STATE_LOCK_TIMEOUT_SECONDS,
+        )
+
    def refresh_server_heartbeat(self, server_id: str) -> None:
        self._redis.set(self.server_key(server_id), "1", ex=SERVER_HEARTBEAT_TTL_SECONDS)

@@ -1164,6 +1164,20 @@ class AgentComposerService:
                agent.active_config_is_published = True
                agent.updated_by = account_id
                binding.current_snapshot_id = version.id
+                normal_draft = cls._get_agent_draft(
+                    session=session,
+                    tenant_id=tenant_id,
+                    agent_id=agent.id,
+                    draft_type=AgentConfigDraftType.DRAFT,
+                    account_id=None,
+                )
+                if normal_draft is not None and cls._rebase_workflow_only_normal_draft(
+                    agent=agent,
+                    draft=normal_draft,
+                    snapshot=version,
+                    updated_by=account_id,
+                ):
+                    session.flush()
            binding.updated_by = account_id
            return binding

@@ -1748,6 +1762,47 @@ class AgentComposerService:
            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
        return session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))

+    @classmethod
+    def get_or_create_normal_agent_draft(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent: Agent,
+        created_by: str | None,
+    ) -> AgentConfigDraft:
+        """Resolve the shared Preview draft, rebasing inline agents when needed."""
+        return cls._get_or_create_agent_draft(
+            session=session,
+            tenant_id=tenant_id,
+            agent=agent,
+            draft_type=AgentConfigDraftType.DRAFT,
+            account_id=None,
+            created_by=created_by,
+        )
+
+    @staticmethod
+    def _rebase_workflow_only_normal_draft(
+        *,
+        agent: Agent,
+        draft: AgentConfigDraft,
+        snapshot: AgentConfigSnapshot,
+        updated_by: str | None,
+    ) -> bool:
+        if (
+            agent.scope != AgentScope.WORKFLOW_ONLY
+            or draft.draft_type != AgentConfigDraftType.DRAFT
+            or draft.account_id is not None
+            or not agent.active_config_snapshot_id
+            or draft.base_snapshot_id == agent.active_config_snapshot_id
+            or snapshot.id != agent.active_config_snapshot_id
+        ):
+            return False
+        draft.base_snapshot_id = snapshot.id
+        draft.config_snapshot = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
+        draft.updated_by = updated_by
+        return True
+
    @classmethod
    def _get_or_create_agent_draft(
        cls,
@@ -1767,6 +1822,26 @@ class AgentComposerService:
            account_id=account_id,
        )
        if draft is not None:
+            if (
+                agent.scope == AgentScope.WORKFLOW_ONLY
+                and draft_type == AgentConfigDraftType.DRAFT
+                and draft.account_id is None
+                and agent.active_config_snapshot_id
+                and draft.base_snapshot_id != agent.active_config_snapshot_id
+            ):
+                active_snapshot = cls._get_version_if_present(
+                    session=session,
+                    tenant_id=tenant_id,
+                    agent_id=agent.id,
+                    version_id=agent.active_config_snapshot_id,
+                )
+                if active_snapshot is not None and cls._rebase_workflow_only_normal_draft(
+                    agent=agent,
+                    draft=draft,
+                    snapshot=active_snapshot,
+                    updated_by=agent.updated_by or agent.created_by,
+                ):
+                    session.flush()
            return draft
        base_snapshot = cls._get_version_if_present(
            session=session,
@@ -9,12 +9,13 @@ import sqlalchemy as sa
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import aliased

+from configs import dify_config
 from core.app.entities.app_invoke_entities import InvokeFrom
 from libs.helper import convert_datetime_to_date, escape_like_pattern, to_timestamp
 from models.agent import WorkflowAgentNodeBinding
-from models.enums import MessageStatus
+from models.enums import CreatorUserRole, MessageStatus
 from models.model import App, Conversation, Message
-from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
+from models.workflow import WorkflowNodeExecutionModel, WorkflowRun, WorkflowType


@dataclass(frozen=True)
@@ -580,9 +581,26 @@ class AgentObservabilityService:

    def _load_daily_statistics(
        self, *, app: App, agent_id: str, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
+    ) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        if source_filter.kind in {"all", "webapp"}:
+            rows.extend(self._load_webapp_daily_statistics(app=app, params=params, source_filter=source_filter))
+        if source_filter.kind in {"all", "workflow"}:
+            rows.extend(
+                self._load_workflow_daily_statistics(
+                    app=app,
+                    agent_id=agent_id,
+                    params=params,
+                    source_filter=source_filter,
+                )
+            )
+        return self._merge_daily_statistics(rows)
+
+    def _load_webapp_daily_statistics(
+        self, *, app: App, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
    ) -> list[dict[str, Any]]:
        converted_created_at = convert_datetime_to_date("m.created_at")
-        message_scope = self._statistics_message_scope_sql(source_filter)
+        message_scope = self._statistics_webapp_message_scope_sql(source_filter)
        sql_query = f"""SELECT
    {converted_created_at} AS date,
    COUNT(m.id) AS message_count,
@@ -602,12 +620,154 @@ WHERE
        args: dict[str, Any] = {
            "tz": params.timezone,
            "app_id": app.id,
-            "tenant_id": app.tenant_id,
-            "agent_id": agent_id,
-            "debugger": InvokeFrom.DEBUGGER,
        }
        if source_filter.invoke_from is not None:
            args["source"] = source_filter.invoke_from
+        if params.start:
+            sql_query += " AND m.created_at >= :start"
+            args["start"] = params.start
+        if params.end:
+            sql_query += " AND m.created_at < :end"
+            args["end"] = params.end
+        sql_query += " GROUP BY date ORDER BY date"
+
+        return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]
+
+    @staticmethod
+    def _statistics_webapp_message_scope_sql(source_filter: AgentSourceFilter) -> str:
+        app_scope = "m.app_id = :app_id"
+        if source_filter.invoke_from is not None:
+            app_scope += " AND m.invoke_from = :source"
+        return app_scope
+
+    def _load_workflow_daily_statistics(
+        self,
+        *,
+        app: App,
+        agent_id: str,
+        params: AgentStatisticsQueryParams,
+        source_filter: AgentSourceFilter,
+    ) -> list[dict[str, Any]]:
+        converted_run_created_at = convert_datetime_to_date("aru.created_at")
+        total_tokens = self._workflow_execution_metadata_numeric_sql(("total_tokens",), "BIGINT")
+        nested_total_tokens = self._workflow_execution_metadata_numeric_sql(
+            ("agent_log", "agent_backend", "usage", "total_tokens"), "BIGINT"
+        )
+        total_price = self._workflow_execution_metadata_numeric_sql(("total_price",), "DECIMAL(65, 30)")
+        nested_total_price = self._workflow_execution_metadata_numeric_sql(
+            ("agent_log", "agent_backend", "usage", "total_price"), "DECIMAL(65, 30)"
+        )
+        completion_tokens = self._workflow_execution_metadata_numeric_sql(
+            ("agent_log", "agent_backend", "usage", "completion_tokens"), "BIGINT"
+        )
+        binding_filters = self._statistics_workflow_binding_filters_sql(source_filter)
+        run_date_filters = ""
+        args: dict[str, Any] = {
+            "tz": params.timezone,
+            "tenant_id": app.tenant_id,
+            "agent_id": agent_id,
+            "chat_workflow_type": WorkflowType.CHAT,
+            "end_user_role": CreatorUserRole.END_USER,
+        }
+        if source_filter.app_id:
+            args["source_app_id"] = source_filter.app_id
+        if source_filter.workflow_id:
+            args["workflow_id"] = source_filter.workflow_id
+        if source_filter.workflow_version:
+            args["workflow_version"] = source_filter.workflow_version
+        if source_filter.node_id:
+            args["node_id"] = source_filter.node_id
+        if params.start:
+            run_date_filters += " AND wr.created_at >= :start"
+            args["start"] = params.start
+        if params.end:
+            run_date_filters += " AND wr.created_at < :end"
+            args["end"] = params.end
+
+        run_query = f"""WITH agent_run_usage AS (
+    SELECT
+        wr.id,
+        wr.created_by_role,
+        wr.created_by,
+        wr.created_at,
+        COALESCE(SUM(COALESCE({total_tokens}, {nested_total_tokens}, 0)), 0) AS token_count,
+        COALESCE(SUM(COALESCE({total_price}, {nested_total_price}, 0)), 0) AS total_price,
+        COALESCE(SUM(COALESCE(wne.elapsed_time, 0)), 0) AS latency,
+        COALESCE(SUM(COALESCE({completion_tokens}, 0)), 0) AS answer_tokens
+    FROM workflow_runs wr
+    JOIN workflow_agent_node_bindings wanb
+        ON wanb.tenant_id = :tenant_id
+        AND wanb.agent_id = :agent_id
+        AND wanb.app_id = wr.app_id
+        AND wanb.workflow_id = wr.workflow_id
+        AND wanb.workflow_version = wr.version
+        {binding_filters}
+    JOIN workflow_node_executions wne
+        ON wne.workflow_run_id = wr.id
+        AND wne.node_id = wanb.node_id
+    WHERE wr.type != :chat_workflow_type{run_date_filters}
+    GROUP BY wr.id, wr.created_by_role, wr.created_by, wr.created_at
+)
+SELECT
+    {converted_run_created_at} AS date,
+    COUNT(aru.id) AS message_count,
+    COUNT(aru.id) AS conversation_count,
+    COUNT(DISTINCT CASE
+        WHEN aru.created_by_role = :end_user_role THEN aru.created_by
+        ELSE NULL
+    END) AS end_user_count,
+    COALESCE(SUM(aru.token_count), 0) AS token_count,
+    COALESCE(SUM(aru.total_price), 0) AS total_price,
+    COALESCE(AVG(aru.latency), 0) AS avg_latency,
+    COALESCE(SUM(aru.latency), 0) AS latency_sum,
+    COALESCE(SUM(aru.answer_tokens), 0) AS answer_tokens,
+    0 AS like_count
+FROM agent_run_usage aru
+GROUP BY date
+ORDER BY date"""
+        rows = [dict(row._mapping) for row in self._session.execute(sa.text(run_query), args).all()]
+        rows.extend(
+            self._load_workflow_chat_daily_context(
+                app=app,
+                agent_id=agent_id,
+                params=params,
+                source_filter=source_filter,
+            )
+        )
+        return self._merge_daily_statistics(rows)
+
+    def _load_workflow_chat_daily_context(
+        self,
+        *,
+        app: App,
+        agent_id: str,
+        params: AgentStatisticsQueryParams,
+        source_filter: AgentSourceFilter,
+    ) -> list[dict[str, Any]]:
+        converted_created_at = convert_datetime_to_date("m.created_at")
+        workflow_scope = self._statistics_workflow_message_scope_sql(source_filter)
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
+    COUNT(m.id) AS message_count,
+    COUNT(DISTINCT m.conversation_id) AS conversation_count,
+    COUNT(DISTINCT m.from_end_user_id) AS end_user_count,
+    COALESCE(SUM(COALESCE(m.message_tokens, 0) + COALESCE(m.answer_tokens, 0)), 0) AS token_count,
+    COALESCE(SUM(COALESCE(m.total_price, 0)), 0) AS total_price,
+    COALESCE(AVG(m.provider_response_latency), 0) AS avg_latency,
+    COALESCE(SUM(m.provider_response_latency), 0) AS latency_sum,
+    COALESCE(SUM(m.answer_tokens), 0) AS answer_tokens,
+    COUNT(mf.id) AS like_count
+FROM messages m
+LEFT JOIN message_feedbacks mf
+    ON mf.message_id = m.id AND mf.rating = 'like'
+WHERE
+    {workflow_scope}"""
+        args: dict[str, Any] = {
+            "tz": params.timezone,
+            "tenant_id": app.tenant_id,
+            "agent_id": agent_id,
+            "chat_workflow_type": WorkflowType.CHAT,
+        }
        if source_filter.app_id:
            args["source_app_id"] = source_filter.app_id
        if source_filter.workflow_id:
@@ -627,10 +787,7 @@ WHERE
        return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]

    @staticmethod
-    def _statistics_message_scope_sql(source_filter: AgentSourceFilter) -> str:
-        app_scope = "m.app_id = :app_id"
-        if source_filter.invoke_from is not None:
-            app_scope += " AND m.invoke_from = :source"
+    def _statistics_workflow_binding_filters_sql(source_filter: AgentSourceFilter) -> str:
        workflow_binding_filters = []
        if source_filter.app_id:
            workflow_binding_filters.append("wanb.app_id = :source_app_id")
@@ -640,8 +797,12 @@ WHERE
            workflow_binding_filters.append("wanb.workflow_version = :workflow_version")
        if source_filter.node_id:
            workflow_binding_filters.append("wanb.node_id = :node_id")
-        extra_workflow_filters = f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
-        workflow_scope = f"""m.workflow_run_id IS NOT NULL
+        return f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
+
+    @classmethod
+    def _statistics_workflow_message_scope_sql(cls, source_filter: AgentSourceFilter) -> str:
+        binding_filters = cls._statistics_workflow_binding_filters_sql(source_filter)
+        return f"""m.workflow_run_id IS NOT NULL
        AND EXISTS (
            SELECT 1
            FROM workflow_runs wr
@@ -651,17 +812,65 @@ WHERE
                AND wanb.app_id = wr.app_id
                AND wanb.workflow_id = wr.workflow_id
                AND wanb.workflow_version = wr.version
-                {extra_workflow_filters}
+                {binding_filters}
            JOIN workflow_node_executions wne
                ON wne.workflow_run_id = wr.id
                AND wne.node_id = wanb.node_id
            WHERE wr.id = m.workflow_run_id
+                AND wr.type = :chat_workflow_type
        )"""
-        if source_filter.kind == "webapp":
-            return app_scope
-        if source_filter.kind == "workflow":
-            return workflow_scope
-        return f"(({app_scope}) OR ({workflow_scope}))"
+
+    @staticmethod
+    def _workflow_execution_metadata_numeric_sql(path: tuple[str, ...], numeric_type: str) -> str:
+        if dify_config.DB_TYPE == "postgresql":
+            json_path = ",".join(path)
+            value = f"CAST(wne.execution_metadata AS JSONB) #>> '{{{json_path}}}'"
+            return f"CAST(NULLIF({value}, '') AS {numeric_type})"
+        if dify_config.DB_TYPE in {"mysql", "oceanbase", "seekdb"}:
+            json_path = "$." + ".".join(path)
+            mysql_numeric_type = "UNSIGNED" if numeric_type == "BIGINT" else numeric_type
+            value = f"JSON_UNQUOTE(JSON_EXTRACT(wne.execution_metadata, '{json_path}'))"
+            return f"CAST(NULLIF(NULLIF({value}, ''), 'null') AS {mysql_numeric_type})"
+        raise NotImplementedError(f"Unsupported database type: {dify_config.DB_TYPE}")
+
+    @staticmethod
+    def _merge_daily_statistics(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        merged: dict[Any, dict[str, Any]] = {}
+        weighted_latency: dict[Any, float] = {}
+        for row in rows:
+            date = row["date"]
+            target = merged.setdefault(
+                date,
+                {
+                    "date": date,
+                    "message_count": 0,
+                    "conversation_count": 0,
+                    "end_user_count": 0,
+                    "token_count": 0,
+                    "total_price": Decimal(0),
+                    "avg_latency": 0.0,
+                    "latency_sum": 0.0,
+                    "answer_tokens": 0,
+                    "like_count": 0,
+                },
+            )
+            message_count = int(row.get("message_count") or 0)
+            target["message_count"] += message_count
+            target["conversation_count"] += int(row.get("conversation_count") or 0)
+            target["end_user_count"] += int(row.get("end_user_count") or 0)
+            target["token_count"] += int(row.get("token_count") or 0)
+            target["total_price"] += Decimal(str(row.get("total_price") or 0))
+            target["latency_sum"] += float(row.get("latency_sum") or 0)
+            target["answer_tokens"] += int(row.get("answer_tokens") or 0)
+            target["like_count"] += int(row.get("like_count") or 0)
+            weighted_latency[date] = (
+                weighted_latency.get(date, 0.0) + float(row.get("avg_latency") or 0) * message_count
+            )
+
+        for date, row in merged.items():
+            message_count = int(row["message_count"])
+            row["avg_latency"] = weighted_latency[date] / message_count if message_count else 0.0
+        return sorted(merged.values(), key=lambda row: str(row["date"]))

    @staticmethod
    def _build_charts(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
--- a/Show More
+++ b/Show More