[autofix.ci] apply automated fixes

.codex/hooks.json

@@ -1,15 +0,0 @@
-{
-  "hooks": {
-    "PreToolUse": [
-      {
-        "matcher": "Bash",
-        "hooks": [
-          {
-            "type": "command",
-            "command": "npx -y block-no-verify@1.1.1"
-          }
-        ]
-      }
-    ]
-  }
-}

.dockerignore

@@ -8,7 +8,6 @@
 **/*.pyc
 **/.mypy_cache
 **/.ruff_cache
-knowledge-fs/
 .git
 .github
 *.md

.github/dependabot.yml

@@ -1,15 +1,6 @@
 version: 2
 
 updates:
-  - package-ecosystem: "npm"
-    directory: "/knowledge-fs"
-    open-pull-requests-limit: 10
-    schedule:
-      interval: "weekly"
-    groups:
-      knowledge-fs-dependencies:
-        patterns:
-          - "*"
   - package-ecosystem: "uv"
     directory: "/api"
     open-pull-requests-limit: 10

.github/workflows/deploy-dev.yml

@@ -2,70 +2,23 @@ name: Deploy Dev
 
 on:
   workflow_run:
-    workflows: ["KnowledgeFS CI"]
+    workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/konwledge"
+      - "deploy/dev"
     types:
       - completed
 
-permissions: {}
-
 jobs:
   deploy:
     runs-on: depot-ubuntu-24.04
-    permissions:
-      actions: read
-      contents: read
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'deploy/konwledge'
+      github.event.workflow_run.head_branch == 'deploy/dev'
     steps:
-      - name: Wait for API and Web image build
-        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
-        timeout-minutes: 35
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          script: |
-            const workflowId = "build-push.yml";
-            const headBranch = context.payload.workflow_run.head_branch;
-            const headSha = context.payload.workflow_run.head_sha;
-            const deadline = Date.now() + 30 * 60 * 1000;
-            const pollIntervalMs = 15 * 1000;
-
-            while (Date.now() < deadline) {
-              const { data } = await github.rest.actions.listWorkflowRuns({
-                owner: context.repo.owner,
-                repo: context.repo.repo,
-                workflow_id: workflowId,
-                branch: headBranch,
-                event: "push",
-                head_sha: headSha,
-                per_page: 10,
-              });
-              const run = data.workflow_runs[0];
-
-              if (!run) {
-                core.info(`Waiting for ${workflowId} to start for ${headSha}.`);
-              } else if (run.status !== "completed") {
-                core.info(`Waiting for ${run.html_url}; current status is ${run.status}.`);
-              } else if (run.conclusion !== "success") {
-                throw new Error(
-                  `${workflowId} did not succeed for ${headSha}: ${run.conclusion} (${run.html_url})`,
-                );
-              } else {
-                core.info(`Both image workflows succeeded for ${headSha}: ${run.html_url}`);
-                return;
-              }
-
-              await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
-            }
-
-            throw new Error(`Timed out waiting for ${workflowId} to succeed for ${headSha}.`);
-
       - name: Deploy to server
         uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
         with:
-          host: ${{ secrets.SSH_NEW_RAG_HOST }}
+          host: ${{ secrets.SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/deploy-knowledge.yml

@@ -1,7 +1,6 @@
 name: Deploy Knowledge
 
 permissions:
-  actions: read
   contents: read
 
 on:
@@ -19,48 +18,6 @@ jobs:
       github.event.workflow_run.conclusion == 'success' &&
       github.event.workflow_run.head_branch == 'deploy/konwledge'
     steps:
-      - name: Wait for KnowledgeFS CI
-        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
-        timeout-minutes: 35
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          script: |
-            const workflowId = "knowledge-fs-ci.yml";
-            const headBranch = context.payload.workflow_run.head_branch;
-            const headSha = context.payload.workflow_run.head_sha;
-            const deadline = Date.now() + 30 * 60 * 1000;
-            const pollIntervalMs = 15 * 1000;
-
-            while (Date.now() < deadline) {
-              const { data } = await github.rest.actions.listWorkflowRuns({
-                owner: context.repo.owner,
-                repo: context.repo.repo,
-                workflow_id: workflowId,
-                branch: headBranch,
-                event: "push",
-                head_sha: headSha,
-                per_page: 10,
-              });
-              const run = data.workflow_runs[0];
-
-              if (!run) {
-                core.info(`Waiting for ${workflowId} to start for ${headSha}.`);
-              } else if (run.status !== "completed") {
-                core.info(`Waiting for ${run.html_url}; current status is ${run.status}.`);
-              } else if (run.conclusion !== "success") {
-                throw new Error(
-                  `${workflowId} did not succeed for ${headSha}: ${run.conclusion} (${run.html_url})`,
-                );
-              } else {
-                core.info(`KnowledgeFS CI succeeded for ${headSha}: ${run.html_url}`);
-                return;
-              }
-
-              await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
-            }
-
-            throw new Error(`Timed out waiting for ${workflowId} to succeed for ${headSha}.`);
-
       - name: Deploy to server
         uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
         with:

.github/workflows/knowledge-fs-ci.yml

@@ -1,514 +0,0 @@
-name: KnowledgeFS CI
-
-on:
-  pull_request:
-    branches: ["main"]
-  merge_group:
-    branches: ["main"]
-    types: [checks_requested]
-  push:
-    branches: ["main", "deploy/konwledge"]
-  workflow_dispatch:
-
-permissions:
-  contents: read
-  pull-requests: read
-
-concurrency:
-  group: knowledge-fs-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
-env:
-  CI: true
-  DIFY_KNOWLEDGE_FS_API_IMAGE_NAME: >-
-    ${{ vars.DIFY_KNOWLEDGE_FS_API_IMAGE_NAME || 'langgenius/dify-knowledge-fs-api' }}
-
-jobs:
-  check-changes:
-    name: Check KnowledgeFS changes
-    runs-on: depot-ubuntu-24.04-4
-    outputs:
-      knowledge-fs: ${{ steps.changes.outputs.knowledge-fs }}
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          fetch-depth: 0
-          persist-credentials: false
-
-      - name: Detect KnowledgeFS changes
-        id: changes
-        uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
-        with:
-          filters: |
-            knowledge-fs:
-              - 'knowledge-fs/**'
-              - 'knowledge-fs/packages/api/src/dify-capability-v2.ts'
-              - 'knowledge-fs/packages/api/src/knowledge-space-routes.ts'
-              - 'knowledge-fs/packages/api/src/upload-session-routes.ts'
-              - 'knowledge-fs/scripts/export-capability-v2-operations.mjs'
-              - 'knowledge-fs/scripts/export-openapi.mjs'
-              - 'api/dev/generate_knowledge_fs_contract.py'
-              - 'api/dev/knowledge_fs_product_contract.py'
-              - 'api/knowledge-fs-contract.lock.json'
-              - 'api/knowledge-fs-product-operation-gaps.json'
-              - 'api/knowledge-fs-product-operations.json'
-              - 'api/**/knowledge_fs/**'
-              - 'api/**/*knowledge_fs*'
-              - 'api/**/*knowledge-fs*'
-              - 'api/.env.example'
-              - 'api/app_factory.py'
-              - 'api/commands/__init__.py'
-              - 'api/controllers/console/__init__.py'
-              - 'api/controllers/console/workspace/rbac.py'
-              - 'api/controllers/service_api/__init__.py'
-              - 'api/core/agent/base_agent_runner.py'
-              - 'api/core/app/apps/agent_app/runtime_request_builder.py'
-              - 'api/core/rbac/entities.py'
-              - 'api/core/tools/__base/tool_runtime.py'
-              - 'api/core/tools/builtin_tool/_position.yaml'
-              - 'api/core/workflow/node_runtime.py'
-              - 'api/core/workflow/nodes/agent_v2/runtime_request_builder.py'
-              - 'api/extensions/ext_celery.py'
-              - 'api/extensions/ext_commands.py'
-              - 'api/models/__init__.py'
-              - 'api/services/account_service.py'
-              - 'api/services/agent_tool_inner_service.py'
-              - 'api/services/enterprise/rbac_service.py'
-              - 'api/services/entities/agent_tool_inner.py'
-              - 'api/services/knowledge_fs/**'
-              - 'api/services/knowledge_fs_capability.py'
-              - 'api/tests/unit_tests/dev/test_generate_knowledge_fs_contract.py'
-              - 'api/tests/unit_tests/controllers/console/workspace/test_rbac.py'
-              - 'api/tests/unit_tests/core/agent/test_base_agent_runner.py'
-              - 'api/tests/unit_tests/core/app/apps/agent_app/test_runtime_request_builder.py'
-              - 'api/tests/unit_tests/core/workflow/nodes/agent_v2/test_runtime_request_builder.py'
-              - 'api/tests/unit_tests/core/workflow/nodes/tool/test_tool_node_runtime.py'
-              - 'api/tests/unit_tests/core/workflow/test_node_runtime.py'
-              - 'api/tests/unit_tests/services/enterprise/test_rbac_service.py'
-              - 'api/tests/unit_tests/services/test_account_service.py'
-              - 'api/tests/unit_tests/services/test_agent_tool_inner_service.py'
-              - 'api/tests/unit_tests/services/test_knowledge_fs_capability.py'
-              - 'api/tests/unit_tests/services/test_knowledge_fs_product_operations.py'
-              - 'api/pyproject.toml'
-              - 'api/uv.lock'
-              - 'dify-agent/src/dify_agent/layers/dify_core_tools/client.py'
-              - 'dify-agent/tests/local/dify_agent/layers/dify_core_tools/test_client.py'
-              - 'packages/contracts/generated/api/console/**'
-              - 'packages/contracts/generated/api/service/**'
-              - 'docker/.env.example'
-              - 'docker/README.md'
-              - 'docker/dify-env-sync.py'
-              - 'docker/dify-env-sync.sh'
-              - 'docker/docker-compose-template.yaml'
-              - 'docker/docker-compose.yaml'
-              - 'docker/envs/core-services/api.env.example'
-              - 'docker/envs/core-services/knowledge-fs.env.example'
-              - 'docker/generate_docker_compose'
-              - 'docs/design/knowledge-fs*'
-              - '.github/dependabot.yml'
-              - '.github/workflows/knowledge-fs-ci.yml'
-
-  build:
-    name: Build KnowledgeFS API production image
-    needs:
-      - check-changes
-      - quality
-    if: >-
-      (needs.check-changes.outputs.knowledge-fs == 'true' || github.event_name == 'workflow_dispatch') &&
-      needs.quality.result == 'success'
-    runs-on: depot-ubuntu-24.04-4
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          persist-credentials: false
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
-
-      - name: Login to Docker Hub
-        if: >-
-          github.event_name == 'workflow_dispatch' ||
-          (github.event_name == 'push' &&
-          (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/deploy/konwledge'))
-        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
-        with:
-          username: ${{ secrets.DOCKERHUB_USER }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-
-      - name: Extract KnowledgeFS image metadata
-        id: meta
-        uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
-        with:
-          images: ${{ env.DIFY_KNOWLEDGE_FS_API_IMAGE_NAME }}
-          tags: |
-            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
-            type=ref,event=branch
-            type=sha,format=long
-
-      - name: Build KnowledgeFS API image
-        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
-        with:
-          context: ./knowledge-fs
-          file: ./knowledge-fs/apps/api/Dockerfile
-          labels: ${{ steps.meta.outputs.labels }}
-          platforms: linux/amd64
-          push: >-
-            ${{ github.event_name == 'workflow_dispatch' ||
-            (github.event_name == 'push' &&
-            (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/deploy/konwledge')) }}
-          tags: ${{ steps.meta.outputs.tags }}
-
-  quality:
-    name: Run KnowledgeFS quality and contract gates
-    needs: check-changes
-    if: needs.check-changes.outputs.knowledge-fs == 'true' || github.event_name == 'workflow_dispatch'
-    runs-on: depot-ubuntu-24.04-4
-    defaults:
-      run:
-        shell: bash
-        working-directory: ./knowledge-fs
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          fetch-depth: 0
-          persist-credentials: false
-
-      - name: Setup pnpm
-        uses: pnpm/action-setup@0ebf47130e4866e96fce0953f49152a61190b271 # v6.0.9
-        with:
-          package_json_file: knowledge-fs/package.json
-          run_install: false
-
-      - name: Setup Node
-        uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
-        with:
-          node-version: 22
-          cache: pnpm
-          cache-dependency-path: knowledge-fs/pnpm-lock.yaml
-
-      - name: Install KnowledgeFS dependencies
-        run: pnpm install --frozen-lockfile
-
-      - name: Scan KnowledgeFS secrets
-        run: pnpm security:secrets
-
-      - name: Audit KnowledgeFS production dependencies
-        run: pnpm security:dependencies
-
-      - name: Run KnowledgeFS checks
-        run: pnpm check
-
-      - name: Build KnowledgeFS
-        run: pnpm build
-
-      - name: Lint KnowledgeFS
-        run: pnpm lint:backend
-
-      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
-        with:
-          enable-cache: true
-          python-version: "3.12"
-          cache-dependency-glob: |
-            api/uv.lock
-            dify-agent/uv.lock
-
-      - name: Verify Dify dependency lock
-        working-directory: .
-        run: uv lock --project api --check
-
-      - name: Install Dify contract dependencies
-        working-directory: .
-        run: uv sync --project api --locked --dev
-
-      - name: Collect Dify KnowledgeFS gate targets
-        working-directory: .
-        run: |
-          set -euo pipefail
-
-          target_dir="${RUNNER_TEMP:?}/knowledge-fs-ci-targets"
-          mkdir -p "$target_dir"
-
-          production_targets=()
-          add_production_target() {
-            local path="$1"
-            if [[ ! -f "$path" ]]; then
-              echo "required Dify KnowledgeFS production target is missing: $path" >&2
-              exit 1
-            fi
-            production_targets+=("$path")
-          }
-
-          while IFS= read -r -d '' path; do
-            if [[ "$path" == *knowledge_fs* ]]; then
-              add_production_target "$path"
-            fi
-          done < <(
-            find api \
-              \( -path 'api/.venv' -o -path 'api/tests' -o -path 'api/storage' \) -prune \
-              -o -type f -name '*.py' -print0
-          )
-
-          production_touchpoints=(
-            api/app_factory.py
-            api/commands/__init__.py
-            api/controllers/console/__init__.py
-            api/controllers/console/workspace/rbac.py
-            api/controllers/service_api/__init__.py
-            api/core/agent/base_agent_runner.py
-            api/core/app/apps/agent_app/runtime_request_builder.py
-            api/core/rbac/entities.py
-            api/core/tools/__base/tool_runtime.py
-            api/core/workflow/node_runtime.py
-            api/core/workflow/nodes/agent_v2/runtime_request_builder.py
-            api/extensions/ext_celery.py
-            api/extensions/ext_commands.py
-            api/models/__init__.py
-            api/services/account_service.py
-            api/services/agent_tool_inner_service.py
-            api/services/enterprise/rbac_service.py
-            api/services/entities/agent_tool_inner.py
-          )
-          for path in "${production_touchpoints[@]}"; do
-            add_production_target "$path"
-          done
-          printf '%s\0' "${production_touchpoints[@]}" > "$target_dir/glue-files"
-
-          if ((${#production_targets[@]} == 0)); then
-            echo "Dify KnowledgeFS production target set is empty" >&2
-            exit 1
-          fi
-          printf '%s\0' "${production_targets[@]}" > "$target_dir/production-files"
-
-          test_targets=()
-          add_test_target() {
-            local path="$1"
-            if [[ ! -f "$path" ]]; then
-              echo "required Dify KnowledgeFS unit test is missing: $path" >&2
-              exit 1
-            fi
-            test_targets+=("$path")
-          }
-
-          while IFS= read -r -d '' path; do
-            if [[ "$path" == *knowledge_fs* ]]; then
-              add_test_target "$path"
-            fi
-          done < <(find api/tests/unit_tests -type f -name '*.py' -print0)
-
-          test_touchpoints=(
-            api/tests/unit_tests/controllers/console/workspace/test_rbac.py
-            api/tests/unit_tests/core/agent/test_base_agent_runner.py
-            api/tests/unit_tests/core/app/apps/agent_app/test_runtime_request_builder.py
-            api/tests/unit_tests/core/workflow/nodes/agent_v2/test_runtime_request_builder.py
-            api/tests/unit_tests/core/workflow/nodes/tool/test_tool_node_runtime.py
-            api/tests/unit_tests/core/workflow/test_node_runtime.py
-            api/tests/unit_tests/services/enterprise/test_rbac_service.py
-            api/tests/unit_tests/services/test_account_service.py
-            api/tests/unit_tests/services/test_agent_tool_inner_service.py
-          )
-          for path in "${test_touchpoints[@]}"; do
-            add_test_target "$path"
-          done
-
-          required_test_scopes=(
-            /commands/
-            /configs/
-            /controllers/
-            /core/agent/
-            /core/app/apps/agent_app/
-            /core/tools/builtin_tool/providers/knowledge_fs/
-            /core/workflow/
-            /dev/
-            /extensions/
-            /migrations/
-            /models/
-            /repositories/
-            /services/
-            /tasks/
-          )
-          for required_scope in "${required_test_scopes[@]}"; do
-            scope_found=false
-            for path in "${test_targets[@]}"; do
-              if [[ "$path" == *"$required_scope"* ]]; then
-                scope_found=true
-                break
-              fi
-            done
-            if [[ "$scope_found" != true ]]; then
-              echo "required Dify KnowledgeFS test scope is empty: $required_scope" >&2
-              exit 1
-            fi
-          done
-
-          if ((${#test_targets[@]} == 0)); then
-            echo "Dify KnowledgeFS unit test target set is empty" >&2
-            exit 1
-          fi
-          printf '%s\0' "${test_targets[@]}" > "$target_dir/unit-test-files"
-
-      - name: Lint Dify KnowledgeFS integration
-        working-directory: .
-        run: |
-          set -euo pipefail
-          targets=()
-          while IFS= read -r -d '' path; do
-            targets+=("$path")
-          done < "${RUNNER_TEMP:?}/knowledge-fs-ci-targets/production-files"
-          if ((${#targets[@]} == 0)); then
-            echo "Dify KnowledgeFS production target manifest is empty" >&2
-            exit 1
-          fi
-          uv run --project api --dev ruff format --check "${targets[@]}"
-          uv run --project api --dev ruff check "${targets[@]}"
-
-      - name: Type-check Dify KnowledgeFS integration
-        working-directory: .
-        run: |
-          set -euo pipefail
-          targets=()
-          while IFS= read -r -d '' path; do
-            targets+=("$path")
-          done < "${RUNNER_TEMP:?}/knowledge-fs-ci-targets/production-files"
-          if ((${#targets[@]} == 0)); then
-            echo "Dify KnowledgeFS production target manifest is empty" >&2
-            exit 1
-          fi
-          PYREFLY_OUTPUT_FORMAT=github ./dev/pyrefly-check-local "${targets[@]}"
-
-          mypy_targets=()
-          for path in "${targets[@]}"; do
-            if [[ "$path" != api/migrations/* ]]; then
-              mypy_targets+=("${path#api/}")
-            fi
-          done
-          if ((${#mypy_targets[@]} == 0)); then
-            echo "Dify KnowledgeFS Mypy target set is empty" >&2
-            exit 1
-          fi
-          uv run --directory api --dev mypy \
-            --explicit-package-bases \
-            --exclude-gitignore \
-            --exclude '(^|/)conftest\.py$' \
-            --exclude 'tests/' \
-            --exclude 'migrations/' \
-            --check-untyped-defs \
-            --disable-error-code=import-untyped \
-            "${mypy_targets[@]}"
-
-      - name: Test Dify KnowledgeFS unit surface
-        working-directory: .
-        env:
-          COVERAGE_FILE: ${{ runner.temp }}/dify-knowledge-fs.coverage
-        run: |
-          set -euo pipefail
-          targets=()
-          while IFS= read -r -d '' path; do
-            targets+=("$path")
-          done < "${RUNNER_TEMP:?}/knowledge-fs-ci-targets/unit-test-files"
-          if ((${#targets[@]} == 0)); then
-            echo "Dify KnowledgeFS unit test target manifest is empty" >&2
-            exit 1
-          fi
-          uv run --project api --dev coverage run --branch --source=api -m pytest "${targets[@]}" --no-cov -q
-
-      - name: Enforce Dify KnowledgeFS focused coverage
-        working-directory: .
-        env:
-          COVERAGE_FILE: ${{ runner.temp }}/dify-knowledge-fs.coverage
-          KNOWLEDGE_FS_COVERAGE_BASE: >-
-            ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.event.before || '' }}
-        run: |
-          set -euo pipefail
-          report="${RUNNER_TEMP:?}/dify-knowledge-fs-coverage.json"
-          uv run --project api --dev coverage json --show-contexts -o "$report"
-          uv run --project api --dev python api/dev/check_knowledge_fs_coverage.py \
-            --coverage-json "$report" \
-            --glue-manifest "${RUNNER_TEMP:?}/knowledge-fs-ci-targets/glue-files" \
-            --base "$KNOWLEDGE_FS_COVERAGE_BASE" \
-            --minimum 90 \
-            --glue-minimum 90
-
-      - name: Verify Dify KnowledgeFS contract
-        working-directory: .
-        run: uv run --project api python api/dev/generate_knowledge_fs_contract.py --check
-
-      - name: Verify Dify Agent dependency lock
-        working-directory: .
-        run: uv lock --project dify-agent --check
-
-      - name: Install Dify Agent gate dependencies
-        working-directory: .
-        run: uv sync --project dify-agent --locked --dev
-
-      - name: Lint Dify Agent KnowledgeFS integration
-        working-directory: ./dify-agent
-        run: |
-          uv run --project . --dev ruff format --check \
-            src/dify_agent/layers/dify_core_tools/client.py \
-            tests/local/dify_agent/layers/dify_core_tools/test_client.py
-          uv run --project . --dev ruff check \
-            src/dify_agent/layers/dify_core_tools/client.py \
-            tests/local/dify_agent/layers/dify_core_tools/test_client.py
-
-      - name: Type-check Dify Agent KnowledgeFS integration
-        working-directory: ./dify-agent
-        run: >-
-          uv run --project . --dev basedpyright --level error
-          src/dify_agent/layers/dify_core_tools/client.py
-          tests/local/dify_agent/layers/dify_core_tools/test_client.py
-
-      - name: Test Dify Agent KnowledgeFS integration
-        working-directory: ./dify-agent
-        run: >-
-          uv run --project . --dev python -m pytest
-          tests/local/dify_agent/layers/dify_core_tools/test_client.py
-          -q
-
-  skip:
-    name: Skip KnowledgeFS quality and contract gates
-    needs: check-changes
-    if: needs.check-changes.outputs.knowledge-fs != 'true' && github.event_name != 'workflow_dispatch'
-    runs-on: depot-ubuntu-24.04-4
-    steps:
-      - name: Report skipped KnowledgeFS checks
-        run: echo "No KnowledgeFS-related changes detected; skipping KnowledgeFS checks."
-
-  final:
-    name: KnowledgeFS CI
-    if: ${{ always() }}
-    needs:
-      - check-changes
-      - build
-      - quality
-      - skip
-    runs-on: depot-ubuntu-24.04-4
-    steps:
-      - name: Finalize KnowledgeFS CI status
-        env:
-          EVENT_NAME: ${{ github.event_name }}
-          BUILD_RESULT: ${{ needs.build.result }}
-          KNOWLEDGE_FS_CHANGED: ${{ needs.check-changes.outputs.knowledge-fs }}
-          QUALITY_RESULT: ${{ needs.quality.result }}
-          SKIP_RESULT: ${{ needs.skip.result }}
-        run: |
-          if [[ "$EVENT_NAME" == 'workflow_dispatch' || "$KNOWLEDGE_FS_CHANGED" == 'true' ]]; then
-            if [[ "$BUILD_RESULT" == 'success' && "$QUALITY_RESULT" == 'success' ]]; then
-              echo "KnowledgeFS build and checks ran successfully."
-              exit 0
-            fi
-
-            echo "KnowledgeFS build or checks failed: build=$BUILD_RESULT quality=$QUALITY_RESULT" >&2
-            exit 1
-          fi
-
-          if [[ "$SKIP_RESULT" == 'success' ]]; then
-            echo "KnowledgeFS checks were skipped because no related files changed."
-            exit 0
-          fi
-
-          echo "KnowledgeFS change detection or skip reporting failed with result: $SKIP_RESULT" >&2
-          exit 1

.gitignore

@@ -30,11 +30,6 @@ share/python-wheels/
 *.egg
 MANIFEST
 
-# KnowledgeFS is an independently rooted TypeScript workspace. Its admin `lib`
-# directory contains source files rather than Python build output.
-!/knowledge-fs/apps/admin/lib/
-!/knowledge-fs/apps/admin/lib/**
-
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

api/.env.example

@@ -686,25 +686,11 @@ AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
 
 # KnowledgeFS (Dataset 2.0)
 KNOWLEDGE_FS_ENABLED=false
-# Production deployments require HTTPS; plain HTTP is limited to non-production or loopback.
 KNOWLEDGE_FS_BASE_URL=
-KNOWLEDGE_FS_DIRECT_ORIGIN=
-KNOWLEDGE_FS_LIFECYCLE_WORKER_ENABLED=false
-KNOWLEDGE_FS_INTEGRATED_PROVISION_READY=false
-KNOWLEDGE_FS_LEGACY_ACL_FREEZE_READY=false
-KNOWLEDGE_FS_LIFECYCLE_POLL_INTERVAL_SECONDS=15
-KNOWLEDGE_FS_LIFECYCLE_LEASE_SECONDS=60
-KNOWLEDGE_FS_LIFECYCLE_BATCH_SIZE=25
-KNOWLEDGE_FS_CAPABILITY_V2_ENABLED=false
-KNOWLEDGE_FS_CAPABILITY_V2_SIGNING_KID=
-KNOWLEDGE_FS_CAPABILITY_V2_PRIVATE_KEY_PEM=
-KNOWLEDGE_FS_CAPABILITY_V2_PREVIOUS_PUBLIC_JWKS=
-KNOWLEDGE_FS_CAPABILITY_V2_ISSUER=dify-control-plane
-KNOWLEDGE_FS_CAPABILITY_V2_AUDIENCE=knowledge-fs
-KNOWLEDGE_FS_CAPABILITY_V2_MAX_TTL_SECONDS=60
+# 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
-KNOWLEDGE_FS_JWKS_CACHE_MAX_AGE_SECONDS=300
-KNOWLEDGE_FS_PRODUCT_MAX_RESPONSE_BYTES=4194304
 
 # Marketplace configuration
 MARKETPLACE_ENABLED=true

api/README.md

@@ -50,9 +50,7 @@ The scripts resolve paths relative to their location, so you can run them from a
    ./dev/start-worker
    ```
 
-1. Start Celery Beat when scheduled tasks are needed. This is required when
-   `KNOWLEDGE_FS_LIFECYCLE_WORKER_ENABLED=true` so provisioning and deletion
-   outbox commands are dispatched to the lifecycle worker.
+1. Optional: start Celery Beat (scheduled tasks).
 
    ```bash
    ./dev/start-beat

api/app_factory.py

@@ -183,7 +183,6 @@ def initialize_extensions(app: DifyApp):
         ext_forward_refs,
         ext_hosting_provider,
         ext_import_modules,
-        ext_knowledge_fs_observability,
         ext_logging,
         ext_login,
         ext_logstore,
@@ -234,7 +233,6 @@ def initialize_extensions(app: DifyApp):
         ext_enterprise_telemetry,
         ext_request_logging,
         ext_session_factory,
-        ext_knowledge_fs_observability,
         ext_oauth_bearer,
     ]
     for ext in extensions:

api/commands/__init__.py

@@ -3,6 +3,7 @@ CLI command modules extracted from `commands.py`.
 """
 
 from .account import create_tenant, reset_email, reset_password
+from .app_maintenance import convert_to_agent_apps, fix_app_site_missing
 from .data_migrate import data_migrate, legacy_model_types
 from .data_migration import (
     export_migration_data,
@@ -10,7 +11,7 @@ from .data_migration import (
     import_migration_data,
     migration_data_wizard,
 )
-from .knowledge_fs import knowledge_fs_control_space
+from .database import upgrade_db
 from .plugin import (
     backfill_plugin_auto_upgrade,
     extract_plugins,
@@ -37,12 +38,6 @@ from .retention import (
     restore_workflow_runs,
 )
 from .storage import clear_orphaned_file_records, file_usage, migrate_oss, remove_orphaned_files_on_storage
-from .system import (
-    convert_to_agent_apps,
-    fix_app_site_missing,
-    reset_encrypt_key_pair,
-    upgrade_db,
-)
 from .vector import (
     add_qdrant_index,
     migrate_annotation_vector_database,
@@ -50,6 +45,8 @@ from .vector import (
     old_metadata_migration,
     vdb_migrate,
 )
+from .workflow_migration import migrate_legacy_sys_files_workflows
+from .workspace import reset_encrypt_key_pair
 
 __all__ = [
     "add_qdrant_index",
@@ -76,12 +73,12 @@ __all__ = [
     "import_migration_data",
     "install_plugins",
     "install_rag_pipeline_plugins",
-    "knowledge_fs_control_space",
     "legacy_model_types",
     "migrate_annotation_vector_database",
     "migrate_data_for_plugin",
     "migrate_dataset_permissions_to_rbac",
     "migrate_knowledge_vector_database",
+    "migrate_legacy_sys_files_workflows",
     "migrate_member_roles_to_rbac",
     "migrate_oss",
     "migration_data_wizard",

api/commands/app_maintenance.py

@@ -1,85 +1,27 @@
+"""App data maintenance CLI commands."""
+
 import logging
 
 import click
 import sqlalchemy as sa
-from sqlalchemy import delete, select, update
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy import select, update
 
-from configs import dify_config
 from events.app_event import app_was_created
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from libs.db_migration_lock import DbMigrationAutoRenewLock
-from libs.rsa import generate_key_pair
 from models import Tenant
 from models.model import App, AppMode, Conversation
-from models.provider import Provider, ProviderModel
-from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider
 
 logger = logging.getLogger(__name__)
 
-DB_UPGRADE_LOCK_TTL_SECONDS = 60
-
-
-@click.command(
-    "reset-encrypt-key-pair",
-    help="Reset the asymmetric key pair of workspace for encrypt LLM credentials. "
-    "After the reset, all LLM credentials and tool provider credentials "
-    "(builtin / API / MCP) will be purged, requiring re-entry. "
-    "Only support SELF_HOSTED mode.",
-)
-@click.confirmation_option(
-    prompt=click.style(
-        "Are you sure you want to reset encrypt key pair? "
-        "This will also purge builtin / API / MCP tool provider records for every tenant. "
-        "This operation cannot be rolled back!",
-        fg="red",
-    )
-)
-def reset_encrypt_key_pair():
-    """
-    Reset the encrypted key pair of workspace for encrypt LLM credentials.
-    After the reset, all LLM credentials will become invalid, requiring re-entry.
-    Only support SELF_HOSTED mode.
-    """
-    if dify_config.EDITION != "SELF_HOSTED":
-        click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
-        return
-    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-        tenants = session.scalars(select(Tenant)).all()
-        for tenant in tenants:
-            if not tenant:
-                click.echo(click.style("No workspaces found. Run /install first.", fg="red"))
-                return
-
-            tenant.encrypt_public_key = generate_key_pair(tenant.id)
-
-            session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
-            session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
-
-            # Purge tool provider records that hold credentials encrypted under the
-            # tenant key. Leaving them in place causes /console/api/workspaces/current/
-            # tool-providers to 500 because decryption fails on stale ciphertext (#35396).
-            session.execute(delete(BuiltinToolProvider).where(BuiltinToolProvider.tenant_id == tenant.id))
-            session.execute(delete(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant.id))
-            session.execute(delete(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant.id))
-
-            click.echo(
-                click.style(
-                    f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",
-                    fg="green",
-                )
-            )
-
 
 @click.command("convert-to-agent-apps", help="Convert Agent Assistant to Agent App.")
-def convert_to_agent_apps():
+def convert_to_agent_apps() -> None:
     """
     Convert Agent Assistant to Agent App.
     """
     click.echo(click.style("Starting convert to agent apps.", fg="green"))
 
-    proceeded_app_ids = []
+    proceeded_app_ids: list[str] = []
 
     while True:
         # fetch first 1000 apps
@@ -132,48 +74,14 @@ def convert_to_agent_apps():
     click.echo(click.style(f"Conversion complete. Converted {len(proceeded_app_ids)} agent apps.", fg="green"))
 
 
-@click.command("upgrade-db", help="Upgrade the database")
-def upgrade_db():
-    click.echo("Preparing database migration...")
-    lock = DbMigrationAutoRenewLock(
-        redis_client=redis_client,
-        name="db_upgrade_lock",
-        ttl_seconds=DB_UPGRADE_LOCK_TTL_SECONDS,
-        logger=logger,
-        log_context="db_migration",
-    )
-    if lock.acquire(blocking=False):
-        migration_succeeded = False
-        try:
-            click.echo(click.style("Starting database migration.", fg="green"))
-
-            # run db migration
-            import flask_migrate
-
-            flask_migrate.upgrade()
-
-            migration_succeeded = True
-            click.echo(click.style("Database migration successful!", fg="green"))
-
-        except Exception as e:
-            logger.exception("Failed to execute database migration")
-            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
-            raise SystemExit(1)
-        finally:
-            status = "successful" if migration_succeeded else "failed"
-            lock.release_safely(status=status)
-    else:
-        click.echo("Database migration skipped")
-
-
 @click.command("fix-app-site-missing", help="Fix app related site missing issue.")
-def fix_app_site_missing():
+def fix_app_site_missing() -> None:
     """
     Fix app related site missing issue.
     """
     click.echo(click.style("Starting fix for missing app-related sites.", fg="green"))
 
-    failed_app_ids = []
+    failed_app_ids: list[str] = []
     while True:
         sql = """select apps.id as id from apps left join sites on sites.app_id=apps.id
 where sites.id is null limit 1000"""

api/commands/database.py

@@ -0,0 +1,45 @@
+"""Database schema migration CLI commands."""
+
+import logging
+
+import click
+
+from extensions.ext_redis import redis_client
+from libs.db_migration_lock import DbMigrationAutoRenewLock
+
+logger = logging.getLogger(__name__)
+
+DB_UPGRADE_LOCK_TTL_SECONDS = 60
+
+
+@click.command("upgrade-db", help="Upgrade the database")
+def upgrade_db() -> None:
+    click.echo("Preparing database migration...")
+    lock = DbMigrationAutoRenewLock(
+        redis_client=redis_client,
+        name="db_upgrade_lock",
+        ttl_seconds=DB_UPGRADE_LOCK_TTL_SECONDS,
+        logger=logger,
+        log_context="db_migration",
+    )
+    if lock.acquire(blocking=False):
+        migration_succeeded = False
+        try:
+            click.echo(click.style("Starting database migration.", fg="green"))
+
+            import flask_migrate
+
+            flask_migrate.upgrade()
+
+            migration_succeeded = True
+            click.echo(click.style("Database migration successful!", fg="green"))
+
+        except Exception as e:
+            logger.exception("Failed to execute database migration")
+            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
+            raise SystemExit(1)
+        finally:
+            status = "successful" if migration_succeeded else "failed"
+            lock.release_safely(status=status)
+    else:
+        click.echo("Database migration skipped")

api/commands/knowledge_fs.py

@@ -1,621 +0,0 @@
-"""Operator commands for the independent KnowledgeFS control-plane."""
-
-from __future__ import annotations
-
-import json
-from collections.abc import Callable
-from datetime import datetime
-from functools import partial
-from pathlib import Path
-
-import click
-from pydantic import BaseModel, ValidationError
-
-from core.db.session_factory import session_factory
-from services.knowledge_fs.cleanup import (
-    CleanupApprovalInput,
-    CleanupCompletionEvidenceInput,
-    CleanupReadinessEvidenceInput,
-    CleanupStartInput,
-    KnowledgeFSCleanupError,
-    KnowledgeFSCleanupService,
-)
-from services.knowledge_fs.control_space_commands import KnowledgeFSControlSpaceCommandService
-from services.knowledge_fs.control_space_lifecycle import KnowledgeFSControlSpaceLifecycleError
-from services.knowledge_fs.control_space_management import (
-    KnowledgeFSControlSpaceManagementService,
-    KnowledgeFSControlSpaceRegistration,
-)
-from services.knowledge_fs.cutover import (
-    CutoverSmokeResultsInput,
-    FinalDeltaInput,
-    KnowledgeFSCutoverError,
-    KnowledgeFSWorkspaceCutoverService,
-    LegacyDependencyInput,
-    QuarantineResolutionInput,
-    ShadowAuthorizationObservationInput,
-    ShadowCompletionInput,
-    WorkspaceInventoryInput,
-)
-from services.knowledge_fs.greenfield_initializer import KnowledgeFSWorkspaceGreenfieldInitializer
-from services.knowledge_fs.orphan_reconciler import KnowledgeFSOrphanReconciler
-from services.knowledge_fs.remote_registry import get_knowledge_fs_lifecycle_remote
-
-
-@click.group("knowledge-fs-control-space")
-def knowledge_fs_control_space() -> None:
-    """Inspect and repair Dify-owned KnowledgeFS control-space state."""
-
-
-@knowledge_fs_control_space.command("dry-run")
-@click.option("--tenant-id", default=None)
-def dry_run(tenant_id: str | None) -> None:
-    report = _management_service().dry_run(tenant_id=tenant_id)
-    click.echo(json.dumps(report._asdict(), sort_keys=True))
-
-
-@knowledge_fs_control_space.command("inventory")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Create ledgers; omitted means read-only inventory.")
-def inventory(input_path: Path, apply: bool) -> None:
-    """Validate strict Workspace inventory JSONL and optionally create ledgers."""
-
-    service = _cutover_service()
-    for payload in _read_jsonl(input_path, WorkspaceInventoryInput):
-        report = _operator_call(partial(service.inventory, payload, apply=apply))
-        _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("register")
-@click.option("--tenant-id", required=True)
-@click.option("--owner-account-id", required=True)
-@click.option("--provisioning-key", required=True)
-@click.option("--knowledge-space-id", required=True)
-@click.option("--knowledge-space-revision", type=click.IntRange(min=0), required=True)
-def register(
-    tenant_id: str,
-    owner_account_id: str,
-    provisioning_key: str,
-    knowledge_space_id: str,
-    knowledge_space_revision: int,
-) -> None:
-    control_space, replayed = _management_service().register(
-        KnowledgeFSControlSpaceRegistration(
-            tenant_id,
-            owner_account_id,
-            provisioning_key,
-            knowledge_space_id,
-            knowledge_space_revision,
-        )
-    )
-    click.echo(json.dumps({"control_space_id": control_space.id, "replayed": replayed}, sort_keys=True))
-
-
-@knowledge_fs_control_space.command("backfill")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist registrations; omitted means dry-run.")
-def backfill(input_path: Path, apply: bool) -> None:
-    """Backfill strict Workspace inventory JSONL; dry-run unless --apply is explicit."""
-
-    service = _cutover_service()
-    for payload in _read_jsonl(input_path, WorkspaceInventoryInput):
-        report = _operator_call(partial(service.backfill, payload, apply=apply))
-        _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("quarantine-resolve")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist resolutions; omitted means dry-run.")
-def quarantine_resolve(input_path: Path, apply: bool) -> None:
-    """Resolve strict tenant-scoped quarantine JSONL with immutable operator evidence."""
-
-    service = _cutover_service()
-    for payload in _read_jsonl(input_path, QuarantineResolutionInput):
-        report = _operator_call(partial(service.resolve_quarantine, payload, apply=apply))
-        _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("shadow-start")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--at", "started_at", default=None, help="Optional explicit timezone-aware shadow start.")
-def shadow_start(tenant_id: str, expected_cas_version: int, started_at: str | None) -> None:
-    service = _cutover_service()
-    _operator_call(
-        lambda: service.begin_shadow(
-            tenant_id=tenant_id,
-            expected_cas_version=expected_cas_version,
-            started_at=_parse_timestamp(started_at) if started_at is not None else None,
-        )
-    )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("shadow-report")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist observations; omitted means dry-run.")
-def shadow_report(input_path: Path, apply: bool) -> None:
-    observations = _read_jsonl(input_path, ShadowAuthorizationObservationInput)
-    report = _operator_call(lambda: _cutover_service().record_shadow_report(observations, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("shadow-complete")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist completion; omitted means dry-run.")
-def shadow_complete(input_path: Path, apply: bool) -> None:
-    payload = _read_one_jsonl(input_path, ShadowCompletionInput)
-    report = _operator_call(lambda: _cutover_service().complete_shadow(payload, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("issue-approve")
-@click.option("--tenant-id", required=True)
-@click.option("--issue-key", required=True)
-@click.option("--account-id", required=True)
-@click.option("--at", "approved_at", required=True)
-def issue_approve(tenant_id: str, issue_key: str, account_id: str, approved_at: str) -> None:
-    _operator_call(
-        lambda: _cutover_service().approve_issue_fail_closed(
-            tenant_id=tenant_id,
-            issue_key=issue_key,
-            account_id=account_id,
-            approved_at=_parse_timestamp(approved_at),
-        )
-    )
-    _echo_json(_cutover_service().status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("issue-resolve")
-@click.option("--tenant-id", required=True)
-@click.option("--issue-key", required=True)
-@click.option("--account-id", required=True)
-@click.option("--at", "resolved_at", required=True)
-def issue_resolve(tenant_id: str, issue_key: str, account_id: str, resolved_at: str) -> None:
-    _operator_call(
-        lambda: _cutover_service().resolve_issue(
-            tenant_id=tenant_id,
-            issue_key=issue_key,
-            account_id=account_id,
-            resolved_at=_parse_timestamp(resolved_at),
-        )
-    )
-    _echo_json(_cutover_service().status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("shadow-approve")
-@click.option("--tenant-id", required=True)
-@click.option("--diff-key", required=True)
-@click.option("--account-id", required=True)
-@click.option("--at", "approved_at", required=True)
-def shadow_approve(tenant_id: str, diff_key: str, account_id: str, approved_at: str) -> None:
-    _operator_call(
-        lambda: _cutover_service().approve_shadow_diff(
-            tenant_id=tenant_id,
-            diff_key=diff_key,
-            account_id=account_id,
-            approved_at=_parse_timestamp(approved_at),
-        )
-    )
-    _echo_json(_cutover_service().status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("shadow-resolve")
-@click.option("--tenant-id", required=True)
-@click.option("--diff-key", required=True)
-@click.option("--account-id", required=True)
-@click.option("--at", "resolved_at", required=True)
-def shadow_resolve(tenant_id: str, diff_key: str, account_id: str, resolved_at: str) -> None:
-    _operator_call(
-        lambda: _cutover_service().resolve_shadow_diff(
-            tenant_id=tenant_id,
-            diff_key=diff_key,
-            account_id=account_id,
-            resolved_at=_parse_timestamp(resolved_at),
-        )
-    )
-    _echo_json(_cutover_service().status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("legacy-dashboard")
-@click.option("--tenant-id", required=True)
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--checked-at", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), default=None)
-@click.option("--apply", is_flag=True, default=False, help="Persist gate evidence; omitted means read-only report.")
-def legacy_dashboard(
-    tenant_id: str,
-    input_path: Path,
-    checked_at: str,
-    expected_cas_version: int | None,
-    apply: bool,
-) -> None:
-    dependencies = _read_jsonl(input_path, LegacyDependencyInput, allow_empty=True)
-    report = _operator_call(
-        lambda: _cutover_service().legacy_dependency_dashboard(
-            tenant_id=tenant_id,
-            dependencies=dependencies,
-            expected_cas_version=expected_cas_version,
-            checked_at=_parse_timestamp(checked_at),
-            apply=apply,
-        )
-    )
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("legacy-check")
-@click.option("--tenant-id", required=True)
-def legacy_check(tenant_id: str) -> None:
-    status_report = _operator_call(lambda: _cutover_service().status(tenant_id=tenant_id))
-    passed = (
-        bool(status_report["legacy_dependency_ready"])
-        and status_report["open_issues"] == 0
-        and status_report["unresolved_cutover_quarantine"] == 0
-    )
-    _echo_json({"tenant_id": tenant_id, "passed": passed, "status": status_report})
-    if not passed:
-        raise click.exceptions.Exit(1)
-
-
-@knowledge_fs_control_space.command("freeze")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--at", "freeze_at", required=True)
-def freeze(tenant_id: str, expected_cas_version: int, freeze_at: str) -> None:
-    service = _cutover_service()
-    _operator_call(
-        lambda: service.freeze(
-            tenant_id=tenant_id,
-            expected_cas_version=expected_cas_version,
-            freeze_at=_parse_timestamp(freeze_at),
-        )
-    )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("final-delta")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-def final_delta(input_path: Path) -> None:
-    payload = _read_one_jsonl(input_path, FinalDeltaInput)
-    service = _cutover_service()
-    _operator_call(lambda: service.apply_final_delta(payload))
-    _echo_json(service.status(tenant_id=str(payload.tenant_id)))
-
-
-@knowledge_fs_control_space.command("cutover")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--at", "cutover_at", required=True)
-@click.option("--rollback-cutoff-at", required=True)
-def cutover(tenant_id: str, expected_cas_version: int, cutover_at: str, rollback_cutoff_at: str) -> None:
-    service = _cutover_service()
-    _operator_call(
-        lambda: service.cutover(
-            tenant_id=tenant_id,
-            expected_cas_version=expected_cas_version,
-            cutover_at=_parse_timestamp(cutover_at),
-            rollback_cutoff_at=_parse_timestamp(rollback_cutoff_at),
-        )
-    )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("smoke")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-def smoke(tenant_id: str, expected_cas_version: int, input_path: Path) -> None:
-    results = _read_one_jsonl(input_path, CutoverSmokeResultsInput)
-    service = _cutover_service()
-    _operator_call(
-        lambda: service.record_smoke_results(
-            tenant_id=tenant_id,
-            expected_cas_version=expected_cas_version,
-            results=results,
-        )
-    )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("observe")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--started-at", default=None)
-@click.option("--window-ends-at", default=None)
-@click.option("--maximum-task-expires-at", default=None)
-@click.option("--observed-at", default=None)
-def observe(
-    tenant_id: str,
-    expected_cas_version: int,
-    started_at: str | None,
-    window_ends_at: str | None,
-    maximum_task_expires_at: str | None,
-    observed_at: str | None,
-) -> None:
-    service = _cutover_service()
-    if observed_at is not None:
-        if any(value is not None for value in (started_at, window_ends_at, maximum_task_expires_at)):
-            raise click.UsageError("--observed-at cannot be combined with observation start options")
-        _operator_call(
-            lambda: service.complete_observation(
-                tenant_id=tenant_id,
-                expected_cas_version=expected_cas_version,
-                observed_at=_parse_timestamp(observed_at),
-            )
-        )
-    else:
-        if started_at is None or window_ends_at is None or maximum_task_expires_at is None:
-            raise click.UsageError(
-                "observation start requires --started-at, --window-ends-at, and --maximum-task-expires-at"
-            )
-        _operator_call(
-            lambda: service.begin_observation(
-                tenant_id=tenant_id,
-                expected_cas_version=expected_cas_version,
-                started_at=_parse_timestamp(started_at),
-                window_ends_at=_parse_timestamp(window_ends_at),
-                maximum_task_expires_at=_parse_timestamp(maximum_task_expires_at),
-            )
-        )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("rollback")
-@click.option("--tenant-id", required=True)
-@click.option("--expected-cas-version", type=click.IntRange(min=0), required=True)
-@click.option("--at", "rolled_back_at", required=True)
-def rollback(tenant_id: str, expected_cas_version: int, rolled_back_at: str) -> None:
-    service = _cutover_service()
-    _operator_call(
-        lambda: service.rollback(
-            tenant_id=tenant_id,
-            expected_cas_version=expected_cas_version,
-            rolled_back_at=_parse_timestamp(rolled_back_at),
-        )
-    )
-    _echo_json(service.status(tenant_id=tenant_id))
-
-
-@knowledge_fs_control_space.command("status")
-@click.option("--tenant-id", required=True)
-def status(tenant_id: str) -> None:
-    _echo_json(_operator_call(lambda: _cutover_service().status(tenant_id=tenant_id)))
-
-
-@knowledge_fs_control_space.command("greenfield-initialize")
-@click.option("--tenant-id", required=True)
-def greenfield_initialize(tenant_id: str) -> None:
-    """Idempotently initialize one Workspace that has no KnowledgeFS state."""
-
-    _operator_call(lambda: _greenfield_initializer().ensure_initialized(tenant_id=tenant_id))
-    _echo_json(_operator_call(lambda: _cutover_service().status(tenant_id=tenant_id)))
-
-
-@knowledge_fs_control_space.command("cleanup-request")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist readiness evidence; omitted means dry-run.")
-def cleanup_request(input_path: Path, apply: bool) -> None:
-    payload = _read_one_jsonl(input_path, CleanupReadinessEvidenceInput)
-    report = _cleanup_call(lambda: _cleanup_service().request_cleanup(payload, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("cleanup-approve")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist four-eyes approval; omitted means dry-run.")
-def cleanup_approve(input_path: Path, apply: bool) -> None:
-    payload = _read_one_jsonl(input_path, CleanupApprovalInput)
-    report = _cleanup_call(lambda: _cleanup_service().approve_cleanup(payload, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("cleanup-start")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist the irreversible fence; never runs deletion.")
-@click.option(
-    "--acknowledge-irreversible",
-    is_flag=True,
-    default=False,
-    help="Required with --apply; confirms rollback will be permanently closed.",
-)
-def cleanup_start(input_path: Path, apply: bool, acknowledge_irreversible: bool) -> None:
-    payload = _read_one_jsonl(input_path, CleanupStartInput)
-    if apply and not acknowledge_irreversible:
-        raise click.UsageError("--apply requires --acknowledge-irreversible")
-    report = _cleanup_call(lambda: _cleanup_service().start_cleanup(payload, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("cleanup-complete")
-@click.option("--input", "input_path", type=click.Path(path_type=Path, exists=True, dir_okay=False), required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist externally verified cleanup completion.")
-@click.option(
-    "--acknowledge-executed",
-    is_flag=True,
-    default=False,
-    help="Required with --apply; confirms the reviewed destructive bundle already executed.",
-)
-def cleanup_complete(input_path: Path, apply: bool, acknowledge_executed: bool) -> None:
-    payload = _read_one_jsonl(input_path, CleanupCompletionEvidenceInput)
-    if apply and not acknowledge_executed:
-        raise click.UsageError("--apply requires --acknowledge-executed")
-    report = _cleanup_call(lambda: _cleanup_service().complete_cleanup(payload, apply=apply))
-    _echo_json(report._asdict())
-
-
-@knowledge_fs_control_space.command("cleanup-status")
-@click.option("--tenant-id", required=True)
-@click.option("--request-id", required=True)
-def cleanup_status(tenant_id: str, request_id: str) -> None:
-    _echo_json(_cleanup_call(lambda: _cleanup_service().status(tenant_id=tenant_id, request_id=request_id)))
-
-
-@knowledge_fs_control_space.command("repair")
-@click.option("--tenant-id", required=True)
-@click.option("--control-space-id", required=True)
-@click.option("--expected-resource-version", type=click.IntRange(min=0), required=True)
-@click.option("--knowledge-space-id", required=True)
-@click.option("--knowledge-space-revision", type=click.IntRange(min=0), required=True)
-def repair(
-    tenant_id: str,
-    control_space_id: str,
-    expected_resource_version: int,
-    knowledge_space_id: str,
-    knowledge_space_revision: int,
-) -> None:
-    control_space = _management_service().repair_registration(
-        tenant_id=tenant_id,
-        control_space_id=control_space_id,
-        expected_resource_version=expected_resource_version,
-        knowledge_space_id=knowledge_space_id,
-        knowledge_space_revision=knowledge_space_revision,
-    )
-    click.echo(json.dumps({"control_space_id": control_space.id, "state": control_space.state.value}, sort_keys=True))
-
-
-@knowledge_fs_control_space.command("orphan-report")
-@click.option("--limit", type=click.IntRange(min=1, max=10_000), default=500, show_default=True)
-def orphan_report(limit: int) -> None:
-    report = KnowledgeFSOrphanReconciler(
-        session_factory.get_session_maker(),
-        get_knowledge_fs_lifecycle_remote(),
-    ).reconcile(limit=limit, apply_repairs=False)
-    click.echo(json.dumps(report._asdict(), sort_keys=True))
-
-
-@knowledge_fs_control_space.command("workspace-delete-request")
-@click.option("--tenant-id", required=True)
-@click.option("--apply", is_flag=True, default=False, help="Persist durable deletion intents; omitted means dry-run.")
-def workspace_delete_request(tenant_id: str, apply: bool) -> None:
-    """Route every KnowledgeFS Space through the canonical lifecycle deletion path."""
-
-    if not apply:
-        report = _management_service().dry_run(tenant_id=tenant_id)
-        _echo_json(
-            {
-                "apply": False,
-                "by_state": report.by_state,
-                "tenant_id": tenant_id,
-                "total": report.total,
-            }
-        )
-        return
-    results = _lifecycle_call(lambda: _lifecycle_service().request_workspace_cleanup(tenant_id=tenant_id))
-    _echo_json(
-        {
-            "apply": True,
-            "control_space_ids": [result.control_space.id for result in results],
-            "operation_ids": [result.outbox.operation_id for result in results if result.outbox is not None],
-            "tenant_id": tenant_id,
-        }
-    )
-
-
-@knowledge_fs_control_space.command("workspace-delete-finalize")
-@click.option("--tenant-id", required=True)
-@click.option("--apply", is_flag=True, default=False, help="Purge terminal local control-plane rows.")
-@click.option(
-    "--acknowledge-control-plane-purge",
-    is_flag=True,
-    default=False,
-    help="Required with --apply after every remote Space has reached deleted.",
-)
-def workspace_delete_finalize(tenant_id: str, apply: bool, acknowledge_control_plane_purge: bool) -> None:
-    """Release the Workspace FK only after all remote deletions are terminal."""
-
-    service = _lifecycle_service()
-    if not apply:
-        _lifecycle_call(lambda: service.assert_workspace_deletion_allowed(tenant_id=tenant_id))
-        _echo_json({"apply": False, "ready": True, "tenant_id": tenant_id})
-        return
-    if not acknowledge_control_plane_purge:
-        raise click.UsageError("--apply requires --acknowledge-control-plane-purge")
-    deleted = _lifecycle_call(lambda: service.finalize_workspace_deletion(tenant_id=tenant_id))
-    _echo_json({"apply": True, "purged_control_spaces": deleted, "tenant_id": tenant_id})
-
-
-def _read_jsonl[InputT: BaseModel](
-    input_path: Path, input_type: type[InputT], *, allow_empty: bool = False
-) -> tuple[InputT, ...]:
-    records: list[InputT] = []
-    for line_number, line in enumerate(input_path.read_text(encoding="utf-8").splitlines(), start=1):
-        if not line.strip():
-            continue
-        try:
-            records.append(input_type.model_validate_json(line))
-        except ValidationError as exc:
-            raise click.ClickException(f"invalid strict JSONL at line {line_number}: {exc}") from exc
-    if not records and not allow_empty:
-        raise click.ClickException("strict JSONL input must contain at least one record")
-    return tuple(records)
-
-
-def _read_one_jsonl[InputT: BaseModel](input_path: Path, input_type: type[InputT]) -> InputT:
-    records = _read_jsonl(input_path, input_type)
-    if len(records) != 1:
-        raise click.ClickException("this command requires exactly one JSONL record")
-    return records[0]
-
-
-def _parse_timestamp(value: str) -> datetime:
-    try:
-        parsed = datetime.fromisoformat(value)
-    except ValueError as exc:
-        raise click.ClickException(f"invalid ISO-8601 timestamp: {value}") from exc
-    if parsed.tzinfo is None:
-        raise click.ClickException("operator timestamps must include an explicit timezone")
-    return parsed
-
-
-def _operator_call[ResultT](operation: Callable[[], ResultT]) -> ResultT:
-    try:
-        return operation()
-    except KnowledgeFSCutoverError as exc:
-        raise click.ClickException(str(exc)) from exc
-
-
-def _cleanup_call[ResultT](operation: Callable[[], ResultT]) -> ResultT:
-    try:
-        return operation()
-    except KnowledgeFSCleanupError as exc:
-        raise click.ClickException(str(exc)) from exc
-
-
-def _lifecycle_call[ResultT](operation: Callable[[], ResultT]) -> ResultT:
-    try:
-        return operation()
-    except KnowledgeFSControlSpaceLifecycleError as exc:
-        raise click.ClickException(str(exc)) from exc
-
-
-def _echo_json(payload: object) -> None:
-    click.echo(json.dumps(payload, default=str, sort_keys=True))
-
-
-def _management_service() -> KnowledgeFSControlSpaceManagementService:
-    return KnowledgeFSControlSpaceManagementService(session_factory.get_session_maker())
-
-
-def _cutover_service() -> KnowledgeFSWorkspaceCutoverService:
-    return KnowledgeFSWorkspaceCutoverService(
-        session_factory.get_session_maker(),
-        remote_factory=get_knowledge_fs_lifecycle_remote,
-    )
-
-
-def _greenfield_initializer() -> KnowledgeFSWorkspaceGreenfieldInitializer:
-    return KnowledgeFSWorkspaceGreenfieldInitializer(
-        session_factory.get_session_maker(),
-        cutover=_cutover_service(),
-    )
-
-
-def _cleanup_service() -> KnowledgeFSCleanupService:
-    return KnowledgeFSCleanupService(session_factory.get_session_maker())
-
-
-def _lifecycle_service() -> KnowledgeFSControlSpaceCommandService:
-    return KnowledgeFSControlSpaceCommandService(session_factory.get_session_maker())
-
-
-__all__ = ["knowledge_fs_control_space"]

api/commands/workflow_migration.py

@@ -0,0 +1,173 @@
+"""Workflow data migration CLI commands.
+
+TODO: Remove the legacy system file workflow migration command after the production migration is complete.
+"""
+
+import logging
+from dataclasses import dataclass
+
+import click
+from sqlalchemy import select
+from sqlalchemy.orm import Session, load_only, sessionmaker
+
+from extensions.ext_database import db
+from models.workflow import Workflow, WorkflowType
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class LegacySysFilesWorkflowMigrationStats:
+    scanned: int = 0
+    migrated: int = 0
+    failed: int = 0
+    batches: int = 0
+    last_id: str | None = None
+
+
+def _build_legacy_sys_files_workflow_query(
+    *,
+    start_after_id: str | None,
+    batch_size: int,
+    tenant_id: str | None,
+    app_id: str | None,
+):
+    # Workflow IDs are UUID4, so this is not chronological pagination. The migration only needs a stable total
+    # order that matches the resume cursor; ordering by the same primary-key column used in the `id > cursor`
+    # predicate lets each batch continue deterministically without offset scans.
+    stmt = (
+        select(Workflow)
+        .options(load_only(Workflow.id, Workflow.type, Workflow.graph))
+        .where(Workflow.type.in_((WorkflowType.WORKFLOW, WorkflowType.CHAT)))
+        .order_by(Workflow.id)
+        .limit(batch_size)
+    )
+    if start_after_id:
+        stmt = stmt.where(Workflow.id > start_after_id)
+    if tenant_id:
+        stmt = stmt.where(Workflow.tenant_id == tenant_id)
+    if app_id:
+        stmt = stmt.where(Workflow.app_id == app_id)
+    return stmt
+
+
+def _migrate_legacy_sys_files_workflow_batch(
+    *,
+    session: Session,
+    start_after_id: str | None,
+    batch_size: int,
+    tenant_id: str | None,
+    app_id: str | None,
+    dry_run: bool,
+) -> LegacySysFilesWorkflowMigrationStats:
+    stats = LegacySysFilesWorkflowMigrationStats()
+    workflows = session.scalars(
+        _build_legacy_sys_files_workflow_query(
+            start_after_id=start_after_id,
+            batch_size=batch_size,
+            tenant_id=tenant_id,
+            app_id=app_id,
+        )
+    ).all()
+
+    for workflow in workflows:
+        stats.scanned += 1
+        stats.last_id = workflow.id
+        try:
+            if workflow.migrate_legacy_sys_files_graph_in_place():
+                stats.migrated += 1
+        except Exception:
+            stats.failed += 1
+            logger.exception("Failed to migrate legacy sys.files workflow, workflow_id=%s", workflow.id)
+
+    if dry_run:
+        session.rollback()
+    else:
+        session.commit()
+    return stats
+
+
+def run_legacy_sys_files_workflow_migration(
+    *,
+    batch_size: int,
+    limit: int | None,
+    start_after_id: str | None,
+    tenant_id: str | None,
+    app_id: str | None,
+    dry_run: bool,
+) -> LegacySysFilesWorkflowMigrationStats:
+    """Scan Workflow and Advanced Chat graphs in keyset-paginated batches."""
+    if batch_size <= 0:
+        raise click.UsageError("--batch-size must be greater than 0")
+    if limit is not None and limit <= 0:
+        raise click.UsageError("--limit must be greater than 0 when provided")
+
+    session_maker = sessionmaker(db.engine, expire_on_commit=False)
+    total = LegacySysFilesWorkflowMigrationStats(last_id=start_after_id)
+    next_start_after_id = start_after_id
+
+    while limit is None or total.scanned < limit:
+        remaining = None if limit is None else limit - total.scanned
+        current_batch_size = batch_size if remaining is None else min(batch_size, remaining)
+        if current_batch_size <= 0:
+            break
+
+        with session_maker() as session:
+            batch_stats = _migrate_legacy_sys_files_workflow_batch(
+                session=session,
+                start_after_id=next_start_after_id,
+                batch_size=current_batch_size,
+                tenant_id=tenant_id,
+                app_id=app_id,
+                dry_run=dry_run,
+            )
+
+        if batch_stats.scanned == 0:
+            break
+
+        total.scanned += batch_stats.scanned
+        total.migrated += batch_stats.migrated
+        total.failed += batch_stats.failed
+        total.batches += 1
+        total.last_id = batch_stats.last_id
+        next_start_after_id = batch_stats.last_id
+
+        if batch_stats.scanned < current_batch_size:
+            break
+
+    return total
+
+
+@click.command(
+    "migrate-legacy-sys-files-workflows",
+    help="Migrate Workflow and Advanced Chat graphs that still reference deprecated sys.files.",
+)
+@click.option("--batch-size", default=1000, show_default=True, type=int, help="Number of workflows to scan per batch.")
+@click.option("--limit", default=None, type=int, help="Maximum number of workflows to scan in this run.")
+@click.option("--start-after-id", default=None, help="Resume scanning after this workflow ID.")
+@click.option("--tenant-id", default=None, help="Limit migration to one tenant.")
+@click.option("--app-id", default=None, help="Limit migration to one app.")
+@click.option("--dry-run", is_flag=True, default=False, help="Scan and report without saving changes.")
+def migrate_legacy_sys_files_workflows(
+    batch_size: int,
+    limit: int | None,
+    start_after_id: str | None,
+    tenant_id: str | None,
+    app_id: str | None,
+    dry_run: bool,
+) -> None:
+    stats = run_legacy_sys_files_workflow_migration(
+        batch_size=batch_size,
+        limit=limit,
+        start_after_id=start_after_id,
+        tenant_id=tenant_id,
+        app_id=app_id,
+        dry_run=dry_run,
+    )
+    click.echo(
+        "Legacy sys.files workflow migration finished: "
+        f"scanned={stats.scanned} migrated={stats.migrated} failed={stats.failed} "
+        f"batches={stats.batches} last_id={stats.last_id or ''}"
+    )
+    if dry_run:
+        click.echo("Dry run only: no workflow graph changes were saved.")

api/commands/workspace.py

@@ -0,0 +1,64 @@
+"""Workspace maintenance CLI commands."""
+
+import click
+from sqlalchemy import delete, select
+from sqlalchemy.orm import sessionmaker
+
+from configs import dify_config
+from enums.deployment_edition import DeploymentEdition
+from extensions.ext_database import db
+from libs.rsa import generate_key_pair
+from models import Tenant
+from models.provider import Provider, ProviderModel
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider
+
+
+@click.command(
+    "reset-encrypt-key-pair",
+    help="Reset the asymmetric key pair of workspace for encrypt LLM credentials. "
+    "After the reset, all LLM credentials and tool provider credentials "
+    "(builtin / API / MCP) will be purged, requiring re-entry. "
+    "Only support SELF_HOSTED mode.",
+)
+@click.confirmation_option(
+    prompt=click.style(
+        "Are you sure you want to reset encrypt key pair? "
+        "This will also purge builtin / API / MCP tool provider records for every tenant. "
+        "This operation cannot be rolled back!",
+        fg="red",
+    )
+)
+def reset_encrypt_key_pair() -> None:
+    """
+    Reset the encrypted key pair of workspace for encrypt LLM credentials.
+    After the reset, all LLM credentials will become invalid, requiring re-entry.
+    Only support SELF_HOSTED mode.
+    """
+    if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
+        click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
+        return
+    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+        tenants = session.scalars(select(Tenant)).all()
+        for tenant in tenants:
+            if not tenant:
+                click.echo(click.style("No workspaces found. Run /install first.", fg="red"))
+                return
+
+            tenant.encrypt_public_key = generate_key_pair(tenant.id)
+
+            session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
+            session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
+
+            # Purge tool provider records that hold credentials encrypted under the
+            # tenant key. Leaving them in place causes /console/api/workspaces/current/
+            # tool-providers to 500 because decryption fails on stale ciphertext (#35396).
+            session.execute(delete(BuiltinToolProvider).where(BuiltinToolProvider.tenant_id == tenant.id))
+            session.execute(delete(ApiToolProvider).where(ApiToolProvider.tenant_id == tenant.id))
+            session.execute(delete(MCPToolProvider).where(MCPToolProvider.tenant_id == tenant.id))
+
+            click.echo(
+                click.style(
+                    f"Congratulations! The asymmetric key pair of workspace {tenant.id} has been reset.",
+                    fg="green",
+                )
+            )

api/configs/extra/knowledge_fs_config.py

@@ -1,72 +1,30 @@
-"""Configuration for the optional KnowledgeFS control-plane integration."""
+"""Configuration for the optional KnowledgeFS Console bridge."""
 
-from ipaddress import ip_address
 from urllib.parse import urlsplit
 
-from pydantic import Field, PositiveFloat, PositiveInt, SecretStr, field_validator, model_validator
+from pydantic import Field, PositiveFloat, SecretStr, field_validator, model_validator
 from pydantic_settings import BaseSettings
 
 
 class KnowledgeFSConfig(BaseSettings):
-    """Server-only KnowledgeFS connection and rollout settings."""
+    """Server-only settings for the KnowledgeFS production connection."""
 
     KNOWLEDGE_FS_ENABLED: bool = Field(
         default=False,
-        description="Enable the KnowledgeFS control-plane product routes.",
+        description="Enable the private KnowledgeFS Console bridge.",
     )
-    KNOWLEDGE_FS_LIFECYCLE_WORKER_ENABLED: bool = Field(
-        default=False,
-        description="Enable delivery of durable KnowledgeFS lifecycle commands after every rollout gate is ready.",
-    )
-    KNOWLEDGE_FS_INTEGRATED_PROVISION_READY: bool = Field(
-        default=False,
-        description="Confirm that the Capability-v2 integrated provision route is deployed and verified.",
-    )
-    KNOWLEDGE_FS_LEGACY_ACL_FREEZE_READY: bool = Field(
-        default=False,
-        description="Confirm that legacy KFS ACL mutation is frozen for integrated mode.",
-    )
-    KNOWLEDGE_FS_LIFECYCLE_POLL_INTERVAL_SECONDS: PositiveInt = Field(default=15, le=300)
-    KNOWLEDGE_FS_LIFECYCLE_LEASE_SECONDS: PositiveInt = Field(default=60, le=600)
-    KNOWLEDGE_FS_LIFECYCLE_BATCH_SIZE: PositiveInt = Field(default=25, le=1_000)
     KNOWLEDGE_FS_BASE_URL: str | None = Field(default=None, description="KnowledgeFS gateway base URL.")
-    KNOWLEDGE_FS_DIRECT_ORIGIN: str | None = Field(
+    KNOWLEDGE_FS_JWT_SECRET: SecretStr | None = Field(
         default=None,
-        description="Public KnowledgeFS origin returned with direct upload capabilities.",
+        min_length=32,
+        description="Shared secret used to sign short-lived KnowledgeFS service JWTs.",
     )
-    KNOWLEDGE_FS_DIRECT_UPLOAD_READY: bool = Field(
-        default=False,
-        description="Confirm that KnowledgeFS direct upload and its browser origin policy are deployed and verified.",
-    )
-    KNOWLEDGE_FS_CAPABILITY_V2_ENABLED: bool = Field(
-        default=False,
-        description="Prepare resource-scoped Capability v2 issuance; disabled until rollout approval.",
-    )
-    KNOWLEDGE_FS_CAPABILITY_V2_SIGNING_KID: str | None = Field(
-        default=None,
-        description="Identifier for the current asymmetric Capability v2 signing key.",
-    )
-    KNOWLEDGE_FS_CAPABILITY_V2_PRIVATE_KEY_PEM: SecretStr | None = Field(
-        default=None,
-        description="Server-only PEM for the current Capability v2 RSA signing key.",
-    )
-    KNOWLEDGE_FS_CAPABILITY_V2_PREVIOUS_PUBLIC_JWKS: str | None = Field(
-        default=None,
-        description="Optional public-only JWKS JSON retained during key rotation overlap.",
-    )
-    KNOWLEDGE_FS_CAPABILITY_V2_ISSUER: str = Field(default="dify-control-plane", min_length=1)
-    KNOWLEDGE_FS_CAPABILITY_V2_AUDIENCE: str = Field(default="knowledge-fs", min_length=1)
-    KNOWLEDGE_FS_CAPABILITY_V2_MAX_TTL_SECONDS: PositiveInt = Field(default=60, le=60)
-    KNOWLEDGE_FS_JWKS_CACHE_MAX_AGE_SECONDS: PositiveInt = Field(default=300, le=86_400)
-    KNOWLEDGE_FS_PRODUCT_MAX_RESPONSE_BYTES: PositiveInt = Field(default=4 * 1024 * 1024, le=16 * 1024 * 1024)
+    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_DIRECT_ORIGIN",
-        "KNOWLEDGE_FS_CAPABILITY_V2_SIGNING_KID",
-        "KNOWLEDGE_FS_CAPABILITY_V2_PRIVATE_KEY_PEM",
-        "KNOWLEDGE_FS_CAPABILITY_V2_PREVIOUS_PUBLIC_JWKS",
+        "KNOWLEDGE_FS_JWT_SECRET",
         mode="before",
     )
     @classmethod
@@ -82,57 +40,25 @@ class KnowledgeFSConfig(BaseSettings):
     @field_validator("KNOWLEDGE_FS_BASE_URL")
     @classmethod
     def validate_base_url(cls, value: str | None) -> str | None:
-        return cls._validate_origin(value, name="KNOWLEDGE_FS_BASE_URL")
-
-    @field_validator("KNOWLEDGE_FS_DIRECT_ORIGIN")
-    @classmethod
-    def validate_direct_origin(cls, value: str | None) -> str | None:
-        return cls._validate_origin(value, name="KNOWLEDGE_FS_DIRECT_ORIGIN")
-
-    @classmethod
-    def _validate_origin(cls, value: str | None, *, name: str) -> str | None:
         if value is None:
             return None
         parsed = urlsplit(value)
         if parsed.scheme not in {"http", "https"} or not parsed.netloc:
-            raise ValueError(f"{name} must be an absolute HTTP(S) URL")
+            raise ValueError("KNOWLEDGE_FS_BASE_URL must be an absolute HTTP(S) URL")
         try:
             _ = parsed.port
         except ValueError as exc:
-            raise ValueError(f"{name} must include a valid port") from exc
-        if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
-            raise ValueError(f"{name} must be an origin without credentials, path, query, or fragment")
+            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 str(getattr(self, "DEPLOY_ENV", "")).strip().upper() == "PRODUCTION":
-            for name, value in (
-                ("KNOWLEDGE_FS_BASE_URL", self.KNOWLEDGE_FS_BASE_URL),
-                ("KNOWLEDGE_FS_DIRECT_ORIGIN", self.KNOWLEDGE_FS_DIRECT_ORIGIN),
-            ):
-                if value and not self._is_secure_or_loopback_origin(value):
-                    raise ValueError(f"{name} must use HTTPS in production unless it targets loopback")
-        if self.KNOWLEDGE_FS_ENABLED:
-            if not self.KNOWLEDGE_FS_BASE_URL:
-                raise ValueError("KnowledgeFS base URL is required when the integration is enabled")
-            if not self.KNOWLEDGE_FS_CAPABILITY_V2_ENABLED:
-                raise ValueError("KnowledgeFS product routes require Capability v2 when enabled")
-        if self.KNOWLEDGE_FS_CAPABILITY_V2_ENABLED and not (
-            self.KNOWLEDGE_FS_CAPABILITY_V2_SIGNING_KID and self.KNOWLEDGE_FS_CAPABILITY_V2_PRIVATE_KEY_PEM
-        ):
-            raise ValueError("Capability v2 signing kid and private key are required when issuance is enabled")
+        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
-
-    @staticmethod
-    def _is_secure_or_loopback_origin(value: str) -> bool:
-        parsed = urlsplit(value)
-        if parsed.scheme == "https":
-            return True
-        hostname = (parsed.hostname or "").rstrip(".").lower()
-        if hostname == "localhost":
-            return True
-        try:
-            return ip_address(hostname).is_loopback
-        except ValueError:
-            return False

api/constants/__init__.py

@@ -77,5 +77,3 @@ COOKIE_NAME_PASSPORT = "passport"
 HEADER_NAME_CSRF_TOKEN = "X-CSRF-Token"
 HEADER_NAME_APP_CODE = "X-App-Code"
 HEADER_NAME_PASSPORT = "X-App-Passport"
-HEADER_NAME_IDEMPOTENCY_KEY = "Idempotency-Key"
-HEADER_NAME_REQUEST_ID = "X-Request-ID"

api/controllers/common/wraps.py

@@ -10,6 +10,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.dataset import Dataset
 from models.model import App
+from services.agent.roster_service import AgentRosterService
 from services.enterprise.rbac_service import RBACService
 
 __all__ = ["RBACPermission", "RBACResourceScope", "enforce_rbac_access", "rbac_permission_required"]
@@ -51,7 +52,7 @@ def enforce_rbac_access(
     check_resource_type = None if resource_type == RBACResourceScope.WORKSPACE else resource_type
     resource_id = None
     if resource_required and check_resource_type:
-        resource_id = _extract_resource_id(resource_type, path_args)
+        resource_id = _extract_resource_id(resource_type, tenant_id, path_args)
         if _is_resource_owned_by_current_user(tenant_id, account_id, resource_type, resource_id):
             return
     allowed = RBACService.CheckAccess.check(
@@ -131,11 +132,14 @@ def _is_resource_owned_by_current_user(
     return False
 
 
-def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str, object] | None = None) -> str:
+def _extract_resource_id(
+    resource_type: RBACResourceScope, tenant_id: str, path_args: dict[str, object] | None = None
+) -> str:
     """Extract the resource ID from matched path arguments.
 
     Some legacy route classes use neutral names such as ``resource_id`` for
-    app/dataset resources, and Agent App routes use ``agent_id`` as the app id.
+    app/dataset resources, and Agent routes carry ``agent_id``, which is
+    resolved to the App backing that Agent.
     Dataset endpoints behind a rag-pipeline route contain ``pipeline_id``
     instead of ``dataset_id``. In that case we look up the associated
     ``Dataset`` row via ``Dataset.pipeline_id``.
@@ -146,10 +150,19 @@ def _extract_resource_id(resource_type: RBACResourceScope, path_args: dict[str,
     matched_args = {**view_args, **(path_args or {})}
 
     if resource_type == RBACResourceScope.APP:
-        app_id = matched_args.get("app_id") or matched_args.get("agent_id") or matched_args.get("resource_id")
-        if not app_id:
-            raise ValueError("Missing app_id in request path")
-        return str(app_id)
+        app_id = matched_args.get("app_id")
+        if app_id:
+            return str(app_id)
+
+        agent_id = matched_args.get("agent_id")
+        if agent_id:
+            authz_app_id = AgentRosterService(db.session).peek_authz_app_id(tenant_id=tenant_id, agent_id=str(agent_id))
+            return authz_app_id or str(agent_id)
+
+        resource_id = matched_args.get("resource_id")
+        if resource_id:
+            return str(resource_id)
+        raise ValueError("Missing app_id in request path")
 
     if resource_type == RBACResourceScope.DATASET:
         dataset_id = matched_args.get("dataset_id") or matched_args.get("resource_id")

api/controllers/console/__init__.py

@@ -38,6 +38,7 @@ from . import (
     feature,
     human_input_form,
     init_validate,
+    knowledge_fs_proxy,
     notification,
     onboarding,
     ping,
@@ -126,7 +127,6 @@ from .explore import (
     saved_message,
     trial,
 )
-from .knowledge_fs import resources as knowledge_fs_resources
 from .snippets import snippet_workflow, snippet_workflow_draft_variable
 from .socketio import workflow as socketio_workflow
 
@@ -197,7 +197,7 @@ __all__ = [
     "human_input_form",
     "init_validate",
     "installed_app",
-    "knowledge_fs_resources",
+    "knowledge_fs_proxy",
     "load_balancing_config",
     "login",
     "mcp_server",

api/controllers/console/agent/composer.py

@@ -230,6 +230,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session
@@ -439,6 +440,7 @@ class SnippetAgentComposerSaveToRosterApi(Resource):
     @rbac_permission_required(
         RBACResourceScope.WORKSPACE, RBACPermission.SNIPPETS_CREATE_AND_MODIFY, resource_required=False
     )
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session
@@ -478,6 +480,7 @@ class AgentComposerApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user_id
     @with_current_tenant_id
     @with_session

api/controllers/console/agent/roster.py

@@ -534,9 +534,23 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:
 
 
 def _query_values(name: str, alias_name: str | None = None) -> list[str]:
-    values = request.args.getlist(name)
+    def _get_values(field_name: str) -> list[str]:
+        values = request.args.getlist(field_name)
+        indexed_values: list[tuple[int, list[str]]] = []
+        prefix = f"{field_name}["
+        for key in request.args:
+            if not key.startswith(prefix) or not key.endswith("]"):
+                continue
+            index = key[len(prefix) : -1]
+            if index.isdigit():
+                indexed_values.append((int(index), request.args.getlist(key)))
+        for _, items in sorted(indexed_values):
+            values.extend(items)
+        return values
+
+    values = _get_values(name)
     if alias_name:
-        values.extend(request.args.getlist(alias_name))
+        values.extend(_get_values(alias_name))
     return [value.strip() for value in values if value.strip()]
 
 
@@ -547,6 +561,7 @@ class AgentAppListApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -584,6 +599,7 @@ class AgentAppListApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -625,6 +641,7 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -650,6 +667,7 @@ class AgentAppApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session
     def delete(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -707,6 +725,7 @@ class AgentPublishApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -729,6 +748,7 @@ class AgentBuildDraftCheckoutApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -805,6 +825,7 @@ class AgentBuildDraftApplyApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -827,6 +848,7 @@ class AgentAppCopyApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -852,6 +874,7 @@ class AgentApiAccessApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -868,6 +891,7 @@ class AgentApiStatusApi(Resource):
     @login_required
     @is_admin_or_owner_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_current_tenant_id
     @with_session
@@ -886,6 +910,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
     token_prefix = "app-"
 
     @console_ns.response(200, "Agent service API keys", console_ns.models[ApiKeyList.__name__])
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID) -> dict[str, object]:
@@ -896,6 +921,7 @@ class AgentApiKeyListApi(BaseApiKeyListResource):
     @console_ns.response(400, "Maximum keys exceeded")
     @with_current_tenant_id
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_session
     def post(self, session: Session, tenant_id: str, agent_id: UUID) -> tuple[dict[str, object], int]:
@@ -915,6 +941,7 @@ class AgentApiKeyApi(BaseApiKeyResource):
     @console_ns.response(204, "Agent service API key deleted")
     @with_current_user
     @with_current_tenant_id
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
     @with_session
     def delete(
@@ -960,6 +987,7 @@ class AgentLogsApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -998,6 +1026,7 @@ class AgentLogMessagesApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1036,6 +1065,7 @@ class AgentLogSourcesApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1056,6 +1086,7 @@ class AgentStatisticsSummaryApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session(write=False)
@@ -1081,6 +1112,7 @@ class AgentRosterVersionsApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID):
@@ -1096,6 +1128,7 @@ class AgentRosterVersionDetailApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_tenant_id
     @with_session(write=False)
     def get(self, session: Session, tenant_id: str, agent_id: UUID, version_id: UUID):
@@ -1116,6 +1149,7 @@ class AgentRosterVersionRestoreApi(Resource):
     @login_required
     @account_initialization_required
     @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
     @with_current_user
     @with_current_tenant_id
     @with_session

api/controllers/console/apikey.py

@@ -12,6 +12,7 @@ from werkzeug.exceptions import Forbidden
 from configs import dify_config
 from controllers.common.schema import register_response_schema_models
 from controllers.common.session import with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app
 from fields.base import ResponseModel
 from libs.helper import dump_response, to_timestamp
 from libs.login import login_required
@@ -194,6 +195,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @console_ns.doc(params={"resource_id": "App ID"})
     @console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
     @with_current_tenant_id
+    @agent_manage_required_for_agent_app
     @with_session(write=False)
     def get(self, session: Session, current_tenant_id: str, resource_id: UUID) -> dict[str, object]:
         """Get all API keys for an app"""
@@ -210,6 +212,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
     @with_current_tenant_id
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     def post(self, session: Session, current_tenant_id: str, resource_id: UUID) -> tuple[dict[str, object], int]:
         """Create a new API key for an app"""
@@ -233,6 +236,7 @@ class AppApiKeyResource(BaseApiKeyResource):
     @with_current_user
     @with_current_tenant_id
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     def delete(
         self,

api/controllers/console/app/app.py

@@ -9,7 +9,7 @@ from flask_restx import Resource
 from pydantic import AliasChoices, BaseModel, Field, ValidationInfo, computed_field, field_validator, model_validator
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, NotFound
+from werkzeug.exceptions import BadRequest, Forbidden, NotFound
 
 from configs import dify_config
 from controllers.common.app_access import resolve_app_access_filter
@@ -23,7 +23,7 @@ from controllers.common.schema import (
     register_schema_models,
 )
 from controllers.console import console_ns
-from controllers.console.app.wraps import get_app_model, with_session
+from controllers.console.app.wraps import agent_manage_required_for_agent_app, get_app_model, with_session
 from controllers.console.workspace.models import LoadBalancingPayload
 from controllers.console.wraps import (
     RBACPermission,
@@ -76,6 +76,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     WeightModel,
     WeightVectorSetting,
 )
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService
 from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
@@ -904,6 +905,7 @@ class AppApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def put(self, session: Session, app_model: App):
@@ -938,6 +940,7 @@ class AppApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_DELETE)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model
     def delete(self, session: Session, app_model: App):
@@ -962,6 +965,7 @@ class AppCopyApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_CREATE_AND_MANAGEMENT)
+    @agent_manage_required_for_agent_app
     @with_current_user
     @with_current_tenant_id
     @get_app_model(mode=None)
@@ -973,16 +977,19 @@ class AppCopyApi(Resource):
         with Session(db.engine, expire_on_commit=False) as session:
             import_service = AppDslService(session)
             yaml_content = import_service.export_dsl(app_model=app_model, session=session, include_secret=True)
-            result = import_service.import_app(
-                account=current_user,
-                import_mode=ImportMode.YAML_CONTENT,
-                yaml_content=yaml_content,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-            )
+            try:
+                result = import_service.import_app(
+                    account=current_user,
+                    import_mode=ImportMode.YAML_CONTENT,
+                    yaml_content=yaml_content,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
                 return dump_response(AppImportResponse, result), 400
@@ -1036,6 +1043,7 @@ class AppExportApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
     @get_app_model
     def get(self, app_model: App):
         """Export app"""
@@ -1060,6 +1068,7 @@ class AppPublishToCreatorsPlatformApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_IMPORT_EXPORT_DSL)
+    @agent_manage_required_for_agent_app
     @with_current_user_id
     @get_app_model(mode=None)
     def post(self, current_user_id: str, app_model: App):
@@ -1090,6 +1099,7 @@ class AppNameApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1117,6 +1127,7 @@ class AppIconApi(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1150,6 +1161,7 @@ class AppSiteStatus(Resource):
     @account_initialization_required
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):
@@ -1177,6 +1189,7 @@ class AppApiStatus(Resource):
     @is_admin_or_owner_required
     @account_initialization_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_RELEASE_AND_VERSION)
+    @agent_manage_required_for_agent_app
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, app_model: App):

api/controllers/console/app/app_import.py

@@ -1,6 +1,7 @@
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from controllers.common.schema import register_enum_models, register_schema_models
@@ -28,6 +29,7 @@ from services.app_dsl_service import (
 )
 from services.enterprise.enterprise_service import EnterpriseService
 from services.entities.dsl_entities import CheckDependenciesResult, ImportStatus
+from services.errors.account import NoPermissionError
 from services.feature_service import FeatureService
 
 from .. import console_ns
@@ -91,18 +93,21 @@ class AppImportApi(Resource):
             import_service = AppDslService(session)
             # Import app
             account = current_user
-            result = import_service.import_app(
-                account=account,
-                import_mode=args.mode,
-                yaml_content=args.yaml_content,
-                yaml_url=args.yaml_url,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
-                app_id=args.app_id,
-            )
+            try:
+                result = import_service.import_app(
+                    account=account,
+                    import_mode=args.mode,
+                    yaml_content=args.yaml_content,
+                    yaml_url=args.yaml_url,
+                    name=args.name,
+                    description=args.description,
+                    icon_type=args.icon_type,
+                    icon=args.icon,
+                    icon_background=args.icon_background,
+                    app_id=args.app_id,
+                )
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:
@@ -157,7 +162,10 @@ class AppImportConfirmApi(Resource):
             import_service = AppDslService(session)
             # Confirm import
             account = current_user
-            result = import_service.confirm_import(import_id=import_id, account=account)
+            try:
+                result = import_service.confirm_import(import_id=import_id, account=account)
+            except NoPermissionError as e:
+                raise Forbidden(str(e))
             if result.status == ImportStatus.FAILED:
                 session.rollback()
             else:

api/controllers/console/app/completion.py

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

api/controllers/console/app/site.py

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

api/controllers/console/app/wraps.py

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

api/controllers/console/auth/activate.py

@@ -7,10 +7,13 @@ from configs import dify_config
 from constants.languages import supported_language
 from controllers.common.schema import query_params_from_model, register_schema_models
 from controllers.console import console_ns
+from controllers.console.auth.error import InvitationAccountMismatchError
 from controllers.console.error import AccountInFreezeError, AlreadyActivateError
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, timezone
+from libs.login import current_account_with_tenant
+from libs.token import extract_access_token
 from models import AccountStatus
 from models.account import TenantAccountJoin, TenantAccountRole
 from services.account_service import RegisterService, TenantService
@@ -136,6 +139,12 @@ class ActivateApi(Resource):
     )
     @console_ns.response(400, "Already activated or invalid token")
     def post(self):
+        """Accept an invitation without letting an existing session act for another account.
+
+        Token-only activation remains available for legacy clients. When the request already
+        carries a console session, that session must belong to the account encoded in the
+        invitation before the token is consumed or tenant membership is changed.
+        """
         args = ActivatePayload.model_validate(console_ns.payload)
 
         normalized_request_email = args.email.lower() if args.email else None
@@ -146,6 +155,11 @@ class ActivateApi(Resource):
             raise AlreadyActivateError()
 
         account = invitation["account"]
+        if extract_access_token(request):
+            current_account, _ = current_account_with_tenant()
+            if current_account.id != account.id:
+                raise InvitationAccountMismatchError()
+
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(account.email):
             raise AccountInFreezeError()
 

api/controllers/console/auth/error.py

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

api/controllers/console/auth/oauth_server.py

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

api/controllers/console/init_validate.py

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

api/controllers/console/knowledge_fs/__init__.py

@@ -1,5 +0,0 @@
-"""Typed Dify-owned KnowledgeFS Console product API."""
-
-from . import resources
-
-__all__ = ["resources"]

api/controllers/console/knowledge_fs/error.py

@@ -1,56 +0,0 @@
-"""Stable, non-enumerating Console error contract for KnowledgeFS."""
-
-from libs.exception import BaseHTTPException
-
-
-class KnowledgeFSSpaceNotFoundHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_space_not_found"
-    description = "KnowledgeFS space was not found."
-    code = 404
-
-
-class KnowledgeFSOperationUnavailableHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_operation_unavailable"
-    description = "KnowledgeFS operation is not available."
-    code = 503
-
-
-class KnowledgeFSUpstreamUnavailableHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_upstream_unavailable"
-    description = "KnowledgeFS is unavailable."
-    code = 502
-
-
-class KnowledgeFSInvalidRequestHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_invalid_request"
-    description = "KnowledgeFS request is invalid."
-    code = 400
-
-
-class KnowledgeFSAccessDeniedHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_access_denied"
-    description = "KnowledgeFS operation is not allowed."
-    code = 403
-
-
-class KnowledgeFSRateLimitHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_rate_limit_exceeded"
-    description = "KnowledgeFS operation rate limit exceeded."
-    code = 429
-
-
-class KnowledgeFSQuotaExceededHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_quota_exceeded"
-    description = "KnowledgeFS operation quota exceeded."
-    code = 403
-
-
-__all__ = [
-    "KnowledgeFSAccessDeniedHTTPError",
-    "KnowledgeFSInvalidRequestHTTPError",
-    "KnowledgeFSOperationUnavailableHTTPError",
-    "KnowledgeFSQuotaExceededHTTPError",
-    "KnowledgeFSRateLimitHTTPError",
-    "KnowledgeFSSpaceNotFoundHTTPError",
-    "KnowledgeFSUpstreamUnavailableHTTPError",
-]

api/controllers/console/knowledge_fs_proxy.py

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

api/controllers/console/setup.py

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

api/controllers/console/workspace/account.py

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

api/controllers/console/workspace/plugin.py

@@ -442,12 +442,10 @@ register_enum_models(
 )
 
 
-def _default_auto_upgrade_settings(
-    tenant_id: str,
-    category: TenantPluginAutoUpgradeCategory,
-) -> AutoUpgradeSettingsResponse:
+def _missing_auto_upgrade_settings(tenant_id: str) -> AutoUpgradeSettingsResponse:
+    """Represent a missing persisted strategy as effectively disabled."""
     return {
-        "strategy_setting": PluginAutoUpgradeService.default_strategy_setting_for_category(category),
+        "strategy_setting": TenantPluginAutoUpgradeStrategySetting.DISABLED,
         "upgrade_time_of_day": PluginAutoUpgradeService.default_upgrade_time_of_day(tenant_id),
         "upgrade_mode": TenantPluginAutoUpgradeMode.EXCLUDE,
         "exclude_plugins": [],
@@ -1135,9 +1133,7 @@ class PluginFetchAutoUpgradeApi(Resource):
         args = ParserAutoUpgradeFetch.model_validate(request.args.to_dict(flat=True))
         auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id, args.category, session=db.session())
         auto_upgrade_dict = (
-            _auto_upgrade_settings_to_dict(auto_upgrade)
-            if auto_upgrade
-            else _default_auto_upgrade_settings(tenant_id, args.category)
+            _auto_upgrade_settings_to_dict(auto_upgrade) if auto_upgrade else _missing_auto_upgrade_settings(tenant_id)
         )
 
         return jsonable_encoder(

api/controllers/console/workspace/rbac.py

@@ -346,13 +346,7 @@ class RBACRoleItemApi(Resource):
     def put(self, role_id):
         tenant_id, account_id = _current_ids()
         request = _payload(_RoleUpsertRequest)
-        role = svc.RBACService.KnowledgeFSRoleMutations.update_role(
-            tenant_id,
-            account_id,
-            str(role_id),
-            request.to_mutation(),
-            session=db.session(),
-        )
+        role = svc.RBACService.Roles.update(tenant_id, account_id, str(role_id), request.to_mutation())
         return _dump(role)
 
     @login_required
@@ -362,12 +356,7 @@ class RBACRoleItemApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[svc.RBACRole.__name__])
     def delete(self, role_id):
         tenant_id, account_id = _current_ids()
-        svc.RBACService.KnowledgeFSRoleMutations.delete_role(
-            tenant_id,
-            account_id,
-            str(role_id),
-            session=db.session(),
-        )
+        svc.RBACService.Roles.delete(tenant_id, account_id, str(role_id))
         return {"result": "success"}
 
 
@@ -926,7 +915,7 @@ class RBACMemberRolesApi(Resource):
         tenant_id, account_id = _current_ids()
         request = _payload(_ReplaceMemberRolesRequest)
         return _dump(
-            svc.RBACService.KnowledgeFSRoleMutations.replace_member_roles(
+            svc.RBACService.MemberRoles.replace(
                 tenant_id,
                 account_id,
                 str(member_id),

api/controllers/console/workspace/workspace.py

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

api/controllers/console/wraps.py

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

api/controllers/inner_api/__init__.py

@@ -20,7 +20,6 @@ from . import runtime_credentials as _runtime_credentials
 from .agent import tools as _agent_tools
 from .app import dsl as _app_dsl
 from .knowledge import retrieval as _knowledge_retrieval
-from .knowledge_fs import storage as _knowledge_fs_storage
 from .plugin import agent_config as _agent_config
 from .plugin import agent_drive as _agent_drive
 from .plugin import plugin as _plugin
@@ -33,7 +32,6 @@ __all__ = [
     "_agent_drive",
     "_agent_tools",
     "_app_dsl",
-    "_knowledge_fs_storage",
     "_knowledge_retrieval",
     "_mail",
     "_plugin",

api/controllers/inner_api/knowledge_fs/__init__.py

@@ -1 +0,0 @@
-"""Trusted KnowledgeFS inner API endpoints."""

api/controllers/inner_api/knowledge_fs/storage.py

@@ -1,303 +0,0 @@
-"""Trusted KnowledgeFS gateway to Dify's configured object-storage backend."""
-
-import json
-from base64 import b64decode
-from binascii import Error as BinasciiError
-from http import HTTPStatus
-from typing import NoReturn
-
-from flask import Response, request
-from flask_restx import Resource
-from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
-from pydantic.alias_generators import to_camel
-
-from controllers.common.schema import query_params_from_model, register_response_schema_models
-from controllers.inner_api import inner_api_ns
-from controllers.inner_api.wraps import knowledge_fs_inner_api_only
-from fields.base import ResponseModel
-from libs.exception import BaseHTTPException
-from libs.helper import dump_response
-from services.knowledge_fs.object_storage import (
-    KNOWLEDGE_FS_OBJECT_MAX_BYTES,
-    KnowledgeFSObjectList,
-    KnowledgeFSObjectMetadata,
-    KnowledgeFSObjectStorageChecksumError,
-    KnowledgeFSObjectStorageCorruptError,
-    KnowledgeFSObjectStorageError,
-    KnowledgeFSObjectStorageInvalidInputError,
-    KnowledgeFSObjectStorageService,
-    KnowledgeFSObjectStorageTooLargeError,
-    KnowledgeFSObjectStorageUnavailableError,
-)
-
-_METADATA_HEADER = "X-Knowledge-FS-Metadata"
-_CHECKSUM_HEADER = "X-Knowledge-FS-Checksum-Sha256"
-_CONTENT_TYPE_HEADER = "X-Knowledge-FS-Content-Type"
-_MAX_ENCODED_METADATA_BYTES = 128 * 1024
-_metadata_adapter = TypeAdapter(dict[str, str])
-
-
-class KnowledgeFSObjectStorageHttpError(BaseHTTPException):
-    """Safe HTTP representation of a KnowledgeFS storage boundary error."""
-
-    error_code = "knowledge_fs_object_storage_failed"
-    description = "KnowledgeFS object storage request failed."
-    code = HTTPStatus.INTERNAL_SERVER_ERROR
-
-    def __init__(self, *, error_code: str, description: str, status_code: HTTPStatus) -> None:
-        self.error_code = error_code
-        self.description = description
-        self.code = status_code
-        super().__init__(description)
-
-
-class _CamelCaseResponse(ResponseModel):
-    model_config = ConfigDict(alias_generator=to_camel)
-
-
-class KnowledgeFSObjectQuery(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    key: str = Field(description="Logical KnowledgeFS object key")
-
-
-class KnowledgeFSObjectListQuery(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    prefix: str = Field(default="", description="Logical object-key prefix")
-    cursor: str | None = Field(default=None, description="Exclusive lexical key cursor")
-    limit: int = Field(default=100, ge=1, le=100, description="Maximum objects to return")
-
-
-class KnowledgeFSObjectMetadataResponse(_CamelCaseResponse):
-    checksum_sha256_base64: str
-    content_type: str | None = None
-    key: str
-    metadata: dict[str, str]
-    size_bytes: int
-
-
-class KnowledgeFSObjectListResponse(_CamelCaseResponse):
-    objects: list[KnowledgeFSObjectMetadataResponse]
-    next_cursor: str | None = None
-
-
-class KnowledgeFSObjectHealthResponse(ResponseModel):
-    ok: bool
-
-
-register_response_schema_models(
-    inner_api_ns,
-    KnowledgeFSObjectMetadataResponse,
-    KnowledgeFSObjectListResponse,
-    KnowledgeFSObjectHealthResponse,
-)
-
-
-@inner_api_ns.route("/knowledge-fs/storage/object")
-class KnowledgeFSObjectApi(Resource):
-    """Read, write, or delete one logical KnowledgeFS object."""
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.doc(params=query_params_from_model(KnowledgeFSObjectQuery))
-    @inner_api_ns.response(
-        HTTPStatus.OK,
-        "Object stored",
-        inner_api_ns.models[KnowledgeFSObjectMetadataResponse.__name__],
-    )
-    def put(self) -> dict[str, object]:
-        try:
-            query = KnowledgeFSObjectQuery.model_validate(request.args.to_dict(flat=True))
-            metadata = _decode_metadata_header(request.headers.get(_METADATA_HEADER))
-            body = request.stream.read(KNOWLEDGE_FS_OBJECT_MAX_BYTES + 1)
-            if len(body) > KNOWLEDGE_FS_OBJECT_MAX_BYTES:
-                raise KnowledgeFSObjectStorageTooLargeError(f"object exceeds max bytes {KNOWLEDGE_FS_OBJECT_MAX_BYTES}")
-            result = KnowledgeFSObjectStorageService().put_object(
-                body=body,
-                checksum_sha256_base64=request.headers.get(_CHECKSUM_HEADER),
-                content_type=request.headers.get(_CONTENT_TYPE_HEADER),
-                key=query.key,
-                metadata=metadata,
-            )
-        except ValidationError as exc:
-            raise _invalid_request_error() from exc
-        except KnowledgeFSObjectStorageError as exc:
-            _raise_http_error(exc)
-        return _metadata_response(result)
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.doc(params=query_params_from_model(KnowledgeFSObjectQuery))
-    @inner_api_ns.produces(["application/octet-stream"])
-    def get(self) -> Response:
-        try:
-            query = KnowledgeFSObjectQuery.model_validate(request.args.to_dict(flat=True))
-            service = KnowledgeFSObjectStorageService()
-            metadata = service.head_object(key=query.key)
-            if metadata is None:
-                raise _not_found_error()
-            body = service.load_stream(key=query.key)
-            if body is None:
-                raise _not_found_error()
-        except ValidationError as exc:
-            raise _invalid_request_error() from exc
-        except KnowledgeFSObjectStorageError as exc:
-            _raise_http_error(exc)
-
-        response = Response(
-            body,
-            content_type=metadata.content_type or "application/octet-stream",
-        )
-        response.content_length = metadata.size_bytes
-        response.headers[_CHECKSUM_HEADER] = metadata.checksum_sha256_base64
-        return response
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.doc(params=query_params_from_model(KnowledgeFSObjectQuery))
-    @inner_api_ns.response(HTTPStatus.NO_CONTENT, "Object deleted")
-    def delete(self) -> tuple[str, int]:
-        try:
-            query = KnowledgeFSObjectQuery.model_validate(request.args.to_dict(flat=True))
-            KnowledgeFSObjectStorageService().delete_object(key=query.key)
-        except ValidationError as exc:
-            raise _invalid_request_error() from exc
-        except KnowledgeFSObjectStorageError as exc:
-            _raise_http_error(exc)
-        return "", HTTPStatus.NO_CONTENT
-
-
-@inner_api_ns.route("/knowledge-fs/storage/object/metadata")
-class KnowledgeFSObjectMetadataApi(Resource):
-    """Read portable metadata for one logical KnowledgeFS object."""
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.doc(params=query_params_from_model(KnowledgeFSObjectQuery))
-    @inner_api_ns.response(
-        HTTPStatus.OK,
-        "Object metadata",
-        inner_api_ns.models[KnowledgeFSObjectMetadataResponse.__name__],
-    )
-    def get(self) -> dict[str, object]:
-        try:
-            query = KnowledgeFSObjectQuery.model_validate(request.args.to_dict(flat=True))
-            result = KnowledgeFSObjectStorageService().head_object(key=query.key)
-            if result is None:
-                raise _not_found_error()
-        except ValidationError as exc:
-            raise _invalid_request_error() from exc
-        except KnowledgeFSObjectStorageError as exc:
-            _raise_http_error(exc)
-        return _metadata_response(result)
-
-
-@inner_api_ns.route("/knowledge-fs/storage/objects")
-class KnowledgeFSObjectListApi(Resource):
-    """List logical KnowledgeFS objects with bounded keyset pagination."""
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.doc(params=query_params_from_model(KnowledgeFSObjectListQuery))
-    @inner_api_ns.response(
-        HTTPStatus.OK,
-        "Object page",
-        inner_api_ns.models[KnowledgeFSObjectListResponse.__name__],
-    )
-    def get(self) -> dict[str, object]:
-        try:
-            query = KnowledgeFSObjectListQuery.model_validate(request.args.to_dict(flat=True))
-            result = KnowledgeFSObjectStorageService().list_objects(
-                cursor=query.cursor,
-                limit=query.limit,
-                prefix=query.prefix,
-            )
-        except ValidationError as exc:
-            raise _invalid_request_error() from exc
-        except KnowledgeFSObjectStorageError as exc:
-            _raise_http_error(exc)
-        return _list_response(result)
-
-
-@inner_api_ns.route("/knowledge-fs/storage/health")
-class KnowledgeFSObjectHealthApi(Resource):
-    """Report whether Dify storage satisfies KnowledgeFS portable requirements."""
-
-    @knowledge_fs_inner_api_only
-    @inner_api_ns.response(
-        HTTPStatus.OK,
-        "Storage available",
-        inner_api_ns.models[KnowledgeFSObjectHealthResponse.__name__],
-    )
-    @inner_api_ns.response(HTTPStatus.SERVICE_UNAVAILABLE, "Storage unavailable")
-    def get(self) -> dict[str, bool] | tuple[dict[str, bool], int]:
-        if KnowledgeFSObjectStorageService().health():
-            return {"ok": True}
-        return {"ok": False}, HTTPStatus.SERVICE_UNAVAILABLE
-
-
-def _decode_metadata_header(value: str | None) -> dict[str, str]:
-    if value is None:
-        return {}
-    if len(value.encode()) > _MAX_ENCODED_METADATA_BYTES:
-        raise KnowledgeFSObjectStorageInvalidInputError("object metadata header is too large")
-    try:
-        padding = "=" * (-len(value) % 4)
-        decoded = b64decode(value + padding, altchars=b"-_", validate=True)
-        return _metadata_adapter.validate_json(decoded)
-    except (BinasciiError, UnicodeEncodeError, ValidationError, json.JSONDecodeError) as exc:
-        raise KnowledgeFSObjectStorageInvalidInputError("object metadata header is invalid") from exc
-
-
-def _metadata_response(metadata: KnowledgeFSObjectMetadata) -> dict[str, object]:
-    return dump_response(KnowledgeFSObjectMetadataResponse, metadata)
-
-
-def _list_response(result: KnowledgeFSObjectList) -> dict[str, object]:
-    return dump_response(KnowledgeFSObjectListResponse, result)
-
-
-def _raise_http_error(error: KnowledgeFSObjectStorageError) -> NoReturn:
-    if isinstance(error, KnowledgeFSObjectStorageTooLargeError):
-        raise KnowledgeFSObjectStorageHttpError(
-            error_code="knowledge_fs_object_too_large",
-            description="KnowledgeFS object exceeds the configured size limit.",
-            status_code=HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
-        ) from error
-    if isinstance(error, KnowledgeFSObjectStorageChecksumError):
-        raise KnowledgeFSObjectStorageHttpError(
-            error_code="knowledge_fs_object_checksum_mismatch",
-            description="KnowledgeFS object checksum does not match the request body.",
-            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
-        ) from error
-    if isinstance(error, KnowledgeFSObjectStorageInvalidInputError):
-        raise _invalid_request_error() from error
-    if isinstance(error, KnowledgeFSObjectStorageCorruptError):
-        raise KnowledgeFSObjectStorageHttpError(
-            error_code="knowledge_fs_object_corrupt",
-            description="KnowledgeFS object metadata is inconsistent.",
-            status_code=HTTPStatus.BAD_GATEWAY,
-        ) from error
-    if isinstance(error, KnowledgeFSObjectStorageUnavailableError):
-        raise KnowledgeFSObjectStorageHttpError(
-            error_code="knowledge_fs_object_storage_unavailable",
-            description="Dify object storage is unavailable for KnowledgeFS.",
-            status_code=HTTPStatus.SERVICE_UNAVAILABLE,
-        ) from error
-    raise KnowledgeFSObjectStorageHttpError(
-        error_code="knowledge_fs_object_storage_failed",
-        description="KnowledgeFS object storage request failed.",
-        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-    ) from error
-
-
-def _invalid_request_error() -> KnowledgeFSObjectStorageHttpError:
-    return KnowledgeFSObjectStorageHttpError(
-        error_code="knowledge_fs_object_storage_invalid_request",
-        description="KnowledgeFS object storage request is invalid.",
-        status_code=HTTPStatus.BAD_REQUEST,
-    )
-
-
-def _not_found_error() -> KnowledgeFSObjectStorageHttpError:
-    return KnowledgeFSObjectStorageHttpError(
-        error_code="knowledge_fs_object_not_found",
-        description="KnowledgeFS object was not found.",
-        status_code=HTTPStatus.NOT_FOUND,
-    )

api/controllers/inner_api/plugin/plugin.py

@@ -8,7 +8,6 @@ from controllers.inner_api.plugin.wraps import get_user_tenant, plugin_data
 from controllers.inner_api.wraps import plugin_inner_api_only
 from core.plugin.backwards_invocation.app import PluginAppBackwardsInvocation
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocationResponse
-from core.plugin.backwards_invocation.datasource import PluginDatasourceBackwardsInvocation
 from core.plugin.backwards_invocation.encrypt import PluginEncrypter
 from core.plugin.backwards_invocation.model import PluginModelBackwardsInvocation
 from core.plugin.backwards_invocation.node import PluginNodeBackwardsInvocation
@@ -16,12 +15,10 @@ from core.plugin.backwards_invocation.tool import PluginToolBackwardsInvocation
 from core.plugin.entities.request import (
     RequestFetchAppInfo,
     RequestInvokeApp,
-    RequestInvokeDatasource,
     RequestInvokeEncrypt,
     RequestInvokeLLM,
     RequestInvokeLLMWithStructuredOutput,
     RequestInvokeModeration,
-    RequestInvokeMultimodalEmbedding,
     RequestInvokeParameterExtractorNode,
     RequestInvokeQuestionClassifierNode,
     RequestInvokeRerank,
@@ -30,7 +27,6 @@ from core.plugin.entities.request import (
     RequestInvokeTextEmbedding,
     RequestInvokeTool,
     RequestInvokeTTS,
-    RequestListModels,
     RequestRequestDownloadFile,
     RequestRequestUploadFile,
 )
@@ -122,36 +118,6 @@ class PluginInvokeTextEmbeddingApi(Resource):
             return jsonable_encoder(BaseBackwardsInvocationResponse(error=str(e)))
 
 
-@inner_api_ns.route("/invoke/multimodal-embedding")
-class PluginInvokeMultimodalEmbeddingApi(Resource):
-    @get_user_tenant
-    @setup_required
-    @plugin_inner_api_only
-    @plugin_data(payload_type=RequestInvokeMultimodalEmbedding)
-    @inner_api_ns.doc("plugin_invoke_multimodal_embedding")
-    @inner_api_ns.doc(description="Invoke multimodal embedding models through Dify model management")
-    @inner_api_ns.doc(
-        responses={
-            200: "Multimodal embedding successful",
-            401: "Unauthorized - invalid API key",
-            404: "Service not available",
-        }
-    )
-    def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestInvokeMultimodalEmbedding):
-        try:
-            return jsonable_encoder(
-                BaseBackwardsInvocationResponse(
-                    data=PluginModelBackwardsInvocation.invoke_multimodal_embedding(
-                        user_id=user_model.id,
-                        tenant=tenant_model,
-                        payload=payload,
-                    )
-                )
-            )
-        except Exception as e:
-            return jsonable_encoder(BaseBackwardsInvocationResponse(error=str(e)))
-
-
 @inner_api_ns.route("/invoke/rerank")
 class PluginInvokeRerankApi(Resource):
     @get_user_tenant
@@ -178,70 +144,6 @@ class PluginInvokeRerankApi(Resource):
             return jsonable_encoder(BaseBackwardsInvocationResponse(error=str(e)))
 
 
-@inner_api_ns.route("/invoke/model-catalog")
-class PluginModelCatalogApi(Resource):
-    @get_user_tenant
-    @setup_required
-    @plugin_inner_api_only
-    @plugin_data(payload_type=RequestListModels)
-    @inner_api_ns.doc("plugin_model_catalog")
-    @inner_api_ns.doc(description="List tenant-active models managed by Dify")
-    @inner_api_ns.doc(
-        responses={
-            200: "Model catalog lookup successful",
-            401: "Unauthorized - invalid API key",
-            404: "Service not available",
-        }
-    )
-    def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestListModels):
-        try:
-            return jsonable_encoder(
-                BaseBackwardsInvocationResponse(
-                    data=PluginModelBackwardsInvocation.list_models(
-                        tenant_id=tenant_model.id,
-                        user_id=user_model.id,
-                        payload=payload,
-                    )
-                )
-            )
-        except Exception as e:
-            return jsonable_encoder(BaseBackwardsInvocationResponse(error=str(e)))
-
-
-@inner_api_ns.route("/invoke/datasource")
-class PluginInvokeDatasourceApi(Resource):
-    """Invoke an installed datasource with credentials resolved inside Dify."""
-
-    @get_user_tenant
-    @setup_required
-    @plugin_inner_api_only
-    @plugin_data(payload_type=RequestInvokeDatasource)
-    @inner_api_ns.doc("plugin_invoke_datasource")
-    @inner_api_ns.doc(description="Invoke datasource plugins through Dify credential management")
-    @inner_api_ns.doc(
-        responses={
-            200: "Datasource invocation successful (streaming response)",
-            401: "Unauthorized - invalid API key",
-            404: "Datasource provider, datasource, or credential not found",
-        }
-    )
-    def post(
-        self,
-        user_model: Account | EndUser,
-        tenant_model: Tenant,
-        payload: RequestInvokeDatasource,
-    ):
-        response = PluginDatasourceBackwardsInvocation.invoke(
-            user_id=user_model.id,
-            tenant=tenant_model,
-            payload=payload,
-        )
-        return length_prefixed_response(
-            0xF,
-            PluginDatasourceBackwardsInvocation.convert_to_event_stream(response),
-        )
-
-
 @inner_api_ns.route("/invoke/tts")
 class PluginInvokeTTSApi(Resource):
     @get_user_tenant

api/controllers/inner_api/plugin/wraps.py

@@ -1,6 +1,5 @@
 from collections.abc import Callable
 from functools import wraps
-from uuid import UUID
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -56,15 +55,14 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
                 # session_id, id is auto-generated) and a fresh EndUser
                 # was created per call, breaking multi-turn chat
                 # continuation (see #36736).
-                if _is_uuid(user_id):
-                    user_model = session.scalar(
-                        select(EndUser)
-                        .where(
-                            EndUser.id == user_id,
-                            EndUser.tenant_id == tenant_id,
-                        )
-                        .limit(1)
+                user_model = session.scalar(
+                    select(EndUser)
+                    .where(
+                        EndUser.id == user_id,
+                        EndUser.tenant_id == tenant_id,
                     )
+                    .limit(1)
+                )
                 if user_model is None:
                     user_model = session.scalar(
                         select(EndUser)
@@ -92,14 +90,6 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     return user_model
 
 
-def _is_uuid(value: str) -> bool:
-    try:
-        UUID(value)
-    except ValueError:
-        return False
-    return True
-
-
 def get_user_tenant[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
     @wraps(view_func)
     def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:

api/controllers/inner_api/wraps.py

@@ -107,9 +107,3 @@ def agent_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     """
 
     return plugin_inner_api_only(view)
-
-
-def knowledge_fs_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
-    """Authenticate the trusted KnowledgeFS process on the shared inner bridge."""
-
-    return plugin_inner_api_only(view)

api/controllers/openapi/auth/data.py

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

api/controllers/service_api/__init__.py

@@ -38,7 +38,6 @@ from .dataset import (
 )
 from .dataset.rag_pipeline import rag_pipeline_workflow
 from .end_user import end_user
-from .knowledge_fs import resources as knowledge_fs_resources
 from .workspace import models
 
 __all__ = [
@@ -55,7 +54,6 @@ __all__ = [
     "hit_testing",
     "human_input_form",
     "index",
-    "knowledge_fs_resources",
     "message",
     "metadata",
     "models",

api/controllers/service_api/app/completion.py

@@ -26,6 +26,10 @@ from controllers.service_api.app.error import (
     ProviderQuotaExceededError,
     WorkflowVersionExecutionNotAllowedError,
 )
+from controllers.service_api.app.legacy_system_files import (
+    attach_legacy_system_file_warning_for_service_api,
+    normalize_legacy_system_file_args_for_service_api,
+)
 from controllers.service_api.schema import (
     InputFileList,
     expect_user_json,
@@ -390,6 +394,7 @@ class ChatApi(Resource):
             args["external_trace_id"] = external_trace_id
 
         streaming = _resolve_agent_app_streaming(app_mode=app_mode, response_mode=payload.response_mode)
+        legacy_system_file_compat = None
 
         try:
             # Eagerly validate conversation to avoid hanging on invalid conversation_id
@@ -401,6 +406,14 @@ class ChatApi(Resource):
                     session=session,
                 )
 
+            if app_mode == AppMode.ADVANCED_CHAT:
+                args, legacy_system_file_compat = normalize_legacy_system_file_args_for_service_api(
+                    session=session,
+                    app_model=app_model,
+                    args=args,
+                    raw_payload=service_api_ns.payload,
+                    workflow_id=args.get("workflow_id"),
+                )
             response = AppGenerateService.generate(
                 session=session,
                 app_model=app_model,
@@ -409,6 +422,7 @@ class ChatApi(Resource):
                 invoke_from=InvokeFrom.SERVICE_API,
                 streaming=streaming,
             )
+            response = attach_legacy_system_file_warning_for_service_api(response, legacy_system_file_compat)
 
             # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)

api/controllers/service_api/app/legacy_system_files.py

@@ -0,0 +1,72 @@
+"""Temporary Service API adapter for the deprecated workflow file input."""
+
+from collections.abc import Generator, Mapping
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
+from core.workflow.legacy_system_files import (
+    LegacySysFilesCompatVariable,
+    attach_legacy_sys_files_warning,
+    normalize_legacy_sys_files_args,
+)
+from models.model import App
+from services.app_generate_service import AppGenerateService
+
+type ServiceAPIGenerateResponse = Mapping[str, Any] | Generator[str, None, None] | RateLimitGenerator
+
+
+def normalize_legacy_system_file_args_for_service_api(
+    *,
+    session: Session,
+    app_model: App,
+    args: dict[str, Any],
+    raw_payload: Mapping[str, Any] | None,
+    workflow_id: str | None = None,
+) -> tuple[dict[str, Any], LegacySysFilesCompatVariable | None]:
+    # TODO: Remove this hidden Service API compatibility path after all persisted workflows are migrated.
+    args_with_hidden_system = _copy_hidden_system_files_arg(args=args, raw_payload=raw_payload)
+    if not _has_legacy_file_arg(args_with_hidden_system):
+        return args, None
+
+    workflow = AppGenerateService.get_workflow(
+        app_model,
+        InvokeFrom.SERVICE_API,
+        workflow_id,
+        session=session,
+    )
+    return normalize_legacy_sys_files_args(graph=workflow.graph_dict, args=args_with_hidden_system)
+
+
+def attach_legacy_system_file_warning_for_service_api(
+    response: ServiceAPIGenerateResponse,
+    compat_variable: LegacySysFilesCompatVariable | None,
+) -> ServiceAPIGenerateResponse:
+    # TODO: Remove this warning once Service API clients no longer need the legacy migration notice.
+    if compat_variable is None:
+        return response
+    return attach_legacy_sys_files_warning(response, compat_variable)
+
+
+def _copy_hidden_system_files_arg(
+    *,
+    args: dict[str, Any],
+    raw_payload: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    system = raw_payload.get("system") if isinstance(raw_payload, Mapping) else None
+    if not isinstance(system, Mapping) or "files" not in system or system["files"] is None:
+        return args
+
+    copied_args = dict(args)
+    copied_args["system"] = {"files": system["files"]}
+    return copied_args
+
+
+def _has_legacy_file_arg(args: Mapping[str, Any]) -> bool:
+    if args.get("files") is not None:
+        return True
+
+    system = args.get("system")
+    return isinstance(system, Mapping) and system.get("files") is not None

api/controllers/service_api/app/workflow.py

@@ -30,6 +30,10 @@ from controllers.service_api.app.error import (
     ProviderQuotaExceededError,
     WorkflowVersionExecutionNotAllowedError,
 )
+from controllers.service_api.app.legacy_system_files import (
+    attach_legacy_system_file_warning_for_service_api,
+    normalize_legacy_system_file_args_for_service_api,
+)
 from controllers.service_api.schema import (
     expect_user_json,
     expect_with_user,
@@ -344,6 +348,12 @@ class WorkflowRunApi(Resource):
         streaming = payload.response_mode == "streaming"
 
         try:
+            args, legacy_system_file_compat = normalize_legacy_system_file_args_for_service_api(
+                session=session,
+                app_model=app_model,
+                args=args,
+                raw_payload=service_api_ns.payload,
+            )
             response = AppGenerateService.generate(
                 session=session,
                 app_model=app_model,
@@ -352,6 +362,7 @@ class WorkflowRunApi(Resource):
                 invoke_from=InvokeFrom.SERVICE_API,
                 streaming=streaming,
             )
+            response = attach_legacy_system_file_warning_for_service_api(response, legacy_system_file_compat)
 
             # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)
@@ -471,6 +482,13 @@ class WorkflowRunByIdApi(Resource):
         streaming = payload.response_mode == "streaming"
 
         try:
+            args, legacy_system_file_compat = normalize_legacy_system_file_args_for_service_api(
+                session=session,
+                app_model=app_model,
+                args=args,
+                raw_payload=service_api_ns.payload,
+                workflow_id=workflow_id,
+            )
             response = AppGenerateService.generate(
                 session=session,
                 app_model=app_model,
@@ -479,6 +497,7 @@ class WorkflowRunByIdApi(Resource):
                 invoke_from=InvokeFrom.SERVICE_API,
                 streaming=streaming,
             )
+            response = attach_legacy_system_file_warning_for_service_api(response, legacy_system_file_compat)
 
             # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)

api/controllers/service_api/knowledge_fs/__init__.py

@@ -1,5 +0,0 @@
-"""KnowledgeFS-specific Service API authenticated by resource credentials."""
-
-from . import resources
-
-__all__ = ["resources"]

api/controllers/service_api/knowledge_fs/error.py

@@ -1,49 +0,0 @@
-"""Stable KnowledgeFS Service API error contract."""
-
-from libs.exception import BaseHTTPException
-
-
-class KnowledgeFSInvalidCredentialHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_invalid_credential"
-    description = "Invalid KnowledgeFS service credential."
-    code = 401
-
-
-class KnowledgeFSServiceOperationUnavailableHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_operation_unavailable"
-    description = "KnowledgeFS operation is not available."
-    code = 503
-
-
-class KnowledgeFSServiceUpstreamUnavailableHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_upstream_unavailable"
-    description = "KnowledgeFS is unavailable."
-    code = 502
-
-
-class KnowledgeFSServiceInvalidRequestHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_invalid_request"
-    description = "KnowledgeFS request is invalid."
-    code = 400
-
-
-class KnowledgeFSServiceRateLimitHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_rate_limit_exceeded"
-    description = "KnowledgeFS operation rate limit exceeded."
-    code = 429
-
-
-class KnowledgeFSServiceQuotaExceededHTTPError(BaseHTTPException):
-    error_code = "knowledge_fs_quota_exceeded"
-    description = "KnowledgeFS operation quota exceeded."
-    code = 403
-
-
-__all__ = [
-    "KnowledgeFSInvalidCredentialHTTPError",
-    "KnowledgeFSServiceInvalidRequestHTTPError",
-    "KnowledgeFSServiceOperationUnavailableHTTPError",
-    "KnowledgeFSServiceQuotaExceededHTTPError",
-    "KnowledgeFSServiceRateLimitHTTPError",
-    "KnowledgeFSServiceUpstreamUnavailableHTTPError",
-]

api/controllers/web/wraps.py

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

api/core/agent/base_agent_runner.py

@@ -14,10 +14,7 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.base_app_runner import AppRunner
 from core.app.entities.app_invoke_entities import (
     AgentChatAppGenerateEntity,
-    DifyRunContext,
-    InvokeFrom,
     ModelConfigWithCredentialsEntity,
-    UserFrom,
 )
 from core.app.file_access import DatabaseFileAccessController
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
@@ -151,15 +148,6 @@ class BaseAgentRunner(AppRunner):
             user_id=self.user_id,
             invoke_from=self.application_generate_entity.invoke_from,
         )
-        invoke_from = self.application_generate_entity.invoke_from
-        if isinstance(invoke_from, InvokeFrom):
-            tool_entity.runtime.dify_run_context = DifyRunContext(
-                tenant_id=self.tenant_id,
-                app_id=self.app_config.app_id,
-                user_id=self.user_id,
-                user_from=(UserFrom.ACCOUNT if invoke_from.runs_as_account() else UserFrom.END_USER),
-                invoke_from=invoke_from,
-            )
         assert tool_entity.entity.description
         message_tool = PromptMessageTool(
             name=tool.tool_name,

api/core/agent/publish_visibility.py

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

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

@@ -46,6 +46,7 @@ from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.legacy_system_files import normalize_legacy_sys_files_args
 from extensions.ext_database import db
 from factories import file_factory
 from graphon.filters import ResponseStreamFilter
@@ -147,6 +148,8 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         if not args.get("query"):
             raise ValueError("query is required")
 
+        # TODO: Remove this compatibility normalization after all persisted workflows are migrated.
+        args, _ = normalize_legacy_sys_files_args(graph=workflow.graph_dict, args=args)
         query = args["query"]
         if not isinstance(query, str):
             raise ValueError("query must be a string")

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

@@ -683,42 +683,27 @@ class AgentAppGenerator(MessageBasedAppGenerator):
             if draft_type == AgentConfigDraftType.DEBUG_BUILD.value
             else AgentConfigDraftType.DRAFT
         )
+        if effective_draft_type == AgentConfigDraftType.DRAFT:
+            from services.agent.composer_service import AgentComposerService
+
+            return AgentComposerService.get_or_create_normal_agent_draft(
+                session=session,
+                tenant_id=tenant_id,
+                agent=agent,
+                created_by=agent.updated_by or agent.created_by,
+            )
+        if not account_id:
+            raise AgentAppGeneratorError("Build draft requires an account user")
         stmt = select(AgentConfigDraft).where(
             AgentConfigDraft.tenant_id == tenant_id,
             AgentConfigDraft.agent_id == agent.id,
-            AgentConfigDraft.draft_type == effective_draft_type,
+            AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
+            AgentConfigDraft.account_id == account_id,
         )
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            if not account_id:
-                raise AgentAppGeneratorError("Build draft requires an account user")
-            stmt = stmt.where(AgentConfigDraft.account_id == account_id)
-        else:
-            stmt = stmt.where(AgentConfigDraft.account_id.is_(None))
         draft = session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))
         if draft is not None:
             return draft
-        if effective_draft_type == AgentConfigDraftType.DEBUG_BUILD:
-            raise AgentAppGeneratorError("Agent build draft not found")
-        _, snapshot, agent_soul = AgentAppGenerator._resolve_agent_by_id(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
-            session=session,
-        )
-        draft = AgentConfigDraft(
-            tenant_id=tenant_id,
-            agent_id=agent.id,
-            draft_type=AgentConfigDraftType.DRAFT,
-            account_id=None,
-            draft_owner_key="",
-            base_snapshot_id=snapshot.id,
-            config_snapshot=agent_soul,
-            created_by=agent.created_by,
-            updated_by=agent.updated_by,
-        )
-        session.add(draft)
-        session.flush()
-        return draft
+        raise AgentAppGeneratorError("Agent build draft not found")
 
     @staticmethod
     def _resolve_agent_by_id(

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

@@ -159,7 +159,6 @@ class AgentAppRuntimeRequestBuilder:
                     user_from=cast(DifyExecutionContextUserFrom, context.dify_context.user_from.value),
                     invoke_from=cast(DifyExecutionContextInvokeFrom, context.dify_context.invoke_from.value),
                     agent_mode="agent_app",
-                    trace_id=context.dify_context.trace_session_id,
                 ),
                 # ENG-616: expand slash-menu mention tokens to canonical names so
                 # no frontend-internal {{#…#}} marker ever reaches the model.

api/core/app/apps/common/workflow_response_converter.py

@@ -155,8 +155,10 @@ class WorkflowResponseConverter:
             # TODO(@future-refactor): store system variables separately from user inputs so we don't
             # need to flatten `sys.*` entries into the input payload just for rerun/export tooling.
             if field_name == SystemVariableKey.CONVERSATION_ID:
-                # Conversation IDs are session-scoped; omitting them keeps workflow inputs
-                # reusable without pinning new runs to a prior conversation.
+                # Conversation IDs are session-scoped; omitting them keeps workflow inputs reusable.
+                continue
+            if field_name == SystemVariableKey.FILES:
+                # When files are exposed as an input, application inputs use the canonical `userinput.files` key.
                 continue
             inputs[f"sys.{field_name}"] = value
         handled = WorkflowEntry.handle_special_values(inputs)

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

@@ -41,6 +41,7 @@ from core.helper.trace_id_helper import (
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import DifyCoreRepositoryFactory
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.legacy_system_files import normalize_legacy_sys_files_args
 from extensions.ext_database import db
 from factories import file_factory
 from graphon.filters import ResponseStreamFilter
@@ -164,6 +165,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         pause_state_config: PauseStateLayerConfig | None = None,
     ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]:
         with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
+            # TODO: Remove this compatibility normalization after all persisted workflows are migrated.
+            args, _ = normalize_legacy_sys_files_args(graph=workflow.graph_dict, args=args)
             files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
             # parse files

api/core/datasource/__base/datasource_provider.py

@@ -38,11 +38,6 @@ class DatasourcePluginProviderController(ABC):
         ):
             raise ToolProviderCredentialValidationError("Invalid credentials")
 
-    def validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> None:
-        """Validate credential shape and value against this installed provider declaration."""
-        self.validate_credentials_format(credentials)
-        self._validate_credentials(user_id, credentials)
-
     @property
     def provider_type(self) -> DatasourceProviderType:
         """

api/core/hosting_configuration.py

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

api/core/indexing_runner.py

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

api/core/logging/context.py

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

api/core/logging/filters.py

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

api/core/plugin/backwards_invocation/datasource.py

@@ -1,110 +0,0 @@
-"""Datasource backward invocation through Dify's tenant-bound runtime.
-
-This module is the internal service boundary used by trusted callers such as
-KnowledgeFS. It resolves installed provider declarations and Dify-owned
-credential references before reaching ``PluginDatasourceManager``; callers
-must never provide raw datasource credentials or a plugin-daemon API key.
-"""
-
-from collections.abc import Generator
-from typing import Any, cast
-
-from pydantic import BaseModel
-
-from core.datasource.datasource_manager import DatasourceManager
-from core.datasource.entities.datasource_entities import (
-    OnlineDriveBrowseFilesRequest,
-    OnlineDriveDownloadFileRequest,
-)
-from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
-from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
-from core.datasource.website_crawl.website_crawl_plugin import WebsiteCrawlDatasourcePlugin
-from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
-from core.plugin.entities.request import RequestInvokeDatasource
-from models.account import Tenant
-from models.provider_ids import DatasourceProviderID
-from services.datasource_provider_service import DatasourceProviderService
-
-
-class PluginDatasourceBackwardsInvocation(BaseBackwardsInvocation):
-    """Resolve and invoke a datasource without exposing credential material to the caller."""
-
-    @classmethod
-    def invoke(
-        cls,
-        *,
-        user_id: str,
-        tenant: Tenant,
-        payload: RequestInvokeDatasource,
-    ) -> Generator[BaseModel | dict[str, Any], None, None]:
-        """Yield datasource messages for one validated inner-runtime request."""
-        provider_id = DatasourceProviderID(payload.provider)
-        canonical_provider_id = str(provider_id)
-        controller = DatasourceManager.get_datasource_plugin_provider(
-            provider_id=canonical_provider_id,
-            tenant_id=tenant.id,
-            datasource_type=payload.datasource_type,
-        )
-        if controller.entity.provider_type != payload.datasource_type:
-            raise ValueError("Datasource provider type mismatch")
-
-        # Resolving the datasource from the installed declaration prevents a caller
-        # from dispatching an arbitrary datasource name under a valid plugin ID.
-        runtime = controller.get_datasource(payload.datasource)
-        credentials = DatasourceProviderService().get_datasource_credentials(
-            tenant_id=tenant.id,
-            provider=provider_id.provider_name,
-            plugin_id=provider_id.plugin_id,
-            credential_id=payload.credential_id,
-        )
-        if controller.need_credentials and not credentials:
-            raise ValueError("Datasource credential not found")
-
-        if payload.operation == "validate_credentials":
-            controller.validate_credentials(user_id=user_id, credentials=credentials)
-            yield {"result": True}
-            return
-
-        runtime.runtime.credentials = credentials
-        provider_type = runtime.datasource_provider_type()
-
-        match payload.operation:
-            case "get_website_crawl":
-                website = cast(WebsiteCrawlDatasourcePlugin, runtime)
-                yield from website.get_website_crawl(
-                    user_id=user_id,
-                    datasource_parameters=payload.datasource_parameters,
-                    provider_type=provider_type,
-                )
-            case "get_online_document_pages":
-                document = cast(OnlineDocumentDatasourcePlugin, runtime)
-                yield from document.get_online_document_pages(
-                    user_id=user_id,
-                    datasource_parameters=payload.datasource_parameters,
-                    provider_type=provider_type,
-                )
-            case "get_online_document_page_content":
-                if payload.page is None:
-                    raise ValueError("Online-document page input is required")
-                document = cast(OnlineDocumentDatasourcePlugin, runtime)
-                yield from document.get_online_document_page_content(
-                    user_id=user_id,
-                    datasource_parameters=payload.page,
-                    provider_type=provider_type,
-                )
-            case "online_drive_browse_files":
-                drive = cast(OnlineDriveDatasourcePlugin, runtime)
-                yield from drive.online_drive_browse_files(
-                    user_id=user_id,
-                    request=OnlineDriveBrowseFilesRequest.model_validate(payload.request),
-                    provider_type=provider_type,
-                )
-            case "online_drive_download_file":
-                drive = cast(OnlineDriveDatasourcePlugin, runtime)
-                yield from drive.online_drive_download_file(
-                    user_id=user_id,
-                    request=OnlineDriveDownloadFileRequest.model_validate(payload.request),
-                    provider_type=provider_type,
-                )
-            case _:
-                raise ValueError(f"Unsupported datasource operation: {payload.operation}")

api/core/plugin/backwards_invocation/model.py

@@ -1,31 +1,22 @@
 import tempfile
 from binascii import hexlify, unhexlify
-from collections.abc import Generator, Mapping
-from enum import Enum
+from collections.abc import Generator
 from typing import Any
 
-from pydantic import BaseModel
-
 from core.app.llm import deduct_llm_quota
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
 from core.model_manager import ModelManager
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
 from core.plugin.entities.request import (
-    InvokableModelCatalogItem,
-    InvokableModelCatalogPage,
     RequestInvokeLLM,
     RequestInvokeLLMWithStructuredOutput,
     RequestInvokeModeration,
-    RequestInvokeMultimodalEmbedding,
     RequestInvokeRerank,
     RequestInvokeSpeech2Text,
     RequestInvokeSummary,
     RequestInvokeTextEmbedding,
     RequestInvokeTTS,
-    RequestListModels,
 )
-from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
-from core.plugin.plugin_service import PluginService
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.utils.model_invocation_utils import ModelInvocationUtils
 from graphon.model_runtime.entities.llm_entities import (
@@ -42,20 +33,6 @@ from graphon.model_runtime.entities.message_entities import (
 )
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.account import Tenant
-from models.provider_ids import ModelProviderID
-
-
-def _json_compatible(value: Any) -> Any:
-    """Convert model-runtime metadata into stable JSON-compatible values."""
-    if isinstance(value, BaseModel):
-        return value.model_dump(mode="json")
-    if isinstance(value, Enum):
-        return value.value
-    if isinstance(value, Mapping):
-        return {str(_json_compatible(key)): _json_compatible(child) for key, child in value.items()}
-    if isinstance(value, list | tuple | set):
-        return [_json_compatible(child) for child in value]
-    return value
 
 
 class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
@@ -206,30 +183,7 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
         )
 
         # invoke model
-        response = model_instance.invoke_text_embedding(texts=payload.texts, input_type=payload.input_type)
-
-        return response
-
-    @classmethod
-    def invoke_multimodal_embedding(
-        cls,
-        user_id: str,
-        tenant: Tenant,
-        payload: RequestInvokeMultimodalEmbedding,
-    ):
-        """Invoke multimodal embedding through the tenant-bound model instance."""
-        model_instance = cls._get_bound_model_instance(
-            tenant_id=tenant.id,
-            user_id=user_id,
-            provider=payload.provider,
-            model_type=payload.model_type,
-            model=payload.model,
-        )
-
-        response = model_instance.invoke_multimodal_embedding(
-            multimodel_documents=[document.model_dump(exclude_none=True) for document in payload.documents],
-            input_type=payload.input_type,
-        )
+        response = model_instance.invoke_text_embedding(texts=payload.texts)
 
         return response
 
@@ -256,67 +210,6 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
 
         return response
 
-    @classmethod
-    def list_models(
-        cls,
-        tenant_id: str,
-        user_id: str,
-        payload: RequestListModels,
-    ) -> InvokableModelCatalogPage:
-        """List only models that are active for the tenant's Dify configuration."""
-        provider_manager = create_plugin_provider_manager(tenant_id=tenant_id, user_id=user_id)
-        active_models = provider_manager.get_configurations(tenant_id).get_models(
-            model_type=payload.model_type,
-            only_active=True,
-        )
-
-        installed_identities: dict[str, str] = {}
-        for plugin in PluginService.list(tenant_id):
-            existing = installed_identities.get(plugin.plugin_id)
-            if existing is not None and existing != plugin.plugin_unique_identifier:
-                raise ValueError(f"Ambiguous installed identity for model plugin {plugin.plugin_id}")
-            installed_identities[plugin.plugin_id] = plugin.plugin_unique_identifier
-
-        requested_provider = str(ModelProviderID(payload.provider)) if payload.provider else None
-        matched_models = [
-            model
-            for model in active_models
-            if (requested_provider is None or model.provider.provider == requested_provider)
-            and (payload.model is None or model.model == payload.model)
-        ]
-        matched_models.sort(key=lambda model: (model.provider.provider, model.model))
-
-        page_models = matched_models[payload.offset : payload.offset + payload.limit]
-        items: list[InvokableModelCatalogItem] = []
-        for model in page_models:
-            provider_id = ModelProviderID(model.provider.provider)
-            unique_identifier = installed_identities.get(provider_id.plugin_id)
-            if unique_identifier is None:
-                raise ValueError(f"Installed identity not found for active model plugin {provider_id.plugin_id}")
-            items.append(
-                InvokableModelCatalogItem(
-                    plugin_id=provider_id.plugin_id,
-                    plugin_unique_identifier=unique_identifier,
-                    provider=provider_id.provider_name,
-                    model=model.model,
-                    model_type=model.model_type,
-                    capabilities={
-                        "deprecated": model.deprecated,
-                        "features": _json_compatible(model.features or []),
-                        "fetchFrom": _json_compatible(model.fetch_from),
-                        "modelProperties": _json_compatible(model.model_properties),
-                        "modelType": model.model_type.value,
-                        "status": _json_compatible(model.status),
-                    },
-                )
-            )
-
-        next_offset = payload.offset + len(page_models)
-        return InvokableModelCatalogPage(
-            items=items,
-            next_offset=next_offset if next_offset < len(matched_models) else None,
-        )
-
     @classmethod
     def invoke_tts(cls, user_id: str, tenant: Tenant, payload: RequestInvokeTTS):
         """

api/core/plugin/entities/request.py

@@ -6,11 +6,6 @@ from typing import Any, Literal
 from flask import Response
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
-from core.datasource.entities.datasource_entities import (
-    DatasourceProviderType,
-    GetOnlineDocumentPageContentRequest,
-)
-from core.entities.embedding_type import EmbeddingInputType
 from core.entities.provider_entities import BasicProviderConfig
 from core.plugin.utils.http_parser import deserialize_response
 from core.workflow.file_reference import is_canonical_file_reference
@@ -59,61 +54,6 @@ class RequestInvokeTool(BaseModel):
     credential_id: str | None = None
 
 
-DatasourceInvocationOperation = Literal[
-    "get_online_document_page_content",
-    "get_online_document_pages",
-    "get_website_crawl",
-    "online_drive_browse_files",
-    "online_drive_download_file",
-    "validate_credentials",
-]
-
-
-class RequestInvokeDatasource(BaseModel):
-    """Invoke one installed datasource using a Dify-owned credential reference.
-
-    Raw credentials are intentionally not part of this contract. ``tenant_id`` and
-    ``user_id`` are consumed by the inner-API request context, while the remaining
-    fields select an installed provider declaration and an operation-specific input.
-    """
-
-    tenant_id: str = Field(min_length=1, max_length=512)
-    user_id: str = Field(min_length=1, max_length=512)
-    provider: str = Field(min_length=1, max_length=768)
-    datasource: str = Field(min_length=1, max_length=256)
-    datasource_type: DatasourceProviderType
-    credential_id: str = Field(min_length=1, max_length=512)
-    operation: DatasourceInvocationOperation
-    datasource_parameters: dict[str, Any] = Field(default_factory=dict)
-    page: GetOnlineDocumentPageContentRequest | None = None
-    request: dict[str, Any] | None = None
-
-    model_config = ConfigDict(extra="forbid")
-
-    @model_validator(mode="after")
-    def validate_operation_payload(self) -> "RequestInvokeDatasource":
-        expected_type = {
-            "get_online_document_page_content": DatasourceProviderType.ONLINE_DOCUMENT,
-            "get_online_document_pages": DatasourceProviderType.ONLINE_DOCUMENT,
-            "get_website_crawl": DatasourceProviderType.WEBSITE_CRAWL,
-            "online_drive_browse_files": DatasourceProviderType.ONLINE_DRIVE,
-            "online_drive_download_file": DatasourceProviderType.ONLINE_DRIVE,
-            "validate_credentials": self.datasource_type,
-        }[self.operation]
-        if self.datasource_type != expected_type:
-            raise ValueError(f"{self.operation} requires datasource_type {expected_type.value}")
-
-        page_required = self.operation == "get_online_document_page_content"
-        if page_required != (self.page is not None):
-            raise ValueError("page is required only for get_online_document_page_content")
-
-        request_required = self.operation in {"online_drive_browse_files", "online_drive_download_file"}
-        if request_required != (self.request is not None):
-            raise ValueError("request is required only for online-drive operations")
-
-        return self
-
-
 class BaseRequestInvokeModel(BaseModel):
     provider: str
     model: str
@@ -175,25 +115,6 @@ class RequestInvokeTextEmbedding(BaseRequestInvokeModel):
 
     model_type: ModelType = ModelType.TEXT_EMBEDDING
     texts: list[str]
-    input_type: EmbeddingInputType = EmbeddingInputType.DOCUMENT
-
-
-class MultimodalEmbeddingDocument(BaseModel):
-    """A document accepted by a multimodal text-embedding model."""
-
-    content: str
-    content_type: str
-    file_id: str | None = None
-
-    model_config = ConfigDict(extra="forbid")
-
-
-class RequestInvokeMultimodalEmbedding(BaseRequestInvokeModel):
-    """Request to invoke a multimodal text-embedding model."""
-
-    model_type: ModelType = ModelType.TEXT_EMBEDDING
-    documents: list[MultimodalEmbeddingDocument] = Field(min_length=1)
-    input_type: EmbeddingInputType = EmbeddingInputType.DOCUMENT
 
 
 class RequestInvokeRerank(BaseRequestInvokeModel):
@@ -204,40 +125,8 @@ class RequestInvokeRerank(BaseRequestInvokeModel):
     model_type: ModelType = ModelType.RERANK
     query: str
     docs: list[str]
-    score_threshold: float | None = None
-    top_n: int | None = None
-
-
-class RequestListModels(BaseModel):
-    """Tenant-scoped query for models that Dify can invoke."""
-
-    model_type: Literal[ModelType.LLM, ModelType.TEXT_EMBEDDING, ModelType.RERANK]
-    provider: str | None = None
-    model: str | None = None
-    offset: int = Field(default=0, ge=0)
-    limit: int = Field(default=50, ge=1, le=100)
-
-    model_config = ConfigDict(protected_namespaces=())
-
-
-class InvokableModelCatalogItem(BaseModel):
-    """Installed identity and active Dify capability metadata for one model."""
-
-    plugin_id: str
-    plugin_unique_identifier: str
-    provider: str
-    model: str
-    model_type: ModelType
-    capabilities: dict[str, Any] = Field(default_factory=dict)
-
-    model_config = ConfigDict(protected_namespaces=())
-
-
-class InvokableModelCatalogPage(BaseModel):
-    """Offset page returned by the internal model catalog endpoint."""
-
-    items: list[InvokableModelCatalogItem] = Field(default_factory=list)
-    next_offset: int | None = None
+    score_threshold: float
+    top_n: int
 
 
 class RequestInvokeTTS(BaseRequestInvokeModel):

api/core/plugin/impl/base.py

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

api/core/plugin/impl/exc.py

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

api/core/provider_manager.py

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

api/core/rag/data_post_processor/data_post_processor.py

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

api/core/rag/datasource/retrieval_service.py

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

api/core/rag/docstore/dataset_docstore.py

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

api/core/rag/embedding/token_counter.py

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

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

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

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

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

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

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

api/core/rag/rerank/rerank_model.py

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

api/core/rag/retrieval/dataset_retrieval.py

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

api/core/rbac/entities.py

@@ -10,7 +10,6 @@ class RBACResourceScope(StrEnum):
 
     APP = "app"
     DATASET = "dataset"
-    KNOWLEDGE_FS = "knowledge_space"
     WORKSPACE = "workspace"
 
 
@@ -58,19 +57,11 @@ class RBACPermission(StrEnum):
     DATASET_EXTERNAL_CONNECT = "dataset_external_connect"
     DATASET_IMPORT_EXPORT_DSL = "dataset_import_export_dsl"
 
-    KNOWLEDGE_FS_READ = "knowledge_space_read"
-    KNOWLEDGE_FS_CREATE = "knowledge_space_create"
-    KNOWLEDGE_FS_EDIT = "knowledge_space_edit"
-    KNOWLEDGE_FS_DELETE = "knowledge_space_delete"
-    KNOWLEDGE_FS_ACCESS_CONFIG = "knowledge_space_access_config"
-    KNOWLEDGE_FS_API_KEY_MANAGE = "knowledge_space_api_key_manage"
-    KNOWLEDGE_FS_DOCUMENT_WRITE = "knowledge_space_document_write"
-    KNOWLEDGE_FS_QUERY = "knowledge_space_query"
-
     WORKSPACE_MEMBER_MANAGE = "workspace_member_manage"
     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"

api/core/tools/__base/tool.py

@@ -58,7 +58,6 @@ class Tool(ABC):
         if self.runtime and self.runtime.runtime_parameters:
             tool_parameters.update(self.runtime.runtime_parameters)
 
-        # try parse tool parameters into the correct type
         tool_parameters = self._transform_tool_parameters_type(tool_parameters)
 
         result = self._invoke(
@@ -87,14 +86,14 @@ class Tool(ABC):
                 return result
 
     def _transform_tool_parameters_type(self, tool_parameters: dict[str, Any]) -> dict[str, Any]:
-        """
-        Transform tool parameters type
-        """
-        # Temp fix for the issue that the tool parameters will be converted to empty while validating the credentials
+        """Transform declared tool parameter values without resolving runtime schemas."""
         result = deepcopy(tool_parameters)
         for parameter in self.entity.parameters or []:
             if parameter.name in tool_parameters:
-                result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])
+                if parameter.multiple:
+                    result[parameter.name] = parameter.init_frontend_parameter(result.get(parameter.name))
+                else:
+                    result[parameter.name] = parameter.type.cast_value(tool_parameters[parameter.name])
 
         return result
 
@@ -196,17 +195,31 @@ class Tool(ABC):
             }:
                 continue
 
-            parameter_schema: dict[str, Any] = (
-                {
-                    "type": parameter.type.as_normal_type(),
-                    "description": parameter.llm_description or "",
-                }
-                if parameter.input_schema is None
-                else deepcopy(parameter.input_schema)
-            )
+            is_multiple_select = parameter.multiple and parameter.type in {
+                ToolParameter.ToolParameterType.SELECT,
+                ToolParameter.ToolParameterType.DYNAMIC_SELECT,
+            }
+            if is_multiple_select:
+                item_schema: dict[str, Any] = {"type": "string"}
+                if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+                    item_schema["enum"] = [option.value for option in parameter.options]
+                parameter_schema: dict[str, Any] = {"type": "array", "items": item_schema}
+            else:
+                parameter_schema = (
+                    {
+                        "type": parameter.type.as_normal_type(),
+                        "description": parameter.llm_description or "",
+                    }
+                    if parameter.input_schema is None
+                    else deepcopy(parameter.input_schema)
+                )
             parameter_schema.setdefault("description", parameter.llm_description or "")
 
-            if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
+            if (
+                not is_multiple_select
+                and parameter.type == ToolParameter.ToolParameterType.SELECT
+                and parameter.options
+            ):
                 parameter_schema["enum"] = [option.value for option in parameter.options]
 
             schema["properties"][parameter.name] = parameter_schema

api/core/tools/__base/tool_runtime.py

@@ -2,7 +2,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from core.app.entities.app_invoke_entities import DifyRunContext, InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.tools.entities.tool_entities import ToolInvokeFrom
 
@@ -20,7 +20,6 @@ class ToolRuntime(BaseModel):
     tool_id: str | None = None
     invoke_from: InvokeFrom | None = None
     tool_invoke_from: ToolInvokeFrom | None = None
-    dify_run_context: DifyRunContext | None = Field(default=None, exclude=True, repr=False)
     credentials: dict[str, Any] = Field(default_factory=dict)
     credential_type: CredentialType = Field(default=CredentialType.API_KEY)
     runtime_parameters: dict[str, Any] = Field(default_factory=dict)

api/core/tools/builtin_tool/_position.yaml

@@ -1,5 +1,4 @@
 - audio
 - code
-- knowledge_fs
 - time
 - webscraper

api/core/tools/builtin_tool/providers/knowledge_fs/_assets/icon.svg

@@ -1,6 +0,0 @@
-<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">
-  <rect width="64" height="64" rx="14" fill="#155EEF"/>
-  <path d="M17 17h20c5.5 0 10 4.5 10 10v20H27c-5.5 0-10-4.5-10-10V17Z" fill="white" fill-opacity=".96"/>
-  <path d="M26 27h12M26 34h12M26 41h7" stroke="#155EEF" stroke-width="4" stroke-linecap="round"/>
-</svg>
-

api/core/tools/builtin_tool/providers/knowledge_fs/knowledge_fs.py

@@ -1,9 +0,0 @@
-from typing import Any, override
-
-from core.tools.builtin_tool.provider import BuiltinToolProviderController
-
-
-class KnowledgeFSProvider(BuiltinToolProviderController):
-    @override
-    def _validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> None:
-        _ = (user_id, credentials)

api/core/tools/builtin_tool/providers/knowledge_fs/knowledge_fs.yaml

@@ -1,13 +0,0 @@
-identity:
-  author: Dify
-  name: knowledge_fs
-  label:
-    en_US: KnowledgeFS
-    zh_Hans: KnowledgeFS
-  description:
-    en_US: Run explicitly bound KnowledgeFS operations from an Agent or Workflow.
-    zh_Hans: 从 Agent 或 Workflow 执行显式绑定的 KnowledgeFS 操作。
-  icon: icon.svg
-  tags:
-    - rag
-

api/core/tools/builtin_tool/providers/knowledge_fs/tools/create_research_task.py

@@ -1,60 +0,0 @@
-from collections.abc import Generator
-from typing import Any, override
-
-from pydantic import ValidationError
-from sqlalchemy.orm import Session, sessionmaker
-
-from core.tools.builtin_tool.tool import BuiltinTool
-from core.tools.entities.tool_entities import ToolInvokeFrom, ToolInvokeMessage
-from core.tools.errors import ToolInvokeError
-from models.knowledge_fs import KnowledgeFSAppSpaceJoinType
-from services.knowledge_fs.app_execution_capability import KnowledgeResourceRef
-from services.knowledge_fs.product_dto import KnowledgeFSResearchTaskCreatePayload
-from services.knowledge_fs.runtime import create_knowledge_fs_runtime
-
-
-class KnowledgeFSCreateResearchTaskTool(BuiltinTool):
-    @override
-    def _invoke(
-        self,
-        session: Session,
-        user_id: str,
-        tool_parameters: dict[str, Any],
-        conversation_id: str | None = None,
-        app_id: str | None = None,
-        message_id: str | None = None,
-    ) -> Generator[ToolInvokeMessage, None, None]:
-        _ = (user_id, conversation_id, app_id, message_id)
-        run_context = self.runtime.dify_run_context
-        if run_context is None or self.runtime.tenant_id != run_context.tenant_id:
-            raise ToolInvokeError("KnowledgeFS requires a trusted Dify run context")
-        match self.runtime.tool_invoke_from:
-            case ToolInvokeFrom.AGENT:
-                caller_kind = KnowledgeFSAppSpaceJoinType.AGENT
-            case ToolInvokeFrom.WORKFLOW:
-                caller_kind = KnowledgeFSAppSpaceJoinType.WORKFLOW
-            case _:
-                raise ToolInvokeError("KnowledgeFS is only available to Agent and Workflow callers")
-
-        try:
-            resource = KnowledgeResourceRef.model_validate(tool_parameters.get("resource"))
-            payload_data: dict[str, object] = {
-                "query": tool_parameters.get("query"),
-            }
-            mode = tool_parameters.get("mode")
-            if mode:
-                payload_data["mode"] = mode
-            payload = KnowledgeFSResearchTaskCreatePayload.model_validate(payload_data)
-            runtime = create_knowledge_fs_runtime(sessionmaker(bind=session.get_bind(), expire_on_commit=False))
-            response = runtime.app_capabilities.create_research_task(
-                run_context=run_context,
-                caller_kind=caller_kind,
-                resource=resource,
-                payload=payload,
-            )
-        except ToolInvokeError:
-            raise
-        except (ValidationError, RuntimeError, ValueError) as exc:
-            raise ToolInvokeError(str(exc)) from exc
-
-        yield self.create_json_message(response.model_dump(mode="json", by_alias=True))

api/core/tools/builtin_tool/providers/knowledge_fs/tools/create_research_task.yaml

@@ -1,75 +0,0 @@
-identity:
-  name: create_research_task
-  author: Dify
-  label:
-    en_US: Create KnowledgeFS Research Task
-    zh_Hans: 创建 KnowledgeFS Research 任务
-description:
-  human:
-    en_US: Create a Research task in an explicitly bound KnowledgeFS space.
-    zh_Hans: 在显式绑定的 KnowledgeFS 空间中创建 Research 任务。
-  llm: Create a durable research task using an explicitly configured KnowledgeFS resource.
-parameters:
-  - name: resource
-    type: object
-    required: true
-    label:
-      en_US: KnowledgeFS resource
-      zh_Hans: KnowledgeFS 资源
-    human_description:
-      en_US: A typed KnowledgeFS control-space reference configured by the app author.
-      zh_Hans: 由应用作者配置的类型化 KnowledgeFS control-space 引用。
-    form: form
-    input_schema:
-      type: object
-      additionalProperties: false
-      properties:
-        kind:
-          type: string
-          const: knowledge_fs
-        control_space_id:
-          type: string
-          minLength: 1
-      required:
-        - kind
-        - control_space_id
-  - name: query
-    type: string
-    required: true
-    label:
-      en_US: Research query
-      zh_Hans: Research 查询
-    human_description:
-      en_US: The question the Research task should investigate.
-      zh_Hans: Research 任务需要调查的问题。
-    llm_description: The question to investigate with KnowledgeFS.
-    form: llm
-  - name: mode
-    type: select
-    required: false
-    label:
-      en_US: Mode
-      zh_Hans: 模式
-    human_description:
-      en_US: Optional KnowledgeFS retrieval mode.
-      zh_Hans: 可选的 KnowledgeFS 检索模式。
-    llm_description: Optional retrieval mode. Use auto unless the task needs a specific mode.
-    form: llm
-    options:
-      - value: auto
-        label:
-          en_US: Auto
-          zh_Hans: 自动
-      - value: fast
-        label:
-          en_US: Fast
-          zh_Hans: 快速
-      - value: deep
-        label:
-          en_US: Deep
-          zh_Hans: 深度
-      - value: research
-        label:
-          en_US: Research
-          zh_Hans: Research
-

api/core/tools/entities/tool_entities.py

@@ -292,9 +292,7 @@ class ToolInvokeMessageBinary(BaseModel):
 
 
 class ToolParameter(PluginParameter):
-    """
-    Overrides type
-    """
+    """Tool-specific parameter declaration and invocation-value normalization."""
 
     class ToolParameterType(StrEnum):
         """
@@ -333,12 +331,28 @@ class ToolParameter(PluginParameter):
         LLM = auto()  # will be set by LLM
 
     type: ToolParameterType = Field(..., description="The type of the parameter")
+    multiple: bool = Field(
+        default=False,
+        description="Whether the parameter is multiple select, only valid for select or dynamic-select type",
+    )
     human_description: I18nObject | None = Field(default=None, description="The description presented to the user")
     form: ToolParameterForm = Field(..., description="The form of the parameter, schema/form/llm")
     llm_description: str | None = None
     # MCP object and array type parameters use this field to store the schema
     input_schema: dict[str, Any] | None = None
 
+    @model_validator(mode="after")
+    def validate_multiple(self) -> ToolParameter:
+        supports_multiple = self.type in {
+            self.ToolParameterType.SELECT,
+            self.ToolParameterType.DYNAMIC_SELECT,
+        }
+        if self.multiple and not supports_multiple:
+            raise ValueError("multiple is only valid for select and dynamic-select parameters")
+        if supports_multiple and self.default is not None and (isinstance(self.default, list) != self.multiple):
+            raise ValueError("default must be a list exactly when multiple is true")
+        return self
+
     @classmethod
     def get_simple_instance(
         cls,
@@ -378,8 +392,25 @@ class ToolParameter(PluginParameter):
             options=option_objs,
         )
 
-    def init_frontend_parameter(self, value: Any):
-        return init_frontend_parameter(self, self.type, value)
+    def init_frontend_parameter(self, value: Any) -> Any:
+        """Normalize a value against this tool parameter's full declaration."""
+        if not self.multiple:
+            return init_frontend_parameter(self, self.type, value)
+
+        parameter_value = self.default if value is None else value
+        if parameter_value is None:
+            parameter_value = []
+        if not isinstance(parameter_value, list):
+            raise ValueError(f"tool parameter {self.name} must be a list when multiple is true")
+        if not all(isinstance(item, str) for item in parameter_value):
+            raise ValueError(f"tool parameter {self.name} must contain only strings")
+        if self.required and not parameter_value:
+            raise ValueError(f"tool parameter {self.name} not found in tool config")
+        if self.type == self.ToolParameterType.SELECT:
+            options = [option.value for option in self.options]
+            if any(item not in options for item in parameter_value):
+                raise ValueError(f"tool parameter {self.name} value {parameter_value} not in options {options}")
+        return parameter_value
 
 
 class ToolProviderIdentity(BaseModel):

api/core/workflow/generator/prompts/builder_prompts.py

@@ -42,9 +42,9 @@ _NODE_SNIPPETS: dict[str, str] = {
     ["local_file", "remote_url"]. Only when you include "custom" must you
     also set ``allowed_file_extensions`` to a non-empty list like
     [".epub", ".rtf"]; otherwise leave it [].
-    In Advanced-Chat mode ``sys.query`` and ``sys.files`` are automatic
-    system variables — downstream nodes may reference them; do NOT add
-    them to ``variables``.""",
+    In Advanced-Chat mode ``sys.query`` is automatic. ``userinput.files`` is
+    the automatic file-upload variable in both app modes. Downstream nodes
+    may reference these variables; do NOT add them to ``variables``.""",
     "end": """\
 - end (Workflow mode only):
     {"outputs": [
@@ -168,8 +168,8 @@ _NODE_SNIPPETS: dict[str, str] = {
     Single output variable ``text``: a string when ``is_array_file`` is false,
     an array of strings (one per file) when it is true. ``variable_selector``
     MUST point at a ``start`` variable declared with type "file" / "file-list"
-    (or ``sys.files`` in Advanced-Chat mode). That start variable MUST set a
-    non-empty ``allowed_file_types`` (use ["document"] for document text).""",
+    (or the automatic ``userinput.files`` variable). A declared start variable
+    MUST set a non-empty ``allowed_file_types`` (use ["document"] for document text).""",
     "variable-aggregator": """\
 - variable-aggregator  (merge mutually-exclusive branches into one output):
     {"output_type": "string",        # VarType of the merged value — one of

api/core/workflow/generator/prompts/node_builder_prompts.py

@@ -91,21 +91,20 @@ def format_parallel_plan(
 def format_mode_section(mode: str) -> str:
     """Tell each builder which app mode it is configuring for.
 
-    Matters most in advanced-chat, where ``sys.query`` / ``sys.files`` are the
-    sanctioned way to reference the user's message — without this the model
-    invents start-node variables that postprocess then materializes as
-    spurious form inputs.
+    ``sys.query`` is available in advanced-chat, while ``userinput.files`` is
+    available in both app modes. Without this guidance the model invents
+    start-node variables that postprocess then materializes as spurious inputs.
     """
     if mode == "advanced-chat":
         return (
             "# App mode\n\n"
             "advanced-chat: the user's chat message is available as sys.query and uploaded files "
-            'as sys.files — placeholder {{#sys.query#}}, selector ["sys", "query"]. Reference them '
-            "directly; do NOT invent start-node variables for the chat message.\n\n"
+            "as userinput.files. Use placeholder {{#sys.query#}} or selector "
+            '["userinput", "files"] directly; do NOT invent start-node variables for them.\n\n'
         )
     return (
         "# App mode\n\n"
-        "workflow: there are NO automatic system variables; reference user input only through "
+        "workflow: uploaded files are available as userinput.files; all other user input must use "
         "the start node's declared variables.\n\n"
     )
 

api/core/workflow/generator/prompts/planner_prompts.py

@@ -96,10 +96,10 @@ minimum set of Dify workflow nodes needed to fulfil it, in execution order.
       - "text-input" for short single-line values (URLs, names),
       - "paragraph" for free-form multi-line text (descriptions, queries),
       - "number" / "select" / "file" / "file-list" for the obvious cases.
-    In Advanced-Chat mode the ``sys.query`` / ``sys.files`` system
-    variables are automatic — downstream nodes may reference them without
-    a ``start_inputs`` entry. In Workflow mode there is NO automatic
-    variable; everything the user supplies must be in ``start_inputs``.
+    In Advanced-Chat mode ``sys.query`` is automatic. ``userinput.files`` is
+    automatic in both app modes. Downstream nodes may reference these values
+    without a ``start_inputs`` entry; every other user-supplied Workflow value
+    must be declared in ``start_inputs``.
 11. Give every node a unique runtime-safe ``id`` using only letters, digits,
     and underscores. In create mode use ``node1``, ``node2``, ... in node-list
     order. In refine mode preserve the existing id for every retained node.

api/core/workflow/generator/runner.py

@@ -1382,8 +1382,8 @@ class WorkflowGenerator:
         multiple outputs remain untouched so validation fails closed instead
         of guessing which value the workflow should consume.
 
-        For Advanced-Chat mode, ``sys.query`` and ``sys.files`` are always
-        treated as resolved without any declaration. Tool nodes' parameter
+        ``sys.query`` in Advanced-Chat mode and ``userinput.files`` in either mode
+        are treated as resolved without declarations. Tool nodes' parameter
         references aren't validated here because we don't know each tool's
         schema — the run time validates those.
         """
@@ -1398,9 +1398,12 @@ class WorkflowGenerator:
         for node in nodes:
             cls._collect_refs_in_data(node.get("data") or {}, refs)
 
+        automatic_refs = {("userinput", "files")}
+        if mode == "advanced-chat":
+            automatic_refs.add(("sys", "query"))
+
         for node_id, var in refs:
-            # Advanced-Chat system variables are always resolved.
-            if mode == "advanced-chat" and node_id == "sys":
+            if (node_id, var) in automatic_refs:
                 continue
             target = nodes_by_id.get(node_id)
             if target is None:

api/core/workflow/legacy_system_files.py

@@ -0,0 +1,218 @@
+"""Compatibility helpers for workflows that still reference deprecated `sys.files`.
+
+TODO: Remove this module after all persisted Workflow and Advanced Chat graphs
+have been migrated from the deprecated system file variable to `userinput.files`.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Generator, Iterable, Mapping
+from dataclasses import dataclass
+from typing import Any
+
+_LEGACY_SYSTEM_NODE_ID = "sys"
+_USER_INPUT_NODE_ID = "userinput"
+_LEGACY_FILES_VARIABLE = "files"
+_USER_INPUT_FILE_SELECTOR = [_USER_INPUT_NODE_ID, _LEGACY_FILES_VARIABLE]
+_USER_INPUT_FILE_INPUT_KEY = ".".join(_USER_INPUT_FILE_SELECTOR)
+_LEGACY_FILES_TEMPLATE = "{{#sys.files#}}"
+_USER_INPUT_FILES_TEMPLATE = "{{#userinput.files#}}"
+
+
+@dataclass(frozen=True)
+class LegacySysFilesCompatVariable:
+    node_id: str
+    variable_name: str
+
+
+@dataclass(frozen=True)
+class LegacySysFilesGraphMigrationResult:
+    graph: dict[str, Any]
+    changed: bool
+
+
+def migrate_legacy_sys_files_graph_with_result(
+    graph: Mapping[str, Any],
+) -> LegacySysFilesGraphMigrationResult:
+    """Return the migrated graph and whether any legacy reference was rewritten."""
+
+    graph_copy = dict(graph)
+    nodes = graph_copy.get("nodes")
+    if not isinstance(nodes, list):
+        return LegacySysFilesGraphMigrationResult(graph=graph_copy, changed=False)
+
+    # Legacy references are stored in node data. Restricting both search and replacement to `nodes`
+    # avoids recursively scanning graph-level metadata and edges for every workflow load.
+    if not _contains_legacy_sys_files_reference(nodes):
+        return LegacySysFilesGraphMigrationResult(graph=graph_copy, changed=False)
+
+    graph_copy["nodes"] = _replace_legacy_sys_files_references(nodes)
+    return LegacySysFilesGraphMigrationResult(graph=graph_copy, changed=True)
+
+
+def resolve_legacy_sys_files_compat_variable(graph: Mapping[str, Any]) -> LegacySysFilesCompatVariable | None:
+    """Resolve the target variable used by the `sys.files` compatibility layer."""
+
+    nodes = graph.get("nodes")
+    if not isinstance(nodes, list):
+        return None
+    if not _contains_file_input_reference(nodes):
+        return None
+    return LegacySysFilesCompatVariable(node_id=_USER_INPUT_NODE_ID, variable_name=_LEGACY_FILES_VARIABLE)
+
+
+def normalize_legacy_sys_files_args(
+    *,
+    graph: Mapping[str, Any],
+    args: Mapping[str, Any],
+) -> tuple[dict[str, Any], LegacySysFilesCompatVariable | None]:
+    """Map Service/Web API file arguments onto the `userinput.files` system alias.
+
+    The top-level `files` argument and hidden `system.files` payload both feed
+    the same runtime file collection. After graph references are migrated, the
+    file collection is exposed in the variable pool as `userinput.files`.
+    """
+
+    normalized_args = dict(args)
+    files_from_input, input_files_used = _extract_userinput_files(args)
+    if input_files_used:
+        normalized_args["files"] = files_from_input
+        return normalized_args, None
+
+    compat_variable = resolve_legacy_sys_files_compat_variable(graph)
+    if compat_variable is None:
+        return normalized_args, None
+
+    files, legacy_files_used = _extract_legacy_files(args)
+    if not legacy_files_used:
+        return normalized_args, None
+
+    if normalized_args.get("files") is None:
+        normalized_args["files"] = files
+
+    raw_inputs = normalized_args.get("inputs")
+    inputs = dict(raw_inputs) if isinstance(raw_inputs, Mapping) else {}
+    inputs.setdefault(_USER_INPUT_FILE_INPUT_KEY, files)
+    normalized_args["inputs"] = inputs
+    return normalized_args, compat_variable
+
+
+def attach_legacy_sys_files_warning(
+    response: Mapping[str, Any] | Iterable[Any],
+    compat_variable: LegacySysFilesCompatVariable,
+) -> Mapping[str, Any] | Generator[str, None, None]:
+    warning = build_legacy_sys_files_warning(compat_variable)
+    if isinstance(response, Mapping):
+        response_with_warning = dict(response)
+        existing_warnings = response_with_warning.get("warnings")
+        warnings = list(existing_warnings) if isinstance(existing_warnings, list) else []
+        warnings.append(warning)
+        response_with_warning["warnings"] = warnings
+        return response_with_warning
+
+    def _with_warning() -> Generator[str, None, None]:
+        try:
+            yield f"data: {json.dumps({'event': 'warning', 'warning': warning})}\n\n"
+            yield from response
+        finally:
+            close = getattr(response, "close", None)
+            if callable(close):
+                close()
+
+    return _with_warning()
+
+
+def build_legacy_sys_files_warning(compat_variable: LegacySysFilesCompatVariable) -> str:
+    variable_selector = ".".join((compat_variable.node_id, compat_variable.variable_name))
+    return (
+        "sys.files is deprecated. This workflow now reads files from "
+        f"`{variable_selector}`; update Service API calls to pass files in "
+        f"`inputs.{variable_selector}` instead of `system.files` or top-level `files`."
+    )
+
+
+def _contains_legacy_sys_files_reference(value: Any) -> bool:
+    if _is_legacy_sys_files_selector(value):
+        return True
+
+    if isinstance(value, str):
+        return _LEGACY_FILES_TEMPLATE in value
+
+    if isinstance(value, Mapping):
+        return any(_contains_legacy_sys_files_reference(item) for item in value.values())
+
+    if isinstance(value, list):
+        return any(_contains_legacy_sys_files_reference(item) for item in value)
+
+    return False
+
+
+def _contains_file_input_reference(value: Any) -> bool:
+    if _is_legacy_sys_files_selector(value) or _is_userinput_files_selector(value):
+        return True
+
+    if isinstance(value, str):
+        return _LEGACY_FILES_TEMPLATE in value or _USER_INPUT_FILES_TEMPLATE in value
+
+    if isinstance(value, Mapping):
+        return any(_contains_file_input_reference(item) for item in value.values())
+
+    if isinstance(value, list):
+        return any(_contains_file_input_reference(item) for item in value)
+
+    return False
+
+
+def _replace_legacy_sys_files_references(value: Any) -> Any:
+    if _is_legacy_sys_files_selector(value):
+        return list(_USER_INPUT_FILE_SELECTOR)
+
+    if isinstance(value, str):
+        return value.replace(_LEGACY_FILES_TEMPLATE, _USER_INPUT_FILES_TEMPLATE)
+
+    if isinstance(value, Mapping):
+        return {key: _replace_legacy_sys_files_references(item) for key, item in value.items()}
+
+    if isinstance(value, list):
+        return [_replace_legacy_sys_files_references(item) for item in value]
+
+    return value
+
+
+def _is_legacy_sys_files_selector(value: Any) -> bool:
+    return (
+        isinstance(value, list)
+        and len(value) == 2
+        and value[0] == _LEGACY_SYSTEM_NODE_ID
+        and value[1] == _LEGACY_FILES_VARIABLE
+    )
+
+
+def _is_userinput_files_selector(value: Any) -> bool:
+    return isinstance(value, list) and value == _USER_INPUT_FILE_SELECTOR
+
+
+def serialized_graph_may_contain_legacy_sys_files(serialized_graph: str) -> bool:
+    """Cheaply reject stored graphs that cannot contain a legacy file reference."""
+
+    return _LEGACY_FILES_TEMPLATE in serialized_graph or ('"sys"' in serialized_graph and '"files"' in serialized_graph)
+
+
+def _extract_legacy_files(args: Mapping[str, Any]) -> tuple[Any, bool]:
+    if "files" in args and args["files"] is not None:
+        return args["files"], True
+
+    system = args.get("system")
+    if isinstance(system, Mapping) and "files" in system and system["files"] is not None:
+        return system["files"], True
+
+    return None, False
+
+
+def _extract_userinput_files(args: Mapping[str, Any]) -> tuple[Any, bool]:
+    inputs = args.get("inputs")
+    if isinstance(inputs, Mapping) and inputs.get(_USER_INPUT_FILE_INPUT_KEY) is not None:
+        return inputs[_USER_INPUT_FILE_INPUT_KEY], True
+
+    return None, False

api/core/workflow/node_runtime.py

@@ -493,7 +493,6 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
                 self._run_context.invoke_from,
                 variable_pool,
             )
-            tool_runtime.runtime.dify_run_context = self._run_context
         except ToolNodeError:
             raise
         except Exception as exc: