fix: fix mcp output_schema is optional (#39453 )

Co-authored-by: yunlu.wen <yunlu.wen@dify.ai>
chore: bump version to 1.16.1 (#39653 )
2026-07-28 02:31:18 +00:00 · 2026-07-28 01:09:25 +00:00 · 2026-07-27 23:53:59 +00:00 · 2026-07-27 23:50:05 +00:00 · 2026-07-27 23:43:03 +00:00 · 2026-07-27 13:00:16 +00:00
242 changed files with 11329 additions and 8102 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")
@@ -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(
        *,
@@ -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,
@@ -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
@@ -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:
@@ -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."""
@@ -497,6 +497,7 @@ class DifyNodeFactory(NodeFactory):
                ),
                "agent_backend_client": create_agent_backend_run_client(
                    base_url=dify_config.AGENT_BACKEND_BASE_URL,
+                    api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                    use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                    fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -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,
@@ -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 = {
@@ -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

@@ -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),
@@ -24,7 +24,10 @@ class TestWorkspaceService:
            patch("services.workspace_service.dify_config") as mock_dify_config,
        ):
            # Setup default mock returns
-            mock_feature_service.get_features.return_value.can_replace_logo = True
+            feature = mock_feature_service.get_features.return_value
+            feature.can_replace_logo = True
+            feature.billing.enabled = True
+            feature.billing.subscription.plan = "professional"
            mock_tenant_service.has_roles.return_value = True
            mock_dify_config.FILES_URL = "https://example.com/files"

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

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

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

        assert result is not None
+        assert result["plan"] is None
        assert "next_credit_reset_date" not in result
        assert "trial_credits" not in result
        assert "trial_credits_used" not in result
@@ -4,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__"):
@@ -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"
@@ -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()
@@ -435,6 +435,7 @@ def test_generate_success_returns_converted(generator, mocker: MockerFixture):
    mocker.patch.object(module, "PipelineQueueManager", return_value=queue_manager)

    worker_thread = MagicMock()
+    worker_thread.is_alive.return_value = False
    mocker.patch.object(module.threading, "Thread", return_value=worker_thread)

    mocker.patch.object(generator, "_get_draft_var_saver_factory", return_value=MagicMock())
@@ -461,6 +462,7 @@ def test_generate_success_returns_converted(generator, mocker: MockerFixture):
    )

    assert result == "converted"
+    worker_thread.join.assert_called_once_with(timeout=300)


 def test_single_iteration_generate_validates_inputs(generator, mocker: MockerFixture):
@@ -1,3 +1,6 @@
+import logging
+from unittest.mock import Mock
+
 import pytest

 from core.app.apps.base_app_generator import BaseAppGenerator
@@ -369,6 +372,58 @@ def test_validate_inputs_optional_file_with_empty_string_ignores_default():


 class TestBaseAppGeneratorExtras:
+    def test_wrap_stream_joins_worker_after_stream_exhaustion(self):
+        base_app_generator = BaseAppGenerator()
+        worker_thread = Mock()
+        worker_thread.is_alive.return_value = False
+
+        def response_stream():
+            yield {"event": "workflow_finished"}
+
+        managed_stream = base_app_generator._wrap_stream_with_worker_thread_join(
+            response_stream(),
+            worker_thread,
+        )
+
+        assert next(managed_stream) == {"event": "workflow_finished"}
+        worker_thread.join.assert_not_called()
+
+        with pytest.raises(StopIteration):
+            next(managed_stream)
+
+        worker_thread.join.assert_called_once_with(timeout=300)
+
+    def test_wrap_stream_joins_worker_when_stream_closes(self):
+        base_app_generator = BaseAppGenerator()
+        worker_thread = Mock()
+        worker_thread.is_alive.return_value = False
+
+        def response_stream():
+            yield {"event": "workflow_started"}
+            yield {"event": "workflow_finished"}
+
+        managed_stream = base_app_generator._wrap_stream_with_worker_thread_join(
+            response_stream(),
+            worker_thread,
+        )
+
+        assert next(managed_stream) == {"event": "workflow_started"}
+        managed_stream.close()
+
+        worker_thread.join.assert_called_once_with(timeout=300)
+
+    def test_join_worker_thread_warns_when_thread_remains_alive(self, caplog: pytest.LogCaptureFixture):
+        worker_thread = Mock()
+        worker_thread.name = "leaked-app-worker"
+        worker_thread.is_alive.return_value = True
+
+        with caplog.at_level(logging.WARNING, logger="core.app.apps.base_app_generator"):
+            BaseAppGenerator._join_worker_thread(worker_thread)
+
+        worker_thread.join.assert_called_once_with(timeout=300)
+        assert "Possible app worker thread leak" in caplog.text
+        assert "leaked-app-worker" in caplog.text
+
    def test_prepare_user_inputs_converts_files_and_lists(self, monkeypatch: pytest.MonkeyPatch):
        base_app_generator = BaseAppGenerator()

@@ -211,6 +211,13 @@ def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
        def start(self):
            return None

+        def join(self, timeout):
+            worker_kwargs["joined"] = True
+            worker_kwargs["join_timeout"] = timeout
+
+        def is_alive(self):
+            return False
+
    mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", DummyThread)

    app_model = SimpleNamespace(mode="workflow", tenant_id="tenant")
@@ -244,6 +251,8 @@ def test_generate_appends_pause_layer_and_forwards_state(mocker: MockerFixture):
    assert result == "converted"
    assert worker_kwargs["kwargs"]["graph_engine_layers"] == ("base-layer", pause_layer)
    assert worker_kwargs["kwargs"]["graph_runtime_state"] is graph_runtime_state
+    assert worker_kwargs["joined"] is True
+    assert worker_kwargs["join_timeout"] == 300
    assert draft_saver_factory.call_args.kwargs["tenant_id"] == app_model.tenant_id


@@ -286,6 +295,8 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):

    mocker.patch("core.app.apps.workflow.app_generator.WorkflowAppRunner", side_effect=runner_ctor)

+    worker_lifecycle: dict[str, bool] = {}
+
    class ImmediateThread:
        def __init__(self, target, kwargs):
            target(**kwargs)
@@ -293,6 +304,13 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
        def start(self):
            return None

+        def join(self, timeout):
+            worker_lifecycle["joined"] = True
+            worker_lifecycle["join_timeout"] = timeout
+
+        def is_alive(self):
+            return False
+
    mocker.patch("core.app.apps.workflow.app_generator.threading.Thread", ImmediateThread)

    mocker.patch(
@@ -331,5 +349,7 @@ def test_resume_path_runs_worker_with_runtime_state(mocker: MockerFixture):
    )

    assert result == "raw-response"
+    assert worker_lifecycle["joined"] is True
+    assert worker_lifecycle["join_timeout"] == 300
    runner_instance.run.assert_called_once()
    queue_manager.graph_runtime_state = runtime_state
@@ -1,5 +1,9 @@
+import threading
+from collections.abc import Generator
+
 import pytest

+from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.workflow.active_workflow_tasks import (
    active_workflow_task,
    get_active_workflow_task_count,
@@ -28,3 +32,51 @@ def test_active_workflow_task_rejects_duplicate_task_id() -> None:
        with pytest.raises(ValueError, match="already active"):
            with active_workflow_task("task-a"):
                pass
+
+
+def test_managed_stream_waits_for_active_worker_cleanup() -> None:
+    worker_started = threading.Event()
+    release_worker = threading.Event()
+    stream_exhausted = threading.Event()
+    consumer_finished = threading.Event()
+    consumer_errors: list[BaseException] = []
+
+    def run_worker() -> None:
+        with active_workflow_task("task-a"):
+            worker_started.set()
+            release_worker.wait()
+
+    def response_stream() -> Generator[dict[str, str], None, None]:
+        yield {"event": "workflow_finished"}
+        stream_exhausted.set()
+
+    worker_thread = threading.Thread(target=run_worker)
+    worker_thread.start()
+    assert worker_started.wait(timeout=2)
+
+    managed_stream = BaseAppGenerator._wrap_stream_with_worker_thread_join(response_stream(), worker_thread)
+    assert next(managed_stream) == {"event": "workflow_finished"}
+
+    def finish_stream() -> None:
+        try:
+            list(managed_stream)
+        except BaseException as exc:
+            consumer_errors.append(exc)
+        finally:
+            consumer_finished.set()
+
+    consumer_thread = threading.Thread(target=finish_stream)
+    consumer_thread.start()
+    try:
+        assert stream_exhausted.wait(timeout=2)
+        assert not consumer_finished.is_set()
+        assert get_active_workflow_task_count() == 1
+    finally:
+        release_worker.set()
+        consumer_thread.join(timeout=2)
+        worker_thread.join(timeout=2)
+
+    assert not consumer_thread.is_alive()
+    assert not worker_thread.is_alive()
+    assert consumer_errors == []
+    assert get_active_workflow_task_count() == 0
@@ -15,6 +15,70 @@ from models.model import AppMode


 class TestWorkflowAppGeneratorValidation:
+    def test_generate_stream_joins_worker_after_response_exhaustion(self, monkeypatch: pytest.MonkeyPatch):
+        generator = WorkflowAppGenerator()
+        worker_thread = Mock()
+        worker_thread.is_alive.return_value = False
+        app_config = WorkflowUIBasedAppConfig(
+            tenant_id="tenant",
+            app_id="app",
+            app_mode=AppMode.WORKFLOW,
+            additional_features=AppAdditionalFeatures(),
+            variables=[],
+            workflow_id="workflow-id",
+        )
+        application_generate_entity = WorkflowAppGenerateEntity.model_construct(
+            task_id="task",
+            app_config=app_config,
+            inputs={},
+            files=[],
+            user_id="user",
+            stream=True,
+            invoke_from=InvokeFrom.WEB_APP,
+            extras={},
+        )
+
+        def response_stream():
+            yield {"event": "workflow_finished"}
+
+        monkeypatch.setattr(generator, "_bind_file_access_scope", lambda **kwargs: contextlib.nullcontext())
+        monkeypatch.setattr(
+            "core.app.apps.workflow.app_generator.WorkflowAppQueueManager",
+            lambda **kwargs: SimpleNamespace(**kwargs),
+        )
+        monkeypatch.setattr(
+            "core.app.apps.workflow.app_generator.current_app",
+            SimpleNamespace(_get_current_object=lambda: SimpleNamespace(name="flask")),
+        )
+        monkeypatch.setattr("core.app.apps.workflow.app_generator.contextvars.copy_context", lambda: "ctx")
+        monkeypatch.setattr("core.app.apps.workflow.app_generator.threading.Thread", lambda **kwargs: worker_thread)
+        monkeypatch.setattr(
+            "core.app.apps.workflow.app_generator.db",
+            SimpleNamespace(session=SimpleNamespace(close=Mock())),
+        )
+        monkeypatch.setattr(generator, "_get_draft_var_saver_factory", lambda *args, **kwargs: "draft-factory")
+        monkeypatch.setattr(generator, "_handle_response", lambda **kwargs: response_stream())
+        monkeypatch.setattr(
+            "core.app.apps.workflow.app_generator.WorkflowAppGenerateResponseConverter.convert",
+            lambda response, invoke_from: response,
+        )
+
+        managed_stream = generator._generate(
+            app_model=SimpleNamespace(mode=AppMode.WORKFLOW, tenant_id="tenant"),
+            workflow=SimpleNamespace(id="workflow-id"),
+            user=SimpleNamespace(id="user"),
+            application_generate_entity=application_generate_entity,
+            invoke_from=InvokeFrom.WEB_APP,
+            workflow_execution_repository=SimpleNamespace(),
+            workflow_node_execution_repository=SimpleNamespace(),
+            streaming=True,
+        )
+
+        worker_thread.start.assert_called_once_with()
+        worker_thread.join.assert_not_called()
+        assert list(managed_stream) == [{"event": "workflow_finished"}]
+        worker_thread.join.assert_called_once_with(timeout=300)
+
    def test_ensure_snippet_start_node_returns_original_for_non_snippet_workflow(self):
        workflow = SimpleNamespace(kind_or_standard="workflow")
        session = SimpleNamespace(scalar=Mock())
@@ -1,24 +1,95 @@
 """Unit tests for the message cycle manager optimization."""

 import logging
+from collections.abc import Iterator
+from dataclasses import dataclass
 from types import SimpleNamespace
 from unittest.mock import Mock, patch

 import pytest
 from flask import Flask, current_app
+from sqlalchemy import Engine, event, select
+from sqlalchemy.orm import Session, sessionmaker

 from core.app.entities.queue_entities import QueueAnnotationReplyEvent, QueueRetrieverResourcesEvent
 from core.app.entities.task_entities import MessageStreamResponse, StreamEvent, TaskStateMetadata
+from core.app.task_pipeline import message_cycle_manager as message_cycle_manager_module
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.rag.entities import RetrievalSourceMetadata
-from models.model import App, AppMode
+from graphon.file import FileTransferMethod, FileType
+from models import model as model_module
+from models.base import TypeBase
+from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
+from models.model import App, AppMode, Conversation, MessageFile


-def _patch_create_session(mock_session):
-    session_cm = Mock()
-    session_cm.__enter__ = Mock(return_value=mock_session)
-    session_cm.__exit__ = Mock(return_value=False)
-    return patch("core.app.task_pipeline.message_cycle_manager.session_factory.create_session", return_value=session_cm)
+@dataclass(frozen=True)
+class _SQLiteDb:
+    engine: Engine
+    session: Session
+
+
+@pytest.fixture
+def cycle_db(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Session]:
+    """Bind request-owned and cycle-manager-owned sessions to isolated SQLite."""
+    TypeBase.metadata.create_all(
+        sqlite_engine,
+        tables=[App.__table__, Conversation.__table__, MessageFile.__table__],
+    )
+    owned_session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    with owned_session_factory() as request_session:
+        sqlite_db = _SQLiteDb(engine=sqlite_engine, session=request_session)
+        monkeypatch.setattr(message_cycle_manager_module, "db", sqlite_db)
+        monkeypatch.setattr(model_module, "db", sqlite_db)
+        monkeypatch.setattr(message_cycle_manager_module.session_factory, "create_session", owned_session_factory)
+        yield request_session
+
+
+def _app(*, app_id: str = "app-id", tenant_id: str = "tenant-1") -> App:
+    return App(
+        id=app_id,
+        tenant_id=tenant_id,
+        name="Test App",
+        description="",
+        mode=AppMode.CHAT,
+        enable_site=True,
+        enable_api=True,
+        max_active_requests=0,
+    )
+
+
+def _conversation(*, conversation_id: str = "conv-1", app_id: str = "app-id") -> Conversation:
+    conversation = Conversation(
+        app_id=app_id,
+        mode=AppMode.CHAT,
+        name="",
+        status="normal",
+        from_source=ConversationFromSource.API,
+        inputs={},
+    )
+    conversation.id = conversation_id
+    return conversation
+
+
+def _message_file(
+    *,
+    file_id: str = "file-1",
+    message_id: str = "test-message-id",
+    belongs_to: MessageFileBelongsTo | None = MessageFileBelongsTo.ASSISTANT,
+    url: str | None = "http://example.com/image.png",
+    file_type: FileType = FileType.IMAGE,
+) -> MessageFile:
+    message_file = MessageFile(
+        message_id=message_id,
+        type=file_type,
+        transfer_method=FileTransferMethod.TOOL_FILE,
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by="account-id",
+        belongs_to=belongs_to,
+        url=url,
+    )
+    message_file.id = file_id
+    return message_file


 class TestMessageCycleManagerOptimization:
@@ -37,30 +108,22 @@ class TestMessageCycleManagerOptimization:
        task_state = Mock()
        return MessageCycleManager(application_generate_entity=mock_application_generate_entity, task_state=task_state)

-    def test_get_message_event_type_with_assistant_file(self, message_cycle_manager):
+    def test_get_message_event_type_with_assistant_file(self, message_cycle_manager, cycle_db: Session):
        """Test get_message_event_type returns MESSAGE_FILE when message has assistant-generated files.

        This ensures that AI-generated images (belongs_to='assistant') trigger the MESSAGE_FILE event,
        allowing the frontend to properly display generated image files with url field.
        """
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            # Setup mock session and message file
-            mock_session = Mock()
-            mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
+        cycle_db.add(_message_file())
+        cycle_db.commit()

-            mock_message_file = Mock()
-            mock_message_file.belongs_to = "assistant"
-            mock_session.scalar.return_value = mock_message_file
+        with current_app.app_context():
+            result = message_cycle_manager.get_message_event_type("test-message-id")

-            # Execute
-            with current_app.app_context():
-                result = message_cycle_manager.get_message_event_type("test-message-id")
+        assert result == StreamEvent.MESSAGE_FILE
+        assert "test-message-id" in message_cycle_manager._message_has_file

-            # Assert
-            assert result == StreamEvent.MESSAGE_FILE
-            mock_session.scalar.assert_called_once()
-
-    def test_get_message_event_type_with_user_file(self, message_cycle_manager):
+    def test_get_message_event_type_with_user_file(self, message_cycle_manager, cycle_db: Session):
        """Test get_message_event_type returns MESSAGE when message only has user-uploaded files.

        This is a regression test for the issue where user-uploaded images (belongs_to='user')
@@ -68,90 +131,81 @@ class TestMessageCycleManagerOptimization:
        resulting in broken images in the chat UI. The query filters for belongs_to='assistant',
        so when only user files exist, the database query returns None, resulting in MESSAGE event type.
        """
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            # Setup mock session and message file
-            mock_session = Mock()
-            mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
+        cycle_db.add(_message_file(belongs_to=MessageFileBelongsTo.USER))
+        cycle_db.commit()

-            # When querying for assistant files with only user files present, return None
-            # (simulates database query with belongs_to='assistant' filter returning no results)
-            mock_session.scalar.return_value = None
+        with current_app.app_context():
+            result = message_cycle_manager.get_message_event_type("test-message-id")

-            # Execute
-            with current_app.app_context():
-                result = message_cycle_manager.get_message_event_type("test-message-id")
+        assert result == StreamEvent.MESSAGE
+        assert "test-message-id" not in message_cycle_manager._message_has_file

-            # Assert
-            assert result == StreamEvent.MESSAGE
-            mock_session.scalar.assert_called_once()
-
-    def test_get_message_event_type_without_message_file(self, message_cycle_manager):
+    def test_get_message_event_type_without_message_file(self, message_cycle_manager, cycle_db: Session):
        """Test get_message_event_type returns MESSAGE when message has no files."""
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            # Setup mock session and no message file
-            mock_session = Mock()
-            mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
-            # Current implementation uses session.scalar(select(...))
-            mock_session.scalar.return_value = None
+        assert list(cycle_db.scalars(select(MessageFile)).all()) == []

-            # Execute
-            with current_app.app_context():
-                result = message_cycle_manager.get_message_event_type("test-message-id")
+        with current_app.app_context():
+            result = message_cycle_manager.get_message_event_type("test-message-id")

-            # Assert
-            assert result == StreamEvent.MESSAGE
-            mock_session.scalar.assert_called_once()
+        assert result == StreamEvent.MESSAGE

-    def test_get_message_event_type_uses_cache_without_query(self, message_cycle_manager):
+    def test_get_message_event_type_uses_cache_without_query(
+        self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
+    ):
        """Return MESSAGE_FILE directly from in-memory cache without opening a DB session."""
        message_cycle_manager._message_has_file.add("cached-message")
+        statements: list[str] = []

-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
+        def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
+            statements.append(statement)
+
+        event.listen(sqlite_engine, "before_cursor_execute", record_statement)
+        try:
            result = message_cycle_manager.get_message_event_type("cached-message")
+        finally:
+            event.remove(sqlite_engine, "before_cursor_execute", record_statement)

        assert result == StreamEvent.MESSAGE_FILE
-        mock_session_factory.create_session.assert_not_called()
+        assert statements == []

-    def test_message_to_stream_response_with_precomputed_event_type(self, message_cycle_manager):
+    def test_message_to_stream_response_with_precomputed_event_type(self, message_cycle_manager, cycle_db: Session):
        """MessageCycleManager.message_to_stream_response expects a valid event_type; callers should precompute it."""
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            # Setup mock session and message file
-            mock_session = Mock()
-            mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
+        cycle_db.add(_message_file())
+        cycle_db.commit()

-            mock_message_file = Mock()
-            mock_message_file.belongs_to = "assistant"
-            mock_session.scalar.return_value = mock_message_file
+        with current_app.app_context():
+            event_type = message_cycle_manager.get_message_event_type("test-message-id")
+            result = message_cycle_manager.message_to_stream_response(
+                answer="Hello world", message_id="test-message-id", event_type=event_type
+            )

-            # Execute: compute event type once, then pass to message_to_stream_response
-            with current_app.app_context():
-                event_type = message_cycle_manager.get_message_event_type("test-message-id")
-                result = message_cycle_manager.message_to_stream_response(
-                    answer="Hello world", message_id="test-message-id", event_type=event_type
-                )
+        assert isinstance(result, MessageStreamResponse)
+        assert result.answer == "Hello world"
+        assert result.id == "test-message-id"
+        assert result.event == StreamEvent.MESSAGE_FILE

-            # Assert
-            assert isinstance(result, MessageStreamResponse)
-            assert result.answer == "Hello world"
-            assert result.id == "test-message-id"
-            assert result.event == StreamEvent.MESSAGE_FILE
-            mock_session.scalar.assert_called_once()
-
-    def test_message_to_stream_response_with_event_type_skips_query(self, message_cycle_manager):
+    def test_message_to_stream_response_with_event_type_skips_query(
+        self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
+    ):
        """Test that message_to_stream_response skips database query when event_type is provided."""
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            # Execute with event_type provided
+        statements: list[str] = []
+
+        def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
+            statements.append(statement)
+
+        event.listen(sqlite_engine, "before_cursor_execute", record_statement)
+        try:
            result = message_cycle_manager.message_to_stream_response(
                answer="Hello world", message_id="test-message-id", event_type=StreamEvent.MESSAGE
            )
+        finally:
+            event.remove(sqlite_engine, "before_cursor_execute", record_statement)

-            # Assert
-            assert isinstance(result, MessageStreamResponse)
-            assert result.answer == "Hello world"
-            assert result.id == "test-message-id"
-            assert result.event == StreamEvent.MESSAGE
-            # Should not open a session when event_type is provided
-            mock_session_factory.create_session.assert_not_called()
+        assert isinstance(result, MessageStreamResponse)
+        assert result.answer == "Hello world"
+        assert result.id == "test-message-id"
+        assert result.event == StreamEvent.MESSAGE
+        assert statements == []

    def test_message_to_stream_response_with_from_variable_selector(self, message_cycle_manager):
        """Test message_to_stream_response with from_variable_selector parameter."""
@@ -168,40 +222,32 @@ class TestMessageCycleManagerOptimization:
        assert result.from_variable_selector == ["var1", "var2"]
        assert result.event == StreamEvent.MESSAGE

-    def test_optimization_usage_example(self, message_cycle_manager):
+    def test_optimization_usage_example(self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine):
        """Test the optimization pattern that should be used by callers."""
-        # Step 1: Get event type once (this queries database)
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            mock_session = Mock()
-            mock_session_factory.create_session.return_value.__enter__.return_value = mock_session
-            # Current implementation uses session.scalar(select(...))
-            mock_session.scalar.return_value = None  # No files
+        statements: list[str] = []
+
+        def record_statement(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
+            statements.append(statement)
+
+        event.listen(sqlite_engine, "before_cursor_execute", record_statement)
+        try:
            with current_app.app_context():
                event_type = message_cycle_manager.get_message_event_type("test-message-id")
-
-        # Should open session once
-        mock_session_factory.create_session.assert_called_once()
-        assert event_type == StreamEvent.MESSAGE
-
-        # Step 2: Use event_type for multiple calls (no additional queries)
-        with patch("core.app.task_pipeline.message_cycle_manager.session_factory") as mock_session_factory:
-            mock_session_factory.create_session.return_value.__enter__.return_value = Mock()
-
            chunk1_response = message_cycle_manager.message_to_stream_response(
                answer="Chunk 1", message_id="test-message-id", event_type=event_type
            )
-
            chunk2_response = message_cycle_manager.message_to_stream_response(
                answer="Chunk 2", message_id="test-message-id", event_type=event_type
            )
+        finally:
+            event.remove(sqlite_engine, "before_cursor_execute", record_statement)

-            # Should not open session again when event_type provided
-            mock_session_factory.create_session.assert_not_called()
-
-            assert chunk1_response.event == StreamEvent.MESSAGE
-            assert chunk2_response.event == StreamEvent.MESSAGE
-            assert chunk1_response.answer == "Chunk 1"
-            assert chunk2_response.answer == "Chunk 2"
+        assert event_type == StreamEvent.MESSAGE
+        assert len([statement for statement in statements if statement.lstrip().upper().startswith("SELECT")]) == 1
+        assert chunk1_response.event == StreamEvent.MESSAGE
+        assert chunk2_response.event == StreamEvent.MESSAGE
+        assert chunk1_response.answer == "Chunk 1"
+        assert chunk2_response.answer == "Chunk 2"

    def test_generate_conversation_name_returns_none_for_completion(self, message_cycle_manager):
        """Return None when completion entities are used for conversation naming.
@@ -269,51 +315,38 @@ class TestMessageCycleManagerOptimization:
        assert message_cycle_manager._application_generate_entity.is_new_conversation is False
        mock_timer.assert_not_called()

-    def test_generate_conversation_name_worker_returns_when_conversation_missing(self, message_cycle_manager):
+    def test_generate_conversation_name_worker_returns_when_conversation_missing(
+        self, message_cycle_manager, cycle_db: Session
+    ):
        """Return early when the conversation cannot be found."""
        flask_app = Flask(__name__)
-        db_session = Mock()
-        db_session.scalar.return_value = None
+        assert list(cycle_db.scalars(select(Conversation)).all()) == []

-        with _patch_create_session(db_session):
-            message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-missing", "hello")
+        message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-missing", "hello")

-        db_session.commit.assert_not_called()
+        assert list(cycle_db.scalars(select(Conversation)).all()) == []

-    def test_generate_conversation_name_worker_returns_when_app_missing(self, message_cycle_manager):
+    def test_generate_conversation_name_worker_returns_when_app_missing(self, message_cycle_manager, cycle_db: Session):
        """Return early when non-completion conversation has no app relation."""
        flask_app = Flask(__name__)
-        conversation = SimpleNamespace(mode=AppMode.CHAT, app=None, app_id="app-id")
-        db_session = Mock()
-        db_session.scalar.return_value = conversation
-        db_session.get.return_value = None
+        conversation = _conversation()
+        cycle_db.add(conversation)
+        cycle_db.commit()

-        with _patch_create_session(db_session):
-            message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")
+        message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")

-        db_session.commit.assert_not_called()
+        assert cycle_db.get(Conversation, "conv-1").name == ""
+        assert cycle_db.get(App, "app-id") is None

-    def test_generate_conversation_name_worker_uses_cached_name(self, message_cycle_manager):
+    def test_generate_conversation_name_worker_uses_cached_name(
+        self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
+    ):
        """Use cached conversation name when present and avoid LLM call."""
        flask_app = Flask(__name__)
-
-        class ConversationWithPoisonedApp:
-            mode = AppMode.CHAT
-            app_id = "app-id"
-            name = ""
-
-            @property
-            def app(self):
-                raise AssertionError("conversation.app must not open an implicit session")
-
-        conversation = ConversationWithPoisonedApp()
-        app_model = SimpleNamespace(tenant_id="tenant-1")
-        db_session = Mock()
-        db_session.scalar.return_value = conversation
-        db_session.get.return_value = app_model
+        cycle_db.add_all([_app(), _conversation()])
+        cycle_db.commit()

        with (
-            _patch_create_session(db_session) as create_session,
            patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
            patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
        ):
@@ -321,27 +354,23 @@ class TestMessageCycleManagerOptimization:

            message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")

+        assert cycle_db.in_transaction() is False
+        with Session(sqlite_engine) as verification_session:
+            conversation = verification_session.get(Conversation, "conv-1")
+        assert conversation is not None
        assert conversation.name == "cached-title"
-        create_session.assert_called_once_with()
-        db_session.get.assert_called_once_with(App, "app-id")
-        db_session.commit.assert_called_once()
        mock_llm_generator.generate_conversation_name.assert_not_called()
        mock_redis.setex.assert_not_called()

-    def test_generate_conversation_name_worker_generates_and_caches_name(self, message_cycle_manager):
+    def test_generate_conversation_name_worker_generates_and_caches_name(
+        self, message_cycle_manager, cycle_db: Session, sqlite_engine: Engine
+    ):
        """Generate conversation name and write it to redis cache on cache miss."""
        flask_app = Flask(__name__)
-        conversation = SimpleNamespace(
-            mode=AppMode.CHAT,
-            app=SimpleNamespace(tenant_id="tenant-1"),
-            app_id="app-id",
-            name="",
-        )
-        db_session = Mock()
-        db_session.scalar.return_value = conversation
+        cycle_db.add_all([_app(), _conversation()])
+        cycle_db.commit()

        with (
-            _patch_create_session(db_session),
            patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
            patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
        ):
@@ -350,27 +379,27 @@ class TestMessageCycleManagerOptimization:

            message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", "hello")

+        assert cycle_db.in_transaction() is False
+        with Session(sqlite_engine) as verification_session:
+            conversation = verification_session.get(Conversation, "conv-1")
+        assert conversation is not None
        assert conversation.name == "generated-title"
-        db_session.commit.assert_called_once()
        mock_redis.setex.assert_called_once()

    def test_generate_conversation_name_worker_falls_back_when_generation_fails(
-        self, message_cycle_manager, caplog: pytest.LogCaptureFixture
+        self,
+        message_cycle_manager,
+        cycle_db: Session,
+        sqlite_engine: Engine,
+        caplog: pytest.LogCaptureFixture,
    ):
        """Fallback to truncated query when LLM generation fails."""
        flask_app = Flask(__name__)
-        conversation = SimpleNamespace(
-            mode=AppMode.CHAT,
-            app=SimpleNamespace(tenant_id="tenant-1"),
-            app_id="app-id",
-            name="",
-        )
-        db_session = Mock()
-        db_session.scalar.return_value = conversation
+        cycle_db.add_all([_app(), _conversation()])
+        cycle_db.commit()
        long_query = "q" * 60

        with (
-            _patch_create_session(db_session),
            patch("core.app.task_pipeline.message_cycle_manager.redis_client") as mock_redis,
            patch("core.app.task_pipeline.message_cycle_manager.LLMGenerator") as mock_llm_generator,
            patch("core.app.task_pipeline.message_cycle_manager.dify_config") as mock_dify_config,
@@ -382,8 +411,11 @@ class TestMessageCycleManagerOptimization:
            with caplog.at_level(logging.ERROR, logger="core.app.task_pipeline.message_cycle_manager"):
                message_cycle_manager._generate_conversation_name_worker(flask_app, "conv-1", long_query)

+        assert cycle_db.in_transaction() is False
+        with Session(sqlite_engine) as verification_session:
+            conversation = verification_session.get(Conversation, "conv-1")
+        assert conversation is not None
        assert conversation.name == (long_query[:47] + "...")
-        db_session.commit.assert_called_once()
        assert any(record.levelno == logging.ERROR for record in caplog.records)

    def test_handle_annotation_reply_sets_metadata(self, message_cycle_manager):
@@ -454,33 +486,25 @@ class TestMessageCycleManagerOptimization:
        assert message_cycle_manager._task_state.metadata.retriever_resources[0].position == 1
        assert message_cycle_manager._task_state.metadata.retriever_resources[1].position == 2

-    def test_message_file_to_stream_response_builds_signed_url(self, message_cycle_manager):
+    def test_message_file_to_stream_response_builds_signed_url(self, message_cycle_manager, cycle_db: Session):
        """Build a stream response with a signed tool file URL.

-        Args: message_cycle_manager with mocked Session/db and sign_tool_file.
+        Args: message_cycle_manager with a persisted MessageFile and mocked sign_tool_file.
        Returns: MessageStreamResponse with signed url and belongs_to normalized to user.
        Side effects: Calls sign_tool_file for tool file ids.
        """
        message_cycle_manager._application_generate_entity.task_id = "task-1"
-
-        message_file = SimpleNamespace(
-            id="file-1",
-            type="image",
-            belongs_to=None,
-            url="tool://file.verylongextension",
-            message_id="msg-1",
+        cycle_db.add(
+            _message_file(
+                file_id="file-1",
+                message_id="msg-1",
+                belongs_to=None,
+                url="tool://file.verylongextension",
+            )
        )
+        cycle_db.commit()

-        session = Mock()
-        session.scalar.return_value = message_file
-
-        with (
-            patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
-            patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign,
-            patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
-        ):
-            mock_db.engine = Mock()
-            mock_session_cls.return_value.__enter__.return_value = session
+        with patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign:
            mock_sign.return_value = "signed-url"

            response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="file-1"))
@@ -514,56 +538,42 @@ class TestMessageCycleManagerOptimization:
        assert len(message_cycle_manager._task_state.metadata.retriever_resources) == 1
        assert message_cycle_manager._task_state.metadata.retriever_resources[0].position == 1

-    def test_message_file_to_stream_response_uses_http_url_directly(self, message_cycle_manager):
+    def test_message_file_to_stream_response_uses_http_url_directly(self, message_cycle_manager, cycle_db: Session):
        """Use original URL when message file URL is already HTTP."""
        message_cycle_manager._application_generate_entity.task_id = "task-http"
-        message_file = SimpleNamespace(
-            id="file-http",
-            type="image",
-            belongs_to="assistant",
-            url="http://example.com/pic.png",
-            message_id="msg-http",
-        )
-
-        session = Mock()
-        session.scalar.return_value = message_file
-
-        with (
-            patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
-            patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
-        ):
-            mock_db.engine = Mock()
-            mock_session_cls.return_value.__enter__.return_value = session
-
-            response = message_cycle_manager.message_file_to_stream_response(
-                SimpleNamespace(message_file_id="file-http")
+        cycle_db.add(
+            _message_file(
+                file_id="file-http",
+                message_id="msg-http",
+                belongs_to=MessageFileBelongsTo.ASSISTANT,
+                url="http://example.com/pic.png",
            )
+        )
+        cycle_db.commit()
+
+        response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="file-http"))

        assert response is not None
        assert response.url == "http://example.com/pic.png"
        assert "msg-http" in message_cycle_manager._message_has_file

-    def test_message_file_to_stream_response_defaults_extension_to_bin_without_dot(self, message_cycle_manager):
+    def test_message_file_to_stream_response_defaults_extension_to_bin_without_dot(
+        self, message_cycle_manager, cycle_db: Session
+    ):
        """Default tool file extension to .bin when URL has no extension part."""
        message_cycle_manager._application_generate_entity.task_id = "task-bin"
-        message_file = SimpleNamespace(
-            id="file-bin",
-            type="file",
-            belongs_to="assistant",
-            url="tool-file-id",
-            message_id="msg-bin",
+        cycle_db.add(
+            _message_file(
+                file_id="file-bin",
+                message_id="msg-bin",
+                belongs_to=MessageFileBelongsTo.ASSISTANT,
+                url="tool-file-id",
+                file_type=FileType.CUSTOM,
+            )
        )
+        cycle_db.commit()

-        session = Mock()
-        session.scalar.return_value = message_file
-
-        with (
-            patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
-            patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign,
-            patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
-        ):
-            mock_db.engine = Mock()
-            mock_session_cls.return_value.__enter__.return_value = session
+        with patch("core.app.task_pipeline.message_cycle_manager.sign_tool_file") as mock_sign:
            mock_sign.return_value = "signed-bin-url"

            response = message_cycle_manager.message_file_to_stream_response(
@@ -574,19 +584,13 @@ class TestMessageCycleManagerOptimization:
        assert response.url == "signed-bin-url"
        mock_sign.assert_called_once_with(tool_file_id="tool-file-id", extension=".bin")

-    def test_message_file_to_stream_response_returns_none_when_file_missing(self, message_cycle_manager):
+    def test_message_file_to_stream_response_returns_none_when_file_missing(
+        self, message_cycle_manager, cycle_db: Session
+    ):
        """Return None when message file lookup does not find a record."""
-        session = Mock()
-        session.scalar.return_value = None
+        assert list(cycle_db.scalars(select(MessageFile)).all()) == []

-        with (
-            patch("core.app.task_pipeline.message_cycle_manager.Session") as mock_session_cls,
-            patch("core.app.task_pipeline.message_cycle_manager.db") as mock_db,
-        ):
-            mock_db.engine = Mock()
-            mock_session_cls.return_value.__enter__.return_value = session
-
-            response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="missing"))
+        response = message_cycle_manager.message_file_to_stream_response(SimpleNamespace(message_file_id="missing"))

        assert response is None

@@ -0,0 +1,67 @@
+"""Verify that Jinja2 transform_caller scripts block unsafe template attacks via SandboxedEnvironment."""
+
+import io
+import sys
+
+import pytest
+
+from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTransformer
+
+MALICIOUS_TEMPLATES = [
+    pytest.param(
+        "{{ lipsum.__globals__.__builtins__.__import__('os').popen('id').read() }}",
+        id="lipsum_globals_builtins",
+    ),
+    pytest.param(
+        "{{ ''.__class__.__mro__[1].__subclasses__() }}",
+        id="string_class_mro",
+    ),
+    pytest.param(
+        "{{ cycler.__init__.__globals__.os.popen('whoami').read() }}",
+        id="cycler_init_globals",
+    ),
+    pytest.param(
+        "{{ namespace.__init__.__globals__['__builtins__']['__import__']('os').system('id') }}",
+        id="namespace_init_globals",
+    ),
+]
+
+
+def _exec_scripts(runner: str, preload: str) -> str:
+    """Execute preload then runner in a shared namespace, return captured stdout."""
+    ns: dict = {}
+    exec(compile(preload, "<preload>", "exec"), ns)  # noqa: S102
+    captured = io.StringIO()
+    old_stdout = sys.stdout
+    sys.stdout = captured
+    try:
+        exec(compile(runner, "<runner>", "exec"), ns)  # noqa: S102
+    finally:
+        sys.stdout = old_stdout
+    return captured.getvalue()
+
+
+class TestJinja2TransformCallerSandbox:
+    """Test transform_caller output (runner + preload) blocks attacks and allows safe templates."""
+
+    @pytest.mark.parametrize("malicious_template", MALICIOUS_TEMPLATES)
+    def test_blocks_unsafe_template(self, malicious_template: str) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller(malicious_template, {})
+        ns: dict = {}
+        exec(compile(preload, "<preload>", "exec"), ns)  # noqa: S102
+        with pytest.raises(Exception) as exc_info:
+            exec(compile(runner, "<runner>", "exec"), ns)  # noqa: S102
+        assert "unsafe" in str(exc_info.value).lower() or "security" in str(exc_info.value).lower()
+
+    def test_renders_safe_template(self) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller(
+            "Hello {{ name }}, you are {{ age }} years old!",
+            {"name": "Alice", "age": 30},
+        )
+        output = _exec_scripts(runner, preload)
+        assert "Hello Alice, you are 30 years old!" in output
+
+    def test_scripts_use_sandboxed_environment(self) -> None:
+        runner, preload = Jinja2TemplateTransformer.transform_caller("{{ x }}", {"x": 1})
+        assert "SandboxedEnvironment" in runner
+        assert "SandboxedEnvironment" in preload
@@ -1,7 +1,57 @@
-import sys
+from datetime import datetime, timedelta
 from unittest.mock import MagicMock, patch

+import pytest
+from sqlalchemy import event
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session, sessionmaker
+
+import core.llm_generator.llm_generator as generator_module
 from core.llm_generator.llm_generator import LLMGenerator, _parse_string_list
+from core.model_manager import ModelInstance, ModelManager
+from core.workflow.generator import tool_catalogue as tool_catalogue_module
+from core.workflow.generator.tool_catalogue import ToolCatalogueEntry
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from graphon.model_runtime.entities.message_entities import AssistantPromptMessage
+from models.dataset import Dataset
+from services.workflow_service import WorkflowService
+
+
+@pytest.fixture
+def dataset_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
+    """Bind the real SQLite session to the production database extension."""
+
+    monkeypatch.setattr(generator_module.db, "session", sqlite_session)
+    return sqlite_session
+
+
+def _llm_result(content: str) -> LLMResult:
+    """Build a real non-streaming LLM response around deterministic test content."""
+
+    return LLMResult(
+        model="test-model",
+        message=AssistantPromptMessage(content=content),
+        usage=LLMUsage.empty_usage(),
+    )
+
+
+def _model_manager() -> tuple[MagicMock, MagicMock]:
+    """Build spec-constrained mocks for the model-manager boundary and its default model."""
+
+    model_manager = MagicMock(spec=ModelManager)
+    model_instance = MagicMock(spec=ModelInstance)
+    model_manager.get_default_model_instance.return_value = model_instance
+    return model_manager, model_instance
+
+
+def _dataset(*, dataset_id: str, tenant_id: str, name: str, created_at: datetime) -> Dataset:
+    return Dataset(
+        id=dataset_id,
+        tenant_id=tenant_id,
+        name=name,
+        created_by="account-id",
+        created_at=created_at,
+    )


 class TestParseStringList:
@@ -34,95 +84,115 @@ class TestParseStringList:
 class TestGenerateWorkflowInstructionSuggestions:
    @patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
    def test_no_default_model(self, mock_for_tenant):
-        mock_for_tenant.return_value.get_default_model_instance.side_effect = Exception("No model")
+        model_manager, _ = _model_manager()
+        model_manager.get_default_model_instance.side_effect = RuntimeError("no default model")
+        mock_for_tenant.return_value = model_manager
+
        assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []

    @patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
    @patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
    def test_llm_success(self, mock_build_context, mock_for_tenant):
        mock_build_context.return_value = "context"
-
-        mock_model = MagicMock()
-        mock_model.invoke_llm.return_value = MagicMock()
-        mock_model.invoke_llm.return_value.message.get_text_content.return_value = '["idea 1", "idea 2"]'
-
-        mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
+        model_manager, model_instance = _model_manager()
+        model_instance.invoke_llm.return_value = _llm_result('["idea 1", "idea 2"]')
+        mock_for_tenant.return_value = model_manager

        result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
        assert result == ["idea 1", "idea 2"]
+        model_instance.invoke_llm.assert_called_once()

    @patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
    @patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
    def test_llm_error(self, mock_build_context, mock_for_tenant):
        mock_build_context.return_value = "context"
+        model_manager, model_instance = _model_manager()
+        model_instance.invoke_llm.side_effect = RuntimeError("API error")
+        mock_for_tenant.return_value = model_manager

-        mock_model = MagicMock()
-        mock_model.invoke_llm.side_effect = Exception("API error")
-
-        mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
-
-        assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []
+        result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
+        assert result == []
+        model_instance.invoke_llm.assert_called_once()

    @patch("core.llm_generator.llm_generator.ModelManager.for_tenant")
    @patch("core.llm_generator.llm_generator.LLMGenerator._build_suggestion_context")
    def test_llm_bad_output(self, mock_build_context, mock_for_tenant):
        mock_build_context.return_value = "context"
+        model_manager, model_instance = _model_manager()
+        model_instance.invoke_llm.return_value = _llm_result("Not a list")
+        mock_for_tenant.return_value = model_manager

-        mock_model = MagicMock()
-        mock_model.invoke_llm.return_value = MagicMock()
-        mock_model.invoke_llm.return_value.message.get_text_content.return_value = "Not a list"
-
-        mock_for_tenant.return_value.get_default_model_instance.return_value = mock_model
-
-        assert LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow") == []
+        result = LLMGenerator.generate_workflow_instruction_suggestions("tenant", mode="workflow")
+        assert result == []
+        model_instance.invoke_llm.assert_called_once()


+@pytest.mark.parametrize("sqlite_session", [(Dataset,)], indirect=True)
 class TestBuildSuggestionContext:
-    @patch("core.llm_generator.llm_generator.db.session.scalars")
-    def test_both_success(self, mock_scalars, monkeypatch):
-        mock_scalars.return_value.all.return_value = ["kb1", "kb2"]
+    def test_both_success(self, dataset_session: Session, monkeypatch: pytest.MonkeyPatch):
+        now = datetime.now()
+        dataset_session.add_all(
+            (
+                _dataset(dataset_id="kb-1", tenant_id="tenant", name="kb1", created_at=now),
+                _dataset(
+                    dataset_id="kb-2",
+                    tenant_id="tenant",
+                    name="kb2",
+                    created_at=now - timedelta(seconds=1),
+                ),
+                _dataset(dataset_id="other-kb", tenant_id="other", name="private", created_at=now),
+            )
+        )
+        dataset_session.commit()

-        # ``_build_suggestion_context`` imports the tool catalogue lazily, so we
-        # stub the module in ``sys.modules``. Use ``monkeypatch.setitem`` so the
-        # ORIGINAL module is RESTORED on teardown — a bare ``del`` would evict it
-        # from sys.modules entirely, after which a sibling test that imported
-        # ``build_tool_catalogue`` at collection time (e.g. test_tool_catalogue)
-        # diverges from a freshly re-imported module and its @patch targets stop
-        # applying, silently breaking it under xdist.
-        mock_tool_catalogue = MagicMock()
-        mock_tool_catalogue.build_tool_catalogue.return_value = "catalog"
-        mock_tool_catalogue.format_tool_catalogue.return_value = "tool1\ntool2"
-        monkeypatch.setitem(sys.modules, "core.workflow.generator.tool_catalogue", mock_tool_catalogue)
+        def build_tool_catalogue(_tenant_id: str) -> list[ToolCatalogueEntry]:
+            return [
+                ToolCatalogueEntry(
+                    provider_name="provider",
+                    provider_type="builtin",
+                    plugin_id="",
+                    tool_name="tool1",
+                    tool_label="tool1",
+                    description="First tool",
+                ),
+                ToolCatalogueEntry(
+                    provider_name="provider",
+                    provider_type="builtin",
+                    plugin_id="",
+                    tool_name="tool2",
+                    tool_label="tool2",
+                    description="Second tool",
+                ),
+            ]
+
+        # Keep the real module and formatter; only isolate provider/plugin discovery.
+        monkeypatch.setattr(tool_catalogue_module, "build_tool_catalogue", build_tool_catalogue)

        result = LLMGenerator._build_suggestion_context("tenant")
        assert "Knowledge bases:\n- kb1\n- kb2" in result
-        assert "Installed tools:\ntool1\ntool2" in result
+        assert "Installed tools:\n- provider/tool1 — First tool\n- provider/tool2 — Second tool" in result

-    @patch("core.llm_generator.llm_generator.db.session.scalars")
-    def test_both_fail(self, mock_scalars, monkeypatch):
-        mock_scalars.side_effect = Exception("DB error")
+    def test_both_fail(self, dataset_session: Session, monkeypatch: pytest.MonkeyPatch):
+        def fail_query(_orm_execute_state: object) -> None:
+            raise SQLAlchemyError("DB error")

-        # See ``test_both_success``: restore the original module via monkeypatch
-        # rather than ``del``-ing it, so we don't evict it for sibling tests.
-        mock_tool_catalogue = MagicMock()
-        mock_tool_catalogue.build_tool_catalogue.side_effect = Exception("Tool error")
-        monkeypatch.setitem(sys.modules, "core.workflow.generator.tool_catalogue", mock_tool_catalogue)
+        def fail_tool_catalogue(_tenant_id: str) -> list[ToolCatalogueEntry]:
+            raise RuntimeError("Tool error")

-        assert LLMGenerator._build_suggestion_context("tenant") == ""
+        event.listen(dataset_session, "do_orm_execute", fail_query)
+        monkeypatch.setattr(tool_catalogue_module, "build_tool_catalogue", fail_tool_catalogue)
+
+        try:
+            assert LLMGenerator._build_suggestion_context("tenant") == ""
+        finally:
+            event.remove(dataset_session, "do_orm_execute", fail_query)


 class TestWorkflowServiceInterface:
-    def test_protocol_methods(self):
-        # Just to cover the 'pass' statements in the Protocol definition
+    def test_real_workflow_service_exposes_protocol_methods(self):
        from core.llm_generator.llm_generator import WorkflowServiceInterface

-        class MockService(WorkflowServiceInterface):
-            def get_draft_workflow(self, app_model, workflow_id=None, *, session):
-                return super().get_draft_workflow(app_model, workflow_id, session=session)
+        service: WorkflowServiceInterface = WorkflowService(sessionmaker())

-            def get_node_last_run(self, app_model, workflow, node_id):
-                return super().get_node_last_run(app_model, workflow, node_id)
-
-        service = MockService()
-        service.get_draft_workflow(None, session=None)
-        service.get_node_last_run(None, None, "node")
+        assert callable(service.get_draft_workflow)
+        assert callable(service.get_node_last_run)
@@ -1,11 +1,19 @@
-"""Comprehensive unit tests for core/memory/token_buffer_memory.py"""
+"""Comprehensive SQLite-backed tests for token-buffer memory."""

+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
 from unittest.mock import MagicMock, patch
 from uuid import uuid4

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

+from core.memory import token_buffer_memory as memory_module
 from core.memory.token_buffer_memory import TokenBufferMemory
+from graphon.file import FileTransferMethod, FileType
 from graphon.model_runtime.entities import (
    AssistantPromptMessage,
    ImagePromptMessageContent,
@@ -13,13 +21,44 @@ from graphon.model_runtime.entities import (
    TextPromptMessageContent,
    UserPromptMessage,
 )
-from models.model import AppMode
+from models.base import TypeBase
+from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
+from models.model import AppMode, Message, MessageFile
+from models.workflow import Workflow, WorkflowType

 # ---------------------------------------------------------------------------
 # Helpers / shared fixtures
 # ---------------------------------------------------------------------------


+@dataclass(frozen=True)
+class Database:
+    """Typed SQLite binding plus executed SQL for query-count assertions."""
+
+    engine: Engine
+    session: Session
+    statements: list[tuple[str, object]]
+
+
+@pytest.fixture
+def database(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> Iterator[Database]:
+    TypeBase.metadata.create_all(
+        sqlite_engine,
+        tables=[Message.__table__, MessageFile.__table__, Workflow.__table__],
+    )
+    statements: list[tuple[str, object]] = []
+
+    def record_statement(_connection, _cursor, statement, parameters, _context, _executemany) -> None:
+        statements.append((statement, parameters))
+
+    event.listen(sqlite_engine, "before_cursor_execute", record_statement)
+    with Session(sqlite_engine, expire_on_commit=False) as session:
+        database = Database(engine=sqlite_engine, session=session, statements=statements)
+        monkeypatch.setattr(memory_module, "db", database)
+        yield database
+    event.remove(sqlite_engine, "before_cursor_execute", record_statement)
+
+
 def _make_conversation(mode: AppMode = AppMode.CHAT) -> MagicMock:
    """Return a minimal Conversation mock."""
    conv = MagicMock()
@@ -47,6 +86,73 @@ def _make_message(answer: str = "hello", answer_tokens: int = 5) -> MagicMock:
    return msg


+def _persist_message(
+    database: Database,
+    conversation_id: str,
+    *,
+    query: str = "user query",
+    answer: str = "hello",
+    answer_tokens: int = 5,
+    created_at: datetime | None = None,
+    workflow_run_id: str | None = None,
+) -> Message:
+    message = Message(
+        id=str(uuid4()),
+        app_id="app-1",
+        conversation_id=conversation_id,
+        _inputs={},
+        query=query,
+        message={},
+        message_unit_price=Decimal(0),
+        answer=answer,
+        answer_tokens=answer_tokens,
+        answer_unit_price=Decimal(0),
+        currency="USD",
+        from_source=ConversationFromSource.API,
+        workflow_run_id=workflow_run_id,
+        created_at=created_at or datetime.now(UTC).replace(tzinfo=None),
+    )
+    database.session.add(message)
+    database.session.commit()
+    return message
+
+
+def _persist_message_file(
+    database: Database,
+    message: Message,
+    *,
+    belongs_to: MessageFileBelongsTo | None,
+) -> MessageFile:
+    message_file = MessageFile(
+        message_id=message.id,
+        type=FileType.IMAGE,
+        transfer_method=FileTransferMethod.REMOTE_URL,
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by="account-1",
+        belongs_to=belongs_to,
+        url="https://example.com/image.png",
+    )
+    database.session.add(message_file)
+    database.session.commit()
+    return message_file
+
+
+def _persist_workflow(database: Database, *, workflow_id: str) -> Workflow:
+    workflow = Workflow(
+        id=workflow_id,
+        tenant_id="tenant-1",
+        app_id="app-1",
+        type=WorkflowType.CHAT,
+        version="1",
+        graph="{}",
+        features="{}",
+        created_by="account-1",
+    )
+    database.session.add(workflow)
+    database.session.commit()
+    return workflow
+
+
 # ===========================================================================
 # Tests for __init__ and workflow_run_repo property
 # ===========================================================================
@@ -61,25 +167,25 @@ class TestInit:
        assert mem.model_instance is mi
        assert mem._workflow_run_repo is None

-    def test_workflow_run_repo_is_created_lazily(self):
+    def test_workflow_run_repo_is_created_lazily(self, database: Database):
        conv = _make_conversation()
        mi = _make_model_instance()
        mem = TokenBufferMemory(conversation=conv, model_instance=mi)

        mock_repo = MagicMock()
-        with (
-            patch("core.memory.token_buffer_memory.sessionmaker") as mock_sm,
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
-                return_value=mock_repo,
-            ),
-        ):
-            mock_db.engine = MagicMock()
+        with patch(
+            "core.memory.token_buffer_memory.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
+            return_value=mock_repo,
+        ) as repository_factory:
            repo = mem.workflow_run_repo
            assert repo is mock_repo
            assert mem._workflow_run_repo is mock_repo

+        session_factory = repository_factory.call_args.args[0]
+        with session_factory() as session:
+            assert isinstance(session, Session)
+            assert session.get_bind() is database.engine
+
    def test_workflow_run_repo_cached_after_first_access(self):
        conv = _make_conversation()
        mi = _make_model_instance()
@@ -410,7 +516,7 @@ class TestBuildPromptMessageWithFiles:
            )

    @pytest.mark.parametrize("mode", [AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def test_workflow_mode_workflow_not_found_raises(self, mode):
+    def test_workflow_mode_workflow_not_found_raises(self, mode, database: Database):
        """Raises ValueError when Workflow lookup returns None."""
        conv = _make_conversation(mode)
        conv.app = MagicMock()
@@ -422,22 +528,17 @@ class TestBuildPromptMessageWithFiles:
        mem._workflow_run_repo = MagicMock()
        mem._workflow_run_repo.get_workflow_run_by_id.return_value = mock_workflow_run

-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-        ):
-            mock_db.session.scalar.return_value = None  # workflow not found
-
-            with pytest.raises(ValueError, match="Workflow not found"):
-                mem._build_prompt_message_with_files(
-                    message_files=[],
-                    text_content="text",
-                    message=_make_message(),
-                    app_record=MagicMock(),
-                    is_user_message=True,
-                )
+        with pytest.raises(ValueError, match="Workflow not found"):
+            mem._build_prompt_message_with_files(
+                message_files=[],
+                text_content="text",
+                message=_make_message(),
+                app_record=MagicMock(),
+                is_user_message=True,
+            )

    @pytest.mark.parametrize("mode", [AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def test_workflow_mode_success_no_files_user(self, mode):
+    def test_workflow_mode_success_no_files_user(self, mode, database: Database):
        """Happy path: workflow mode, no message files → plain UserPromptMessage."""
        conv = _make_conversation(mode)
        conv.app = MagicMock()
@@ -445,22 +546,16 @@ class TestBuildPromptMessageWithFiles:
        mock_workflow_run = MagicMock()
        mock_workflow_run.workflow_id = str(uuid4())

-        mock_workflow = MagicMock()
-        mock_workflow.features_dict = {}
+        workflow = _persist_workflow(database, workflow_id=mock_workflow_run.workflow_id)

        mem = TokenBufferMemory(conversation=conv, model_instance=_make_model_instance())
        mem._workflow_run_repo = MagicMock()
        mem._workflow_run_repo.get_workflow_run_by_id.return_value = mock_workflow_run

-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
+        with patch(
+            "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
+            return_value=None,
        ):
-            mock_db.session.scalar.return_value = mock_workflow
-
            result = mem._build_prompt_message_with_files(
                message_files=[],
                text_content="wf text",
@@ -471,6 +566,7 @@ class TestBuildPromptMessageWithFiles:

        assert isinstance(result, UserPromptMessage)
        assert result.content == "wf text"
+        assert database.session.get(Workflow, workflow.id) is workflow

    # ------------------------------------------------------------------
    # Invalid mode
@@ -498,417 +594,140 @@ class TestBuildPromptMessageWithFiles:


 class TestGetHistoryPromptMessages:
-    """Tests for get_history_prompt_messages."""
+    """Tests for persisted history retrieval, file batching, and pruning."""

    def _make_memory(self, mode: AppMode = AppMode.CHAT) -> TokenBufferMemory:
        conv = _make_conversation(mode)
        conv.app = MagicMock()
        return TokenBufferMemory(conversation=conv, model_instance=_make_model_instance())

-    def test_returns_empty_when_no_messages(self):
+    def test_returns_empty_when_no_messages(self, database: Database) -> None:
+        assert self._make_memory().get_history_prompt_messages() == []
+
+    def test_skips_newest_message_without_answer(self, database: Database) -> None:
        mem = self._make_memory()
-        with patch("core.memory.token_buffer_memory.db") as mock_db:
-            mock_db.session.scalars.return_value.all.return_value = []
-            result = mem.get_history_prompt_messages()
-        assert result == []
+        message = _persist_message(database, mem.conversation.id, answer="", answer_tokens=0)

-    def test_skips_first_message_without_answer(self):
-        """The newest message (index 0 after extraction) without answer and tokens==0 is skipped."""
+        assert mem.get_history_prompt_messages() == []
+        assert database.session.get(Message, message.id) is message
+
+    def test_message_with_answer_returns_user_and_assistant_prompts(self, database: Database) -> None:
        mem = self._make_memory()
+        _persist_message(database, mem.conversation.id, query="My query", answer="My answer", answer_tokens=10)

-        msg_no_answer = _make_message(answer="", answer_tokens=0)
-        msg_no_answer.parent_message_id = None  # ensures extract_thread_messages returns it
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg_no_answer],
-            ),
-        ):
-            mock_db.session.scalars.return_value.all.side_effect = [
-                [msg_no_answer],  # first call: messages query
-                [],  # second call: user files query (never hit, but safe)
-            ]
-            result = mem.get_history_prompt_messages()
-
-        assert result == []
-
-    def test_message_with_answer_not_skipped(self):
-        """A message with a non-empty answer is NOT popped."""
-        mem = self._make_memory()
-
-        msg = _make_message(answer="some answer", answer_tokens=10)
-        msg.parent_message_id = None
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            # user files query → empty; assistant files query → empty
-            mock_db.session.scalars.return_value.all.return_value = []
-            result = mem.get_history_prompt_messages()
-
-        assert len(result) == 2  # one user + one assistant
-
-    def test_message_limit_default_is_500(self):
-        """When message_limit is None the stmt is limited to 500."""
-        mem = self._make_memory()
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch("core.memory.token_buffer_memory.select") as mock_select,
-            patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
-        ):
-            mock_stmt = MagicMock()
-            mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
-            mock_stmt.limit.return_value = mock_stmt
-            mock_db.session.scalars.return_value.all.return_value = []
-
-            mem.get_history_prompt_messages(message_limit=None)
-            mock_stmt.limit.assert_called_with(500)
-
-    def test_message_limit_clipped_to_500(self):
-        """A message_limit > 500 is clamped to 500."""
-        mem = self._make_memory()
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch("core.memory.token_buffer_memory.select") as mock_select,
-            patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
-        ):
-            mock_stmt = MagicMock()
-            mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
-            mock_stmt.limit.return_value = mock_stmt
-            mock_db.session.scalars.return_value.all.return_value = []
-
-            mem.get_history_prompt_messages(message_limit=9999)
-            mock_stmt.limit.assert_called_with(500)
-
-    def test_message_limit_positive_used(self):
-        """A positive message_limit < 500 is used as-is."""
-        mem = self._make_memory()
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch("core.memory.token_buffer_memory.select") as mock_select,
-            patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
-        ):
-            mock_stmt = MagicMock()
-            mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
-            mock_stmt.limit.return_value = mock_stmt
-            mock_db.session.scalars.return_value.all.return_value = []
-
-            mem.get_history_prompt_messages(message_limit=10)
-            mock_stmt.limit.assert_called_with(10)
-
-    def test_message_limit_zero_uses_default(self):
-        """message_limit=0 triggers the else branch → default 500."""
-        mem = self._make_memory()
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch("core.memory.token_buffer_memory.select") as mock_select,
-            patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=[]),
-        ):
-            mock_stmt = MagicMock()
-            mock_select.return_value.where.return_value.order_by.return_value = mock_stmt
-            mock_stmt.limit.return_value = mock_stmt
-            mock_db.session.scalars.return_value.all.return_value = []
-
-            mem.get_history_prompt_messages(message_limit=0)
-            mock_stmt.limit.assert_called_with(500)
-
-    def test_user_files_cause_build_with_files_call(self):
-        """When user_files is non-empty _build_prompt_message_with_files is invoked."""
-        mem = self._make_memory()
-        msg = _make_message()
-        msg.parent_message_id = None
-
-        mock_user_file = MagicMock()
-        mock_user_file.message_id = msg.id  # must match so batched grouping keys it to this message
-        mock_user_prompt = UserPromptMessage(content="from build")
-        mock_assistant_prompt = AssistantPromptMessage(content="answer")
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                # messages query
-                r.all.return_value = [msg]
-            elif call_count["n"] == 1:
-                # user files
-                r.all.return_value = [mock_user_file]
-            else:
-                # assistant files
-                r.all.return_value = []
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch.object(
-                mem,
-                "_build_prompt_message_with_files",
-                side_effect=[mock_user_prompt, mock_assistant_prompt],
-            ) as mock_build,
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages()
-
-        assert mock_build.call_count >= 1
-        # First call should be user message
-        first_call_kwargs = mock_build.call_args_list[0][1]
-        assert first_call_kwargs["is_user_message"] is True
-
-    def test_assistant_files_cause_build_with_files_call(self):
-        """When assistant_files is non-empty, build is called with is_user_message=False."""
-        mem = self._make_memory()
-        msg = _make_message()
-        msg.parent_message_id = None
-
-        mock_assistant_file = MagicMock()
-        mock_assistant_file.message_id = msg.id  # must match so batched grouping keys it to this message
-        mock_user_prompt = UserPromptMessage(content="query")
-        mock_assistant_prompt = AssistantPromptMessage(content="built")
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                r.all.return_value = [msg]
-            elif call_count["n"] == 1:
-                r.all.return_value = []  # no user files
-            else:
-                r.all.return_value = [mock_assistant_file]
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch.object(
-                mem,
-                "_build_prompt_message_with_files",
-                return_value=mock_assistant_prompt,
-            ) as mock_build,
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages()
-
-        mock_build.assert_called_once()
-        call_kwargs = mock_build.call_args[1]
-        assert call_kwargs["is_user_message"] is False
-
-    def test_message_files_loaded_with_constant_query_count(self):
-        """Regression guard against N+1: message files must be batch-loaded.
-
-        Regardless of the number of messages in the thread, file loading must use a
-        constant number of queries (1 messages query + 2 batched file queries),
-        never 2 queries per message.
-        """
-        mem = self._make_memory()
-
-        messages = [_make_message() for _ in range(5)]
-        for m in messages:
-            m.parent_message_id = None
-
-        scalars_calls = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            # First call returns the thread messages; the batched file queries return none.
-            r.all.return_value = messages if scalars_calls["n"] == 0 else []
-            scalars_calls["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=messages),
-            patch("core.memory.token_buffer_memory.FileUploadConfigManager.convert", return_value=None),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            mem.get_history_prompt_messages()
-
-        # 1 (messages) + 2 (batched user/assistant files) = 3, independent of message count.
-        # Before this fix it would have been 1 + 2 * 5 = 11 (an N+1 pattern).
-        assert scalars_calls["n"] == 3
-
-    def test_token_pruning_removes_oldest_messages(self):
-        """If tokens exceed limit, oldest messages are removed until within limit."""
-        conv = _make_conversation()
-        conv.app = MagicMock()
-
-        # Model returns tokens that decrease only after removing pairs
-        token_values = [3000, 1500]  # first call over limit, second within
-        mi = MagicMock()
-        mi.get_llm_num_tokens.side_effect = token_values
-
-        mem = TokenBufferMemory(conversation=conv, model_instance=mi)
-
-        msg = _make_message()
-        msg.parent_message_id = None
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                r.all.return_value = [msg]
-            else:
-                r.all.return_value = []
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages(max_token_limit=2000)
-
-        # After pruning, we should have fewer than the 2 initial messages
-        assert len(result) <= 1
-
-    def test_token_pruning_stops_at_single_message(self):
-        """Pruning stops when only 1 message remains (to prevent empty list)."""
-        conv = _make_conversation()
-        conv.app = MagicMock()
-
-        # Always over limit
-        mi = MagicMock()
-        mi.get_llm_num_tokens.return_value = 99999
-
-        mem = TokenBufferMemory(conversation=conv, model_instance=mi)
-
-        msg = _make_message()
-        msg.parent_message_id = None
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                r.all.return_value = [msg]
-            else:
-                r.all.return_value = []
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages(max_token_limit=1)
-
-        # At least 1 message should remain
-        assert len(result) >= 1
-
-    def test_no_pruning_when_within_limit(self):
-        """When tokens ≤ limit, no pruning occurs."""
-        mem = self._make_memory()
-        mem.model_instance.get_llm_num_tokens.return_value = 50  # well under default 2000
-
-        msg = _make_message()
-        msg.parent_message_id = None
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                r.all.return_value = [msg]
-            else:
-                r.all.return_value = []
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages(max_token_limit=2000)
-
-        assert len(result) == 2  # user + assistant
-
-    def test_plain_user_and_assistant_messages_returned(self):
-        """Without files, plain UserPromptMessage and AssistantPromptMessage appear."""
-        mem = self._make_memory()
-
-        msg = _make_message(answer="My answer")
-        msg.query = "My query"
-        msg.parent_message_id = None
-
-        call_count = {"n": 0}
-
-        def scalars_side_effect(stmt):
-            r = MagicMock()
-            if call_count["n"] == 0:
-                r.all.return_value = [msg]
-            else:
-                r.all.return_value = []
-            call_count["n"] += 1
-            return r
-
-        with (
-            patch("core.memory.token_buffer_memory.db") as mock_db,
-            patch(
-                "core.memory.token_buffer_memory.extract_thread_messages",
-                return_value=[msg],
-            ),
-            patch(
-                "core.memory.token_buffer_memory.FileUploadConfigManager.convert",
-                return_value=None,
-            ),
-        ):
-            mock_db.session.scalars.side_effect = scalars_side_effect
-            result = mem.get_history_prompt_messages()
+        result = mem.get_history_prompt_messages()

        assert len(result) == 2
-        user_msg, ai_msg = result
-        assert isinstance(user_msg, UserPromptMessage)
-        assert user_msg.content == "My query"
-        assert isinstance(ai_msg, AssistantPromptMessage)
-        assert ai_msg.content == "My answer"
+        assert isinstance(result[0], UserPromptMessage)
+        assert result[0].content == "My query"
+        assert isinstance(result[1], AssistantPromptMessage)
+        assert result[1].content == "My answer"
+
+    def test_history_is_conversation_scoped(self, database: Database) -> None:
+        mem = self._make_memory()
+        _persist_message(database, mem.conversation.id, answer="visible")
+        _persist_message(database, "other-conversation", answer="hidden")
+
+        result = mem.get_history_prompt_messages()
+
+        assert [prompt.content for prompt in result] == ["user query", "visible"]
+
+    @pytest.mark.parametrize(
+        ("message_limit", "expected_limit"),
+        [(None, 500), (9999, 500), (10, 10), (0, 500)],
+    )
+    def test_message_limit_is_applied_to_executable_query(
+        self,
+        database: Database,
+        message_limit: int | None,
+        expected_limit: int,
+    ) -> None:
+        mem = self._make_memory()
+        before = len(database.statements)
+
+        mem.get_history_prompt_messages(message_limit=message_limit)
+
+        statements = database.statements[before:]
+        assert len(statements) == 1
+        sql, parameters = statements[0]
+        assert "LIMIT" in sql
+        assert expected_limit in parameters
+
+    @pytest.mark.parametrize(
+        ("belongs_to", "is_user_message"),
+        [
+            (MessageFileBelongsTo.USER, True),
+            (None, True),
+            (MessageFileBelongsTo.ASSISTANT, False),
+        ],
+    )
+    def test_message_files_use_persisted_ownership(
+        self,
+        database: Database,
+        belongs_to: MessageFileBelongsTo | None,
+        is_user_message: bool,
+    ) -> None:
+        mem = self._make_memory()
+        message = _persist_message(database, mem.conversation.id)
+        message_file = _persist_message_file(database, message, belongs_to=belongs_to)
+        built_prompt = (
+            UserPromptMessage(content="built user")
+            if is_user_message
+            else AssistantPromptMessage(content="built assistant")
+        )
+
+        with patch.object(mem, "_build_prompt_message_with_files", return_value=built_prompt) as build_prompt:
+            result = mem.get_history_prompt_messages()
+
+        build_prompt.assert_called_once()
+        assert build_prompt.call_args.kwargs["message_files"] == [message_file]
+        assert build_prompt.call_args.kwargs["is_user_message"] is is_user_message
+        assert built_prompt in result
+
+    def test_message_files_are_batch_loaded_with_constant_query_count(self, database: Database) -> None:
+        mem = self._make_memory()
+        base_time = datetime.now(UTC).replace(tzinfo=None)
+        messages = [
+            _persist_message(
+                database,
+                mem.conversation.id,
+                query=f"query-{index}",
+                answer=f"answer-{index}",
+                created_at=base_time + timedelta(seconds=index),
+            )
+            for index in range(5)
+        ]
+        before = len(database.statements)
+
+        with patch("core.memory.token_buffer_memory.extract_thread_messages", return_value=messages):
+            result = mem.get_history_prompt_messages()
+
+        selects = [sql for sql, _ in database.statements[before:] if sql.lstrip().upper().startswith("SELECT")]
+        assert len(selects) == 3
+        assert len(result) == 10
+
+    @pytest.mark.parametrize(
+        ("token_values", "max_token_limit", "expected_length"),
+        [
+            ([3000, 1500], 2000, 1),
+            ([99999, 99999], 1, 1),
+            ([50], 2000, 2),
+        ],
+    )
+    def test_token_pruning_uses_persisted_history(
+        self,
+        database: Database,
+        token_values: list[int],
+        max_token_limit: int,
+        expected_length: int,
+    ) -> None:
+        mem = self._make_memory()
+        mem.model_instance.get_llm_num_tokens.side_effect = token_values
+        _persist_message(database, mem.conversation.id)
+
+        result = mem.get_history_prompt_messages(max_token_limit=max_token_limit)
+
+        assert len(result) == expected_length


 # ===========================================================================
@@ -2,13 +2,24 @@ from unittest.mock import MagicMock, patch

 import pytest
 from pydantic import ValidationError
+from sqlalchemy.orm import Session

+import core.moderation.api.api as moderation_module
 from core.extension.api_based_extension_requestor import APIBasedExtensionPoint
 from core.moderation.api.api import ApiModeration, ModerationInputParams, ModerationOutputParams
 from core.moderation.base import ModerationAction, ModerationInputsResult, ModerationOutputsResult
 from models.api_based_extension import APIBasedExtension


+class _DatabaseBinding:
+    """Expose the real SQLite session used by extension lookup."""
+
+    session: Session
+
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+
 class TestApiModeration:
    @pytest.fixture
    def api_config(self):
@@ -165,17 +176,27 @@ class TestApiModeration:
        with pytest.raises(ValueError, match="API-based Extension not found"):
            api_moderation._get_config_by_requestor(APIBasedExtensionPoint.APP_MODERATION_INPUT, {})

-    @patch("core.moderation.api.api.db.session.scalar")
-    def test_get_api_based_extension(self, mock_scalar):
-        mock_ext = MagicMock(spec=APIBasedExtension)
-        mock_scalar.return_value = mock_ext
+    @pytest.mark.parametrize("sqlite_session", [(APIBasedExtension,)], indirect=True)
+    def test_get_api_based_extension(self, sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+        target = APIBasedExtension(
+            tenant_id="tenant-1",
+            name="Target extension",
+            api_endpoint="https://example.com/moderate",
+            api_key="encrypted-key",
+        )
+        target.id = "ext-1"
+        other_tenant = APIBasedExtension(
+            tenant_id="tenant-2",
+            name="Other extension",
+            api_endpoint="https://example.com/other",
+            api_key="other-key",
+        )
+        other_tenant.id = "ext-2"
+        sqlite_session.add_all((target, other_tenant))
+        sqlite_session.commit()
+        monkeypatch.setattr(moderation_module, "db", _DatabaseBinding(sqlite_session))

        result = ApiModeration._get_api_based_extension("tenant-1", "ext-1")

-        assert result == mock_ext
-        mock_scalar.assert_called_once()
-        # Verify the call has the correct filters
-        args, kwargs = mock_scalar.call_args
-        stmt = args[0]
-        # We can't easily inspect the statement without complex sqlalchemy tricks,
-        # but calling it is usually enough for unit tests if we mock the result.
+        assert result is target
+        assert ApiModeration._get_api_based_extension("tenant-1", "ext-2") is None
@@ -7,36 +7,212 @@ Covers:
 - TraceTask._get_user_id_from_metadata
 """

-from unittest.mock import MagicMock, patch
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+from unittest.mock import PropertyMock, patch

 import pytest
+from sqlalchemy import Engine, event
+from sqlalchemy.orm import Session
+
+from core.tools.entities.tool_entities import ApiProviderSchemaType
+from extensions.ext_database import db
+from graphon.model_runtime.entities.model_entities import ModelType
+from models.account import Tenant
+from models.base import TypeBase
+from models.model import App, AppMode, IconType
+from models.provider import Provider, ProviderCredential, ProviderModel, ProviderModelCredential, ProviderType
+from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider

 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------


-def _make_db_and_session_patches(scalar_side_effect=None, scalar_return_value=None):
-    """Return (mock_db, cm, session) ready to patch 'core.ops.ops_trace_manager.db'
-    and 'core.ops.ops_trace_manager.Session'.
+@pytest.fixture
+def orm_session(sqlite_engine: Engine) -> Iterator[Session]:
+    models = (
+        App,
+        Tenant,
+        Provider,
+        ProviderCredential,
+        ProviderModel,
+        ProviderModelCredential,
+        BuiltinToolProvider,
+        ApiToolProvider,
+        WorkflowToolProvider,
+        MCPToolProvider,
+    )
+    tables = [model.metadata.tables[model.__tablename__] for model in models]
+    TypeBase.metadata.create_all(sqlite_engine, tables=tables)

-    Provide either scalar_side_effect (list, for multiple calls) or
-    scalar_return_value (single value).
-    """
-    mock_db = MagicMock()
-    mock_db.engine = MagicMock()
+    with patch.object(type(db), "engine", new_callable=PropertyMock, return_value=sqlite_engine):
+        with Session(sqlite_engine, expire_on_commit=False) as session:
+            yield session

-    session = MagicMock()
-    if scalar_side_effect is not None:
-        session.scalar.side_effect = scalar_side_effect
+
+def _persist_app(session: Session, *, tenant_id: str, name: str = "MyApp") -> App:
+    app = App(
+        id=str(uuid.uuid4()),
+        tenant_id=tenant_id,
+        name=name,
+        mode=AppMode.WORKFLOW,
+        icon_type=IconType.EMOJI,
+        icon="workflow",
+        icon_background="#FFFFFF",
+        enable_site=True,
+        enable_api=False,
+    )
+    session.add(app)
+    session.commit()
+    return app
+
+
+def _persist_tenant(session: Session, *, name: str = "MyWorkspace") -> Tenant:
+    tenant = Tenant(name=name)
+    session.add(tenant)
+    session.commit()
+    return tenant
+
+
+def _persist_tool_provider(
+    session: Session, provider_type: str
+) -> BuiltinToolProvider | ApiToolProvider | WorkflowToolProvider | MCPToolProvider:
+    tenant_id = str(uuid.uuid4())
+    user_id = str(uuid.uuid4())
+    if provider_type in {"builtin", "plugin"}:
+        provider = BuiltinToolProvider(
+            name="CredentialA",
+            tenant_id=tenant_id,
+            user_id=user_id,
+            provider="test/provider",
+        )
+    elif provider_type == "api":
+        provider = ApiToolProvider(
+            name="CredentialA",
+            icon="icon.svg",
+            schema="{}",
+            schema_type_str=ApiProviderSchemaType.OPENAPI,
+            user_id=user_id,
+            tenant_id=tenant_id,
+            description="API provider",
+            tools_str="[]",
+            credentials_str="{}",
+        )
+    elif provider_type == "workflow":
+        provider = WorkflowToolProvider(
+            name="CredentialA",
+            label="CredentialA",
+            icon="icon.svg",
+            app_id=str(uuid.uuid4()),
+            version="1",
+            user_id=user_id,
+            tenant_id=tenant_id,
+            description="Workflow provider",
+        )
+    elif provider_type == "mcp":
+        provider = MCPToolProvider(
+            name="CredentialA",
+            server_identifier="credential-a",
+            server_url="https://example.com/mcp",
+            server_url_hash="credential-a-hash",
+            icon="icon.svg",
+            tenant_id=tenant_id,
+            user_id=user_id,
+        )
    else:
-        session.scalar.return_value = scalar_return_value
+        raise ValueError(f"unsupported provider type: {provider_type}")

-    cm = MagicMock()
-    cm.__enter__ = MagicMock(return_value=session)
-    cm.__exit__ = MagicMock(return_value=False)
+    session.add(provider)
+    session.commit()
+    return provider

-    return mock_db, cm, session
+
+def _persist_provider_credential(
+    session: Session,
+    *,
+    tenant_id: str,
+    credential_name: str = "ProvCredName",
+) -> ProviderCredential:
+    credential = ProviderCredential(
+        tenant_id=tenant_id,
+        provider_name="openai",
+        credential_name=credential_name,
+        encrypted_config="{}",
+    )
+    session.add(credential)
+    session.commit()
+    return credential
+
+
+def _persist_model_credential(
+    session: Session,
+    *,
+    tenant_id: str,
+    credential_name: str = "ModelCredName",
+) -> ProviderModelCredential:
+    credential = ProviderModelCredential(
+        tenant_id=tenant_id,
+        provider_name="openai",
+        model_name="gpt-4",
+        model_type=ModelType.LLM,
+        credential_name=credential_name,
+        encrypted_config="{}",
+    )
+    session.add(credential)
+    session.commit()
+    return credential
+
+
+def _persist_provider(
+    session: Session,
+    *,
+    tenant_id: str,
+    credential_id: str | None,
+) -> Provider:
+    provider = Provider(
+        tenant_id=tenant_id,
+        provider_name="openai",
+        provider_type=ProviderType.CUSTOM,
+        credential_id=credential_id,
+    )
+    session.add(provider)
+    session.commit()
+    return provider
+
+
+def _persist_provider_model(
+    session: Session,
+    *,
+    tenant_id: str,
+    credential_id: str | None,
+) -> ProviderModel:
+    model = ProviderModel(
+        tenant_id=tenant_id,
+        provider_name="openai",
+        model_name="gpt-4",
+        model_type=ModelType.LLM,
+        credential_id=credential_id,
+    )
+    session.add(model)
+    session.commit()
+    return model
+
+
+@contextmanager
+def _raise_on_table(engine: Engine, table_name: str) -> Iterator[None]:
+    """Raise only when SQL targets the named table, leaving other real lookups intact."""
+
+    def fail_target_query(_conn, _cursor, statement, _parameters, _context, _executemany):
+        if f"FROM {table_name}" in statement:
+            raise RuntimeError(f"forced failure for {table_name}")
+
+    event.listen(engine, "before_cursor_execute", fail_target_query)
+    try:
+        yield
+    finally:
+        event.remove(engine, "before_cursor_execute", fail_target_query)


 # ---------------------------------------------------------------------------
@@ -47,62 +223,42 @@ def _make_db_and_session_patches(scalar_side_effect=None, scalar_return_value=No
 class TestLookupAppAndWorkspaceNames:
    """Tests for _lookup_app_and_workspace_names(app_id, tenant_id)."""

-    def test_both_found(self):
+    def test_both_found(self, orm_session: Session):
        """Returns (app_name, workspace_name) when both records exist."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=["MyApp", "MyWorkspace"])
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
+        tenant = _persist_tenant(orm_session)
+        app = _persist_app(orm_session, tenant_id=tenant.id)
+        app_name, workspace_name = _lookup_app_and_workspace_names(app.id, tenant.id)

        assert app_name == "MyApp"
        assert workspace_name == "MyWorkspace"

-    def test_app_only_found(self):
+    def test_app_only_found(self, orm_session: Session):
        """Returns (app_name, '') when tenant record is absent."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=["MyApp", None])
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
+        app = _persist_app(orm_session, tenant_id=str(uuid.uuid4()))
+        app_name, workspace_name = _lookup_app_and_workspace_names(app.id, str(uuid.uuid4()))

        assert app_name == "MyApp"
        assert workspace_name == ""

-    def test_tenant_only_found(self):
+    def test_tenant_only_found(self, orm_session: Session):
        """Returns ('', workspace_name) when app record is absent."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[None, "MyWorkspace"])
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
+        tenant = _persist_tenant(orm_session)
+        app_name, workspace_name = _lookup_app_and_workspace_names(str(uuid.uuid4()), tenant.id)

        assert app_name == ""
        assert workspace_name == "MyWorkspace"

-    def test_neither_found(self):
+    def test_neither_found(self, orm_session: Session):
        """Returns ('', '') when both DB lookups return None."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[None, None])
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names("app-123", "tenant-456")
+        app_name, workspace_name = _lookup_app_and_workspace_names(str(uuid.uuid4()), str(uuid.uuid4()))

        assert app_name == ""
        assert workspace_name == ""
@@ -111,50 +267,30 @@ class TestLookupAppAndWorkspaceNames:
        """Returns ('', '') immediately when both IDs are None — no DB access."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        app_name, workspace_name = _lookup_app_and_workspace_names(None, None)

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names(None, None)
-
-        mock_session_cls.assert_not_called()
        assert app_name == ""
        assert workspace_name == ""

-    def test_app_id_none_only_queries_tenant(self):
+    def test_app_id_none_only_queries_tenant(self, orm_session: Session):
        """When app_id is None, only the tenant query is issued."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="OnlyWorkspace")
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names(None, "tenant-456")
+        tenant = _persist_tenant(orm_session, name="OnlyWorkspace")
+        app_name, workspace_name = _lookup_app_and_workspace_names(None, tenant.id)

        assert app_name == ""
        assert workspace_name == "OnlyWorkspace"
-        assert session.scalar.call_count == 1

-    def test_tenant_id_none_only_queries_app(self):
+    def test_tenant_id_none_only_queries_app(self, orm_session: Session):
        """When tenant_id is None, only the app query is issued."""
        from core.ops.ops_trace_manager import _lookup_app_and_workspace_names

-        mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="OnlyApp")
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            app_name, workspace_name = _lookup_app_and_workspace_names("app-123", None)
+        app = _persist_app(orm_session, tenant_id=str(uuid.uuid4()), name="OnlyApp")
+        app_name, workspace_name = _lookup_app_and_workspace_names(app.id, None)

        assert app_name == "OnlyApp"
        assert workspace_name == ""
-        assert session.scalar.call_count == 1


 # ---------------------------------------------------------------------------
@@ -166,32 +302,20 @@ class TestLookupCredentialName:
    """Tests for _lookup_credential_name(credential_id, provider_type)."""

    @pytest.mark.parametrize("provider_type", ["builtin", "plugin", "api", "workflow", "mcp"])
-    def test_known_provider_types_return_name(self, provider_type):
+    def test_known_provider_types_return_name(self, provider_type: str, orm_session: Session):
        """Each valid provider_type results in a DB query and returns the credential name."""
        from core.ops.ops_trace_manager import _lookup_credential_name

-        mock_db, cm, session = _make_db_and_session_patches(scalar_return_value="CredentialA")
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            result = _lookup_credential_name("cred-123", provider_type)
+        provider = _persist_tool_provider(orm_session, provider_type)
+        result = _lookup_credential_name(provider.id, provider_type)

        assert result == "CredentialA"
-        session.scalar.assert_called_once()

-    def test_credential_not_found_returns_empty_string(self):
+    def test_credential_not_found_returns_empty_string(self, orm_session: Session):
        """Returns '' when DB yields None for the given credential_id."""
        from core.ops.ops_trace_manager import _lookup_credential_name

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_return_value=None)
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            result = _lookup_credential_name("cred-999", "api")
+        result = _lookup_credential_name(str(uuid.uuid4()), "api")

        assert result == ""

@@ -199,48 +323,24 @@ class TestLookupCredentialName:
        """Returns '' immediately for an unrecognised provider_type — no DB access."""
        from core.ops.ops_trace_manager import _lookup_credential_name

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        result = _lookup_credential_name(str(uuid.uuid4()), "unknown_type")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            result = _lookup_credential_name("cred-123", "unknown_type")
-
-        mock_session_cls.assert_not_called()
        assert result == ""

    def test_none_credential_id_returns_empty_string_without_db(self):
        """Returns '' immediately when credential_id is None — no DB access."""
        from core.ops.ops_trace_manager import _lookup_credential_name

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        result = _lookup_credential_name(None, "api")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            result = _lookup_credential_name(None, "api")
-
-        mock_session_cls.assert_not_called()
        assert result == ""

    def test_none_provider_type_returns_empty_string_without_db(self):
        """Returns '' immediately when provider_type is None — no DB access."""
        from core.ops.ops_trace_manager import _lookup_credential_name

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        result = _lookup_credential_name(str(uuid.uuid4()), None)

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            result = _lookup_credential_name("cred-123", None)
-
-        mock_session_cls.assert_not_called()
        assert result == ""

    def test_builtin_and_plugin_map_to_same_model(self):
@@ -281,106 +381,78 @@ class TestLookupCredentialName:
 class TestLookupLlmCredentialInfo:
    """Tests for _lookup_llm_credential_info(tenant_id, provider, model, model_type)."""

-    def _provider_record(self, credential_id: str | None = None) -> MagicMock:
-        record = MagicMock()
-        record.credential_id = credential_id
-        return record
-
-    def _model_record(self, credential_id: str | None = None) -> MagicMock:
-        record = MagicMock()
-        record.credential_id = credential_id
-        return record
-
-    def test_model_level_credential_found(self):
+    def test_model_level_credential_found(self, orm_session: Session):
        """Returns model-level credential_id and name when ProviderModel has a credential."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id=None)
-        model_record = self._model_record(credential_id="model-cred-id")
+        tenant_id = str(uuid.uuid4())
+        model_credential = _persist_model_credential(orm_session, tenant_id=tenant_id)
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=None)
+        _persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=model_credential.id)

-        # scalar calls: (1) Provider, (2) ProviderModel, (3) ProviderModelCredential.credential_name
-        mock_db, cm, _session = _make_db_and_session_patches(
-            scalar_side_effect=[provider_record, model_record, "ModelCredName"]
+        decoy_tenant_id = str(uuid.uuid4())
+        decoy_credential = _persist_model_credential(
+            orm_session,
+            tenant_id=decoy_tenant_id,
+            credential_name="WrongTenantCredential",
        )
+        _persist_provider(orm_session, tenant_id=decoy_tenant_id, credential_id=None)
+        _persist_provider_model(orm_session, tenant_id=decoy_tenant_id, credential_id=decoy_credential.id)

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

-        assert cred_id == "model-cred-id"
+        assert cred_id == model_credential.id
        assert cred_name == "ModelCredName"

-    def test_provider_level_fallback_when_no_model_credential(self):
+    def test_provider_level_fallback_when_no_model_credential(self, orm_session: Session):
        """Falls back to provider-level credential when ProviderModel has no credential_id."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id="prov-cred-id")
-        model_record = self._model_record(credential_id=None)
+        tenant_id = str(uuid.uuid4())
+        provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)
+        _persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=None)

-        # scalar calls: (1) Provider, (2) ProviderModel (no cred), (3) ProviderCredential.credential_name
-        mock_db, cm, _session = _make_db_and_session_patches(
-            scalar_side_effect=[provider_record, model_record, "ProvCredName"]
-        )
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
-
-        assert cred_id == "prov-cred-id"
+        assert cred_id == provider_credential.id
        assert cred_name == "ProvCredName"

-    def test_provider_level_fallback_when_no_model_record(self):
+    def test_provider_level_fallback_when_no_model_record(self, orm_session: Session):
        """Falls back to provider-level credential when no ProviderModel row exists."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id="prov-cred-id")
+        tenant_id = str(uuid.uuid4())
+        provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)

-        # scalar calls: (1) Provider, (2) ProviderModel → None, (3) ProviderCredential.credential_name
-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[provider_record, None, "ProvCredName"])
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
-
-        assert cred_id == "prov-cred-id"
+        assert cred_id == provider_credential.id
        assert cred_name == "ProvCredName"

-    def test_no_model_arg_uses_provider_level_only(self):
+    def test_no_model_arg_uses_provider_level_only(self, orm_session: Session):
        """When model is None, skips ProviderModel query and uses provider credential."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id="prov-cred-id")
+        tenant_id = str(uuid.uuid4())
+        provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)

-        # scalar calls: (1) Provider, (2) ProviderCredential.credential_name — no ProviderModel
-        mock_db, cm, session = _make_db_and_session_patches(scalar_side_effect=[provider_record, "ProvCredName"])
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", None)

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", None)
-
-        assert cred_id == "prov-cred-id"
+        assert cred_id == provider_credential.id
        assert cred_name == "ProvCredName"
-        assert session.scalar.call_count == 2

-    def test_provider_not_found_returns_none_and_empty(self):
+    def test_provider_not_found_returns_none_and_empty(self, orm_session: Session):
        """Returns (None, '') when Provider record does not exist."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_return_value=None)
+        other_tenant_id = str(uuid.uuid4())
+        _persist_provider(orm_session, tenant_id=other_tenant_id, credential_id=None)
+        tenant_id = str(uuid.uuid4())

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

        assert cred_id is None
        assert cred_name == ""
@@ -389,16 +461,8 @@ class TestLookupLlmCredentialInfo:
        """Returns (None, '') immediately when tenant_id is None — no DB access."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        cred_id, cred_name = _lookup_llm_credential_info(None, "openai", "gpt-4")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info(None, "openai", "gpt-4")
-
-        mock_session_cls.assert_not_called()
        assert cred_id is None
        assert cred_name == ""

@@ -406,69 +470,46 @@ class TestLookupLlmCredentialInfo:
        """Returns (None, '') immediately when provider is None — no DB access."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        mock_db = MagicMock()
-        mock_session_cls = MagicMock()
+        cred_id, cred_name = _lookup_llm_credential_info(str(uuid.uuid4()), None, "gpt-4")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", mock_session_cls),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", None, "gpt-4")
-
-        mock_session_cls.assert_not_called()
        assert cred_id is None
        assert cred_name == ""

-    def test_db_error_on_outer_query_returns_none_and_empty(self):
+    def test_db_error_on_outer_query_returns_none_and_empty(self, orm_session: Session, sqlite_engine: Engine):
        """Returns (None, '') and logs a warning when the outer DB query raises."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        mock_db, cm, session = _make_db_and_session_patches()
-        session.scalar.side_effect = Exception("DB connection failed")
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
+        with _raise_on_table(sqlite_engine, "providers"):
+            cred_id, cred_name = _lookup_llm_credential_info(str(uuid.uuid4()), "openai", "gpt-4")

        assert cred_id is None
        assert cred_name == ""

-    def test_credential_name_lookup_failure_returns_id_with_empty_name(self):
+    def test_credential_name_lookup_failure_returns_id_with_empty_name(
+        self, orm_session: Session, sqlite_engine: Engine
+    ):
        """When credential name sub-query fails, returns cred_id but '' for name."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id="prov-cred-id")
+        tenant_id = str(uuid.uuid4())
+        provider_credential = _persist_provider_credential(orm_session, tenant_id=tenant_id)
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=provider_credential.id)

-        # Provider found, no model record, then name lookup raises
-        mock_db, cm, _session = _make_db_and_session_patches(
-            scalar_side_effect=[provider_record, None, Exception("deleted")]
-        )
+        with _raise_on_table(sqlite_engine, "provider_credentials"):
+            cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
-
-        assert cred_id == "prov-cred-id"
+        assert cred_id == provider_credential.id
        assert cred_name == ""

-    def test_no_credential_on_provider_or_model_returns_none_id(self):
+    def test_no_credential_on_provider_or_model_returns_none_id(self, orm_session: Session):
        """Returns (None, '') when neither provider nor model has a credential_id."""
        from core.ops.ops_trace_manager import _lookup_llm_credential_info

-        provider_record = self._provider_record(credential_id=None)
-        model_record = self._model_record(credential_id=None)
+        tenant_id = str(uuid.uuid4())
+        _persist_provider(orm_session, tenant_id=tenant_id, credential_id=None)
+        _persist_provider_model(orm_session, tenant_id=tenant_id, credential_id=None)

-        mock_db, cm, _session = _make_db_and_session_patches(scalar_side_effect=[provider_record, model_record])
-
-        with (
-            patch("core.ops.ops_trace_manager.db", mock_db),
-            patch("core.ops.ops_trace_manager.Session", return_value=cm),
-        ):
-            cred_id, cred_name = _lookup_llm_credential_info("tenant-1", "openai", "gpt-4")
+        cred_id, cred_name = _lookup_llm_credential_info(tenant_id, "openai", "gpt-4")

        assert cred_id is None
        assert cred_name == ""
@@ -1,9 +1,11 @@
 import re
 from datetime import datetime
-from unittest.mock import MagicMock, patch
+from decimal import Decimal

 import pytest
+from sqlalchemy.orm import Session

+import core.ops.utils as utils_module
 from core.ops.utils import (
    filter_none_values,
    generate_dotted_order,
@@ -15,6 +17,42 @@ from core.ops.utils import (
    validate_url,
    validate_url_with_path,
 )
+from models.enums import ConversationFromSource
+from models.model import Message
+
+
+class _DatabaseBinding:
+    """Expose the real SQLite session used by the message lookup helper."""
+
+    session: Session
+
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+
+@pytest.fixture
+def message_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
+    """Bind the message lookup helper to the shared SQLite test session."""
+
+    monkeypatch.setattr(utils_module, "db", _DatabaseBinding(sqlite_session))
+    return sqlite_session
+
+
+def _message(message_id: str) -> Message:
+    message = Message(
+        id=message_id,
+        app_id="app-id",
+        conversation_id="conversation-id",
+        query="question",
+        message={"role": "user", "content": "question"},
+        answer="answer",
+        message_unit_price=Decimal("0.0001"),
+        answer_unit_price=Decimal("0.0001"),
+        currency="USD",
+        from_source=ConversationFromSource.API,
+    )
+    message._inputs = {}
+    return message


 class TestValidateUrl:
@@ -220,22 +258,20 @@ class TestFilterNoneValues:
        assert filter_none_values({}) == {}


+@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
 class TestGetMessageData:
    """Test cases for get_message_data function"""

-    @patch("core.ops.utils.db")
-    @patch("core.ops.utils.Message")
-    @patch("core.ops.utils.select")
-    def test_get_message_data(self, mock_select, mock_message, mock_db):
-        mock_scalar = mock_db.session.scalar
-        mock_msg_instance = MagicMock()
-        mock_scalar.return_value = mock_msg_instance
+    def test_get_message_data(self, message_session: Session):
+        target = _message("message-id")
+        unrelated = _message("other-message-id")
+        message_session.add_all((target, unrelated))
+        message_session.commit()

        result = get_message_data("message-id")

-        assert result == mock_msg_instance
-        mock_select.assert_called_once()
-        mock_scalar.assert_called_once()
+        assert result is target
+        assert result.id == "message-id"


 class TestMeasureTime:
@@ -1,9 +1,42 @@
-from unittest.mock import MagicMock
+from datetime import datetime, timedelta
+from decimal import Decimal
 from uuid import uuid4

+import pytest
+from sqlalchemy.orm import Session
+
 from constants import UUID_NIL
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
+from models.enums import ConversationFromSource
+from models.model import Message
+
+
+def _persisted_message(
+    *,
+    message_id: str,
+    conversation_id: str,
+    parent_message_id: str,
+    answer: str,
+    created_at: datetime,
+) -> Message:
+    message = Message(
+        id=message_id,
+        app_id="app-id",
+        conversation_id=conversation_id,
+        query="question",
+        message={"role": "user", "content": "question"},
+        answer=answer,
+        message_unit_price=Decimal("0.0001"),
+        answer_unit_price=Decimal("0.0001"),
+        currency="USD",
+        from_source=ConversationFromSource.API,
+        parent_message_id=parent_message_id,
+        created_at=created_at,
+        updated_at=created_at,
+    )
+    message._inputs = {}
+    return message


 class MockMessage:
@@ -104,33 +137,64 @@ def test_extract_thread_messages_breaks_when_parent_is_none():
    assert result[0].id == id2


-def test_get_thread_messages_length_excludes_newly_created_empty_answer():
+@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
+def test_get_thread_messages_length_excludes_newly_created_empty_answer(sqlite_session: Session):
    id1, id2 = str(uuid4()), str(uuid4())
+    now = datetime.now()
    messages = [
-        MockMessage(id2, id1, answer=""),  # newest generated message should be excluded
-        MockMessage(id1, UUID_NIL, answer="ok"),
+        _persisted_message(
+            message_id=id2,
+            conversation_id="conversation-1",
+            parent_message_id=id1,
+            answer="",
+            created_at=now,
+        ),
+        _persisted_message(
+            message_id=id1,
+            conversation_id="conversation-1",
+            parent_message_id=UUID_NIL,
+            answer="ok",
+            created_at=now - timedelta(seconds=1),
+        ),
+        _persisted_message(
+            message_id=str(uuid4()),
+            conversation_id="other-conversation",
+            parent_message_id=UUID_NIL,
+            answer="unrelated",
+            created_at=now + timedelta(seconds=1),
+        ),
    ]
+    sqlite_session.add_all(messages)
+    sqlite_session.commit()

-    session = MagicMock()
-    session.scalars.return_value.all.return_value = messages
-
-    length = get_thread_messages_length("conversation-1", session=session)
+    length = get_thread_messages_length("conversation-1", session=sqlite_session)

    assert length == 1
-    session.scalars.assert_called_once()


-def test_get_thread_messages_length_keeps_non_empty_latest_answer():
+@pytest.mark.parametrize("sqlite_session", [(Message,)], indirect=True)
+def test_get_thread_messages_length_keeps_non_empty_latest_answer(sqlite_session: Session):
    id1, id2 = str(uuid4()), str(uuid4())
+    now = datetime.now()
    messages = [
-        MockMessage(id2, id1, answer="latest-answer"),
-        MockMessage(id1, UUID_NIL, answer="older-answer"),
+        _persisted_message(
+            message_id=id2,
+            conversation_id="conversation-2",
+            parent_message_id=id1,
+            answer="latest-answer",
+            created_at=now,
+        ),
+        _persisted_message(
+            message_id=id1,
+            conversation_id="conversation-2",
+            parent_message_id=UUID_NIL,
+            answer="older-answer",
+            created_at=now - timedelta(seconds=1),
+        ),
    ]
+    sqlite_session.add_all(messages)
+    sqlite_session.commit()

-    session = MagicMock()
-    session.scalars.return_value.all.return_value = messages
-
-    length = get_thread_messages_length("conversation-2", session=session)
+    length = get_thread_messages_length("conversation-2", session=sqlite_session)

    assert length == 2
-    session.scalars.assert_called_once()
@@ -22,6 +22,23 @@ class TestCleanProcessor:
        expected = "normalpadding"
        assert CleanProcessor.clean(text_with_ufffe, None) == expected

+    def test_clean_preserves_valid_extended_characters(self):
+        """Default cleaning must not strip valid printable characters.
+
+        The invalid-symbol filter used to include the UTF-8 bytes of U+FFFE
+        (0xEF 0xBF 0xBE) inside a character class. On a decoded string those
+        bytes are the code points U+00EF, U+00BF and U+00BE, i.e. the valid
+        characters 'ï', '¿' and '¾', so words like "naïve" and Spanish
+        questions like "¿Cómo?" were being silently corrupted on ingest.
+        """
+        assert CleanProcessor.clean("naïve", None) == "naïve"
+        assert CleanProcessor.clean("¿Cómo estás?", None) == "¿Cómo estás?"
+        assert CleanProcessor.clean("¾ cup sugar", None) == "¾ cup sugar"
+        assert CleanProcessor.clean("ï¿¾", None) == "ï¿¾"
+
+        # The U+FFFE noncharacter is still stripped by its dedicated substitution.
+        assert CleanProcessor.clean("keep\ufffedrop", None) == "keepdrop"
+
    def test_clean_with_none_process_rule(self):
        """Test cleaning with None process_rule - only default cleaning applied."""
        text = "Hello<|World\x00"
@@ -1372,6 +1372,19 @@ class TestIndexingRunnerDocumentCleaning:
        assert "\ufffe" not in result
        assert "Text with" in result

+    def test_filter_string_preserves_valid_extended_characters(self):
+        """filter_string must keep valid printable characters like 'ï', '¿', '¾'."""
+        # Arrange
+        text = "naïve ¿Cómo? ¾ done"
+
+        # Act
+        result = IndexingRunner.filter_string(text)
+
+        # Assert
+        assert result == text
+        # The U+FFFE noncharacter is still stripped.
+        assert IndexingRunner.filter_string("keep\ufffedrop") == "keepdrop"
+

 class TestIndexingRunnerSplitter:
    """Unit tests for text splitter configuration.
@@ -952,6 +952,21 @@ class TestFixedRecursiveCharacterTextSplitter:
        assert "word1" in combined
        assert "word2" in combined

+    def test_preserves_spaces_when_recursively_splitting_long_paragraph(self):
+        """Ensure recursive space splitting preserves word boundaries."""
+        text = "여름철에는 항상 기상상황에 주목하며 주변 사람들과 함께 정보를 공유합니다."
+        splitter = FixedRecursiveCharacterTextSplitter(
+            fixed_separator="\n\n",
+            chunk_size=20,
+            chunk_overlap=0,
+            keep_separator=True,
+        )
+
+        result = splitter.split_text(text)
+
+        assert len(result) > 1
+        assert " ".join(result) == text
+
    def test_character_level_splitting(self):
        """Test character-level splitting when no separator works."""
        text = "verylongwordwithoutspaces"
@@ -1,16 +1,16 @@
-"""Unit tests for workflow-as-tool behavior.
-
-StubSession/StubScalars emulate SQLAlchemy session/scalars with minimal methods
-(`scalar`, `scalars`, `expunge`, `commit`, `refresh`, context manager) to keep
-database access mocked and predictable in tests.
-"""
+"""Unit tests for workflow-as-tool behavior with real SQLite ORM boundaries."""

 import json
+import uuid
+from collections.abc import Iterator
+from dataclasses import dataclass
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock, Mock, patch

 import pytest
+from sqlalchemy import Engine, inspect
+from sqlalchemy.orm import Session, sessionmaker

 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -23,74 +23,142 @@ from core.tools.entities.tool_entities import (
    ToolProviderType,
 )
 from core.tools.errors import ToolInvokeError
+from core.tools.workflow_as_tool import tool as workflow_tool_module
 from core.tools.workflow_as_tool.tool import WorkflowTool
 from graphon.file import FILE_MODEL_IDENTITY, FileTransferMethod, FileType
+from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
+from models.base import TypeBase
+from models.enums import EndUserType
+from models.model import App, AppMode, EndUser
+from models.workflow import Workflow, WorkflowType
+
+TENANT_ID = "00000000-0000-0000-0000-000000000001"
+OTHER_TENANT_ID = "00000000-0000-0000-0000-000000000002"
+APP_ID = "00000000-0000-0000-0000-000000000003"
+ACCOUNT_ID = "00000000-0000-0000-0000-000000000004"
+END_USER_ID = "00000000-0000-0000-0000-000000000005"
+CREATOR_ID = "00000000-0000-0000-0000-000000000006"


-class StubScalars:
-    """Minimal stub for SQLAlchemy scalar results."""
-
-    _value: Any
-
-    def __init__(self, value: Any) -> None:
-        self._value = value
-
-    def first(self) -> Any:
-        return self._value
+@dataclass(frozen=True)
+class SqliteToolDb:
+    engine: Engine
+    session_maker: sessionmaker[Session]
+    caller_session: Session


-class StubSession:
-    """Minimal stub for session_factory-created sessions."""
+@pytest.fixture
+def sqlite_tool_db(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_engine: Engine,
+) -> Iterator[SqliteToolDb]:
+    """Bind service-owned sessions and Account tenant reloads to SQLite."""
+    models = (App, Workflow, EndUser, Account, Tenant, TenantAccountJoin)
+    TypeBase.metadata.create_all(sqlite_engine, tables=[model.__table__ for model in models])
+    session_maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr(workflow_tool_module.session_factory, "create_session", session_maker)

-    scalar_results: list[Any]
-    scalars_results: list[Any]
-    expunge_calls: list[object]
+    from models import account as account_module

-    def __init__(self, *, scalar_results: list[Any] | None = None, scalars_results: list[Any] | None = None) -> None:
-        self.scalar_results = list(scalar_results or [])
-        self.scalars_results = list(scalars_results or [])
-        self.expunge_calls: list[object] = []
-
-    def scalar(self, _stmt: Any) -> Any:
-        return self.scalar_results.pop(0)
-
-    def scalars(self, _stmt: Any) -> StubScalars:
-        return StubScalars(self.scalars_results.pop(0))
-
-    def expunge(self, value: Any) -> None:
-        self.expunge_calls.append(value)
-
-    def begin(self) -> "StubSession":
-        return self
-
-    def commit(self) -> None:
-        pass
-
-    def refresh(self, _value: Any) -> None:
-        pass
-
-    def close(self) -> None:
-        pass
-
-    def __enter__(self) -> "StubSession":
-        return self
-
-    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool:
-        return False
+    monkeypatch.setattr(account_module, "db", SimpleNamespace(engine=sqlite_engine))
+    with session_maker() as caller_session:
+        yield SqliteToolDb(engine=sqlite_engine, session_maker=session_maker, caller_session=caller_session)


-def _build_tool() -> WorkflowTool:
+def _persist_tenant(db: SqliteToolDb, *, tenant_id: str = TENANT_ID) -> Tenant:
+    tenant = Tenant(name="Tenant")
+    tenant.id = tenant_id
+    db.caller_session.add(tenant)
+    db.caller_session.commit()
+    return tenant
+
+
+def _persist_account(db: SqliteToolDb, *, tenant_id: str = TENANT_ID) -> Account:
+    account = Account(name="Account", email="account@example.com")
+    account.id = ACCOUNT_ID
+    join = TenantAccountJoin(
+        tenant_id=tenant_id,
+        account_id=account.id,
+        current=True,
+        role=TenantAccountRole.NORMAL,
+    )
+    db.caller_session.add_all([account, join])
+    db.caller_session.commit()
+    return account
+
+
+def _persist_end_user(
+    db: SqliteToolDb,
+    *,
+    end_user_id: str = END_USER_ID,
+    tenant_id: str = TENANT_ID,
+) -> EndUser:
+    end_user = EndUser(
+        id=end_user_id,
+        tenant_id=tenant_id,
+        app_id=APP_ID,
+        type=EndUserType.SERVICE_API,
+        name="End user",
+        session_id="end-user-session",
+    )
+    db.caller_session.add(end_user)
+    db.caller_session.commit()
+    return end_user
+
+
+def _persist_app(db: SqliteToolDb) -> App:
+    app = App(
+        id=APP_ID,
+        tenant_id=TENANT_ID,
+        name="Workflow app",
+        description="",
+        mode=AppMode.WORKFLOW,
+        icon_type=None,
+        icon="",
+        icon_background=None,
+        app_model_config_id=None,
+        workflow_id=None,
+        enable_site=False,
+        enable_api=True,
+        max_active_requests=None,
+        created_by=CREATOR_ID,
+    )
+    db.caller_session.add(app)
+    db.caller_session.commit()
+    return app
+
+
+def _persist_workflow(db: SqliteToolDb, *, version: str, workflow_id: str | None = None) -> Workflow:
+    workflow = Workflow.new(
+        tenant_id=TENANT_ID,
+        app_id=APP_ID,
+        type=WorkflowType.WORKFLOW.value,
+        version=version,
+        graph=json.dumps({"nodes": [], "edges": []}),
+        features="{}",
+        created_by=CREATOR_ID,
+        environment_variables=[],
+        conversation_variables=[],
+        rag_pipeline_variables=[],
+    )
+    workflow.id = workflow_id or str(uuid.uuid4())
+    db.caller_session.add(workflow)
+    db.caller_session.commit()
+    return workflow
+
+
+def _build_tool(*, tenant_id: str = "test_tool", workflow_app_id: str = "app-1", version: str = "1") -> WorkflowTool:
    entity = ToolEntity(
        identity=ToolIdentity(author="test", name="test tool", label=I18nObject(en_US="test tool"), provider="test"),
        parameters=[],
        description=None,
        has_runtime_parameters=False,
    )
-    runtime = ToolRuntime(tenant_id="test_tool", invoke_from=InvokeFrom.EXPLORE)
+    runtime = ToolRuntime(tenant_id=tenant_id, invoke_from=InvokeFrom.EXPLORE)
    return WorkflowTool(
-        workflow_app_id="app-1",
+        workflow_app_id=workflow_app_id,
        workflow_as_tool_id="wf-tool-1",
-        version="1",
+        version=version,
        workflow_entities={},
        workflow_call_depth=1,
        entity=entity,
@@ -98,7 +166,10 @@ def _build_tool() -> WorkflowTool:
    )


-def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_field(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_field(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure that WorkflowTool will throw a `ToolInvokeError` exception when
    `WorkflowAppGenerator.generate` returns a result with `error` key inside
    the `data` element.
@@ -122,11 +193,14 @@ def test_workflow_tool_should_raise_tool_invoke_error_when_result_has_error_fiel
    with pytest.raises(ToolInvokeError) as exc_info:
        # WorkflowTool always returns a generator, so we need to iterate to
        # actually `run` the tool.
-        list(tool.invoke(MagicMock(), "test_user", {}))
+        list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))
    assert exc_info.value.args == ("oops",)


-def test_workflow_tool_does_not_use_pause_state_config(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_does_not_use_pause_state_config(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure pause_state_config is passed as None."""
    tool = _build_tool()

@@ -140,14 +214,17 @@ def test_workflow_tool_does_not_use_pause_state_config(monkeypatch: pytest.Monke
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    call_kwargs = generate_mock.call_args.kwargs
    assert "pause_state_config" in call_kwargs
    assert call_kwargs["pause_state_config"] is None


-def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_passes_parent_trace_context_from_runtime(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure nested workflow runtime metadata is forwarded as parent trace context."""
    tool = _build_tool()
    tool.set_parent_trace_context(
@@ -165,7 +242,7 @@ def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pyt
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    call_kwargs = generate_mock.call_args.kwargs
    assert call_kwargs["args"]["parent_trace_context"].model_dump() == {
@@ -174,7 +251,10 @@ def test_workflow_tool_passes_parent_trace_context_from_runtime(monkeypatch: pyt
    }


-def test_workflow_tool_passes_parent_trace_session_id(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_passes_parent_trace_session_id(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure nested workflows inherit the parent observability session ID."""
    tool = _build_tool()
    tool.entity.parameters = [
@@ -197,14 +277,17 @@ def test_workflow_tool_passes_parent_trace_session_id(monkeypatch: pytest.Monkey
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {"trace_session_id": "user-input-session"}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {"trace_session_id": "user-input-session"}))

    call_kwargs = generate_mock.call_args.kwargs
    assert call_kwargs["args"]["inputs"]["trace_session_id"] == "user-input-session"
    assert call_kwargs["args"]["trace_session_id"] == "session-1"


-def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure private trace context does not overwrite same-named workflow inputs."""
    tool = _build_tool()
    tool.entity.parameters = [
@@ -238,7 +321,7 @@ def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypat

    list(
        tool.invoke(
-            MagicMock(),
+            sqlite_tool_db.caller_session,
            "test_user",
            {
                "outer_workflow_run_id": "user-workflow-input",
@@ -256,7 +339,10 @@ def test_workflow_tool_keeps_user_inputs_named_like_trace_runtime_keys(monkeypat
    }


-def test_workflow_tool_can_clear_parent_trace_context(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_can_clear_parent_trace_context(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure reused WorkflowTool instances do not keep stale parent trace context."""
    tool = _build_tool()
    tool.set_parent_trace_context(
@@ -275,13 +361,16 @@ def test_workflow_tool_can_clear_parent_trace_context(monkeypatch: pytest.Monkey
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    call_kwargs = generate_mock.call_args.kwargs
    assert "parent_trace_context" not in call_kwargs["args"]


-def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_can_clear_trace_session_id(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure reused WorkflowTool instances do not keep stale trace session IDs."""
    tool = _build_tool()
    tool.set_trace_session_id("session-1")
@@ -297,7 +386,7 @@ def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatc
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    call_kwargs = generate_mock.call_args.kwargs
    assert "trace_session_id" not in call_kwargs["args"]
@@ -315,6 +404,7 @@ def test_workflow_tool_can_clear_trace_session_id(monkeypatch: pytest.MonkeyPatc
 def test_workflow_tool_omits_parent_trace_context_when_runtime_is_incomplete(
    monkeypatch: pytest.MonkeyPatch,
    runtime_parameters: dict[str, Any],
+    sqlite_tool_db: SqliteToolDb,
 ):
    """Ensure incomplete runtime metadata does not leak parent trace context into generator args."""
    tool = _build_tool()
@@ -330,13 +420,16 @@ def test_workflow_tool_omits_parent_trace_context_when_runtime_is_incomplete(
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

-    list(tool.invoke(MagicMock(), "test_user", {}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    call_kwargs = generate_mock.call_args.kwargs
    assert "parent_trace_context" not in call_kwargs["args"]


-def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_should_generate_variable_messages_for_outputs(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Test that WorkflowTool should generate variable messages when there are outputs"""
    tool = _build_tool()

@@ -359,7 +452,7 @@ def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

    # Execute tool invocation
-    messages = list(tool.invoke(MagicMock(), "test_user", {}))
+    messages = list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    # Verify variable messages
    variable_messages = [msg for msg in messages if msg.type == ToolInvokeMessage.MessageType.VARIABLE]
@@ -382,7 +475,10 @@ def test_workflow_tool_should_generate_variable_messages_for_outputs(monkeypatch
    assert json_messages[0].message.json_object == mock_outputs


-def test_workflow_tool_should_handle_empty_outputs(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_should_handle_empty_outputs(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Test that WorkflowTool should handle empty outputs correctly"""
    tool = _build_tool()

@@ -402,7 +498,7 @@ def test_workflow_tool_should_handle_empty_outputs(monkeypatch: pytest.MonkeyPat
    monkeypatch.setattr("libs.login.current_user", lambda *args, **kwargs: None)

    # Execute tool invocation
-    messages = list(tool.invoke(MagicMock(), "test_user", {}))
+    messages = list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {}))

    # Verify generated messages
    # Should contain: 0 variable messages + 1 text message + 1 JSON message = 2 messages
@@ -458,41 +554,32 @@ def test_create_file_message_should_include_file_marker():
    assert message.meta == {"file": file_obj}


-def test_resolve_user_from_database_falls_back_to_end_user(monkeypatch: pytest.MonkeyPatch):
+def test_resolve_user_from_database_falls_back_to_end_user(sqlite_tool_db: SqliteToolDb):
    """Ensure worker context can resolve EndUser when Account is missing."""
-
-    tenant = SimpleNamespace(id="tenant_id")
-    end_user = SimpleNamespace(id="end_user_id", tenant_id="tenant_id")
-
-    # Monkeypatch session factory to return our stub session
-    stub_session = StubSession(scalar_results=[tenant, None, end_user])
-    monkeypatch.setattr(
-        "core.tools.workflow_as_tool.tool.session_factory.create_session",
-        lambda: stub_session,
+    _persist_tenant(sqlite_tool_db)
+    end_user = _persist_end_user(sqlite_tool_db)
+    other_tenant_end_user = _persist_end_user(
+        sqlite_tool_db,
+        end_user_id="00000000-0000-0000-0000-000000000007",
+        tenant_id=OTHER_TENANT_ID,
    )

-    tool = _build_tool()
+    tool = _build_tool(tenant_id=TENANT_ID)
    tool.runtime.invoke_from = InvokeFrom.SERVICE_API
-    tool.runtime.tenant_id = "tenant_id"

    resolved_user = tool._resolve_user_from_database(user_id=end_user.id)

-    assert resolved_user is end_user
-    assert stub_session.expunge_calls == [end_user]
+    assert isinstance(resolved_user, EndUser)
+    assert resolved_user.id == end_user.id
+    assert resolved_user.tenant_id == TENANT_ID
+    assert inspect(resolved_user).detached is True
+    assert tool._resolve_user_from_database(user_id=other_tenant_end_user.id) is None


-def test_resolve_user_from_database_returns_none_when_no_tenant(monkeypatch: pytest.MonkeyPatch):
+def test_resolve_user_from_database_returns_none_when_no_tenant(sqlite_tool_db: SqliteToolDb):
    """Return None if tenant cannot be found in worker context."""
-
-    # Monkeypatch session factory to return our stub session with no tenant
-    monkeypatch.setattr(
-        "core.tools.workflow_as_tool.tool.session_factory.create_session",
-        lambda: StubSession(scalar_results=[None]),
-    )
-
-    tool = _build_tool()
+    tool = _build_tool(tenant_id=OTHER_TENANT_ID)
    tool.runtime.invoke_from = InvokeFrom.SERVICE_API
-    tool.runtime.tenant_id = "missing_tenant"

    resolved_user = tool._resolve_user_from_database(user_id="any")

@@ -544,7 +631,10 @@ def test_extract_usage_from_nested():
    assert nested == {"total_tokens": 3}


-def test_invoke_raises_when_user_not_found(monkeypatch: pytest.MonkeyPatch):
+def test_invoke_raises_when_user_not_found(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Raise ToolInvokeError when user resolution fails."""
    tool = _build_tool()
    monkeypatch.setattr(tool, "_get_app", lambda *args, **kwargs: None)
@@ -552,58 +642,45 @@ def test_invoke_raises_when_user_not_found(monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setattr(tool, "_resolve_user", lambda *args, **kwargs: None)

    with pytest.raises(ToolInvokeError, match="User not found"):
-        list(tool.invoke(MagicMock(), "missing", {}))
+        list(tool.invoke(sqlite_tool_db.caller_session, "missing", {}))


-def test_resolve_user_from_database_returns_account(monkeypatch: pytest.MonkeyPatch):
+def test_resolve_user_from_database_returns_account(sqlite_tool_db: SqliteToolDb):
    """Resolve Account and set tenant in worker context."""
-    tenant = SimpleNamespace(id="tenant_id")
-    account = SimpleNamespace(id="account_id", current_tenant=None)
-    set_current_tenant = Mock(side_effect=lambda tenant, *, session: setattr(account, "current_tenant", tenant))
-    account.set_current_tenant_with_session = set_current_tenant
-    session = StubSession(scalar_results=[tenant, account])
+    tenant = _persist_tenant(sqlite_tool_db)
+    account = _persist_account(sqlite_tool_db)
+    tool = _build_tool(tenant_id=TENANT_ID)

-    monkeypatch.setattr("core.tools.workflow_as_tool.tool.session_factory.create_session", lambda: session)
-    tool = _build_tool()
-    tool.runtime.tenant_id = "tenant_id"
-
-    resolved = tool._resolve_user_from_database(user_id="account_id")
-    assert resolved is account
-    assert account.current_tenant is tenant
-    set_current_tenant.assert_called_once_with(tenant, session=session)
-    assert session.expunge_calls == [account]
+    resolved = tool._resolve_user_from_database(user_id=account.id)
+    assert isinstance(resolved, Account)
+    assert resolved.id == account.id
+    assert resolved.current_tenant_id == tenant.id
+    assert inspect(resolved).detached is True


-def test_get_workflow_and_get_app_db_branches(monkeypatch: pytest.MonkeyPatch):
+def test_get_workflow_and_get_app_db_branches(sqlite_tool_db: SqliteToolDb):
    """Cover workflow/app retrieval branches and error cases."""
-    tool = _build_tool()
-    latest_workflow = SimpleNamespace(id="wf-latest")
-    specific_workflow = SimpleNamespace(id="wf-v1")
-    app = SimpleNamespace(id="app-1")
-    sessions = iter(
-        [
-            StubSession(scalar_results=[], scalars_results=[latest_workflow]),
-            StubSession(scalar_results=[specific_workflow], scalars_results=[]),
-            StubSession(scalar_results=[app], scalars_results=[]),
-        ]
-    )
-    monkeypatch.setattr(
-        "core.tools.workflow_as_tool.tool.session_factory.create_session",
-        lambda: next(sessions),
-    )
+    app = _persist_app(sqlite_tool_db)
+    specific_workflow = _persist_workflow(sqlite_tool_db, version="1")
+    latest_workflow = _persist_workflow(sqlite_tool_db, version="2")
+    _persist_workflow(sqlite_tool_db, version=Workflow.VERSION_DRAFT)
+    tool = _build_tool(tenant_id=TENANT_ID, workflow_app_id=APP_ID)

-    assert tool._get_workflow("app-1", "") is latest_workflow
-    assert tool._get_workflow("app-1", "1") is specific_workflow
-    assert tool._get_app("app-1") is app
+    latest = tool._get_workflow(APP_ID, "")
+    specific = tool._get_workflow(APP_ID, "1")
+    resolved_app = tool._get_app(APP_ID)
+
+    assert latest.id == latest_workflow.id
+    assert specific.id == specific_workflow.id
+    assert resolved_app.id == app.id
+    assert inspect(latest).detached is True
+    assert inspect(specific).detached is True
+    assert inspect(resolved_app).detached is True

-    monkeypatch.setattr(
-        "core.tools.workflow_as_tool.tool.session_factory.create_session",
-        lambda: StubSession(scalar_results=[None, None], scalars_results=[None]),
-    )
    with pytest.raises(ValueError, match="workflow not found"):
-        tool._get_workflow("app-1", "1")
+        tool._get_workflow(APP_ID, "missing")
    with pytest.raises(ValueError, match="app not found"):
-        tool._get_app("app-1")
+        tool._get_app("00000000-0000-0000-0000-000000000099")


 def _setup_transform_args_tool(monkeypatch: pytest.MonkeyPatch) -> WorkflowTool:
@@ -722,7 +799,10 @@ def test_transform_args_normalizes_optional_files_parameter(
    assert files == []


-def test_workflow_tool_invocation_normalizes_optional_files_parameter(monkeypatch: pytest.MonkeyPatch):
+def test_workflow_tool_invocation_normalizes_optional_files_parameter(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_tool_db: SqliteToolDb,
+):
    """Ensure casted empty FILES values do not reach workflow input validation as [None]."""
    tool = _build_tool()
    images_param = ToolParameter.get_simple_instance(
@@ -741,7 +821,7 @@ def test_workflow_tool_invocation_normalizes_optional_files_parameter(monkeypatc
    generate_mock = MagicMock(return_value={"data": {}})
    monkeypatch.setattr("core.app.apps.workflow.app_generator.WorkflowAppGenerator.generate", generate_mock)

-    list(tool.invoke(MagicMock(), "test_user", {"images": None}))
+    list(tool.invoke(sqlite_tool_db.caller_session, "test_user", {"images": None}))

    call_kwargs = generate_mock.call_args.kwargs
    assert call_kwargs["args"]["inputs"]["images"] == []
@@ -18,7 +18,6 @@ from models.provider import ProviderType
@pytest.fixture
 def credit_pool_session_factory(sqlite_engine: Engine) -> Iterator[sessionmaker[Session]]:
    """Bind message-created accounting to fixture-owned SQLite sessions."""
-    TenantCreditPool.__table__.create(sqlite_engine)
    session_factory = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
    with patch("events.event_handlers.update_provider_when_message_created.db.session", session_factory):
        yield session_factory
@@ -0,0 +1,43 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from extensions.ext_celery import _enqueue_initial_community_telemetry_heartbeat
+
+
+def test_beat_start_enqueues_community_telemetry_heartbeat() -> None:
+    task = Mock()
+    sender = SimpleNamespace(
+        app=SimpleNamespace(
+            conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
+            tasks={"community_telemetry.send_heartbeat": task},
+        )
+    )
+
+    _enqueue_initial_community_telemetry_heartbeat(sender)
+
+    task.apply_async.assert_called_once_with()
+
+
+def test_beat_start_skips_community_telemetry_when_not_scheduled() -> None:
+    task = Mock()
+    sender = SimpleNamespace(
+        app=SimpleNamespace(
+            conf=SimpleNamespace(beat_schedule={}),
+            tasks={"community_telemetry.send_heartbeat": task},
+        )
+    )
+
+    _enqueue_initial_community_telemetry_heartbeat(sender)
+
+    task.apply_async.assert_not_called()
+
+
+def test_beat_start_skips_community_telemetry_when_task_is_unavailable() -> None:
+    sender = SimpleNamespace(
+        app=SimpleNamespace(
+            conf=SimpleNamespace(beat_schedule={"community_telemetry_heartbeat": {}}),
+            tasks={},
+        )
+    )
+
+    _enqueue_initial_community_telemetry_heartbeat(sender)
@@ -548,6 +548,7 @@ def test_load_agent_app_composer_exposes_draft_save_only(monkeypatch: pytest.Mon
    agent = SimpleNamespace(
        id="agent-1",
        active_config_snapshot_id="version-1",
+        active_config_is_published=True,
        updated_by="account-1",
        created_by="account-1",
        app_id="app-1",
@@ -567,6 +568,7 @@ def test_load_agent_app_composer_exposes_draft_save_only(monkeypatch: pytest.Mon
    result = AgentComposerService.load_agent_app_composer(session=session, tenant_id="tenant-1", app_id="app-1")

    assert result["save_options"] == [ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value]
+    assert result["active_config_is_published"] is True


 def test_save_agent_app_composer_rejects_version_save_strategy():
@@ -610,7 +612,11 @@ def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.Monkey
        lambda **kwargs: saved.update(kwargs) or SimpleNamespace(id="draft-1"),
    )
    monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **_kwargs: active_version)
-    monkeypatch.setattr(AgentComposerService, "load_agent_composer", lambda **kwargs: {"loaded": True})
+    monkeypatch.setattr(
+        AgentComposerService,
+        "load_agent_composer",
+        lambda **kwargs: {"loaded": True, "active_config_is_published": agent.active_config_is_published},
+    )
    payload = ComposerSavePayload.model_validate(
        {
            "variant": ComposerVariant.AGENT_APP.value,
@@ -628,7 +634,7 @@ def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.Monkey
    )

    assert result.pop("validation") == {"warnings": [], "knowledge_retrieval_placeholder": []}
-    assert result == {"loaded": True}
+    assert result == {"loaded": True, "active_config_is_published": False}
    assert saved["draft_type"] == AgentConfigDraftType.DRAFT
    assert saved["agent_soul"].model_dump(mode="json") == _agent_soul_with_model().model_dump(mode="json")
    assert agent.active_config_is_published is False
@@ -657,7 +663,11 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
        lambda **_kwargs: SimpleNamespace(id="draft-1"),
    )
    monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **_kwargs: active_version)
-    monkeypatch.setattr(AgentComposerService, "load_agent_composer", lambda **_kwargs: {"loaded": True})
+    monkeypatch.setattr(
+        AgentComposerService,
+        "load_agent_composer",
+        lambda **_kwargs: {"loaded": True, "active_config_is_published": agent.active_config_is_published},
+    )
    payload = ComposerSavePayload.model_validate(
        {
            "variant": ComposerVariant.AGENT_APP.value,
@@ -666,7 +676,7 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
        }
    )

-    AgentComposerService.save_agent_app_composer(
+    result = AgentComposerService.save_agent_app_composer(
        session=session,
        tenant_id="tenant-1",
        app_id="app-1",
@@ -675,6 +685,7 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
    )

    assert agent.active_config_is_published is True
+    assert result["active_config_is_published"] is True
    assert fake_session.flushes >= 1


@@ -246,6 +246,26 @@ class TestPluginModelProviderCache:
            call([cache_key]),
        ]

+    def test_fetch_plugin_model_providers_bypasses_redis_when_cache_disabled(self) -> None:
+        """With the cache disabled the daemon is the only source, and Redis is never touched."""
+        with patch(f"{MODULE}.redis_client") as redis_client, patch(f"{MODULE}.dify_config") as config:
+            config.PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED = False
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            first = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+            second = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert [provider.provider for provider in first] == ["langgenius/openai/openai"]
+        assert [provider.provider for provider in second] == ["langgenius/openai/openai"]
+        assert client.fetch_model_providers.call_count == 2
+        redis_client.get.assert_not_called()
+        redis_client.mget.assert_not_called()
+        redis_client.setex.assert_not_called()
+        redis_client.lock.assert_not_called()
+
    def test_fetch_plugin_model_providers_refetches_when_cache_read_fails(self) -> None:
        """Redis read failures do not block provider discovery for the tenant."""
        with patch(f"{MODULE}.redis_client") as redis_client:
@@ -1,7 +1,7 @@
 import json
-from collections.abc import Iterator
 from datetime import datetime, timedelta
 from unittest.mock import MagicMock, patch
+from uuid import UUID

 import pytest
 from sqlalchemy import event, select
@@ -10,7 +10,6 @@ from sqlalchemy.orm import Session
 from configs import dify_config
 from models.account import (
    Account,
-    AccountIntegrate,
    AccountStatus,
    Tenant,
    TenantAccountJoin,
@@ -113,22 +112,6 @@ class TestAccountService:
    - Error conditions and edge cases
    """

-    @pytest.fixture
-    def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
-        """SQLite session with the account/workspace tables these service tests touch."""
-        tables = [
-            model.metadata.tables[model.__tablename__]
-            for model in (
-                Account,
-                Tenant,
-                TenantAccountJoin,
-                TenantPluginAutoUpgradeStrategy,
-            )
-        ]
-        Account.metadata.create_all(sqlite_engine, tables=tables)
-        with Session(sqlite_engine, expire_on_commit=False) as session:
-            yield session
-
    @pytest.fixture
    def mock_password_dependencies(self):
        """Mock setup for password-related functions."""
@@ -1263,24 +1246,6 @@ class TestRegisterService:
    - Error conditions and edge cases
    """

-    @pytest.fixture
-    def sqlite_session(self, sqlite_engine) -> Iterator[Session]:
-        """SQLite session with the account/workspace tables registration flows touch."""
-        tables = [
-            model.metadata.tables[model.__tablename__]
-            for model in (
-                Account,
-                AccountIntegrate,
-                Tenant,
-                TenantAccountJoin,
-                TenantPluginAutoUpgradeStrategy,
-                DifySetup,
-            )
-        ]
-        Account.metadata.create_all(sqlite_engine, tables=tables)
-        with Session(sqlite_engine, expire_on_commit=False) as session:
-            yield session
-
    @pytest.fixture
    def mock_redis_dependencies(self):
        """Mock setup for Redis-related functions."""
@@ -1325,7 +1290,10 @@ class TestRegisterService:
        with patch("services.account_service.AccountService.create_account") as mock_create_account:
            mock_create_account.return_value = mock_account

-            with patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant:
+            with (
+                patch("services.account_service.TenantService.create_owner_tenant_if_not_exist") as mock_create_tenant,
+                patch("services.account_service.CommunityTelemetryService.report_install") as mock_report_install,
+            ):
                RegisterService.setup(
                    "admin@example.com",
                    "Admin User",
@@ -1344,7 +1312,39 @@ class TestRegisterService:
                    session=sqlite_session,
                )
                mock_create_tenant.assert_called_once_with(account=mock_account, is_setup=True, session=sqlite_session)
-                assert sqlite_session.scalar(select(DifySetup)) is not None
+                dify_setup = sqlite_session.scalar(select(DifySetup))
+                assert dify_setup is not None
+                assert dify_setup.instance_id is not None
+                assert str(UUID(dify_setup.instance_id)) == dify_setup.instance_id
+                assert dify_setup.install_reported_at is None
+                assert dify_setup.last_heartbeat_at is None
+                mock_report_install.assert_called_once_with(session=sqlite_session)
+
+    def test_setup_succeeds_when_telemetry_install_report_fails(
+        self, sqlite_session: Session, mock_external_service_dependencies
+    ):
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock()
+
+        with (
+            patch("services.account_service.AccountService.create_account", return_value=mock_account),
+            patch("services.account_service.TenantService.create_owner_tenant_if_not_exist"),
+            patch(
+                "services.account_service.CommunityTelemetryService.report_install",
+                side_effect=RuntimeError("telemetry unavailable"),
+            ),
+        ):
+            RegisterService.setup(
+                "admin@example.com",
+                "Admin User",
+                "password123",
+                "192.168.1.1",
+                "en-US",
+                session=sqlite_session,
+            )
+
+        assert sqlite_session.scalar(select(DifySetup)) is not None

    def test_setup_failure_rollback(self, sqlite_session: Session, mock_external_service_dependencies):
        """Test setup failure with proper rollback."""
@@ -1,19 +1,23 @@
 from __future__ import annotations

 from collections.abc import Callable
+from datetime import datetime
 from types import SimpleNamespace
 from typing import cast
 from unittest.mock import MagicMock, patch
+from uuid import uuid4

 import pytest
+from sqlalchemy import event
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session

 from graphon.model_runtime.entities.model_entities import ModelType
 from models import Account
-from models.model import App, AppMode, AppModelConfig
+from models.model import App, AppMode, AppModelConfig, IconType
 from models.workflow import Workflow
 from services.agent.errors import AgentNameConflictError
-from services.app_service import AppService, CreateAppParams
+from services.app_service import AppListParams, AppService, CreateAppParams


 class TestCreateAppTransactionBoundary:
@@ -236,6 +240,92 @@ class TestOpenapiVisibilityHelpers:
        mock_session.execute.assert_called_once()


+@pytest.mark.parametrize("sqlite_session", [(Account, App, AppModelConfig)], indirect=True)
+def test_get_recent_apps_uses_one_tenant_scoped_projection_query(sqlite_session: Session) -> None:
+    tenant_id = str(uuid4())
+    other_tenant_id = str(uuid4())
+    account = Account(name="Recent Apps Author", email="recent-apps@example.com")
+    sqlite_session.add(account)
+    sqlite_session.flush()
+
+    def create_app(*, name: str, tenant_id: str, updated_at: datetime, mode: AppMode = AppMode.CHAT) -> App:
+        app = App()
+        app.id = str(uuid4())
+        app.tenant_id = tenant_id
+        app.name = name
+        app.description = ""
+        app.mode = mode
+        app.icon_type = IconType.EMOJI
+        app.icon = "🚀"
+        app.icon_background = "#FFFFFF"
+        app.enable_site = False
+        app.enable_api = False
+        app.created_by = account.id
+        app.maintainer = account.id
+        app.created_at = updated_at
+        app.updated_at = updated_at
+        app.use_icon_as_answer_icon = False
+        return app
+
+    newest = create_app(name="Newest", tenant_id=tenant_id, updated_at=datetime(2026, 7, 3))
+    legacy_agent = AppModelConfig(app_id=newest.id)
+    legacy_agent.agent_mode = '{"enabled": true, "strategy": "react"}'
+    newest.app_model_config_id = legacy_agent.id
+    second = create_app(
+        name="Second",
+        tenant_id=tenant_id,
+        updated_at=datetime(2026, 7, 2),
+        mode=AppMode.WORKFLOW,
+    )
+    second.icon_type = None
+    second.icon = None
+    second.icon_background = None
+    second.created_by = None
+    second.maintainer = None
+    channel = create_app(
+        name="Channel",
+        tenant_id=tenant_id,
+        updated_at=datetime(2026, 7, 5),
+        mode=AppMode.CHANNEL,
+    )
+    rag_pipeline = create_app(
+        name="RAG Pipeline",
+        tenant_id=tenant_id,
+        updated_at=datetime(2026, 7, 4),
+        mode=AppMode.RAG_PIPELINE,
+    )
+    oldest = create_app(name="Oldest", tenant_id=tenant_id, updated_at=datetime(2026, 7, 1))
+    foreign = create_app(name="Foreign", tenant_id=other_tenant_id, updated_at=datetime(2026, 7, 4))
+    sqlite_session.add_all([newest, legacy_agent, second, channel, rag_pipeline, oldest, foreign])
+    sqlite_session.commit()
+
+    statements: list[str] = []
+    bind = sqlite_session.get_bind()
+
+    def record_sql(_conn, _cursor, statement, _parameters, _context, _executemany) -> None:
+        statements.append(statement)
+
+    event.listen(bind, "before_cursor_execute", record_sql)
+    try:
+        recent_apps = AppService().get_recent_apps(
+            account.id,
+            tenant_id,
+            AppListParams(limit=2),
+            sqlite_session,
+        )
+    finally:
+        event.remove(bind, "before_cursor_execute", record_sql)
+
+    assert [(app.name, app.mode, app.icon_type, app.author_name, app.maintainer) for app in recent_apps] == [
+        ("Newest", AppMode.CHAT, IconType.EMOJI, "Recent Apps Author", account.id),
+        ("Second", AppMode.WORKFLOW, None, None, None),
+    ]
+    select_statements = [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")]
+    assert len(select_statements) == 1
+    assert "count(" not in select_statements[0].lower()
+    assert "app_model_configs" not in select_statements[0].lower()
+
+
 class TestAppMeta:
    def test_loads_workflow_with_caller_session(self):
        session = MagicMock()
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
wangxiaolei GitHub yunlu.wen	6f8ed69ee1	fix: fix mcp output_schema is optional (#39453 ) Co-authored-by: yunlu.wen <yunlu.wen@dify.ai>	2026-07-28 02:31:18 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	59b879d2df	chore: bump version to 1.16.1 (#39653 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-28 01:09:25 +00:00
Xiyuan ChenandGitHub	a57b0b9b58	feat(plugin): allow disabling the tenant plugin model providers cache (#39632 )	2026-07-27 23:53:59 +00:00
Xiyuan ChenandGitHub	d8506efed6	fix(cli): decouple release script tests from the live compat window (#39658 )	2026-07-27 23:50:05 +00:00
JingyiandGitHub	b25b28cc76	fix(workflow): align block icon vector sizes (#39657 )	2026-07-27 23:43:03 +00:00
yyh GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	f68cabe226	fix(web): align tour trigger DOM order (#39654 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 13:00:16 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	f73f83c5d6	test: use SQLite sessions in core app (#39075 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:45:16 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	29801d6a65	test: use SQLite sessions in core tools (#39077 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:45:11 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	75016e8bfe	test: use SQLite sessions in core ops (#39072 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:41:17 +00:00
yyhandGitHub	b6dea7b2ba	fix(workflow): preserve latest collaboration session (#39646 )	2026-07-27 10:37:45 +00:00
yyhandGitHub	b2b1cd7e97	chore: update workspace dependencies (#39641 )	2026-07-27 10:27:31 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	ca63e01d45	test: use SQLite sessions in controllers service api (#39069 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:26:23 +00:00
yyhandGitHub	271b6e1f5c	fix(ui): unify combobox trigger focus rings (#39643 )	2026-07-27 10:21:32 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	609ddf9e01	test: use SQLite sessions in core memory (#39066 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 10:15:08 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	9fd1fea58c	test: use SQLite sessions in services core (#39057 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 09:36:44 +00:00
JoelandGitHub	8de3b4d033	fix: prevent the agent build draft UI from flashing during apply (#39642 )	2026-07-27 09:32:52 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	d9c038daf2	chore(deps): bump the storage group across 1 directory with 3 updates (#39604 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 09:24:58 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron.wang	c9c057fd04	test: use SQLite sessions in rag retrieval (#39049 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Byron.wang <byron@dify.ai>	2026-07-27 09:12:30 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	85189be53d	chore: bump pymdown-extensions from 10.21.2 to 11.0 in /dify-agent (#39618 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 09:10:22 +00:00
Asuka Minato GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	01e736aaf7	test: use sqlite3 session in test_extract_thread_messages (#38739 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 09:00:47 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	a28969f564	chore(deps): bump golang.org/x/net from 0.53.0 to 0.55.0 in /dify-agent-runtime (#39025 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 08:58:40 +00:00
Asuka MinatoandGitHub	42f9610ba5	test: use sqlite3 session in test_api (#38733 )	2026-07-27 08:57:51 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	1a7ffbe5ee	chore: bump gitpython from 3.1.52 to 3.1.54 in /api (#39619 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 08:57:44 +00:00
JoelandGitHub	481354ca70	refactor: make agent composer save and publish state consistent (#39637 )	2026-07-27 08:43:45 +00:00
Yunlu Wen GitHub Copilot Autofix powered by AI	3c80857ea3	feat: add bearer auth to agent backend (#39622 ) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>	2026-07-27 08:37:30 +00:00
Xiyuan ChenandGitHub	94702efbc2	fix: gate service API, MCP and trigger surfaces on enterprise license (#39635 )	2026-07-27 08:23:15 +00:00
Asuka MinatoandGitHub	755f7b0e8b	test: use sqlite3 session in test_workflow_comment_service (#38700 )	2026-07-27 08:22:45 +00:00
Asuka MinatoandGitHub	a1c9564b30	test: use sqlite3 session in test_llm_generator_missing (#38742 )	2026-07-27 08:22:16 +00:00
Asuka MinatoandGitHub	97acd9c70b	test: use sqlite3 session in test_utils (#38729 )	2026-07-27 08:21:39 +00:00
Asuka MinatoandGitHub	71601bf76c	test: use sqlite3 session in test_draft_var_loader_simple (#38702 )	2026-07-27 08:20:18 +00:00
Asuka MinatoandGitHub	460efbf285	test: use sqlite3 session in test_workflow_run_service (#38701 )	2026-07-27 08:19:32 +00:00
Escape0707 GitHub Asuka Minato	a752f43b8e	test: use pristine file-backed SQLite fixtures (#39624 ) Co-authored-by: Asuka Minato <i@asukaminato.eu.org>	2026-07-27 08:07:05 +00:00
非法操作 GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	b3774bfe1c	perf: add lightweight recent apps endpoint (#39625 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 08:00:39 +00:00
yyhandGitHub	4313d23889	perf(web): lazy-load Home creation modals (#39634 )	2026-07-27 07:48:35 +00:00
林玮 (Jade Lin)andGitHub	26a43db6a4	fix(api): wait for workflow worker cleanup before task completion (#39614 )	2026-07-27 07:34:25 +00:00
yyhandGitHub	e5d40336b3	fix(workflow): lazy-load Loro collaboration runtime (#39631 )	2026-07-27 07:15:13 +00:00
yyhandGitHub	58e2bcbba1	fix(web): stop reporting workspace status (#39626 )	2026-07-27 06:22:37 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	e1b55f54e6	feat(agent): add a squid proxy for agent sandbox (#39544 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 06:21:20 +00:00
Yunlu Wen GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	989039db6e	fix: use jinja sandbox (#39609 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 06:03:01 +00:00
Joel GitHub 林玮autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	52428df1bd	feat: add amptitude to new agent (#39608 ) Co-authored-by: 林玮 (Jade Lin) <linw1995@icloud.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 05:54:45 +00:00
yyh GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	28d174603f	fix: align workspace card plan ownership (#39616 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 05:01:02 +00:00
Xiyuan ChenandGitHub	b6099d09ff	feat(inner_api): add endpoint to invalidate plugin model providers cache (#39468 )	2026-07-27 04:33:36 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2962a7ea93	chore(deps): bump the github-actions-dependencies group with 6 updates (#39603 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 04:32:09 +00:00
zyssyz123andGitHub	cc01189966	fix(agent): expose publish state from composer (#39615 )	2026-07-27 04:05:38 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	d563d6e7df	chore: bump pypdf from 6.10.2 to 6.14.2 in /dify-agent (#39502 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-27 03:56:57 +00:00
CYJ1226andGitHub	e4e99d198b	fix: preserve spaces during recursive text splitting (#39498 )	2026-07-27 03:21:26 +00:00
yyhandGitHub	34e6e5a049	refactor(web): simplify Dify logo theming (#39580 )	2026-07-27 03:01:24 +00:00
林玮 (Jade Lin)GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	7e6ba05464	feat(api): expose app mode in webapp site response (#39607 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 02:57:18 +00:00
Madan kumarandGitHub	aeda37db68	fix(rag): stop the document cleaner from stripping valid characters ï, ¿, ¾ (#39215 )	2026-07-27 02:53:33 +00:00
FFXNandGitHub	441f9f9ec0	fix: The expiration time for web app login to JWT is incorrectly set. (#39537 )	2026-07-27 02:43:32 +00:00
Taranum Wasu GitHub Taranum Wasu Cursor	f16c249b1e	fix(api): honor array-element limit and byte budget for list[File] in VariableTruncator (#39218 ) (#39220 ) Co-authored-by: Taranum Wasu <taranumwasu@Taranums-MacBook-Pro.local> Co-authored-by: Cursor <cursoragent@cursor.com>	2026-07-27 02:12:57 +00:00
FFXN GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	858b3b74dc	feat: add CE telemetry report (#39452 ) Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>	2026-07-27 02:04:28 +00:00
David Park GitHub Claude Opus 4.8	d5c0e927c6	fix(web): stop chunk length/overlap inputs collapsing to an unusable width in narrow containers (#39600 ) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-27 02:03:17 +00:00
Asuka MinatoandGitHub	b81737cfb3	test: use sqlite3 session in test_message_service (#38698 )	2026-07-27 02:01:47 +00:00