[autofix.ci] apply automated fixes

fix(workflow): migrate sys files to user input
fix: validate plugin installation scope (#39669 )
2026-07-28 05:33:30 +00:00 · 2026-07-28 13:29:09 +08:00 · 2026-07-28 04:13:20 +00:00 · 2026-07-28 04:09:32 +00:00 · 2026-07-28 04:02:28 +00:00 · 2026-07-28 04:01:12 +00:00
343 changed files with 14806 additions and 10155 deletions
@@ -29,13 +29,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -88,13 +88,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -139,13 +139,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -20,7 +20,7 @@ jobs:
        run: echo "autofix.ci updates pull request branches, not merge group refs."

      - if: github.event_name != 'merge_group'
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

      - name: Check Docker Compose inputs
        if: github.event_name != 'merge_group'
@@ -84,12 +84,12 @@ jobs:
            dify-agent/pyproject.toml
            dify-agent/uv.lock
      - if: github.event_name != 'merge_group'
-        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
        with:
          python-version: "3.11"

      - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0

      - name: Generate Docker Compose
        if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -97,7 +97,7 @@ jobs:
          echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV

      - name: Login to Docker Hub
-        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
+        uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
        with:
          username: ${{ env.DOCKERHUB_USER }}
          password: ${{ env.DOCKERHUB_TOKEN }}
@@ -199,7 +199,7 @@ jobs:
          merge-multiple: true

      - name: Login to Docker Hub
-        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
+        uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
        with:
          username: ${{ env.DOCKERHUB_USER }}
          password: ${{ env.DOCKERHUB_TOKEN }}
@@ -79,7 +79,7 @@ jobs:
      ws2_app_id:                 ${{ steps.out.outputs.DIFY_E2E_WS2_APP_ID }}

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -123,7 +123,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -170,7 +170,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -233,7 +233,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -295,7 +295,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -351,7 +351,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -23,7 +23,7 @@ jobs:
        working-directory: ./cli
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 0
@@ -35,7 +35,7 @@ jobs:
      dify_tag: ${{ steps.resolve.outputs.dify_tag }}
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -98,7 +98,7 @@ jobs:
      DIFY_TAG: ${{ needs.validate.outputs.dify_tag }}
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 1
@@ -24,7 +24,7 @@ jobs:
        shell: bash
    steps:
      - name: Checkout cli ref
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -30,7 +30,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -13,13 +13,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -63,13 +63,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -24,7 +24,7 @@ jobs:
    name: Require cherry-pick provenance
    runs-on: depot-ubuntu-24.04
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

@@ -9,6 +9,6 @@ jobs:
      pull-requests: write
    runs-on: depot-ubuntu-24.04
    steps:
-      - uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
+      - uses: actions/labeler@bf12e9b00b37c5c0ca2b87b79b2daf7891dbda13 # v7.0.0
        with:
          sync-labels: true
@@ -47,7 +47,7 @@ jobs:
      migration-changed: ${{ steps.changes.outputs.migration }}
      sandbox-runtime-changed: ${{ steps.changes.outputs.sandbox-runtime }}
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
        id: changes
        with:
@@ -18,7 +18,7 @@ jobs:
    outputs:
      external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
        id: changes
        with:
@@ -17,12 +17,12 @@ jobs:
      pull-requests: write
    steps:
      - name: Checkout PR branch
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -21,10 +21,10 @@ jobs:
    if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
    steps:
      - name: Checkout default branch (trusted code)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -17,12 +17,12 @@ jobs:
      pull-requests: write
    steps:
      - name: Checkout PR branch
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -21,7 +21,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -45,7 +45,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -72,7 +72,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -23,7 +23,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 0
@@ -45,7 +45,7 @@ jobs:

      - name: Setup UV and Python
        if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: false
          python-version: "3.12"
@@ -93,7 +93,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -144,7 +144,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -186,7 +186,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -24,7 +24,7 @@ jobs:
        working-directory: sdks/nodejs-client

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -40,7 +40,7 @@ jobs:

    steps:
      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          token: ${{ secrets.GITHUB_TOKEN }}
@@ -158,7 +158,7 @@ jobs:

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

    steps:
      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

@@ -24,7 +24,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -36,7 +36,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -21,7 +21,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -33,7 +33,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -26,7 +26,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -34,7 +34,7 @@ jobs:
        uses: ./.github/actions/setup-web

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -29,7 +29,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -62,7 +62,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -100,7 +100,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -132,7 +132,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -666,6 +666,7 @@ PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
 PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
+PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED=true
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
 # Comma-separated marketplace plugin IDs whose latest versions are installed for newly registered users.
 # Example: langgenius/openai,langgenius/gemini
@@ -677,6 +678,8 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y

 # Dify Agent backend
 AGENT_BACKEND_BASE_URL=http://localhost:5050
+# Bearer token sent to the Agent backend /runs API. Must match DIFY_AGENT_API_TOKEN on the server side.
+AGENT_BACKEND_API_TOKEN=dify-agent-run-token-for-dev-only
 AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS=30
 AGENT_BACKEND_STREAM_MAX_RECONNECTS=3
 AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
@@ -729,7 +732,6 @@ OTEL_MAX_EXPORT_BATCH_SIZE=512
 OTEL_METRIC_EXPORT_INTERVAL=60000
 OTEL_BATCH_EXPORT_TIMEOUT=10000
 OTEL_METRIC_EXPORT_TIMEOUT=30000
-
 # Prevent Clickjacking
 ALLOW_EMBED=false

@@ -1,10 +1,13 @@
 import logging
 import time
+from collections.abc import Callable
+from typing import NamedTuple

 import socketio
 from flask import request
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
+from werkzeug.exceptions import Forbidden, HTTPException, ServiceUnavailable

 from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
@@ -42,6 +45,53 @@ _CONSOLE_EXEMPT_PREFIXES = (
    "/console/api/activate/check",
 )

+_WEBAPP_EXEMPT_PREFIXES = ("/api/system-features",)
+
+_INVALID_LICENSE_STATUSES = (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST)
+
+
+def _session_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    if license_status is None:
+        return UnauthorizedAndForceLogout("Unable to verify enterprise license. Please contact your administrator.")
+    return UnauthorizedAndForceLogout(f"Enterprise license is {license_status}. Please contact your administrator.")
+
+
+def _bearer_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    """Token-authed: forcing a logout is meaningless and license state must not leak."""
+    return Forbidden(description="license_required")
+
+
+def _retryable_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    """Webhook senders retry on 5xx but treat 4xx as permanent, disabling the subscription."""
+    return ServiceUnavailable(description="license_required")
+
+
+class _LicenseGatedSurface(NamedTuple):
+    prefix: str
+    exempt_prefixes: tuple[str, ...]
+    build_error: Callable[[LicenseStatus | None], HTTPException]
+
+
+# /files (plugin-daemon data plane), /inner/api (enterprise control plane) and /health
+# stay ungated: blocking them breaks workflow execution or license recovery itself.
+_LICENSE_GATED_SURFACES = (
+    _LicenseGatedSurface("/console/api/", _CONSOLE_EXEMPT_PREFIXES, _session_surface_error),
+    _LicenseGatedSurface("/api/", _WEBAPP_EXEMPT_PREFIXES, _session_surface_error),
+    _LicenseGatedSurface("/v1", (), _bearer_surface_error),
+    _LicenseGatedSurface("/mcp", (), _bearer_surface_error),
+    _LicenseGatedSurface("/triggers", (), _retryable_surface_error),
+)
+
+
+def _match_license_gated_surface(path: str) -> _LicenseGatedSurface | None:
+    for surface in _LICENSE_GATED_SURFACES:
+        if not path.startswith(surface.prefix):
+            continue
+        if any(path.startswith(exempt) for exempt in surface.exempt_prefixes):
+            return None
+        return surface
+    return None
+

 # ----------------------------
 # Application Factory Function
@@ -62,38 +112,17 @@ def create_flask_app_with_configs() -> DifyApp:
        init_request_context()
        RecyclableContextVar.increment_thread_recycles()

-        # Enterprise license validation for API endpoints (both console and webapp)
-        # When license expires, block all API access except bootstrap endpoints needed
-        # for the frontend to load the license expiration page without infinite reloads.
        if dify_config.ENTERPRISE_ENABLED:
-            is_console_api = request.path.startswith("/console/api/")
-            is_webapp_api = request.path.startswith("/api/")
+            surface = _match_license_gated_surface(request.path)
+            if surface is not None:
+                try:
+                    license_status = EnterpriseService.get_cached_license_status()
+                except Exception:
+                    logger.exception("Failed to check enterprise license status")
+                    license_status = None

-            if is_console_api or is_webapp_api:
-                if is_console_api:
-                    is_exempt = any(request.path.startswith(p) for p in _CONSOLE_EXEMPT_PREFIXES)
-                else:  # webapp API
-                    is_exempt = request.path.startswith("/api/system-features")
-
-                if not is_exempt:
-                    try:
-                        # Check license status (cached — see EnterpriseService for TTL details)
-                        license_status = EnterpriseService.get_cached_license_status()
-                        if license_status in (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST):
-                            raise UnauthorizedAndForceLogout(
-                                f"Enterprise license is {license_status}. Please contact your administrator."
-                            )
-                        if license_status is None:
-                            raise UnauthorizedAndForceLogout(
-                                "Unable to verify enterprise license. Please contact your administrator."
-                            )
-                    except UnauthorizedAndForceLogout:
-                        raise
-                    except Exception:
-                        logger.exception("Failed to check enterprise license status")
-                        raise UnauthorizedAndForceLogout(
-                            "Unable to verify enterprise license. Please contact your administrator."
-                        )
+                if license_status is None or license_status in _INVALID_LICENSE_STATUSES:
+                    raise surface.build_error(license_status)

    # add after request hook for injecting trace headers from OpenTelemetry span context
    # Only adds headers when OTEL is enabled and has valid context
@@ -11,6 +11,7 @@ from clients.agent_backend.fake_client import FakeAgentBackendRunClient, FakeAge
 def create_agent_backend_run_client(
    *,
    base_url: str | None = None,
+    api_token: str | None = None,
    use_fake: bool = False,
    fake_scenario: str | FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCESS,
    stream_read_timeout_seconds: float = 30,
@@ -22,8 +23,11 @@ def create_agent_backend_run_client(
        return FakeAgentBackendRunClient(scenario=FakeAgentBackendScenario(fake_scenario))
    if base_url is None:
        raise ValueError("base_url is required when creating a real Agent backend client")
+    headers: dict[str, str] = {}
+    if api_token:
+        headers["Authorization"] = f"Bearer {api_token}"
    return DifyAgentBackendRunClient(
-        Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds),
+        Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds, headers=headers),
        stream_max_reconnects=stream_max_reconnects,
        stream_timeout_seconds=stream_run_timeout_seconds,
    )
@@ -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,6 +11,7 @@ from .data_migration import (
    import_migration_data,
    migration_data_wizard,
 )
+from .database import upgrade_db
 from .plugin import (
    backfill_plugin_auto_upgrade,
    extract_plugins,
@@ -36,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,
@@ -49,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",
@@ -80,6 +78,7 @@ __all__ = [
    "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",
@@ -1,86 +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 enums.deployment_edition import DeploymentEdition
 from events.app_event import app_was_created
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-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.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",
-                )
-            )
-

@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
@@ -133,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"""
@@ -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")
@@ -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.")
@@ -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",
+                )
+            )
@@ -12,6 +12,11 @@ class AgentBackendConfig(BaseSettings):
        default=None,
    )

+    AGENT_BACKEND_API_TOKEN: str | None = Field(
+        description="Bearer token for authenticating with the Agent backend /runs API.",
+        default=None,
+    )
+
    AGENT_BACKEND_USE_FAKE: bool = Field(
        description="Use the deterministic in-process fake Agent backend client.",
        default=False,
@@ -266,6 +266,12 @@ class PluginConfig(BaseSettings):
        default=60 * 60,
    )

+    PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED: bool = Field(
+        description="Whether tenant plugin model providers are cached in Redis. Disable when plugins are installed "
+        "by a system other than this one, which cannot invalidate the cache when a tenant's plugins change.",
+        default=True,
+    )
+
    PLUGIN_MODEL_PROVIDERS_CACHE_TTL: PositiveInt = Field(
        description="TTL in seconds for caching tenant plugin model providers in Redis",
        default=60 * 60 * 24,
@@ -816,6 +822,41 @@ class UpdateConfig(BaseSettings):
    )


+class CommunityTelemetryConfig(BaseSettings):
+    """
+    Configuration for anonymous self-hosted community telemetry.
+    """
+
+    DISABLE_TELEMETRY: bool = Field(
+        description="Disable anonymous community telemetry",
+        default=False,
+    )
+    DO_NOT_TRACK: bool = Field(
+        description="Respect the standard do-not-track opt-out signal for telemetry",
+        default=False,
+    )
+    TELEMETRY_ENDPOINT: str = Field(
+        description="Endpoint for anonymous community telemetry events",
+        default="https://otel.dify.ai/v1/events",
+    )
+    TELEMETRY_FALLBACK_ENDPOINT: str = Field(
+        description="Fallback endpoint for anonymous community telemetry events",
+        default="https://otel.dify.cn/v1/events",
+    )
+    TELEMETRY_TIMEOUT_SECONDS: PositiveInt = Field(
+        description="HTTP timeout in seconds for anonymous community telemetry requests",
+        default=3,
+    )
+    TELEMETRY_HEARTBEAT_INTERVAL_MINUTES: PositiveInt = Field(
+        description="Celery beat interval in minutes for checking whether heartbeat telemetry is due",
+        default=30,
+    )
+    CI: bool = Field(
+        description="Whether the process is running in CI; telemetry is skipped when true",
+        default=False,
+    )
+
+
 class WorkflowVariableTruncationConfig(BaseSettings):
    WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
        # 1000 KiB
@@ -1599,6 +1640,7 @@ class FeatureConfig(
    TenantIsolatedTaskQueueConfig,
    ToolConfig,
    UpdateConfig,
+    CommunityTelemetryConfig,
    WorkflowConfig,
    WorkflowNodeExecutionConfig,
    WorkspaceConfig,
@@ -257,7 +257,6 @@ class AgentAppDetailWithSite(GenericAppDetailWithSite):
    debug_conversation_has_messages: bool = False
    debug_conversation_message_count: int = 0
    role: str | None = None
-    active_config_is_published: bool = False


 class AgentDebugConversationRefreshResponse(BaseModel):
@@ -410,10 +409,6 @@ def _serialize_agent_app_detail(
    payload["debug_conversation_has_messages"] = message_count > 0
    payload["debug_conversation_message_count"] = message_count
    payload["role"] = agent.role or ""
-    payload["active_config_is_published"] = roster_service.active_config_is_published(
-        tenant_id=app_model.tenant_id,
-        agent=agent,
-    )
    return payload


@@ -58,6 +58,7 @@ from services.app_service import (
    AppResponseView,
    AppService,
    CreateAppParams,
+    RecentAppMode,
    StarredAppListParams,
 )
 from services.enterprise import rbac_service as enterprise_rbac_service
@@ -139,6 +140,10 @@ class AppListBaseQuery(BaseModel):
            raise ValueError("Invalid UUID format in creator_ids.") from exc


+class RecentAppListQuery(BaseModel):
+    limit: int = Field(default=8, ge=1, le=8, description="Number of recently modified apps to return (1-8)")
+
+
 class AppListQuery(AppListBaseQuery):
    pass

@@ -411,6 +416,33 @@ class AppPartial(AppResponseModel):
        return to_timestamp(value)


+class RecentAppResponse(ResponseModel):
+    id: str
+    name: str
+    icon_type: IconType | None = None
+    icon: str | None = None
+    icon_background: str | None = None
+    mode: RecentAppMode
+    author_name: str | None = None
+    updated_at: int
+    permission_keys: list[str] = Field(default_factory=list)
+    maintainer: str | None = None
+
+    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
+    @property
+    def icon_url(self) -> str | None:
+        return build_icon_url(self.icon_type, self.icon)
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int) -> int:
+        return to_timestamp(value)
+
+
+class RecentAppListResponse(ResponseModel):
+    data: list[RecentAppResponse]
+
+
 class AppDetail(AppResponseModel):
    id: str
    name: str
@@ -575,6 +607,8 @@ register_schema_models(
 register_response_schema_models(
    console_ns,
    AppPartial,
+    RecentAppResponse,
+    RecentAppListResponse,
    AppDetailWithSite,
    AppPagination,
 )
@@ -699,6 +733,49 @@ class AppListApi(Resource):
        return app_detail.model_dump(mode="json"), 201


+@console_ns.route("/apps/recent")
+class RecentAppListApi(Resource):
+    @console_ns.doc("list_recent_apps")
+    @console_ns.doc(description="Get recently modified apps for the home Continue Work section")
+    @console_ns.doc(params=query_params_from_model(RecentAppListQuery))
+    @console_ns.response(200, "Success", console_ns.models[RecentAppListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @with_session(write=False)
+    @with_current_user_id
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, current_user_id: str, session: Session):
+        """Return the lightweight app cards needed by the Explore home page."""
+        args = query_params_from_request(RecentAppListQuery)
+        params = AppListParams(limit=args.limit)
+
+        permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
+            current_tenant_id,
+            current_user_id,
+            session=session,
+        )
+        if dify_config.RBAC_ENABLED:
+            access_filter = resolve_app_access_filter(
+                current_tenant_id,
+                current_user_id,
+                session=session,
+                permissions=permissions,
+            )
+            access_filter.apply_to_params(params)
+
+        recent_apps = AppService().get_recent_apps(current_user_id, current_tenant_id, params, session)
+        permission_keys_map = permissions.app.permission_keys_by_resource_ids([app.id for app in recent_apps])
+        response_items = [
+            RecentAppResponse.model_validate(app, from_attributes=True).model_copy(
+                update={"permission_keys": permission_keys_map.get(app.id, [])}
+            )
+            for app in recent_apps
+        ]
+        return dump_response(RecentAppListResponse, {"data": response_items}), 200
+
+
@console_ns.route("/apps/starred")
 class StarredAppListApi(Resource):
    @console_ns.doc("list_starred_apps")
@@ -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)
@@ -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
@@ -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)
@@ -24,7 +24,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import RateLimiter, dump_response, extract_remote_ip, to_timestamp
 from models.account import TenantStatus
-from models.model import App, Site
+from models.model import App, AppMode, Site
 from repositories.factory import DifyAPIRepositoryFactory
 from services.feature_service import FeatureService
 from services.human_input_file_upload_service import HumanInputFileUploadService
@@ -207,6 +207,7 @@ class HumanInputFormApi(Resource):
                site=WebAppSiteResponse.from_app_site(
                    tenant=tenant,
                    app_model=app_model,
+                    mode=AppMode.value_of(app_model.mode),
                    site=site,
                    end_user_id=None,
                    features=features,
@@ -14,7 +14,7 @@ from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
 from models.account import Tenant, TenantStatus
-from models.model import App, EndUser, IconType, Site
+from models.model import App, AppMode, EndUser, IconType, Site
 from services.feature_service import FeatureModel, FeatureService
 from services.file_service import FileService

@@ -67,6 +67,7 @@ class WebAppCustomConfigResponse(ResponseModel):

 class WebAppSiteResponse(ResponseModel):
    app_id: str
+    mode: AppMode
    end_user_id: str | None = None
    enable_site: bool
    site: WebSiteResponse
@@ -83,6 +84,7 @@ class WebAppSiteResponse(ResponseModel):
        *,
        tenant: Tenant,
        app_model: App,
+        mode: AppMode,
        site: Site,
        end_user_id: str | None,
        features: FeatureModel,
@@ -109,6 +111,7 @@ class WebAppSiteResponse(ResponseModel):

        return cls(
            app_id=app_model.id,
+            mode=mode,
            end_user_id=end_user_id,
            enable_site=app_model.enable_site,
            site=site_response,
@@ -167,6 +170,7 @@ class AppSiteApi(WebApiResource):
        return WebAppSiteResponse.from_app_site(
            tenant=tenant,
            app_model=app_model,
+            mode=AppMode.value_of(app_model.mode_compatible_with_agent_with_session(session=db.session())),
            site=site,
            end_user_id=end_user.id,
            features=features,
@@ -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")
@@ -616,23 +619,34 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            message_snapshot = MessageSnapshot.from_message(message)
            session.close()

-            # return response or stream generator
-            response = self._handle_advanced_chat_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow_snapshot,
-                queue_manager=queue_manager,
-                conversation=conversation_snapshot,
-                message=message_snapshot,
-                user=user,
-                stream=stream,
-                draft_var_saver_factory=self._get_draft_var_saver_factory(
-                    invoke_from,
-                    account=user,
-                    tenant_id=application_generate_entity.app_config.tenant_id,
-                ),
-            )
+            try:
+                response = self._handle_advanced_chat_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow_snapshot,
+                    queue_manager=queue_manager,
+                    conversation=conversation_snapshot,
+                    message=message_snapshot,
+                    user=user,
+                    stream=stream,
+                    draft_var_saver_factory=self._get_draft_var_saver_factory(
+                        invoke_from,
+                        account=user,
+                        tenant_id=application_generate_entity.app_config.tenant_id,
+                    ),
+                )
+                converted_response = AdvancedChatAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def _generate_worker(
        self,
@@ -538,6 +538,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            request_builder=AgentAppRuntimeRequestBuilder(credentials_provider=credentials_provider),
            agent_backend_client=create_agent_backend_run_client(
                base_url=dify_config.AGENT_BACKEND_BASE_URL,
+                api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -1,3 +1,5 @@
+import logging
+import threading
 from collections.abc import Generator, Mapping, Sequence
 from contextlib import AbstractContextManager, nullcontext
 from typing import TYPE_CHECKING, Any, Union, final
@@ -23,6 +25,10 @@ from services.workflow_draft_variable_service import DraftVariableSaver as Draft
 if TYPE_CHECKING:
    from graphon.variables.input_entities import VariableEntity

+logger = logging.getLogger(__name__)
+
+_WORKER_THREAD_JOIN_TIMEOUT_SECONDS = 300
+

@final
 class _DebuggerDraftVariableSaver:
@@ -64,6 +70,29 @@ class _DebuggerDraftVariableSaver:
 class BaseAppGenerator:
    _file_access_controller: DatabaseFileAccessController = DatabaseFileAccessController()

+    @staticmethod
+    def _join_worker_thread(worker_thread: threading.Thread) -> None:
+        # Bound the wait so a leaked app worker cannot occupy an execution slot indefinitely.
+        worker_thread.join(timeout=_WORKER_THREAD_JOIN_TIMEOUT_SECONDS)
+        if worker_thread.is_alive():
+            logger.warning(
+                "Possible app worker thread leak: thread_name=%s timeout_seconds=%s; "
+                "continuing without waiting further to avoid occupying an execution slot indefinitely",
+                worker_thread.name,
+                _WORKER_THREAD_JOIN_TIMEOUT_SECONDS,
+            )
+
+    @staticmethod
+    def _wrap_stream_with_worker_thread_join[ResponseT](
+        response_stream: Generator[ResponseT, None, None],
+        worker_thread: threading.Thread,
+    ) -> Generator[ResponseT, None, None]:
+        """Keep the producer owned by the response stream until both finish."""
+        try:
+            yield from response_stream
+        finally:
+            BaseAppGenerator._join_worker_thread(worker_thread)
+
    @staticmethod
    def _bind_file_access_scope(
        *,
@@ -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)
@@ -351,17 +351,28 @@ class PipelineGenerator(BaseAppGenerator):
                user,
                tenant_id=pipeline.tenant_id,
            )
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow,
-                queue_manager=queue_manager,
-                user=user,
-                stream=streaming,
-                draft_var_saver_factory=draft_var_saver_factory,
-            )
+            try:
+                response = self._handle_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow,
+                    queue_manager=queue_manager,
+                    user=user,
+                    stream=streaming,
+                    draft_var_saver_factory=draft_var_saver_factory,
+                )
+                converted_response = WorkflowAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def single_iteration_generate(
        self,
@@ -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
@@ -405,17 +408,28 @@ class WorkflowAppGenerator(BaseAppGenerator):
                tenant_id=app_model.tenant_id,
            )

-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow,
-                queue_manager=queue_manager,
-                user=user,
-                draft_var_saver_factory=draft_var_saver_factory,
-                stream=streaming,
-            )
+            try:
+                response = self._handle_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow,
+                    queue_manager=queue_manager,
+                    user=user,
+                    draft_var_saver_factory=draft_var_saver_factory,
+                    stream=streaming,
+                )
+                converted_response = WorkflowAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def single_iteration_generate(
        self,
@@ -39,15 +39,16 @@ class Jinja2TemplateTransformer(TemplateTransformer):
    @override
    def get_runner_script(cls) -> str:
        runner_script = dedent(f"""
-            import jinja2
            import json
            from base64 import b64decode
+            from jinja2.sandbox import SandboxedEnvironment

            # declare main function
            def main(**inputs):
                # Decode base64-encoded template to handle special characters safely
                template_code = b64decode('{cls._template_b64_placeholder}').decode('utf-8')
-                template = jinja2.Template(template_code)
+                env = SandboxedEnvironment()
+                template = env.from_string(template_code)
                return template.render(**inputs)

            # decode and prepare input dict
@@ -67,12 +68,13 @@ class Jinja2TemplateTransformer(TemplateTransformer):
    @override
    def get_preload_script(cls) -> str:
        preload_script = dedent("""
-            import jinja2
+            from jinja2.sandbox import SandboxedEnvironment
            from base64 import b64decode

            def _jinja2_preload_():
-                # prepare jinja2 environment, load template and render before to avoid sandbox issue
-                template = jinja2.Template('{{s}}')
+                # prepare jinja2 sandboxed environment, load template and render
+                env = SandboxedEnvironment()
+                template = env.from_string('{{s}}')
                template.render(s='a')

            if __name__ == '__main__':
@@ -519,7 +519,7 @@ class IndexingRunner:
    def filter_string(text):
        text = re.sub(r"<\|", "<", text)
        text = re.sub(r"\|>", ">", text)
-        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
+        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
        # Unicode  U+FFFE
        text = re.sub("\ufffe", "", text)
        return text
@@ -66,7 +66,7 @@ from services.enterprise.plugin_manager_service import (
    PreUninstallPluginRequest,
 )
 from services.errors.plugin import PluginInstallationForbiddenError
-from services.feature_service import FeatureService, PluginInstallationScope
+from services.feature_service import FeatureService, PluginInstallationPermissionModel, PluginInstallationScope

 logger = logging.getLogger(__name__)
 _provider_entities_adapter: TypeAdapter[list[ProviderEntity]] = TypeAdapter(list[ProviderEntity])
@@ -434,14 +434,18 @@ class PluginService:
                    exc_info=True,
                )

+    @classmethod
+    def _fetch_plugin_model_providers_uncached(
+        cls, tenant_id: str, client: PluginModelClient | None
+    ) -> tuple[ProviderEntity, ...]:
+        model_client = client or PluginModelClient()
+        return tuple(cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id))
+
    @classmethod
    def _fetch_and_cache_plugin_model_providers(
        cls, tenant_id: str, client: PluginModelClient | None, *, refresh_generation: int | None
    ) -> tuple[ProviderEntity, ...]:
-        model_client = client or PluginModelClient()
-        providers = tuple(
-            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
-        )
+        providers = cls._fetch_plugin_model_providers_uncached(tenant_id, client)
        generation = cls._load_plugin_model_providers_generation(tenant_id)
        if generation is not None and generation == refresh_generation:
            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
@@ -471,6 +475,9 @@ class PluginService:
        are intentionally owned by this service so tenant isolation and cache
        expiry are handled in one place.
        """
+        if not dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED:
+            return cls._fetch_plugin_model_providers_uncached(tenant_id, client)
+
        deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT

        while True:
@@ -597,22 +604,30 @@ class PluginService:
            return result

    @staticmethod
-    def _check_marketplace_only_permission():
+    def _check_marketplace_only_permission() -> None:
        """
        Check if the marketplace only permission is enabled
        """
-        features = FeatureService.get_system_features()
-        if features.plugin_installation_permission.restrict_to_marketplace_only:
+        permission = PluginService._get_plugin_installation_permission()
+        if permission.restrict_to_marketplace_only:
            raise PluginInstallationForbiddenError("Plugin installation is restricted to marketplace only")

    @staticmethod
-    def _check_plugin_installation_scope(plugin_verification: PluginVerification | None):
+    def _get_plugin_installation_permission() -> PluginInstallationPermissionModel:
+        """Resolve the validated policy and reject deny-all before any installation side effect."""
+        permission = FeatureService.get_plugin_installation_permission()
+        if permission.plugin_installation_scope == PluginInstallationScope.NONE:
+            raise PluginInstallationForbiddenError("Installing plugins is not allowed")
+        return permission
+
+    @staticmethod
+    def _check_plugin_installation_scope(plugin_verification: PluginVerification | None) -> None:
        """
        Check the plugin installation scope
        """
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

-        match features.plugin_installation_permission.plugin_installation_scope:
+        match permission.plugin_installation_scope:
            case PluginInstallationScope.OFFICIAL_ONLY:
                if (
                    plugin_verification is None
@@ -627,10 +642,10 @@ class PluginService:
                    raise PluginInstallationForbiddenError(
                        "Plugin installation is restricted to official and specific partners"
                    )
-            case PluginInstallationScope.NONE:
-                raise PluginInstallationForbiddenError("Installing plugins is not allowed")
            case PluginInstallationScope.ALL:
                pass
+            case _:
+                raise PluginInstallationForbiddenError("Plugin installation policy is invalid")

    @staticmethod
    def get_debugging_key(tenant_id: str) -> str:
@@ -900,7 +915,7 @@ class PluginService:
        # check if plugin pkg is already downloaded
        manager = PluginInstaller()

-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        try:
            manager.fetch_plugin_manifest(tenant_id, new_plugin_unique_identifier)
@@ -912,7 +927,7 @@ class PluginService:
            response = manager.upload_pkg(
                tenant_id,
                pkg,
-                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                verify_signature=permission.restrict_to_marketplace_only,
            )

            # check if the plugin is available to install
@@ -967,11 +982,11 @@ class PluginService:
        """
        PluginService._check_marketplace_only_permission()
        manager = PluginInstaller()
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()
        response = manager.upload_pkg(
            tenant_id,
            pkg,
-            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            verify_signature=permission.restrict_to_marketplace_only,
        )
        PluginService._check_plugin_installation_scope(response.verification)

@@ -989,13 +1004,13 @@ class PluginService:
        pkg = download_with_size_limit(
            f"https://github.com/{repo}/releases/download/{version}/{package}", dify_config.PLUGIN_MAX_PACKAGE_SIZE
        )
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        manager = PluginInstaller()
        response = manager.upload_pkg(
            tenant_id,
            pkg,
-            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            verify_signature=permission.restrict_to_marketplace_only,
        )
        PluginService._check_plugin_installation_scope(response.verification)

@@ -1069,7 +1084,7 @@ class PluginService:
        if not dify_config.MARKETPLACE_ENABLED:
            raise ValueError("marketplace is not enabled")

-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        manager = PluginInstaller()
        try:
@@ -1079,7 +1094,7 @@ class PluginService:
            response = manager.upload_pkg(
                tenant_id,
                pkg,
-                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                verify_signature=permission.restrict_to_marketplace_only,
            )
            # check if the plugin is available to install
            PluginService._check_plugin_installation_scope(response.verification)
@@ -1101,7 +1116,7 @@ class PluginService:
        # collect actual plugin_unique_identifiers
        actual_plugin_unique_identifiers = []
        metas = []
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        # check if already downloaded
        for plugin_unique_identifier in plugin_unique_identifiers:
@@ -1119,7 +1134,7 @@ class PluginService:
                response = manager.upload_pkg(
                    tenant_id,
                    pkg,
-                    verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                    verify_signature=permission.restrict_to_marketplace_only,
                )
                # check if the plugin is available to install
                PluginService._check_plugin_installation_scope(response.verification)
@@ -9,7 +9,7 @@ class CleanProcessor:
        # remove invalid symbol
        text = re.sub(r"<\|", "<", text)
        text = re.sub(r"\|>", ">", text)
-        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
+        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
        # Unicode  U+FFFE
        text = re.sub("\ufffe", "", text)

@@ -2023,6 +2023,8 @@ class DatasetRetrieval:
            redis_client.zremrangebyscore(key, 0, current_time - 60000)
            request_count = redis_client.zcard(key)
            if request_count > knowledge_rate_limit.limit:
+                # The rate-limit exception is raised after this block, so commit the audit row
+                # explicitly instead of relying on the Session context, which only closes it.
                with session_factory.create_session() as session:
                    rate_limit_log = RateLimitLog(
                        tenant_id=tenant_id,
@@ -2030,6 +2032,7 @@ class DatasetRetrieval:
                        operation="knowledge",
                    )
                    session.add(rate_limit_log)
+                    session.commit()
                raise exc.RateLimitExceededError(
                    "you have reached the knowledge base request rate limit of your subscription."
                )
@@ -91,8 +91,8 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
                splits = re.split(r" +", text)
            else:
                splits = text.split(separator)
-                if self._keep_separator:
-                    splits = [s + separator for s in splits[:-1]] + splits[-1:]
+            if self._keep_separator:
+                splits = [s + separator for s in splits[:-1]] + splits[-1:]
        else:
            splits = list(text)
        if separator == "\n":
@@ -107,6 +107,8 @@ class MCPTool(Tool):
        if self.entity.output_schema and result.structuredContent:
            for k, v in result.structuredContent.items():
                yield self.create_variable_message(k, v)
+        elif result.structuredContent:
+            yield self.create_json_message(result.structuredContent)

    def _process_text_content(self, content: TextContent) -> Generator[ToolInvokeMessage, None, None]:
        """Process text content and yield appropriate messages."""
@@ -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
@@ -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"
    )

@@ -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.
@@ -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:
@@ -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
@@ -497,6 +497,7 @@ class DifyNodeFactory(NodeFactory):
                ),
                "agent_backend_client": create_agent_backend_run_client(
                    base_url=dify_config.AGENT_BACKEND_BASE_URL,
+                    api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                    use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                    fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -368,7 +368,7 @@ class WorkflowAgentRuntimeRequestBuilder:

        if uploaded_files is not None:
            lines.append("- Uploaded workflow files:")
-            lines.append(f"  - sys.files: {uploaded_files}")
+            lines.append(f"  - userinput.files: {uploaded_files}")

        if resolved_outputs:
            lines.append("- Previous node outputs:")
@@ -16,6 +16,7 @@ from .variable_prefixes import (
    ENVIRONMENT_VARIABLE_NODE_ID,
    RAG_PIPELINE_VARIABLE_NODE_ID,
    SYSTEM_VARIABLE_NODE_ID,
+    USER_INPUT_VARIABLE_NODE_ID,
 )


@@ -118,6 +119,12 @@ def build_bootstrap_variables(
        *(_with_selector(variable, ENVIRONMENT_VARIABLE_NODE_ID) for variable in environment_variables),
        *(_with_selector(variable, CONVERSATION_VARIABLE_NODE_ID) for variable in conversation_variables),
    ]
+    # TODO: Stop emitting the legacy `sys.files` selector after stored graphs and Service API callers are migrated.
+    # `userinput.files` remains the canonical file-upload variable.
+    for variable in system_variables:
+        if variable.name == SystemVariableKey.FILES.value:
+            variables.append(_with_selector(variable, USER_INPUT_VARIABLE_NODE_ID))
+            break

    rag_pipeline_variables_map: defaultdict[str, dict[str, Any]] = defaultdict(dict)
    for rag_var in rag_pipeline_variables:
@@ -1,4 +1,5 @@
 SYSTEM_VARIABLE_NODE_ID = "sys"
+USER_INPUT_VARIABLE_NODE_ID = "userinput"
 ENVIRONMENT_VARIABLE_NODE_ID = "env"
 CONVERSATION_VARIABLE_NODE_ID = "conversation"
 RAG_PIPELINE_VARIABLE_NODE_ID = "rag"
@@ -5,6 +5,7 @@ from typing import Any
 import pytz  # type: ignore[import-untyped]
 from celery import Celery, Task
 from celery.schedules import crontab
+from celery.signals import beat_init
 from typing_extensions import TypedDict

 from configs import dify_config
@@ -36,6 +37,19 @@ class CeleryBeatScheduleEntry(TypedDict):
    schedule: crontab | timedelta


+def _enqueue_initial_community_telemetry_heartbeat(sender: Any, **_: Any) -> None:
+    task_name = "community_telemetry.send_heartbeat"
+    if "community_telemetry_heartbeat" not in sender.app.conf.beat_schedule:
+        return
+
+    task = sender.app.tasks.get(task_name)
+    if task is not None:
+        task.apply_async()
+
+
+beat_init.connect(_enqueue_initial_community_telemetry_heartbeat, weak=False)
+
+
 def get_celery_ssl_options() -> CelerySSLOptionsDict | None:
    """Get SSL configuration for Celery broker/backend connections."""
    # Only apply SSL if we're using Redis as broker/backend
@@ -260,6 +274,19 @@ def init_app(app: DifyApp) -> Celery:
            "schedule": timedelta(minutes=dify_config.API_TOKEN_LAST_USED_UPDATE_INTERVAL),
        }

+    if (
+        dify_config.EDITION == "SELF_HOSTED"
+        and not dify_config.ENTERPRISE_ENABLED
+        and not dify_config.DISABLE_TELEMETRY
+        and not dify_config.DO_NOT_TRACK
+        and not dify_config.CI
+    ):
+        imports.append("tasks.community_telemetry_task")
+        beat_schedule["community_telemetry_heartbeat"] = {
+            "task": "community_telemetry.send_heartbeat",
+            "schedule": timedelta(minutes=dify_config.TELEMETRY_HEARTBEAT_INTERVAL_MINUTES),
+        }
+
    if dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED:
        imports.append("tasks.enterprise_telemetry_task")
    celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)
@@ -29,6 +29,7 @@ def init_app(app: DifyApp):
        install_rag_pipeline_plugins,
        migrate_data_for_plugin,
        migrate_dataset_permissions_to_rbac,
+        migrate_legacy_sys_files_workflows,
        migrate_member_roles_to_rbac,
        migrate_oss,
        migration_data_wizard,
@@ -57,6 +58,7 @@ def init_app(app: DifyApp):
        data_migrate,
        upgrade_db,
        fix_app_site_missing,
+        migrate_legacy_sys_files_workflows,
        migrate_data_for_plugin,
        migrate_dataset_permissions_to_rbac,
        migrate_member_roles_to_rbac,
@@ -383,6 +383,7 @@ class AgentAppComposerResponse(ResponseModel):
    variant: Literal[ComposerVariant.AGENT_APP]
    agent: AgentComposerAgentResponse
    active_config_snapshot: AgentConfigSnapshotSummaryResponse | None = None
+    active_config_is_published: bool
    draft: AgentConfigDraftSummaryResponse | None = None
    agent_soul: AgentSoulConfig
    save_options: list[ComposerSaveStrategy]
@@ -0,0 +1,30 @@
+"""add telemetry fields to dify_setups
+
+Revision ID: 6f5a9c2d8e1b
+Revises: d2825e7b9c10
+Create Date: 2026-07-23 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "6f5a9c2d8e1b"
+down_revision = "d2825e7b9c10"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("dify_setups", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("instance_id", sa.String(length=255), nullable=True))
+        batch_op.add_column(sa.Column("install_reported_at", sa.DateTime(), nullable=True))
+        batch_op.add_column(sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True))
+
+
+def downgrade():
+    with op.batch_alter_table("dify_setups", schema=None) as batch_op:
+        batch_op.drop_column("last_heartbeat_at")
+        batch_op.drop_column("install_reported_at")
+        batch_op.drop_column("instance_id")
@@ -362,6 +362,9 @@ class DifySetup(TypeBase):
    __table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)

    version: Mapped[str] = mapped_column(String(255), nullable=False)
+    instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+    install_reported_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
+    last_heartbeat_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
    setup_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
@@ -1114,14 +1117,14 @@ class ExporleBanner(TypeBase):
    status: Mapped[BannerStatus] = mapped_column(
        EnumText(BannerStatus, length=255),
        nullable=False,
-        server_default=sa.text("'enabled'::character varying"),
+        server_default=sa.text("'enabled'"),
        default=BannerStatus.ENABLED,
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    language: Mapped[str] = mapped_column(
-        String(255), nullable=False, server_default=sa.text("'en-US'::character varying"), default="en-US"
+        String(255), nullable=False, server_default=sa.text("'en-US'"), default="en-US"
    )


@@ -25,6 +25,10 @@ from typing_extensions import deprecated

 from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE
 from core.workflow.human_input_adapter import adapt_node_config_for_graph
+from core.workflow.legacy_system_files import (
+    migrate_legacy_sys_files_graph_with_result,
+    serialized_graph_may_contain_legacy_sys_files,
+)
 from core.workflow.nodes.human_input.pause_reason import (
    HumanInputRequired,
 )
@@ -325,7 +329,39 @@ class Workflow(Base):  # bug
        # Currently, the following functions / methods would mutate the returned dict:
        #
        # - `_get_graph_and_variable_pool_for_single_node_run`.
-        return json.loads(self.graph) if self.graph else {}
+        if not self.graph:
+            return {}
+
+        graph = json.loads(self.graph)
+        if not self._supports_legacy_sys_files_compatibility() or not serialized_graph_may_contain_legacy_sys_files(
+            self.graph
+        ):
+            return graph
+
+        # TODO: Remove this load-time compatibility rewrite after all persisted workflows are migrated.
+        return migrate_legacy_sys_files_graph_with_result(graph).graph
+
+    def migrate_legacy_sys_files_graph_in_place(self) -> bool:
+        if (
+            not self.graph
+            or not self._supports_legacy_sys_files_compatibility()
+            or not serialized_graph_may_contain_legacy_sys_files(self.graph)
+        ):
+            return False
+
+        # TODO: Remove this in-place compatibility rewrite after all persisted workflows are migrated.
+        migration_result = migrate_legacy_sys_files_graph_with_result(json.loads(self.graph))
+        if migration_result.changed:
+            self.graph = json.dumps(migration_result.graph)
+        return migration_result.changed
+
+    def _supports_legacy_sys_files_compatibility(self) -> bool:
+        return self.type in {
+            WorkflowType.WORKFLOW,
+            WorkflowType.CHAT,
+            WorkflowType.WORKFLOW.value,
+            WorkflowType.CHAT.value,
+        }

    def get_node_config_by_id(self, node_id: str) -> NodeConfigDict:
        """Extract a node configuration from the workflow graph by node ID.
@@ -487,7 +523,7 @@ class Workflow(Base):  # bug
            "memory":
              {
                "window": { "enabled": false, "size": 10 },
-                "query_prompt_template": "{{#sys.query#}}\n\n{{#sys.files#}}",
+                "query_prompt_template": "{{#sys.query#}}\n\n{{#userinput.files#}}",
                "role_prefix": { "user": "", "assistant": "" },
              },
            "selected": false,
@@ -1520,7 +1556,7 @@ class ConversationVariable(TypeBase):
        return variable_factory.build_conversation_variable_from_mapping(mapping)


-# Only `sys.query` and `sys.files` could be modified.
+# TODO: Remove file-system-variable editability after all persisted workflows are migrated.
 _EDITABLE_SYSTEM_VARIABLE = frozenset(("query", "files"))


@@ -1672,6 +1672,23 @@ Create a new application
 | 200 | Import confirmed | **application/json**: [Import](#import)<br> |
 | 400 | Import failed | **application/json**: [Import](#import)<br> |

+### [GET] /apps/recent
+**Return the lightweight app cards needed by the Explore home page**
+
+Get recently modified apps for the home Continue Work section
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| limit | query | Number of recently modified apps to return (1-8) | No | integer, <br>**Default:** 8 |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [RecentAppListResponse](#recentapplistresponse)<br> |
+
 ### [GET] /apps/starred
 Get applications starred by the current account

@@ -13243,6 +13260,7 @@ Model class for AI model.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
+| active_config_is_published | boolean |  | Yes |
 | active_config_snapshot | [AgentConfigSnapshotSummaryResponse](#agentconfigsnapshotsummaryresponse) |  | No |
 | agent | [AgentComposerAgentResponse](#agentcomposeragentresponse) |  | Yes |
 | agent_soul | [AgentSoulConfig](#agentsoulconfig) |  | Yes |
@@ -13282,7 +13300,6 @@ Model class for AI model.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | access_mode | string |  | No |
-| active_config_is_published | boolean |  | No |
 | api_base_url | string |  | No |
 | app_id | string |  | No |
 | backing_app_id | string |  | No |
@@ -21018,6 +21035,28 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
 | result | string |  | Yes |
 | updated_at | integer |  | Yes |

+#### RecentAppListResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| data | [ [RecentAppResponse](#recentappresponse) ] |  | Yes |
+
+#### RecentAppResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| author_name | string |  | No |
+| icon | string |  | No |
+| icon_background | string |  | No |
+| icon_type | [IconType](#icontype) |  | No |
+| icon_url | string |  | Yes |
+| id | string |  | Yes |
+| maintainer | string |  | No |
+| mode | string, <br>**Available values:** "advanced-chat", "agent-chat", "chat", "completion", "workflow" | *Enum:* `"advanced-chat"`, `"agent-chat"`, `"chat"`, `"completion"`, `"workflow"` | Yes |
+| name | string |  | Yes |
+| permission_keys | [ string ] |  | No |
+| updated_at | integer |  | Yes |
+
 #### RecommendedAppDetailNullableResponse

 | Name | Type | Description | Required |
@@ -965,6 +965,12 @@ Returns Server-Sent Events stream.
 | ---- | ---- | ----------- | -------- |
 | tool_icons | object | Tool icon metadata keyed by tool name | No |

+#### AppMode
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| AppMode | string |  |  |
+
 #### AppPermissionQuery

 | Name | Type | Description | Required |
@@ -1646,6 +1652,7 @@ in form definition, or a variable while the workflow is running.
 | custom_config | [WebAppCustomConfigResponse](#webappcustomconfigresponse) |  | No |
 | enable_site | boolean |  | Yes |
 | end_user_id | string |  | No |
+| mode | [AppMode](#appmode) |  | Yes |
 | model_config | [WebModelConfigResponse](#webmodelconfigresponse) |  | No |
 | plan | string |  | Yes |
 | site | [WebSiteResponse](#websiteresponse) |  | Yes |
@@ -1,12 +1,12 @@
 [project]
 name = "dify-api"
-version = "1.16.0"
+version = "1.16.1"
 requires-python = "~=3.12.0"

 dependencies = [
    # Legacy: mature and widely deployed
    "bleach>=6.4.0,<7.0.0",
-    "boto3>=1.43.46,<2.0.0",
+    "boto3>=1.43.56,<2.0.0",
    "celery>=5.6.3,<6.0.0",
    "croniter>=6.2.2,<7.0.0",
    "dify-agent",
@@ -193,10 +193,10 @@ dev = [
 ############################################################
 storage = [
    "azure-storage-blob>=12.30.0,<13.0.0",
-    "bce-python-sdk==0.9.72",
+    "bce-python-sdk==0.9.76",
    "cos-python-sdk-v5>=1.9.44,<2.0.0",
    "esdk-obs-python>=3.26.6,<4.0.0",
-    "google-cloud-storage>=3.12.1,<4.0.0",
+    "google-cloud-storage>=3.13.0,<4.0.0",
    "opendal==0.46.0",
    "oss2>=2.19.1,<3.0.0",
    "supabase>=2.31.0,<3.0.0",
@@ -75,6 +75,7 @@ from services.errors.account import (
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
+from services.telemetry_service import CommunityTelemetryService
 from tasks.delete_account_task import delete_account_task
 from tasks.mail_account_deletion_task import send_account_deletion_verification_code
 from tasks.mail_change_mail_task import (
@@ -1953,7 +1954,7 @@ class RegisterService:

            TenantService.create_owner_tenant_if_not_exist(account=account, is_setup=True, session=session)

-            dify_setup = DifySetup(version=dify_config.project.version)
+            dify_setup = DifySetup(version=dify_config.project.version, instance_id=str(uuid.uuid4()))
            session.add(dify_setup)
            session.commit()
        except Exception as e:
@@ -1966,6 +1967,11 @@ class RegisterService:
            logger.exception("Setup account failed, email: %s, name: %s", email, name)
            raise ValueError(f"Setup failed: {e}")

+        try:
+            CommunityTelemetryService.report_install(session=session)
+        except Exception:
+            logger.debug("Failed to report install telemetry", exc_info=True)
+
    @classmethod
    def register(
        cls,
@@ -405,6 +405,7 @@ class AgentComposerService:
            "variant": ComposerVariant.AGENT_APP.value,
            "agent": cls._serialize_agent(agent),
            "active_config_snapshot": cls._serialize_version(version),
+            "active_config_is_published": bool(agent.active_config_snapshot_id and agent.active_config_is_published),
            "draft": cls._serialize_draft(draft),
            "agent_soul": draft.config_snapshot_dict,
            "save_options": [ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value],
@@ -488,7 +488,7 @@ class AppGenerateService:
        )

    @classmethod
-    def _get_workflow(
+    def get_workflow(
        cls,
        app_model: App,
        invoke_from: InvokeFrom,
@@ -533,6 +533,17 @@ class AppGenerateService:

        return workflow

+    @classmethod
+    def _get_workflow(
+        cls,
+        app_model: App,
+        invoke_from: InvokeFrom,
+        workflow_id: str | None = None,
+        *,
+        session: Session,
+    ) -> Workflow:
+        return cls.get_workflow(app_model, invoke_from, workflow_id, session=session)
+
    @classmethod
    def get_response_generator(
        cls,
@@ -1,6 +1,7 @@
 import json
 import logging
 from collections.abc import Sequence
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Literal, NotRequired, TypedDict, cast, override

@@ -41,6 +42,20 @@ from tasks.remove_app_and_related_data_task import remove_app_and_related_data_t
 logger = logging.getLogger(__name__)

 AppListSortBy = Literal["last_modified", "recently_created", "earliest_created"]
+RecentAppMode = Literal[
+    AppMode.COMPLETION,
+    AppMode.WORKFLOW,
+    AppMode.CHAT,
+    AppMode.ADVANCED_CHAT,
+    AppMode.AGENT_CHAT,
+]
+RECENT_APP_MODES: tuple[RecentAppMode, ...] = (
+    AppMode.COMPLETION,
+    AppMode.WORKFLOW,
+    AppMode.CHAT,
+    AppMode.ADVANCED_CHAT,
+    AppMode.AGENT_CHAT,
+)


 class AppListBaseParams(BaseModel):
@@ -65,6 +80,19 @@ class StarredAppListParams(AppListBaseParams):
    pass


+@dataclass(frozen=True)
+class RecentAppListItem:
+    id: str
+    name: str
+    icon_type: IconType | None
+    icon: str | None
+    icon_background: str | None
+    mode: RecentAppMode
+    author_name: str | None
+    updated_at: datetime
+    maintainer: str | None
+
+
 class CreateAppParams(BaseModel):
    name: str = Field(min_length=1)
    description: str | None = None
@@ -323,6 +351,62 @@ class AppService:

        return app_models

+    def get_recent_apps(
+        self,
+        user_id: str,
+        tenant_id: str,
+        params: AppListParams,
+        session: Session,
+    ) -> list[RecentAppListItem]:
+        """Return recently modified apps as one lightweight, non-paginated projection."""
+        filters = self._build_app_list_filters(user_id, tenant_id, params, session)
+        if not filters:
+            return []
+
+        stmt = (
+            sa.select(
+                App.id,
+                App.name,
+                App.icon_type,
+                App.icon,
+                App.icon_background,
+                App.mode,
+                Account.name.label("author_name"),
+                App.updated_at,
+                App.maintainer,
+            )
+            .outerjoin(Account, Account.id == App.created_by)
+            .where(*filters, App.mode.in_(RECENT_APP_MODES))
+            .order_by(App.updated_at.desc())
+            .limit(params.limit)
+        )
+        rows = session.execute(stmt).all()
+
+        return [
+            RecentAppListItem(
+                id=str(app_id),
+                name=name,
+                icon_type=icon_type,
+                icon=icon,
+                icon_background=icon_background,
+                mode=cast(RecentAppMode, mode),
+                author_name=author_name,
+                updated_at=updated_at,
+                maintainer=maintainer,
+            )
+            for (
+                app_id,
+                name,
+                icon_type,
+                icon,
+                icon_background,
+                mode,
+                author_name,
+                updated_at,
+                maintainer,
+            ) in rows
+        ]
+
    def get_paginate_starred_apps(
        self,
        user_id: str,
@@ -1,6 +1,8 @@
+import logging
+from collections.abc import Mapping
 from enum import StrEnum

-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, ValidationError

 from configs import dify_config
 from constants.dsl_version import CURRENT_APP_DSL_VERSION
@@ -10,6 +12,8 @@ from enums.hosted_provider import HostedTrialProvider
 from services.billing_service import BillingInfo, BillingService
 from services.enterprise.enterprise_service import EnterpriseService

+logger = logging.getLogger(__name__)
+

 class FeatureResponseModel(BaseModel):
    model_config = ConfigDict(json_schema_serialization_defaults_required=True, protected_namespaces=())
@@ -131,6 +135,13 @@ class PluginInstallationPermissionModel(FeatureResponseModel):
    restrict_to_marketplace_only: bool = False


+class _EnterprisePluginInstallationPermission(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    plugin_installation_scope: PluginInstallationScope = Field(alias="pluginInstallationScope")
+    restrict_to_marketplace_only: bool = Field(alias="restrictToMarketplaceOnly", strict=True)
+
+
 class FeatureModel(FeatureResponseModel):
    billing: BillingModel = BillingModel()
    education: EducationModel = EducationModel()
@@ -285,6 +296,14 @@ class FeatureService:
        """Return whether Enterprise plugin credential policies must be enforced."""
        return dify_config.ENTERPRISE_ENABLED

+    @classmethod
+    def get_plugin_installation_permission(cls) -> PluginInstallationPermissionModel:
+        """Resolve the validated deployment-wide plugin installation policy."""
+        if not dify_config.ENTERPRISE_ENABLED:
+            return PluginInstallationPermissionModel()
+
+        return cls._resolve_plugin_installation_permission(EnterpriseService.get_info())
+
    @classmethod
    def get_license(cls) -> LicenseModel:
        """Return full license detail. Enterprise-only; requires an authenticated caller.
@@ -452,6 +471,33 @@ class FeatureService:
                )
        return license_model

+    @classmethod
+    def _resolve_plugin_installation_permission(
+        cls, enterprise_info: Mapping[str, object]
+    ) -> PluginInstallationPermissionModel:
+        if "PluginInstallationPermission" not in enterprise_info:
+            return PluginInstallationPermissionModel()
+
+        try:
+            permission = _EnterprisePluginInstallationPermission.model_validate(
+                enterprise_info["PluginInstallationPermission"]
+            )
+        except ValidationError as exc:
+            # Do not attach the exception because it may contain raw Enterprise configuration values.
+            logger.error(  # noqa: TRY400
+                "Invalid Enterprise plugin installation permission; denying all plugin installations: %s",
+                exc.errors(include_input=False),
+            )
+            return PluginInstallationPermissionModel(
+                plugin_installation_scope=PluginInstallationScope.NONE,
+                restrict_to_marketplace_only=True,
+            )
+
+        return PluginInstallationPermissionModel(
+            plugin_installation_scope=permission.plugin_installation_scope,
+            restrict_to_marketplace_only=permission.restrict_to_marketplace_only,
+        )
+
    @classmethod
    def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel):
        enterprise_info = EnterpriseService.get_info()
@@ -499,11 +545,4 @@ class FeatureService:
                status=LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
            )

-        if "PluginInstallationPermission" in enterprise_info:
-            plugin_installation_info = enterprise_info["PluginInstallationPermission"]
-            features.plugin_installation_permission.plugin_installation_scope = plugin_installation_info[
-                "pluginInstallationScope"
-            ]
-            features.plugin_installation_permission.restrict_to_marketplace_only = plugin_installation_info[
-                "restrictToMarketplaceOnly"
-            ]
+        features.plugin_installation_permission = cls._resolve_plugin_installation_permission(enterprise_info)
@@ -0,0 +1,165 @@
+import logging
+import platform
+import uuid
+from datetime import datetime
+from typing import Literal
+
+import httpx
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from configs import dify_config
+from libs.datetime_utils import naive_utc_now
+from models.model import DifySetup
+
+logger = logging.getLogger(__name__)
+
+TelemetryEvent = Literal["install", "heartbeat"]
+
+SCHEMA_VERSION = 1
+
+
+class CommunityTelemetryService:
+    @classmethod
+    def report_install(cls, *, session: Session) -> bool:
+        setup = cls._get_setup(session)
+        if setup is None:
+            return False
+
+        if setup.instance_id is None:
+            setup.instance_id = str(uuid.uuid4())
+            session.add(setup)
+            session.commit()
+
+        payload = cls._build_payload(setup, "install")
+        if not cls._send_event(payload):
+            return False
+
+        setup.install_reported_at = naive_utc_now()
+        session.add(setup)
+        session.commit()
+        return True
+
+    @classmethod
+    def report_heartbeat(cls, *, session: Session, now: datetime | None = None) -> bool:
+        setup = cls._get_setup(session)
+        if setup is None:
+            return False
+
+        if setup.instance_id is None:
+            setup.instance_id = str(uuid.uuid4())
+            session.add(setup)
+            session.commit()
+
+        now = now or naive_utc_now()
+        if not cls._is_heartbeat_due(setup, now):
+            return False
+
+        if setup.install_reported_at is None:
+            cls.report_install(session=session)
+
+        payload = cls._build_payload(setup, "heartbeat")
+        if not cls._send_event(payload):
+            return False
+
+        setup.last_heartbeat_at = now
+        session.add(setup)
+        session.commit()
+        return True
+
+    @classmethod
+    def _get_setup(cls, session: Session) -> DifySetup | None:
+        return session.scalar(select(DifySetup).order_by(DifySetup.setup_at.asc()).limit(1))
+
+    @classmethod
+    def _is_enabled(cls) -> bool:
+        return (
+            dify_config.EDITION == "SELF_HOSTED"
+            and not dify_config.ENTERPRISE_ENABLED
+            and not dify_config.DISABLE_TELEMETRY
+            and not dify_config.DO_NOT_TRACK
+            and not dify_config.CI
+            and bool(dify_config.TELEMETRY_ENDPOINT)
+        )
+
+    @classmethod
+    def _build_payload(cls, setup: DifySetup, event: TelemetryEvent) -> dict[str, str | int]:
+        payload: dict[str, str | int] = {
+            "event": event,
+            "instance_id": setup.instance_id or "",
+            "version": setup.version if event == "install" else dify_config.project.version,
+            "edition": dify_config.EDITION,
+            "deployment_type": "unknown",
+            "schema_version": SCHEMA_VERSION,
+            "os": cls._normalize_os(platform.system()),
+            "arch": cls._normalize_arch(platform.machine()),
+            "sent_at": cls._format_datetime(naive_utc_now()),
+        }
+
+        if event == "install":
+            payload["installed_at"] = cls._format_datetime(setup.setup_at)
+
+        return payload
+
+    @classmethod
+    def _send_event(cls, payload: dict[str, str | int]) -> bool:
+        if not cls._is_enabled():
+            return False
+
+        endpoints = [dify_config.TELEMETRY_ENDPOINT]
+        if dify_config.TELEMETRY_FALLBACK_ENDPOINT not in endpoints:
+            endpoints.append(dify_config.TELEMETRY_FALLBACK_ENDPOINT)
+
+        for endpoint in endpoints:
+            if not endpoint:
+                continue
+
+            try:
+                response = httpx.post(
+                    endpoint,
+                    json=payload,
+                    timeout=dify_config.TELEMETRY_TIMEOUT_SECONDS,
+                )
+                response.raise_for_status()
+                return True
+            except httpx.RequestError:
+                logger.debug("Failed to send community telemetry event to %s", endpoint, exc_info=True)
+            except httpx.HTTPStatusError:
+                logger.debug("Community telemetry endpoint returned an error: %s", endpoint, exc_info=True)
+                return False
+
+        return False
+
+    @classmethod
+    def _is_heartbeat_due(cls, setup: DifySetup, now: datetime) -> bool:
+        if setup.instance_id is None:
+            return False
+
+        if setup.last_heartbeat_at is not None and setup.last_heartbeat_at.date() >= now.date():
+            return False
+
+        return True
+
+    @staticmethod
+    def _format_datetime(value: datetime) -> str:
+        return value.replace(microsecond=0).isoformat() + "Z"
+
+    @staticmethod
+    def _normalize_os(value: str) -> str:
+        os_name = value.lower()
+        if os_name in {"linux", "darwin", "windows"}:
+            return os_name
+        return "unknown"
+
+    @staticmethod
+    def _normalize_arch(value: str) -> str:
+        arch = value.lower()
+        if arch in {"x86_64", "amd64"}:
+            return "amd64"
+        if arch in {"aarch64", "arm64"}:
+            return "arm64"
+        if arch.startswith("arm"):
+            return "arm"
+        if arch in {"i386", "i686", "x86"}:
+            return "386"
+        return "unknown"
@@ -278,14 +278,14 @@ class VariableTruncator(BaseTruncator):
        target_length = self._array_element_limit

        for i, item in enumerate(value):
-            # Dirty fix:
-            # The output of `Start` node may contain list of `File` elements,
-            # causing `AssertionError` while invoking `_truncate_json_primitives`.
-            #
-            # This check ensures that `list[File]` are handled separately
-            if isinstance(item, File):
-                truncated_value.append(item)
-                continue
+            # ``File`` is routed through ``_truncate_json_primitives`` (whose
+            # dedicated ``File`` branch returns the file as-is with its real
+            # serialized size). That preserves the count cap
+            # (``array_element_limit``) and the byte budget (``target_size``)
+            # for ``list[File]`` — the original "Dirty fix" branch above this
+            # loop bypassed both guarantees and reported ``used_size=2`` even
+            # when the returned array serialized to well over the budget.
+            # See https://github.com/langgenius/dify/issues/39218.
            if i >= target_length:
                return _PartResult(truncated_value, used_size, True)
            if i > 0:
@@ -295,7 +295,7 @@ class VariableTruncator(BaseTruncator):
                break

            remaining_budget = target_size - used_size
-            if item is None or isinstance(item, (str, list, dict, bool, int, float, UpdatedVariable)):
+            if item is None or isinstance(item, (str, list, dict, bool, int, float, File, UpdatedVariable)):
                part_result = self._truncate_json_primitives(item, remaining_budget)
            else:
                raise UnknownTypeError(f"got unknown type {type(item)} in array truncation")
@@ -116,7 +116,7 @@ class WebAppAuthService:

    @classmethod
    def _get_account_jwt_token(cls, account: Account) -> str:
-        exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES * 24)
+        exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)
        exp = int(exp_dt.timestamp())

        payload = {
@@ -18,6 +18,7 @@ from core.app.apps.completion.app_config_manager import CompletionAppConfigManag
 from core.helper import encrypter
 from core.prompt.simple_prompt_transform import SimplePromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.variable_prefixes import USER_INPUT_VARIABLE_NODE_ID
 from events.app_event import app_was_created
 from graphon.file import FileUploadConfig
 from graphon.model_runtime.entities.llm_entities import LLMMode
@@ -597,7 +598,7 @@ class WorkflowConverter:
                },
                "vision": {
                    "enabled": file_upload is not None,
-                    "variable_selector": ["sys", "files"] if file_upload is not None else None,
+                    "variable_selector": [USER_INPUT_VARIABLE_NODE_ID, "files"] if file_upload is not None else None,
                    "configs": {"detail": file_upload.image_config.detail}
                    if file_upload is not None and file_upload.image_config is not None
                    else None,
@@ -6,8 +6,10 @@ from collections.abc import Callable, Generator, Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, cast

-from sqlalchemy import exists, select
+from sqlalchemy import exists, inspect, select, update
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm.attributes import set_committed_value

 from configs import dify_config
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
@@ -143,6 +145,8 @@ from .human_input_delivery_test_service import (
 from .workflow_draft_variable_service import DraftVariableSaver, DraftVarLoader, WorkflowDraftVariableService
 from .workflow_restore import apply_published_workflow_snapshot_to_draft

+logger = logging.getLogger(__name__)
+
 _file_access_controller = DatabaseFileAccessController()


@@ -155,6 +159,7 @@ class WorkflowService:
        """Initialize WorkflowService with repository dependencies."""
        if session_maker is None:
            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+        self._session_maker = session_maker
        self._node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
            session_maker
        )
@@ -211,7 +216,7 @@ class WorkflowService:
        )

        # return draft workflow
-        return workflow
+        return self._persist_legacy_sys_files_migration_on_load(workflow)

    def get_published_workflow_by_id(self, app_model: App, workflow_id: str, *, session: Session) -> Workflow | None:
        """
@@ -236,6 +241,7 @@ class WorkflowService:
                f"Cannot use draft workflow version. Workflow ID: {workflow_id}. "
                f"Please use a published workflow version or leave workflow_id empty."
            )
+        self._persist_legacy_sys_files_migration_on_load(workflow)
        return workflow

    def get_published_workflow(self, app_model: App, *, session: Session) -> Workflow | None:
@@ -259,6 +265,53 @@ class WorkflowService:
            .limit(1)
        )

+        return self._persist_legacy_sys_files_migration_on_load(workflow)
+
+    def _persist_legacy_sys_files_migration_on_load(self, workflow: Workflow | None) -> Workflow | None:
+        """Persist a load-time graph rewrite without joining or dirtying the caller's transaction."""
+
+        if workflow is None:
+            return None
+        if inspect(workflow, raiseerr=False) is None:
+            return workflow
+
+        # TODO: Remove this load-time persistence path after the historical workflow migration is complete.
+        original_graph = workflow.graph
+        if not workflow.migrate_legacy_sys_files_graph_in_place():
+            return workflow
+
+        migrated_graph = workflow.graph
+        try:
+            with self._session_maker.begin() as session:
+                result = session.execute(
+                    update(Workflow)
+                    .where(
+                        Workflow.id == workflow.id,
+                        Workflow.tenant_id == workflow.tenant_id,
+                        Workflow.graph == original_graph,
+                    )
+                    .values(graph=migrated_graph)
+                )
+                if getattr(result, "rowcount", None) == 0:
+                    logger.warning(
+                        "Skipped persisting legacy sys.files workflow migration because the workflow changed "
+                        "concurrently, "
+                        "workflow_id=%s tenant_id=%s",
+                        workflow.id,
+                        workflow.tenant_id,
+                    )
+        except SQLAlchemyError:
+            logger.warning(
+                "Failed to persist legacy sys.files workflow migration, workflow_id=%s tenant_id=%s",
+                workflow.id,
+                workflow.tenant_id,
+                exc_info=True,
+            )
+        finally:
+            # The conditional update owns persistence. Mark the caller's instance clean so its later flush cannot
+            # overwrite a concurrent workflow edit with the compatibility rewrite.
+            set_committed_value(workflow, "graph", migrated_graph)
+
        return workflow

    def get_accessible_app_ids(self, app_ids: Sequence[str], tenant_id: str, *, session: Session) -> set[str]:
@@ -27,7 +27,6 @@ class WorkspaceService:
        tenant_info: dict[str, object] = {
            "id": tenant.id,
            "name": tenant.name,
-            "plan": tenant.plan,
            "status": tenant.status,
            "created_at": tenant.created_at,
            "trial_end_reason": None,
@@ -44,6 +43,7 @@ class WorkspaceService:
        tenant_info["role"] = tenant_account_join.role

        feature = FeatureService.get_features(tenant.id, exclude_vector_space=True)
+        tenant_info["plan"] = feature.billing.subscription.plan if feature.billing.enabled else None
        can_replace_logo = feature.can_replace_logo

        if can_replace_logo and TenantService.has_roles(
@@ -22,6 +22,7 @@ def _create_agent_backend_client():
        return None
    return create_agent_backend_run_client(
        base_url=dify_config.AGENT_BACKEND_BASE_URL,
+        api_token=dify_config.AGENT_BACKEND_API_TOKEN,
        use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
        fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
        stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -457,7 +457,7 @@ def _publish_streaming_response(
@shared_task(queue=WORKFLOW_BASED_APP_EXECUTION_QUEUE)
 def workflow_based_app_execution_task(
    payload: str,
-) -> Generator[Mapping[str, Any] | str, None, None] | Mapping[str, Any] | None:
+) -> Mapping[str, Any] | None:
    exec_params = AppExecutionParams.model_validate_json(payload)

    logger.info("workflow_based_app_execution_task run with params: %s", exec_params)
@@ -0,0 +1,19 @@
+import logging
+
+from celery import shared_task
+from sqlalchemy.orm import sessionmaker
+
+from extensions.ext_database import db
+from services.telemetry_service import CommunityTelemetryService
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(name="community_telemetry.send_heartbeat", queue="schedule_executor")
+def send_community_telemetry_heartbeat() -> None:
+    session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+    with session_factory() as session:
+        try:
+            CommunityTelemetryService.report_heartbeat(session=session)
+        except Exception:
+            logger.debug("Failed to process community telemetry heartbeat", exc_info=True)
@@ -61,7 +61,7 @@ workflow:
          query_prompt_template: '{{#sys.query#}}


-            {{#sys.files#}}'
+            {{#userinput.files#}}'
          window:
            enabled: false
            size: 10
@@ -162,7 +162,7 @@ workflow:
          query_prompt_template: '{{#sys.query#}}


-            {{#sys.files#}}'
+            {{#userinput.files#}}'
          role_prefix:
            assistant: ''
            user: ''
@@ -207,7 +207,7 @@ workflow:
          query_prompt_template: '{{#sys.query#}}


-            {{#sys.files#}}'
+            {{#userinput.files#}}'
          role_prefix:
            assistant: ''
            user: ''
@@ -178,7 +178,7 @@ workflow:
          query_prompt_template: '{{#sys.query#}}


-            {{#sys.files#}}'
+            {{#userinput.files#}}'
          role_prefix:
            assistant: ''
            user: ''
@@ -17,10 +17,11 @@ class _SimpleJinja2Renderer:
    """Minimal Jinja2-based renderer for integration tests (no code executor)."""

    def render_template(self, template: str, variables: dict[str, object]) -> str:
-        from jinja2 import Template
+        from jinja2.sandbox import SandboxedEnvironment

        try:
-            return Template(template).render(**variables)
+            env = SandboxedEnvironment()
+            return env.from_string(template).render(**variables)
        except Exception as exc:
            raise TemplateRenderError(str(exc)) from exc

@@ -1,494 +1,85 @@
-"""Testcontainers integration tests for controllers.console.datasets.data_source endpoints."""
+"""Integration coverage for Notion page bindings backed by persisted documents."""

-from __future__ import annotations
-
-import inspect
-from collections.abc import Iterator
-from datetime import UTC, datetime
-from unittest.mock import MagicMock, PropertyMock, patch
+from inspect import unwrap
+from unittest.mock import MagicMock, patch
 from uuid import uuid4

-import pytest
 from flask import Flask
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import NotFound

-from controllers.console.datasets import data_source
-from controllers.console.datasets.data_source import (
-    DataSourceApi,
-    DataSourceNotionDatasetSyncApi,
-    DataSourceNotionDocumentSyncApi,
-    DataSourceNotionIndexingEstimateApi,
-    DataSourceNotionListApi,
-    DataSourceNotionPreviewApi,
-)
-from core.rag.index_processor.constant.index_type import IndexStructureType
-from models import Account, DataSourceOauthBinding
+from controllers.console.datasets.data_source import DataSourceNotionListApi
+from models import Account
 from models.dataset import Document
 from models.enums import DataSourceType, DocumentCreatedFrom, IndexingStatus


-@pytest.fixture
-def current_user() -> Account:
-    account = Account(name="Test User", email="u1@example.com")
-    account.id = "u1"
-    return account
+def test_notion_page_is_marked_bound_from_persisted_document(
+    flask_app_with_containers: Flask,
+    db_session_with_containers: Session,
+) -> None:
+    tenant_id = str(uuid4())
+    dataset_id = str(uuid4())
+    account = Account(name="Test User", email="user@example.com")
+    account.id = str(uuid4())
+    document = Document(
+        tenant_id=tenant_id,
+        dataset_id=dataset_id,
+        position=1,
+        data_source_type=DataSourceType.NOTION_IMPORT,
+        data_source_info='{"notion_page_id": "page-1"}',
+        batch=f"batch-{uuid4()}",
+        name="Notion Page",
+        created_from=DocumentCreatedFrom.WEB,
+        created_by=str(uuid4()),
+        indexing_status=IndexingStatus.COMPLETED,
+        enabled=True,
+    )
+    db_session_with_containers.add(document)
+    db_session_with_containers.commit()
+    runtime = MagicMock(
+        get_online_document_pages=lambda **_kwargs: iter(
+            [
+                MagicMock(
+                    result=[
+                        MagicMock(
+                            workspace_id="workspace-1",
+                            workspace_name="Workspace",
+                            workspace_icon=None,
+                            pages=[
+                                MagicMock(
+                                    page_id="page-1",
+                                    page_name="Page",
+                                    type="page",
+                                    parent_id="parent",
+                                    page_icon=None,
+                                )
+                            ],
+                        )
+                    ]
+                )
+            ]
+        ),
+        datasource_provider_type=lambda: None,
+    )

-
-@pytest.fixture
-def mock_engine() -> Iterator[None]:
-    with patch.object(
-        type(data_source.db),
-        "engine",
-        new_callable=PropertyMock,
-        return_value=MagicMock(),
+    with (
+        flask_app_with_containers.test_request_context(f"/?credential_id=c1&dataset_id={dataset_id}"),
+        patch(
+            "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
+            return_value={"token": "token"},
+        ),
+        patch(
+            "controllers.console.datasets.data_source.DatasetService.get_dataset",
+            return_value=MagicMock(data_source_type="notion_import"),
+        ),
+        patch(
+            "core.datasource.datasource_manager.DatasourceManager.get_datasource_runtime",
+            return_value=runtime,
+        ),
    ):
-        yield
-
-
-class TestDataSourceApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_get_success(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.get)
-
-        binding = DataSourceOauthBinding(
-            tenant_id="tenant-1",
-            access_token="token",
-            provider="notion",
-            source_info={
-                "workspace_name": "Workspace",
-                "workspace_id": "workspace-1",
-                "workspace_icon": None,
-                "total": 1,
-                "pages": [
-                    {
-                        "page_id": "page-1",
-                        "page_name": "Page",
-                        "page_icon": {"type": "emoji", "emoji": "P", "url": None},
-                        "parent_id": "parent-1",
-                        "type": "page",
-                    }
-                ],
-            },
-        )
-        binding.id = "b1"
-        binding.created_at = datetime(2026, 5, 25, 1, 2, 3, tzinfo=UTC)
-        binding.disabled = False
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.db.session.scalars",
-                return_value=MagicMock(all=lambda: [binding]),
-            ),
-        ):
-            response, status = method(api, "tenant-1")
-
-        assert status == 200
-        assert response["data"][0] == {
-            "id": "b1",
-            "provider": "notion",
-            "created_at": 1779670923,
-            "is_bound": True,
-            "disabled": False,
-            "source_info": {
-                "workspace_name": "Workspace",
-                "workspace_id": "workspace-1",
-                "workspace_icon": None,
-                "pages": [
-                    {
-                        "page_name": "Page",
-                        "page_id": "page-1",
-                        "page_icon": {"type": "emoji", "url": None, "emoji": "P"},
-                        "parent_id": "parent-1",
-                        "type": "page",
-                    }
-                ],
-                "total": 1,
-            },
-            "link": "http://localhost/console/api/oauth/data-source/notion",
-        }
-
-    def test_get_no_bindings(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.db.session.scalars",
-                return_value=MagicMock(all=lambda: []),
-            ),
-        ):
-            response, status = method(api, "tenant-1")
-
-        assert status == 200
-        assert response["data"] == []
-
-    def test_patch_enable_binding(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.patch)
-
-        binding = MagicMock(id="b1", disabled=True)
-        session = MagicMock()
-        session.scalar.return_value = binding
-
-        with app.test_request_context("/"):
-            response, status = method(api, session, "tenant-1", "b1", "enable")
-
-        assert status == 200
-        assert binding.disabled is False
-
-    def test_patch_disable_binding(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.patch)
-
-        binding = MagicMock(id="b1", disabled=False)
-        session = MagicMock()
-        session.scalar.return_value = binding
-
-        with app.test_request_context("/"):
-            response, status = method(api, session, "tenant-1", "b1", "disable")
-
-        assert status == 200
-        assert binding.disabled is True
-
-    def test_patch_binding_not_found(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.patch)
-        session = MagicMock()
-        session.scalar.return_value = None
-
-        with app.test_request_context("/"):
-            with pytest.raises(NotFound):
-                method(api, session, "tenant-1", "b1", "enable")
-
-    def test_patch_enable_already_enabled(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.patch)
-
-        binding = MagicMock(id="b1", disabled=False)
-        session = MagicMock()
-        session.scalar.return_value = binding
-
-        with app.test_request_context("/"):
-            with pytest.raises(ValueError):
-                method(api, session, "tenant-1", "b1", "enable")
-
-    def test_patch_disable_already_disabled(self, app: Flask) -> None:
-        api = DataSourceApi()
-        method = inspect.unwrap(api.patch)
-
-        binding = MagicMock(id="b1", disabled=True)
-        session = MagicMock()
-        session.scalar.return_value = binding
-
-        with app.test_request_context("/"):
-            with pytest.raises(ValueError):
-                method(api, session, "tenant-1", "b1", "disable")
-
-
-class TestDataSourceNotionListApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_get_credential_not_found(self, app: Flask, current_user: Account) -> None:
-        api = DataSourceNotionListApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/?credential_id=c1"),
-            patch(
-                "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
-                return_value=None,
-            ),
-        ):
-            with pytest.raises(NotFound):
-                method(api, MagicMock(), "tenant-1", current_user)
-
-    def test_get_success_no_dataset_id(self, app: Flask, current_user: Account, mock_engine: None) -> None:
-        api = DataSourceNotionListApi()
-        method = inspect.unwrap(api.get)
-
-        page = MagicMock(
-            page_id="p1",
-            page_name="Page 1",
-            type="page",
-            parent_id="parent",
-            page_icon=None,
+        response, status = unwrap(DataSourceNotionListApi().get)(
+            DataSourceNotionListApi(), db_session_with_containers, tenant_id, account
        )

-        online_document_message = MagicMock(
-            result=[
-                MagicMock(
-                    workspace_id="w1",
-                    workspace_name="My Workspace",
-                    workspace_icon="icon",
-                    pages=[page],
-                )
-            ]
-        )
-
-        with (
-            app.test_request_context("/?credential_id=c1"),
-            patch(
-                "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
-                return_value={"token": "t"},
-            ),
-            patch(
-                "core.datasource.datasource_manager.DatasourceManager.get_datasource_runtime",
-                return_value=MagicMock(
-                    get_online_document_pages=lambda **kw: iter([online_document_message]),
-                    datasource_provider_type=lambda: None,
-                ),
-            ),
-        ):
-            response, status = method(api, MagicMock(), "tenant-1", current_user)
-
-        assert status == 200
-
-    def test_get_success_with_dataset_id(
-        self, app: Flask, current_user: Account, mock_engine: None, db_session_with_containers: Session
-    ) -> None:
-        api = DataSourceNotionListApi()
-        method = inspect.unwrap(api.get)
-        tenant_id = str(uuid4())
-        dataset_id = str(uuid4())
-
-        page = MagicMock(
-            page_id="p1",
-            page_name="Page 1",
-            type="page",
-            parent_id="parent",
-            page_icon=None,
-        )
-
-        online_document_message = MagicMock(
-            result=[
-                MagicMock(
-                    workspace_id="w1",
-                    workspace_name="My Workspace",
-                    workspace_icon="icon",
-                    pages=[page],
-                )
-            ]
-        )
-
-        dataset = MagicMock(data_source_type="notion_import")
-        document = Document(
-            tenant_id=tenant_id,
-            dataset_id=dataset_id,
-            position=1,
-            data_source_type=DataSourceType.NOTION_IMPORT,
-            data_source_info='{"notion_page_id": "p1"}',
-            batch=f"batch-{uuid4()}",
-            name="Notion Page",
-            created_from=DocumentCreatedFrom.WEB,
-            created_by=str(uuid4()),
-            indexing_status=IndexingStatus.COMPLETED,
-            enabled=True,
-        )
-        db_session_with_containers.add(document)
-        db_session_with_containers.commit()
-
-        with (
-            app.test_request_context(f"/?credential_id=c1&dataset_id={dataset_id}"),
-            patch(
-                "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
-                return_value={"token": "t"},
-            ),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=dataset,
-            ),
-            patch(
-                "core.datasource.datasource_manager.DatasourceManager.get_datasource_runtime",
-                return_value=MagicMock(
-                    get_online_document_pages=lambda **kw: iter([online_document_message]),
-                    datasource_provider_type=lambda: None,
-                ),
-            ),
-        ):
-            response, status = method(api, db_session_with_containers, tenant_id, current_user)
-
-        assert status == 200
-
-    def test_get_invalid_dataset_type(self, app: Flask, current_user: Account, mock_engine: None) -> None:
-        api = DataSourceNotionListApi()
-        method = inspect.unwrap(api.get)
-
-        dataset = MagicMock(data_source_type="other_type")
-
-        with (
-            app.test_request_context("/?credential_id=c1&dataset_id=ds1"),
-            patch(
-                "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
-                return_value={"token": "t"},
-            ),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=dataset,
-            ),
-        ):
-            with pytest.raises(ValueError):
-                method(api, MagicMock(), "tenant-1", current_user)
-
-
-class TestDataSourceNotionPreviewApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_get_preview_success(self, app: Flask) -> None:
-        api = DataSourceNotionPreviewApi()
-        method = inspect.unwrap(api.get)
-
-        extractor = MagicMock(extract=lambda: [MagicMock(page_content="hello")])
-
-        with (
-            app.test_request_context("/?credential_id=c1"),
-            patch(
-                "controllers.console.datasets.data_source.DatasourceProviderService.get_datasource_credentials",
-                return_value={"integration_secret": "t"},
-            ),
-            patch(
-                "controllers.console.datasets.data_source.NotionExtractor",
-                return_value=extractor,
-            ),
-        ):
-            response, status = method(api, "tenant-1", "p1", "page")
-
-        assert status == 200
-
-
-class TestDataSourceNotionIndexingEstimateApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_post_indexing_estimate_success(self, app: Flask) -> None:
-        api = DataSourceNotionIndexingEstimateApi()
-        method = inspect.unwrap(api.post)
-
-        empty_rules: dict[str, object] = {}
-        payload: dict[str, object] = {
-            "notion_info_list": [
-                {
-                    "workspace_id": "w1",
-                    "credential_id": "c1",
-                    "pages": [{"page_id": "p1", "type": "page"}],
-                }
-            ],
-            "process_rule": {"rules": empty_rules},
-            "doc_form": IndexStructureType.PARAGRAPH_INDEX,
-            "doc_language": "English",
-        }
-
-        with (
-            app.test_request_context("/", method="POST", json=payload, headers={"Content-Type": "application/json"}),
-            patch(
-                "controllers.console.datasets.data_source.DocumentService.estimate_args_validate",
-            ),
-            patch(
-                "controllers.console.datasets.data_source.IndexingRunner.indexing_estimate",
-                return_value=MagicMock(model_dump=lambda: {"total_pages": 1}),
-            ),
-        ):
-            response, status = method(api, MagicMock(), "tenant-1")
-
-        assert status == 200
-
-
-class TestDataSourceNotionDatasetSyncApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_get_success(self, app: Flask) -> None:
-        api = DataSourceNotionDatasetSyncApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=MagicMock(),
-            ),
-            patch(
-                "controllers.console.datasets.data_source.DocumentService.get_document_by_dataset_id",
-                return_value=[MagicMock(id="d1")],
-            ),
-            patch(
-                "controllers.console.datasets.data_source.document_indexing_sync_task.delay",
-                return_value=None,
-            ),
-        ):
-            response, status = method(api, MagicMock(), "ds-1")
-
-        assert status == 200
-
-    def test_get_dataset_not_found(self, app: Flask) -> None:
-        api = DataSourceNotionDatasetSyncApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=None,
-            ),
-        ):
-            with pytest.raises(NotFound):
-                method(api, MagicMock(), "ds-1")
-
-
-class TestDataSourceNotionDocumentSyncApi:
-    @pytest.fixture
-    def app(self, flask_app_with_containers: Flask) -> Flask:
-        return flask_app_with_containers
-
-    def test_get_success(self, app: Flask) -> None:
-        api = DataSourceNotionDocumentSyncApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=MagicMock(),
-            ),
-            patch(
-                "controllers.console.datasets.data_source.DocumentService.get_document",
-                return_value=MagicMock(),
-            ),
-            patch(
-                "controllers.console.datasets.data_source.document_indexing_sync_task.delay",
-                return_value=None,
-            ),
-        ):
-            response, status = method(api, MagicMock(), "ds-1", "doc-1")
-
-        assert status == 200
-
-    def test_get_document_not_found(self, app: Flask) -> None:
-        api = DataSourceNotionDocumentSyncApi()
-        method = inspect.unwrap(api.get)
-
-        with (
-            app.test_request_context("/"),
-            patch(
-                "controllers.console.datasets.data_source.DatasetService.get_dataset",
-                return_value=MagicMock(),
-            ),
-            patch(
-                "controllers.console.datasets.data_source.DocumentService.get_document",
-                return_value=None,
-            ),
-        ):
-            with pytest.raises(NotFound):
-                method(api, MagicMock(), "ds-1", "doc-1")
+    assert status == 200
+    assert response["notion_info"][0]["pages"][0]["is_bound"] is True
@@ -97,6 +97,7 @@ class TestAppSiteApi:
        assert result["end_user_id"] == end_user.id
        assert result["plan"] == "basic"
        assert result["enable_site"] is True
+        assert result["mode"] == AppMode.CHAT

    @patch("controllers.web.site.FileService.get_file_presigned_url")
    @patch("controllers.web.site.FeatureService.get_features")
@@ -178,6 +179,7 @@ class TestWebAppSiteResponse:
        response = WebAppSiteResponse.from_app_site(
            tenant=tenant,
            app_model=app_model,
+            mode=AppMode.CHAT,
            site=_site_model(app_id=app_model.id),
            end_user_id="eu-1",
            features=FeatureModel(can_replace_logo=False, webapp_copyright_enabled=True),
@@ -185,6 +187,7 @@ class TestWebAppSiteResponse:
        )

        assert response.app_id == app_model.id
+        assert response.mode == AppMode.CHAT
        assert response.end_user_id == "eu-1"
        assert response.enable_site is True
        assert response.plan == "basic"
@@ -209,6 +212,7 @@ class TestWebAppSiteResponse:
        response = WebAppSiteResponse.from_app_site(
            tenant=tenant,
            app_model=app_model,
+            mode=AppMode.CHAT,
            site=site,
            end_user_id=None,
            features=FeatureModel(can_replace_logo=False, webapp_copyright_enabled=True),
@@ -236,6 +240,7 @@ class TestWebAppSiteResponse:
        response = WebAppSiteResponse.from_app_site(
            tenant=tenant,
            app_model=app_model,
+            mode=AppMode.CHAT,
            site=_site_model(app_id=app_model.id),
            end_user_id="eu-1",
            features=FeatureModel(can_replace_logo=True, webapp_copyright_enabled=True),
@@ -4,7 +4,7 @@ import datetime
 import json
 import uuid
 from decimal import Decimal
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch

 import pytest
 from faker import Faker
@@ -1172,65 +1172,8 @@ class TestMessagesCleanServiceIntegration:
        # Verify all messages were deleted
        assert db_session_with_containers.query(Message).where(Message.id.in_(msg_ids)).count() == 0

-    def test_from_time_range_validation(self):
-        """Test that from_time_range raises ValueError for invalid inputs."""
-        policy = MagicMock(spec=BillingDisabledPolicy)
-        now = datetime.datetime.now()
-
-        with pytest.raises(ValueError, match="start_from .* must be less than end_before"):
-            MessagesCleanService.from_time_range(policy, now, now)
-
-        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
-            MessagesCleanService.from_time_range(policy, now - datetime.timedelta(days=1), now, batch_size=0)
-
-    def test_from_time_range_success(self):
-        """Test that from_time_range creates a service with correct parameters."""
-        policy = MagicMock(spec=BillingDisabledPolicy)
-        start = datetime.datetime(2024, 1, 1)
-        end = datetime.datetime(2024, 2, 1)
-
-        service = MessagesCleanService.from_time_range(policy, start, end)
-        assert service._start_from == start
-        assert service._end_before == end
-
-    def test_from_days_validation(self):
-        """Test that from_days raises ValueError for invalid inputs."""
-        policy = MagicMock(spec=BillingDisabledPolicy)
-
-        with pytest.raises(ValueError, match="days .* must be greater than or equal to 0"):
-            MessagesCleanService.from_days(policy, days=-1)
-
-        with pytest.raises(ValueError, match="batch_size .* must be greater than 0"):
-            MessagesCleanService.from_days(policy, days=30, batch_size=0)
-
-    def test_from_days_success(self):
-        """Test that from_days creates a service with correct parameters."""
-        policy = MagicMock(spec=BillingDisabledPolicy)
-
-        with patch("services.retention.conversation.messages_clean_service.naive_utc_now") as mock_now:
-            fixed_now = datetime.datetime(2024, 6, 1)
-            mock_now.return_value = fixed_now
-
-            service = MessagesCleanService.from_days(policy, days=10)
-            assert service._start_from is None
-            assert service._end_before == fixed_now - datetime.timedelta(days=10)
-
    def test_batch_delete_message_relations_empty(self, db_session_with_containers: Session):
        """Test that batch_delete_message_relations with empty list does nothing."""
        # Get execute call count before
        MessagesCleanService._batch_delete_message_relations(db_session_with_containers, [])
        # No exception means success — empty list is a no-op
-
-    def test_run_calls_clean_messages(self):
-        """Test that run() delegates to _clean_messages_by_time_range."""
-        policy = MagicMock(spec=BillingDisabledPolicy)
-        service = MessagesCleanService(
-            policy=policy,
-            end_before=datetime.datetime.now(),
-            batch_size=10,
-        )
-        with patch.object(service, "_clean_messages_by_time_range") as mock_clean:
-            mock_clean.return_value = {"total_deleted": 5}
-            result = service.run()
-            assert result == {"total_deleted": 5}
-            mock_clean.assert_called_once()
@@ -24,7 +24,10 @@ class TestWorkspaceService:
            patch("services.workspace_service.dify_config") as mock_dify_config,
        ):
            # Setup default mock returns
-            mock_feature_service.get_features.return_value.can_replace_logo = True
+            feature = mock_feature_service.get_features.return_value
+            feature.can_replace_logo = True
+            feature.billing.enabled = True
+            feature.billing.subscription.plan = "professional"
            mock_tenant_service.has_roles.return_value = True
            mock_dify_config.FILES_URL = "https://example.com/files"

@@ -112,7 +115,7 @@ class TestWorkspaceService:
            assert result is not None
            assert result["id"] == tenant.id
            assert result["name"] == tenant.name
-            assert result["plan"] == tenant.plan
+            assert result["plan"] == "professional"
            assert result["status"] == tenant.status
            assert result["role"] == TenantAccountRole.OWNER
            assert result["created_at"] == tenant.created_at
@@ -159,7 +162,7 @@ class TestWorkspaceService:
            assert result is not None
            assert result["id"] == tenant.id
            assert result["name"] == tenant.name
-            assert result["plan"] == tenant.plan
+            assert result["plan"] == "professional"
            assert result["status"] == tenant.status
            assert result["role"] == TenantAccountRole.OWNER
            assert result["created_at"] == tenant.created_at
@@ -214,7 +217,7 @@ class TestWorkspaceService:
            assert result is not None
            assert result["id"] == tenant.id
            assert result["name"] == tenant.name
-            assert result["plan"] == tenant.plan
+            assert result["plan"] == "professional"
            assert result["status"] == tenant.status
            assert result["role"] == TenantAccountRole.NORMAL
            assert result["created_at"] == tenant.created_at
@@ -606,20 +609,23 @@ class TestWorkspaceService:
    def test_get_tenant_info_should_not_include_cloud_fields_in_self_hosted(
        self, db_session_with_containers: Session, mock_external_service_dependencies
    ):
-        """next_credit_reset_date and trial_credits should NOT appear in SELF_HOSTED mode."""
+        """Cloud-only billing data should not appear in SELF_HOSTED mode."""
        fake = Faker()
        account, tenant = self._create_test_account_and_tenant(
            db_session_with_containers, mock_external_service_dependencies
        )

        mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
-        mock_external_service_dependencies["feature_service"].get_features.return_value.can_replace_logo = False
+        feature = mock_external_service_dependencies["feature_service"].get_features.return_value
+        feature.can_replace_logo = False
+        feature.billing.enabled = False
        mock_external_service_dependencies["tenant_service"].has_roles.return_value = False

        with patch("services.workspace_service.current_user", account):
            result = WorkspaceService.get_tenant_info(tenant, db_session_with_containers)

        assert result is not None
+        assert result["plan"] is None
        assert "next_credit_reset_date" not in result
        assert "trial_credits" not in result
        assert "trial_credits_used" not in result
@@ -4,7 +4,7 @@ from unittest.mock import MagicMock
 import pytest
 from sqlalchemy.orm import Session

-from commands import system as system_commands
+from commands import app_maintenance as app_maintenance_commands


 def test_fix_app_site_missing_passes_loaded_session_to_signal(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -30,15 +30,15 @@ def test_fix_app_site_missing_passes_loaded_session_to_signal(monkeypatch: pytes
    engine = MagicMock()
    engine.begin.return_value.__enter__.return_value = connection

-    monkeypatch.setattr(system_commands, "db", SimpleNamespace(engine=engine, session=scoped_session))
+    monkeypatch.setattr(app_maintenance_commands, "db", SimpleNamespace(engine=engine, session=scoped_session))
    send = MagicMock(side_effect=lambda *_args, **_kwargs: phase_events.append("signal"))
-    monkeypatch.setattr(system_commands.app_was_created, "send", send)
+    monkeypatch.setattr(app_maintenance_commands.app_was_created, "send", send)

-    system_commands.fix_app_site_missing.callback()
+    app_maintenance_commands.fix_app_site_missing.callback()

    scoped_session.assert_called_once_with()
    scalar.assert_called_once()
-    get.assert_called_once_with(system_commands.Tenant, app.tenant_id)
+    get.assert_called_once_with(app_maintenance_commands.Tenant, app.tenant_id)
    tenant.get_accounts.assert_called_once_with(session=session)
    send.assert_called_once_with(app, account=account, session=session)
    commit.assert_called_once_with()
@@ -62,15 +62,19 @@ def test_fix_app_site_missing_rolls_back_when_signal_fails(monkeypatch: pytest.M
    engine = MagicMock()
    engine.begin.return_value.__enter__.return_value = connection

-    monkeypatch.setattr(system_commands, "db", SimpleNamespace(engine=engine, session=MagicMock(return_value=session)))
+    monkeypatch.setattr(
+        app_maintenance_commands,
+        "db",
+        SimpleNamespace(engine=engine, session=MagicMock(return_value=session)),
+    )

    def fail_signal(*_args, **_kwargs) -> None:
        phase_events.append("signal")
        raise RuntimeError("failed")

-    monkeypatch.setattr(system_commands.app_was_created, "send", MagicMock(side_effect=fail_signal))
+    monkeypatch.setattr(app_maintenance_commands.app_was_created, "send", MagicMock(side_effect=fail_signal))

-    system_commands.fix_app_site_missing.callback()
+    app_maintenance_commands.fix_app_site_missing.callback()

    session.rollback.assert_called_once_with()
    session.commit.assert_not_called()
@@ -6,6 +6,7 @@ import json
 import os
 import threading
 import time
+from collections.abc import Iterator
 from datetime import datetime, timedelta
 from pathlib import Path
 from types import SimpleNamespace
@@ -15,9 +16,12 @@ import pytest
 import sqlalchemy as sa
 from click.testing import CliRunner
 from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session, SessionTransaction, sessionmaker

 from graphon.model_runtime.entities.model_entities import ModelType
+from models import Dataset, DatasetPermission, DatasetPermissionEnum
 from models.account import Tenant
+from models.base import TypeBase
 from models.enums import CredentialSourceType
 from models.provider import ProviderModel
 from tests.helpers.legacy_model_type_migration import (
@@ -59,6 +63,40 @@ def command_module():
        )


+@pytest.fixture
+def rbac_session(sqlite_engine: sa.Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Session]:
+    """Bind RBAC command reads to persisted SQLite dataset rows."""
+
+    TypeBase.metadata.create_all(
+        sqlite_engine,
+        tables=[Dataset.__table__, DatasetPermission.__table__],
+    )
+    factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr("commands.rbac.session_factory.create_session", factory)
+    with factory() as session:
+        yield session
+
+
+def _persist_dataset(
+    session: Session,
+    *,
+    dataset_id: str = "dataset-1",
+    tenant_id: str = "tenant-1",
+    permission: DatasetPermissionEnum = DatasetPermissionEnum.ONLY_ME,
+    created_by: str = "creator-account-1",
+) -> Dataset:
+    dataset = Dataset(
+        id=dataset_id,
+        tenant_id=tenant_id,
+        name=f"Dataset {dataset_id}",
+        permission=permission,
+        created_by=created_by,
+    )
+    session.add(dataset)
+    session.commit()
+    return dataset
+
+
 def _parse_json_lines(output: io.StringIO) -> list[dict[str, object]]:
    return [json.loads(line) for line in output.getvalue().splitlines() if line.strip()]

@@ -363,56 +401,35 @@ def test_dataset_permission_rbac_migration_maps_legacy_permissions_to_enum_scope

 def test_dataset_permission_rbac_migration_uses_dataset_creator_as_operator(
    command_module,
+    rbac_session: Session,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    rbac_module = importlib.import_module("commands.rbac")
-    dataset_row = SimpleNamespace(
-        id="dataset-1",
-        tenant_id="tenant-1",
-        permission="only_me",
-        created_by="creator-account-1",
-    )
-    execute_results = [[dataset_row], [], []]
+    _persist_dataset(rbac_session)
    calls: list[dict[str, object]] = []
-    session_closed = False
-
-    class FakeExecuteResult:
-        def __init__(self, rows: list[object]) -> None:
-            self._rows = rows
-
-        def all(self) -> list[object]:
-            return self._rows
-
-    class FakeSession:
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, traceback) -> None:
-            nonlocal session_closed
-            session_closed = True
-            pass
-
-        def execute(self, stmt):
-            return FakeExecuteResult(execute_results.pop(0))
-
-    class FakeSessionFactory:
-        @staticmethod
-        def create_session() -> FakeSession:
-            return FakeSession()
+    read_transaction_ended = False

    def fake_replace_whitelist(**kwargs):
-        assert session_closed is True
+        assert read_transaction_ended is True
        calls.append(kwargs)

-    monkeypatch.setattr(rbac_module, "session_factory", FakeSessionFactory)
-    monkeypatch.setattr(rbac_module.RBACService.DatasetAccess, "replace_whitelist", fake_replace_whitelist)
+    def _record_transaction_end(session: Session, transaction: object) -> None:
+        nonlocal read_transaction_ended
+        del transaction
+        if session.get_bind() is rbac_session.get_bind():
+            read_transaction_ended = True

-    command_module.migrate_dataset_permissions_to_rbac.callback(
-        tenant_id=None,
-        dataset_id=None,
-        batch_size=500,
-        dry_run=False,
-    )
+    sa.event.listen(Session, "after_transaction_end", _record_transaction_end)
+    monkeypatch.setattr(rbac_module.RBACService.DatasetAccess, "replace_whitelist", fake_replace_whitelist)
+    try:
+        command_module.migrate_dataset_permissions_to_rbac.callback(
+            tenant_id=None,
+            dataset_id=None,
+            batch_size=500,
+            dry_run=False,
+        )
+    finally:
+        sa.event.remove(Session, "after_transaction_end", _record_transaction_end)

    assert calls[0]["tenant_id"] == "tenant-1"
    assert calls[0]["account_id"] == "creator-account-1"
@@ -422,41 +439,19 @@ def test_dataset_permission_rbac_migration_uses_dataset_creator_as_operator(

 def test_dataset_permission_rbac_migration_dry_run_outputs_structured_proposed_changes(
    command_module,
+    rbac_session: Session,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    rbac_module = importlib.import_module("commands.rbac")
-    dataset_row = SimpleNamespace(
-        id="dataset-1",
-        tenant_id="tenant-1",
-        permission="partial_members",
-        created_by="creator-account-1",
+    dataset = _persist_dataset(rbac_session, permission=DatasetPermissionEnum.PARTIAL_TEAM)
+    rbac_session.add(
+        DatasetPermission(
+            dataset_id=dataset.id,
+            account_id="member-account-1",
+            tenant_id=dataset.tenant_id,
+        )
    )
-    permission_row = SimpleNamespace(dataset_id="dataset-1", account_id="member-account-1")
-    execute_results = [[dataset_row], [permission_row], []]
-
-    class FakeExecuteResult:
-        def __init__(self, rows: list[object]) -> None:
-            self._rows = rows
-
-        def all(self) -> list[object]:
-            return self._rows
-
-    class FakeSession:
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, traceback) -> None:
-            pass
-
-        def execute(self, stmt):
-            return FakeExecuteResult(execute_results.pop(0))
-
-    class FakeSessionFactory:
-        @staticmethod
-        def create_session() -> FakeSession:
-            return FakeSession()
-
-    monkeypatch.setattr(rbac_module, "session_factory", FakeSessionFactory)
+    rbac_session.commit()
    monkeypatch.setattr(
        rbac_module.RBACService.DatasetAccess,
        "replace_whitelist",
@@ -1306,50 +1301,36 @@ def test_provider_models_processing_uses_same_plan_locking_and_transaction_entry
    begin_calls: list[str] = []
    configure_calls: list[str] = []

-    class _FakeBeginContext:
-        def __init__(self, phase: str) -> None:
-            self._phase = phase
+    def _record_begin(session: Session, transaction: SessionTransaction) -> None:
+        if session.get_bind() is sqlite_engine and transaction.parent is None:
+            begin_calls.append(current_phase["name"])

-        def __enter__(self) -> None:
-            begin_calls.append(self._phase)
-
-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-    class _FakeSession:
-        def __init__(self, phase: str) -> None:
-            self._phase = phase
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-        def begin(self) -> _FakeBeginContext:
-            return _FakeBeginContext(self._phase)
-
-    def _fake_session_factory(engine: sa.Engine) -> _FakeSession:
-        return _FakeSession(current_phase["name"])
-
-    def _fake_build_plan(self, session, candidate, *, lock_rows: bool):
+    def _fake_build_plan(self, session: Session, candidate, *, lock_rows: bool):
+        assert session.get_bind() is sqlite_engine
        lock_rows_seen.append((current_phase["name"], lock_rows))
-        return SimpleNamespace(group_row_ids=[str(candidate.row.id)], winner=None, loser_rows=[])
+        return migration_module._ProviderModelGroupPlan(
+            group_row_ids=[str(candidate.row.id)],
+            winner=None,
+            loser_rows=[],
+        )

    def _fake_emit_plan(self, plan, *, session, tx_id: str, business_key: dict[str, object]) -> None:
        return None

-    def _fake_configure(self, session) -> None:
+    def _fake_configure(self, session: Session) -> None:
+        assert session.get_bind() is sqlite_engine
        configure_calls.append(current_phase["name"])

-    monkeypatch.setattr(migration_module, "_session_factory", _fake_session_factory)
    monkeypatch.setattr(migration_module.Migration, "_build_provider_model_group_plan", _fake_build_plan)
    monkeypatch.setattr(migration_module.Migration, "_emit_provider_model_group_plan", _fake_emit_plan)
    monkeypatch.setattr(migration_module.Migration, "_configure_lock_timeout", _fake_configure)
-
-    dry_migration._process_provider_model_group(candidate, business_key)
-    current_phase["name"] = "apply"
-    apply_migration._process_provider_model_group(candidate, business_key)
+    sa.event.listen(Session, "after_transaction_create", _record_begin)
+    try:
+        dry_migration._process_provider_model_group(candidate, business_key)
+        current_phase["name"] = "apply"
+        apply_migration._process_provider_model_group(candidate, business_key)
+    finally:
+        sa.event.remove(Session, "after_transaction_create", _record_begin)

    assert [phase for phase, _ in lock_rows_seen] == ["dry", "apply"]
    assert lock_rows_seen[0][1] == lock_rows_seen[1][1]
@@ -1392,6 +1373,22 @@ def test_process_load_balancing_model_config_row_logs_stacktrace_for_lock_timeou
    sqlite_engine: sa.Engine,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    create_minimal_legacy_model_type_schema(sqlite_engine)
+    created_at = datetime(2025, 1, 1, 12, 0, 0)
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id="40000000-0000-0000-0000-000000000001",
+        tenant_id="tenant-1",
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        name="credential",
+        encrypted_config="{}",
+        credential_id="50000000-0000-0000-0000-000000000001",
+        enabled=True,
+        created_at=created_at,
+        updated_at=created_at,
+    )
    output = io.StringIO()
    migration = migration_module.Migration(
        tenant_id="tenant-1",
@@ -1401,37 +1398,18 @@ def test_process_load_balancing_model_config_row_logs_stacktrace_for_lock_timeou
        model_types=(ModelType.LLM,),
        orm_models=(migration_module.LoadBalancingModelConfig,),
    )
-    candidate = migration_module._RowWithRawModelType(
-        row=SimpleNamespace(id="lb-row-1"),
-        raw_model_type="text-generation",
-        canonical_model_type=ModelType.LLM,
-    )
+    candidate = migration._load_load_balancing_model_config_candidates(None)[0]
    lock_timeout_exc = OperationalError("SELECT 1", {}, SimpleNamespace(pgcode="55P03"))
+    transaction_begins = 0

-    class _FakeBeginContext:
-        def __enter__(self) -> None:
-            return None
-
-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-    class _FakeSession:
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-        def begin(self) -> _FakeBeginContext:
-            return _FakeBeginContext()
-
-    def _fake_session_factory(engine: sa.Engine) -> _FakeSession:
-        return _FakeSession()
+    def _record_begin(session: Session, transaction: SessionTransaction) -> None:
+        nonlocal transaction_begins
+        if session.get_bind() is sqlite_engine and transaction.parent is None:
+            transaction_begins += 1

    def _fake_reload(self, session, original_candidate, *, lock_rows: bool):
        raise lock_timeout_exc

-    monkeypatch.setattr(migration_module, "_session_factory", _fake_session_factory)
    monkeypatch.setattr(migration_module.Migration, "_configure_lock_timeout", lambda self, session: None)
    monkeypatch.setattr(
        migration_module.Migration,
@@ -1439,17 +1417,22 @@ def test_process_load_balancing_model_config_row_logs_stacktrace_for_lock_timeou
        _fake_reload,
    )

-    migration._process_load_balancing_model_config_row(candidate)
+    sa.event.listen(Session, "after_transaction_create", _record_begin)
+    try:
+        migration._process_load_balancing_model_config_row(candidate)
+    finally:
+        sa.event.remove(Session, "after_transaction_create", _record_begin)

    lines = _parse_json_lines(output)
    assert len(lines) == 1
    assert lines[0]["event"] == "lock_timeout_skipped"
    attrs = cast(dict[str, object], lines[0]["attrs"])
    assert attrs["table_name"] == "load_balancing_model_configs"
-    assert attrs["id"] == "lb-row-1"
+    assert attrs["id"] == str(candidate.row.id)
    assert attrs["error"] == str(lock_timeout_exc)
    assert isinstance(attrs["stacktrace"], str)
    assert "OperationalError" in attrs["stacktrace"]
+    assert transaction_begins == 1


 def test_process_load_balancing_model_config_row_logs_update_after_sql_execution(
@@ -1457,6 +1440,23 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
    sqlite_engine: sa.Engine,
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    create_minimal_legacy_model_type_schema(sqlite_engine)
+    created_at = datetime(2025, 1, 1, 12, 0, 0)
+    row_id = "40000000-0000-0000-0000-000000000002"
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id=row_id,
+        tenant_id="tenant-1",
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        name="credential",
+        encrypted_config="{}",
+        credential_id="50000000-0000-0000-0000-000000000002",
+        enabled=True,
+        created_at=created_at,
+        updated_at=created_at,
+    )
    migration = migration_module.Migration(
        tenant_id="tenant-1",
        engine=sqlite_engine,
@@ -1465,42 +1465,33 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
        model_types=(ModelType.LLM,),
        orm_models=(migration_module.LoadBalancingModelConfig,),
    )
-    candidate = migration_module._RowWithRawModelType(
-        row=SimpleNamespace(id="lb-row-1"),
-        raw_model_type="text-generation",
-        canonical_model_type=ModelType.LLM,
-    )
+    candidate = migration._load_load_balancing_model_config_candidates(None)[0]
    action_log: list[str] = []

-    class _FakeBeginContext:
-        def __enter__(self) -> None:
+    def _record_begin(session: Session, transaction: SessionTransaction) -> None:
+        if session.get_bind() is sqlite_engine and transaction.parent is None:
            action_log.append("begin")

-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-    class _FakeSession:
-        def __enter__(self):
-            return self
-
-        def __exit__(self, exc_type, exc, tb) -> bool:
-            return False
-
-        def begin(self) -> _FakeBeginContext:
-            return _FakeBeginContext()
-
-        def execute(self, stmt) -> None:
+    def _record_sql(
+        connection: sa.Connection,
+        cursor: object,
+        statement: str,
+        parameters: object,
+        context: object,
+        executemany: bool,
+    ) -> None:
+        del connection, cursor, parameters, context, executemany
+        if statement.lstrip().upper().startswith("UPDATE"):
            action_log.append("sql_execute")

-    def _fake_session_factory(engine: sa.Engine) -> _FakeSession:
-        return _FakeSession()
-
    def _fake_configure(self, session) -> None:
        action_log.append("configure_lock_timeout")

-    def _fake_reload(self, session, original_candidate, *, lock_rows: bool):
+    original_reload = migration_module.Migration._reload_load_balancing_model_config_candidate
+
+    def _record_reload(self, session: Session, original_candidate, *, lock_rows: bool):
        action_log.append(f"reload_candidate:{lock_rows}")
-        return candidate
+        return original_reload(self, session, original_candidate, lock_rows=lock_rows)

    def _fake_log_row_updated(self, *args, **kwargs) -> None:
        action_log.append("log_row_updated")
@@ -1508,12 +1499,11 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
    def _fake_cache_cleanup(self, *, row_id: str, tx_id: str) -> None:
        action_log.append("cache_cleanup")

-    monkeypatch.setattr(migration_module, "_session_factory", _fake_session_factory)
    monkeypatch.setattr(migration_module.Migration, "_configure_lock_timeout", _fake_configure)
    monkeypatch.setattr(
        migration_module.Migration,
        "_reload_load_balancing_model_config_candidate",
-        _fake_reload,
+        _record_reload,
    )
    monkeypatch.setattr(migration_module.Migration, "_log_row_updated", _fake_log_row_updated)
    monkeypatch.setattr(
@@ -1522,7 +1512,13 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
        _fake_cache_cleanup,
    )

-    migration._process_load_balancing_model_config_row(candidate)
+    sa.event.listen(Session, "after_transaction_create", _record_begin)
+    sa.event.listen(sqlite_engine, "before_cursor_execute", _record_sql)
+    try:
+        migration._process_load_balancing_model_config_row(candidate)
+    finally:
+        sa.event.remove(sqlite_engine, "before_cursor_execute", _record_sql)
+        sa.event.remove(Session, "after_transaction_create", _record_begin)

    assert action_log == [
        "begin",
@@ -1532,6 +1528,10 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
        "log_row_updated",
        "cache_cleanup",
    ]
+    with Session(sqlite_engine) as session:
+        persisted = session.get(migration_module.LoadBalancingModelConfig, row_id)
+        assert persisted is not None
+        assert persisted.model_type == ModelType.LLM


 def test_load_balancing_model_config_cache_delete_failure_logs_stacktrace(
@@ -0,0 +1,236 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import click
+import pytest
+
+from commands import migrate_legacy_sys_files_workflows
+from commands import workflow_migration as workflow_migration_commands
+
+
+def test_migrate_legacy_sys_files_workflows_command_passes_batch_options(mocker, capsys):
+    runner = mocker.patch.object(
+        workflow_migration_commands,
+        "run_legacy_sys_files_workflow_migration",
+        return_value=workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+            scanned=10,
+            migrated=2,
+            failed=0,
+            batches=1,
+            last_id="workflow-10",
+        ),
+    )
+
+    migrate_legacy_sys_files_workflows.callback(
+        batch_size=200,
+        limit=500,
+        start_after_id="workflow-1",
+        tenant_id="tenant-1",
+        app_id="app-1",
+        dry_run=True,
+    )
+
+    runner.assert_called_once_with(
+        batch_size=200,
+        limit=500,
+        start_after_id="workflow-1",
+        tenant_id="tenant-1",
+        app_id="app-1",
+        dry_run=True,
+    )
+    captured = capsys.readouterr()
+    assert "scanned=10" in captured.out
+    assert "migrated=2" in captured.out
+    assert "last_id=workflow-10" in captured.out
+
+
+def test_migrate_legacy_sys_files_workflows_rejects_non_positive_batch_size():
+    with pytest.raises(click.UsageError, match="batch-size"):
+        migrate_legacy_sys_files_workflows.callback(
+            batch_size=0,
+            limit=None,
+            start_after_id=None,
+            tenant_id=None,
+            app_id=None,
+            dry_run=False,
+        )
+
+
+def test_migrate_legacy_sys_files_workflows_rejects_non_positive_limit():
+    with pytest.raises(click.UsageError, match="limit"):
+        migrate_legacy_sys_files_workflows.callback(
+            batch_size=100,
+            limit=0,
+            start_after_id=None,
+            tenant_id=None,
+            app_id=None,
+            dry_run=False,
+        )
+
+
+def test_build_legacy_sys_files_workflow_query_uses_keyset_pagination():
+    stmt = workflow_migration_commands._build_legacy_sys_files_workflow_query(
+        start_after_id="workflow-1",
+        batch_size=200,
+        tenant_id="tenant-1",
+        app_id="app-1",
+    )
+    compiled = str(stmt.compile(compile_kwargs={"literal_binds": True}))
+
+    assert "workflows.id > 'workflow-1'" in compiled
+    assert "workflows.tenant_id = 'tenant-1'" in compiled
+    assert "workflows.app_id = 'app-1'" in compiled
+    assert "ORDER BY workflows.id" in compiled
+    assert "LIMIT 200" in compiled
+    assert "workflows.environment_variables" not in compiled
+
+
+def test_migrate_legacy_sys_files_workflow_batch_dry_run_rolls_back():
+    migrated_workflow = MagicMock()
+    migrated_workflow.id = "workflow-1"
+    migrated_workflow.migrate_legacy_sys_files_graph_in_place.return_value = True
+    untouched_workflow = MagicMock()
+    untouched_workflow.id = "workflow-2"
+    untouched_workflow.migrate_legacy_sys_files_graph_in_place.return_value = False
+    session = MagicMock()
+    session.scalars.return_value.all.return_value = [migrated_workflow, untouched_workflow]
+
+    stats = workflow_migration_commands._migrate_legacy_sys_files_workflow_batch(
+        session=session,
+        start_after_id=None,
+        batch_size=200,
+        tenant_id=None,
+        app_id=None,
+        dry_run=True,
+    )
+
+    assert stats.scanned == 2
+    assert stats.migrated == 1
+    assert stats.failed == 0
+    assert stats.last_id == "workflow-2"
+    session.rollback.assert_called_once()
+    session.commit.assert_not_called()
+
+
+def test_migrate_legacy_sys_files_workflow_batch_commits_and_counts_failures(caplog):
+    migrated_workflow = MagicMock()
+    migrated_workflow.id = "workflow-1"
+    migrated_workflow.migrate_legacy_sys_files_graph_in_place.return_value = True
+    failing_workflow = MagicMock()
+    failing_workflow.id = "workflow-2"
+    failing_workflow.migrate_legacy_sys_files_graph_in_place.side_effect = RuntimeError("boom")
+    session = MagicMock()
+    session.scalars.return_value.all.return_value = [migrated_workflow, failing_workflow]
+
+    stats = workflow_migration_commands._migrate_legacy_sys_files_workflow_batch(
+        session=session,
+        start_after_id=None,
+        batch_size=200,
+        tenant_id=None,
+        app_id=None,
+        dry_run=False,
+    )
+
+    assert stats.scanned == 2
+    assert stats.migrated == 1
+    assert stats.failed == 1
+    assert stats.last_id == "workflow-2"
+    assert "Failed to migrate legacy" in caplog.text
+    session.commit.assert_called_once()
+    session.rollback.assert_not_called()
+
+
+def test_run_legacy_sys_files_workflow_migration_uses_keyset_batches(mocker):
+    session_maker = MagicMock()
+    sessions = [MagicMock(), MagicMock()]
+    session_maker.side_effect = sessions
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    migrate_batch = mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        side_effect=[
+            workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+                scanned=2,
+                migrated=1,
+                failed=0,
+                last_id="workflow-2",
+            ),
+            workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+                scanned=1,
+                migrated=1,
+                failed=0,
+                last_id="workflow-3",
+            ),
+        ],
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=3,
+        start_after_id="workflow-0",
+        tenant_id="tenant-1",
+        app_id="app-1",
+        dry_run=True,
+    )
+
+    assert stats.scanned == 3
+    assert stats.migrated == 2
+    assert stats.batches == 2
+    assert stats.last_id == "workflow-3"
+    assert migrate_batch.call_args_list[0].kwargs["start_after_id"] == "workflow-0"
+    assert migrate_batch.call_args_list[0].kwargs["batch_size"] == 2
+    assert migrate_batch.call_args_list[1].kwargs["start_after_id"] == "workflow-2"
+    assert migrate_batch.call_args_list[1].kwargs["batch_size"] == 1
+
+
+def test_run_legacy_sys_files_workflow_migration_stops_on_empty_batch(mocker):
+    session_maker = MagicMock(return_value=MagicMock())
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        return_value=workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(scanned=0),
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=None,
+        start_after_id=None,
+        tenant_id=None,
+        app_id=None,
+        dry_run=False,
+    )
+
+    assert stats.scanned == 0
+    assert stats.batches == 0
+
+
+def test_run_legacy_sys_files_workflow_migration_stops_on_short_batch(mocker):
+    session_maker = MagicMock(return_value=MagicMock())
+    mocker.patch.object(workflow_migration_commands, "sessionmaker", return_value=session_maker)
+    mocker.patch.object(workflow_migration_commands, "db", SimpleNamespace(engine=object()))
+    migrate_batch = mocker.patch.object(
+        workflow_migration_commands,
+        "_migrate_legacy_sys_files_workflow_batch",
+        return_value=workflow_migration_commands.LegacySysFilesWorkflowMigrationStats(
+            scanned=1,
+            migrated=1,
+            failed=0,
+            last_id="workflow-1",
+        ),
+    )
+
+    stats = workflow_migration_commands.run_legacy_sys_files_workflow_migration(
+        batch_size=2,
+        limit=None,
+        start_after_id=None,
+        tenant_id=None,
+        app_id=None,
+        dry_run=False,
+    )
+
+    assert stats.scanned == 1
+    assert stats.batches == 1
+    migrate_batch.assert_called_once()
@@ -14,7 +14,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session

 import commands
-from commands import system as system_commands
+from commands import workspace as workspace_commands
 from core.tools.entities.tool_entities import ApiProviderSchemaType
 from graphon.model_runtime.entities.model_entities import ModelType
 from models import Tenant
@@ -83,11 +83,11 @@ def _encrypted_rows(tenant_id: str, *, suffix: str = "1") -> tuple[object, ...]:


 def _bind_command_to_sqlite(monkeypatch: pytest.MonkeyPatch, session: Session) -> None:
-    monkeypatch.setattr(system_commands, "db", SimpleNamespace(engine=session.get_bind()))
+    monkeypatch.setattr(workspace_commands, "db", SimpleNamespace(engine=session.get_bind()))


 def test_reset_aborts_when_not_self_hosted(monkeypatch, capsys):
-    monkeypatch.setattr(system_commands.dify_config, "EDITION", "CLOUD")
+    monkeypatch.setattr(workspace_commands.dify_config, "EDITION", "CLOUD")

    exit_code = _invoke_reset()
    captured = capsys.readouterr()
@@ -106,8 +106,8 @@ def test_reset_purges_provider_and_tool_tables_for_each_tenant(
 ) -> None:
    """The command must purge LLM provider rows AND every tool provider table
    that stores ciphertext encrypted under the tenant key (#35396)."""
-    monkeypatch.setattr(system_commands.dify_config, "EDITION", "SELF_HOSTED")
-    monkeypatch.setattr(system_commands, "generate_key_pair", lambda tenant_id: f"new-key-{tenant_id}")
+    monkeypatch.setattr(workspace_commands.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(workspace_commands, "generate_key_pair", lambda tenant_id: f"new-key-{tenant_id}")
    _bind_command_to_sqlite(monkeypatch, sqlite_session)

    tenant = _tenant(TENANT_ID)
@@ -146,8 +146,8 @@ def test_reset_purges_provider_and_tool_tables_for_each_tenant(
 )
 def test_reset_iterates_all_tenants(monkeypatch: pytest.MonkeyPatch, sqlite_session: Session) -> None:
    """Multi-tenant deployments must purge every tenant, not just the first."""
-    monkeypatch.setattr(system_commands.dify_config, "EDITION", "SELF_HOSTED")
-    monkeypatch.setattr(system_commands, "generate_key_pair", lambda tenant_id: f"new-key-{tenant_id}")
+    monkeypatch.setattr(workspace_commands.dify_config, "EDITION", "SELF_HOSTED")
+    monkeypatch.setattr(workspace_commands, "generate_key_pair", lambda tenant_id: f"new-key-{tenant_id}")

    _bind_command_to_sqlite(monkeypatch, sqlite_session)
    tenant_ids = [f"11111111-1111-1111-1111-{index:012d}" for index in range(3)]
@@ -4,7 +4,7 @@ import types
 from unittest.mock import MagicMock

 import commands
-from commands import system as system_commands
+from commands import database as database_commands
 from libs.db_migration_lock import LockNotOwnedError, RedisError

 HEARTBEAT_WAIT_TIMEOUT_SECONDS = 5.0
@@ -25,11 +25,11 @@ def _invoke_upgrade_db() -> int:


 def test_upgrade_db_skips_when_lock_not_acquired(monkeypatch, capsys):
-    monkeypatch.setattr(system_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 1234)
+    monkeypatch.setattr(database_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 1234)

    lock = MagicMock()
    lock.acquire.return_value = False
-    system_commands.redis_client.lock.return_value = lock
+    database_commands.redis_client.lock.return_value = lock

    exit_code = _invoke_upgrade_db()
    captured = capsys.readouterr()
@@ -37,18 +37,20 @@ def test_upgrade_db_skips_when_lock_not_acquired(monkeypatch, capsys):
    assert exit_code == 0
    assert "Database migration skipped" in captured.out

-    system_commands.redis_client.lock.assert_called_once_with(name="db_upgrade_lock", timeout=1234, thread_local=False)
+    database_commands.redis_client.lock.assert_called_once_with(
+        name="db_upgrade_lock", timeout=1234, thread_local=False
+    )
    lock.acquire.assert_called_once_with(blocking=False)
    lock.release.assert_not_called()


 def test_upgrade_db_failure_not_masked_by_lock_release(monkeypatch, capsys):
-    monkeypatch.setattr(system_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 321)
+    monkeypatch.setattr(database_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 321)

    lock = MagicMock()
    lock.acquire.return_value = True
    lock.release.side_effect = LockNotOwnedError("simulated")
-    system_commands.redis_client.lock.return_value = lock
+    database_commands.redis_client.lock.return_value = lock

    def _upgrade():
        raise RuntimeError("boom")
@@ -61,18 +63,18 @@ def test_upgrade_db_failure_not_masked_by_lock_release(monkeypatch, capsys):
    assert exit_code == 1
    assert "Database migration failed: boom" in captured.out

-    system_commands.redis_client.lock.assert_called_once_with(name="db_upgrade_lock", timeout=321, thread_local=False)
+    database_commands.redis_client.lock.assert_called_once_with(name="db_upgrade_lock", timeout=321, thread_local=False)
    lock.acquire.assert_called_once_with(blocking=False)
    lock.release.assert_called_once()


 def test_upgrade_db_success_ignores_lock_not_owned_on_release(monkeypatch, capsys):
-    monkeypatch.setattr(system_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 999)
+    monkeypatch.setattr(database_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 999)

    lock = MagicMock()
    lock.acquire.return_value = True
    lock.release.side_effect = LockNotOwnedError("simulated")
-    system_commands.redis_client.lock.return_value = lock
+    database_commands.redis_client.lock.return_value = lock

    _install_fake_flask_migrate(monkeypatch, lambda: None)

@@ -82,7 +84,7 @@ def test_upgrade_db_success_ignores_lock_not_owned_on_release(monkeypatch, capsy
    assert exit_code == 0
    assert "Database migration successful!" in captured.out

-    system_commands.redis_client.lock.assert_called_once_with(name="db_upgrade_lock", timeout=999, thread_local=False)
+    database_commands.redis_client.lock.assert_called_once_with(name="db_upgrade_lock", timeout=999, thread_local=False)
    lock.acquire.assert_called_once_with(blocking=False)
    lock.release.assert_called_once()

@@ -93,11 +95,11 @@ def test_upgrade_db_renews_lock_during_migration(monkeypatch, capsys):
    """

    # Use a small TTL so the heartbeat interval triggers quickly.
-    monkeypatch.setattr(system_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 0.3)
+    monkeypatch.setattr(database_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 0.3)

    lock = MagicMock()
    lock.acquire.return_value = True
-    system_commands.redis_client.lock.return_value = lock
+    database_commands.redis_client.lock.return_value = lock

    renewed = threading.Event()

@@ -121,11 +123,11 @@ def test_upgrade_db_renews_lock_during_migration(monkeypatch, capsys):

 def test_upgrade_db_ignores_reacquire_errors(monkeypatch, capsys):
    # Use a small TTL so heartbeat runs during the upgrade call.
-    monkeypatch.setattr(system_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 0.3)
+    monkeypatch.setattr(database_commands, "DB_UPGRADE_LOCK_TTL_SECONDS", 0.3)

    lock = MagicMock()
    lock.acquire.return_value = True
-    system_commands.redis_client.lock.return_value = lock
+    database_commands.redis_client.lock.return_value = lock

    attempted = threading.Event()

--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
autofix-ci[bot]andGitHub	756ce8a10e	[autofix.ci] apply automated fixes	2026-07-28 05:33:30 +00:00
-LAN-	e07de297bf	fix(workflow): migrate sys files to user input	2026-07-28 13:29:09 +08:00
yyhandGitHub	1e5e47b889	fix: validate plugin installation scope (#39669 )	2026-07-28 04:13:20 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	dd28b0d165	test: move OAuth server service coverage to unit tests (#38931 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-28 04:09:32 +00:00
JoelandGitHub	49e74e2f58	fix: prevent the model selector footer from covering options (#39668 )	2026-07-28 04:02:28 +00:00
Joel GitHub yyh	e6e5d761c2	fix: prevent Safari from clipping the settings close button (#39664 ) Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>	2026-07-28 04:01:12 +00:00
Asuka MinatoandGitHub	e1ce808567	test: move data source controller coverage to unit tests (#38924 )	2026-07-28 03:51:22 +00:00
Asuka MinatoandGitHub	3c7ad816d9	test: use SQLite sessions in commands (#39080 )	2026-07-28 03:29:02 +00:00
Asuka MinatoandGitHub	f44dd343da	test: use sqlite3 session in test_plugin_service (#38727 )	2026-07-28 03:27:58 +00:00
Asuka MinatoandGitHub	2d9b2d50f3	test: use sqlite3 session in test_wraps (#38770 )	2026-07-28 03:27:22 +00:00
Asuka MinatoandGitHub	003e0f9614	test: move message cleanup coverage to unit tests (#38932 )	2026-07-28 03:26:43 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	da0979b373	test: use SQLite sessions in services core (#39090 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-28 03:20:27 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	81fb57639e	test: use SQLite sessions in controllers service api (#39098 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-28 03:19:17 +00:00
Asuka MinatoandGitHub	1e61078e93	test: use SQLite sessions in services core (#39088 )	2026-07-28 03:18:32 +00:00
Escape0707andGitHub	b597bb1b17	test: separate human input unit and database paths (#39655 )	2026-07-28 03:18:01 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	65ead05dfc	test: use SQLite sessions in services plugin (#39084 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-28 03:16:19 +00:00
wangxiaolei GitHub yunlu.wen	6f8ed69ee1	fix: fix mcp output_schema is optional (#39453 ) Co-authored-by: yunlu.wen <yunlu.wen@dify.ai>	2026-07-28 02:31:18 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	59b879d2df	chore: bump version to 1.16.1 (#39653 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-28 01:09:25 +00:00
Xiyuan ChenandGitHub	a57b0b9b58	feat(plugin): allow disabling the tenant plugin model providers cache (#39632 )	2026-07-27 23:53:59 +00:00
Xiyuan ChenandGitHub	d8506efed6	fix(cli): decouple release script tests from the live compat window (#39658 )	2026-07-27 23:50:05 +00:00
JingyiandGitHub	b25b28cc76	fix(workflow): align block icon vector sizes (#39657 )	2026-07-27 23:43:03 +00:00
yyh GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	f68cabe226	fix(web): align tour trigger DOM order (#39654 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 13:00:16 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	f73f83c5d6	test: use SQLite sessions in core app (#39075 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:45:16 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	29801d6a65	test: use SQLite sessions in core tools (#39077 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:45:11 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	75016e8bfe	test: use SQLite sessions in core ops (#39072 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:41:17 +00:00
yyhandGitHub	b6dea7b2ba	fix(workflow): preserve latest collaboration session (#39646 )	2026-07-27 10:37:45 +00:00
yyhandGitHub	b2b1cd7e97	chore: update workspace dependencies (#39641 )	2026-07-27 10:27:31 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	ca63e01d45	test: use SQLite sessions in controllers service api (#39069 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:26:23 +00:00
yyhandGitHub	271b6e1f5c	fix(ui): unify combobox trigger focus rings (#39643 )	2026-07-27 10:21:32 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	609ddf9e01	test: use SQLite sessions in core memory (#39066 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:15:08 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	9fd1fea58c	test: use SQLite sessions in services core (#39057 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 09:36:44 +00:00
JoelandGitHub	8de3b4d033	fix: prevent the agent build draft UI from flashing during apply (#39642 )	2026-07-27 09:32:52 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	d9c038daf2	chore(deps): bump the storage group across 1 directory with 3 updates (#39604 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 09:24:58 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	c9c057fd04	test: use SQLite sessions in rag retrieval (#39049 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 09:12:30 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	85189be53d	chore: bump pymdown-extensions from 10.21.2 to 11.0 in /dify-agent (#39618 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 09:10:22 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	01e736aaf7	test: use sqlite3 session in test_extract_thread_messages (#38739 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 09:00:47 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	a28969f564	chore(deps): bump golang.org/x/net from 0.53.0 to 0.55.0 in /dify-agent-runtime (#39025 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 08:58:40 +00:00
Asuka MinatoandGitHub	42f9610ba5	test: use sqlite3 session in test_api (#38733 )	2026-07-27 08:57:51 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	1a7ffbe5ee	chore: bump gitpython from 3.1.52 to 3.1.54 in /api (#39619 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 08:57:44 +00:00
JoelandGitHub	481354ca70	refactor: make agent composer save and publish state consistent (#39637 )	2026-07-27 08:43:45 +00:00
Yunlu Wen GitHub Copilot Autofix powered by AI	3c80857ea3	feat: add bearer auth to agent backend (#39622 ) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>	2026-07-27 08:37:30 +00:00
Xiyuan ChenandGitHub	94702efbc2	fix: gate service API, MCP and trigger surfaces on enterprise license (#39635 )	2026-07-27 08:23:15 +00:00
Asuka MinatoandGitHub	755f7b0e8b	test: use sqlite3 session in test_workflow_comment_service (#38700 )	2026-07-27 08:22:45 +00:00
Asuka MinatoandGitHub	a1c9564b30	test: use sqlite3 session in test_llm_generator_missing (#38742 )	2026-07-27 08:22:16 +00:00
Asuka MinatoandGitHub	97acd9c70b	test: use sqlite3 session in test_utils (#38729 )	2026-07-27 08:21:39 +00:00
Asuka MinatoandGitHub	71601bf76c	test: use sqlite3 session in test_draft_var_loader_simple (#38702 )	2026-07-27 08:20:18 +00:00
Asuka MinatoandGitHub	460efbf285	test: use sqlite3 session in test_workflow_run_service (#38701 )	2026-07-27 08:19:32 +00:00
Escape0707 GitHub Asuka Minato	a752f43b8e	test: use pristine file-backed SQLite fixtures (#39624 ) Co-authored-by: Asuka Minato <i@asukaminato.eu.org>	2026-07-27 08:07:05 +00:00
非法操作 GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	b3774bfe1c	perf: add lightweight recent apps endpoint (#39625 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 08:00:39 +00:00
yyhandGitHub	4313d23889	perf(web): lazy-load Home creation modals (#39634 )	2026-07-27 07:48:35 +00:00
林玮 (Jade Lin)andGitHub	26a43db6a4	fix(api): wait for workflow worker cleanup before task completion (#39614 )	2026-07-27 07:34:25 +00:00
yyhandGitHub	e5d40336b3	fix(workflow): lazy-load Loro collaboration runtime (#39631 )	2026-07-27 07:15:13 +00:00
yyhandGitHub	58e2bcbba1	fix(web): stop reporting workspace status (#39626 )	2026-07-27 06:22:37 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	e1b55f54e6	feat(agent): add a squid proxy for agent sandbox (#39544 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 06:21:20 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	989039db6e	fix: use jinja sandbox (#39609 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 06:03:01 +00:00
Joel GitHub 林玮autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	52428df1bd	feat: add amptitude to new agent (#39608 ) Co-authored-by: 林玮 (Jade Lin) <linw1995@icloud.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 05:54:45 +00:00
yyh GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	28d174603f	fix: align workspace card plan ownership (#39616 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 05:01:02 +00:00
Xiyuan ChenandGitHub	b6099d09ff	feat(inner_api): add endpoint to invalidate plugin model providers cache (#39468 )	2026-07-27 04:33:36 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2962a7ea93	chore(deps): bump the github-actions-dependencies group with 6 updates (#39603 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 04:32:09 +00:00
zyssyz123andGitHub	cc01189966	fix(agent): expose publish state from composer (#39615 )	2026-07-27 04:05:38 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	d563d6e7df	chore: bump pypdf from 6.10.2 to 6.14.2 in /dify-agent (#39502 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 03:56:57 +00:00
CYJ1226andGitHub	e4e99d198b	fix: preserve spaces during recursive text splitting (#39498 )	2026-07-27 03:21:26 +00:00
yyhandGitHub	34e6e5a049	refactor(web): simplify Dify logo theming (#39580 )	2026-07-27 03:01:24 +00:00
林玮 (Jade Lin)GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	7e6ba05464	feat(api): expose app mode in webapp site response (#39607 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 02:57:18 +00:00
Madan kumarandGitHub	aeda37db68	fix(rag): stop the document cleaner from stripping valid characters ï, ¿, ¾ (#39215 )	2026-07-27 02:53:33 +00:00
FFXNandGitHub	441f9f9ec0	fix: The expiration time for web app login to JWT is incorrectly set. (#39537 )	2026-07-27 02:43:32 +00:00
Taranum Wasu GitHub Taranum Wasu Cursor	f16c249b1e	fix(api): honor array-element limit and byte budget for list[File] in VariableTruncator (#39218 ) (#39220 ) Co-authored-by: Taranum Wasu <taranumwasu@Taranums-MacBook-Pro.local> Co-authored-by: Cursor <cursoragent@cursor.com>	2026-07-27 02:12:57 +00:00
FFXN GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	858b3b74dc	feat: add CE telemetry report (#39452 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 02:04:28 +00:00
David Park GitHub Claude Opus 4.8	d5c0e927c6	fix(web): stop chunk length/overlap inputs collapsing to an unusable width in narrow containers (#39600 ) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-27 02:03:17 +00:00
Asuka MinatoandGitHub	b81737cfb3	test: use sqlite3 session in test_message_service (#38698 )	2026-07-27 02:01:47 +00:00