refactor(dataset): validate declared KnowledgeFS operations

Fixes WTA-1411

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

@@ -32,11 +32,12 @@ 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.
-- 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.
+- 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
 - 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
 
@@ -65,7 +66,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 the package-required static checks documented in `e2e/AGENTS.md`.
+   - Run `vpr lint --fix --quiet` from the repository root and `pnpm -C e2e type-check`.
    - Broaden verification only when the change affects hooks, tags, setup, or shared step semantics.
 
 ## Review Checklist
@@ -76,8 +77,6 @@ Keep this skill focused on Cucumber, Playwright, and package-level E2E guidance.
 - Are locators user-facing and assertions web-first?
 - Does the change introduce hidden coupling across scenarios, tags, or instance state?
 - Does it document or implement behavior that differs from the real hooks or configuration?
-- Does setup/cleanup use the generated client directly, with any remaining helper owning more than a one-to-one endpoint forward?
-- Is every raw HTTP call a documented protocol or infrastructure exception rather than an ordinary Console operation?
 
 Lead findings with correctness, flake risk, and architecture drift.
 

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

@@ -9,11 +9,12 @@ 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 provides an execution checklist and must not redefine or extend that policy.
+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.
 
 ## 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.
@@ -32,4 +33,4 @@ vp test run path/to/spec-or-directory
 vp test run --project unit src/path/to/spec
 ```
 
-Run Dify UI Storybook tests with `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.
+For styled Dify UI behavior, run `vp test --project storybook --run`. Run broader checks only after the focused behavior passes.

.github/CODEOWNERS

@@ -8,6 +8,7 @@
 
 # Lint bulk suppression baselines.
 /oxlint-suppressions.json
+/eslint-suppressions.json
 
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola
@@ -32,9 +33,31 @@
 # 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
@@ -88,6 +111,7 @@
 /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
@@ -112,11 +136,11 @@
 /api/controllers/console/billing/ @hj24 @zyssyz123
 
 # Backend - Enterprise
-/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
+/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
 
 # Backend - Database Migrations
 /api/migrations/ @snakevash @laipz8200 @MRZHUH
@@ -129,6 +153,7 @@
 
 # Frontend - Platform and Features
 /web/config/ @lyzno1
+/web/contract/ @lyzno1
 /web/env.ts @lyzno1
 /web/features/ @lyzno1
 /web/hooks/ @lyzno1
@@ -187,6 +212,7 @@
 /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
@@ -205,22 +231,22 @@
 /web/app/components/plugins/marketplace/ @iamjoel @Yessenia-d
 
 # Frontend - Login and Registration
-/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
+/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
 
 # Frontend - Service Authentication
-/web/service/base.ts @iamjoel
+/web/service/base.ts @douxc @iamjoel
 
 # Frontend - WebApp Authentication and Access Control
-/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
+/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
 
 # Frontend - Explore Page
 /web/app/components/explore/ @CodingOnStar @iamjoel
@@ -239,6 +265,7 @@
 /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
@@ -250,22 +277,5 @@
 # Frontend - Workspace
 /web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh
 
-# Frontend - App Shell and Console Bootstrap
-/web/app/layout.tsx @iamjoel @lyzno1
-/web/app/error.tsx @iamjoel @lyzno1
-/web/app/(commonLayout)/layout.tsx @iamjoel @lyzno1
-/web/app/(commonLayout)/providers.tsx @iamjoel @lyzno1
-/web/app/(commonLayout)/hydration-boundary.tsx @iamjoel @lyzno1
-/web/app/(commonLayout)/profile-bootstrap-gate.tsx @iamjoel @lyzno1
-/web/app/(commonLayout)/error.tsx @iamjoel @lyzno1
-/web/app/account/(commonLayout)/layout.tsx @iamjoel @lyzno1
-/web/app/components/main-nav/* @iamjoel @lyzno1
-/web/app/components/main-nav/components/* @iamjoel @lyzno1
-/web/context/query-client.tsx @iamjoel @lyzno1
-/web/context/query-client-server.ts @iamjoel @lyzno1
-/web/proxy.ts @iamjoel @lyzno1
-/web/app/auth/refresh/route.ts @iamjoel @lyzno1
-/web/service/server.ts @iamjoel @lyzno1
-
 # Docker
 /docker/* @laipz8200

.github/workflows/api-tests.yml

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

.github/workflows/deploy-knowledge.yml

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

.github/workflows/main-ci.yml

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

.github/workflows/post-merge.yml

@@ -26,30 +26,19 @@ 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/**'
@@ -59,13 +48,8 @@ jobs:
               - 'api/services/plugin/**'
               - 'api/core/tools/**'
               - 'api/services/tools/**'
-              - 'packages/contracts/package.json'
               - 'packages/contracts/generated/api/console/agent/**'
-              - 'packages/contracts/generated/api/console/apps/**'
-              - 'packages/contracts/generated/api/console/datasets/**'
               - 'packages/contracts/generated/api/console/orpc.gen.ts'
-              - 'packages/contracts/generated/api/console/workspaces/**'
-              - 'packages/contracts/generated/api/service/**'
               - 'web/features/agent-v2/**'
               - 'web/app/(commonLayout)/agents/**'
               - 'web/app/(commonLayout)/@detailSidebar/agents/**'

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

@@ -27,7 +27,7 @@ jobs:
           persist-credentials: false
 
       - name: Setup Go
-        uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.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@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.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@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a # v6.5.0
+        uses: golangci/golangci-lint-action@4afd733a84b1f43292c63897423277bb7f4313a9 # 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@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+        uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
         with:
           go-version-file: dify-agent-runtime/go.mod
           cache-dependency-path: dify-agent-runtime/go.sum

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

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

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

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

.github/workflows/web-e2e.yml

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

.github/workflows/web-tests.yml

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

Makefile

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

README.md

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

api/.env.example

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

api/clients/agent_backend/client.py

@@ -8,7 +8,7 @@ creating another wire contract.
 
 from __future__ import annotations
 
-from collections.abc import Callable, Iterator
+from collections.abc import Iterator
 from typing import Protocol
 
 from dify_agent.client import (
@@ -45,13 +45,7 @@ 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,
-        should_stop: Callable[[], bool] | None = None,
-    ) -> Iterator[RunEvent]:
+    def stream_events(self, run_id: str, *, after: str | 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:
@@ -67,15 +61,7 @@ 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,
-        max_reconnects: int | None = None,
-        timeout_seconds: float | None = None,
-        should_stop: Callable[[], bool] | None = None,
-    ) -> Iterator[RunEvent]:
+    def stream_events_sync(self, run_id: str, *, after: str | None = None) -> Iterator[RunEvent]:
         """Stream run events synchronously."""
 
     def wait_run_sync(self, run_id: str, *, timeout_seconds: float | None = None) -> RunStatusResponse:
@@ -87,16 +73,8 @@ class DifyAgentBackendRunClient:
 
     client: _DifyAgentSyncClient
 
-    def __init__(
-        self,
-        client: _DifyAgentSyncClient,
-        *,
-        stream_max_reconnects: int = 3,
-        stream_timeout_seconds: float = 1200,
-    ) -> None:
+    def __init__(self, client: _DifyAgentSyncClient) -> 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."""
@@ -112,22 +90,10 @@ 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,
-        should_stop: Callable[[], bool] | None = None,
-    ) -> Iterator[RunEvent]:
+    def stream_events(self, run_id: str, *, after: str | 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,
-                max_reconnects=self._stream_max_reconnects,
-                timeout_seconds=self._stream_timeout_seconds,
-                should_stop=should_stop,
-            )
+            yield from self.client.stream_events_sync(run_id, after=after)
         except Exception as exc:
             raise _normalize_dify_agent_error(exc) from exc
 

api/clients/agent_backend/factory.py

@@ -13,17 +13,10 @@ 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, stream_timeout=stream_read_timeout_seconds),
-        stream_max_reconnects=stream_max_reconnects,
-        stream_timeout_seconds=stream_run_timeout_seconds,
-    )
+    return DifyAgentBackendRunClient(Client(base_url=base_url))

api/clients/agent_backend/fake_client.py

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

api/commands/retention.py

@@ -1,8 +1,6 @@
 import datetime
 import logging
-import re
 import time
-import uuid
 from collections.abc import Callable
 from typing import TypedDict
 
@@ -23,7 +21,6 @@ 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):
@@ -69,7 +66,6 @@ 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()})
@@ -78,27 +74,6 @@ 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,
@@ -835,11 +810,9 @@ 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, *, dry_run: bool) -> None:
+def _echo_bundle_archive_operation_summary(summary) -> 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}. "
@@ -848,12 +821,10 @@ def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> 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} {cursor_label}={cursor_value or 'none'}",
+            f"bytes_per_second={summary.bytes_per_second:.2f}",
             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",
@@ -871,8 +842,7 @@ def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> 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"catalog_id={result.catalog_id} time={result.elapsed_time:.2f}s "
-                    f"validation={result.validation_time:.2f}s",
+                    f"time={result.elapsed_time:.2f}s validation={result.validation_time:.2f}s",
                     fg="white",
                 )
             )
@@ -880,7 +850,7 @@ def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> None:
             click.echo(
                 click.style(
                     f"  failed bundle={result.bundle_id} tenant={result.tenant_id} "
-                    f"catalog_id={result.catalog_id} object_prefix={result.object_prefix} error={result.error}",
+                    f"object_prefix={result.object_prefix} error={result.error}",
                     fg="red",
                 )
             )
@@ -897,24 +867,25 @@ def _echo_bundle_archive_operation_summary(summary, *, dry_run: bool) -> None:
 )
 @click.option("--run-id", required=False, help="Workflow run ID to restore.")
 @click.option(
-    "--target-month",
-    metavar="YYYY-MM",
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
     default=None,
-    help="V2 catalog month to restore; required unless --run-id is used.",
+    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
 )
 @click.option(
-    "--after-catalog-id",
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
     default=None,
-    help="Exclusive V2 cursor from the same restore month and tenant scope.",
+    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
 )
 @click.option("--workers", default=1, show_default=True, type=int, help="V1 --run-id compatibility only.")
-@click.option("--limit", type=click.IntRange(min=1), default=100, show_default=True, help="Maximum V2 catalog rows.")
+@click.option("--limit", type=int, default=100, show_default=True, help="Maximum number of V2 bundles to restore.")
 @click.option("--dry-run", is_flag=True, help="Preview without restoring.")
 def restore_workflow_runs(
     tenant_ids: str | None,
     run_id: str | None,
-    target_month: str | None,
-    after_catalog_id: str | None,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
     workers: int,
     limit: int,
     dry_run: bool,
@@ -934,20 +905,23 @@ 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 = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
+    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")
 
+    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 {target_desc} at {start_time.isoformat()}.",
+            f"Starting restore of workflow run {run_id} at {start_time.isoformat()}.",
             fg="white",
         )
     )
@@ -981,20 +955,17 @@ def restore_workflow_runs(
         click.echo(
             click.style("--workers is ignored for V2 bundle restore; bundles are processed serially.", fg="yellow")
         )
-    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)
+    assert start_from is not None
+    assert end_before is not None
     bundle_restorer = WorkflowRunBundleArchiveMaintenance(dry_run=dry_run, strict_content_validation=True)
     summary = bundle_restorer.restore_batch(
         tenant_ids=parsed_tenant_ids,
-        target_year=target_year,
-        target_month=target_month_number,
-        after_catalog_id=catalog_cursor,
+        start_date=start_from,
+        end_date=end_before,
         limit=limit,
     )
-    _echo_bundle_archive_operation_summary(summary, dry_run=dry_run)
-    if summary.bundles_failed:
-        raise click.exceptions.Exit(1)
+    _echo_bundle_archive_operation_summary(summary)
+    return
 
 
 @click.command(
@@ -1008,41 +979,23 @@ def restore_workflow_runs(
 )
 @click.option("--run-id", required=False, help="Workflow run ID to delete.")
 @click.option(
-    "--target-month",
-    metavar="YYYY-MM",
+    "--start-from",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
     default=None,
-    help="V2 catalog month to delete; required unless --run-id is used.",
+    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
 )
 @click.option(
-    "--after-catalog-id",
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
     default=None,
-    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.",
+    help="Optional upper bound (exclusive) for created_at; must be paired with --start-from.",
 )
+@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="V1 --run-id only: continue when one archive object fails validation.",
+    help="Continue batch deletion when one archive object fails validation.",
 )
 @click.option(
     "--restore-sample-interval",
@@ -1054,11 +1007,8 @@ def restore_workflow_runs(
 def delete_archived_workflow_runs(
     tenant_ids: str | None,
     run_id: str | None,
-    target_month: str | None,
-    after_catalog_id: str | None,
-    run_shard_index: int | None,
-    run_shard_total: int | None,
-    all_pages: bool,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime | None,
     limit: int,
     dry_run: bool,
     skip_bad_archives: bool,
@@ -1068,38 +1018,26 @@ 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. 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.
+    counts, and source/archive content checksums before deleting source rows. `--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 = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
+    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")
 
+    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 f"workflow archive catalog month {target_month}"
+    target_desc = f"workflow run {run_id}" if run_id else "workflow runs"
     click.echo(
         click.style(
             f"Starting delete of {target_desc} at {start_time.isoformat()}.",
@@ -1172,104 +1110,20 @@ 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 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
+    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,
     )
-    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",
-            )
-        )
+    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)
 
 
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:

api/commands/system.py

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

api/configs/app_config.py

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

api/configs/enterprise/__init__.py

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

api/configs/extra/__init__.py

@@ -1,6 +1,5 @@
 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
 
@@ -9,7 +8,6 @@ class ExtraServiceConfig(
     # place the configs in alphabet order
     AgentBackendConfig,
     ArchiveStorageConfig,
-    KnowledgeFSConfig,
     NotionConfig,
     SentryConfig,
 ):

api/configs/extra/agent_backend_config.py

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

api/configs/extra/knowledge_fs_config.py

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

api/configs/feature/__init__.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import timedelta
 from enum import StrEnum
 from typing import Literal
 
@@ -11,7 +11,6 @@ from pydantic import (
     PositiveFloat,
     PositiveInt,
     computed_field,
-    field_validator,
 )
 from pydantic_settings import BaseSettings
 
@@ -281,27 +280,6 @@ 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()]
@@ -1160,16 +1138,6 @@ 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):
     """
@@ -1497,11 +1465,6 @@ class LoginConfig(BaseSettings):
 
 
 class AccountConfig(BaseSettings):
-    ENABLE_CHANGE_EMAIL: bool = Field(
-        description="whether users can change their email address",
-        default=True,
-    )
-
     ACCOUNT_DELETION_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
         description="Duration in minutes for which a account deletion token remains valid",
         default=5,

api/controllers/common/wraps.py

@@ -10,7 +10,6 @@ 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"]
@@ -52,7 +51,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, tenant_id, path_args)
+        resource_id = _extract_resource_id(resource_type, path_args)
         if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
             return
     allowed = RBACService.CheckAccess.check(
@@ -132,14 +131,11 @@ def _is_resource_owned_by_current_user(
     return False
 
 
-def _extract_resource_id(
-    resource_type: RBACResourceScope, tenant_id: str, path_args: dict[str, object] | None = None
-) -> str:
+def _extract_resource_id(resource_type: RBACResourceScope, 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 routes carry ``agent_id``, which is
-    resolved to the App backing that Agent.
+    app/dataset resources, and Agent App routes use ``agent_id`` as the app id.
     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``.
@@ -150,19 +146,10 @@ def _extract_resource_id(
     matched_args = {**view_args, **(path_args or {})}
 
     if resource_type == RBACResourceScope.APP:
-        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")
+        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)
 
     if resource_type == RBACResourceScope.DATASET:
         dataset_id = matched_args.get("dataset_id") or matched_args.get("resource_id")

api/controllers/console/__init__.py

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

api/controllers/console/agent/composer.py

@@ -230,7 +230,6 @@ 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
@@ -440,7 +439,6 @@ 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
@@ -480,7 +478,6 @@ class AgentComposerApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session

api/controllers/console/agent/roster.py

@@ -62,7 +62,7 @@ from libs.datetime_utils import parse_time_range
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
-from models.agent import Agent, AgentConfigDraftType, AgentStatus
+from models.agent import Agent, AgentStatus
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import ApiTokenType
 from models.model import ApiToken, App, IconType
@@ -266,13 +266,6 @@ 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")
 
@@ -316,7 +309,6 @@ register_schema_models(
     AgentAppCopyPayload,
     AgentPublishPayload,
     AgentBuildDraftCheckoutPayload,
-    AgentDebugConversationRefreshPayload,
     ComposerSavePayload,
     AgentApiStatusPayload,
     AgentInviteOptionsQuery,
@@ -400,7 +392,6 @@ 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(
@@ -448,7 +439,6 @@ 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,
@@ -539,23 +529,9 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:
 
 
 def _query_values(name: str, alias_name: str | None = None) -> list[str]:
-    def _get_values(field_name: str) -> list[str]:
-        values = request.args.getlist(field_name)
-        indexed_values: list[tuple[int, list[str]]] = []
-        prefix = f"{field_name}["
-        for key in request.args:
-            if not key.startswith(prefix) or not key.endswith("]"):
-                continue
-            index = key[len(prefix) : -1]
-            if index.isdigit():
-                indexed_values.append((int(index), request.args.getlist(key)))
-        for _, items in sorted(indexed_values):
-            values.extend(items)
-        return values
-
-    values = _get_values(name)
+    values = request.args.getlist(name)
     if alias_name:
-        values.extend(_get_values(alias_name))
+        values.extend(request.args.getlist(alias_name))
     return [value.strip() for value in values if value.strip()]
 
 
@@ -566,7 +542,6 @@ 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
@@ -604,7 +579,6 @@ 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
@@ -646,7 +620,6 @@ 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
@@ -672,7 +645,6 @@ 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):
@@ -683,16 +655,6 @@ 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",
@@ -707,12 +669,10 @@ 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,
@@ -730,7 +690,6 @@ 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
@@ -753,7 +712,6 @@ 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
@@ -771,7 +729,6 @@ 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
@@ -830,7 +787,6 @@ 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
@@ -853,7 +809,6 @@ 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
@@ -879,7 +834,6 @@ 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):
@@ -896,7 +850,6 @@ 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
@@ -915,7 +868,6 @@ 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]:
@@ -926,7 +878,6 @@ 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]:
@@ -946,7 +897,6 @@ 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(
@@ -992,7 +942,6 @@ 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)
@@ -1031,7 +980,6 @@ 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)
@@ -1070,7 +1018,6 @@ 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)
@@ -1091,7 +1038,6 @@ 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)
@@ -1117,7 +1063,6 @@ 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):
@@ -1133,7 +1078,6 @@ 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):
@@ -1154,7 +1098,6 @@ class AgentRosterVersionRestoreApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session

api/controllers/console/apikey.py

@@ -12,7 +12,6 @@ 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
@@ -195,7 +194,6 @@ 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"""
@@ -212,7 +210,6 @@ 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"""
@@ -236,7 +233,6 @@ class AppApiKeyResource(BaseApiKeyResource):
     @with_current_user
     @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
-    @agent_manage_required_for_agent_app
     @with_session
     def delete(
         self,

api/controllers/console/app/app.py

@@ -9,7 +9,7 @@ from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, ValidationInfo, computed_field, field_validator, model_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, Forbidden, NotFound
+from werkzeug.exceptions import BadRequest, 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 agent_manage_required_for_agent_app, get_app_model, with_session
+from controllers.console.app.wraps import get_app_model, with_session
 from controllers.console.workspace.models import LoadBalancingPayload
 from controllers.console.wraps import (
     RBACPermission,
@@ -75,7 +75,6 @@ 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
 
@@ -247,7 +246,7 @@ class ModelConfigPartial(ResponseModel):
         return to_timestamp(value)
 
 
-class AppModelConfigResponse(ResponseModel):
+class ModelConfig(ResponseModel):
     opening_statement: str | None = None
     suggested_questions: Any | None = Field(
         default=None, validation_alias=AliasChoices("suggested_questions_list", "suggested_questions")
@@ -420,7 +419,7 @@ class AppDetail(AppResponseModel):
     icon_background: str | None = None
     enable_site: bool
     enable_api: bool
-    model_config_: AppModelConfigResponse | None = Field(
+    model_config_: ModelConfig | None = Field(
         default=None,
         validation_alias=AliasChoices("app_model_config", "model_config"),
         alias="model_config",
@@ -526,13 +525,7 @@ 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,
-    AppModelConfigResponse,
-    AppDetail,
+    console_ns, RedirectUrlResponse, SimpleResultResponse, AppImportResponse, AppTraceResponse
 )
 
 register_schema_models(
@@ -551,8 +544,10 @@ register_schema_models(
     Tag,
     WorkflowPartial,
     ModelConfigPartial,
+    ModelConfig,
     AppDetailSiteResponse,
     DeletedTool,
+    AppDetail,
     AppExportResponse,
     Segmentation,
     PreProcessingRule,
@@ -828,7 +823,6 @@ 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):
@@ -863,7 +857,6 @@ 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):
@@ -888,7 +881,6 @@ 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)
@@ -900,19 +892,16 @@ 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)
-            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))
+            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,
+            )
             if result.status == ImportStatus.FAILED:
                 session.rollback()
                 return dump_response(AppImportResponse, result), 400
@@ -966,7 +955,6 @@ 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"""
@@ -991,7 +979,6 @@ 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):
@@ -1022,7 +1009,6 @@ 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):
@@ -1050,7 +1036,6 @@ 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):
@@ -1084,7 +1069,6 @@ 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):
@@ -1112,7 +1096,6 @@ class AppApiStatus(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
-    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):

api/controllers/console/app/app_import.py

@@ -1,7 +1,6 @@
 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
@@ -29,7 +28,6 @@ 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
@@ -93,21 +91,18 @@ class AppImportApi(Resource):
             import_service = AppDslService(session)
             # Import app
             account = current_user
-            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))
+            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,
+            )
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:
@@ -162,10 +157,7 @@ class AppImportConfirmApi(Resource):
             import_service = AppDslService(session)
             # Confirm import
             account = current_user
-            try:
-                result = import_service.confirm_import(import_id=import_id, account=account)
-            except NoPermissionError as e:
-                raise Forbidden(str(e))
+            result = import_service.confirm_import(import_id=import_id, account=account)
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:

api/controllers/console/app/completion.py

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

api/controllers/console/app/site.py

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

api/controllers/console/app/workflow.py

@@ -105,7 +105,6 @@ 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,
     )
@@ -359,12 +358,6 @@ class WorkflowPublishResponse(ResponseModel):
     created_at: int
 
 
-class SyncDraftWorkflowResponse(ResponseModel):
-    result: str
-    hash: str
-    updated_at: int
-
-
 class WorkflowRestoreResponse(ResponseModel):
     result: str
     hash: str
@@ -447,7 +440,6 @@ register_response_schema_models(
     WorkflowOnlineUsersByApp,
     WorkflowOnlineUsersResponse,
     WorkflowPublishResponse,
-    SyncDraftWorkflowResponse,
     WorkflowRestoreResponse,
     DefaultBlockConfigsResponse,
     DefaultBlockConfigResponse,
@@ -563,7 +555,14 @@ class DraftWorkflowApi(Resource):
     @console_ns.response(
         200,
         "Draft workflow synced successfully",
-        console_ns.models[SyncDraftWorkflowResponse.__name__],
+        console_ns.model(
+            "SyncDraftWorkflowResponse",
+            {
+                "result": fields.String,
+                "hash": fields.String,
+                "updated_at": fields.String,
+            },
+        ),
     )
     @console_ns.response(400, "Invalid workflow configuration")
     @console_ns.response(403, "Permission denied")
@@ -611,21 +610,17 @@ 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 dump_response(
-            SyncDraftWorkflowResponse,
-            {
-                "result": "success",
-                "hash": workflow.unique_hash,
-                "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-            },
-        )
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
 
 
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/run")

api/controllers/console/app/wraps.py

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

api/controllers/console/auth/activate.py

@@ -7,13 +7,10 @@ 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
@@ -139,12 +136,6 @@ 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
@@ -155,11 +146,6 @@ class ActivateApi(Resource):
             raise AlreadyActivateError()
 
         account = invitation["account"]
-        if extract_access_token(request):
-            current_account, _ = current_account_with_tenant()
-            if current_account.id != account.id:
-                raise InvitationAccountMismatchError()
-
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(account.email):
             raise AccountInFreezeError()
 

api/controllers/console/auth/error.py

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

api/controllers/console/auth/login.py

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

api/controllers/console/auth/oauth.py

@@ -6,7 +6,6 @@ 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
@@ -128,20 +127,6 @@ 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")
@@ -210,26 +195,16 @@ 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_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}")
+            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.")
 
-            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)
+            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin/invite-settings?invite_token={invite_token}")
 
         try:
             account, oauth_new_user = _generate_account(provider, user_info, timezone=timezone, language=language)
@@ -264,10 +239,21 @@ 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()}"
-        return _redirect_with_console_session(account, target_url)
+        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
 
 
 def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) -> Account | None:

api/controllers/console/auth/oauth_server.py

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

api/controllers/console/datasets/datasets.py

@@ -607,13 +607,6 @@ 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,
@@ -882,7 +875,7 @@ class DatasetIndexingEstimateApi(Resource):
                 file_details = session.scalars(
                     select(UploadFile).where(UploadFile.tenant_id == current_tenant_id, UploadFile.id.in_(file_ids))
                 ).all()
-                if not file_details:
+                if file_details is None:
                     raise NotFound("File not found.")
 
                 if file_details:

api/controllers/console/explore/trial.py

@@ -47,9 +47,7 @@ from controllers.console.explore.error import (
     NotWorkflowAppError,
 )
 from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-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.console.wraps import 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
@@ -63,13 +61,12 @@ 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, App
+from models import Account
 from models.account import TenantStatus
 from models.model import AppMode, Site, load_annotation_reply_config
 from models.workflow import Workflow
@@ -431,36 +428,6 @@ 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__])
@@ -645,7 +612,7 @@ class TrialChatAudioApi(TrialAppResource):
     def post(self, current_user: Account, trial_app):
         app_model = trial_app
 
-        file = request.files.get("file")
+        file = request.files["file"]
 
         try:
             # Get IDs before they might be detached from session
@@ -920,18 +887,6 @@ class DatasetListApi(Resource):
 
 console_ns.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
 
-console_ns.add_resource(
-    TrialAppFileUploadApi,
-    "/trial-apps/<uuid:app_id>/files/upload",
-    endpoint="trial_app_file_upload",
-)
-
-console_ns.add_resource(
-    TrialAppRemoteFileUploadApi,
-    "/trial-apps/<uuid:app_id>/remote-files/upload",
-    endpoint="trial_app_remote_file_upload",
-)
-
 console_ns.add_resource(
     TrialMessageSuggestedQuestionApi,
     "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",

api/controllers/console/feature.py

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

api/controllers/console/files.py

@@ -29,7 +29,7 @@ from extensions.ext_database import db
 from fields.file_fields import FileResponse, UploadConfig
 from libs.helper import dump_response
 from libs.login import login_required
-from models import Account, UploadFile
+from models import Account
 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,43 +56,6 @@ 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
@@ -118,11 +81,42 @@ 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):
-        upload_file = upload_file_from_request(current_user=current_user)
+        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)
 
         return dump_response(FileResponse, upload_file), 201
 

api/controllers/console/init_validate.py

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

api/controllers/console/knowledge_fs_proxy.py

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

api/controllers/console/onboarding.py

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

api/controllers/console/remote_files.py

@@ -46,61 +46,6 @@ 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__])
@@ -108,8 +53,53 @@ class RemoteFileUpload(Resource):
     @login_required
     @with_current_user
     def post(self, current_user: Account):
-        remote_file = upload_remote_file_from_request(current_user=current_user)
+        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
         return (
-            remote_file.model_dump(mode="json"),
+            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"),
             201,
         )

api/controllers/console/setup.py

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

api/controllers/console/socketio/workflow.py

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

api/controllers/console/workflow_run_archive.py

@@ -4,16 +4,20 @@ from http import HTTPStatus
 from flask import redirect
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from werkzeug.exceptions import Conflict, Forbidden, NotFound
+from werkzeug.exceptions import Conflict, 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
@@ -21,7 +25,6 @@ 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,
@@ -95,13 +98,11 @@ register_response_schema_models(
 )
 
 
-def _current_owner_or_admin_ids() -> tuple[str, str]:
-    """Return current Cloud workspace IDs for an owner or admin, independently of enterprise RBAC."""
+def _current_ids() -> tuple[str, str]:
+    """Return current `(tenant_id, account_id)` or raise when no workspace is selected."""
     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
 
 
@@ -123,8 +124,12 @@ 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_owner_or_admin_ids()
+        tenant_id, _ = _current_ids()
         return dump_response(WorkflowRunArchiveListResponse, list_workflow_run_archives(db.session(), tenant_id))
 
 
@@ -144,8 +149,12 @@ 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_owner_or_admin_ids()
+        tenant_id, account_id = _current_ids()
         payload = WorkflowRunArchiveDownloadPayload.model_validate(console_ns.payload or {})
         try:
             task = create_workflow_run_archive_download_task(
@@ -171,8 +180,12 @@ 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_owner_or_admin_ids()
+        tenant_id, _ = _current_ids()
         try:
             task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
         except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
@@ -196,8 +209,12 @@ 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_owner_or_admin_ids()
+        tenant_id, _ = _current_ids()
         try:
             task = get_ready_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
         except WorkflowRunArchiveDownloadTaskNotFoundError as exc:

api/controllers/console/workspace/account.py

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

api/controllers/console/workspace/members.py

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

api/controllers/console/workspace/plugin.py

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

api/controllers/console/workspace/workspace.py

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

api/controllers/console/wraps.py

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

api/controllers/openapi/auth/data.py

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

api/controllers/service_api/app/audio.py

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

api/controllers/service_api/dataset/dataset.py

@@ -531,22 +531,14 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        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),
-                )
+        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)
 
         return _dump_service_dataset_detail(dataset, session=session), 200
 

api/controllers/web/audio.py

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

api/controllers/web/site.py

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

api/controllers/web/wraps.py

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

api/core/agent/publish_visibility.py

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

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

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

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

@@ -540,9 +540,6 @@ 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(),
@@ -682,27 +679,42 @@ 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 == AgentConfigDraftType.DEBUG_BUILD,
-            AgentConfigDraft.account_id == account_id,
+            AgentConfigDraft.draft_type == effective_draft_type,
         )
+        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
-        raise AgentAppGeneratorError("Agent build draft not found")
+        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
 
     @staticmethod
     def _resolve_agent_by_id(

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

@@ -941,64 +941,48 @@ class AgentAppRunner:
             if pending_text:
                 persist_answer_text(pending_text)
 
-        try:
-            public_events = self._agent_backend_client.stream_events(
-                run_id,
-                should_stop=queue_manager.is_stopped,
-            )
-            for public_event in public_events:
+        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):
                 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):
-                    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):
-                            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
+                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
-                    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
+
+                    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
+                flush_pending_agent_message_text()
+                terminal = internal_event
+                break
+            if terminal is not None:
+                break
         flush_pending_agent_message_text()
-        if queue_manager.is_stopped():
-            self._cancel_run(run_id)
-            raise GenerateTaskStoppedError()
         return terminal, process_recorder
 
     def _cancel_run(self, run_id: str) -> None:

api/core/app/apps/base_app_generator.py

@@ -32,7 +32,6 @@ class _DebuggerDraftVariableSaver:
         self,
         *,
         account: Account,
-        tenant_id: str,
         app_id: str,
         node_id: str,
         node_type: NodeType,
@@ -40,7 +39,6 @@ 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
@@ -51,7 +49,6 @@ 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,
@@ -290,12 +287,7 @@ class BaseAppGenerator:
 
     @final
     @staticmethod
-    def _get_draft_var_saver_factory(
-        invoke_from: InvokeFrom,
-        account: Account | EndUser,
-        *,
-        tenant_id: str,
-    ) -> DraftVariableSaverFactory:
+    def _get_draft_var_saver_factory(invoke_from: InvokeFrom, account: Account | EndUser) -> DraftVariableSaverFactory:
         if invoke_from == InvokeFrom.DEBUGGER:
             assert isinstance(account, Account)
 
@@ -308,7 +300,6 @@ class BaseAppGenerator:
             ) -> DraftVariableSaver:
                 return _DebuggerDraftVariableSaver(
                     account=account,
-                    tenant_id=tenant_id,
                     app_id=app_id,
                     node_id=node_id,
                     node_type=node_type,

api/core/app/apps/base_app_queue_manager.py

@@ -21,7 +21,6 @@ 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__)
@@ -52,9 +51,6 @@ 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):
         """
@@ -63,7 +59,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.monotonic()
+        start_time = time.time()
         last_ping_time: int | float = 0
         try:
             while True:
@@ -76,14 +72,8 @@ class AppQueueManager(ABC):
                 except queue.Empty:
                     continue
                 finally:
-                    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)
+                    elapsed_time = time.time() - start_time
+                    if elapsed_time >= listen_timeout or self._is_stopped():
                         # publish two messages to make sure the client can receive the stop signal
                         # and stop listening after the stop signal processed
                         self.publish(
@@ -94,33 +84,16 @@ 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, *, execution_terminal: bool = False):
+    def stop_listen(self):
         """
         Stop listen to queue
         :return:
         """
-        if execution_terminal:
-            self._execution_terminal.set()
         self._clear_task_belong_cache()
         self._q.put(None)
 
-    def _abort_execution(self, reason: str) -> None:
-        """Propagate response timeout/disconnect to legacy and GraphEngine runners."""
-        with self._lifecycle_lock:
-            if self._execution_terminal.is_set() or self._abort_sent.is_set():
-                return
-            self._abort_sent.set()
-
-        try:
-            self.set_stop_flag_no_user_check(self._task_id)
-            GraphEngineManager(redis_client).send_stop_command(self._task_id, reason=reason)
-        except Exception:
-            logger.exception("Failed to abort app execution for task %s", self._task_id)
-
     def _clear_task_belong_cache(self) -> None:
         """
         Remove the task belong cache key once listening is finished.

api/core/app/apps/message_based_app_queue_manager.py

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

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

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

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

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

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

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

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

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

api/core/app/apps/workflow_app_runner.py

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

api/core/app/entities/queue_entities.py

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

api/core/helper/ssrf_proxy.py

@@ -47,24 +47,6 @@ 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
 
@@ -160,31 +142,7 @@ def _inject_trace_headers(headers: Headers | None) -> Headers:
     return headers
 
 
-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.
-    """
+def make_request(method: str, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     # Convert requests-style allow_redirects to httpx-style follow_redirects
     if "allow_redirects" in kwargs:
         allow_redirects = kwargs.pop("allow_redirects")
@@ -217,11 +175,6 @@ def make_request(
     # 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:
@@ -232,11 +185,7 @@ def make_request(
             if user_provided_host is not None:
                 headers["host"] = user_provided_host
             kwargs["headers"] = headers
-            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)
+            response = client.request(method=method, url=url, **kwargs)
 
             # Check for SSRF protection by Squid proxy
             if response.status_code in (401, 403):
@@ -246,7 +195,6 @@ def make_request(
 
                 # 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. "
@@ -260,7 +208,6 @@ def make_request(
                     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)
@@ -273,42 +220,6 @@ def make_request(
     raise MaxRetriesExceededError(f"Reached maximum retries ({max_retries}) for URL {url}")
 
 
-def buffer_response(response: httpx.Response, *, max_response_bytes: int) -> httpx.Response:
-    """Consume one open identity response under a decoded byte limit and close its stream."""
-    if max_response_bytes <= 0:
-        raise ValueError("max_response_bytes must be positive")
-
-    try:
-        content_encoding = response.headers.get("content-encoding", "identity").strip().lower()
-        if content_encoding not in {"", "identity"}:
-            raise UnsupportedResponseEncodingError(f"content encoding {content_encoding} cannot be safely bounded")
-        content = bytearray()
-        for chunk in response.iter_bytes():
-            if len(content) + len(chunk) > max_response_bytes:
-                raise ResponseTooLargeError(f"response exceeded {max_response_bytes} bytes")
-            content.extend(chunk)
-        decoded_headers = {
-            name: value
-            for name, value in response.headers.items()
-            if name.lower() not in {"content-encoding", "content-length", "transfer-encoding"}
-        }
-        try:
-            request = response.request
-        except RuntimeError:
-            request = None
-        return httpx.Response(
-            response.status_code,
-            headers=decoded_headers,
-            content=bytes(content),
-            request=request,
-            extensions=response.extensions,
-            history=response.history,
-            default_encoding=response.default_encoding,
-        )
-    finally:
-        response.close()
-
-
 def get(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
     return make_request("GET", url, max_retries=max_retries, **kwargs)
 

api/core/hosting_configuration.py

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

api/core/indexing_runner.py

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

api/core/logging/context.py

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

api/core/logging/filters.py

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

api/core/plugin/entities/plugin_daemon.py

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

api/core/plugin/impl/base.py

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

api/core/plugin/impl/exc.py

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

api/core/plugin/impl/plugin.py

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

api/core/plugin/plugin_service.py

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

api/core/provider_manager.py

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

api/core/rag/data_post_processor/data_post_processor.py

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

api/core/rag/datasource/retrieval_service.py

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

api/core/rag/docstore/dataset_docstore.py

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

api/core/rag/embedding/token_counter.py

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

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

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

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

@@ -1,15 +1,5 @@
-"""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 Any, override
-
-from httpx import Response
+from typing import override
 
 
 class WaterCrawlError(Exception):
@@ -17,16 +7,11 @@ class WaterCrawlError(Exception):
 
 
 class WaterCrawlBadRequestError(WaterCrawlError):
-    def __init__(self, response: Response):
+    def __init__(self, response):
         self.status_code = response.status_code
         self.response = response
-        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"
+        data = response.json()
+        self.message = data.get("message", "Unknown error occurred")
         self.errors = data.get("errors", {})
         super().__init__(self.message)
 

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

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

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

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

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

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

api/core/rag/rerank/rerank_model.py

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

api/core/rag/retrieval/dataset_retrieval.py

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

api/core/rbac/entities.py

@@ -61,7 +61,6 @@ 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"