fix: harden workflow archive delete resume

feat: support sharded workflow archive deletion
Merge branch 'main' into zhaohao1004/workflow-archive-catalog-discovery
2026-07-20 12:06:54 +08:00 · 2026-07-20 11:34:01 +08:00 · 2026-07-20 09:48:53 +08:00 · 2026-07-18 10:07:22 +08:00 · 2026-07-16 15:05:24 +08:00 · 2026-07-15 16:51:42 +08:00
1462 changed files with 27710 additions and 63632 deletions
@@ -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.

@@ -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.
@@ -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
@@ -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 \
@@ -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:
@@ -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/**'
@@ -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
@@ -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: ''
@@ -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 }}
@@ -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: e2e-admin@example.com
@@ -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: e2e-admin@example.com
@@ -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
@@ -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:
@@ -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 \
@@ -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
@@ -681,14 +681,6 @@ 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
 MARKETPLACE_API_URL=https://marketplace.dify.ai
@@ -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,
@@ -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,
 ):
@@ -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
@@ -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,
@@ -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")
@@ -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",
@@ -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
@@ -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,
@@ -552,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
@@ -590,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
@@ -632,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
@@ -658,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):
@@ -669,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",
@@ -693,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,
@@ -716,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
@@ -739,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
@@ -757,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
@@ -816,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
@@ -839,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
@@ -865,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):
@@ -882,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
@@ -901,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]:
@@ -912,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]:
@@ -932,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(
@@ -978,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)
@@ -1017,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)
@@ -1056,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)
@@ -1077,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)
@@ -1103,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):
@@ -1119,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):
@@ -1140,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
@@ -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,
@@ -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):
@@ -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:
@@ -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,23 +343,14 @@ 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,
+    *, session: Session, current_tenant_id: str, current_user: Account, app_model: App, agent_id: str | None
 ) -> str:
-    """Resolve the current editor's conversation without crossing draft surfaces."""
-
    roster_service = AgentRosterService(session)
    if agent_id:
        return roster_service.get_or_create_agent_app_debug_conversation_id(
            tenant_id=current_tenant_id,
            agent_id=agent_id,
            account_id=current_user.id,
-            draft_type=draft_type,
        )

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


@@ -393,7 +382,6 @@ def _create_chat_message(
            current_user=current_user,
            app_model=app_model,
            agent_id=agent_id,
-            draft_type=AgentConfigDraftType(args_model.draft_type),
        )
        if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
            raise NotFound("Conversation Not Exists.")
@@ -430,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,
@@ -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
@@ -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")
@@ -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:
@@ -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()

@@ -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."
@@ -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:
@@ -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:
@@ -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",
@@ -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

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

@@ -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:
@@ -67,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
@@ -654,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
@@ -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(
@@ -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

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

@@ -9,13 +9,11 @@ from controllers.common.schema import register_response_schema_models
 from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
 from extensions.ext_database import db
-from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
 from models.account import Tenant, TenantStatus
-from models.model import App, EndUser, 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):
@@ -34,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):
@@ -86,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:
@@ -101,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
@@ -123,15 +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.EDITION == "CLOUD" and StorageType(dify_config.STORAGE_TYPE) == StorageType.S3:
-        return FileService(db.engine).get_file_presigned_url(file_id=site.icon, tenant_id=tenant_id)
-    return build_icon_url(site.icon_type, site.icon)
-
-
@web_ns.route("/site")
 class AppSiteApi(WebApiResource):
    @web_ns.doc("Get App Site Info")
@@ -168,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")
@@ -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)
@@ -682,27 +682,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(
@@ -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,
@@ -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(
@@ -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(
@@ -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)

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

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

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

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

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

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

        return result

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

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

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

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


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

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

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

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

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


 class ToolProviderIdentity(BaseModel):
@@ -2,7 +2,7 @@ from __future__ import annotations

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

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

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

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

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

    @staticmethod
    def _tool_layer_destination(tool_config: AgentSoulDifyToolConfig) -> Literal["plugin", "core"]:
-        provider_type = tool_config.provider_type
+        provider_type = ToolProviderType.value_of(tool_config.provider_type)
        if provider_type is ToolProviderType.PLUGIN or (
            provider_type is ToolProviderType.BUILT_IN and _is_plugin_provider_id(tool_config.provider_id)
        ):
            return "plugin"
-        if provider_type in _CORE_TOOL_PROVIDER_TYPES:
+        if provider_type in {
+            ToolProviderType.BUILT_IN,
+            ToolProviderType.API,
+            ToolProviderType.WORKFLOW,
+            ToolProviderType.MCP,
+        }:
            return "core"
        if provider_type is ToolProviderType.DATASET_RETRIEVAL:
            raise WorkflowAgentDifyToolsBuildError(
@@ -416,7 +417,7 @@ class WorkflowAgentDifyToolsBuilder:
    ) -> DifyCoreToolConfig:
        parameters = self._prepared_parameters(tool_runtime)
        return DifyCoreToolConfig(
-            provider_type=_CORE_TOOL_PROVIDER_TYPES[tool_config.provider_type],
+            provider_type=cast(DifyCoreToolProviderType, tool_config.provider_type),
            provider_id=self._provider_id(tool_config),
            tool_name=tool_config.tool_name or exposed_name,
            credential_id=tool_config.credential_ref.id if tool_config.credential_ref else None,
@@ -1,385 +0,0 @@
-"""Validate Dify Console KnowledgeFS declarations against a pinned OpenAPI document.
-
-The OpenAPI document is exported only during explicit development validation. Runtime declarations live with Dify
-product policy; this module validates their transport metadata without generating a complete operation catalog.
-"""
-
-from __future__ import annotations
-
-import argparse
-import hashlib
-import json
-import subprocess
-import sys
-import tempfile
-from copy import deepcopy
-from pathlib import Path
-from typing import Any, Literal, TypedDict
-
-API_ROOT = Path(__file__).resolve().parents[1]
-if str(API_ROOT) not in sys.path:
-    sys.path.insert(0, str(API_ROOT))
-
-WORKSPACE_ROOT = API_ROOT.parent
-LOCK_PATH = API_ROOT / "knowledge-fs-contract.lock.json"
-DEFAULT_REPOSITORY = WORKSPACE_ROOT.parent / "knowledge-fs"
-OPENAPI_METHODS = ("delete", "get", "head", "options", "patch", "post", "put", "trace")
-PROXY_METHODS = frozenset({"delete", "get", "patch", "post", "put"})
-CONSOLE_PROXY_ERROR_SCHEMA_NAME = "ConsoleProxyError"
-CONSOLE_PROXY_ERROR_SCHEMA: dict[str, Any] = {
-    "type": "object",
-    "required": ["code", "message", "status"],
-    "properties": {
-        "code": {"type": "string"},
-        "message": {"type": "string"},
-        "status": {"type": "integer"},
-    },
-}
-
-
-class ContractDeclaration(TypedDict):
-    """KnowledgeFS transport contract declared by one Dify Console registry entry."""
-
-    operation_id: str
-    method: str
-    path: str
-    required_scope: str | None
-    response_kind: str
-    max_response_bytes: int
-    request_headers: tuple[str, ...]
-    response_headers: tuple[str, ...]
-    response_media_types: tuple[str, ...]
-    error_status_map: tuple[tuple[int, int], ...]
-
-
-type DeclarationField = Literal[
-    "method",
-    "path",
-    "required_scope",
-    "response_kind",
-    "max_response_bytes",
-    "request_headers",
-    "response_headers",
-    "response_media_types",
-]
-
-DECLARATION_FIELDS: tuple[DeclarationField, ...] = (
-    "method",
-    "path",
-    "required_scope",
-    "response_kind",
-    "max_response_bytes",
-    "request_headers",
-    "response_headers",
-    "response_media_types",
-)
-
-
-def main() -> None:
-    """Update or verify the pin and validate Console declarations against its OpenAPI document."""
-    parser = argparse.ArgumentParser()
-    mode = parser.add_mutually_exclusive_group()
-    mode.add_argument("--check", action="store_true")
-    mode.add_argument("--update-lock", action="store_true")
-    parser.add_argument("--repository", type=Path, default=DEFAULT_REPOSITORY)
-    parser.add_argument("--output-openapi", type=Path)
-    args = parser.parse_args()
-
-    repository = args.repository.resolve()
-    lock = json.loads(LOCK_PATH.read_text())
-    tracked_changes = run("git", "status", "--porcelain", "--untracked-files=no", cwd=repository).strip()
-    if tracked_changes:
-        raise RuntimeError("KnowledgeFS checkout must not contain tracked changes during contract export")
-
-    commit = run("git", "rev-parse", "HEAD", cwd=repository).strip()
-    if not args.update_lock and commit != lock["commit"]:
-        raise RuntimeError(
-            f"KnowledgeFS checkout mismatch: expected {lock['commit']}, received {commit}. "
-            "Use the pinned commit or pass --update-lock intentionally."
-        )
-
-    with tempfile.TemporaryDirectory(prefix="dify-knowledge-fs-contract-") as directory:
-        openapi_path = Path(directory) / "knowledge-fs.openapi.json"
-        subprocess.run(
-            ["pnpm", "openapi:export", "--", "--output", str(openapi_path)],
-            cwd=repository,
-            check=True,
-        )
-        openapi_content = openapi_path.read_bytes()
-
-    openapi_sha256 = sha256(openapi_content)
-    if not args.update_lock and openapi_sha256 != lock["openapiSha256"]:
-        raise RuntimeError(
-            f"KnowledgeFS OpenAPI hash mismatch: expected {lock['openapiSha256']}, received {openapi_sha256}"
-        )
-
-    document: dict[str, Any] = json.loads(openapi_content)
-    declarations = console_contract_declarations()
-    validate_declarations(document, declarations)
-
-    if args.output_openapi:
-        filtered_document = filter_openapi_document(document, declarations)
-        filtered_document["x-dify-source-openapi-sha256"] = openapi_sha256
-        filtered_document["x-dify-console-declarations-sha256"] = contract_declarations_sha256(declarations)
-        args.output_openapi.parent.mkdir(parents=True, exist_ok=True)
-        args.output_openapi.write_text(json.dumps(filtered_document, indent=2) + "\n")
-
-    if args.update_lock:
-        LOCK_PATH.write_text(
-            json.dumps(
-                {
-                    "commit": commit,
-                    "openapiSha256": openapi_sha256,
-                    "repository": lock["repository"],
-                },
-                indent=2,
-            )
-            + "\n"
-        )
-
-
-def validate_declarations(document: dict[str, Any], declarations: tuple[ContractDeclaration, ...]) -> None:
-    """Validate Dify Console declarations against matching pinned OpenAPI operations."""
-    operations_by_id: dict[str, list[tuple[str, str, dict[str, Any], dict[str, Any]]]] = {}
-    for path, path_item in document.get("paths", {}).items():
-        for method in OPENAPI_METHODS:
-            operation = path_item.get(method)
-            if operation is None:
-                continue
-            operation_id = operation.get("operationId")
-            if isinstance(operation_id, str) and operation_id:
-                operations_by_id.setdefault(operation_id, []).append((method, path, path_item, operation))
-
-    declared_ids: set[str] = set()
-    for declaration in declarations:
-        operation_id = declaration["operation_id"]
-        if operation_id in declared_ids:
-            raise ValueError(f"Dify Console registry has duplicate operationId: {operation_id}")
-        declared_ids.add(operation_id)
-
-        matches = operations_by_id.get(operation_id, [])
-        if not matches:
-            raise ValueError(f"KnowledgeFS OpenAPI has no operationId: {operation_id}")
-        if len(matches) > 1:
-            raise ValueError(f"KnowledgeFS OpenAPI has duplicate operationId: {operation_id}")
-
-        method, path, path_item, operation = matches[0]
-        if not path.startswith("/"):
-            raise ValueError(f"KnowledgeFS OpenAPI path must be absolute: {path}")
-        if method not in PROXY_METHODS:
-            raise ValueError(f"KnowledgeFS proxy does not support {method.upper()} {path}")
-        expected: dict[DeclarationField, object] = {
-            "method": method.upper(),
-            "path": path[1:],
-            "required_scope": required_scope(operation),
-            "response_kind": response_kind(operation),
-            "max_response_bytes": required_max_response_bytes(operation),
-            "request_headers": request_header_names(path_item, operation),
-            "response_headers": response_header_names(operation),
-            "response_media_types": response_media_types(operation),
-        }
-        for field in DECLARATION_FIELDS:
-            expected_value = expected[field]
-            received_value = declaration[field]
-            if received_value != expected_value:
-                raise ValueError(
-                    f"KnowledgeFS operation {operation_id} field {field} drifted: "
-                    f"expected {expected_value!r}, received {received_value!r}"
-                )
-        validate_error_status_map(operation_id, declaration["error_status_map"])
-
-
-def filter_openapi_document(
-    document: dict[str, Any],
-    declarations: tuple[ContractDeclaration, ...],
-) -> dict[str, Any]:
-    """Return a code-generation document containing only Console-allowlisted operations."""
-    filtered_document: dict[str, Any] = {
-        key: value for key, value in document.items() if key not in {"components", "paths"}
-    }
-    source_paths = document.get("paths", {})
-    filtered_paths: dict[str, Any] = {}
-
-    for declaration in declarations:
-        path = f"/{declaration['path']}"
-        method = declaration["method"].lower()
-        source_path_item = source_paths[path]
-        path_metadata = {key: value for key, value in source_path_item.items() if key not in OPENAPI_METHODS}
-        filtered_path_item = filtered_paths.setdefault(path, path_metadata)
-        filtered_operation = deepcopy(source_path_item[method])
-        _rewrite_proxy_error_responses(filtered_operation, declaration["error_status_map"])
-        filtered_path_item[method] = filtered_operation
-
-    filtered_document["paths"] = filtered_paths
-
-    source_components = document.get("components", {})
-    filtered_components = {key: value for key, value in source_components.items() if key != "schemas"}
-    source_schemas = source_components.get("schemas", {})
-    available_schemas = {**source_schemas, CONSOLE_PROXY_ERROR_SCHEMA_NAME: CONSOLE_PROXY_ERROR_SCHEMA}
-    schema_names = _referenced_schema_names(filtered_paths, available_schemas)
-    filtered_components["schemas"] = {
-        name: schema for name, schema in available_schemas.items() if name in schema_names
-    }
-    filtered_document["components"] = filtered_components
-    return filtered_document
-
-
-def validate_error_status_map(operation_id: str, error_status_map: tuple[tuple[int, int], ...]) -> None:
-    """Validate the status normalization advertised by one Console operation."""
-    upstream_statuses: set[int] = set()
-    for upstream_status, console_status in error_status_map:
-        if upstream_status in upstream_statuses:
-            raise ValueError(f"KnowledgeFS operation {operation_id} has duplicate error status: {upstream_status}")
-        if not 400 <= upstream_status <= 599 or not 400 <= console_status <= 599:
-            raise ValueError(f"KnowledgeFS operation {operation_id} has invalid error status mapping")
-        upstream_statuses.add(upstream_status)
-
-
-def _rewrite_proxy_error_responses(
-    operation: dict[str, Any],
-    error_status_map: tuple[tuple[int, int], ...],
-) -> None:
-    responses = operation.setdefault("responses", {})
-    proxy_error_response = {
-        "description": "Error normalized by the Dify Console KnowledgeFS proxy.",
-        "content": {
-            "application/json": {"schema": {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"}}
-        },
-    }
-    for upstream_status, console_status in error_status_map:
-        existing_target = responses.get(str(console_status)) if upstream_status != console_status else None
-        responses.pop(str(upstream_status), None)
-        normalized_response: dict[str, Any] = deepcopy(proxy_error_response)
-        existing_schema = (
-            existing_target.get("content", {}).get("application/json", {}).get("schema")
-            if isinstance(existing_target, dict)
-            else None
-        )
-        if existing_schema is not None:
-            normalized_response["content"]["application/json"]["schema"] = {
-                "oneOf": [
-                    deepcopy(existing_schema),
-                    {"$ref": f"#/components/schemas/{CONSOLE_PROXY_ERROR_SCHEMA_NAME}"},
-                ]
-            }
-        responses[str(console_status)] = normalized_response
-
-
-def _referenced_schema_names(value: Any, schemas: dict[str, Any]) -> set[str]:
-    reference_prefix = "#/components/schemas/"
-    selected: set[str] = set()
-    pending: list[Any] = [value]
-
-    while pending:
-        current = pending.pop()
-        if isinstance(current, list):
-            pending.extend(current)
-            continue
-        if not isinstance(current, dict):
-            continue
-
-        reference = current.get("$ref")
-        if isinstance(reference, str) and reference.startswith(reference_prefix):
-            name = reference.removeprefix(reference_prefix)
-            if name not in selected:
-                if name not in schemas:
-                    raise ValueError(f"KnowledgeFS OpenAPI references missing schema: {name}")
-                selected.add(name)
-                pending.append(schemas[name])
-        pending.extend(current.values())
-
-    return selected
-
-
-def console_contract_declarations() -> tuple[ContractDeclaration, ...]:
-    """Return transport declarations from the runtime Console operation registry."""
-    from services.knowledge_fs_operations import KNOWLEDGE_FS_CONSOLE_OPERATIONS
-
-    return tuple(
-        {
-            "operation_id": operation.operation_id,
-            "method": operation.method,
-            "path": operation.path,
-            "required_scope": operation.required_scope,
-            "response_kind": operation.response_kind,
-            "max_response_bytes": operation.max_response_bytes,
-            "request_headers": operation.request_headers,
-            "response_headers": operation.response_headers,
-            "response_media_types": operation.response_media_types,
-            "error_status_map": operation.error_status_map,
-        }
-        for operation in KNOWLEDGE_FS_CONSOLE_OPERATIONS
-    )
-
-
-def contract_declarations_sha256(declarations: tuple[ContractDeclaration, ...]) -> str:
-    """Return a stable digest for the runtime Console operation declarations."""
-    content = json.dumps(declarations, separators=(",", ":"), sort_keys=True).encode()
-    return sha256(content)
-
-
-def response_kind(operation: dict[str, Any]) -> str:
-    media_types = response_media_types(operation)
-    if "text/event-stream" in media_types:
-        return "stream"
-    if "application/octet-stream" in media_types:
-        return "binary"
-    return "buffered"
-
-
-def response_media_types(operation: dict[str, Any]) -> tuple[str, ...]:
-    media_types: set[str] = set()
-    for status, response in operation.get("responses", {}).items():
-        if status == "2XX" or (len(status) == 3 and status.startswith("2") and status.isdigit()):
-            media_types.update(response.get("content", {}))
-    return tuple(sorted(media_types))
-
-
-def required_scope(operation: dict[str, Any]) -> str | None:
-    scope = operation.get("x-knowledge-fs-required-scope")
-    if scope in ("knowledge-spaces:read", "knowledge-spaces:write"):
-        return scope
-    if operation.get("security") == []:
-        return None
-    raise ValueError(f"KnowledgeFS operation has no supported required scope: {scope}")
-
-
-def required_max_response_bytes(operation: dict[str, Any]) -> int:
-    value = operation.get("x-knowledge-fs-max-response-bytes")
-    if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
-        raise ValueError(f"KnowledgeFS operation has no valid response byte limit: {value}")
-    return value
-
-
-def request_header_names(path_item: dict[str, Any], operation: dict[str, Any]) -> tuple[str, ...]:
-    names: set[str] = set()
-    for parameter in [*path_item.get("parameters", []), *operation.get("parameters", [])]:
-        if "$ref" in parameter:
-            raise ValueError(f"KnowledgeFS request header references are not supported: {parameter['$ref']}")
-        if parameter.get("in") == "header":
-            names.add(parameter["name"].lower())
-    return tuple(sorted(names))
-
-
-def response_header_names(operation: dict[str, Any]) -> tuple[str, ...]:
-    return tuple(
-        sorted(
-            {
-                name.lower()
-                for response in operation.get("responses", {}).values()
-                for name in response.get("headers", {})
-            }
-        )
-    )
-
-
-def sha256(content: bytes) -> str:
-    return hashlib.sha256(content).hexdigest()
-
-
-def run(*command: str, cwd: Path) -> str:
-    return subprocess.run(command, cwd=cwd, check=True, capture_output=True, text=True).stdout
-
-
-if __name__ == "__main__":
-    main()
@@ -157,7 +157,6 @@ def init_app(app: DifyApp) -> Celery:
        "tasks.regenerate_summary_index_task",  # summary index regeneration
        "tasks.initialize_created_app_rbac_access_task",  # app access initialization
        "tasks.install_default_plugins_task",  # tenant default plugin installation
-        "tasks.refresh_billing_vector_space_task",  # billing vector-space cache refresh
        "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
        "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
    ]
@@ -119,19 +119,6 @@ class Storage:
    def delete(self, filename: str):
        return self.storage_runner.delete(filename)

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

@@ -92,21 +92,3 @@ class AwsS3Storage(BaseStorage):
    @override
    def delete(self, filename: str):
        self.client.delete_object(Bucket=self.bucket_name, Key=filename)
-
-    @override
-    def generate_presigned_url(
-        self,
-        filename: str,
-        *,
-        expires_in: int,
-        content_type: str | None = None,
-    ) -> str:
-        params = {"Bucket": self.bucket_name, "Key": filename}
-        if content_type:
-            params["ResponseContentType"] = content_type
-
-        return self.client.generate_presigned_url(
-            "get_object",
-            Params=params,
-            ExpiresIn=expires_in,
-        )
@@ -31,16 +31,6 @@ class BaseStorage(ABC):
    def delete(self, filename: str):
        raise NotImplementedError

-    def generate_presigned_url(
-        self,
-        filename: str,
-        *,
-        expires_in: int,
-        content_type: str | None = None,
-    ) -> str:
-        """Generate a temporary direct-download URL when the backend supports it."""
-        raise NotImplementedError("This storage backend doesn't support presigned URLs")
-
    def scan(self, path, files=True, directories=False) -> list[str]:
        """
        Scan files and directories in the given path.
@@ -1,5 +0,0 @@
-{
-  "commit": "a0f50470612cc0b3656f89e4f2435aaf412e6e3b",
-  "openapiSha256": "f18910e9c45a64f0855e0643a7a626fb2889021b4f943458de86c6bd2469facb",
-  "repository": "https://github.com/langgenius/knowledge-fs"
-}
@@ -9,8 +9,6 @@ from werkzeug.http import HTTP_STATUS_CODES

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

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

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

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


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

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

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

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


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

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

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


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

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

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

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

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


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

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

    @property
    def api_base_url(self) -> str:
@@ -1,59 +0,0 @@
-"""Account-level onboarding state models."""
-
-from datetime import datetime
-
-import sqlalchemy as sa
-from sqlalchemy import DateTime, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from .base import TypeBase, gen_uuidv7_string
-from .types import AdjustedJSON, StringUUID
-
-
-class AccountStepByStepTourState(TypeBase):
-    """Persistent account-level Step-by-step Tour state.
-
-    The tour is account-owned, with workspace IDs stored only as presentation
-    overrides. The first workspace is the workspace context where an eligible
-    account first asks for tour state; subsequent workspaces are opt-in only.
-    """
-
-    __tablename__ = "account_step_by_step_tour_states"
-    __table_args__ = (
-        sa.PrimaryKeyConstraint("id", name="account_step_by_step_tour_state_pkey"),
-        sa.UniqueConstraint("account_id", name="account_step_by_step_tour_state_account_id_key"),
-    )
-
-    id: Mapped[str] = mapped_column(
-        StringUUID,
-        insert_default=gen_uuidv7_string,
-        default_factory=gen_uuidv7_string,
-        init=False,
-    )
-    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    first_workspace_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
-    skipped: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"), default=False)
-    completed_task_ids: Mapped[list[str]] = mapped_column(AdjustedJSON, nullable=False, default_factory=list)
-    manually_enabled_workspace_ids: Mapped[list[str]] = mapped_column(
-        AdjustedJSON,
-        nullable=False,
-        default_factory=list,
-    )
-    manually_disabled_workspace_ids: Mapped[list[str]] = mapped_column(
-        AdjustedJSON,
-        nullable=False,
-        default_factory=list,
-    )
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime,
-        server_default=func.current_timestamp(),
-        nullable=False,
-        init=False,
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime,
-        server_default=func.current_timestamp(),
-        nullable=False,
-        init=False,
-        onupdate=func.current_timestamp(),
-    )
@@ -260,12 +260,7 @@ Get account avatar url
 | 200 | Success | **application/json**: [AccountResponse](#accountresponse)<br> |

 ### [POST] /activate
-**Accept an invitation without letting an existing session act for another account**
-
 Activate account with invitation token
-Token-only activation remains available for legacy clients. When the request already
-carries a console session, that session must belong to the account encoded in the
-invitation before the token is consumed or tenant membership is changed.

 #### Request Body

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

 #### SystemConfigurationResponse

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

 #### ToolParameter

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

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

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

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

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

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


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

        return users

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

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

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

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

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


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

    def _load_daily_statistics(
        self, *, app: App, agent_id: str, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
-    ) -> list[dict[str, Any]]:
-        rows: list[dict[str, Any]] = []
-        if source_filter.kind in {"all", "webapp"}:
-            rows.extend(self._load_webapp_daily_statistics(app=app, params=params, source_filter=source_filter))
-        if source_filter.kind in {"all", "workflow"}:
-            rows.extend(
-                self._load_workflow_daily_statistics(
-                    app=app,
-                    agent_id=agent_id,
-                    params=params,
-                    source_filter=source_filter,
-                )
-            )
-        return self._merge_daily_statistics(rows)
-
-    def _load_webapp_daily_statistics(
-        self, *, app: App, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
    ) -> list[dict[str, Any]]:
        converted_created_at = convert_datetime_to_date("m.created_at")
-        message_scope = self._statistics_webapp_message_scope_sql(source_filter)
+        message_scope = self._statistics_message_scope_sql(source_filter)
        sql_query = f"""SELECT
    {converted_created_at} AS date,
    COUNT(m.id) AS message_count,
@@ -620,9 +602,20 @@ WHERE
        args: dict[str, Any] = {
            "tz": params.timezone,
            "app_id": app.id,
+            "tenant_id": app.tenant_id,
+            "agent_id": agent_id,
+            "debugger": InvokeFrom.DEBUGGER,
        }
        if source_filter.invoke_from is not None:
            args["source"] = source_filter.invoke_from
+        if source_filter.app_id:
+            args["source_app_id"] = source_filter.app_id
+        if source_filter.workflow_id:
+            args["workflow_id"] = source_filter.workflow_id
+        if source_filter.workflow_version:
+            args["workflow_version"] = source_filter.workflow_version
+        if source_filter.node_id:
+            args["node_id"] = source_filter.node_id
        if params.start:
            sql_query += " AND m.created_at >= :start"
            args["start"] = params.start
@@ -634,160 +627,10 @@ WHERE
        return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]

    @staticmethod
-    def _statistics_webapp_message_scope_sql(source_filter: AgentSourceFilter) -> str:
+    def _statistics_message_scope_sql(source_filter: AgentSourceFilter) -> str:
        app_scope = "m.app_id = :app_id"
        if source_filter.invoke_from is not None:
            app_scope += " AND m.invoke_from = :source"
-        return app_scope
-
-    def _load_workflow_daily_statistics(
-        self,
-        *,
-        app: App,
-        agent_id: str,
-        params: AgentStatisticsQueryParams,
-        source_filter: AgentSourceFilter,
-    ) -> list[dict[str, Any]]:
-        converted_run_created_at = convert_datetime_to_date("aru.created_at")
-        total_tokens = self._workflow_execution_metadata_numeric_sql(("total_tokens",), "BIGINT")
-        nested_total_tokens = self._workflow_execution_metadata_numeric_sql(
-            ("agent_log", "agent_backend", "usage", "total_tokens"), "BIGINT"
-        )
-        total_price = self._workflow_execution_metadata_numeric_sql(("total_price",), "DECIMAL(65, 30)")
-        nested_total_price = self._workflow_execution_metadata_numeric_sql(
-            ("agent_log", "agent_backend", "usage", "total_price"), "DECIMAL(65, 30)"
-        )
-        completion_tokens = self._workflow_execution_metadata_numeric_sql(
-            ("agent_log", "agent_backend", "usage", "completion_tokens"), "BIGINT"
-        )
-        binding_filters = self._statistics_workflow_binding_filters_sql(source_filter)
-        run_date_filters = ""
-        args: dict[str, Any] = {
-            "tz": params.timezone,
-            "tenant_id": app.tenant_id,
-            "agent_id": agent_id,
-            "chat_workflow_type": WorkflowType.CHAT,
-            "end_user_role": CreatorUserRole.END_USER,
-        }
-        if source_filter.app_id:
-            args["source_app_id"] = source_filter.app_id
-        if source_filter.workflow_id:
-            args["workflow_id"] = source_filter.workflow_id
-        if source_filter.workflow_version:
-            args["workflow_version"] = source_filter.workflow_version
-        if source_filter.node_id:
-            args["node_id"] = source_filter.node_id
-        if params.start:
-            run_date_filters += " AND wr.created_at >= :start"
-            args["start"] = params.start
-        if params.end:
-            run_date_filters += " AND wr.created_at < :end"
-            args["end"] = params.end
-
-        run_query = f"""WITH agent_run_usage AS (
-    SELECT
-        wr.id,
-        wr.created_by_role,
-        wr.created_by,
-        wr.created_at,
-        COALESCE(SUM(COALESCE({total_tokens}, {nested_total_tokens}, 0)), 0) AS token_count,
-        COALESCE(SUM(COALESCE({total_price}, {nested_total_price}, 0)), 0) AS total_price,
-        COALESCE(SUM(COALESCE(wne.elapsed_time, 0)), 0) AS latency,
-        COALESCE(SUM(COALESCE({completion_tokens}, 0)), 0) AS answer_tokens
-    FROM workflow_runs wr
-    JOIN workflow_agent_node_bindings wanb
-        ON wanb.tenant_id = :tenant_id
-        AND wanb.agent_id = :agent_id
-        AND wanb.app_id = wr.app_id
-        AND wanb.workflow_id = wr.workflow_id
-        AND wanb.workflow_version = wr.version
-        {binding_filters}
-    JOIN workflow_node_executions wne
-        ON wne.workflow_run_id = wr.id
-        AND wne.node_id = wanb.node_id
-    WHERE wr.type != :chat_workflow_type{run_date_filters}
-    GROUP BY wr.id, wr.created_by_role, wr.created_by, wr.created_at
-)
-SELECT
-    {converted_run_created_at} AS date,
-    COUNT(aru.id) AS message_count,
-    COUNT(aru.id) AS conversation_count,
-    COUNT(DISTINCT CASE
-        WHEN aru.created_by_role = :end_user_role THEN aru.created_by
-        ELSE NULL
-    END) AS end_user_count,
-    COALESCE(SUM(aru.token_count), 0) AS token_count,
-    COALESCE(SUM(aru.total_price), 0) AS total_price,
-    COALESCE(AVG(aru.latency), 0) AS avg_latency,
-    COALESCE(SUM(aru.latency), 0) AS latency_sum,
-    COALESCE(SUM(aru.answer_tokens), 0) AS answer_tokens,
-    0 AS like_count
-FROM agent_run_usage aru
-GROUP BY date
-ORDER BY date"""
-        rows = [dict(row._mapping) for row in self._session.execute(sa.text(run_query), args).all()]
-        rows.extend(
-            self._load_workflow_chat_daily_context(
-                app=app,
-                agent_id=agent_id,
-                params=params,
-                source_filter=source_filter,
-            )
-        )
-        return self._merge_daily_statistics(rows)
-
-    def _load_workflow_chat_daily_context(
-        self,
-        *,
-        app: App,
-        agent_id: str,
-        params: AgentStatisticsQueryParams,
-        source_filter: AgentSourceFilter,
-    ) -> list[dict[str, Any]]:
-        converted_created_at = convert_datetime_to_date("m.created_at")
-        workflow_scope = self._statistics_workflow_message_scope_sql(source_filter)
-        sql_query = f"""SELECT
-    {converted_created_at} AS date,
-    COUNT(m.id) AS message_count,
-    COUNT(DISTINCT m.conversation_id) AS conversation_count,
-    COUNT(DISTINCT m.from_end_user_id) AS end_user_count,
-    COALESCE(SUM(COALESCE(m.message_tokens, 0) + COALESCE(m.answer_tokens, 0)), 0) AS token_count,
-    COALESCE(SUM(COALESCE(m.total_price, 0)), 0) AS total_price,
-    COALESCE(AVG(m.provider_response_latency), 0) AS avg_latency,
-    COALESCE(SUM(m.provider_response_latency), 0) AS latency_sum,
-    COALESCE(SUM(m.answer_tokens), 0) AS answer_tokens,
-    COUNT(mf.id) AS like_count
-FROM messages m
-LEFT JOIN message_feedbacks mf
-    ON mf.message_id = m.id AND mf.rating = 'like'
-WHERE
-    {workflow_scope}"""
-        args: dict[str, Any] = {
-            "tz": params.timezone,
-            "tenant_id": app.tenant_id,
-            "agent_id": agent_id,
-            "chat_workflow_type": WorkflowType.CHAT,
-        }
-        if source_filter.app_id:
-            args["source_app_id"] = source_filter.app_id
-        if source_filter.workflow_id:
-            args["workflow_id"] = source_filter.workflow_id
-        if source_filter.workflow_version:
-            args["workflow_version"] = source_filter.workflow_version
-        if source_filter.node_id:
-            args["node_id"] = source_filter.node_id
-        if params.start:
-            sql_query += " AND m.created_at >= :start"
-            args["start"] = params.start
-        if params.end:
-            sql_query += " AND m.created_at < :end"
-            args["end"] = params.end
-        sql_query += " GROUP BY date ORDER BY date"
-
-        return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]
-
-    @staticmethod
-    def _statistics_workflow_binding_filters_sql(source_filter: AgentSourceFilter) -> str:
        workflow_binding_filters = []
        if source_filter.app_id:
            workflow_binding_filters.append("wanb.app_id = :source_app_id")
@@ -797,12 +640,8 @@ WHERE
            workflow_binding_filters.append("wanb.workflow_version = :workflow_version")
        if source_filter.node_id:
            workflow_binding_filters.append("wanb.node_id = :node_id")
-        return f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
-
-    @classmethod
-    def _statistics_workflow_message_scope_sql(cls, source_filter: AgentSourceFilter) -> str:
-        binding_filters = cls._statistics_workflow_binding_filters_sql(source_filter)
-        return f"""m.workflow_run_id IS NOT NULL
+        extra_workflow_filters = f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
+        workflow_scope = f"""m.workflow_run_id IS NOT NULL
        AND EXISTS (
            SELECT 1
            FROM workflow_runs wr
@@ -812,65 +651,17 @@ WHERE
                AND wanb.app_id = wr.app_id
                AND wanb.workflow_id = wr.workflow_id
                AND wanb.workflow_version = wr.version
-                {binding_filters}
+                {extra_workflow_filters}
            JOIN workflow_node_executions wne
                ON wne.workflow_run_id = wr.id
                AND wne.node_id = wanb.node_id
            WHERE wr.id = m.workflow_run_id
-                AND wr.type = :chat_workflow_type
        )"""
-
-    @staticmethod
-    def _workflow_execution_metadata_numeric_sql(path: tuple[str, ...], numeric_type: str) -> str:
-        if dify_config.DB_TYPE == "postgresql":
-            json_path = ",".join(path)
-            value = f"CAST(wne.execution_metadata AS JSONB) #>> '{{{json_path}}}'"
-            return f"CAST(NULLIF({value}, '') AS {numeric_type})"
-        if dify_config.DB_TYPE in {"mysql", "oceanbase", "seekdb"}:
-            json_path = "$." + ".".join(path)
-            mysql_numeric_type = "UNSIGNED" if numeric_type == "BIGINT" else numeric_type
-            value = f"JSON_UNQUOTE(JSON_EXTRACT(wne.execution_metadata, '{json_path}'))"
-            return f"CAST(NULLIF(NULLIF({value}, ''), 'null') AS {mysql_numeric_type})"
-        raise NotImplementedError(f"Unsupported database type: {dify_config.DB_TYPE}")
-
-    @staticmethod
-    def _merge_daily_statistics(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
-        merged: dict[Any, dict[str, Any]] = {}
-        weighted_latency: dict[Any, float] = {}
-        for row in rows:
-            date = row["date"]
-            target = merged.setdefault(
-                date,
-                {
-                    "date": date,
-                    "message_count": 0,
-                    "conversation_count": 0,
-                    "end_user_count": 0,
-                    "token_count": 0,
-                    "total_price": Decimal(0),
-                    "avg_latency": 0.0,
-                    "latency_sum": 0.0,
-                    "answer_tokens": 0,
-                    "like_count": 0,
-                },
-            )
-            message_count = int(row.get("message_count") or 0)
-            target["message_count"] += message_count
-            target["conversation_count"] += int(row.get("conversation_count") or 0)
-            target["end_user_count"] += int(row.get("end_user_count") or 0)
-            target["token_count"] += int(row.get("token_count") or 0)
-            target["total_price"] += Decimal(str(row.get("total_price") or 0))
-            target["latency_sum"] += float(row.get("latency_sum") or 0)
-            target["answer_tokens"] += int(row.get("answer_tokens") or 0)
-            target["like_count"] += int(row.get("like_count") or 0)
-            weighted_latency[date] = (
-                weighted_latency.get(date, 0.0) + float(row.get("avg_latency") or 0) * message_count
-            )
-
-        for date, row in merged.items():
-            message_count = int(row["message_count"])
-            row["avg_latency"] = weighted_latency[date] / message_count if message_count else 0.0
-        return sorted(merged.values(), key=lambda row: str(row["date"]))
+        if source_filter.kind == "webapp":
+            return app_scope
+        if source_filter.kind == "workflow":
+            return workflow_scope
+        return f"(({app_scope}) OR ({workflow_scope}))"

    @staticmethod
    def _build_charts(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
@@ -430,11 +430,7 @@ class AgentRosterService:
        agent.active_config_has_model = agent_soul_has_model(soul)
        agent.active_config_is_published = False
        self._session.flush()
-        self._get_or_create_agent_app_debug_conversation(
-            agent=agent,
-            account_id=account_id,
-            draft_type=AgentConfigDraftType.DEBUG_BUILD,
-        )
+        self._get_or_create_agent_app_debug_conversation(agent=agent, account_id=account_id)
        return agent

    def create_hidden_backing_app_for_workflow_agent(
@@ -531,11 +527,7 @@ class AgentRosterService:
        self._session.flush()
        return backing_app.id

-    def _get_or_create_agent_app_debug_conversation(
-        self, *, agent: Agent, account_id: str, draft_type: AgentConfigDraftType
-    ) -> str:
-        """Return the editor's conversation for one Agent draft surface."""
-
+    def _get_or_create_agent_app_debug_conversation(self, *, agent: Agent, account_id: str) -> str:
        backing_app_id = self._ensure_workflow_agent_backing_app(agent=agent, account_id=account_id)
        if not backing_app_id:
            raise AgentNotFoundError()
@@ -545,7 +537,6 @@ class AgentRosterService:
                AgentDebugConversation.tenant_id == agent.tenant_id,
                AgentDebugConversation.agent_id == agent.id,
                AgentDebugConversation.account_id == account_id,
-                AgentDebugConversation.draft_type == draft_type,
            )
        )
        if mapping is not None:
@@ -579,7 +570,6 @@ class AgentRosterService:
                agent_id=agent.id,
                app_id=backing_app_id,
                account_id=account_id,
-                draft_type=draft_type,
                conversation_id=conversation_id,
            )
        )
@@ -587,15 +577,9 @@ class AgentRosterService:
        return conversation_id

    def get_or_create_agent_app_debug_conversation_id(
-        self,
-        *,
-        tenant_id: str,
-        agent_id: str,
-        account_id: str,
-        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
-        commit: bool = True,
+        self, *, tenant_id: str, agent_id: str, account_id: str, commit: bool = True
    ) -> str:
-        """Return the current editor's Build or Preview conversation for an Agent App."""
+        """Return the current editor's debug conversation for an Agent App."""

        agent = self._session.scalar(
            select(Agent).where(
@@ -607,24 +591,13 @@ class AgentRosterService:
        if agent is None:
            raise AgentNotFoundError()

-        conversation_id = self._get_or_create_agent_app_debug_conversation(
-            agent=agent,
-            account_id=account_id,
-            draft_type=draft_type,
-        )
+        conversation_id = self._get_or_create_agent_app_debug_conversation(agent=agent, account_id=account_id)
        if commit:
            self._session.commit()
        return conversation_id

-    def load_agent_app_debug_conversation_id(
-        self,
-        *,
-        tenant_id: str,
-        agent_id: str,
-        account_id: str,
-        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
-    ) -> str | None:
-        """Return the editor's existing scoped conversation without creating or repairing rows."""
+    def load_agent_app_debug_conversation_id(self, *, tenant_id: str, agent_id: str, account_id: str) -> str | None:
+        """Return the current editor's existing debug conversation without creating or repairing rows."""

        return self._session.scalar(
            select(Conversation.id)
@@ -633,7 +606,6 @@ class AgentRosterService:
                AgentDebugConversation.tenant_id == tenant_id,
                AgentDebugConversation.agent_id == agent_id,
                AgentDebugConversation.account_id == account_id,
-                AgentDebugConversation.draft_type == draft_type,
                AgentDebugConversation.app_id == Conversation.app_id,
                Conversation.from_source == ConversationFromSource.CONSOLE,
                Conversation.from_account_id == account_id,
@@ -654,21 +626,16 @@ class AgentRosterService:
        )

    def refresh_agent_app_debug_conversation_id(
-        self,
-        *,
-        tenant_id: str,
-        agent_id: str,
-        account_id: str,
-        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
-        commit: bool = True,
+        self, *, tenant_id: str, agent_id: str, account_id: str, commit: bool = True
    ) -> str:
-        """Start a new scoped console conversation for the current Agent App editor.
+        """Start a new console debug conversation for the current Agent App editor.

-        If this account already has a mapping for the requested draft surface, the previous
+        If this account already has a debug conversation mapping, the previous
        conversation is abandoned first: any ACTIVE conversation-owned Agent
        runtime sessions for that old conversation are sent through best-effort
-        backend cleanup and then retired locally even when enqueueing fails. The
-        other draft surface is left untouched.
+        backend cleanup and then retired locally even when enqueueing fails.
+        The debug mapping is then repointed to the freshly created
+        conversation.
        """

        agent = self._session.scalar(
@@ -696,7 +663,6 @@ class AgentRosterService:
                AgentDebugConversation.tenant_id == tenant_id,
                AgentDebugConversation.agent_id == agent_id,
                AgentDebugConversation.account_id == account_id,
-                AgentDebugConversation.draft_type == draft_type,
            )
        )
        if mapping is None:
@@ -706,7 +672,6 @@ class AgentRosterService:
                    agent_id=agent_id,
                    app_id=backing_app_id,
                    account_id=account_id,
-                    draft_type=draft_type,
                    conversation_id=conversation_id,
                )
            )
@@ -718,7 +683,6 @@ class AgentRosterService:
                    tenant_id=tenant_id,
                    agent_id=agent_id,
                    account_id=account_id,
-                    draft_type=draft_type,
                    app_id=previous_app_id or backing_app_id,
                    conversation_id=previous_conversation_id,
                )
@@ -735,7 +699,6 @@ class AgentRosterService:
        tenant_id: str,
        agent_id: str,
        account_id: str,
-        draft_type: AgentConfigDraftType,
        app_id: str,
        conversation_id: str,
    ) -> None:
@@ -764,8 +727,7 @@ class AgentRosterService:
                        session_snapshot=stored_session.session_snapshot,
                        runtime_layer_specs=stored_session.runtime_layer_specs,
                        idempotency_key=(
-                            f"{tenant_id}:{agent_id}:{account_id}:{draft_type.value}:{conversation_id}:"
-                            "debug-session-cleanup:"
+                            f"{tenant_id}:{agent_id}:{account_id}:{conversation_id}:debug-session-cleanup:"
                            f"{stored_session.scope.agent_id}:"
                            f"{stored_session.scope.agent_config_snapshot_id or 'no-config'}:"
                            f"{stored_session.backend_run_id or 'no-run'}"
@@ -776,7 +738,6 @@ class AgentRosterService:
                            "conversation_id": stored_session.scope.conversation_id,
                            "agent_id": stored_session.scope.agent_id,
                            "agent_config_snapshot_id": stored_session.scope.agent_config_snapshot_id,
-                            "draft_type": draft_type.value,
                            "previous_agent_backend_run_id": stored_session.backend_run_id,
                        },
                    )
@@ -811,14 +772,9 @@ class AgentRosterService:
                    )

    def load_or_create_agent_app_debug_conversation_ids_by_agent_id(
-        self,
-        *,
-        tenant_id: str,
-        agents: list[Agent],
-        account_id: str,
-        draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
+        self, *, tenant_id: str, agents: list[Agent], account_id: str
    ) -> dict[str, str]:
-        """Return per-account scoped conversations for a page of Agent Apps."""
+        """Return per-account debug conversations for a page of Agent Apps."""

        conversation_ids_by_agent_id: dict[str, str] = {}
        changed = False
@@ -828,7 +784,6 @@ class AgentRosterService:
            conversation_ids_by_agent_id[agent.id] = self._get_or_create_agent_app_debug_conversation(
                agent=agent,
                account_id=account_id,
-                draft_type=draft_type,
            )
            changed = True
        if changed:
@@ -911,14 +866,16 @@ class AgentRosterService:
            raise AgentNotFoundError()
        return app

-    def _get_runtime_resolvable_agent(self, *, tenant_id: str, agent_id: str) -> Agent | None:
-        """Load an Agent that is eligible to resolve to a runtime backing App.
+    def get_agent_runtime_app_model(self, *, tenant_id: str, agent_id: str) -> App:
+        """Resolve the App that backs an Agent runtime surface.

-        Shared by the runtime resolver and the read-only authorization resolver
-        so both agree on what counts as a resolvable Agent.
+        Roster Agents use their public Agent App. Workflow-only Agents use a
+        hidden Agent App stored in ``backing_app_id`` so console chat/logs can
+        reuse the app runtime without exposing the resource in workspace app
+        lists.
        """

-        return self._session.scalar(
+        agent = self._session.scalar(
            select(Agent)
            .where(
                Agent.tenant_id == tenant_id,
@@ -936,34 +893,6 @@ class AgentRosterService:
            )
            .limit(1)
        )
-
-    def peek_authz_app_id(self, *, tenant_id: str, agent_id: str) -> str | None:
-        """Resolve the App id whose access policy governs an Agent.
-
-        Roster Agents are governed by their own Agent App, while workflow-only
-        Agents are governed by their parent workflow App: the hidden runtime
-        backing App never receives a resource access policy, so it must not be
-        used for authorization. Stays read-only — unlike
-        :meth:`get_agent_runtime_app_model`, this never materializes the hidden
-        backing App. Returns ``None`` when the Agent does not resolve, leaving
-        the caller to decide how to treat it.
-        """
-
-        agent = self._get_runtime_resolvable_agent(tenant_id=tenant_id, agent_id=agent_id)
-        if agent is None:
-            return None
-        return agent.app_id
-
-    def get_agent_runtime_app_model(self, *, tenant_id: str, agent_id: str) -> App:
-        """Resolve the App that backs an Agent runtime surface.
-
-        Roster Agents use their public Agent App. Workflow-only Agents use a
-        hidden Agent App stored in ``backing_app_id`` so console chat/logs can
-        reuse the app runtime without exposing the resource in workspace app
-        lists.
-        """
-
-        agent = self._get_runtime_resolvable_agent(tenant_id=tenant_id, agent_id=agent_id)
        if agent is None:
            raise AgentNotFoundError()
        should_commit_backing_app = agent.scope == AgentScope.WORKFLOW_ONLY and not agent.backing_app_id
@@ -19,7 +19,6 @@ from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
 from core.file import remote_fetcher
 from core.plugin.entities.plugin import PluginDependency
-from core.rbac import RBACPermission
 from core.trigger.constants import (
    TRIGGER_PLUGIN_NODE_TYPE,
    TRIGGER_SCHEDULE_NODE_TYPE,
@@ -44,9 +43,7 @@ from services.agent.dsl_service import AgentDslService, AgentPackage
 from services.agent.workflow_publish_service import WorkflowAgentPublishService
 from services.dsl_content import DSL_MAX_SIZE, dsl_content_size
 from services.dsl_version import check_version_compatibility
-from services.enterprise.rbac_service import RBACService
 from services.entities.dsl_entities import CheckDependenciesResult, DslImportWarning, ImportMode, ImportStatus
-from services.errors.account import NoPermissionError
 from services.errors.app import WorkflowNotFoundError
 from services.plugin.dependencies_analysis import DependenciesAnalysisService
 from services.workflow_draft_variable_service import WorkflowDraftVariableService
@@ -304,9 +301,6 @@ class AppDslService:
                error=f"Invalid YAML format: {str(e)}",
            )

-        except NoPermissionError:
-            raise
-
        except Exception as e:
            logger.exception("Failed to import app")
            return Import(
@@ -370,9 +364,6 @@ class AppDslService:
                warnings=self._warnings,
            )

-        except NoPermissionError:
-            raise
-
        except Exception as e:
            logger.exception("Error confirming import")
            return Import(
@@ -404,21 +395,6 @@ class AppDslService:
            leaked_dependencies=leaked_dependencies,
        )

-    @staticmethod
-    def _ensure_agent_manage_permission(account: Account) -> None:
-        """Importing an Agent DSL creates a roster Agent, which requires ``agent.manage``."""
-        if not dify_config.RBAC_ENABLED:
-            return
-        if account.current_tenant_id is None:
-            raise ValueError("Current tenant is not set")
-        allowed = RBACService.CheckAccess.check(
-            account.current_tenant_id,
-            account.id,
-            scene=RBACPermission.AGENT_MANAGE,
-        )
-        if not allowed:
-            raise NoPermissionError("Agent management permission is required to import an Agent App")
-
    def _create_or_update_app(
        self,
        *,
@@ -439,8 +415,6 @@ class AppDslService:
        if not app_mode:
            raise ValueError("loss app mode")
        app_mode = AppMode(app_mode)
-        if app_mode == AppMode.AGENT:
-            self._ensure_agent_manage_permission(account)

        # Set icon type
        icon_type_value = icon_type or app_data.get("icon_type")
@@ -408,7 +408,6 @@ class AppService:
        default_model_config = app_template.get("model_config")
        default_model_config = default_model_config.copy() if default_model_config else None
        if default_model_config and "model" in default_model_config:
-            default_model_dict = default_model_config["model"]
            # get model provider
            model_manager = ModelManager.for_tenant(tenant_id=account.current_tenant_id or "")

@@ -423,7 +422,7 @@ class AppService:
                logger.exception("Get default model instance failed, tenant_id: %s", tenant_id)
                model_instance = None

-            if model_instance is not None:
+            if model_instance:
                if (
                    model_instance.model_name == default_model_config["model"]["name"]
                    and model_instance.provider == default_model_config["model"]["provider"]
@@ -431,28 +430,17 @@ class AppService:
                    default_model_dict = default_model_config["model"]
                else:
                    llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
-                    try:
-                        model_schema = llm_model.get_model_schema(model_instance.model_name, model_instance.credentials)
-                        if model_schema is None:
-                            raise ValueError(f"model schema not found for model {model_instance.model_name}")
-                    except Exception:
-                        # A removed provider model must not prevent creating an app.
-                        logger.warning(
-                            "Default model schema is unavailable, tenant_id: %s, provider: %s, model: %s",
-                            tenant_id,
-                            model_instance.provider,
-                            model_instance.model_name,
-                            exc_info=True,
-                        )
-                        model_instance = None
-                    else:
-                        default_model_dict = {
-                            "provider": model_instance.provider,
-                            "name": model_instance.model_name,
-                            "mode": model_schema.model_properties.get(ModelPropertyKey.MODE),
-                            "completion_params": {},
-                        }
-            if model_instance is None:
+                    model_schema = llm_model.get_model_schema(model_instance.model_name, model_instance.credentials)
+                    if model_schema is None:
+                        raise ValueError(f"model schema not found for model {model_instance.model_name}")
+
+                    default_model_dict = {
+                        "provider": model_instance.provider,
+                        "name": model_instance.model_name,
+                        "mode": model_schema.model_properties.get(ModelPropertyKey.MODE),
+                        "completion_params": {},
+                    }
+            else:
                try:
                    provider, model = model_manager.get_default_provider_model_name(
                        tenant_id=account.current_tenant_id or "", model_type=ModelType.LLM
@@ -217,18 +217,12 @@ class BillingService:
        return _billing_info_adapter.validate_python(billing_info)

    @classmethod
-    def get_vector_space(cls, tenant_id: str, bypass_cache: bool = False) -> _VectorSpaceQuota:
+    def get_vector_space(cls, tenant_id: str) -> _VectorSpaceQuota:
        params = {"tenant_id": tenant_id}
-        if bypass_cache:
-            params["bypass_cache"] = "true"
        return _vector_space_quota_adapter.validate_python(
            cls._send_request("GET", "/subscription/vector-space", params=params)
        )

-    @classmethod
-    def invalidate_vector_space_cache(cls, tenant_id: str) -> None:
-        cls.get_vector_space(tenant_id, bypass_cache=True)
-
    @classmethod
    def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
        """Deprecated: Use get_quota_info instead."""
@@ -1974,10 +1974,7 @@ class DocumentService:
                if data_source_info and "upload_file_id" in data_source_info:
                    file_id = data_source_info["upload_file_id"]
        document_was_deleted.send(
-            document.id,
-            dataset_id=document.dataset_id,
-            doc_form=document.doc_form,
-            file_id=file_id,
+            document.id, dataset_id=document.dataset_id, doc_form=document.doc_form, file_id=file_id
        )

        session.delete(document)
@@ -2016,12 +2013,7 @@ class DocumentService:
        # Dispatch cleanup task after commit to avoid lock contention
        # Task cleans up segments, files, and vector indexes
        if deleted_document_ids and doc_form is not None:
-            batch_clean_document_task.delay(
-                deleted_document_ids,
-                dataset_ref.dataset_id,
-                doc_form,
-                file_ids,
-            )
+            batch_clean_document_task.delay(deleted_document_ids, dataset_ref.dataset_id, doc_form, file_ids)

    @staticmethod
    def rename_document(dataset_id: str, document_id: str, name: str, session: Session) -> Document:
@@ -5,7 +5,6 @@ from typing import Any

 import httpx

-from configs import dify_config
 from core.helper.trace_id_helper import generate_traceparent_header
 from services.errors.enterprise import (
    EnterpriseAPIBadRequestError,
@@ -97,14 +96,12 @@ class BaseRequest:
            logger.debug("Failed to generate traceparent header", exc_info=True)

        with httpx.Client(mounts=mounts) as client:
-            # Callers that pass an explicit timeout keep it; everyone else gets the
-            # configured budget rather than httpx's implicit 5s default.
-            request_kwargs: dict[str, Any] = {
-                "json": json,
-                "params": params,
-                "headers": headers,
-                "timeout": timeout if timeout is not None else dify_config.ENTERPRISE_REQUEST_TIMEOUT,
-            }
+            # IMPORTANT:
+            # - In httpx, passing timeout=None disables timeouts (infinite) and overrides the library default.
+            # - To preserve httpx's default timeout behavior for existing call sites, only pass the kwarg when set.
+            request_kwargs: dict[str, Any] = {"json": json, "params": params, "headers": headers}
+            if timeout is not None:
+                request_kwargs["timeout"] = timeout

            response = client.request(method, url, **request_kwargs)

@@ -209,8 +206,9 @@ class EnterpriseRequest(BaseRequest):
                "json": json,
                "params": params,
                "headers": {"Content-Type": "application/json", cls.secret_key_header: cls.secret_key, **inner_headers},
-                "timeout": timeout if timeout is not None else dify_config.ENTERPRISE_RBAC_REQUEST_TIMEOUT,
            }
+            if timeout is not None:
+                request_kwargs["timeout"] = timeout
            response = client.request(method, url, **request_kwargs)
            if not response.is_success:
                cls._handle_error_response(response)
@@ -317,6 +317,9 @@ _LEGACY_WORKSPACE_OWNER_KEYS: list[str] = [
    "credential.use",
    "credential.create",
    "credential.manage",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
    "app.acl.preview",
    "app_library.access",
    "app.create_and_management",
@@ -330,7 +333,6 @@ _LEGACY_WORKSPACE_OWNER_KEYS: list[str] = [
    "snippets.management",
    "tool.manage",
    "mcp.manage",
-    "agent.manage",
 ]

 _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
@@ -347,6 +349,9 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
    "credential.use",
    "credential.create",
    "credential.manage",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
    "app_library.access",
    "app.create_and_management",
    "app.tag.manage",
@@ -358,7 +363,6 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
    "snippets.management",
    "tool.manage",
    "mcp.manage",
-    "agent.manage",
 ]

 _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
@@ -373,7 +377,9 @@ _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
    "dataset.external.connect",
    "snippets.create_and_modify",
    "tool.manage",
-    "agent.manage",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
 ]

 _LEGACY_WORKSPACE_NORMAL_KEYS: list[str] = [
@@ -381,6 +387,9 @@ _LEGACY_WORKSPACE_NORMAL_KEYS: list[str] = [
    "plugin.install",
    "credential.use",
    "app_library.access",
+    "billing.view",
+    "billing.subscription.manage",
+    "billing.manage",
 ]

 _LEGACY_WORKSPACE_DATASET_OPERATOR_KEYS: list[str] = [
@@ -796,6 +805,7 @@ class RBACService:
            data = _inner_call(
                "GET",
                f"{_INNER_PREFIX}/role-permissions/catalog",
+                params={"billing_enabled": dify_config.BILLING_ENABLED},
                tenant_id=tenant_id,
                account_id=account_id,
            )
@@ -93,20 +93,8 @@ class ExternalDatasetService:
                raise ValueError(f"invalid endpoint: {endpoint} must start with http:// or https://")
            else:
                raise ValueError(f"invalid endpoint: {endpoint}")
-        # Send a minimal body shaped like the External Knowledge API retrieval contract so providers
-        # that require a JSON payload (e.g. RAGFlow) accept the validation probe instead of rejecting
-        # a body-less POST. Mirrors the request built in fetch_external_knowledge_retrieval.
-        validation_payload = {
-            "knowledge_id": "",
-            "query": "",
-            "retrieval_setting": {"top_k": 1, "score_threshold": 0.0},
-        }
        try:
-            response = ssrf_proxy.post(
-                endpoint,
-                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
-                data=json.dumps(validation_payload),
-            )
+            response = ssrf_proxy.post(endpoint, headers={"Authorization": f"Bearer {api_key}"})
        except Exception as e:
            raise ValueError(f"failed to connect to the endpoint: {endpoint}") from e
        if response.status_code == 502:
@@ -75,12 +75,6 @@ class LicenseStatus(StrEnum):
    LOST = "lost"


-class DeploymentEdition(StrEnum):
-    COMMUNITY = "COMMUNITY"
-    ENTERPRISE = "ENTERPRISE"
-    CLOUD = "CLOUD"
-
-
 class LicenseModel(FeatureResponseModel):
    status: LicenseStatus = LicenseStatus.NONE
    expired_at: str = ""
@@ -106,7 +100,6 @@ class WebAppAuthModel(FeatureResponseModel):
    sso_config: WebAppAuthSSOModel = WebAppAuthSSOModel()
    allow_email_code_login: bool = False
    allow_email_password_login: bool = False
-    allow_public_access: bool = True


 class KnowledgePipeline(FeatureResponseModel):
@@ -168,7 +161,6 @@ class PluginManagerModel(FeatureResponseModel):


 class SystemFeatureModel(FeatureResponseModel):
-    deployment_edition: DeploymentEdition
    enable_app_deploy: bool = False
    sso_enforced_for_signin: bool = False
    sso_enforced_for_signin_protocol: str = ""
@@ -191,9 +183,7 @@ class SystemFeatureModel(FeatureResponseModel):
    enable_trial_app: bool = False
    enable_explore_banner: bool = False
    enable_learn_app: bool = True
-    enable_step_by_step_tour: bool = False
    rbac_enabled: bool = False
-    knowledge_fs_enabled: bool = False


 class FeatureService:
@@ -259,7 +249,7 @@ class FeatureService:

    @classmethod
    def get_system_features(cls, is_authenticated: bool = False) -> SystemFeatureModel:
-        system_features = SystemFeatureModel(deployment_edition=cls._resolve_deployment_edition())
+        system_features = SystemFeatureModel()
        system_features.rbac_enabled = dify_config.RBAC_ENABLED

        cls._fulfill_system_params_from_env(system_features)
@@ -279,14 +269,6 @@ class FeatureService:

        return system_features

-    @classmethod
-    def _resolve_deployment_edition(cls) -> DeploymentEdition:
-        if dify_config.EDITION == "CLOUD":
-            return DeploymentEdition.CLOUD
-        if dify_config.ENTERPRISE_ENABLED:
-            return DeploymentEdition.ENTERPRISE
-        return DeploymentEdition.COMMUNITY
-
    @classmethod
    def get_app_dsl_version(cls) -> str:
        return CURRENT_APP_DSL_VERSION
@@ -300,13 +282,9 @@ class FeatureService:
        system_features.is_allow_register = dify_config.ALLOW_REGISTER
        system_features.is_allow_create_workspace = dify_config.ALLOW_CREATE_WORKSPACE
        system_features.is_email_setup = dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""
-        system_features.enable_change_email = dify_config.ENABLE_CHANGE_EMAIL
        system_features.enable_trial_app = dify_config.ENABLE_TRIAL_APP
        system_features.enable_explore_banner = dify_config.ENABLE_EXPLORE_BANNER
        system_features.enable_learn_app = dify_config.ENABLE_LEARN_APP
-        system_features.webapp_auth.allow_public_access = dify_config.WEBAPP_PUBLIC_ACCESS_ENABLED
-        system_features.enable_step_by_step_tour = dify_config.ENABLE_STEP_BY_STEP_TOUR
-        system_features.knowledge_fs_enabled = dify_config.KNOWLEDGE_FS_ENABLED

    @classmethod
    def _fulfill_trial_models_from_env(cls) -> list[str]:
@@ -141,29 +141,6 @@ class FileService:
        blob = storage.load_once(upload_file_key)
        return base64.b64encode(blob).decode()

-    def get_file_presigned_url(self, *, file_id: str, tenant_id: str) -> str:
-        """Generate a direct storage URL for a tenant-owned upload file."""
-        with self._session_maker(expire_on_commit=False) as session:
-            upload_file = session.scalar(
-                select(UploadFile)
-                .where(
-                    UploadFile.id == file_id,
-                    UploadFile.tenant_id == tenant_id,
-                )
-                .limit(1)
-            )
-            if upload_file is None:
-                raise NotFound("File not found")
-
-            file_key = upload_file.key
-            content_type = upload_file.mime_type
-
-        return storage.generate_presigned_url(
-            file_key,
-            expires_in=dify_config.FILES_ACCESS_TIMEOUT,
-            content_type=content_type,
-        )
-
    def upload_text(self, text: str, text_name: str, user_id: str, tenant_id: str) -> UploadFile:
        if len(text_name) > 200:
            text_name = text_name[:200]
@@ -1,502 +0,0 @@
-"""Product-facing KnowledgeFS operation and authorization declarations.
-
-This registry is Dify's explicit Console surface. Transport concerns live in
-knowledge_fs_proxy so contract review does not require reading proxy mechanics.
-"""
-
-from __future__ import annotations
-
-from typing import Final, Literal, NamedTuple
-
-from core.rbac import RBACPermission
-
-type KnowledgeFSMethod = Literal["DELETE", "GET", "PATCH", "POST", "PUT"]
-type KnowledgeFSResponseKind = Literal["binary", "buffered", "stream"]
-type KnowledgeFSRequiredScope = Literal["knowledge-spaces:read", "knowledge-spaces:write"]
-type KnowledgeFSLegacyRole = Literal["reader", "dataset_editor", "admin"]
-type KnowledgeFSErrorStatusMap = tuple[tuple[int, int], ...]
-
-
-class KnowledgeFSOperation(NamedTuple):
-    operation_id: str
-    method: KnowledgeFSMethod
-    path: str
-    response_kind: KnowledgeFSResponseKind
-    required_scope: KnowledgeFSRequiredScope
-    rbac_permission: RBACPermission
-    legacy_role: KnowledgeFSLegacyRole
-    max_response_bytes: int
-    request_headers: tuple[str, ...]
-    response_headers: tuple[str, ...]
-    response_media_types: tuple[str, ...]
-    error_status_map: KnowledgeFSErrorStatusMap
-
-
-def _console_operation(
-    operation_id: str,
-    method: KnowledgeFSMethod,
-    path: str,
-    *,
-    rbac_permission: RBACPermission,
-    legacy_role: KnowledgeFSLegacyRole,
-    max_response_bytes: int = 1_048_576,
-    request_headers: tuple[str, ...] = ("x-trace-id",),
-    response_kind: KnowledgeFSResponseKind = "buffered",
-    response_media_types: tuple[str, ...] = ("application/json",),
-    error_status_map: KnowledgeFSErrorStatusMap = ((401, 502), (403, 403)),
-) -> KnowledgeFSOperation:
-    """Declare one contract-pinned operation with an explicit Dify authorization policy."""
-    is_read = method == "GET"
-    return KnowledgeFSOperation(
-        operation_id=operation_id,
-        method=method,
-        path=path,
-        response_kind=response_kind,
-        required_scope="knowledge-spaces:read" if is_read else "knowledge-spaces:write",
-        rbac_permission=rbac_permission,
-        legacy_role=legacy_role,
-        max_response_bytes=max_response_bytes,
-        request_headers=request_headers,
-        response_headers=("x-trace-id",),
-        response_media_types=response_media_types,
-        error_status_map=error_status_map,
-    )
-
-
-def _dataset_read_operation(operation_id: str, path: str) -> KnowledgeFSOperation:
-    """Declare a dataset-readable buffered JSON operation."""
-    return _console_operation(
-        operation_id,
-        "GET",
-        path,
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    )
-
-
-def _dataset_edit_operation(
-    operation_id: str,
-    method: KnowledgeFSMethod,
-    path: str,
-    *,
-    request_headers: tuple[str, ...] = ("x-trace-id",),
-) -> KnowledgeFSOperation:
-    """Declare a dataset-editable buffered JSON operation."""
-    return _console_operation(
-        operation_id,
-        method,
-        path,
-        rbac_permission=RBACPermission.DATASET_EDIT,
-        legacy_role="dataset_editor",
-        request_headers=request_headers,
-    )
-
-
-def _external_source_operation(
-    operation_id: str,
-    method: KnowledgeFSMethod,
-    path: str,
-    *,
-    request_headers: tuple[str, ...] = ("x-trace-id",),
-) -> KnowledgeFSOperation:
-    """Declare a source-connection operation restricted to dataset editors."""
-    return _console_operation(
-        operation_id,
-        method,
-        path,
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-        request_headers=request_headers,
-    )
-
-
-KNOWLEDGE_FS_CONSOLE_OPERATIONS: Final[tuple[KnowledgeFSOperation, ...]] = (
-    _console_operation(
-        operation_id="listKnowledgeSpaces",
-        method="GET",
-        path="knowledge-spaces",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _console_operation(
-        operation_id="createKnowledgeSpace",
-        method="POST",
-        path="knowledge-spaces",
-        rbac_permission=RBACPermission.DATASET_CREATE_AND_MANAGEMENT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesById",
-        method="GET",
-        path="knowledge-spaces/{id}",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _dataset_edit_operation("patchKnowledgeSpacesById", "PATCH", "knowledge-spaces/{id}"),
-    _dataset_edit_operation(
-        "deleteKnowledgeSpacesById",
-        "DELETE",
-        "knowledge-spaces/{id}",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _dataset_read_operation("getKnowledgeSpacesByIdStats", "knowledge-spaces/{id}/stats"),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdAccessPolicy",
-        method="GET",
-        path="knowledge-spaces/{id}/access-policy",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _console_operation(
-        operation_id="patchKnowledgeSpacesByIdAccessPolicy",
-        method="PATCH",
-        path="knowledge-spaces/{id}/access-policy",
-        rbac_permission=RBACPermission.DATASET_ACCESS_CONFIG,
-        legacy_role="admin",
-    ),
-    _console_operation(
-        operation_id="getSourceProviders",
-        method="GET",
-        path="source-providers",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdSourceConnections",
-        method="GET",
-        path="knowledge-spaces/{id}/source-connections",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourceConnections",
-        method="POST",
-        path="knowledge-spaces/{id}/source-connections",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourceConnectionsOauth",
-        "POST",
-        "knowledge-spaces/{id}/source-connections/oauth",
-    ),
-    _external_source_operation("postSourceOauthCallback", "POST", "source-oauth/callback"),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourceConnectionsByConnectionId",
-        "GET",
-        "knowledge-spaces/{id}/source-connections/{connectionId}",
-    ),
-    _external_source_operation(
-        "deleteKnowledgeSpacesByIdSourceConnectionsByConnectionId",
-        "DELETE",
-        "knowledge-spaces/{id}/source-connections/{connectionId}",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourceConnectionsByConnectionIdRefresh",
-        method="POST",
-        path="knowledge-spaces/{id}/source-connections/{connectionId}/refresh",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdSources",
-        method="GET",
-        path="knowledge-spaces/{id}/sources",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSources",
-        method="POST",
-        path="knowledge-spaces/{id}/sources",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourcesBySourceId",
-        "GET",
-        "knowledge-spaces/{id}/sources/{sourceId}",
-    ),
-    _external_source_operation(
-        "patchKnowledgeSpacesByIdSourcesBySourceId",
-        "PATCH",
-        "knowledge-spaces/{id}/sources/{sourceId}",
-    ),
-    _external_source_operation(
-        "deleteKnowledgeSpacesByIdSourcesBySourceId",
-        "DELETE",
-        "knowledge-spaces/{id}/sources/{sourceId}",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _external_source_operation(
-        "putKnowledgeSpacesByIdSourcesBySourceIdCredentials",
-        "PUT",
-        "knowledge-spaces/{id}/sources/{sourceId}/credentials",
-    ),
-    _external_source_operation(
-        "deleteKnowledgeSpacesByIdSourcesBySourceIdCredentials",
-        "DELETE",
-        "knowledge-spaces/{id}/sources/{sourceId}/credentials",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdSync",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/sync",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourcesBySourceIdCrawlPreview",
-        method="POST",
-        path="knowledge-spaces/{id}/sources/{sourceId}/crawl-preview",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdWorkflowImports",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/workflow-imports",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourcesBySourceIdPages",
-        "GET",
-        "knowledge-spaces/{id}/sources/{sourceId}/pages",
-    ),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourcesBySourceIdFiles",
-        "GET",
-        "knowledge-spaces/{id}/sources/{sourceId}/files",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdCrawl",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/crawl",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdImport",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/import",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdTest",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/test",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBySourceIdImportFiles",
-        "POST",
-        "knowledge-spaces/{id}/sources/{sourceId}/import-files",
-    ),
-    _external_source_operation(
-        "postKnowledgeSpacesByIdSourcesBulk",
-        "POST",
-        "knowledge-spaces/{id}/sources/bulk",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourceWorkflows",
-        "GET",
-        "knowledge-spaces/{id}/source-workflows",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdSourceWorkflowsByRunId",
-        method="GET",
-        path="knowledge-spaces/{id}/source-workflows/{runId}",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _external_source_operation(
-        "getKnowledgeSpacesByIdSourceWorkflowsByRunIdBulkItems",
-        "GET",
-        "knowledge-spaces/{id}/source-workflows/{runId}/bulk-items",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdSourceWorkflowsByRunIdPages",
-        method="GET",
-        path="knowledge-spaces/{id}/source-workflows/{runId}/pages",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdCancel",
-        method="POST",
-        path="knowledge-spaces/{id}/source-workflows/{runId}/cancel",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdRetry",
-        method="POST",
-        path="knowledge-spaces/{id}/source-workflows/{runId}/retry",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdSourceWorkflowsByRunIdSelection",
-        method="POST",
-        path="knowledge-spaces/{id}/source-workflows/{runId}/selection",
-        rbac_permission=RBACPermission.DATASET_EXTERNAL_CONNECT,
-        legacy_role="dataset_editor",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
-        method="GET",
-        path="knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _console_operation(
-        operation_id="putKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
-        method="PUT",
-        path="knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
-        rbac_permission=RBACPermission.DATASET_EDIT,
-        legacy_role="dataset_editor",
-    ),
-    _dataset_read_operation("getKnowledgeSpacesByIdDocuments", "knowledge-spaces/{id}/documents"),
-    _dataset_edit_operation("postKnowledgeSpacesByIdDocuments", "POST", "knowledge-spaces/{id}/documents"),
-    _dataset_edit_operation(
-        "deleteKnowledgeSpacesByIdDocumentsBulk",
-        "DELETE",
-        "knowledge-spaces/{id}/documents/bulk",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _dataset_edit_operation(
-        "postKnowledgeSpacesByIdDocumentsBulk",
-        "POST",
-        "knowledge-spaces/{id}/documents/bulk",
-    ),
-    _dataset_edit_operation(
-        "postKnowledgeSpacesByIdDocumentsBulkReindex",
-        "POST",
-        "knowledge-spaces/{id}/documents/bulk/reindex",
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentId",
-        "knowledge-spaces/{id}/documents/{documentId}",
-    ),
-    _dataset_edit_operation(
-        "deleteKnowledgeSpacesByIdDocumentsByDocumentId",
-        "DELETE",
-        "knowledge-spaces/{id}/documents/{documentId}",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdLogicalDocuments",
-        method="GET",
-        path="knowledge-spaces/{id}/logical-documents",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _dataset_edit_operation(
-        "deleteKnowledgeSpacesByIdLogicalDocumentsByDocumentId",
-        "DELETE",
-        "knowledge-spaces/{id}/logical-documents/{documentId}",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentIdOutline",
-        "knowledge-spaces/{id}/documents/{documentId}/outline",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdLogicalDocumentsByDocumentId",
-        method="GET",
-        path="knowledge-spaces/{id}/logical-documents/{documentId}",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdRevisions",
-        method="GET",
-        path="knowledge-spaces/{id}/documents/{documentId}/revisions",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _dataset_edit_operation(
-        "postKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionRollback",
-        "POST",
-        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/rollback",
-    ),
-    _dataset_edit_operation(
-        "patchKnowledgeSpacesByIdDocumentsByDocumentIdMetadata",
-        "PATCH",
-        "knowledge-spaces/{id}/documents/{documentId}/metadata",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunks",
-        method="GET",
-        path="knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunksByChunkId",
-        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks/{chunkId}",
-    ),
-    _dataset_edit_operation(
-        "postKnowledgeSpacesByIdDocumentsByDocumentIdRevisionsByRevisionChunksByChunkIdState",
-        "POST",
-        "knowledge-spaces/{id}/documents/{documentId}/revisions/{revision}/chunks/{chunkId}/state",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdProcessingTasks",
-        method="GET",
-        path="knowledge-spaces/{id}/processing-tasks",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasks",
-        "knowledge-spaces/{id}/documents/{documentId}/processing-tasks",
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskId",
-        "knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}",
-    ),
-    _console_operation(
-        operation_id="getKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskIdEvents",
-        method="GET",
-        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}/events",
-        rbac_permission=RBACPermission.DATASET_READONLY,
-        legacy_role="reader",
-        max_response_bytes=67_108_864,
-        request_headers=("last-event-id", "x-trace-id"),
-        response_kind="stream",
-        response_media_types=("text/event-stream",),
-    ),
-    _console_operation(
-        operation_id="deleteKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskId",
-        method="DELETE",
-        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}",
-        rbac_permission=RBACPermission.DATASET_EDIT,
-        legacy_role="dataset_editor",
-    ),
-    _console_operation(
-        operation_id="postKnowledgeSpacesByIdDocumentsByDocumentIdProcessingTasksByTaskIdRetry",
-        method="POST",
-        path="knowledge-spaces/{id}/documents/{documentId}/processing-tasks/{taskId}/retry",
-        rbac_permission=RBACPermission.DATASET_EDIT,
-        legacy_role="dataset_editor",
-    ),
-    _dataset_read_operation(
-        "getKnowledgeSpacesByIdDocumentsByDocumentIdSettings",
-        "knowledge-spaces/{id}/documents/{documentId}/settings",
-    ),
-    _dataset_edit_operation(
-        "putKnowledgeSpacesByIdDocumentsByDocumentIdSettings",
-        "PUT",
-        "knowledge-spaces/{id}/documents/{documentId}/settings",
-    ),
-    _dataset_read_operation("getJobsById", "jobs/{id}"),
-    _dataset_edit_operation("deleteJobsById", "DELETE", "jobs/{id}"),
-    _dataset_edit_operation("postJobsByIdRetry", "POST", "jobs/{id}/retry"),
-    _dataset_read_operation("getDeletionJobsByJobId", "deletion-jobs/{jobId}"),
-    _dataset_edit_operation(
-        "postDeletionJobsByJobIdRetry",
-        "POST",
-        "deletion-jobs/{jobId}/retry",
-        request_headers=("idempotency-key", "x-trace-id"),
-    ),
-    _dataset_read_operation("getBulkJobsById", "bulk-jobs/{id}"),
-)
@@ -1,408 +0,0 @@
-"""Authorize and forward the explicitly enabled KnowledgeFS Console operations.
-
-The dedicated request path uses Dify's shared SSRF policy, never follows redirects,
-bounds buffered responses, and rejects compressed streaming responses.
-"""
-
-from __future__ import annotations
-
-from collections.abc import Iterable, Mapping
-from dataclasses import dataclass
-from datetime import UTC, datetime, timedelta
-from http import HTTPStatus
-from typing import NamedTuple, Protocol
-
-import httpx
-import jwt
-
-from configs import dify_config
-from core.helper import ssrf_proxy
-from core.rbac import RBACResourceScope
-from core.tools.errors import ToolSSRFError
-from models import Account
-from services.enterprise.rbac_service import RBACService
-from services.knowledge_fs_operations import (
-    KNOWLEDGE_FS_CONSOLE_OPERATIONS,
-    KnowledgeFSMethod,
-    KnowledgeFSOperation,
-    KnowledgeFSResponseKind,
-)
-
-_JWT_AUDIENCE = "knowledge-fs"
-_JWT_ISSUER = "dify"
-_JWT_TTL_SECONDS = 60
-_MAX_BUFFERED_RESPONSE_BYTES = 1024 * 1024
-
-
-class KnowledgeFSUpstreamResponse(NamedTuple):
-    response: httpx.Response
-    response_kind: KnowledgeFSResponseKind
-    operation: KnowledgeFSOperation
-
-
-_AUTHORIZATION_MARKER = object()
-
-
-@dataclass(eq=False, frozen=True, init=False, slots=True)
-class KnowledgeFSAuthorization:
-    """Single-use forwarding capability created after Dify workspace policy checks.
-
-    Callers obtain this value from :func:`authorize_knowledge_fs_request`. Direct
-    construction and repeated forwarding are rejected before outbound I/O.
-    """
-
-    account_id: str
-    tenant_id: str
-    operation: KnowledgeFSOperation
-    _used: bool
-
-    def __init__(
-        self,
-        account_id: str,
-        tenant_id: str,
-        operation: KnowledgeFSOperation,
-        *,
-        _marker: object | None = None,
-    ) -> None:
-        if _marker is not _AUTHORIZATION_MARKER:
-            raise KnowledgeFSAccessDeniedError("KnowledgeFS authorization must be created by workspace authorization")
-        object.__setattr__(self, "account_id", account_id)
-        object.__setattr__(self, "tenant_id", tenant_id)
-        object.__setattr__(self, "operation", operation)
-        object.__setattr__(self, "_used", False)
-
-    def consume(self) -> tuple[str, str, KnowledgeFSOperation]:
-        """Return the authorized principals and canonical operation exactly once.
-
-        Raises:
-            KnowledgeFSAccessDeniedError: The capability was already consumed.
-        """
-        if self._used:
-            raise KnowledgeFSAccessDeniedError("KnowledgeFS authorization has already been used")
-        object.__setattr__(self, "_used", True)
-        return self.account_id, self.tenant_id, self.operation
-
-
-class _RequestHeaders(Protocol):
-    def items(self) -> Iterable[tuple[str, str]]: ...
-
-
-class KnowledgeFSConfigurationError(RuntimeError):
-    """KnowledgeFS is incompletely configured or blocked by outbound policy."""
-
-
-class KnowledgeFSTimeoutError(RuntimeError):
-    """KnowledgeFS exceeded the configured request timeout."""
-
-
-class KnowledgeFSTransportError(RuntimeError):
-    """KnowledgeFS could not be reached or returned a response outside safety bounds."""
-
-
-class KnowledgeFSRouteNotAllowedError(RuntimeError):
-    """The requested path is outside the Console-visible KnowledgeFS surface."""
-
-
-class KnowledgeFSAccessDeniedError(RuntimeError):
-    """The Dify account lacks the workspace permission required by the operation."""
-
-
-def authorize_knowledge_fs_request(
-    *,
-    account: Account,
-    tenant_id: str,
-    method: KnowledgeFSMethod,
-    path: str,
-) -> KnowledgeFSAuthorization:
-    """Enforce Dify's workspace policy before KFS performs resource authorization.
-
-    Args:
-        account: Authenticated Dify account with its current workspace role.
-        tenant_id: Current Dify workspace identifier.
-        method: Requested upstream HTTP method.
-        path: Requested relative KnowledgeFS path.
-
-    Raises:
-        KnowledgeFSRouteNotAllowedError: The method and path do not resolve to a declared operation.
-        KnowledgeFSAccessDeniedError: The account lacks a required legacy or enterprise permission.
-
-    Returns:
-        A request-scoped capability binding the authorized account, workspace, and operation.
-    """
-    operation = get_knowledge_fs_operation(method, path)
-    if operation.legacy_role == "dataset_editor" and not account.is_dataset_editor:
-        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation requires dataset edit access")
-    if operation.legacy_role == "admin" and not account.is_admin_or_owner:
-        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation requires workspace administration access")
-    if not RBACService.CheckAccess.check(
-        tenant_id,
-        account.id,
-        scene=operation.rbac_permission.value,
-        resource_type=RBACResourceScope.DATASET.value,
-    ):
-        raise KnowledgeFSAccessDeniedError("KnowledgeFS operation is denied by workspace RBAC")
-    return KnowledgeFSAuthorization(account.id, tenant_id, operation, _marker=_AUTHORIZATION_MARKER)
-
-
-def proxy_knowledge_fs_request(
-    *,
-    account: Account,
-    method: KnowledgeFSMethod,
-    path: str,
-    tenant_id: str,
-    accept: str | None = None,
-    content_type: str | None = None,
-    query: bytes | None = None,
-    body: bytes | None = None,
-    request_headers: _RequestHeaders | None = None,
-) -> KnowledgeFSUpstreamResponse:
-    """Authorize and forward one allowlisted KnowledgeFS request as a single use case."""
-    authorization = authorize_knowledge_fs_request(
-        account=account,
-        tenant_id=tenant_id,
-        method=method,
-        path=path,
-    )
-
-    return proxy_authorized_knowledge_fs_request(
-        authorization=authorization,
-        accept=accept,
-        content_type=content_type,
-        query=query,
-        body=body,
-        request_headers=request_headers,
-    )
-
-
-def proxy_authorized_knowledge_fs_request(
-    *,
-    authorization: KnowledgeFSAuthorization,
-    accept: str | None = None,
-    content_type: str | None = None,
-    query: bytes | None = None,
-    body: bytes | None = None,
-    request_headers: _RequestHeaders | None = None,
-) -> KnowledgeFSUpstreamResponse:
-    """Forward one request whose operation and workspace policy were already authorized.
-
-    This performs one outbound KnowledgeFS request and does not repeat Dify RBAC checks.
-
-    Args:
-        authorization: Request-scoped capability returned by :func:`authorize_knowledge_fs_request`.
-        accept: Original Accept header, when present.
-        content_type: Original request Content-Type header, when present.
-        query: Original encoded query string from the Console request.
-        body: Original request body, when present.
-        request_headers: Incoming headers; only names declared by the operation are forwarded.
-
-    Returns:
-        The bounded KnowledgeFS response together with its transport metadata.
-
-    Raises:
-        KnowledgeFSConfigurationError: The connection is incomplete or blocked by outbound policy.
-        KnowledgeFSRouteNotAllowedError: A forwarded request header is outside the operation contract.
-        KnowledgeFSTimeoutError: KnowledgeFS exceeds the configured timeout.
-        KnowledgeFSTransportError: The request fails or its response violates transport bounds.
-    """
-    account_id, tenant_id, operation = authorization.consume()
-    incoming_request_headers = {name.lower(): value for name, value in (request_headers or {}).items()}
-    contract_request_headers = {
-        name: incoming_request_headers[name] for name in operation.request_headers if name in incoming_request_headers
-    }
-    return _forward_knowledge_fs_request(
-        account_id=account_id,
-        method=operation.method,
-        path=operation.path,
-        tenant_id=tenant_id,
-        accept=accept,
-        content_type=content_type,
-        query=query,
-        body=body,
-        request_headers=contract_request_headers,
-    )
-
-
-def _forward_knowledge_fs_request(
-    *,
-    account_id: str,
-    method: KnowledgeFSMethod,
-    path: str,
-    tenant_id: str,
-    accept: str | None = None,
-    content_type: str | None = None,
-    query: bytes | None = None,
-    body: bytes | None = None,
-    request_headers: Mapping[str, str] | None = None,
-) -> KnowledgeFSUpstreamResponse:
-    """Forward one fixed-route request without parsing its KnowledgeFS payload.
-
-    Args:
-        account_id: Current Dify account used as the KFS member identity.
-        method: Allowlisted upstream HTTP method.
-        path: Relative KnowledgeFS path under an allowlisted product surface.
-        tenant_id: Current Dify workspace used as the KFS tenant identity.
-        accept: Original Accept header, when present.
-        content_type: Original request Content-Type header, when present.
-        query: Original encoded query string from the Console request.
-        body: Original request body, when present.
-        request_headers: Contract-declared request headers forwarded by the Console adapter.
-
-    Returns:
-        The KnowledgeFS response and its actual transport kind. Non-success responses are buffered.
-
-    Raises:
-        KnowledgeFSConfigurationError: The connection is incomplete or blocked by outbound policy.
-        KnowledgeFSRouteNotAllowedError: The path is outside the allowlisted product surface.
-        KnowledgeFSTimeoutError: KnowledgeFS exceeds the configured timeout.
-        KnowledgeFSTransportError: The request fails or its response cannot be safely bounded.
-
-    Each request is bound to stable Dify account and workspace principals with a short expiration.
-    """
-    operation = get_knowledge_fs_operation(method, path)
-    base_url = dify_config.KNOWLEDGE_FS_BASE_URL
-    jwt_secret = dify_config.KNOWLEDGE_FS_JWT_SECRET
-    if base_url is None or jwt_secret is None:
-        raise KnowledgeFSConfigurationError("KnowledgeFS connection configuration is incomplete")
-    now = datetime.now(UTC)
-    token = jwt.encode(
-        {
-            "aud": _JWT_AUDIENCE,
-            "caller_kind": "interactive",
-            "dify_account_id": f"dify-account:{account_id}",
-            "exp": now + timedelta(seconds=_JWT_TTL_SECONDS),
-            "iat": now,
-            "iss": _JWT_ISSUER,
-            "scopes": [operation.required_scope],
-            "sub": f"dify-workspace:{tenant_id}",
-            "tenant_id": tenant_id,
-        },
-        jwt_secret.get_secret_value(),
-        algorithm="HS256",
-    )
-    headers = {
-        "Accept": accept or "application/json",
-        "Accept-Encoding": "identity",
-        "Authorization": f"Bearer {token}",
-    }
-    if body is not None:
-        headers["Content-Type"] = content_type or "application/json"
-    allowed_request_headers = set(operation.request_headers)
-    for name, value in (request_headers or {}).items():
-        normalized_name = name.lower()
-        if normalized_name not in allowed_request_headers:
-            raise KnowledgeFSRouteNotAllowedError("KnowledgeFS request header is not allowed")
-        headers[normalized_name] = value
-
-    try:
-        upstream_url = httpx.URL(f"{base_url}/").join(operation.path)
-        response = ssrf_proxy.make_request(
-            method=operation.method,
-            url=str(upstream_url),
-            params=query,
-            content=body,
-            headers=headers,
-            timeout=dify_config.KNOWLEDGE_FS_TIMEOUT_SECONDS,
-            follow_redirects=False,
-            max_retries=0,
-            stream_response=True,
-        )
-        response_kind = _classify_response(operation, response)
-        if response_kind == "stream":
-            content_encoding = response.headers.get("content-encoding", "identity").strip().lower()
-            if content_encoding not in {"", "identity"}:
-                response.close()
-                raise KnowledgeFSTransportError("KnowledgeFS streaming response used an unsupported encoding")
-            _set_response_read_timeout(response, dify_config.KNOWLEDGE_FS_SSE_READ_TIMEOUT_SECONDS)
-            return KnowledgeFSUpstreamResponse(response, response_kind, operation)
-
-        max_response_bytes = (
-            operation.max_response_bytes
-            if HTTPStatus.OK <= response.status_code < HTTPStatus.MULTIPLE_CHOICES
-            else _MAX_BUFFERED_RESPONSE_BYTES
-        )
-        buffered_response = ssrf_proxy.buffer_response(response, max_response_bytes=max_response_bytes)
-        if buffered_response.content and not buffered_response.headers.get("content-type", "").strip():
-            buffered_response.close()
-            raise KnowledgeFSTransportError("KnowledgeFS buffered response used an unsupported media type")
-        return KnowledgeFSUpstreamResponse(buffered_response, response_kind, operation)
-    except ssrf_proxy.ResponseLimitError as exc:
-        raise KnowledgeFSTransportError("KnowledgeFS response violated the proxy limit") from exc
-    except ToolSSRFError as exc:
-        raise KnowledgeFSConfigurationError("KnowledgeFS origin was blocked by outbound policy") from exc
-    except httpx.TimeoutException as exc:
-        raise KnowledgeFSTimeoutError("KnowledgeFS request timed out") from exc
-    except httpx.RequestError as exc:
-        raise KnowledgeFSTransportError("KnowledgeFS transport request failed") from exc
-
-
-def get_knowledge_fs_operation(method: KnowledgeFSMethod, path: str) -> KnowledgeFSOperation:
-    """Resolve an exact operation and its transport/access contract metadata."""
-    for operation in KNOWLEDGE_FS_CONSOLE_OPERATIONS:
-        if method == operation.method and _matches_route_template(operation.path, path):
-            return operation._replace(path=path)
-    raise KnowledgeFSRouteNotAllowedError("KnowledgeFS route is not allowed")
-
-
-def _classify_response(operation: KnowledgeFSOperation, response: httpx.Response) -> KnowledgeFSResponseKind:
-    """Resolve the actual response kind from status and Content-Type before reading its body."""
-    content_type = response.headers.get("content-type", "").partition(";")[0].strip().lower()
-    is_success = HTTPStatus.OK <= response.status_code < HTTPStatus.MULTIPLE_CHOICES
-    if not is_success:
-        if content_type and not _is_json_content_type(content_type):
-            response.close()
-            raise KnowledgeFSTransportError("KnowledgeFS error response used an unsupported media type")
-        return "buffered"
-
-    if operation.response_kind == "stream":
-        if content_type != "text/event-stream":
-            response.close()
-            raise KnowledgeFSTransportError("KnowledgeFS stream response used an unsupported media type")
-        return "stream"
-    if operation.response_kind == "binary":
-        if content_type not in operation.response_media_types:
-            response.close()
-            raise KnowledgeFSTransportError("KnowledgeFS binary response used an unsupported media type")
-        return "binary"
-    if content_type and not _is_json_content_type(content_type):
-        response.close()
-        raise KnowledgeFSTransportError("KnowledgeFS buffered response used an unsupported media type")
-    return "buffered"
-
-
-def _is_json_content_type(content_type: str) -> bool:
-    return content_type == "application/json" or content_type.endswith("+json")
-
-
-def _set_response_read_timeout(response: httpx.Response, timeout_seconds: float | None) -> None:
-    """Set the body-read timeout after headers identify a valid SSE response."""
-    try:
-        request = response.request
-    except RuntimeError:
-        return
-    timeout = request.extensions.get("timeout")
-    if isinstance(timeout, dict):
-        timeout["read"] = timeout_seconds
-
-
-def _matches_route_template(template: str, path: str) -> bool:
-    """Match path parameters without permitting encoded or traversal-like segments."""
-    template_segments = template.split("/")
-    path_segments = path.split("/")
-    if len(template_segments) != len(path_segments):
-        return False
-
-    for template_segment, path_segment in zip(template_segments, path_segments, strict=True):
-        if template_segment.startswith("{") and template_segment.endswith("}"):
-            if (
-                not path_segment
-                or path_segment in {".", ".."}
-                or "\\" in path_segment
-                or "%" in path_segment
-                or "?" in path_segment
-                or "#" in path_segment
-            ):
-                return False
-            continue
-        if template_segment != path_segment:
-            return False
-    return True
@@ -601,7 +601,6 @@ class RagPipelineService:
        with sessionmaker(bind=db.engine).begin() as session:
            draft_var_saver = DraftVariableSaver(
                session=session,
-                tenant_id=pipeline.tenant_id,
                app_id=pipeline.id,
                node_id=workflow_node_execution.node_id,
                node_type=workflow_node_execution.node_type,
@@ -1392,7 +1391,6 @@ class RagPipelineService:
        with sessionmaker(bind=db.engine).begin() as session:
            draft_var_saver = DraftVariableSaver(
                session=session,
-                tenant_id=pipeline.tenant_id,
                app_id=pipeline.id,
                node_id=workflow_node_execution_db_model.node_id,
                node_type=workflow_node_execution_db_model.node_type,
@@ -1,221 +0,0 @@
-"""Account-level Step-by-step Tour persistence."""
-
-from datetime import datetime
-from typing import NotRequired, TypedDict
-
-from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session, scoped_session
-
-from configs import dify_config
-from libs.datetime_utils import ensure_naive_utc
-from models.account import Account
-from models.onboarding import AccountStepByStepTourState
-
-STEP_BY_STEP_TOUR_TASK_IDS = frozenset(("home", "studio", "knowledge", "integration"))
-
-
-class StepByStepTourStateResponse(TypedDict):
-    first_workspace_id: str | None
-    skipped: bool
-    completed_task_ids: list[str]
-    manually_enabled_workspace_ids: list[str]
-    manually_disabled_workspace_ids: list[str]
-    updated_at: datetime | None
-
-
-class StepByStepTourPatch(TypedDict):
-    action: str
-    task_id: NotRequired[str | None]
-
-
-class StepByStepTourService:
-    """Coordinate persisted tour state with account eligibility rules."""
-
-    @classmethod
-    def get_state(
-        cls,
-        *,
-        account: Account,
-        current_tenant_id: str,
-        session: Session | scoped_session,
-    ) -> StepByStepTourStateResponse:
-        eligible = cls.is_eligible(account)
-        state = cls._get_state(account.id, session=session)
-
-        if eligible:
-            state = cls._ensure_state(account.id, session=session, state=state)
-            if state.first_workspace_id is None:
-                state.first_workspace_id = current_tenant_id
-                session.commit()
-                session.refresh(state)
-
-        return cls._build_response(state=state)
-
-    @classmethod
-    def patch_state(
-        cls,
-        *,
-        account: Account,
-        current_tenant_id: str,
-        patch: StepByStepTourPatch,
-        session: Session | scoped_session,
-    ) -> StepByStepTourStateResponse:
-        state = cls._ensure_state(account.id, session=session, state=None)
-        cls._apply_action(
-            state=state,
-            action=patch["action"],
-            task_id=patch.get("task_id"),
-            current_tenant_id=current_tenant_id,
-        )
-
-        session.commit()
-        session.refresh(state)
-        return cls._build_response(state=state)
-
-    @classmethod
-    def is_eligible(cls, account: Account) -> bool:
-        if not dify_config.ENABLE_STEP_BY_STEP_TOUR:
-            return False
-
-        rollout_started_at = dify_config.STEP_BY_STEP_TOUR_ROLLOUT_STARTED_AT
-        if rollout_started_at is None:
-            return False
-
-        account_started_at = account.initialized_at or account.created_at
-        if account_started_at is None:
-            return False
-
-        return ensure_naive_utc(account_started_at) >= ensure_naive_utc(rollout_started_at)
-
-    @classmethod
-    def _get_state(
-        cls,
-        account_id: str,
-        *,
-        session: Session | scoped_session,
-    ) -> AccountStepByStepTourState | None:
-        stmt = select(AccountStepByStepTourState).where(AccountStepByStepTourState.account_id == account_id).limit(1)
-        return session.execute(stmt).scalar_one_or_none()
-
-    @classmethod
-    def _ensure_state(
-        cls,
-        account_id: str,
-        *,
-        session: Session | scoped_session,
-        state: AccountStepByStepTourState | None,
-    ) -> AccountStepByStepTourState:
-        if state is None:
-            state = cls._get_state(account_id, session=session)
-        if state is not None:
-            return state
-
-        state = AccountStepByStepTourState(account_id=account_id)
-        session.add(state)
-        try:
-            session.flush()
-        except IntegrityError:
-            # Another tab/device can create the account row between our read and insert.
-            session.rollback()
-            state = cls._get_state(account_id, session=session)
-            if state is None:
-                raise
-        return state
-
-    @classmethod
-    def _apply_action(
-        cls,
-        *,
-        state: AccountStepByStepTourState,
-        action: str,
-        task_id: str | None,
-        current_tenant_id: str,
-    ) -> None:
-        match action:
-            case "skip":
-                state.skipped = True
-                state.manually_enabled_workspace_ids = cls._remove_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
-                )
-            case "complete_task":
-                if task_id is None:
-                    raise ValueError("task_id is required")
-                cls._validate_task_id(task_id)
-                state.completed_task_ids = cls._add_id(state.completed_task_ids, task_id)
-            case "uncomplete_task":
-                if task_id is None:
-                    raise ValueError("task_id is required")
-                cls._validate_task_id(task_id)
-                state.completed_task_ids = cls._remove_id(state.completed_task_ids, task_id)
-            case "enable_current_workspace":
-                state.skipped = False
-                state.manually_enabled_workspace_ids = cls._add_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
-                )
-                state.manually_disabled_workspace_ids = cls._remove_id(
-                    state.manually_disabled_workspace_ids,
-                    current_tenant_id,
-                )
-            case "disable_current_workspace":
-                state.manually_enabled_workspace_ids = cls._remove_id(
-                    state.manually_enabled_workspace_ids,
-                    current_tenant_id,
-                )
-                state.manually_disabled_workspace_ids = cls._add_id(
-                    state.manually_disabled_workspace_ids,
-                    current_tenant_id,
-                )
-            case _:
-                raise ValueError(f"Unsupported action: {action}")
-
-    @classmethod
-    def _build_response(
-        cls,
-        *,
-        state: AccountStepByStepTourState | None,
-    ) -> StepByStepTourStateResponse:
-        if state is None:
-            return {
-                "first_workspace_id": None,
-                "skipped": False,
-                "completed_task_ids": [],
-                "manually_enabled_workspace_ids": [],
-                "manually_disabled_workspace_ids": [],
-                "updated_at": None,
-            }
-
-        return {
-            "first_workspace_id": state.first_workspace_id,
-            "skipped": state.skipped,
-            "completed_task_ids": cls._normalize_ids(state.completed_task_ids),
-            "manually_enabled_workspace_ids": cls._normalize_ids(state.manually_enabled_workspace_ids),
-            "manually_disabled_workspace_ids": cls._normalize_ids(state.manually_disabled_workspace_ids),
-            "updated_at": state.updated_at,
-        }
-
-    @staticmethod
-    def _validate_task_id(task_id: str) -> None:
-        if task_id not in STEP_BY_STEP_TOUR_TASK_IDS:
-            raise ValueError(f"Unsupported task_id: {task_id}")
-
-    @classmethod
-    def _add_id(cls, values: list[str], value: str) -> list[str]:
-        normalized = cls._normalize_ids(values)
-        if value in normalized:
-            return normalized
-        return [*normalized, value]
-
-    @classmethod
-    def _remove_id(cls, values: list[str], value: str) -> list[str]:
-        return [item for item in cls._normalize_ids(values) if item != value]
-
-    @staticmethod
-    def _normalize_ids(values: list[str]) -> list[str]:
-        normalized: list[str] = []
-        for value in values:
-            if value not in normalized:
-                normalized.append(value)
-        return normalized
@@ -8,7 +8,6 @@ import uuid
 from collections.abc import Mapping
 from typing import Any, override

-from socketio.exceptions import TimeoutError as SocketIOTimeoutError  # type: ignore[reportMissingTypeStubs]
 from sqlalchemy import select
 from sqlalchemy.orm import Session

@@ -19,7 +18,6 @@ from repositories.workflow_collaboration_repository import WorkflowCollaboration
 logger = logging.getLogger(__name__)

 SERVER_HEARTBEAT_INTERVAL_SECONDS = 30
-SYNC_REQUEST_TIMEOUT_SECONDS = 15
 _PROCESS_SERVER_ID: str | None = None
 _PROCESS_SERVER_ID_PID: int | None = None

@@ -40,8 +38,7 @@ class WorkflowCollaborationService:

    Socket.IO rooms are process-local unless backed by a message queue, while online users and leader election live in
    Redis. Each websocket worker writes a small heartbeat keyed by `server_id`; session rows store their owner so a
-    worker can distinguish a live remote sid from a stale sid left behind by a dead worker. Visibility events are
-    ordered per socket, and sync requests wait for the selected saver to acknowledge its draft write.
+    worker can distinguish a live remote sid from a stale sid left behind by a dead worker.
    """

    _heartbeat_started: bool
@@ -131,9 +128,6 @@ class WorkflowCollaborationService:
            "sid": sid,
            "connected_at": int(time.time()),
            "server_id": self.server_id,
-            # Joins are assumed visible; hidden tabs re-report via a graph_view_state
-            # event right after receiving the post-join "status" emit.
-            "graph_active": True,
        }

        self._repository.set_session_info(workflow_id, session_info)
@@ -164,8 +158,7 @@ class WorkflowCollaborationService:
        self.handle_leader_disconnect(workflow_id, sid)
        self.broadcast_online_users(workflow_id)

-    def relay_collaboration_event(self, sid: str, data: Mapping[str, object]) -> tuple[dict[str, object], int]:
-        """Route collaboration control events, directing save and graph-resync requests to the active leader."""
+    def relay_collaboration_event(self, sid: str, data: Mapping[str, object]) -> tuple[dict[str, str], int]:
        mapping = self._repository.get_sid_mapping(sid)
        if not mapping:
            return {"msg": "unauthorized"}, 401
@@ -181,54 +174,11 @@ class WorkflowCollaborationService:
        if not event_type:
            return {"msg": "invalid event type"}, 400

-        if event_type == "graph_view_state":
-            if not isinstance(event_data, Mapping):
-                return {"msg": "invalid graph_view_state"}, 400
-
-            graph_active = event_data.get("graphActive")
-            sequence = event_data.get("sequence")
-            if not isinstance(graph_active, bool):
-                return {"msg": "invalid graph_view_state"}, 400
-            if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 0:
-                return {"msg": "invalid graph_view_state"}, 400
-
-            # Sequence the visibility write together with its leader side effect. Otherwise a newer
-            # visible event can land after the Lua write but before an older hidden handler demotes.
-            with self._repository.graph_view_state_lock(workflow_id):
-                applied = self._repository.update_session_graph_active(workflow_id, sid, graph_active, sequence)
-                if not applied:
-                    return {"msg": "graph_view_state_ignored"}, 200
-
-                # Write the flag before re-electing so tier-1 selection already excludes the
-                # session that just went hidden (including one _ensure_leader just promoted).
-                if not graph_active:
-                    self._demote_leader_if_hidden(workflow_id, sid)
-            return {"msg": "graph_view_state_updated"}, 200
-
        if event_type == "sync_request":
-            if not isinstance(event_data, Mapping):
-                return {"msg": "invalid sync_request"}, 400
-            request_id = event_data.get("requestId")
-            if not isinstance(request_id, str) or not request_id.strip():
-                return {"msg": "invalid sync_request"}, 400
-
            leader_sid = self._repository.get_current_leader(workflow_id)
            target_sid: str | None
            if leader_sid and self.is_session_active(workflow_id, leader_sid):
-                if self._is_session_graph_active(workflow_id, leader_sid):
-                    target_sid = leader_sid
-                else:
-                    # The leader is connected but its tab is hidden: its canvas is frozen
-                    # (rAF paused), so saving through it would persist stale data. Hand
-                    # leadership to a visible session — or, if every tab is hidden, to the
-                    # requester, whose canvas at least contains its own edits.
-                    replacement = self._select_graph_leader(workflow_id, preferred_sid=sid)
-                    if replacement and replacement != leader_sid:
-                        self._repository.set_leader(workflow_id, replacement)
-                        self.broadcast_leader_change(workflow_id, replacement)
-                        target_sid = replacement
-                    else:
-                        target_sid = leader_sid
+                target_sid = leader_sid
            else:
                if leader_sid:
                    self._repository.delete_leader(workflow_id)
@@ -238,73 +188,15 @@ class WorkflowCollaborationService:
                    self.broadcast_leader_change(workflow_id, target_sid)

            if not target_sid:
-                return {"msg": "no_active_leader", "requestId": request_id}, 503
-
-            target_data = dict(event_data)
-            target_data["requestId"] = request_id
-            try:
-                result = self._socketio.call(
-                    "collaboration_update",
-                    {"type": event_type, "userId": user_id, "data": target_data, "timestamp": timestamp},
-                    to=target_sid,
-                    timeout=SYNC_REQUEST_TIMEOUT_SECONDS,
-                )
-            except SocketIOTimeoutError:
-                logger.warning(
-                    "Workflow collaboration sync request timed out: workflow_id=%s requester_sid=%s target_sid=%s",
-                    workflow_id,
-                    sid,
-                    target_sid,
-                )
-                return {"msg": "sync_request_timeout", "requestId": request_id}, 504
-
-            if not isinstance(result, Mapping) or result.get("success") is not True:
-                response: dict[str, object] = {
-                    "msg": "workflow_sync_failed",
-                    "requestId": request_id,
-                    "success": False,
-                }
-                if isinstance(result, Mapping) and isinstance(result.get("error"), str):
-                    response["error"] = result["error"]
-                return response, 502
-
-            workflow_hash = result.get("hash")
-            updated_at = result.get("updatedAt")
-            if not isinstance(workflow_hash, str) or not workflow_hash:
-                return {"msg": "invalid_sync_response", "requestId": request_id}, 502
-            if not isinstance(updated_at, int) or isinstance(updated_at, bool):
-                return {"msg": "invalid_sync_response", "requestId": request_id}, 502
-
-            return {
-                "msg": "workflow_synced",
-                "requestId": request_id,
-                "success": True,
-                "hash": workflow_hash,
-                "updatedAt": updated_at,
-            }, 200
-
-        if event_type == "graph_resync_request":
-            leader_sid = self._repository.get_current_leader(workflow_id)
-            resync_target_sid: str | None
-            if leader_sid and self.is_session_active(workflow_id, leader_sid):
-                resync_target_sid = leader_sid
-            else:
-                if leader_sid:
-                    self._repository.delete_leader(workflow_id)
-                resync_target_sid = self._select_graph_leader(workflow_id, preferred_sid=sid)
-                if resync_target_sid:
-                    self._repository.set_leader(workflow_id, resync_target_sid)
-                    self.broadcast_leader_change(workflow_id, resync_target_sid)
-
-            if not resync_target_sid:
-                return {"msg": "no_active_leader"}, 503
+                return {"msg": "no_active_leader"}, 200

            self._socketio.emit(
                "collaboration_update",
                {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
-                to=resync_target_sid,
+                room=target_sid,
            )
-            return {"msg": "graph_resync_request_forwarded"}, 200
+
+            return {"msg": "sync_request_forwarded"}, 200

        self._socketio.emit(
            "collaboration_update",
@@ -437,56 +329,17 @@ class WorkflowCollaborationService:
        self._repository.set_leader(workflow_id, sid)
        self.broadcast_leader_change(workflow_id, sid)

-    def _select_graph_leader(
-        self,
-        workflow_id: str,
-        preferred_sid: str | None = None,
-        *,
-        require_graph_active: bool = False,
-    ) -> str | None:
-        """Pick a leader, preferring sessions whose canvas tab is visible.
-
-        Hidden tabs freeze rAF-driven CRDT->canvas application, so a visible session is
-        always the freshest saver. When every tab is hidden the room must still keep a
-        leader (followers drop sync_requests unless they believe they are the leader),
-        so tier 2 falls back to any active session unless require_graph_active is set.
-        """
-        active_sessions = [
-            session
+    def _select_graph_leader(self, workflow_id: str, preferred_sid: str | None = None) -> str | None:
+        session_sids = [
+            session["sid"]
            for session in self._repository.list_sessions(workflow_id)
-            if self.is_session_active(workflow_id, session["sid"])
+            if session.get("graph_active", True) and self.is_session_active(workflow_id, session["sid"])
        ]
-        visible_sids = [session["sid"] for session in active_sessions if session.get("graph_active", True)]
-
-        candidate_sids = visible_sids
-        if not candidate_sids and not require_graph_active:
-            candidate_sids = [session["sid"] for session in active_sessions]
-
-        if not candidate_sids:
+        if not session_sids:
            return None
-        if preferred_sid and preferred_sid in candidate_sids:
+        if preferred_sid and preferred_sid in session_sids:
            return preferred_sid
-        return candidate_sids[0]
-
-    def _is_session_graph_active(self, workflow_id: str, sid: str) -> bool:
-        """Default to True on missing/unreadable session data so read failures never churn leadership."""
-        session_info = self._repository.get_session_info(workflow_id, sid)
-        if session_info is None:
-            return True
-        return bool(session_info.get("graph_active", True))
-
-    def _demote_leader_if_hidden(self, workflow_id: str, hidden_sid: str) -> None:
-        current_leader = self._repository.get_current_leader(workflow_id)
-        if current_leader != hidden_sid:
-            return
-
-        new_leader = self._select_graph_leader(workflow_id, require_graph_active=True)
-        # No visible session: keep the hidden leader rather than leaving the room leaderless.
-        if not new_leader or new_leader == hidden_sid:
-            return
-
-        self._repository.set_leader(workflow_id, new_leader)
-        self.broadcast_leader_change(workflow_id, new_leader)
+        return session_sids[0]

    def is_session_active(self, workflow_id: str, sid: str) -> bool:
        if not sid:
@@ -823,8 +823,6 @@ _FILENAME_TRANS_TABLE = _make_filename_trans_table()


 class DraftVariableSaver:
-    """Persist draft outputs under the tenant that owns the app or pipeline."""
-
    # _DUMMY_OUTPUT_IDENTITY is a placeholder output for workflow nodes.
    # Its sole possible value is `None`.
    #
@@ -844,10 +842,6 @@ class DraftVariableSaver:
    # Database session used for persisting draft variables.
    _session: Session

-    # Resource owner tenant. An account's current tenant may be unset or point elsewhere
-    # when draft variables are persisted by an asynchronous workflow execution.
-    _tenant_id: str
-
    # The application ID associated with the draft variables.
    # This should match the `Workflow.app_id` of the workflow to which the current node belongs.
    _app_id: str
@@ -873,7 +867,6 @@ class DraftVariableSaver:
    def __init__(
        self,
        session: Session,
-        tenant_id: str,
        app_id: str,
        node_id: str,
        node_type: NodeType,
@@ -885,7 +878,6 @@ class DraftVariableSaver:
        # WorkflowNodeExecutionModel/WorkflowNodeExecution, not their `node_execution_id`
        # field. These are distinct database fields with different purposes.
        self._session = session
-        self._tenant_id = tenant_id
        self._app_id = app_id
        self._node_id = node_id
        self._node_type = node_type
@@ -893,6 +885,12 @@ class DraftVariableSaver:
        self._user = user
        self._enclosing_node_id = enclosing_node_id

+    def _resolve_app_tenant_id(self) -> str:
+        tenant_id = self._session.scalar(select(App.tenant_id).where(App.id == self._app_id))
+        if not tenant_id:
+            raise ValueError(f"Unable to resolve tenant_id for app {self._app_id}")
+        return tenant_id
+
    def _create_dummy_output_variable(self):
        return WorkflowDraftVariable.new_node_variable(
            app_id=self._app_id,
@@ -951,10 +949,11 @@ class DraftVariableSaver:
                if name == SystemVariableKey.FILES:
                    # Here we know the type of variable must be `array[file]`, we
                    # just rebuild files from the serialized payload.
+                    tenant_id = self._resolve_app_tenant_id()
                    files = [
                        build_file_from_stored_mapping(
                            file_mapping=v,
-                            tenant_id=self._tenant_id,
+                            tenant_id=tenant_id,
                        )
                        for v in value
                    ]
@@ -1097,8 +1096,8 @@ class DraftVariableSaver:
            content=original_content_serialized.encode(),
            mimetype=content_type,
            user=self._user,
-            tenant_id=self._tenant_id,
        )
+        assert self._user.current_tenant_id
        # Create WorkflowDraftVariableFile record
        variable_file = WorkflowDraftVariableFile(
            upload_file_id=upload_file.id,
@@ -1106,7 +1105,7 @@ class DraftVariableSaver:
            length=original_length,
            value_type=value_seg.value_type,
            app_id=self._app_id,
-            tenant_id=self._tenant_id,
+            tenant_id=self._user.current_tenant_id,
            user_id=self._user.id,
        )
        variable_file.id = str(uuidv7())
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
zhaohao1004	1236d2b063	fix: harden workflow archive delete resume	2026-07-20 12:06:54 +08:00
zhaohao1004	0a80c3f8e3	feat: support sharded workflow archive deletion	2026-07-20 11:34:01 +08:00
Ingram ZandGitHub	c4bc9abc2c	Merge branch 'main' into zhaohao1004/workflow-archive-catalog-discovery	2026-07-20 09:48:53 +08:00
Ingram ZandGitHub	0c30918760	Merge branch 'main' into zhaohao1004/workflow-archive-catalog-discovery	2026-07-18 10:07:22 +08:00
zhaohao1004	b3c7a706cd	fix: handle partially deleted workflow archive bundles	2026-07-16 15:05:24 +08:00
zhaohao1004	9554037fec	fix: harden workflow archive maintenance	2026-07-15 16:51:42 +08:00
zhaohao1004	a813e093d6	chore: remove archive discovery design doc	2026-07-15 16:02:31 +08:00
zhaohao1004	f7a3b9b283	refactor: use catalog for workflow archive maintenance	2026-07-15 15:52:07 +08:00