fix(completion): preserve legacy workflow compatibility

fix(workflow): keep converter imports sorted
fix(workflow): preserve polling llm invoke hook
2026-07-28 13:04:11 +08:00 · 2026-07-28 13:03:12 +08:00 · 2026-07-28 13:03:12 +08:00 · 2026-07-28 13:03:12 +08:00 · 2026-07-28 13:03:11 +08:00 · 2026-07-28 13:03:11 +08:00
316 changed files with 14562 additions and 10986 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,
    )
@@ -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")
@@ -278,7 +278,9 @@ class ChatMessageTextApi(Resource):
    @get_app_model
    def post(self, app_model: App):
        try:
-            payload = TextToSpeechPayload.model_validate(console_ns.payload)
+            payload_data = dict(console_ns.payload or {})
+            payload_data.setdefault("text", "")
+            payload = TextToSpeechPayload.model_validate(payload_data)
            message_ref = None
            if payload.message_id:
                app_ref = AppRefService.create_app_ref(app_model)
@@ -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,
@@ -616,23 +616,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(
        *,
@@ -1,7 +1,7 @@
 import base64
 import logging
 import time
-from collections.abc import Generator, Mapping, Sequence
+from collections.abc import Generator, Mapping, MutableMapping, Sequence
 from mimetypes import guess_extension
 from typing import TYPE_CHECKING, Any, Union

@@ -54,9 +54,15 @@ _logger = logging.getLogger(__name__)

 class AppRunner:
    def recalc_llm_max_tokens(
-        self, model_config: ModelConfigWithCredentialsEntity, prompt_messages: list[PromptMessage]
-    ):
+        self,
+        model_config: ModelConfigWithCredentialsEntity,
+        prompt_messages: list[PromptMessage],
+        *,
+        model_parameters: MutableMapping[str, Any] | None = None,
+    ) -> int | None:
+        """Clamp max tokens against the final prompt on the selected parameter mapping."""
        # recalc max_tokens if sum(prompt_token +  max_tokens) over model token limit
+        parameters = model_parameters if model_parameters is not None else model_config.parameters
        model_instance = ModelInstance(
            provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
        )
@@ -69,8 +75,7 @@ class AppRunner:
                parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
            ):
                max_tokens = (
-                    model_config.parameters.get(parameter_rule.name)
-                    or model_config.parameters.get(parameter_rule.use_template or "")
+                    parameters.get(parameter_rule.name) or parameters.get(parameter_rule.use_template or "")
                ) or 0

        if model_context_tokens is None:
@@ -85,7 +90,9 @@ class AppRunner:
                if parameter_rule.name == "max_tokens" or (
                    parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
                ):
-                    model_config.parameters[parameter_rule.name] = max_tokens
+                    parameters[parameter_rule.name] = max_tokens
+
+        return None

    def organize_prompt_messages(
        self,
@@ -15,8 +15,8 @@ from core.app.app_config.easy_ui_based_app.model_config.converter import ModelCo
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
-from core.app.apps.completion.app_runner import CompletionAppRunner
 from core.app.apps.completion.generate_response_converter import CompletionAppGenerateResponseConverter
+from core.app.apps.completion.workflow_runner import CompletionWorkflowRunner
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
@@ -243,7 +243,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
                message = self._get_message(message_id)

                # chatbot app
-                runner = CompletionAppRunner()
+                runner = CompletionWorkflowRunner()
                with session_factory.create_session() as session:
                    runner.run(
                        application_generate_entity=application_generate_entity,
@@ -1,208 +0,0 @@
-import logging
-from typing import cast
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.apps.base_app_runner import AppRunner
-from core.app.apps.completion.app_config_manager import CompletionAppConfig
-from core.app.entities.app_invoke_entities import (
-    CompletionAppGenerateEntity,
-)
-from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.db.session_factory import create_session
-from core.model_manager import ModelInstance
-from core.moderation.base import ModerationError
-from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
-from graphon.file import File
-from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent
-from models.model import App, Message
-
-logger = logging.getLogger(__name__)
-
-
-class CompletionAppRunner(AppRunner):
-    """
-    Completion Application Runner
-    """
-
-    def run(
-        self,
-        application_generate_entity: CompletionAppGenerateEntity,
-        queue_manager: AppQueueManager,
-        message: Message,
-        session: Session,
-    ):
-        """Run the application without retaining ``session`` during model I/O.
-
-        Database preparation is committed and the connection is released before
-        the provider response is requested or consumed.
-
-        :param application_generate_entity: application generate entity
-        :param queue_manager: application queue manager
-        :param message: message
-        :return:
-        """
-        app_config = application_generate_entity.app_config
-        app_config = cast(CompletionAppConfig, app_config)
-        stmt = select(App).where(App.id == app_config.app_id)
-        with create_session() as read_session:
-            app_record = read_session.scalar(stmt)
-            if app_record:
-                read_session.expunge(app_record)
-        if not app_record:
-            raise ValueError("App not found")
-
-        inputs = application_generate_entity.inputs
-        query = application_generate_entity.query
-        files = application_generate_entity.files
-
-        image_detail_config = (
-            application_generate_entity.file_upload_config.image_config.detail
-            if (
-                application_generate_entity.file_upload_config
-                and application_generate_entity.file_upload_config.image_config
-            )
-            else None
-        )
-        image_detail_config = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
-
-        # organize all inputs and template to prompt messages
-        # Include: prompt template, inputs, query(optional), files(optional)
-        prompt_messages, stop = self.organize_prompt_messages(
-            app_record=app_record,
-            model_config=application_generate_entity.model_conf,
-            prompt_template_entity=app_config.prompt_template,
-            inputs=inputs,
-            files=files,
-            query=query,
-            image_detail_config=image_detail_config,
-        )
-
-        # moderation
-        try:
-            # process sensitive_word_avoidance
-            _, inputs, query = self.moderation_for_inputs(
-                app_id=app_record.id,
-                tenant_id=app_config.tenant_id,
-                app_generate_entity=application_generate_entity,
-                inputs=inputs,
-                query=query or "",
-                message_id=message.id,
-            )
-        except ModerationError as e:
-            self.direct_output(
-                queue_manager=queue_manager,
-                app_generate_entity=application_generate_entity,
-                prompt_messages=prompt_messages,
-                text=str(e),
-                stream=application_generate_entity.stream,
-            )
-            return
-
-        # fill in variable inputs from external data tools if exists
-        external_data_tools = app_config.external_data_variables
-        if external_data_tools:
-            inputs = self.fill_in_inputs_from_external_data_tools(
-                tenant_id=app_record.tenant_id,
-                app_id=app_record.id,
-                external_data_tools=external_data_tools,
-                inputs=inputs,
-                query=query,
-            )
-
-        # get context from datasets
-        context = None
-        context_files: list[File] = []
-        if app_config.dataset and app_config.dataset.dataset_ids:
-            hit_callback = DatasetIndexToolCallbackHandler(
-                queue_manager,
-                app_record.id,
-                message.id,
-                application_generate_entity.user_id,
-                application_generate_entity.invoke_from,
-            )
-
-            dataset_config = app_config.dataset
-            if dataset_config and dataset_config.retrieve_config.query_variable:
-                query = inputs.get(dataset_config.retrieve_config.query_variable, "")
-
-            dataset_retrieval = DatasetRetrieval(application_generate_entity)
-            context, retrieved_files = dataset_retrieval.retrieve(
-                session=session,
-                app_id=app_record.id,
-                user_id=application_generate_entity.user_id,
-                tenant_id=app_record.tenant_id,
-                model_config=application_generate_entity.model_conf,
-                config=dataset_config,
-                query=query or "",
-                invoke_from=application_generate_entity.invoke_from,
-                show_retrieve_source=app_config.additional_features.show_retrieve_source
-                if app_config.additional_features
-                else False,
-                hit_callback=hit_callback,
-                message_id=message.id,
-                inputs=inputs,
-                vision_enabled=bool(
-                    application_generate_entity.app_config.app_model_config_dict.get("file_upload", {})
-                    .get("image", {})
-                    .get("enabled", False)
-                ),
-            )
-            context_files = retrieved_files or []
-
-        session.commit()
-        session.close()
-
-        # reorganize all inputs and template to prompt messages
-        # Include: prompt template, inputs, query(optional), files(optional)
-        #          memory(optional), external data, dataset context(optional)
-        prompt_messages, stop = self.organize_prompt_messages(
-            app_record=app_record,
-            model_config=application_generate_entity.model_conf,
-            prompt_template_entity=app_config.prompt_template,
-            inputs=inputs,
-            files=files,
-            query=query,
-            context=context,
-            image_detail_config=image_detail_config,
-            context_files=context_files,
-        )
-
-        # check hosting moderation
-        hosting_moderation_result = self.check_hosting_moderation(
-            application_generate_entity=application_generate_entity,
-            queue_manager=queue_manager,
-            prompt_messages=prompt_messages,
-        )
-
-        if hosting_moderation_result:
-            return
-
-        # Re-calculate the max tokens if sum(prompt_token +  max_tokens) over model token limit
-        self.recalc_llm_max_tokens(model_config=application_generate_entity.model_conf, prompt_messages=prompt_messages)
-
-        # Invoke model
-        model_instance = ModelInstance(
-            provider_model_bundle=application_generate_entity.model_conf.provider_model_bundle,
-            model=application_generate_entity.model_conf.model,
-        )
-
-        invoke_result = model_instance.invoke_llm(
-            prompt_messages=prompt_messages,
-            model_parameters=application_generate_entity.model_conf.parameters,
-            stop=stop,
-            stream=application_generate_entity.stream,
-            request_metadata={"app_id": app_config.app_id},
-        )
-
-        # handle invoke result
-        self._handle_invoke_result(
-            invoke_result=invoke_result,
-            queue_manager=queue_manager,
-            stream=application_generate_entity.stream,
-            message_id=message.id,
-            user_id=application_generate_entity.user_id,
-            tenant_id=app_config.tenant_id,
-        )
@@ -0,0 +1,144 @@
+from collections.abc import Mapping, Sequence
+
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity
+from core.app.entities.queue_entities import (
+    QueueErrorEvent,
+    QueueLLMChunkEvent,
+    QueueMessageEndEvent,
+    QueueRetrieverResourcesEvent,
+    QueueStopEvent,
+)
+from core.rag.entities import RetrievalSourceMetadata
+from graphon.graph_events import (
+    GraphEngineEvent,
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunSucceededEvent,
+    NodeRunRetrieverResourceEvent,
+    NodeRunStreamChunkEvent,
+    NodeRunSucceededEvent,
+)
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
+from graphon.model_runtime.entities.message_entities import AssistantPromptMessage, PromptMessage
+
+_LLM_TEXT_SELECTOR_PREFIX = ("llm", "text")
+
+
+class CompletionGraphEventAdapter:
+    """Translate one runtime graph run into legacy Completion queue events."""
+
+    _application_generate_entity: CompletionAppGenerateEntity
+    _queue_manager: AppQueueManager
+    _answer: str
+    _usage: LLMUsage
+    _prompt_messages: list[PromptMessage]
+    _chunk_index: int
+
+    def __init__(
+        self,
+        *,
+        application_generate_entity: CompletionAppGenerateEntity,
+        queue_manager: AppQueueManager,
+    ) -> None:
+        self._application_generate_entity = application_generate_entity
+        self._queue_manager = queue_manager
+        self._answer = ""
+        self._usage = LLMUsage.empty_usage()
+        self._prompt_messages = []
+        self._chunk_index = 0
+
+    def set_prompt_messages(self, prompt_messages: Sequence[PromptMessage]) -> None:
+        """Capture the final GraphOn prompt for legacy chunks and message persistence."""
+        self._prompt_messages = list(prompt_messages)
+
+    def handle_event(self, event: GraphEngineEvent) -> None:
+        match event:
+            case NodeRunStreamChunkEvent():
+                self._handle_stream_chunk(event)
+            case NodeRunRetrieverResourceEvent():
+                self._handle_retriever_resource(event)
+            case NodeRunSucceededEvent():
+                self._handle_node_succeeded(event)
+            case GraphRunSucceededEvent():
+                self._publish_message_end(event.outputs)
+            case GraphRunFailedEvent():
+                self._publish_error(event.error)
+            case GraphRunAbortedEvent():
+                self._queue_manager.publish(
+                    QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL),
+                    PublishFrom.APPLICATION_MANAGER,
+                )
+            case _:
+                return
+
+    def _handle_stream_chunk(self, event: NodeRunStreamChunkEvent) -> None:
+        if tuple(event.selector)[:2] != _LLM_TEXT_SELECTOR_PREFIX:
+            return
+        if event.is_final and not event.chunk:
+            return
+
+        self._answer += event.chunk
+        self._queue_manager.publish(
+            QueueLLMChunkEvent(
+                chunk=LLMResultChunk(
+                    model=self._application_generate_entity.model_conf.model,
+                    prompt_messages=self._prompt_messages,
+                    delta=LLMResultChunkDelta(
+                        index=self._chunk_index,
+                        message=AssistantPromptMessage(content=event.chunk),
+                    ),
+                )
+            ),
+            PublishFrom.APPLICATION_MANAGER,
+        )
+        self._chunk_index += 1
+
+    def _handle_retriever_resource(self, event: NodeRunRetrieverResourceEvent) -> None:
+        additional_features = self._application_generate_entity.app_config.additional_features
+        if not additional_features or not additional_features.show_retrieve_source:
+            return
+
+        self._queue_manager.publish(
+            QueueRetrieverResourcesEvent(
+                retriever_resources=[
+                    RetrievalSourceMetadata.model_validate(resource) for resource in event.retriever_resources
+                ],
+                in_iteration_id=event.in_iteration_id,
+                in_loop_id=event.in_loop_id,
+            ),
+            PublishFrom.APPLICATION_MANAGER,
+        )
+
+    def _handle_node_succeeded(self, event: NodeRunSucceededEvent) -> None:
+        if event.node_id != "llm":
+            return
+
+        result = event.node_run_result
+        text = result.outputs.get("text")
+        if isinstance(text, str):
+            self._answer = text
+        self._usage = result.llm_usage
+
+    def _publish_message_end(self, outputs: Mapping[str, object]) -> None:
+        result = outputs.get("result")
+        if isinstance(result, str) and not self._answer:
+            self._answer = result
+
+        self._queue_manager.publish(
+            QueueMessageEndEvent(
+                llm_result=LLMResult(
+                    model=self._application_generate_entity.model_conf.model,
+                    prompt_messages=self._prompt_messages,
+                    message=AssistantPromptMessage(content=self._answer),
+                    usage=self._usage,
+                )
+            ),
+            PublishFrom.APPLICATION_MANAGER,
+        )
+
+    def _publish_error(self, error: object) -> None:
+        self._queue_manager.publish(
+            QueueErrorEvent(error=ValueError(str(error))),
+            PublishFrom.APPLICATION_MANAGER,
+        )
@@ -0,0 +1,69 @@
+import json
+from dataclasses import dataclass
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy.orm import Session
+
+from core.app.apps.completion.app_config_manager import CompletionAppConfig
+from graphon.nodes import BuiltinNodeTypes
+from models.model import App, AppMode
+from services.workflow.workflow_converter import WorkflowConverter, WorkflowGraph
+
+
+@dataclass(frozen=True, slots=True)
+class RuntimeCompletionWorkflow:
+    workflow_id: str
+    root_node_id: str
+    graph_dict: WorkflowGraph
+
+
+def build_runtime_completion_workflow(
+    *,
+    app_model: App,
+    app_config: CompletionAppConfig,
+    session: Session,
+    workflow_converter: WorkflowConverter | None = None,
+) -> RuntimeCompletionWorkflow:
+    """Build the transient WorkflowEntry graph used by Completion execution."""
+    converter = workflow_converter or WorkflowConverter()
+    graph, _ = converter.build_graph_from_app_config(
+        app_model=app_model,
+        app_config=app_config,
+        target_app_mode=AppMode.WORKFLOW,
+        session=session,
+    )
+    _route_external_data_query_to_sys_query(graph)
+    return RuntimeCompletionWorkflow(
+        workflow_id=f"completion-runtime-{uuid4()}",
+        root_node_id="start",
+        graph_dict=graph,
+    )
+
+
+def _route_external_data_query_to_sys_query(graph: WorkflowGraph) -> None:
+    """Preserve Completion API-based variable behavior in the runtime graph."""
+    for node in graph["nodes"]:
+        data = node.get("data", {})
+        if data.get("type") != BuiltinNodeTypes.HTTP_REQUEST:
+            continue
+
+        body = data.get("body")
+        if not isinstance(body, dict) or body.get("type") != "json":
+            continue
+
+        raw_body_data = body.get("data")
+        if not isinstance(raw_body_data, str):
+            continue
+
+        try:
+            body_data: dict[str, Any] = json.loads(raw_body_data)
+        except json.JSONDecodeError:
+            continue
+
+        params = body_data.get("params")
+        if not isinstance(params, dict) or params.get("query") != "":
+            continue
+
+        params["query"] = "{{#sys.query#}}"
+        body["data"] = json.dumps(body_data)
@@ -0,0 +1,242 @@
+import time
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any, cast
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.base_app_runner import AppRunner
+from core.app.apps.completion.app_config_manager import CompletionAppConfig
+from core.app.apps.completion.graph_event_adapter import CompletionGraphEventAdapter
+from core.app.apps.completion.runtime_workflow_builder import build_runtime_completion_workflow
+from core.app.apps.exc import GenerateTaskStoppedError
+from core.app.apps.workflow_app_runner import init_graph
+from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, UserFrom
+from core.moderation.base import ModerationError
+from core.workflow.node_runtime import DIFY_BEFORE_LLM_INVOKE_KEY
+from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
+from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
+from core.workflow.workflow_entry import WorkflowEntry
+from extensions.ext_redis import redis_client
+from graphon.graph_engine.command_channels import RedisChannel
+from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessage
+from graphon.runtime import GraphRuntimeState, VariablePool
+from models.model import App, Message
+
+
+@dataclass(frozen=True, slots=True)
+class ModeratedCompletionInputs:
+    stopped: bool
+    inputs: Mapping[str, Any]
+    query: str
+
+
+class CompletionWorkflowRunner(AppRunner):
+    """Run a transient WorkflowEntry graph while the legacy task pipeline owns persistence."""
+
+    def run(
+        self,
+        application_generate_entity: CompletionAppGenerateEntity,
+        queue_manager: AppQueueManager,
+        message: Message,
+        session: Session,
+    ) -> None:
+        app_config = cast(CompletionAppConfig, application_generate_entity.app_config)
+        app_record = self._get_app(app_id=app_config.app_id, tenant_id=app_config.tenant_id, session=session)
+
+        moderation_result = self._run_input_moderation(
+            app_record=app_record,
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            message=message,
+        )
+        if moderation_result.stopped:
+            return
+
+        runtime_workflow = build_runtime_completion_workflow(
+            app_model=app_record,
+            app_config=app_config,
+            session=session,
+        )
+        variable_pool = self._build_variable_pool(
+            application_generate_entity=application_generate_entity,
+            message=message,
+            workflow_id=runtime_workflow.workflow_id,
+            root_node_id=runtime_workflow.root_node_id,
+            inputs=moderation_result.inputs,
+            query=moderation_result.query,
+        )
+        graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
+        user_from = self._resolve_user_from(application_generate_entity)
+        adapter = CompletionGraphEventAdapter(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+        )
+        extra_context = {
+            DIFY_BEFORE_LLM_INVOKE_KEY: self._build_before_llm_invoke_hook(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                adapter=adapter,
+            )
+        }
+
+        graph = init_graph(
+            app_id=app_config.app_id,
+            graph_config=runtime_workflow.graph_dict,
+            graph_runtime_state=graph_runtime_state,
+            user_from=user_from,
+            invoke_from=application_generate_entity.invoke_from,
+            workflow_id=runtime_workflow.workflow_id,
+            tenant_id=app_config.tenant_id,
+            user_id=application_generate_entity.user_id,
+            root_node_id=runtime_workflow.root_node_id,
+            trace_session_id=application_generate_entity.extras.get("trace_session_id"),
+            call_depth=application_generate_entity.call_depth,
+            extra_context=extra_context,
+        )
+
+        queue_manager.graph_runtime_state = graph_runtime_state
+        command_channel = RedisChannel(redis_client, f"workflow:{application_generate_entity.task_id}:commands")
+        workflow_entry = WorkflowEntry(
+            tenant_id=app_config.tenant_id,
+            app_id=app_config.app_id,
+            workflow_id=runtime_workflow.workflow_id,
+            graph_config=runtime_workflow.graph_dict,
+            graph=graph,
+            user_id=application_generate_entity.user_id,
+            user_from=user_from,
+            invoke_from=application_generate_entity.invoke_from,
+            call_depth=application_generate_entity.call_depth,
+            variable_pool=variable_pool,
+            graph_runtime_state=graph_runtime_state,
+            command_channel=command_channel,
+        )
+        # Do not hold a database connection during graph execution or provider streaming.
+        session.commit()
+        session.close()
+        for event in workflow_entry.run():
+            adapter.handle_event(event)
+
+    def _get_app(self, *, app_id: str, tenant_id: str, session: Session) -> App:
+        app_record = session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id))
+        if not app_record:
+            raise ValueError("App not found")
+        return app_record
+
+    def _run_input_moderation(
+        self,
+        *,
+        app_record: App,
+        application_generate_entity: CompletionAppGenerateEntity,
+        queue_manager: AppQueueManager,
+        message: Message,
+    ) -> ModeratedCompletionInputs:
+        app_config = cast(CompletionAppConfig, application_generate_entity.app_config)
+        prompt_messages, _ = self.organize_prompt_messages(
+            app_record=app_record,
+            model_config=application_generate_entity.model_conf,
+            prompt_template_entity=app_config.prompt_template,
+            inputs=application_generate_entity.inputs,
+            files=application_generate_entity.files,
+            query=application_generate_entity.query,
+            image_detail_config=self._resolve_image_detail_config(application_generate_entity),
+        )
+
+        try:
+            _, inputs, query = self.moderation_for_inputs(
+                app_id=app_record.id,
+                tenant_id=app_config.tenant_id,
+                app_generate_entity=application_generate_entity,
+                inputs=application_generate_entity.inputs,
+                query=application_generate_entity.query or "",
+                message_id=message.id,
+            )
+        except ModerationError as exc:
+            self.direct_output(
+                queue_manager=queue_manager,
+                app_generate_entity=application_generate_entity,
+                prompt_messages=prompt_messages,
+                text=str(exc),
+                stream=application_generate_entity.stream,
+            )
+            return ModeratedCompletionInputs(
+                stopped=True,
+                inputs=application_generate_entity.inputs,
+                query=application_generate_entity.query or "",
+            )
+
+        return ModeratedCompletionInputs(stopped=False, inputs=inputs, query=query)
+
+    def _build_before_llm_invoke_hook(
+        self,
+        *,
+        application_generate_entity: CompletionAppGenerateEntity,
+        queue_manager: AppQueueManager,
+        adapter: CompletionGraphEventAdapter,
+    ) -> Callable[[Sequence[PromptMessage], Mapping[str, Any]], Mapping[str, Any]]:
+        def check(
+            prompt_messages: Sequence[PromptMessage],
+            model_parameters: Mapping[str, Any],
+        ) -> Mapping[str, Any]:
+            adapter.set_prompt_messages(prompt_messages)
+            if self.check_hosting_moderation(
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                prompt_messages=list(prompt_messages),
+            ):
+                raise GenerateTaskStoppedError()
+
+            adjusted_parameters = dict(model_parameters)
+            self.recalc_llm_max_tokens(
+                model_config=application_generate_entity.model_conf,
+                prompt_messages=list(prompt_messages),
+                model_parameters=adjusted_parameters,
+            )
+            return adjusted_parameters
+
+        return check
+
+    def _build_variable_pool(
+        self,
+        *,
+        application_generate_entity: CompletionAppGenerateEntity,
+        message: Message,
+        workflow_id: str,
+        root_node_id: str,
+        inputs: Mapping[str, Any],
+        query: str,
+    ) -> VariablePool:
+        variable_pool = VariablePool()
+        system_inputs = build_system_variables(
+            files=application_generate_entity.files,
+            user_id=application_generate_entity.user_id,
+            app_id=application_generate_entity.app_config.app_id,
+            workflow_id=workflow_id,
+            workflow_execution_id=application_generate_entity.task_id,
+            timestamp=int(time.time()),
+            query=query,
+            conversation_id=message.conversation_id,
+        )
+        add_variables_to_pool(
+            variable_pool,
+            build_bootstrap_variables(system_variables=system_inputs, environment_variables=[]),
+        )
+        add_node_inputs_to_pool(variable_pool, node_id=root_node_id, inputs=inputs)
+        return variable_pool
+
+    @staticmethod
+    def _resolve_user_from(application_generate_entity: CompletionAppGenerateEntity) -> UserFrom:
+        if application_generate_entity.invoke_from.runs_as_account():
+            return UserFrom.ACCOUNT
+        return UserFrom.END_USER
+
+    @staticmethod
+    def _resolve_image_detail_config(
+        application_generate_entity: CompletionAppGenerateEntity,
+    ) -> ImagePromptMessageContent.DETAIL:
+        file_upload_config = application_generate_entity.file_upload_config
+        if file_upload_config and file_upload_config.image_config:
+            return file_upload_config.image_config.detail or ImagePromptMessageContent.DETAIL.LOW
+        return ImagePromptMessageContent.DETAIL.LOW
@@ -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,
@@ -405,17 +405,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,
@@ -93,6 +93,60 @@ from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
 logger = logging.getLogger(__name__)


+def init_graph(
+    *,
+    app_id: str,
+    graph_config: Mapping[str, Any],
+    graph_runtime_state: GraphRuntimeState,
+    user_from: UserFrom,
+    invoke_from: InvokeFrom,
+    workflow_id: str = "",
+    tenant_id: str = "",
+    user_id: str = "",
+    root_node_id: str | None = None,
+    trace_session_id: str | None = None,
+    call_depth: int = 0,
+    extra_context: Mapping[str, Any] | None = None,
+) -> Graph:
+    if "nodes" not in graph_config or "edges" not in graph_config:
+        raise ValueError("nodes or edges not found in workflow graph")
+
+    if not isinstance(graph_config.get("nodes"), list):
+        raise ValueError("nodes in workflow graph must be a list")
+
+    if not isinstance(graph_config.get("edges"), list):
+        raise ValueError("edges in workflow graph must be a list")
+
+    run_context = build_dify_run_context(
+        tenant_id=tenant_id or "",
+        app_id=app_id,
+        user_id=user_id,
+        user_from=user_from,
+        invoke_from=invoke_from,
+        trace_session_id=trace_session_id,
+        extra_context=extra_context,
+    )
+    graph_init_context = DifyGraphInitContext(
+        workflow_id=workflow_id,
+        graph_config=graph_config,
+        run_context=run_context,
+        call_depth=call_depth,
+    )
+    node_factory = DifyNodeFactory.from_graph_init_context(
+        graph_init_context=graph_init_context,
+        graph_runtime_state=graph_runtime_state,
+    )
+
+    if root_node_id is None:
+        root_node_id = get_default_root_node_id(graph_config)
+
+    graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
+    if not graph:
+        raise ValueError("graph not found in workflow")
+
+    return graph
+
+
 class WorkflowBasedAppRunner:
    def __init__(
        self,
@@ -128,48 +182,18 @@ class WorkflowBasedAppRunner:
        """
        Init graph
        """
-        if "nodes" not in graph_config or "edges" not in graph_config:
-            raise ValueError("nodes or edges not found in workflow graph")
-
-        if not isinstance(graph_config.get("nodes"), list):
-            raise ValueError("nodes in workflow graph must be a list")
-
-        if not isinstance(graph_config.get("edges"), list):
-            raise ValueError("edges in workflow graph must be a list")
-
-        # Create explicit graph init context for Graph.init.
-        run_context = build_dify_run_context(
-            tenant_id=tenant_id or "",
+        return init_graph(
            app_id=self._app_id,
-            user_id=user_id,
+            graph_config=graph_config,
+            graph_runtime_state=graph_runtime_state,
            user_from=user_from,
            invoke_from=invoke_from,
+            workflow_id=workflow_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            root_node_id=root_node_id,
            trace_session_id=trace_session_id,
        )
-        graph_init_context = DifyGraphInitContext(
-            workflow_id=workflow_id,
-            graph_config=graph_config,
-            run_context=run_context,
-            call_depth=0,
-        )
-
-        # Use the provided graph_runtime_state for consistent state management
-
-        node_factory = DifyNodeFactory.from_graph_init_context(
-            graph_init_context=graph_init_context,
-            graph_runtime_state=graph_runtime_state,
-        )
-
-        if root_node_id is None:
-            root_node_id = get_default_root_node_id(graph_config)
-
-        # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
-
-        if not graph:
-            raise ValueError("graph not found in workflow")
-
-        return graph

    def _prepare_single_node_execution(
        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."""
@@ -23,6 +23,8 @@ from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from core.workflow.human_input_adapter import adapt_node_config_for_graph
 from core.workflow.node_runtime import (
+    DIFY_BEFORE_LLM_INVOKE_KEY,
+    BeforeLLMInvoke,
    DifyFileReferenceFactory,
    DifyHumanInputNodeRuntime,
    DifyPreparedLLM,
@@ -497,6 +499,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,
@@ -548,6 +551,10 @@ class DifyNodeFactory(NodeFactory):
    ) -> dict[str, object]:
        validated_node_data = cast(LLMCompatibleNodeData, node_data)
        model_instance = self._build_model_instance_for_llm_node(validated_node_data)
+        before_llm_invoke = cast(
+            BeforeLLMInvoke | None,
+            self.graph_init_params.run_context.get(DIFY_BEFORE_LLM_INVOKE_KEY),
+        )
        node_init_kwargs: dict[str, object] = {
            "credentials_provider": self._llm_credentials_provider,
            "model_factory": self._llm_model_factory,
@@ -556,6 +563,7 @@ class DifyNodeFactory(NodeFactory):
                    node_data=validated_node_data,
                    model_instance=model_instance,
                    request_metadata={"app_id": self._dify_context.app_id},
+                    before_invoke=before_llm_invoke,
                )
                if wrap_model_instance
                else model_instance
@@ -589,13 +597,22 @@ class DifyNodeFactory(NodeFactory):
        node_data: LLMCompatibleNodeData,
        model_instance: ModelInstance,
        request_metadata: Mapping[str, object] | None = None,
+        before_invoke: BeforeLLMInvoke | None = None,
    ) -> DifyPreparedLLM:
        # Only graphon's LLM node consumes the polling protocol. Keep classifier
        # and extractor nodes on the existing wrapper even if the same model
        # advertises polling support.
        if node_data.type == BuiltinNodeTypes.LLM and DifyNodeFactory._supports_plugin_llm_polling(model_instance):
-            return DifyPreparedPollingLLM(model_instance, request_metadata=request_metadata)
-        return DifyPreparedLLM(model_instance, request_metadata=request_metadata)
+            return DifyPreparedPollingLLM(
+                model_instance,
+                request_metadata=request_metadata,
+                before_invoke=before_invoke,
+            )
+        return DifyPreparedLLM(
+            model_instance,
+            request_metadata=request_metadata,
+            before_invoke=before_invoke,
+        )

    @staticmethod
    def _supports_plugin_llm_polling(model_instance: ModelInstance) -> bool:
@@ -94,6 +94,8 @@ if TYPE_CHECKING:
    from graphon.nodes.tool.entities import ToolNodeData


+DIFY_BEFORE_LLM_INVOKE_KEY = "_dify_before_llm_invoke"
+BeforeLLMInvoke = Callable[[Sequence[PromptMessage], Mapping[str, Any]], Mapping[str, Any]]
 _file_access_controller = DatabaseFileAccessController()


@@ -150,9 +152,15 @@ class DifyFileReferenceFactory(FileReferenceFactoryProtocol):
 class DifyPreparedLLM(LLMProtocol):
    """Workflow-layer adapter that hides the full `ModelInstance` API from `graphon` nodes."""

-    def __init__(self, model_instance: ModelInstance, request_metadata: Mapping[str, object] | None = None) -> None:
+    def __init__(
+        self,
+        model_instance: ModelInstance,
+        request_metadata: Mapping[str, object] | None = None,
+        before_invoke: BeforeLLMInvoke | None = None,
+    ) -> None:
        self._model_instance = model_instance
        self._request_metadata = request_metadata
+        self._before_invoke = before_invoke

    @property
    @override
@@ -193,6 +201,15 @@ class DifyPreparedLLM(LLMProtocol):
    def get_llm_num_tokens(self, prompt_messages: Sequence[PromptMessage]) -> int:
        return self._model_instance.get_llm_num_tokens(prompt_messages)

+    def _run_before_invoke(
+        self,
+        prompt_messages: Sequence[PromptMessage],
+        model_parameters: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        if self._before_invoke is None:
+            return model_parameters
+        return self._before_invoke(prompt_messages, model_parameters)
+
    @overload
    def invoke_llm(
        self,
@@ -225,6 +242,7 @@ class DifyPreparedLLM(LLMProtocol):
        stop: Sequence[str] | None,
        stream: bool,
    ) -> LLMResult | Generator[LLMResultChunk, None, None]:
+        model_parameters = self._run_before_invoke(prompt_messages, model_parameters)
        return self._model_instance.invoke_llm(
            prompt_messages=list(prompt_messages),
            model_parameters=dict(model_parameters),
@@ -266,6 +284,7 @@ class DifyPreparedLLM(LLMProtocol):
        stop: Sequence[str] | None,
        stream: bool,
    ) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]:
+        model_parameters = self._run_before_invoke(prompt_messages, model_parameters)
        return invoke_llm_with_structured_output(
            provider=self.provider,
            model_schema=self.get_model_schema(),
@@ -285,10 +304,15 @@ class DifyPreparedLLM(LLMProtocol):
 class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
    """Prepared workflow LLM adapter that exposes Graphon's polling protocol."""

-    def __init__(self, model_instance: ModelInstance, request_metadata: Mapping[str, object] | None = None) -> None:
+    def __init__(
+        self,
+        model_instance: ModelInstance,
+        request_metadata: Mapping[str, object] | None = None,
+        before_invoke: BeforeLLMInvoke | None = None,
+    ) -> None:
        from core.plugin.impl.model_runtime import PluginModelRuntime

-        super().__init__(model_instance, request_metadata=request_metadata)
+        super().__init__(model_instance, request_metadata=request_metadata, before_invoke=before_invoke)
        model_type_instance = model_instance.model_type_instance
        if not isinstance(model_type_instance, LargeLanguageModel):
            raise TypeError("Polling wrapper requires a large-language-model instance.")
@@ -309,6 +333,7 @@ class DifyPreparedPollingLLM(DifyPreparedLLM, LLMPollingCapableProtocol):
        stop: Sequence[str] | None,
        json_schema: Mapping[str, Any] | None,
    ) -> LLMPollingResult:
+        model_parameters = self._run_before_invoke(prompt_messages, model_parameters)
        return self._plugin_model_runtime.start_llm_polling(
            provider=self.provider,
            model=self.model_name,
@@ -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)
@@ -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"
    )


@@ -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],
@@ -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,
@@ -40,7 +40,7 @@ class AppTaskService:
        # Legacy mechanism: Set stop flag in Redis
        AppQueueManager.set_stop_flag(task_id, invoke_from, user_id)

-        # New mechanism: Send stop command via GraphEngine for workflow-based apps
-        # This ensures proper workflow status recording in the persistence layer
-        if app_mode in (AppMode.ADVANCED_CHAT, AppMode.WORKFLOW):
+        # New mechanism: send stop command via GraphEngine for graph-backed apps.
+        # Completion uses WorkflowEntry at runtime but keeps legacy message persistence.
+        if app_mode in (AppMode.ADVANCED_CHAT, AppMode.WORKFLOW, AppMode.COMPLETION):
            GraphEngineManager(redis_client).send_stop_command(task_id)
@@ -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 = {
@@ -135,7 +135,45 @@ class WorkflowConverter:
            app_model=app_model, app_model_config=app_model_config, session=session
        )

-        # init workflow graph
+        graph, features = self.build_graph_from_app_config(
+            app_model=app_model,
+            app_config=app_config,
+            target_app_mode=new_app_mode,
+            session=session,
+        )
+
+        # create workflow record
+        workflow = Workflow(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            type=WorkflowType.from_app_mode(new_app_mode).value,
+            version=Workflow.VERSION_DRAFT,
+            graph=json.dumps(graph),
+            features=json.dumps(features),
+            created_by=account_id,
+            environment_variables=[],
+            conversation_variables=[],
+        )
+
+        session.add(workflow)
+        session.commit()
+
+        return workflow
+
+    def build_graph_from_app_config(
+        self,
+        *,
+        app_model: App,
+        app_config: EasyUIBasedAppConfig,
+        target_app_mode: AppMode,
+        session: Session,
+    ) -> tuple[WorkflowGraph, dict[str, Any]]:
+        """
+        Build a workflow graph from an EasyUI app config without persisting it.
+
+        This is shared by the persisted app-conversion flow and runtime-only
+        execution paths that need a graph but must not create a Workflow row.
+        """
        graph: WorkflowGraph = {"nodes": [], "edges": []}

        # Convert list:
@@ -168,7 +206,7 @@ class WorkflowConverter:
        # convert to knowledge retrieval node
        if app_config.dataset:
            knowledge_retrieval_node = self._convert_to_knowledge_retrieval_node(
-                new_app_mode=new_app_mode, dataset_config=app_config.dataset, model_config=app_config.model
+                new_app_mode=target_app_mode, dataset_config=app_config.dataset, model_config=app_config.model
            )

            if knowledge_retrieval_node:
@@ -177,7 +215,7 @@ class WorkflowConverter:
        # convert to llm node
        llm_node = self._convert_to_llm_node(
            original_app_mode=AppMode.value_of(app_model.mode),
-            new_app_mode=new_app_mode,
+            new_app_mode=target_app_mode,
            graph=graph,
            model_config=app_config.model,
            prompt_template=app_config.prompt_template,
@@ -189,7 +227,7 @@ class WorkflowConverter:

        app_model_config_dict = app_config.app_model_config_dict

-        match new_app_mode:
+        match target_app_mode:
            case AppMode.WORKFLOW:
                end_node = self._convert_to_end_node()
                graph = self._append_node(graph, end_node)
@@ -220,23 +258,7 @@ class WorkflowConverter:
                    "sensitive_word_avoidance": app_model_config_dict.get("sensitive_word_avoidance"),
                }

-        # create workflow record
-        workflow = Workflow(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            type=WorkflowType.from_app_mode(new_app_mode).value,
-            version=Workflow.VERSION_DRAFT,
-            graph=json.dumps(graph),
-            features=json.dumps(features),
-            created_by=account_id,
-            environment_variables=[],
-            conversation_variables=[],
-        )
-
-        session.add(workflow)
-        session.commit()
-
-        return workflow
+        return graph, features

    def _convert_to_app_config(
        self, app_model: App, app_model_config: AppModelConfig, *, session: Session
@@ -573,8 +595,7 @@ class WorkflowConverter:
        if new_app_mode == AppMode.ADVANCED_CHAT:
            memory = {"role_prefix": role_prefix, "window": {"enabled": False}}

-        completion_params = model_config.parameters
-        completion_params.update({"stop": model_config.stop})
+        completion_params = {**model_config.parameters, "stop": model_config.stop}
        return {
            "id": "llm",
            "position": None,
@@ -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)
@@ -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="[email protected]")
-    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="[email protected]")
+    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
@@ -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(
@@ -4,6 +4,7 @@ from dotenv import dotenv_values

 BASE_API_AND_DOCKER_CONFIG_SET_DIFF: frozenset[str] = frozenset(
    (
+        "AGENT_BACKEND_API_TOKEN",
        "APP_MAX_EXECUTION_TIME",
        "BATCH_UPLOAD_LIMIT",
        "CELERY_BEAT_SCHEDULER_TIME",
@@ -43,6 +44,7 @@ BASE_API_AND_DOCKER_CONFIG_SET_DIFF: frozenset[str] = frozenset(

 BASE_API_AND_DOCKER_COMPOSE_CONFIG_SET_DIFF: frozenset[str] = frozenset(
    (
+        "AGENT_BACKEND_API_TOKEN",
        "BATCH_UPLOAD_LIMIT",
        "CELERY_BEAT_SCHEDULER_TIME",
        "HTTP_REQUEST_MAX_CONNECT_TIMEOUT",
@@ -1,11 +1,13 @@
 import os
+import shutil
 from collections.abc import Iterator
+from pathlib import Path
 from unittest.mock import MagicMock, patch

 import pytest
 from flask import Flask
 from sqlalchemy import create_engine
-from sqlalchemy.engine import Engine
+from sqlalchemy.engine import URL, Engine
 from sqlalchemy.orm import Session, sessionmaker

 # Getting the absolute path of the current file's directory
@@ -35,7 +37,7 @@ os.environ.setdefault("OPENDAL_SCHEME", "fs")
 os.environ.setdefault("OPENDAL_FS_ROOT", "/tmp/dify-storage")
 os.environ.setdefault("STORAGE_TYPE", "opendal")

-from core.db.session_factory import configure_session_factory, session_factory
+import core.db.session_factory as session_factory_module
 from extensions import ext_redis
 from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
 from models.base import TypeBase
@@ -111,42 +113,70 @@ def reset_secret_key():
        dify_config.SECRET_KEY = original


-@pytest.fixture(scope="session")
-def _unit_test_engine():
-    engine = create_engine("sqlite:///:memory:")
-    yield engine
-    engine.dispose()
-
-
@pytest.fixture
-def sqlite_engine() -> Iterator[Engine]:
-    """Create an isolated in-memory SQLite engine for tests that need a disposable database."""
+def _sqlite_engine(_sqlite_database_template: Path, tmp_path: Path) -> Iterator[Engine]:
+    """Create an engine over a pristine per-test copy of the SQLite schema."""
+
+    database_path = tmp_path / "unit-tests.sqlite3"
+    shutil.copyfile(_sqlite_database_template, database_path)
+    engine = create_engine(URL.create("sqlite", database=str(database_path)))

-    engine = create_engine("sqlite:///:memory:")
    try:
        yield engine
    finally:
        engine.dispose()
+        database_path.unlink(missing_ok=True)


-@pytest.fixture
-def sqlite_session(request: pytest.FixtureRequest, sqlite_engine: Engine) -> Iterator[Session]:
-    """Yield a SQLite session after creating the model tables passed through ``request.param``."""
+@pytest.fixture(scope="session")
+def _sqlite_database_template(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Create one empty full-schema SQLite database per pytest worker."""

-    models: tuple[type[TypeBase], ...] = request.param
-    tables = [model.metadata.tables[model.__tablename__] for model in models]
-    TypeBase.metadata.create_all(sqlite_engine, tables=tables)
-    session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
-    with session_factory() as session:
-        yield session
+    database_path = tmp_path_factory.mktemp("sqlite-template") / "unit-tests.sqlite3"
+    engine = create_engine(URL.create("sqlite", database=str(database_path)))
+    try:
+        TypeBase.metadata.create_all(engine)
+    finally:
+        engine.dispose()
+    return database_path


@pytest.fixture(autouse=True)
-def _configure_session_factory(_unit_test_engine):
-    try:
-        session_factory.get_session_maker()
-    except RuntimeError:
-        configure_session_factory(_unit_test_engine, expire_on_commit=False)
+def _sqlite_session_factory(
+    _sqlite_engine: Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> sessionmaker[Session]:
+    """Bind all unit-test Sessions to the pristine full-schema SQLite database."""
+
+    factory = sessionmaker(bind=_sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr(session_factory_module, "_session_maker", factory)
+    return factory
+
+
+@pytest.fixture
+def sqlite_engine(_sqlite_engine: Engine) -> Engine:
+    """Expose the pristine full-schema SQLite engine to tests."""
+
+    return _sqlite_engine
+
+
+@pytest.fixture
+def sqlite_session_factory(_sqlite_session_factory: sessionmaker[Session]) -> sessionmaker[Session]:
+    """Expose the shared SQLite session factory to tests."""
+
+    return _sqlite_session_factory
+
+
+@pytest.fixture
+def sqlite_session(_sqlite_session_factory: sessionmaker[Session]) -> Iterator[Session]:
+    """Yield a session over the pristine full-schema SQLite database.
+
+    Legacy indirect model parameters remain accepted by pytest but are ignored.
+    Remove those decorators as their test files receive individual review.
+    """
+
+    with _sqlite_session_factory() as session:
+        yield session


 def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:
@@ -115,6 +115,7 @@ def _agent_app_composer_response() -> dict:
            "active_config_snapshot_id": "version-1",
        },
        "active_config_snapshot": _version_response(),
+        "active_config_is_published": True,
        "agent_soul": {},
        "save_options": ["save_to_current_version"],
    }
@@ -376,7 +377,7 @@ def test_agent_app_list_and_create_use_agent_route(
    assert created["app_id"] == "app-created"
    assert created["debug_conversation_id"] == "debug-conversation-created"
    assert created["role"] == "Created role"
-    assert created["active_config_is_published"] is False
+    assert "active_config_is_published" not in created
    assert "bound_agent_id" not in created
    create_call = cast(dict[str, object], captured["create"])
    create_params = cast(Any, create_call["params"])
@@ -487,7 +488,7 @@ def test_agent_app_detail_update_delete_resolve_app_from_agent_id(
    assert detail["debug_conversation_has_messages"] is True
    assert detail["debug_conversation_message_count"] == 2
    assert detail["role"] == "Resolved role"
-    assert detail["active_config_is_published"] is False
+    assert "active_config_is_published" not in detail
    assert "bound_agent_id" not in detail
    assert captured["get_app"] == {"app": app_model, "session": session}
    with app.test_request_context(
@@ -502,7 +503,7 @@ def test_agent_app_detail_update_delete_resolve_app_from_agent_id(
    assert updated["debug_conversation_has_messages"] is True
    assert updated["debug_conversation_message_count"] == 2
    assert updated["role"] == "Resolved role"
-    assert updated["active_config_is_published"] is False
+    assert "active_config_is_published" not in updated
    assert "bound_agent_id" not in updated
    update_call = cast(dict[str, object], captured["update"])
    assert update_call["app"] is app_model
@@ -845,9 +846,6 @@ def test_agent_app_update_allows_empty_role(app: Flask, monkeypatch: pytest.Monk
    monkeypatch.setattr(
        roster_controller.AgentRosterService, "count_agent_app_debug_conversation_messages", lambda _self, **kwargs: 0
    )
-    monkeypatch.setattr(
-        roster_controller.AgentRosterService, "active_config_is_published", lambda _self, **kwargs: False
-    )
    monkeypatch.setattr(
        roster_controller.FeatureService,
        "get_system_features",
@@ -1299,13 +1297,14 @@ def test_agent_composer_routes_resolve_app_from_agent_id(
        composer_controller.AgentComposerService, "collect_validation_findings", collect_validation_findings
    )
    monkeypatch.setattr(composer_controller.AgentComposerService, "get_agent_app_candidates", get_agent_app_candidates)
-    assert unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", agent_id)["variant"] == "agent_app"
+    composer = unwrap(AgentComposerApi.get)(AgentComposerApi(), MagicMock(), "tenant-1", agent_id)
+    assert composer["variant"] == "agent_app"
+    assert composer["active_config_is_published"] is True
    assert cast(dict[str, object], captured["load"])["agent_id"] == agent_id
    with app.test_request_context(json=payload):
-        assert (
-            unwrap(AgentComposerApi.put)(AgentComposerApi(), MagicMock(), "tenant-1", account_id, agent_id)["variant"]
-            == "agent_app"
-        )
+        saved_composer = unwrap(AgentComposerApi.put)(AgentComposerApi(), MagicMock(), "tenant-1", account_id, agent_id)
+        assert saved_composer["variant"] == "agent_app"
+        assert saved_composer["active_config_is_published"] is True
        assert cast(dict[str, object], captured["save"])["agent_id"] == agent_id
        assert unwrap(AgentComposerValidateApi.post)(AgentComposerValidateApi(), MagicMock(), "tenant-1", agent_id) == {
            "result": "success",
@@ -708,6 +708,121 @@ def test_app_list_api_attaches_permission_keys(app, app_module):
    assert resp["data"][0]["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]


+def test_recent_app_list_api_returns_only_home_card_fields(app, app_module):
+    method = app_module.RecentAppListApi.get
+    while hasattr(method, "__wrapped__"):
+        method = method.__wrapped__
+
+    recent_app = SimpleNamespace(
+        id="app-1",
+        name="Recent App",
+        icon_type="emoji",
+        icon="🚀",
+        icon_background="#FFFFFF",
+        mode="chat",
+        author_name="Recent Author",
+        updated_at=_ts(15),
+        maintainer="acct-1",
+    )
+    get_recent_apps = MagicMock(return_value=[recent_app])
+
+    with app.test_request_context("/apps/recent?limit=8"):
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(dify_config, "RBAC_ENABLED", False)
+            monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
+            monkeypatch.setattr(
+                app_module.enterprise_rbac_service.RBACService.MyPermissions,
+                "get",
+                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
+                    app=app_module.enterprise_rbac_service.ResourcePermissionSnapshot(
+                        overrides=[
+                            app_module.enterprise_rbac_service.ResourcePermissionKeys(
+                                resource_id="app-1",
+                                permission_keys=["app.acl.monitor"],
+                            )
+                        ]
+                    )
+                ),
+            )
+
+            resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())
+
+    assert status == 200
+    assert resp == {
+        "data": [
+            {
+                "id": "app-1",
+                "name": "Recent App",
+                "icon_type": "emoji",
+                "icon": "🚀",
+                "icon_background": "#FFFFFF",
+                "mode": "chat",
+                "author_name": "Recent Author",
+                "updated_at": int(_ts(15).timestamp()),
+                "permission_keys": ["app.acl.monitor"],
+                "maintainer": "acct-1",
+                "icon_url": None,
+            }
+        ]
+    }
+    params = get_recent_apps.call_args.args[2]
+    assert params.limit == 8
+    assert "total" not in resp
+    assert "description" not in resp["data"][0]
+    assert "tags" not in resp["data"][0]
+    assert "workflow" not in resp["data"][0]
+
+
+@pytest.mark.parametrize("mode", ["channel", "rag-pipeline", "agent"])
+def test_recent_app_response_rejects_non_home_app_modes(app_module, mode: str) -> None:
+    with pytest.raises(ValidationError):
+        app_module.RecentAppResponse.model_validate(
+            {
+                "id": "app-1",
+                "name": "Recent App",
+                "mode": mode,
+                "updated_at": _ts(),
+            }
+        )
+
+
+def test_recent_app_list_api_applies_rbac_visibility_filter(app, app_module):
+    method = app_module.RecentAppListApi.get
+    while hasattr(method, "__wrapped__"):
+        method = method.__wrapped__
+
+    get_recent_apps = MagicMock(return_value=[])
+    with app.test_request_context("/apps/recent"):
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
+            monkeypatch.setattr(app_module.AppService, "get_recent_apps", get_recent_apps)
+            monkeypatch.setattr(
+                app_module.enterprise_rbac_service.RBACService.MyPermissions,
+                "get",
+                lambda tenant_id, account_id, session: app_module.enterprise_rbac_service.MyPermissionsResponse(
+                    workspace=app_module.enterprise_rbac_service.WorkspacePermissionSnapshot(
+                        permission_keys=["app.create_and_management"]
+                    )
+                ),
+            )
+            monkeypatch.setattr(
+                app_module.enterprise_rbac_service.RBACService.AppAccess,
+                "whitelist_resources",
+                lambda tenant_id, account_id: SimpleNamespace(
+                    unrestricted=False,
+                    resource_ids=["app-shared"],
+                ),
+            )
+
+            resp, status = method(app_module.RecentAppListApi(), "tenant-1", "acct-1", MagicMock())
+
+    assert status == 200
+    assert resp == {"data": []}
+    params = get_recent_apps.call_args.args[2]
+    assert params.accessible_app_ids == ["app-shared"]
+    assert params.include_own_apps is True
+
+
 def test_app_list_api_limits_to_apps_created_by_current_user_without_view_permission(app, app_module):
    method = app_module.AppListApi.get
    while hasattr(method, "__wrapped__"):
@@ -321,6 +321,33 @@ def test_console_text_api_builds_message_ref(app: Flask, monkeypatch: pytest.Mon
    assert calls["message_ref"] == MessageRef(AppRef("tenant-1", "app-1"), "message-1", account_id="account-1")


+def test_console_text_api_accepts_message_id_without_text(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = {}
+    monkeypatch.setattr(AudioService, "transcript_tts", lambda **kwargs: calls.update(kwargs) or {"audio": "ok"})
+
+    api = ChatMessageTextApi()
+    handler = unwrap(api.post)
+    app_model = SimpleNamespace(id="app-1", tenant_id="tenant-1")
+
+    with (
+        app.test_request_context(
+            "/console/api/apps/app/text-to-audio",
+            method="POST",
+            json={"message_id": "0f67f8c5-8f7c-4ebd-b549-7ac8e972d37e", "streaming": True},
+        ),
+        patch("controllers.console.app.audio.current_user", SimpleNamespace(id="account-1")),
+    ):
+        response = handler(api, app_model=app_model)
+
+    assert response == {"audio": "ok"}
+    assert calls["text"] == ""
+    assert calls["message_ref"] == MessageRef(
+        AppRef("tenant-1", "app-1"),
+        "0f67f8c5-8f7c-4ebd-b549-7ac8e972d37e",
+        account_id="account-1",
+    )
+
+
 def test_console_text_api_error_mapping(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(AudioService, "transcript_tts", lambda **_kwargs: (_ for _ in ()).throw(QuotaExceededError()))

@@ -1,18 +1,22 @@
 from __future__ import annotations

 import inspect
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from datetime import UTC, datetime
-from typing import cast
+from typing import Literal, cast
 from unittest.mock import MagicMock, PropertyMock, patch
-from uuid import uuid4
+from uuid import UUID

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

 from controllers.console.datasets import data_source as module
 from controllers.console.datasets.data_source import DataSourceApi, DataSourceNotionListApi
 from models import Account, DataSourceOauthBinding
+from models.engine import db

 ControllerMethod = Callable[..., tuple[dict[str, object], int]]

@@ -22,10 +26,15 @@ def unwrap(func: object) -> ControllerMethod:


@pytest.fixture
-def flask_app() -> Flask:
+def flask_app() -> Iterator[Flask]:
    app = Flask(__name__)
    app.config["TESTING"] = True
-    return app
+    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
+    db.init_app(app)
+
+    with app.app_context():
+        DataSourceOauthBinding.__table__.create(db.engine)
+        yield app


@pytest.fixture
@@ -35,9 +44,13 @@ def current_user() -> Account:
    return account


-def test_get_data_source_integrates_serializes_orm_binding(flask_app: Flask) -> None:
+TENANT_ID = "11111111-1111-1111-1111-111111111111"
+BINDING_ID = "22222222-2222-2222-2222-222222222222"
+
+
+def _add_binding(session: Session, *, disabled: bool) -> DataSourceOauthBinding:
    binding = DataSourceOauthBinding(
-        tenant_id="tenant-1",
+        tenant_id=TENANT_ID,
        access_token="token",
        provider="notion",
        source_info={
@@ -55,24 +68,31 @@ def test_get_data_source_integrates_serializes_orm_binding(flask_app: Flask) ->
                }
            ],
        },
+        disabled=disabled,
    )
-    binding.id = "binding-1"
+    binding.id = BINDING_ID
    binding.created_at = datetime(2026, 5, 25, 1, 2, 3, tzinfo=UTC)
-    binding.disabled = False
+    session.add(binding)
+    session.commit()
+    return binding

-    with (
-        flask_app.test_request_context("/"),
-        patch.object(module.db.session, "scalars", return_value=MagicMock(all=lambda: [binding])),
-    ):
-        response, status = unwrap(DataSourceApi().get)(DataSourceApi(), "tenant-1")
+
+def test_get_data_source_integrates_serializes_orm_binding(
+    flask_app: Flask,
+) -> None:
+    binding = _add_binding(db.session, disabled=False)
+    expected_created_at = int(binding.created_at.timestamp())
+
+    with flask_app.test_request_context("/"):
+        response, status = unwrap(DataSourceApi().get)(DataSourceApi(), TENANT_ID)

    assert status == 200
    assert response == {
        "data": [
            {
-                "id": "binding-1",
+                "id": BINDING_ID,
                "provider": "notion",
-                "created_at": 1779670923,
+                "created_at": expected_created_at,
                "is_bound": True,
                "disabled": False,
                "source_info": {
@@ -96,34 +116,75 @@ def test_get_data_source_integrates_serializes_orm_binding(flask_app: Flask) ->
    }


-def test_get_data_source_integrates_preserves_empty_list_when_no_binding(flask_app: Flask) -> None:
-    with (
-        flask_app.test_request_context("/"),
-        patch.object(module.db.session, "scalars", return_value=MagicMock(all=lambda: [])),
-    ):
-        response, status = unwrap(DataSourceApi().get)(DataSourceApi(), "tenant-1")
+def test_get_data_source_integrates_preserves_empty_list_when_no_binding(
+    flask_app: Flask,
+) -> None:
+    with flask_app.test_request_context("/"):
+        response, status = unwrap(DataSourceApi().get)(DataSourceApi(), TENANT_ID)

    assert status == 200
    assert response == {"data": []}


-def test_patch_data_source_binding_uses_injected_session(flask_app: Flask) -> None:
-    binding = MagicMock(disabled=True)
-    session = MagicMock()
-    session.scalar.return_value = binding
+@pytest.mark.parametrize(
+    ("disabled", "action", "expected_disabled"),
+    [(True, "enable", False), (False, "disable", True)],
+)
+@pytest.mark.parametrize("sqlite_session", [(DataSourceOauthBinding,)], indirect=True)
+def test_patch_data_source_binding_updates_state(
+    flask_app: Flask,
+    sqlite_session: Session,
+    disabled: bool,
+    action: Literal["enable", "disable"],
+    expected_disabled: bool,
+) -> None:
+    _add_binding(sqlite_session, disabled=disabled)
+    sqlite_session.expunge_all()

    with flask_app.test_request_context("/"):
-        response, status = unwrap(DataSourceApi().patch)(DataSourceApi(), session, "tenant-1", uuid4(), "enable")
+        response, status = unwrap(DataSourceApi().patch)(
+            DataSourceApi(), sqlite_session, TENANT_ID, UUID(BINDING_ID), action
+        )

+    sqlite_session.flush()
+    sqlite_session.expire_all()
+    binding = sqlite_session.scalar(select(DataSourceOauthBinding).where(DataSourceOauthBinding.id == BINDING_ID))
    assert status == 200
    assert response == {"result": "success"}
-    assert binding.disabled is False
-    session.scalar.assert_called_once()
-    session.add.assert_not_called()
-    session.commit.assert_not_called()
+    assert binding is not None
+    assert binding.disabled is expected_disabled


-def test_notion_pre_import_pages_serializes_frontend_list_shape(flask_app: Flask, current_user: Account) -> None:
+@pytest.mark.parametrize("sqlite_session", [(DataSourceOauthBinding,)], indirect=True)
+def test_patch_data_source_binding_rejects_unknown_binding(
+    flask_app: Flask,
+    sqlite_session: Session,
+) -> None:
+    with flask_app.test_request_context("/"), pytest.raises(NotFound, match="Data source binding not found"):
+        unwrap(DataSourceApi().patch)(DataSourceApi(), sqlite_session, TENANT_ID, UUID(BINDING_ID), "enable")
+
+
+@pytest.mark.parametrize(("disabled", "action"), [(False, "enable"), (True, "disable")])
+@pytest.mark.parametrize("sqlite_session", [(DataSourceOauthBinding,)], indirect=True)
+def test_patch_data_source_binding_rejects_current_state(
+    flask_app: Flask,
+    sqlite_session: Session,
+    disabled: bool,
+    action: Literal["enable", "disable"],
+) -> None:
+    _add_binding(sqlite_session, disabled=disabled)
+    sqlite_session.expunge_all()
+
+    with flask_app.test_request_context("/"), pytest.raises(ValueError):
+        unwrap(DataSourceApi().patch)(DataSourceApi(), sqlite_session, TENANT_ID, UUID(BINDING_ID), action)
+
+
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_notion_pre_import_pages_serializes_frontend_list_shape(
+    flask_app: Flask,
+    current_user: Account,
+    sqlite_session: Session,
+) -> None:
    page = MagicMock(
        page_id="page-1",
        page_name="Page",
@@ -145,8 +206,6 @@ def test_notion_pre_import_pages_serializes_frontend_list_shape(flask_app: Flask
        get_online_document_pages=MagicMock(return_value=iter([online_document_message])),
        datasource_provider_type=MagicMock(return_value="online_document"),
    )
-    session = MagicMock()
-
    with (
        flask_app.test_request_context("/?credential_id=credential-1"),
        patch.object(
@@ -158,7 +217,7 @@ def test_notion_pre_import_pages_serializes_frontend_list_shape(flask_app: Flask
        patch("core.datasource.datasource_manager.DatasourceManager.get_datasource_runtime", return_value=runtime),
    ):
        response, status = unwrap(DataSourceNotionListApi().get)(
-            DataSourceNotionListApi(), session, "tenant-1", current_user
+            DataSourceNotionListApi(), sqlite_session, "tenant-1", current_user
        )

    assert status == 200
@@ -183,3 +242,38 @@ def test_notion_pre_import_pages_serializes_frontend_list_shape(flask_app: Flask
    }
    runtime.get_online_document_pages.assert_called_once()
    assert runtime.get_online_document_pages.call_args.kwargs["datasource_parameters"] == {}
+
+
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_notion_pre_import_pages_rejects_missing_credential(
+    flask_app: Flask,
+    current_user: Account,
+    sqlite_session: Session,
+) -> None:
+    with (
+        flask_app.test_request_context("/?credential_id=credential-1"),
+        patch.object(module.DatasourceProviderService, "get_datasource_credentials", return_value=None),
+        pytest.raises(NotFound, match="Credential not found"),
+    ):
+        unwrap(DataSourceNotionListApi().get)(DataSourceNotionListApi(), sqlite_session, TENANT_ID, current_user)
+
+
+@pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+def test_notion_pre_import_pages_rejects_non_notion_dataset(
+    flask_app: Flask,
+    current_user: Account,
+    sqlite_session: Session,
+) -> None:
+    dataset = MagicMock(data_source_type="other_type")
+
+    with (
+        flask_app.test_request_context("/?credential_id=credential-1&dataset_id=dataset-1"),
+        patch.object(
+            module.DatasourceProviderService,
+            "get_datasource_credentials",
+            return_value={"token": "token"},
+        ),
+        patch.object(module.DatasetService, "get_dataset", return_value=dataset),
+        pytest.raises(ValueError, match="Dataset is not notion type"),
+    ):
+        unwrap(DataSourceNotionListApi().get)(DataSourceNotionListApi(), sqlite_session, TENANT_ID, current_user)
@@ -0,0 +1,171 @@
+"""Unit tests for controllers.console.datasets.data_source Notion endpoints."""
+
+from __future__ import annotations
+
+import inspect
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.console.datasets.data_source import (
+    DataSourceNotionDatasetSyncApi,
+    DataSourceNotionDocumentSyncApi,
+    DataSourceNotionIndexingEstimateApi,
+    DataSourceNotionPreviewApi,
+)
+from core.rag.index_processor.constant.index_type import IndexStructureType
+from models import Account
+
+
+@pytest.fixture
+def current_user() -> Account:
+    account = Account(name="Test User", email="[email protected]")
+    account.id = "u1"
+    return account
+
+
+class TestDataSourceNotionPreviewApi:
+    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.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_post_indexing_estimate_success(self, app: Flask, sqlite_session: Session) -> 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, sqlite_session, "tenant-1")
+
+        assert status == 200
+
+
+class TestDataSourceNotionDatasetSyncApi:
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_get_success(self, app: Flask, sqlite_session: Session) -> 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, sqlite_session, "ds-1")
+
+        assert status == 200
+
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_get_dataset_not_found(self, app: Flask, sqlite_session: Session) -> 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, sqlite_session, "ds-1")
+
+
+class TestDataSourceNotionDocumentSyncApi:
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_get_success(self, app: Flask, sqlite_session: Session) -> 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, sqlite_session, "ds-1", "doc-1")
+
+        assert status == 200
+
+    @pytest.mark.parametrize("sqlite_session", [()], indirect=True)
+    def test_get_document_not_found(self, app: Flask, sqlite_session: Session) -> 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, sqlite_session, "ds-1", "doc-1")
@@ -1,17 +1,31 @@
-"""
-Unit tests for Service API File Preview endpoint
+"""Unit tests for the Service API file-preview endpoint.
+
+Ownership checks run against persisted message, file, app, and upload rows so the
+tests exercise the same SQLAlchemy statements and tenant boundary as production.
+Storage remains mocked because it is the external I/O boundary of the endpoint.
 """

 import logging
-import uuid
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime
+from decimal import Decimal
 from typing import Protocol, cast
 from unittest.mock import Mock, patch
+from uuid import uuid4

 import pytest
+from sqlalchemy import event
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session

 from controllers.service_api.app.error import FileAccessDeniedError, FileNotFoundError
 from controllers.service_api.app.file_preview import FilePreviewApi
-from models.model import App, EndUser, Message, MessageFile, UploadFile
+from extensions.storage.storage_type import StorageType
+from graphon.file import FileTransferMethod, FileType
+from models.base import TypeBase
+from models.enums import ConversationFromSource, CreatorUserRole
+from models.model import App, AppMode, Message, MessageFile, UploadFile


 class _FilePreviewLogRecord(Protocol):
@@ -20,367 +34,252 @@ class _FilePreviewLogRecord(Protocol):
    error: str


+@dataclass(frozen=True)
+class _Database:
+    """Expose the real test session through the interface used by the controller."""
+
+    session: Session
+
+
+@dataclass(frozen=True)
+class _PreviewRecords:
+    app: App
+    message: Message
+    message_file: MessageFile
+    upload_file: UploadFile
+
+
+@pytest.fixture
+def database(sqlite_engine: Engine) -> Iterator[_Database]:
+    """Create only the tables required by file ownership validation."""
+
+    models = (App, Message, MessageFile, UploadFile)
+    tables = [TypeBase.metadata.tables[model.__tablename__] for model in models]
+    TypeBase.metadata.create_all(sqlite_engine, tables=tables)
+    with Session(sqlite_engine, expire_on_commit=False) as session:
+        yield _Database(session)
+
+
+@pytest.fixture
+def file_preview_api() -> FilePreviewApi:
+    """Create the resource instance under test."""
+
+    return FilePreviewApi()
+
+
+def _upload_file(*, tenant_id: str, file_id: str | None = None) -> UploadFile:
+    upload_file = UploadFile(
+        tenant_id=tenant_id,
+        storage_type=StorageType.LOCAL,
+        key="storage/key/test_file.jpg",
+        name="test_file.jpg",
+        size=1024,
+        extension="jpg",
+        mime_type="image/jpeg",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=str(uuid4()),
+        created_at=datetime(2026, 1, 1),
+        used=True,
+    )
+    if file_id is not None:
+        upload_file.id = file_id
+    return upload_file
+
+
+def _persist_preview_records(
+    session: Session,
+    *,
+    app_id: str | None = None,
+    app_tenant_id: str | None = None,
+    upload_tenant_id: str | None = None,
+) -> _PreviewRecords:
+    app_id = app_id or str(uuid4())
+    app_tenant_id = app_tenant_id or str(uuid4())
+    upload_file = _upload_file(tenant_id=upload_tenant_id or app_tenant_id)
+    app = App(
+        id=app_id,
+        tenant_id=app_tenant_id,
+        name="Preview app",
+        description="",
+        mode=AppMode.CHAT,
+        icon_type=None,
+        icon="",
+        icon_background=None,
+        enable_site=True,
+        enable_api=True,
+    )
+    message = Message(
+        id=str(uuid4()),
+        app_id=app_id,
+        conversation_id=str(uuid4()),
+        _inputs={},
+        query="preview",
+        message={},
+        message_unit_price=Decimal(0),
+        answer="answer",
+        answer_unit_price=Decimal(0),
+        currency="USD",
+        from_source=ConversationFromSource.API,
+    )
+    message_file = MessageFile(
+        message_id=message.id,
+        type=FileType.IMAGE,
+        transfer_method=FileTransferMethod.LOCAL_FILE,
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=str(uuid4()),
+        upload_file_id=upload_file.id,
+    )
+    session.add_all([app, message, message_file, upload_file])
+    session.commit()
+    return _PreviewRecords(app=app, message=message, message_file=message_file, upload_file=upload_file)
+
+
 class TestFilePreviewApi:
-    """Test suite for FilePreviewApi"""
+    """Exercise ownership validation and response construction."""

-    @pytest.fixture
-    def file_preview_api(self):
-        """Create FilePreviewApi instance for testing"""
-        return FilePreviewApi()
+    def test_validate_file_ownership_success(self, file_preview_api: FilePreviewApi, database: _Database):
+        records = _persist_preview_records(database.session)

-    @pytest.fixture
-    def mock_app(self):
-        """Mock App model"""
-        app = Mock(spec=App)
-        app.id = str(uuid.uuid4())
-        app.tenant_id = str(uuid.uuid4())
-        return app
+        with patch("controllers.service_api.app.file_preview.db", database):
+            message_file, upload_file = file_preview_api._validate_file_ownership(
+                records.upload_file.id, records.app.id
+            )

-    @pytest.fixture
-    def mock_end_user(self):
-        """Mock EndUser model"""
-        end_user = Mock(spec=EndUser)
-        end_user.id = str(uuid.uuid4())
-        return end_user
+        assert message_file.id == records.message_file.id
+        assert upload_file.id == records.upload_file.id
+        assert upload_file.tenant_id == records.app.tenant_id

-    @pytest.fixture
-    def mock_upload_file(self):
-        """Mock UploadFile model"""
-        upload_file = Mock(spec=UploadFile)
-        upload_file.id = str(uuid.uuid4())
-        upload_file.name = "test_file.jpg"
-        upload_file.extension = "jpg"
-        upload_file.mime_type = "image/jpeg"
-        upload_file.size = 1024
-        upload_file.key = "storage/key/test_file.jpg"
-        upload_file.tenant_id = str(uuid.uuid4())
-        return upload_file
+    def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database):
+        with patch("controllers.service_api.app.file_preview.db", database):
+            with pytest.raises(FileNotFoundError, match="File not found in message context"):
+                file_preview_api._validate_file_ownership(str(uuid4()), str(uuid4()))

-    @pytest.fixture
-    def mock_message_file(self):
-        """Mock MessageFile model"""
-        message_file = Mock(spec=MessageFile)
-        message_file.id = str(uuid.uuid4())
-        message_file.upload_file_id = str(uuid.uuid4())
-        message_file.message_id = str(uuid.uuid4())
-        return message_file
+    def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, database: _Database):
+        records = _persist_preview_records(database.session)

-    @pytest.fixture
-    def mock_message(self):
-        """Mock Message model"""
-        message = Mock(spec=Message)
-        message.id = str(uuid.uuid4())
-        message.app_id = str(uuid.uuid4())
-        return message
+        with patch("controllers.service_api.app.file_preview.db", database):
+            with pytest.raises(FileAccessDeniedError, match="not owned by requesting app"):
+                file_preview_api._validate_file_ownership(records.upload_file.id, str(uuid4()))

-    def test_validate_file_ownership_success(
-        self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message
-    ):
-        """Test successful file ownership validation"""
-        file_id = str(uuid.uuid4())
-        app_id = mock_app.id
+    def test_validate_file_ownership_upload_file_not_found(self, file_preview_api: FilePreviewApi, database: _Database):
+        records = _persist_preview_records(database.session)
+        database.session.delete(records.upload_file)
+        database.session.commit()

-        # Set up the mocks
-        mock_upload_file.tenant_id = mock_app.tenant_id
-        mock_message.app_id = app_id
-        mock_message_file.upload_file_id = file_id
-        mock_message_file.message_id = mock_message.id
+        with patch("controllers.service_api.app.file_preview.db", database):
+            with pytest.raises(FileNotFoundError, match="Upload file record not found"):
+                file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)

-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock scalar() for MessageFile and Message queries
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query
-                mock_message,  # Message query
-            ]
-            # Mock get() for UploadFile and App PK lookups
-            mock_db.session.get.side_effect = [
-                mock_upload_file,  # UploadFile query
-                mock_app,  # App query for tenant validation
-            ]
+    def test_validate_file_ownership_tenant_mismatch(self, file_preview_api: FilePreviewApi, database: _Database):
+        records = _persist_preview_records(database.session, upload_tenant_id=str(uuid4()))

-            # Execute the method
-            result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
-
-            # Assertions
-            assert result_message_file == mock_message_file
-            assert result_upload_file == mock_upload_file
-
-    def test_validate_file_ownership_file_not_found(self, file_preview_api: FilePreviewApi):
-        """Test file ownership validation when MessageFile not found"""
-        file_id = str(uuid.uuid4())
-        app_id = str(uuid.uuid4())
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock MessageFile not found via scalar()
-            mock_db.session.scalar.return_value = None
-
-            # Execute and assert exception
-            with pytest.raises(FileNotFoundError) as exc_info:
-                file_preview_api._validate_file_ownership(file_id, app_id)
-
-            assert "File not found in message context" in str(exc_info.value)
-
-    def test_validate_file_ownership_access_denied(self, file_preview_api: FilePreviewApi, mock_message_file):
-        """Test file ownership validation when Message not owned by app"""
-        file_id = str(uuid.uuid4())
-        app_id = str(uuid.uuid4())
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock MessageFile found but Message not owned by app via scalar()
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query - found
-                None,  # Message query - not found (access denied)
-            ]
-
-            # Execute and assert exception
-            with pytest.raises(FileAccessDeniedError) as exc_info:
-                file_preview_api._validate_file_ownership(file_id, app_id)
-
-            assert "not owned by requesting app" in str(exc_info.value)
-
-    def test_validate_file_ownership_upload_file_not_found(
-        self, file_preview_api: FilePreviewApi, mock_message_file, mock_message
-    ):
-        """Test file ownership validation when UploadFile not found"""
-        file_id = str(uuid.uuid4())
-        app_id = str(uuid.uuid4())
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock scalar() for MessageFile and Message
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query - found
-                mock_message,  # Message query - found
-            ]
-            # Mock get() for UploadFile - not found
-            mock_db.session.get.return_value = None
-
-            # Execute and assert exception
-            with pytest.raises(FileNotFoundError) as exc_info:
-                file_preview_api._validate_file_ownership(file_id, app_id)
-
-            assert "Upload file record not found" in str(exc_info.value)
-
-    def test_validate_file_ownership_tenant_mismatch(
-        self, file_preview_api: FilePreviewApi, mock_app, mock_upload_file, mock_message_file, mock_message
-    ):
-        """Test file ownership validation with tenant mismatch"""
-        file_id = str(uuid.uuid4())
-        app_id = mock_app.id
-
-        # Set up tenant mismatch
-        mock_upload_file.tenant_id = "different_tenant_id"
-        mock_app.tenant_id = "app_tenant_id"
-        mock_message.app_id = app_id
-        mock_message_file.upload_file_id = file_id
-        mock_message_file.message_id = mock_message.id
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock scalar() for MessageFile and Message queries
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query
-                mock_message,  # Message query
-            ]
-            # Mock get() for UploadFile and App PK lookups
-            mock_db.session.get.side_effect = [
-                mock_upload_file,  # UploadFile query
-                mock_app,  # App query for tenant validation
-            ]
-
-            # Execute and assert exception
-            with pytest.raises(FileAccessDeniedError) as exc_info:
-                file_preview_api._validate_file_ownership(file_id, app_id)
-
-            assert "tenant mismatch" in str(exc_info.value)
+        with patch("controllers.service_api.app.file_preview.db", database):
+            with pytest.raises(FileAccessDeniedError, match="tenant mismatch"):
+                file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)

    def test_validate_file_ownership_invalid_input(self, file_preview_api: FilePreviewApi):
-        """Test file ownership validation with invalid input"""
-
-        # Test with empty file_id
-        with pytest.raises(FileAccessDeniedError) as exc_info:
+        with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"):
            file_preview_api._validate_file_ownership("", "app_id")
-        assert "Invalid file or app identifier" in str(exc_info.value)

-        # Test with empty app_id
-        with pytest.raises(FileAccessDeniedError) as exc_info:
+        with pytest.raises(FileAccessDeniedError, match="Invalid file or app identifier"):
            file_preview_api._validate_file_ownership("file_id", "")
-        assert "Invalid file or app identifier" in str(exc_info.value)

-    def test_build_file_response_basic(self, file_preview_api: FilePreviewApi, mock_upload_file):
-        """Test basic file response building"""
-        mock_generator = Mock()
+    @pytest.mark.parametrize(
+        ("as_attachment", "mime_type", "name", "extension", "size"),
+        [
+            (False, "image/jpeg", "test_file.jpg", "jpg", 1024),
+            (True, "image/jpeg", "test_file.jpg", "jpg", 1024),
+            (False, "text/html", "unsafe.html", "html", 1024),
+            (False, "video/mp4", "test_file.mp4", "mp4", 1024),
+            (False, "image/jpeg", "test_file.jpg", "jpg", 0),
+        ],
+    )
+    def test_build_file_response(
+        self,
+        file_preview_api: FilePreviewApi,
+        as_attachment: bool,
+        mime_type: str,
+        name: str,
+        extension: str,
+        size: int,
+    ):
+        upload_file = _upload_file(tenant_id=str(uuid4()))
+        upload_file.mime_type = mime_type
+        upload_file.name = name
+        upload_file.extension = extension
+        upload_file.size = size

-        response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
+        response = file_preview_api._build_file_response(Mock(), upload_file, as_attachment)

-        # Check response properties
-        assert response.mimetype == mock_upload_file.mime_type
        assert response.direct_passthrough is True
-        assert response.headers["Content-Length"] == str(mock_upload_file.size)
        assert "Cache-Control" in response.headers
-
-    def test_build_file_response_as_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file):
-        """Test file response building with attachment flag"""
-        mock_generator = Mock()
-
-        response = file_preview_api._build_file_response(mock_generator, mock_upload_file, True)
-
-        # Check attachment-specific headers
-        assert "attachment" in response.headers["Content-Disposition"]
-        assert mock_upload_file.name in response.headers["Content-Disposition"]
-        assert response.headers["Content-Type"] == "application/octet-stream"
-
-    def test_build_file_response_html_forces_attachment(self, file_preview_api: FilePreviewApi, mock_upload_file):
-        """Test HTML files are forced to download"""
-        mock_generator = Mock()
-        mock_upload_file.mime_type = "text/html"
-        mock_upload_file.name = "unsafe.html"
-        mock_upload_file.extension = "html"
-
-        response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
-
-        assert "attachment" in response.headers["Content-Disposition"]
-        assert response.headers["Content-Type"] == "application/octet-stream"
-        assert response.headers["X-Content-Type-Options"] == "nosniff"
-
-    def test_build_file_response_audio_video(self, file_preview_api: FilePreviewApi, mock_upload_file):
-        """Test file response building for audio/video files"""
-        mock_generator = Mock()
-        mock_upload_file.mime_type = "video/mp4"
-
-        response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
-
-        # Check Range support for media files
-        assert response.headers["Accept-Ranges"] == "bytes"
-
-    def test_build_file_response_no_size(self, file_preview_api: FilePreviewApi, mock_upload_file):
-        """Test file response building when size is unknown"""
-        mock_generator = Mock()
-        mock_upload_file.size = 0  # Unknown size
-
-        response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
-
-        # Content-Length should not be set when size is unknown
-        assert "Content-Length" not in response.headers
+        assert ("Content-Length" in response.headers) is bool(size)
+        if as_attachment or mime_type == "text/html":
+            assert "attachment" in response.headers["Content-Disposition"]
+            assert response.headers["Content-Type"] == "application/octet-stream"
+        else:
+            assert response.mimetype == mime_type
+        if mime_type == "text/html":
+            assert response.headers["X-Content-Type-Options"] == "nosniff"
+        if mime_type.startswith("video/"):
+            assert response.headers["Accept-Ranges"] == "bytes"

    @patch("controllers.service_api.app.file_preview.storage")
-    def test_get_method_integration(
-        self,
-        mock_storage,
-        file_preview_api: FilePreviewApi,
-        mock_app,
-        mock_end_user,
-        mock_upload_file,
-        mock_message_file,
-        mock_message,
+    def test_components_use_validated_file(
+        self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database
    ):
-        """Test the full GET method integration (without decorator)"""
-        file_id = str(uuid.uuid4())
-        app_id = mock_app.id
+        records = _persist_preview_records(database.session)
+        generator = Mock()

-        # Set up mocks
-        mock_upload_file.tenant_id = mock_app.tenant_id
-        mock_message.app_id = app_id
-        mock_message_file.upload_file_id = file_id
-        mock_message_file.message_id = mock_message.id
+        with patch("controllers.service_api.app.file_preview.db", database):
+            message_file, upload_file = file_preview_api._validate_file_ownership(
+                records.upload_file.id, records.app.id
+            )
+            response = file_preview_api._build_file_response(generator, upload_file, False)

-        mock_generator = Mock()
-        mock_storage.load.return_value = mock_generator
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock scalar() for MessageFile and Message queries
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query
-                mock_message,  # Message query
-            ]
-            # Mock get() for UploadFile and App PK lookups
-            mock_db.session.get.side_effect = [
-                mock_upload_file,  # UploadFile query
-                mock_app,  # App query for tenant validation
-            ]
-
-            # Test the core logic directly without Flask decorators
-            # Validate file ownership
-            result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
-            assert result_message_file == mock_message_file
-            assert result_upload_file == mock_upload_file
-
-            # Test file response building
-            response = file_preview_api._build_file_response(mock_generator, mock_upload_file, False)
-            assert response is not None
-
-            # Verify storage was called correctly
-            mock_storage.load.assert_not_called()  # Since we're testing components separately
+        assert message_file.id == records.message_file.id
+        assert response.mimetype == "image/jpeg"
+        mock_storage.load.assert_not_called()

    @patch("controllers.service_api.app.file_preview.storage")
-    def test_storage_error_handling(
-        self,
-        mock_storage,
-        file_preview_api: FilePreviewApi,
-        mock_app,
-        mock_upload_file,
-        mock_message_file,
-        mock_message,
+    def test_storage_error_remains_external(
+        self, mock_storage: Mock, file_preview_api: FilePreviewApi, database: _Database
    ):
-        """Test storage error handling in the core logic"""
-        file_id = str(uuid.uuid4())
-        app_id = mock_app.id
+        records = _persist_preview_records(database.session)
+        mock_storage.load.side_effect = OSError("Storage error")

-        # Set up mocks
-        mock_upload_file.tenant_id = mock_app.tenant_id
-        mock_message.app_id = app_id
-        mock_message_file.upload_file_id = file_id
-        mock_message_file.message_id = mock_message.id
+        with patch("controllers.service_api.app.file_preview.db", database):
+            _, upload_file = file_preview_api._validate_file_ownership(records.upload_file.id, records.app.id)

-        # Mock storage error
-        mock_storage.load.side_effect = Exception("Storage error")
-
-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock scalar() for MessageFile and Message queries
-            mock_db.session.scalar.side_effect = [
-                mock_message_file,  # MessageFile query
-                mock_message,  # Message query
-            ]
-            # Mock get() for UploadFile and App PK lookups
-            mock_db.session.get.side_effect = [
-                mock_upload_file,  # UploadFile query
-                mock_app,  # App query for tenant validation
-            ]
-
-            # First validate file ownership works
-            result_message_file, result_upload_file = file_preview_api._validate_file_ownership(file_id, app_id)
-            assert result_message_file == mock_message_file
-            assert result_upload_file == mock_upload_file
-
-            # Test storage error handling
-            with pytest.raises(Exception) as exc_info:
-                mock_storage.load(mock_upload_file.key, stream=True)
-
-            assert "Storage error" in str(exc_info.value)
+        with pytest.raises(OSError, match="Storage error"):
+            mock_storage.load(upload_file.key, stream=True)

    def test_validate_file_ownership_unexpected_error_logging(
-        self, file_preview_api: FilePreviewApi, caplog: pytest.LogCaptureFixture
+        self,
+        file_preview_api: FilePreviewApi,
+        database: _Database,
+        sqlite_engine: Engine,
+        caplog: pytest.LogCaptureFixture,
    ):
-        """Test that unexpected errors are logged properly"""
-        file_id = str(uuid.uuid4())
-        app_id = str(uuid.uuid4())
+        file_id = str(uuid4())
+        app_id = str(uuid4())

-        with patch("controllers.service_api.app.file_preview.db") as mock_db:
-            # Mock database scalar to raise unexpected exception
-            mock_db.session.scalar.side_effect = Exception("Unexpected database error")
+        def fail_statement(*_args: object) -> None:
+            raise RuntimeError("Unexpected database error")

-            # Execute and assert exception
-            with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"):
-                with pytest.raises(FileAccessDeniedError) as exc_info:
-                    file_preview_api._validate_file_ownership(file_id, app_id)
+        event.listen(sqlite_engine, "before_cursor_execute", fail_statement)
+        try:
+            with patch("controllers.service_api.app.file_preview.db", database):
+                with caplog.at_level(logging.ERROR, logger="controllers.service_api.app.file_preview"):
+                    with pytest.raises(FileAccessDeniedError, match="File access validation failed"):
+                        file_preview_api._validate_file_ownership(file_id, app_id)
+        finally:
+            event.remove(sqlite_engine, "before_cursor_execute", fail_statement)

-            # Verify error message
-            assert "File access validation failed" in str(exc_info.value)
-
-            # Verify logging was called with the structured context fields. The ``extra`` keys
-            # are attached to the LogRecord as attributes, so they are not in ``caplog.text``.
-            assert len(caplog.records) == 1
-            log_record = caplog.records[0]
-            assert log_record.getMessage() == "Unexpected error during file ownership validation"
-            record = cast(_FilePreviewLogRecord, log_record)
-            assert record.file_id == file_id
-            assert record.app_id == app_id
-            assert record.error == "Unexpected database error"
+        assert len(caplog.records) == 1
+        log_record = caplog.records[0]
+        assert log_record.getMessage() == "Unexpected error during file ownership validation"
+        record = cast(_FilePreviewLogRecord, log_record)
+        assert record.file_id == file_id
+        assert record.app_id == app_id
+        assert record.error == "Unexpected database error"
@@ -15,12 +15,15 @@ Focus on:
 """

 import uuid
+from collections.abc import Iterator
 from inspect import unwrap
 from types import SimpleNamespace
 from unittest.mock import Mock, patch

 import pytest
 from flask import Flask
+from sqlalchemy import Engine
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound

 from controllers.service_api.app.error import NotChatAppError
@@ -44,6 +47,14 @@ from services.errors.message import (
 from services.message_service import MessageService


+@pytest.fixture
+def orm_session(sqlite_engine: Engine) -> Iterator[Session]:
+    """Provide a real caller-owned session for MessageService interface tests."""
+
+    with Session(sqlite_engine, expire_on_commit=False) as session:
+        yield session
+
+
 class TestMessageListQuery:
    """Test suite for MessageListQuery Pydantic model."""

@@ -253,7 +264,7 @@ class TestMessageService:
        assert callable(MessageService.get_suggested_questions_after_answer)

    @patch.object(MessageService, "pagination_by_first_id")
-    def test_pagination_by_first_id_returns_pagination_result(self, mock_pagination):
+    def test_pagination_by_first_id_returns_pagination_result(self, mock_pagination, orm_session: Session):
        """Test pagination_by_first_id returns expected format."""
        mock_result = Mock()
        mock_result.data = []
@@ -267,7 +278,7 @@ class TestMessageService:
            conversation_id=str(uuid.uuid4()),
            first_id=None,
            limit=20,
-            session=Mock(),
+            session=orm_session,
        )

        assert hasattr(result, "data")
@@ -275,7 +286,7 @@ class TestMessageService:
        assert hasattr(result, "has_more")

    @patch.object(MessageService, "pagination_by_first_id")
-    def test_pagination_raises_conversation_not_exists_error(self, mock_pagination):
+    def test_pagination_raises_conversation_not_exists_error(self, mock_pagination, orm_session: Session):
        """Test pagination raises ConversationNotExistsError."""
        import services.errors.conversation

@@ -288,11 +299,11 @@ class TestMessageService:
                conversation_id="invalid_id",
                first_id=None,
                limit=20,
-                session=Mock(),
+                session=orm_session,
            )

    @patch.object(MessageService, "pagination_by_first_id")
-    def test_pagination_raises_first_message_not_exists_error(self, mock_pagination):
+    def test_pagination_raises_first_message_not_exists_error(self, mock_pagination, orm_session: Session):
        """Test pagination raises FirstMessageNotExistsError."""
        mock_pagination.side_effect = FirstMessageNotExistsError()

@@ -303,11 +314,11 @@ class TestMessageService:
                conversation_id=str(uuid.uuid4()),
                first_id="invalid_first_id",
                limit=20,
-                session=Mock(),
+                session=orm_session,
            )

    @patch.object(MessageService, "create_feedback")
-    def test_create_feedback_with_rating_and_content(self, mock_create_feedback):
+    def test_create_feedback_with_rating_and_content(self, mock_create_feedback, orm_session: Session):
        """Test create_feedback with rating and content."""
        mock_create_feedback.return_value = None

@@ -317,13 +328,13 @@ class TestMessageService:
            user=Mock(spec=EndUser),
            rating=FeedbackRating.LIKE,
            content="Great response!",
-            session=Mock(),
+            session=orm_session,
        )

        mock_create_feedback.assert_called_once()

    @patch.object(MessageService, "create_feedback")
-    def test_create_feedback_raises_message_not_exists_error(self, mock_create_feedback):
+    def test_create_feedback_raises_message_not_exists_error(self, mock_create_feedback, orm_session: Session):
        """Test create_feedback raises MessageNotExistsError."""
        mock_create_feedback.side_effect = MessageNotExistsError()

@@ -334,11 +345,11 @@ class TestMessageService:
                user=Mock(spec=EndUser),
                rating=FeedbackRating.LIKE,
                content=None,
-                session=Mock(),
+                session=orm_session,
            )

    @patch.object(MessageService, "get_all_messages_feedbacks")
-    def test_get_all_messages_feedbacks_returns_list(self, mock_get_feedbacks):
+    def test_get_all_messages_feedbacks_returns_list(self, mock_get_feedbacks, orm_session: Session):
        """Test get_all_messages_feedbacks returns list of feedbacks."""
        mock_feedbacks = [
            {"message_id": str(uuid.uuid4()), "rating": "like"},
@@ -346,13 +357,15 @@ class TestMessageService:
        ]
        mock_get_feedbacks.return_value = mock_feedbacks

-        result = MessageService.get_all_messages_feedbacks(app_model=Mock(spec=App), page=1, limit=20, session=Mock())
+        result = MessageService.get_all_messages_feedbacks(
+            app_model=Mock(spec=App), page=1, limit=20, session=orm_session
+        )

        assert len(result) == 2
        assert result[0]["rating"] == "like"

    @patch.object(MessageService, "get_suggested_questions_after_answer")
-    def test_get_suggested_questions_returns_questions_list(self, mock_get_questions):
+    def test_get_suggested_questions_returns_questions_list(self, mock_get_questions, orm_session: Session):
        """Test get_suggested_questions_after_answer returns list of questions."""
        mock_questions = ["What about this aspect?", "Can you elaborate on that?", "How does this relate to...?"]
        mock_get_questions.return_value = mock_questions
@@ -362,14 +375,14 @@ class TestMessageService:
            user=Mock(spec=EndUser),
            message_id=str(uuid.uuid4()),
            invoke_from=Mock(),
-            session=Mock(),
+            session=orm_session,
        )

        assert len(result) == 3
        assert isinstance(result[0], str)

    @patch.object(MessageService, "get_suggested_questions_after_answer")
-    def test_get_suggested_questions_raises_disabled_error(self, mock_get_questions):
+    def test_get_suggested_questions_raises_disabled_error(self, mock_get_questions, orm_session: Session):
        """Test get_suggested_questions_after_answer raises SuggestedQuestionsAfterAnswerDisabledError."""
        mock_get_questions.side_effect = SuggestedQuestionsAfterAnswerDisabledError()

@@ -379,11 +392,11 @@ class TestMessageService:
                user=Mock(spec=EndUser),
                message_id=str(uuid.uuid4()),
                invoke_from=Mock(),
-                session=Mock(),
+                session=orm_session,
            )

    @patch.object(MessageService, "get_suggested_questions_after_answer")
-    def test_get_suggested_questions_raises_message_not_exists_error(self, mock_get_questions):
+    def test_get_suggested_questions_raises_message_not_exists_error(self, mock_get_questions, orm_session: Session):
        """Test get_suggested_questions_after_answer raises MessageNotExistsError."""
        mock_get_questions.side_effect = MessageNotExistsError()

@@ -393,7 +406,7 @@ class TestMessageService:
                user=Mock(spec=EndUser),
                message_id="invalid_message_id",
                invoke_from=Mock(),
-                session=Mock(),
+                session=orm_session,
            )


@@ -24,6 +24,7 @@ from unittest.mock import Mock, patch

 import pytest
 from flask import Flask
+from sqlalchemy.orm import Session
 from werkzeug.datastructures import FileStorage
 from werkzeug.exceptions import Forbidden, NotFound

@@ -38,6 +39,7 @@ from controllers.service_api.dataset.rag_pipeline.rag_pipeline_workflow import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom
 from models.account import Account
+from models.dataset import Dataset
 from services.errors.file import FileTooLargeError, UnsupportedFileTypeError
 from services.rag_pipeline.entity.pipeline_service_api_entities import (
    DatasourceNodeRunApiEntity,
@@ -46,6 +48,20 @@ from services.rag_pipeline.entity.pipeline_service_api_entities import (
 from services.rag_pipeline.rag_pipeline import RagPipelineService


+def _persist_dataset(session: Session, *, tenant_id: str, dataset_id: str) -> Dataset:
+    dataset = Dataset(
+        id=dataset_id,
+        tenant_id=tenant_id,
+        name="Pipeline dataset",
+        created_by="account-1",
+        data_source_type=None,
+        indexing_technique=None,
+    )
+    session.add(dataset)
+    session.commit()
+    return dataset
+
+
 class TestDatasourceNodeRunPayload:
    """Test suite for DatasourceNodeRunPayload Pydantic model."""

@@ -550,13 +566,15 @@ class TestPipelineRunApiPost:
    )
    @patch("controllers.service_api.dataset.rag_pipeline.rag_pipeline_workflow.RagPipelineService")
    @patch("controllers.service_api.dataset.rag_pipeline.rag_pipeline_workflow.service_api_ns")
-    def test_post_success_streaming(self, mock_ns, mock_svc_cls, mock_current_user, mock_gen_svc, mock_helper, app):
+    @pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
+    def test_post_success_streaming(
+        self, mock_ns, mock_svc_cls, mock_current_user, mock_gen_svc, mock_helper, app, sqlite_session: Session
+    ):
        """Test successful pipeline run with streaming response."""
        tenant_id = str(uuid.uuid4())
        dataset_id = str(uuid.uuid4())

-        session = Mock()
-        session.scalar.return_value = Mock()
+        _persist_dataset(sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id)

        mock_ns.payload = {
            "inputs": {"key": "val"},
@@ -577,33 +595,33 @@ class TestPipelineRunApiPost:

        with app.test_request_context("/datasets/test/pipeline/run", method="POST"):
            api = PipelineRunApi()
-            response = api.post.__wrapped__(api, session, tenant_id=tenant_id, dataset_id=dataset_id)
+            response = api.post.__wrapped__(api, sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id)

        assert response == {"result": "ok"}
-        mock_svc_cls.assert_called_once_with(session)
+        mock_svc_cls.assert_called_once_with(sqlite_session)
        mock_gen_svc.generate.assert_called_once()

-    def test_post_not_found(self, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
+    def test_post_not_found(self, app: Flask, sqlite_session: Session):
        """Test NotFound when dataset check fails."""
-        session = Mock()
-        session.scalar.return_value = None
-
        with app.test_request_context("/datasets/test/pipeline/run", method="POST"):
            api = PipelineRunApi()
            with pytest.raises(NotFound):
                api.post.__wrapped__(
                    api,
-                    session,
+                    sqlite_session,
                    tenant_id=str(uuid.uuid4()),
                    dataset_id=str(uuid.uuid4()),
                )

    @patch("controllers.service_api.dataset.rag_pipeline.rag_pipeline_workflow.current_user", new="not_account")
    @patch("controllers.service_api.dataset.rag_pipeline.rag_pipeline_workflow.service_api_ns")
-    def test_post_forbidden_non_account_user(self, mock_ns, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
+    def test_post_forbidden_non_account_user(self, mock_ns, app: Flask, sqlite_session: Session):
        """Test Forbidden when current_user is not an Account."""
-        session = Mock()
-        session.scalar.return_value = Mock()
+        tenant_id = str(uuid.uuid4())
+        dataset_id = str(uuid.uuid4())
+        _persist_dataset(sqlite_session, tenant_id=tenant_id, dataset_id=dataset_id)
        mock_ns.payload = {
            "inputs": {},
            "datasource_type": "online_document",
@@ -618,9 +636,9 @@ class TestPipelineRunApiPost:
            with pytest.raises(Forbidden):
                api.post.__wrapped__(
                    api,
-                    session,
-                    tenant_id=str(uuid.uuid4()),
-                    dataset_id=str(uuid.uuid4()),
+                    sqlite_session,
+                    tenant_id=tenant_id,
+                    dataset_id=dataset_id,
                )


@@ -3,10 +3,13 @@ Unit tests for Service API wraps (authentication decorators)
 """

 import uuid
+from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch

 import pytest
 from flask import Flask
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized

 from controllers.service_api.wraps import (
@@ -21,12 +24,11 @@ from controllers.service_api.wraps import (
    validate_dataset_token,
 )
 from enums.cloud_plan import CloudPlan
-from models.account import TenantStatus
-from models.model import ApiToken
-from tests.unit_tests.conftest import (
-    setup_mock_dataset_owner_execute_result,
-    setup_mock_tenant_owner_execute_result,
-)
+from models import Account, Tenant, TenantAccountJoin
+from models.account import TenantAccountRole
+from models.dataset import Dataset, RateLimitLog
+from models.enums import ApiTokenType
+from models.model import ApiToken, App, AppMode, IconType


 def _configure_current_app_mock(mock_current_app):
@@ -34,6 +36,51 @@ def _configure_current_app_mock(mock_current_app):
    mock_current_app._get_current_object = Mock(return_value=Mock())


+def _session_proxy(session: Session) -> MagicMock:
+    """Emulate Flask-SQLAlchemy's callable scoped-session proxy around a test session."""
+    proxy = MagicMock(wraps=session)
+    proxy.return_value = session
+    return proxy
+
+
+def _api_token(*, tenant_id: str, app_id: str | None = None, token_type: ApiTokenType) -> ApiToken:
+    return ApiToken(
+        id=str(uuid.uuid4()),
+        tenant_id=tenant_id,
+        app_id=app_id,
+        type=token_type,
+        token="test_token",
+    )
+
+
+def _persist_workspace(session: Session) -> tuple[Tenant, Account, TenantAccountJoin]:
+    tenant = Tenant(name="Workspace")
+    account = Account(name="Owner", email=f"owner-{uuid.uuid4()}@example.com")
+    membership = TenantAccountJoin(
+        tenant_id=tenant.id,
+        account_id=account.id,
+        current=True,
+        role=TenantAccountRole.OWNER,
+    )
+    session.add_all([tenant, account, membership])
+    session.commit()
+    return tenant, account, membership
+
+
+def _app_model(*, tenant_id: str, enable_api: bool = True) -> App:
+    return App(
+        id=str(uuid.uuid4()),
+        tenant_id=tenant_id,
+        name="Service API App",
+        mode=AppMode.CHAT,
+        icon_type=IconType.EMOJI,
+        icon="chat",
+        icon_background="#FFFFFF",
+        enable_site=False,
+        enable_api=enable_api,
+    )
+
+
 class TestValidateAndGetApiToken:
    """Test suite for validate_and_get_api_token function"""

@@ -70,21 +117,24 @@ class TestValidateAndGetApiToken:
    def test_valid_token_returns_api_token(self, mock_fetch_token, mock_cache_cls, mock_record_usage, app: Flask):
        """Test that valid token returns the ApiToken object."""
        # Arrange
-        mock_api_token = Mock(spec=ApiToken)
-        mock_api_token.token = "valid_token_123"
-        mock_api_token.type = "app"
+        api_token = _api_token(
+            tenant_id=str(uuid.uuid4()),
+            app_id=str(uuid.uuid4()),
+            token_type=ApiTokenType.APP,
+        )
+        api_token.token = "valid_token_123"

        mock_cache_instance = Mock()
        mock_cache_instance.get.return_value = None  # Cache miss
        mock_cache_cls.get = mock_cache_instance.get
-        mock_fetch_token.return_value = mock_api_token
+        mock_fetch_token.return_value = api_token

        # Act
        with app.test_request_context("/", method="GET", headers={"Authorization": "Bearer valid_token_123"}):
            result = validate_and_get_api_token("app")

        # Assert
-        assert result == mock_api_token
+        assert result == api_token

    @patch("controllers.service_api.wraps.record_token_usage")
    @patch("controllers.service_api.wraps.ApiTokenCache")
@@ -117,116 +167,124 @@ class TestValidateAppToken:
        return app

    @patch("controllers.service_api.wraps.user_logged_in")
-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
    @patch("controllers.service_api.wraps.current_app")
+    @pytest.mark.parametrize(
+        "sqlite_session",
+        [(App, ApiToken, Tenant, Account, TenantAccountJoin)],
+        indirect=True,
+    )
    def test_valid_app_token_allows_access(
-        self, mock_current_app, mock_validate_token, mock_db, mock_user_logged_in, app
+        self,
+        mock_current_app,
+        mock_validate_token,
+        mock_user_logged_in,
+        app: Flask,
+        sqlite_session: Session,
    ):
        """Test that valid app token allows access to decorated view."""
        # Arrange
        _configure_current_app_mock(mock_current_app)

-        mock_api_token = Mock()
-        mock_api_token.app_id = str(uuid.uuid4())
-        mock_api_token.tenant_id = str(uuid.uuid4())
-        mock_validate_token.return_value = mock_api_token
-
-        mock_app = Mock()
-        mock_app.id = mock_api_token.app_id
-        mock_app.status = "normal"
-        mock_app.enable_api = True
-        mock_app.tenant_id = mock_api_token.tenant_id
-
-        mock_tenant = Mock()
-        mock_tenant.status = TenantStatus.NORMAL
-        mock_tenant.id = mock_api_token.tenant_id
-
-        mock_account = Mock()
-        mock_account.id = str(uuid.uuid4())
-
-        # Use side_effect to return app first, then tenant via session.get()
-        mock_db.session.get.side_effect = [mock_app, mock_tenant]
-
-        # Mock the tenant owner execute result (execute(select(...)).one_or_none())
-        setup_mock_tenant_owner_execute_result(mock_db, mock_tenant, mock_account)
+        tenant, account, _ = _persist_workspace(sqlite_session)
+        app_model = _app_model(tenant_id=tenant.id)
+        api_token = _api_token(tenant_id=tenant.id, app_id=app_model.id, token_type=ApiTokenType.APP)
+        sqlite_session.add_all([app_model, api_token])
+        sqlite_session.commit()
+        mock_validate_token.return_value = api_token

        @validate_app_token
        def protected_view(app_model):
            return {"success": True, "app_id": app_model.id}

        # Act
-        with app.test_request_context("/", method="GET", headers={"Authorization": "Bearer test_token"}):
+        with (
+            app.test_request_context("/", method="GET", headers={"Authorization": "Bearer test_token"}),
+            patch("controllers.service_api.wraps.db.session", _session_proxy(sqlite_session)),
+        ):
            result = protected_view()

        # Assert
        assert result["success"] is True
-        assert result["app_id"] == mock_app.id
+        assert result["app_id"] == app_model.id
+        assert account.current_tenant_id == tenant.id

-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
-    def test_app_not_found_raises_forbidden(self, mock_validate_token, mock_db, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
+    def test_app_not_found_raises_forbidden(self, mock_validate_token, app: Flask, sqlite_session: Session):
        """Test that Forbidden is raised when app no longer exists."""
        # Arrange
-        mock_api_token = Mock()
-        mock_api_token.app_id = str(uuid.uuid4())
-        mock_validate_token.return_value = mock_api_token
-
-        mock_db.session.get.return_value = None
+        api_token = _api_token(
+            tenant_id=str(uuid.uuid4()),
+            app_id=str(uuid.uuid4()),
+            token_type=ApiTokenType.APP,
+        )
+        mock_validate_token.return_value = api_token

        @validate_app_token
        def protected_view(**kwargs):
            return {"success": True}

        # Act & Assert
-        with app.test_request_context("/", method="GET"):
+        with (
+            app.test_request_context("/", method="GET"),
+            patch("controllers.service_api.wraps.db.session", sqlite_session),
+        ):
            with pytest.raises(Forbidden) as exc_info:
                protected_view()
            assert "no longer exists" in str(exc_info.value)

-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
-    def test_app_status_abnormal_raises_forbidden(self, mock_validate_token, mock_db, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
+    def test_app_status_abnormal_raises_forbidden(self, mock_validate_token, app: Flask, sqlite_session: Session):
        """Test that Forbidden is raised when app status is abnormal."""
        # Arrange
-        mock_api_token = Mock()
-        mock_api_token.app_id = str(uuid.uuid4())
-        mock_validate_token.return_value = mock_api_token
-
-        mock_app = Mock()
-        mock_app.status = "abnormal"
-        mock_db.session.get.return_value = mock_app
+        app_model = _app_model(tenant_id=str(uuid.uuid4()))
+        sqlite_session.add(app_model)
+        sqlite_session.commit()
+        app_model.status = "abnormal"
+        mock_validate_token.return_value = _api_token(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            token_type=ApiTokenType.APP,
+        )

        @validate_app_token
        def protected_view(**kwargs):
            return {"success": True}

        # Act & Assert
-        with app.test_request_context("/", method="GET"):
+        with (
+            app.test_request_context("/", method="GET"),
+            patch("controllers.service_api.wraps.db.session", sqlite_session),
+        ):
            with pytest.raises(Forbidden) as exc_info:
                protected_view()
            assert "status is abnormal" in str(exc_info.value)

-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
-    def test_app_api_disabled_raises_forbidden(self, mock_validate_token, mock_db, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
+    def test_app_api_disabled_raises_forbidden(self, mock_validate_token, app: Flask, sqlite_session: Session):
        """Test that Forbidden is raised when app API is disabled."""
        # Arrange
-        mock_api_token = Mock()
-        mock_api_token.app_id = str(uuid.uuid4())
-        mock_validate_token.return_value = mock_api_token
-
-        mock_app = Mock()
-        mock_app.status = "normal"
-        mock_app.enable_api = False
-        mock_db.session.get.return_value = mock_app
+        app_model = _app_model(tenant_id=str(uuid.uuid4()), enable_api=False)
+        sqlite_session.add(app_model)
+        sqlite_session.commit()
+        mock_validate_token.return_value = _api_token(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            token_type=ApiTokenType.APP,
+        )

        @validate_app_token
        def protected_view(**kwargs):
            return {"success": True}

        # Act & Assert
-        with app.test_request_context("/", method="GET"):
+        with (
+            app.test_request_context("/", method="GET"),
+            patch("controllers.service_api.wraps.db.session", sqlite_session),
+        ):
            with pytest.raises(Forbidden) as exc_info:
                protected_view()
            assert "API service has been disabled" in str(exc_info.value)
@@ -468,26 +526,35 @@ class TestCloudEditionBillingRateLimitCheck:

    @patch("controllers.service_api.wraps.validate_and_get_api_token")
    @patch("controllers.service_api.wraps.FeatureService.get_knowledge_rate_limit")
-    @patch("controllers.service_api.wraps.db")
-    @patch("controllers.service_api.wraps.sessionmaker")
+    @pytest.mark.parametrize("sqlite_session", [(RateLimitLog,)], indirect=True)
    def test_rejects_over_rate_limit(
-        self, mock_sessionmaker, mock_db, mock_get_rate_limit, mock_validate_token, app: Flask
+        self,
+        mock_get_rate_limit,
+        mock_validate_token,
+        app: Flask,
+        sqlite_session: Session,
    ):
        """Test that Forbidden is raised when over rate limit."""
        # Arrange
-        mock_validate_token.return_value = Mock(tenant_id="tenant123")
+        tenant_id = str(uuid.uuid4())
+        mock_validate_token.return_value = _api_token(
+            tenant_id=tenant_id,
+            token_type=ApiTokenType.DATASET,
+        )

        mock_rate_limit = Mock()
        mock_rate_limit.enabled = True
        mock_rate_limit.limit = 10
        mock_rate_limit.subscription_plan = "pro"
        mock_get_rate_limit.return_value = mock_rate_limit
-        rate_limit_log_session = MagicMock()
-        session_factory = MagicMock()
-        session_factory.begin.return_value.__enter__.return_value = rate_limit_log_session
-        mock_sessionmaker.return_value = session_factory

-        with patch("controllers.service_api.wraps.redis_client") as mock_redis:
+        with (
+            patch("controllers.service_api.wraps.redis_client") as mock_redis,
+            patch(
+                "controllers.service_api.wraps.db",
+                SimpleNamespace(engine=sqlite_session.get_bind()),
+            ),
+        ):
            mock_redis.zcard.return_value = 15  # Over limit

            @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
@@ -499,9 +566,12 @@ class TestCloudEditionBillingRateLimitCheck:
                with pytest.raises(Forbidden) as exc_info:
                    knowledge_request()
                assert "rate limit" in str(exc_info.value)
-            mock_sessionmaker.assert_called_once_with(bind=mock_db.engine, expire_on_commit=False)
-            rate_limit_log_session.add.assert_called_once()
-            mock_db.session.commit.assert_not_called()
+
+        persisted_logs = sqlite_session.scalars(select(RateLimitLog)).all()
+        assert len(persisted_logs) == 1
+        assert persisted_logs[0].tenant_id == tenant_id
+        assert persisted_logs[0].subscription_plan == "pro"
+        assert persisted_logs[0].operation == "knowledge"


 class TestValidateDatasetToken:
@@ -515,65 +585,62 @@ class TestValidateDatasetToken:
        return app

    @patch("controllers.service_api.wraps.user_logged_in")
-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
    @patch("controllers.service_api.wraps.current_app")
-    def test_valid_dataset_token(self, mock_current_app, mock_validate_token, mock_db, mock_user_logged_in, app: Flask):
+    @pytest.mark.parametrize(
+        "sqlite_session",
+        [(Tenant, Account, TenantAccountJoin)],
+        indirect=True,
+    )
+    def test_valid_dataset_token(
+        self,
+        mock_current_app,
+        mock_validate_token,
+        mock_user_logged_in,
+        app: Flask,
+        sqlite_session: Session,
+    ):
        """Test that valid dataset token allows access."""
        # Arrange
        _configure_current_app_mock(mock_current_app)

-        tenant_id = str(uuid.uuid4())
-        mock_api_token = Mock()
-        mock_api_token.tenant_id = tenant_id
-        mock_validate_token.return_value = mock_api_token
-
-        mock_tenant = Mock()
-        mock_tenant.id = tenant_id
-        mock_tenant.status = TenantStatus.NORMAL
-
-        mock_ta = Mock()
-        mock_ta.account_id = str(uuid.uuid4())
-
-        mock_account = Mock()
-        mock_account.id = mock_ta.account_id
-        mock_account.current_tenant = mock_tenant
-
-        # Mock the tenant account join query (execute(select(...)).one_or_none())
-        setup_mock_dataset_owner_execute_result(mock_db, mock_tenant, mock_ta)
-
-        # Mock the account lookup via session.get()
-        mock_db.session.get.return_value = mock_account
+        tenant, account, _ = _persist_workspace(sqlite_session)
+        api_token = _api_token(tenant_id=tenant.id, token_type=ApiTokenType.DATASET)
+        mock_validate_token.return_value = api_token

        @validate_dataset_token
        def protected_view(tenant_id):
            return {"success": True, "tenant_id": tenant_id}

        # Act
-        with app.test_request_context("/", method="GET", headers={"Authorization": "Bearer test_token"}):
+        with (
+            app.test_request_context("/", method="GET", headers={"Authorization": "Bearer test_token"}),
+            patch("controllers.service_api.wraps.db.session", _session_proxy(sqlite_session)),
+        ):
            result = protected_view()

        # Assert
        assert result["success"] is True
-        assert result["tenant_id"] == tenant_id
+        assert result["tenant_id"] == tenant.id
+        assert account.current_tenant_id == tenant.id

-    @patch("controllers.service_api.wraps.db")
    @patch("controllers.service_api.wraps.validate_and_get_api_token")
-    def test_dataset_not_found_raises_not_found(self, mock_validate_token, mock_db, app: Flask):
+    @pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
+    def test_dataset_not_found_raises_not_found(self, mock_validate_token, app: Flask, sqlite_session: Session):
        """Test that NotFound is raised when dataset doesn't exist."""
        # Arrange
-        mock_api_token = Mock()
-        mock_api_token.tenant_id = str(uuid.uuid4())
-        mock_validate_token.return_value = mock_api_token
-
-        mock_db.session.scalar.return_value = None
+        api_token = _api_token(tenant_id=str(uuid.uuid4()), token_type=ApiTokenType.DATASET)
+        mock_validate_token.return_value = api_token

        @validate_dataset_token
        def protected_view(dataset_id=None, **kwargs):
            return {"success": True}

        # Act & Assert
-        with app.test_request_context("/", method="GET"):
+        with (
+            app.test_request_context("/", method="GET"),
+            patch("controllers.service_api.wraps.db.session", sqlite_session),
+        ):
            with pytest.raises(NotFound) as exc_info:
                protected_view(dataset_id=str(uuid.uuid4()))
            assert "Dataset not found" in str(exc_info.value)
@@ -163,6 +163,7 @@ def test_get_form_includes_site(monkeypatch: pytest.MonkeyPatch, app: Flask, dat
    assert body["expiration_time"] == int(expiration_time.timestamp())
    assert body["site"] == {
        "app_id": app_model.id,
+        "mode": "chat",
        "end_user_id": None,
        "enable_site": True,
        "site": {
@@ -383,6 +384,7 @@ def test_get_form_allows_backstage_token(monkeypatch: pytest.MonkeyPatch, app: F
    assert body["expiration_time"] == int(expiration_time.timestamp())
    assert body["site"] == {
        "app_id": app_model.id,
+        "mode": "chat",
        "end_user_id": None,
        "enable_site": True,
        "site": {
@@ -3,7 +3,42 @@ from unittest.mock import MagicMock, patch
 from configs import dify_config
 from controllers.web import site as site_module
 from extensions.storage.storage_type import StorageType
-from models.model import IconType, Site
+from models.model import AppMode, IconType, Site
+from services.feature_service import FeatureModel
+
+
+def test_app_site_api_returns_legacy_agent_compatible_mode() -> None:
+    app_model = MagicMock()
+    app_model.id = "app-id"
+    app_model.tenant_id = "tenant-id"
+    app_model.tenant = MagicMock(id="tenant-id", status="normal")
+    app_model.mode_compatible_with_agent_with_session.return_value = AppMode.AGENT_CHAT
+    end_user = MagicMock(id="end-user-id")
+    site = MagicMock(spec=Site)
+    response = MagicMock()
+    response.model_dump.return_value = {"mode": AppMode.AGENT_CHAT}
+
+    with (
+        patch.object(site_module, "db") as mock_db,
+        patch.object(site_module.FeatureService, "get_features", return_value=FeatureModel(can_replace_logo=False)),
+        patch.object(site_module, "_build_site_icon_url", return_value=None),
+        patch.object(site_module.WebAppSiteResponse, "from_app_site", return_value=response) as mock_from_app_site,
+    ):
+        mock_db.session.scalar.return_value = site
+        result = site_module.AppSiteApi().get(app_model, end_user)
+
+    assert result["mode"] == AppMode.AGENT_CHAT
+    app_model.mode_compatible_with_agent_with_session.assert_called_once_with(session=mock_db.session())
+    mock_from_app_site.assert_called_once_with(
+        tenant=app_model.tenant,
+        app_model=app_model,
+        mode=AppMode.AGENT_CHAT,
+        site=site,
+        end_user_id=end_user.id,
+        features=FeatureModel(can_replace_logo=False),
+        can_replace_logo=False,
+        icon_url=None,
+    )


 def test_build_site_icon_url_uses_s3_presigned_url() -> None:
@@ -442,6 +442,13 @@ class TestAdvancedChatAppGeneratorInternals:
            def start(self):
                thread_data["started"] = True

+            def join(self, timeout):
+                thread_data["joined"] = True
+                thread_data["join_timeout"] = timeout
+
+            def is_alive(self):
+                return False
+
        monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.threading.Thread", _Thread)
        monkeypatch.setattr(
            "core.app.apps.advanced_chat.app_generator.db", SimpleNamespace(engine=object(), session=db_session)
@@ -475,6 +482,8 @@ class TestAdvancedChatAppGeneratorInternals:

        assert response["response"] == {"raw": True}
        assert thread_data["started"] is True
+        assert thread_data["joined"] is True
+        assert thread_data["join_timeout"] == 300
        assert "pause-layer" in thread_data["kwargs"]["graph_engine_layers"]
        assert generator._dialogue_count == 3
        assert init_records.call_args.kwargs["session"] is db_session
@@ -542,6 +551,13 @@ class TestAdvancedChatAppGeneratorInternals:
            def start(self):
                thread_data["started"] = True

+            def join(self, timeout):
+                thread_data["joined"] = True
+                thread_data["join_timeout"] = timeout
+
+            def is_alive(self):
+                return False
+
        monkeypatch.setattr("core.app.apps.advanced_chat.app_generator.threading.Thread", _Thread)
        monkeypatch.setattr(
            "core.app.apps.advanced_chat.app_generator.db", SimpleNamespace(engine=object(), session=db_session)
@@ -574,6 +590,8 @@ class TestAdvancedChatAppGeneratorInternals:
        init_records.assert_not_called()
        get_thread_messages_length.assert_called_once_with(conversation.id, session=db_session)
        assert thread_data["started"] is True
+        assert thread_data["joined"] is True
+        assert thread_data["join_timeout"] == 300
        db_session.commit.assert_not_called()
        db_session.refresh.assert_not_called()
        db_session.close.assert_called_once()
@@ -1,21 +1,11 @@
-from contextlib import contextmanager
 from types import SimpleNamespace
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import MagicMock

-import pytest
-from pytest_mock import MockerFixture
-
-import core.app.apps.completion.app_runner as module
-from core.app.apps.completion.app_runner import CompletionAppRunner
+from core.app.apps.completion.workflow_runner import CompletionWorkflowRunner
 from core.moderation.base import ModerationError
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent


-@pytest.fixture
-def runner():
-    return CompletionAppRunner()
-
-
 def _build_app_config(dataset=None, external_tools=None, additional_features=None):
    app_config = MagicMock()
    app_config.app_id = "app1"
@@ -24,7 +14,7 @@ def _build_app_config(dataset=None, external_tools=None, additional_features=Non
    app_config.dataset = dataset
    app_config.external_data_variables = external_tools or []
    app_config.additional_features = additional_features
-    app_config.app_model_config_dict = {"file_upload": {"enabled": True}}
+    app_config.app_model_config_dict = {"file_upload": {"image": {"enabled": True}}}
    return app_config


@@ -38,163 +28,40 @@ def _build_generate_entity(app_config, file_upload_config=None):
    return SimpleNamespace(
        app_config=app_config,
        model_conf=model_conf,
-        inputs={"qvar": "query_from_input"},
+        inputs={"qvar": "original_query_from_input"},
        query="original_query",
        files=[],
        file_upload_config=file_upload_config,
        stream=True,
        user_id="user",
        invoke_from=MagicMock(),
+        trace_manager=None,
    )


-@contextmanager
-def patched_create_session(*, return_value=None):
-    session = MagicMock()
-    session.scalar.return_value = return_value
-    session_context = MagicMock()
-    session_context.__enter__.return_value = session
-    with patch.object(module, "create_session", return_value=session_context):
-        yield session
+def test_workflow_runner_direct_outputs_on_input_moderation() -> None:
+    runner = CompletionWorkflowRunner()
+    app_record = MagicMock(id="app1", tenant_id="tenant")
+    app_generate_entity = _build_generate_entity(_build_app_config())
+    queue_manager = MagicMock()
+    message = MagicMock(id="msg")
+    runner.organize_prompt_messages = MagicMock(return_value=([], None))
+    runner.moderation_for_inputs = MagicMock(side_effect=ModerationError("blocked"))
+    runner.direct_output = MagicMock()
+
+    result = runner._run_input_moderation(
+        app_record=app_record,
+        application_generate_entity=app_generate_entity,
+        queue_manager=queue_manager,
+        message=message,
+    )
+
+    assert result.stopped is True
+    runner.direct_output.assert_called_once()


-class TestCompletionAppRunner:
-    def test_run_app_not_found(self, runner, mocker: MockerFixture):
-        app_config = _build_app_config()
-        app_generate_entity = _build_generate_entity(app_config)
+def test_workflow_runner_uses_low_image_detail_default() -> None:
+    runner = CompletionWorkflowRunner()
+    app_generate_entity = _build_generate_entity(_build_app_config(), file_upload_config=None)

-        with patched_create_session(return_value=None):
-            with pytest.raises(ValueError):
-                runner.run(app_generate_entity, MagicMock(), MagicMock(), MagicMock())
-
-    def test_run_moderation_error_outputs_direct(self, runner, mocker: MockerFixture):
-        app_record = MagicMock(id="app1", tenant_id="tenant")
-
-        app_config = _build_app_config()
-        app_generate_entity = _build_generate_entity(app_config)
-
-        runner.organize_prompt_messages = MagicMock(return_value=([], None))
-        runner.moderation_for_inputs = MagicMock(side_effect=ModerationError("blocked"))
-        runner.direct_output = MagicMock()
-        runner._handle_invoke_result = MagicMock()
-
-        with patched_create_session(return_value=app_record):
-            runner.run(app_generate_entity, MagicMock(), MagicMock(id="msg"), MagicMock())
-
-        runner.direct_output.assert_called_once()
-        runner._handle_invoke_result.assert_not_called()
-
-    def test_run_hosting_moderation_stops(self, runner, mocker: MockerFixture):
-        app_record = MagicMock(id="app1", tenant_id="tenant")
-
-        app_config = _build_app_config()
-        app_generate_entity = _build_generate_entity(app_config)
-
-        runner.organize_prompt_messages = MagicMock(return_value=([], None))
-        runner.moderation_for_inputs = MagicMock(return_value=(None, app_generate_entity.inputs, "query"))
-        runner.check_hosting_moderation = MagicMock(return_value=True)
-        runner._handle_invoke_result = MagicMock()
-
-        with patched_create_session(return_value=app_record):
-            runner.run(app_generate_entity, MagicMock(), MagicMock(id="msg"), MagicMock())
-
-        runner._handle_invoke_result.assert_not_called()
-
-    def test_run_dataset_and_external_tools_flow(self, runner, mocker: MockerFixture):
-        app_record = MagicMock(id="app1", tenant_id="tenant")
-
-        retrieve_config = MagicMock(query_variable="qvar")
-        dataset_config = MagicMock(dataset_ids=["ds"], retrieve_config=retrieve_config)
-        additional_features = MagicMock(show_retrieve_source=True)
-        app_config = _build_app_config(
-            dataset=dataset_config,
-            external_tools=["tool"],
-            additional_features=additional_features,
-        )
-
-        file_upload_config = MagicMock()
-        file_upload_config.image_config.detail = ImagePromptMessageContent.DETAIL.HIGH
-
-        app_generate_entity = _build_generate_entity(app_config, file_upload_config=file_upload_config)
-
-        runner.organize_prompt_messages = MagicMock(side_effect=[(["pm1"], ["stop"]), (["pm2"], ["stop"])])
-        runner.moderation_for_inputs = MagicMock(return_value=(None, app_generate_entity.inputs, "query"))
-        runner.fill_in_inputs_from_external_data_tools = MagicMock(return_value=app_generate_entity.inputs)
-        runner.check_hosting_moderation = MagicMock(return_value=False)
-        runner.recalc_llm_max_tokens = MagicMock()
-        runner._handle_invoke_result = MagicMock()
-
-        dataset_retrieval = MagicMock()
-        dataset_retrieval.retrieve.return_value = ("ctx", ["file1"])
-        mocker.patch.object(module, "DatasetRetrieval", return_value=dataset_retrieval)
-
-        model_instance = MagicMock()
-        model_instance.invoke_llm.return_value = "invoke_result"
-        mocker.patch.object(module, "ModelInstance", return_value=model_instance)
-
-        with patched_create_session(return_value=app_record):
-            runner.run(app_generate_entity, MagicMock(), MagicMock(id="msg", tenant_id="tenant"), MagicMock())
-
-        dataset_retrieval.retrieve.assert_called_once()
-        assert dataset_retrieval.retrieve.call_args.kwargs["query"] == "query_from_input"
-        runner._handle_invoke_result.assert_called_once()
-
-    def test_run_closes_explicit_session_before_stream_consumption(self, runner, mocker: MockerFixture):
-        app_record = MagicMock(id="app1", tenant_id="tenant")
-        app_config = _build_app_config()
-        app_generate_entity = _build_generate_entity(app_config)
-        queue_manager = MagicMock()
-
-        events = []
-        session = MagicMock()
-        session.commit.side_effect = lambda: events.append("commit")
-        session.close.side_effect = lambda: events.append("close")
-        runner.organize_prompt_messages = MagicMock(return_value=([], None))
-        runner.moderation_for_inputs = MagicMock(return_value=(None, app_generate_entity.inputs, "query"))
-        runner.check_hosting_moderation = MagicMock(return_value=False)
-        runner.recalc_llm_max_tokens = MagicMock()
-        runner._handle_invoke_result = MagicMock(side_effect=lambda invoke_result, **kwargs: list(invoke_result))
-
-        model_instance = MagicMock()
-
-        def invoke_stream():
-            events.append("first-chunk")
-            yield "chunk"
-
-        def invoke_llm(**kwargs):
-            events.append("invoke")
-            return invoke_stream()
-
-        model_instance.invoke_llm.side_effect = invoke_llm
-        mocker.patch.object(module, "ModelInstance", return_value=model_instance)
-
-        with patched_create_session(return_value=app_record):
-            runner.run(app_generate_entity, queue_manager, MagicMock(id="msg"), session)
-
-        assert events == ["commit", "close", "invoke", "first-chunk"]
-        runner._handle_invoke_result.assert_called_once_with(
-            invoke_result=ANY,
-            queue_manager=queue_manager,
-            stream=True,
-            message_id="msg",
-            user_id="user",
-            tenant_id="tenant",
-        )
-
-    def test_run_uses_low_image_detail_default(self, runner, mocker: MockerFixture):
-        app_record = MagicMock(id="app1", tenant_id="tenant")
-
-        app_config = _build_app_config()
-        app_generate_entity = _build_generate_entity(app_config, file_upload_config=None)
-
-        runner.organize_prompt_messages = MagicMock(return_value=([], None))
-        runner.moderation_for_inputs = MagicMock(return_value=(None, app_generate_entity.inputs, "query"))
-        runner.check_hosting_moderation = MagicMock(return_value=True)
-
-        with patched_create_session(return_value=app_record):
-            runner.run(app_generate_entity, MagicMock(), MagicMock(id="msg"), MagicMock())
-
-        assert (
-            runner.organize_prompt_messages.call_args.kwargs["image_detail_config"]
-            == ImagePromptMessageContent.DETAIL.LOW
-        )
+    assert runner._resolve_image_detail_config(app_generate_entity) == ImagePromptMessageContent.DETAIL.LOW
@@ -376,18 +376,26 @@ class TestCompletionAppGenerator:
        create_session.return_value = session_context
        mocker.patch.object(module.db, "session")

-        mocker.patch.object(generator, "_get_message", return_value=MagicMock())
+        message = MagicMock()
+        mocker.patch.object(generator, "_get_message", return_value=message)

        runner_instance = MagicMock()
        runner_instance.run.side_effect = error
-        mocker.patch.object(module, "CompletionAppRunner", return_value=runner_instance)
+        mocker.patch.object(module, "CompletionWorkflowRunner", return_value=runner_instance)

        queue_manager = MagicMock()
+        application_generate_entity = MagicMock()
        generator._generate_worker(
            flask_app=flask_app,
-            application_generate_entity=MagicMock(),
+            application_generate_entity=application_generate_entity,
            queue_manager=queue_manager,
            message_id="msg",
        )

+        runner_instance.run.assert_called_once_with(
+            application_generate_entity=application_generate_entity,
+            queue_manager=queue_manager,
+            message=message,
+            session=session,
+        )
        assert queue_manager.publish_error.called is should_publish
@@ -0,0 +1,70 @@
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from core.app.apps.completion.runtime_workflow_builder import build_runtime_completion_workflow
+from graphon.nodes import BuiltinNodeTypes
+from models.model import AppMode
+
+
+def test_builder_returns_runtime_graph_without_workflow_record() -> None:
+    app_model = SimpleNamespace(mode=AppMode.COMPLETION)
+    app_config = MagicMock()
+    workflow_converter = MagicMock(
+        build_graph_from_app_config=MagicMock(return_value=({"nodes": [{"id": "start"}], "edges": []}, {}))
+    )
+    session = MagicMock()
+
+    result = build_runtime_completion_workflow(
+        app_model=app_model,
+        app_config=app_config,
+        session=session,
+        workflow_converter=workflow_converter,
+    )
+
+    assert result.workflow_id.startswith("completion-runtime-")
+    assert result.root_node_id == "start"
+    assert result.graph_dict == {"nodes": [{"id": "start"}], "edges": []}
+    workflow_converter.build_graph_from_app_config.assert_called_once_with(
+        app_model=app_model,
+        app_config=app_config,
+        target_app_mode=AppMode.WORKFLOW,
+        session=session,
+    )
+
+
+def test_builder_routes_api_based_variable_query_to_runtime_sys_query() -> None:
+    app_model = SimpleNamespace(mode=AppMode.COMPLETION)
+    app_config = MagicMock()
+    request_body = {"params": {"query": ""}}
+    workflow_converter = MagicMock(
+        build_graph_from_app_config=MagicMock(
+            return_value=(
+                {
+                    "nodes": [
+                        {
+                            "id": "http_request_1",
+                            "data": {
+                                "type": BuiltinNodeTypes.HTTP_REQUEST,
+                                "body": {"type": "json", "data": json.dumps(request_body)},
+                            },
+                        }
+                    ],
+                    "edges": [],
+                },
+                {},
+            )
+        )
+    )
+    session = MagicMock()
+
+    result = build_runtime_completion_workflow(
+        app_model=app_model,
+        app_config=app_config,
+        session=session,
+        workflow_converter=workflow_converter,
+    )
+
+    http_node = result.graph_dict["nodes"][0]
+    body = json.loads(http_node["data"]["body"]["data"])
+    assert body["params"]["query"] == "{{#sys.query#}}"
@@ -0,0 +1,172 @@
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from core.app.apps.base_app_queue_manager import PublishFrom
+from core.app.apps.completion.graph_event_adapter import CompletionGraphEventAdapter
+from core.app.entities.queue_entities import (
+    QueueErrorEvent,
+    QueueLLMChunkEvent,
+    QueueMessageEndEvent,
+    QueueRetrieverResourcesEvent,
+    QueueStopEvent,
+)
+from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
+from graphon.graph_events import (
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunSucceededEvent,
+    NodeRunRetrieverResourceEvent,
+    NodeRunStreamChunkEvent,
+    NodeRunSucceededEvent,
+)
+from graphon.model_runtime.entities.llm_entities import LLMUsage
+from graphon.model_runtime.entities.message_entities import UserPromptMessage
+from graphon.node_events import NodeRunResult
+
+
+def _adapter(*, show_retrieve_source: bool = True) -> tuple[CompletionGraphEventAdapter, MagicMock]:
+    queue_manager = MagicMock()
+    entity = SimpleNamespace(
+        model_conf=SimpleNamespace(model="model"),
+        app_config=SimpleNamespace(
+            additional_features=SimpleNamespace(show_retrieve_source=show_retrieve_source),
+        ),
+    )
+    return (
+        CompletionGraphEventAdapter(application_generate_entity=entity, queue_manager=queue_manager),
+        queue_manager,
+    )
+
+
+def test_stream_chunk_event_publishes_llm_chunk() -> None:
+    adapter, queue_manager = _adapter()
+    prompt_message = UserPromptMessage(content="prompt")
+    adapter.set_prompt_messages([prompt_message])
+
+    adapter.handle_event(
+        NodeRunStreamChunkEvent(
+            id="run",
+            node_id="llm",
+            node_type=BuiltinNodeTypes.LLM,
+            selector=["llm", "text"],
+            chunk="hello",
+        )
+    )
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueLLMChunkEvent)
+    assert event.chunk.delta.message.content == "hello"
+    assert event.chunk.prompt_messages == [prompt_message]
+    assert queue_manager.publish.call_args.args[1] == PublishFrom.APPLICATION_MANAGER
+
+
+def test_stream_chunk_event_skips_final_empty_chunk() -> None:
+    adapter, queue_manager = _adapter()
+
+    adapter.handle_event(
+        NodeRunStreamChunkEvent(
+            id="run",
+            node_id="llm",
+            node_type=BuiltinNodeTypes.LLM,
+            selector=["llm", "text"],
+            chunk="",
+            is_final=True,
+        )
+    )
+
+    queue_manager.publish.assert_not_called()
+
+
+def test_retriever_resource_event_publishes_legacy_retriever_resources() -> None:
+    adapter, queue_manager = _adapter()
+
+    adapter.handle_event(
+        NodeRunRetrieverResourceEvent(
+            id="run",
+            node_id="knowledge_retrieval",
+            node_type=BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL,
+            retriever_resources=[{"dataset_id": "dataset", "content": "hit"}],
+            context="hit",
+        )
+    )
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueRetrieverResourcesEvent)
+    assert event.retriever_resources[0].dataset_id == "dataset"
+
+
+def test_retriever_resource_event_is_hidden_when_feature_is_disabled() -> None:
+    adapter, queue_manager = _adapter(show_retrieve_source=False)
+
+    adapter.handle_event(
+        NodeRunRetrieverResourceEvent(
+            id="run",
+            node_id="knowledge_retrieval",
+            node_type=BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL,
+            retriever_resources=[{"dataset_id": "dataset", "content": "hit"}],
+            context="hit",
+        )
+    )
+
+    queue_manager.publish.assert_not_called()
+
+
+def test_llm_success_then_graph_success_publishes_message_end() -> None:
+    adapter, queue_manager = _adapter()
+    usage = LLMUsage.empty_usage()
+    prompt_message = UserPromptMessage(content="prompt")
+    adapter.set_prompt_messages([prompt_message])
+    adapter.handle_event(
+        NodeRunSucceededEvent(
+            id="run",
+            node_id="llm",
+            node_type=BuiltinNodeTypes.LLM,
+            start_at=datetime.now(UTC),
+            node_run_result=NodeRunResult(
+                status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                outputs={"text": "final"},
+                llm_usage=usage,
+            ),
+        )
+    )
+
+    adapter.handle_event(GraphRunSucceededEvent(outputs={"result": "final"}))
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueMessageEndEvent)
+    assert event.llm_result is not None
+    assert event.llm_result.message.content == "final"
+    assert event.llm_result.prompt_messages == [prompt_message]
+    assert event.llm_result.usage is usage
+
+
+def test_graph_success_uses_outputs_result_when_llm_success_was_not_seen() -> None:
+    adapter, queue_manager = _adapter()
+
+    adapter.handle_event(GraphRunSucceededEvent(outputs={"result": "final from graph"}))
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueMessageEndEvent)
+    assert event.llm_result is not None
+    assert event.llm_result.message.content == "final from graph"
+
+
+def test_failed_graph_publishes_error() -> None:
+    adapter, queue_manager = _adapter()
+
+    adapter.handle_event(GraphRunFailedEvent(error="boom"))
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueErrorEvent)
+    assert str(event.error) == "boom"
+
+
+def test_user_abort_publishes_legacy_stop() -> None:
+    adapter, queue_manager = _adapter()
+
+    adapter.handle_event(GraphRunAbortedEvent(reason="Stopped by user."))
+
+    event = queue_manager.publish.call_args.args[0]
+    assert isinstance(event, QueueStopEvent)
+    assert event.stopped_by == QueueStopEvent.StopBy.USER_MANUAL
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
-LAN-	62518a5e55	fix(completion): preserve legacy workflow compatibility	2026-07-28 13:04:11 +08:00
-LAN-	e2c769c6cc	fix(workflow): keep converter imports sorted	2026-07-28 13:03:12 +08:00
-LAN-	6ad1a106c0	fix(workflow): preserve polling llm invoke hook	2026-07-28 13:03:12 +08:00
-LAN-	659c486e12	test(completion): cover workflow runner compatibility branches	2026-07-28 13:03:12 +08:00
-LAN-	eaa55a4292	fix(completion): keep tts text required in schema	2026-07-28 13:03:11 +08:00
-LAN-	4122233ba0	fix(task-pipeline): tolerate message end events without saved prompt	2026-07-28 13:03:11 +08:00
autofix-ci[bot]and-LAN-	dfc2726e44	[autofix.ci] apply automated fixes	2026-07-28 13:03:11 +08:00
-LAN-	de8ae2a6ab	refactor(completion): run through workflow entry Replace the dedicated Completion runner with a Workflow Entry backed execution path. Adapt GraphOn events back into legacy Completion queue events and keep existing message persistence in the EasyUI task pipeline. Refs #37572	2026-07-28 13:03:11 +08:00
-LAN-	eb82fa8e04	docs(completion): add workflow entry implementation plan	2026-07-28 13:03:10 +08:00
-LAN-	8ca84a7b45	docs(completion): add workflow entry reuse design	2026-07-28 13:03:10 +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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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 <[email protected]>	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] <[email protected]> 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 <[email protected]>	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] <[email protected]> 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] <[email protected]> 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] <[email protected]> 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 <[email protected]>	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 <[email protected]>	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) <[email protected]> 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] <[email protected]> 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] <[email protected]> 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 <[email protected]> Co-authored-by: Cursor <[email protected]>	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) <[email protected]>	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