fix: read marketplace install count from payload

fix: handle null plugin badges (#36767 )
Fix/marketplace collection templates limit (#36709 )
2026-07-24 13:28:32 +08:00 · 2026-05-28 15:12:15 +08:00 · 2026-05-27 12:07:27 +08:00 · 2026-05-26 16:01:45 +08:00 · 2026-05-26 15:53:10 +08:00 · 2026-05-26 15:28:07 +08:00
583 changed files with 24994 additions and 16266 deletions
@@ -6,7 +6,6 @@ runs:
    - name: Setup Vite+
      uses: voidzero-dev/setup-vp@20553a7a7429c429a74894104a2835d7fed28a72 # v1.3.0
      with:
-        working-directory: web
        node-version-file: .nvmrc
        cache: true
        run-install: true
@@ -14,11 +14,11 @@ concurrency:
  cancel-in-progress: true

 jobs:
-  test:
-    name: API Tests
+  api-unit:
+    name: API Unit Tests
    runs-on: ubuntu-latest
    env:
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      COVERAGE_FILE: coverage-unit
    defaults:
      run:
        shell: bash
@@ -35,7 +35,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -50,6 +50,52 @@ jobs:
      - name: Run dify config tests
        run: uv run --project api dev/pytest/pytest_config_tests.py

+      - name: Run Unit Tests
+        run: uv run --project api bash dev/pytest/pytest_unit_tests.sh
+
+      - name: Upload unit coverage data
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+        with:
+          name: api-coverage-unit
+          path: coverage-unit
+          retention-days: 1
+
+  api-integration:
+    name: API Integration Tests
+    runs-on: ubuntu-latest
+    env:
+      COVERAGE_FILE: coverage-integration
+      STORAGE_TYPE: opendal
+      OPENDAL_SCHEME: fs
+      OPENDAL_FS_ROOT: /tmp/dify-storage
+    defaults:
+      run:
+        shell: bash
+    strategy:
+      matrix:
+        python-version:
+          - "3.12"
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+        with:
+          enable-cache: true
+          python-version: ${{ matrix.python-version }}
+          cache-dependency-glob: api/uv.lock
+
+      - name: Check UV lockfile
+        run: uv lock --project api --check
+
+      - name: Install dependencies
+        run: uv sync --project api --dev
+
      - name: Set up dotenvs
        run: |
          cp docker/.env.example docker/.env
@@ -73,23 +119,91 @@ jobs:
        run: |
          cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env

-      - name: Run API Tests
-        env:
-          STORAGE_TYPE: opendal
-          OPENDAL_SCHEME: fs
-          OPENDAL_FS_ROOT: /tmp/dify-storage
+      - name: Run Integration Tests
        run: |
          uv run --project api pytest \
            -n auto \
            --timeout "${PYTEST_TIMEOUT:-180}" \
            api/tests/integration_tests/workflow \
            api/tests/integration_tests/tools \
-            api/tests/test_containers_integration_tests \
-            api/tests/unit_tests
+            api/tests/test_containers_integration_tests
+
+      - name: Upload integration coverage data
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+        with:
+          name: api-coverage-integration
+          path: coverage-integration
+          retention-days: 1
+
+  api-coverage:
+    name: API Coverage
+    runs-on: ubuntu-latest
+    needs:
+      - api-unit
+      - api-integration
+    env:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      COVERAGE_FILE: .coverage
+    defaults:
+      run:
+        shell: bash
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+        with:
+          enable-cache: true
+          python-version: "3.12"
+          cache-dependency-glob: api/uv.lock
+
+      - name: Install dependencies
+        run: uv sync --project api --dev
+
+      - name: Download coverage data
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          path: coverage-data
+          pattern: api-coverage-*
+          merge-multiple: true
+
+      - name: Combine coverage
+        run: |
+          set -euo pipefail
+
+          echo "### API Coverage" >> "$GITHUB_STEP_SUMMARY"
+          echo "" >> "$GITHUB_STEP_SUMMARY"
+          echo "Merged backend coverage report generated for Codecov project status." >> "$GITHUB_STEP_SUMMARY"
+          echo "" >> "$GITHUB_STEP_SUMMARY"
+
+          unit_coverage="$(find coverage-data -type f -name coverage-unit -print -quit)"
+          integration_coverage="$(find coverage-data -type f -name coverage-integration -print -quit)"
+          : "${unit_coverage:?coverage-unit artifact not found}"
+          : "${integration_coverage:?coverage-integration artifact not found}"
+
+          report_file="$(mktemp)"
+          uv run --project api coverage combine "$unit_coverage" "$integration_coverage"
+          uv run --project api coverage report --show-missing | tee "$report_file"
+          echo "Summary: \`$(tail -n 1 "$report_file")\`" >> "$GITHUB_STEP_SUMMARY"
+          {
+            echo ""
+            echo "<details><summary>Coverage report</summary>"
+            echo ""
+            echo '```'
+            cat "$report_file"
+            echo '```'
+            echo "</details>"
+          } >> "$GITHUB_STEP_SUMMARY"
+          uv run --project api coverage xml -o coverage.xml

      - name: Report coverage
-        if: ${{ env.CODECOV_TOKEN != '' && matrix.python-version == '3.12' }}
-        uses: codecov/codecov-action@1af58845a975a7985b0beb0cbe6fbbb71a41dbad # v5.5.3
+        if: ${{ env.CODECOV_TOKEN != '' }}
+        uses: codecov/codecov-action@57e3a136b779b570ffcdbf80b3bdc90e7fab3de2 # v6.0.0
        with:
          files: ./coverage.xml
          disable_search: true
@@ -39,6 +39,10 @@ jobs:
        with:
          files: |
            web/**
+            package.json
+            pnpm-lock.yaml
+            pnpm-workspace.yaml
+            .nvmrc
      - name: Check api inputs
        if: github.event_name != 'merge_group'
        id: api-changes
@@ -52,7 +56,7 @@ jobs:
          python-version: "3.11"

      - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0

      - name: Generate Docker Compose
        if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -24,27 +24,39 @@ env:

 jobs:
  build:
-    runs-on: ${{ matrix.platform == 'linux/arm64' && 'arm64_runner' || 'ubuntu-latest' }}
+    runs-on: ${{ matrix.runs_on }}
    if: github.repository == 'langgenius/dify'
    strategy:
      matrix:
        include:
          - service_name: "build-api-amd64"
            image_name_env: "DIFY_API_IMAGE_NAME"
-            context: "api"
+            artifact_context: "api"
+            build_context: "{{defaultContext}}:api"
+            file: "Dockerfile"
            platform: linux/amd64
+            runs_on: ubuntu-latest
          - service_name: "build-api-arm64"
            image_name_env: "DIFY_API_IMAGE_NAME"
-            context: "api"
+            artifact_context: "api"
+            build_context: "{{defaultContext}}:api"
+            file: "Dockerfile"
            platform: linux/arm64
+            runs_on: ubuntu-24.04-arm
          - service_name: "build-web-amd64"
            image_name_env: "DIFY_WEB_IMAGE_NAME"
-            context: "web"
+            artifact_context: "web"
+            build_context: "{{defaultContext}}"
+            file: "web/Dockerfile"
            platform: linux/amd64
+            runs_on: ubuntu-latest
          - service_name: "build-web-arm64"
            image_name_env: "DIFY_WEB_IMAGE_NAME"
-            context: "web"
+            artifact_context: "web"
+            build_context: "{{defaultContext}}"
+            file: "web/Dockerfile"
            platform: linux/arm64
+            runs_on: ubuntu-24.04-arm

    steps:
      - name: Prepare
@@ -58,9 +70,6 @@ jobs:
          username: ${{ env.DOCKERHUB_USER }}
          password: ${{ env.DOCKERHUB_TOKEN }}

-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0
-
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0

@@ -74,7 +83,8 @@ jobs:
        id: build
        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
        with:
-          context: "{{defaultContext}}:${{ matrix.context }}"
+          context: ${{ matrix.build_context }}
+          file: ${{ matrix.file }}
          platforms: ${{ matrix.platform }}
          build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
          labels: ${{ steps.meta.outputs.labels }}
@@ -93,7 +103,7 @@ jobs:
      - name: Upload digest
        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
        with:
-          name: digests-${{ matrix.context }}-${{ env.PLATFORM_PAIR }}
+          name: digests-${{ matrix.artifact_context }}-${{ env.PLATFORM_PAIR }}
          path: /tmp/digests/*
          if-no-files-found: error
          retention-days: 1
@@ -19,7 +19,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -69,7 +69,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -6,7 +6,12 @@ on:
      - "main"
    paths:
      - api/Dockerfile
+      - web/docker/**
      - web/Dockerfile
+      - package.json
+      - pnpm-lock.yaml
+      - pnpm-workspace.yaml
+      - .nvmrc

 concurrency:
  group: docker-build-${{ github.head_ref || github.run_id }}
@@ -14,26 +19,31 @@ concurrency:

 jobs:
  build-docker:
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.runs_on }}
    strategy:
      matrix:
        include:
          - service_name: "api-amd64"
            platform: linux/amd64
-            context: "api"
+            runs_on: ubuntu-latest
+            context: "{{defaultContext}}:api"
+            file: "Dockerfile"
          - service_name: "api-arm64"
            platform: linux/arm64
-            context: "api"
+            runs_on: ubuntu-24.04-arm
+            context: "{{defaultContext}}:api"
+            file: "Dockerfile"
          - service_name: "web-amd64"
            platform: linux/amd64
-            context: "web"
+            runs_on: ubuntu-latest
+            context: "{{defaultContext}}"
+            file: "web/Dockerfile"
          - service_name: "web-arm64"
            platform: linux/arm64
-            context: "web"
+            runs_on: ubuntu-24.04-arm
+            context: "{{defaultContext}}"
+            file: "web/Dockerfile"
    steps:
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0
-
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0

@@ -41,8 +51,8 @@ jobs:
        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
        with:
          push: false
-          context: "{{defaultContext}}:${{ matrix.context }}"
-          file: "${{ matrix.file }}"
+          context: ${{ matrix.context }}
+          file: ${{ matrix.file }}
          platforms: ${{ matrix.platform }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
@@ -42,6 +42,7 @@ jobs:
    runs-on: ubuntu-latest
    outputs:
      api-changed: ${{ steps.changes.outputs.api }}
+      e2e-changed: ${{ steps.changes.outputs.e2e }}
      web-changed: ${{ steps.changes.outputs.web }}
      vdb-changed: ${{ steps.changes.outputs.vdb }}
      migration-changed: ${{ steps.changes.outputs.migration }}
@@ -53,21 +54,71 @@ jobs:
          filters: |
            api:
              - 'api/**'
-              - 'docker/**'
              - '.github/workflows/api-tests.yml'
+              - '.github/workflows/expose_service_ports.sh'
+              - 'docker/.env.example'
+              - 'docker/middleware.env.example'
+              - 'docker/docker-compose.middleware.yaml'
+              - 'docker/docker-compose-template.yaml'
+              - 'docker/generate_docker_compose'
+              - 'docker/ssrf_proxy/**'
+              - 'docker/volumes/sandbox/conf/**'
            web:
              - 'web/**'
+              - 'package.json'
+              - 'pnpm-lock.yaml'
+              - 'pnpm-workspace.yaml'
+              - '.nvmrc'
              - '.github/workflows/web-tests.yml'
              - '.github/actions/setup-web/**'
+            e2e:
+              - 'api/**'
+              - 'api/pyproject.toml'
+              - 'api/uv.lock'
+              - 'e2e/**'
+              - 'web/**'
+              - 'package.json'
+              - 'pnpm-lock.yaml'
+              - 'pnpm-workspace.yaml'
+              - '.nvmrc'
+              - 'docker/docker-compose.middleware.yaml'
+              - 'docker/middleware.env.example'
+              - '.github/workflows/web-e2e.yml'
+              - '.github/actions/setup-web/**'
            vdb:
              - 'api/core/rag/datasource/**'
-              - 'docker/**'
+              - 'api/tests/integration_tests/vdb/**'
              - '.github/workflows/vdb-tests.yml'
+              - '.github/workflows/expose_service_ports.sh'
+              - 'docker/.env.example'
+              - 'docker/middleware.env.example'
+              - 'docker/docker-compose.yaml'
+              - 'docker/docker-compose-template.yaml'
+              - 'docker/generate_docker_compose'
+              - 'docker/certbot/**'
+              - 'docker/couchbase-server/**'
+              - 'docker/elasticsearch/**'
+              - 'docker/iris/**'
+              - 'docker/nginx/**'
+              - 'docker/pgvector/**'
+              - 'docker/ssrf_proxy/**'
+              - 'docker/startupscripts/**'
+              - 'docker/tidb/**'
+              - 'docker/volumes/**'
              - 'api/uv.lock'
              - 'api/pyproject.toml'
            migration:
              - 'api/migrations/**'
+              - 'api/.env.example'
              - '.github/workflows/db-migration-test.yml'
+              - '.github/workflows/expose_service_ports.sh'
+              - 'docker/.env.example'
+              - 'docker/middleware.env.example'
+              - 'docker/docker-compose.middleware.yaml'
+              - 'docker/docker-compose-template.yaml'
+              - 'docker/generate_docker_compose'
+              - 'docker/ssrf_proxy/**'
+              - 'docker/volumes/sandbox/conf/**'

  # Run tests in parallel while always emitting stable required checks.
  api-tests-run:
@@ -190,6 +241,65 @@ jobs:
          echo "Web tests were not required, but the skip job finished with result: $SKIP_RESULT" >&2
          exit 1

+  web-e2e-run:
+    name: Run Web Full-Stack E2E
+    needs:
+      - pre_job
+      - check-changes
+    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
+    uses: ./.github/workflows/web-e2e.yml
+
+  web-e2e-skip:
+    name: Skip Web Full-Stack E2E
+    needs:
+      - pre_job
+      - check-changes
+    if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed != 'true'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Report skipped web full-stack e2e
+        run: echo "No E2E-related changes detected; skipping web full-stack E2E."
+
+  web-e2e:
+    name: Web Full-Stack E2E
+    if: ${{ always() }}
+    needs:
+      - pre_job
+      - check-changes
+      - web-e2e-run
+      - web-e2e-skip
+    runs-on: ubuntu-latest
+    steps:
+      - name: Finalize Web Full-Stack E2E status
+        env:
+          SHOULD_SKIP_WORKFLOW: ${{ needs.pre_job.outputs.should_skip }}
+          TESTS_CHANGED: ${{ needs.check-changes.outputs.e2e-changed }}
+          RUN_RESULT: ${{ needs.web-e2e-run.result }}
+          SKIP_RESULT: ${{ needs.web-e2e-skip.result }}
+        run: |
+          if [[ "$SHOULD_SKIP_WORKFLOW" == 'true' ]]; then
+            echo "Web full-stack E2E was skipped because this workflow run duplicated a successful or newer run."
+            exit 0
+          fi
+
+          if [[ "$TESTS_CHANGED" == 'true' ]]; then
+            if [[ "$RUN_RESULT" == 'success' ]]; then
+              echo "Web full-stack E2E ran successfully."
+              exit 0
+            fi
+
+            echo "Web full-stack E2E was required but finished with result: $RUN_RESULT" >&2
+            exit 1
+          fi
+
+          if [[ "$SKIP_RESULT" == 'success' ]]; then
+            echo "Web full-stack E2E was skipped because no E2E-related files changed."
+            exit 0
+          fi
+
+          echo "Web full-stack E2E was not required, but the skip job finished with result: $SKIP_RESULT" >&2
+          exit 1
+
  style-check:
    name: Style Check
    needs: pre_job
@@ -22,7 +22,7 @@ jobs:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: true

@@ -50,6 +50,17 @@ jobs:
        run: |
          diff -u /tmp/pyrefly_base.txt /tmp/pyrefly_pr.txt > pyrefly_diff.txt || true

+      - name: Check if line counts match
+        id: line_count_check
+        run: |
+          base_lines=$(wc -l < /tmp/pyrefly_base.txt)
+          pr_lines=$(wc -l < /tmp/pyrefly_pr.txt)
+          if [ "$base_lines" -eq "$pr_lines" ]; then
+            echo "same=true" >> $GITHUB_OUTPUT
+          else
+            echo "same=false" >> $GITHUB_OUTPUT
+          fi
+
      - name: Save PR number
        run: |
          echo ${{ github.event.pull_request.number }} > pr_number.txt
@@ -63,7 +74,7 @@ jobs:
            pr_number.txt

      - name: Comment PR with pyrefly diff
-        if: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
+        if: ${{ github.event.pull_request.head.repo.full_name == github.repository && steps.line_count_check.outputs.same == 'false' }}
        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
        with:
          github-token: ${{ secrets.GITHUB_TOKEN }}
@@ -33,7 +33,7 @@ jobs:

      - name: Setup UV and Python
        if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: false
          python-version: "3.12"
@@ -49,7 +49,7 @@ jobs:

      - name: Run Type Checks
        if: steps.changed-files.outputs.any_changed == 'true'
-        run: make type-check
+        run: make type-check-core

      - name: Dotenv check
        if: steps.changed-files.outputs.any_changed == 'true'
@@ -77,6 +77,10 @@ jobs:
        with:
          files: |
            web/**
+            package.json
+            pnpm-lock.yaml
+            pnpm-workspace.yaml
+            .nvmrc
            .github/workflows/style.yml
            .github/actions/setup-web/**

@@ -90,9 +94,9 @@ jobs:
        uses: actions/cache/restore@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
        with:
          path: web/.eslintcache
-          key: ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'web/pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
+          key: ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
          restore-keys: |
-            ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'web/pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-
+            ${{ runner.os }}-web-eslint-${{ hashFiles('web/package.json', 'pnpm-lock.yaml', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-

      - name: Web style check
        if: steps.changed-files.outputs.any_changed == 'true'
@@ -6,6 +6,9 @@ on:
      - main
    paths:
      - sdks/**
+      - package.json
+      - pnpm-lock.yaml
+      - pnpm-workspace.yaml

 concurrency:
  group: sdk-tests-${{ github.head_ref || github.run_id }}
@@ -1,26 +1,24 @@
 name: Translate i18n Files with Claude Code

 # Note: claude-code-action doesn't support push events directly.
-# Push events are handled by trigger-i18n-sync.yml which sends repository_dispatch.
-# See: https://github.com/langgenius/dify/issues/30743
-
+# Push events are bridged by trigger-i18n-sync.yml via repository_dispatch.
 on:
  repository_dispatch:
    types: [i18n-sync]
  workflow_dispatch:
    inputs:
      files:
-        description: 'Specific files to translate (space-separated, e.g., "app common"). Leave empty for all files.'
+        description: 'Specific files to translate (space-separated, e.g., "app common"). Required for full mode; leave empty in incremental mode to use en-US files changed since HEAD~1.'
        required: false
        type: string
      languages:
-        description: 'Specific languages to translate (space-separated, e.g., "zh-Hans ja-JP"). Leave empty for all supported languages.'
+        description: 'Specific languages to translate (space-separated, e.g., "zh-Hans ja-JP"). Leave empty for all supported target languages except en-US.'
        required: false
        type: string
      mode:
-        description: 'Sync mode: incremental (only changes) or full (re-check all keys)'
+        description: 'Sync mode: incremental (compare with previous en-US revision) or full (sync all keys in scope)'
        required: false
-        default: 'incremental'
+        default: incremental
        type: choice
        options:
          - incremental
@@ -30,11 +28,15 @@ permissions:
  contents: write
  pull-requests: write

+concurrency:
+  group: translate-i18n-${{ github.event_name }}-${{ github.ref }}
+  cancel-in-progress: false
+
 jobs:
  translate:
    if: github.repository == 'langgenius/dify'
    runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 120

    steps:
      - name: Checkout repository
@@ -51,380 +53,374 @@ jobs:
      - name: Setup web environment
        uses: ./.github/actions/setup-web

-      - name: Detect changed files and generate diff
-        id: detect_changes
+      - name: Prepare sync context
+        id: context
+        shell: bash
        run: |
-          if [ "${{ github.event_name }}" == "workflow_dispatch" ]; then
-            # Manual trigger
-            if [ -n "${{ github.event.inputs.files }}" ]; then
-              echo "CHANGED_FILES=${{ github.event.inputs.files }}" >> $GITHUB_OUTPUT
-            else
-              # Get all JSON files in en-US directory
-              files=$(ls web/i18n/en-US/*.json 2>/dev/null | xargs -n1 basename | sed 's/.json$//' | tr '\n' ' ')
-              echo "CHANGED_FILES=$files" >> $GITHUB_OUTPUT
-            fi
-            echo "TARGET_LANGS=${{ github.event.inputs.languages }}" >> $GITHUB_OUTPUT
-            echo "SYNC_MODE=${{ github.event.inputs.mode || 'incremental' }}" >> $GITHUB_OUTPUT
+          DEFAULT_TARGET_LANGS=$(awk "
+            /value: '/ {
+              value=\$2
+              gsub(/[',]/, \"\", value)
+            }
+            /supported: true/ && value != \"en-US\" {
+              printf \"%s \", value
+            }
+          " web/i18n-config/languages.ts | sed 's/[[:space:]]*$//')

-            # For manual trigger with incremental mode, get diff from last commit
-            # For full mode, we'll do a complete check anyway
-            if [ "${{ github.event.inputs.mode }}" == "full" ]; then
-              echo "Full mode: will check all keys" > /tmp/i18n-diff.txt
-              echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
-            else
-              git diff HEAD~1..HEAD -- 'web/i18n/en-US/*.json' > /tmp/i18n-diff.txt 2>/dev/null || echo "" > /tmp/i18n-diff.txt
-              if [ -s /tmp/i18n-diff.txt ]; then
-                echo "DIFF_AVAILABLE=true" >> $GITHUB_OUTPUT
-              else
-                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
-              fi
-            fi
-          elif [ "${{ github.event_name }}" == "repository_dispatch" ]; then
-            # Triggered by push via trigger-i18n-sync.yml workflow
-            # Validate required payload fields
-            if [ -z "${{ github.event.client_payload.changed_files }}" ]; then
-              echo "Error: repository_dispatch payload missing required 'changed_files' field" >&2
-              exit 1
-            fi
-            echo "CHANGED_FILES=${{ github.event.client_payload.changed_files }}" >> $GITHUB_OUTPUT
-            echo "TARGET_LANGS=" >> $GITHUB_OUTPUT
-            echo "SYNC_MODE=${{ github.event.client_payload.sync_mode || 'incremental' }}" >> $GITHUB_OUTPUT
+          generate_changes_json() {
+            node <<'NODE'
+            const { execFileSync } = require('node:child_process')
+            const fs = require('node:fs')
+            const path = require('node:path')

-            # Decode the base64-encoded diff from the trigger workflow
-            if [ -n "${{ github.event.client_payload.diff_base64 }}" ]; then
-              if ! echo "${{ github.event.client_payload.diff_base64 }}" | base64 -d > /tmp/i18n-diff.txt 2>&1; then
-                echo "Warning: Failed to decode base64 diff payload" >&2
-                echo "" > /tmp/i18n-diff.txt
-                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
-              elif [ -s /tmp/i18n-diff.txt ]; then
-                echo "DIFF_AVAILABLE=true" >> $GITHUB_OUTPUT
-              else
-                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
-              fi
+            const repoRoot = process.cwd()
+            const baseSha = process.env.BASE_SHA || ''
+            const headSha = process.env.HEAD_SHA || ''
+            const files = (process.env.CHANGED_FILES || '').split(/\s+/).filter(Boolean)
+
+            const englishPath = fileStem => path.join(repoRoot, 'web', 'i18n', 'en-US', `${fileStem}.json`)
+
+            const readCurrentJson = (fileStem) => {
+              const filePath = englishPath(fileStem)
+              if (!fs.existsSync(filePath))
+                return null
+
+              return JSON.parse(fs.readFileSync(filePath, 'utf8'))
+            }
+
+            const readBaseJson = (fileStem) => {
+              if (!baseSha)
+                return null
+
+              try {
+                const relativePath = `web/i18n/en-US/${fileStem}.json`
+                const content = execFileSync('git', ['show', `${baseSha}:${relativePath}`], { encoding: 'utf8' })
+                return JSON.parse(content)
+              }
+              catch (error) {
+                return null
+              }
+            }
+
+            const compareJson = (beforeValue, afterValue) => JSON.stringify(beforeValue) === JSON.stringify(afterValue)
+
+            const changes = {}
+
+            for (const fileStem of files) {
+              const currentJson = readCurrentJson(fileStem)
+              const beforeJson = readBaseJson(fileStem) || {}
+              const afterJson = currentJson || {}
+              const added = {}
+              const updated = {}
+              const deleted = []
+
+              for (const [key, value] of Object.entries(afterJson)) {
+                if (!(key in beforeJson)) {
+                  added[key] = value
+                  continue
+                }
+
+                if (!compareJson(beforeJson[key], value)) {
+                  updated[key] = {
+                    before: beforeJson[key],
+                    after: value,
+                  }
+                }
+              }
+
+              for (const key of Object.keys(beforeJson)) {
+                if (!(key in afterJson))
+                  deleted.push(key)
+              }
+
+              changes[fileStem] = {
+                fileDeleted: currentJson === null,
+                added,
+                updated,
+                deleted,
+              }
+            }
+
+            fs.writeFileSync(
+              '/tmp/i18n-changes.json',
+              JSON.stringify({
+                baseSha,
+                headSha,
+                files,
+                changes,
+              })
+            )
+            NODE
+          }
+
+          if [ "${{ github.event_name }}" = "repository_dispatch" ]; then
+            BASE_SHA="${{ github.event.client_payload.base_sha }}"
+            HEAD_SHA="${{ github.event.client_payload.head_sha }}"
+            CHANGED_FILES="${{ github.event.client_payload.changed_files }}"
+            TARGET_LANGS="$DEFAULT_TARGET_LANGS"
+            SYNC_MODE="${{ github.event.client_payload.sync_mode || 'incremental' }}"
+
+            if [ -n "${{ github.event.client_payload.changes_base64 }}" ]; then
+              printf '%s' '${{ github.event.client_payload.changes_base64 }}' | base64 -d > /tmp/i18n-changes.json
+              CHANGES_AVAILABLE="true"
+              CHANGES_SOURCE="embedded"
+            elif [ -n "$BASE_SHA" ] && [ -n "$CHANGED_FILES" ]; then
+              export BASE_SHA HEAD_SHA CHANGED_FILES
+              generate_changes_json
+              CHANGES_AVAILABLE="true"
+              CHANGES_SOURCE="recomputed"
            else
-              echo "" > /tmp/i18n-diff.txt
-              echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+              printf '%s' '{"baseSha":"","headSha":"","files":[],"changes":{}}' > /tmp/i18n-changes.json
+              CHANGES_AVAILABLE="false"
+              CHANGES_SOURCE="unavailable"
            fi
          else
-            echo "Unsupported event type: ${{ github.event_name }}"
-            exit 1
+            BASE_SHA=""
+            HEAD_SHA=$(git rev-parse HEAD)
+            if [ -n "${{ github.event.inputs.languages }}" ]; then
+              TARGET_LANGS="${{ github.event.inputs.languages }}"
+            else
+              TARGET_LANGS="$DEFAULT_TARGET_LANGS"
+            fi
+            SYNC_MODE="${{ github.event.inputs.mode || 'incremental' }}"
+            if [ -n "${{ github.event.inputs.files }}" ]; then
+              CHANGED_FILES="${{ github.event.inputs.files }}"
+            elif [ "$SYNC_MODE" = "incremental" ]; then
+              BASE_SHA=$(git rev-parse HEAD~1 2>/dev/null || true)
+              if [ -n "$BASE_SHA" ]; then
+                CHANGED_FILES=$(git diff --name-only "$BASE_SHA" "$HEAD_SHA" -- 'web/i18n/en-US/*.json' 2>/dev/null | sed -n 's@^.*/@@p' | sed 's/\.json$//' | tr '\n' ' ' | sed 's/[[:space:]]*$//')
+              else
+                CHANGED_FILES=$(find web/i18n/en-US -maxdepth 1 -type f -name '*.json' -print | sed -n 's@^.*/@@p' | sed 's/\.json$//' | sort | tr '\n' ' ' | sed 's/[[:space:]]*$//')
+              fi
+            elif [ "$SYNC_MODE" = "full" ]; then
+              echo "workflow_dispatch full mode requires the files input to stay within CI limits." >&2
+              exit 1
+            else
+              CHANGED_FILES=""
+            fi
+
+            if [ "$SYNC_MODE" = "incremental" ] && [ -n "$CHANGED_FILES" ]; then
+              export BASE_SHA HEAD_SHA CHANGED_FILES
+              generate_changes_json
+              CHANGES_AVAILABLE="true"
+              CHANGES_SOURCE="local"
+            else
+              printf '%s' '{"baseSha":"","headSha":"","files":[],"changes":{}}' > /tmp/i18n-changes.json
+              CHANGES_AVAILABLE="false"
+              CHANGES_SOURCE="unavailable"
+            fi
          fi

-          # Truncate diff if too large (keep first 50KB)
-          if [ -f /tmp/i18n-diff.txt ]; then
-            head -c 50000 /tmp/i18n-diff.txt > /tmp/i18n-diff-truncated.txt
-            mv /tmp/i18n-diff-truncated.txt /tmp/i18n-diff.txt
+          FILE_ARGS=""
+          if [ -n "$CHANGED_FILES" ]; then
+            FILE_ARGS="--file $CHANGED_FILES"
          fi

-          echo "Detected files: $(cat $GITHUB_OUTPUT | grep CHANGED_FILES || echo 'none')"
+          LANG_ARGS=""
+          if [ -n "$TARGET_LANGS" ]; then
+            LANG_ARGS="--lang $TARGET_LANGS"
+          fi
+
+          {
+            echo "DEFAULT_TARGET_LANGS=$DEFAULT_TARGET_LANGS"
+            echo "BASE_SHA=$BASE_SHA"
+            echo "HEAD_SHA=$HEAD_SHA"
+            echo "CHANGED_FILES=$CHANGED_FILES"
+            echo "TARGET_LANGS=$TARGET_LANGS"
+            echo "SYNC_MODE=$SYNC_MODE"
+            echo "CHANGES_AVAILABLE=$CHANGES_AVAILABLE"
+            echo "CHANGES_SOURCE=$CHANGES_SOURCE"
+            echo "FILE_ARGS=$FILE_ARGS"
+            echo "LANG_ARGS=$LANG_ARGS"
+          } >> "$GITHUB_OUTPUT"
+
+          echo "Files: ${CHANGED_FILES:-<none>}"
+          echo "Languages: ${TARGET_LANGS:-<none>}"
+          echo "Mode: $SYNC_MODE"

      - name: Run Claude Code for Translation Sync
-        if: steps.detect_changes.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@ff9acae5886d41a99ed4ec14b7dc147d55834722 # v1.0.77
+        if: steps.context.outputs.CHANGED_FILES != ''
+        uses: anthropics/claude-code-action@88c168b39e7e64da0286d812b6e9fbebb6708185 # v1.0.82
        with:
          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
          github_token: ${{ secrets.GITHUB_TOKEN }}
-          # Allow github-actions bot to trigger this workflow via repository_dispatch
-          # See: https://github.com/anthropics/claude-code-action/blob/main/docs/usage.md
          allowed_bots: 'github-actions[bot]'
+          show_full_output: ${{ github.event_name == 'workflow_dispatch' }}
          prompt: |
-            You are a professional i18n synchronization engineer for the Dify project.
-            Your task is to keep all language translations in sync with the English source (en-US).
+            You are the i18n sync agent for the Dify repository.
+            Your job is to keep translations synchronized with the English source files under `${{ github.workspace }}/web/i18n/en-US/`.

-            ## CRITICAL TOOL RESTRICTIONS
-            - Use **Read** tool to read files (NOT cat or bash)
-            - Use **Edit** tool to modify JSON files (NOT node, jq, or bash scripts)
-            - Use **Bash** ONLY for: git commands, gh commands, pnpm commands
-            - Run bash commands ONE BY ONE, never combine with && or ||
-            - NEVER use `$()` command substitution - it's not supported. Split into separate commands instead.
+            Use absolute paths at all times:
+            - Repo root: `${{ github.workspace }}`
+            - Web directory: `${{ github.workspace }}/web`
+            - Language config: `${{ github.workspace }}/web/i18n-config/languages.ts`

-            ## WORKING DIRECTORY & ABSOLUTE PATHS
-            Claude Code sandbox working directory may vary. Always use absolute paths:
-            - For pnpm: `pnpm --dir ${{ github.workspace }}/web <command>`
-            - For git: `git -C ${{ github.workspace }} <command>`
-            - For gh: `gh --repo ${{ github.repository }} <command>`
-            - For file paths: `${{ github.workspace }}/web/i18n/`
+            Inputs:
+            - Files in scope: `${{ steps.context.outputs.CHANGED_FILES }}`
+            - Target languages: `${{ steps.context.outputs.TARGET_LANGS }}`
+            - Sync mode: `${{ steps.context.outputs.SYNC_MODE }}`
+            - Base SHA: `${{ steps.context.outputs.BASE_SHA }}`
+            - Head SHA: `${{ steps.context.outputs.HEAD_SHA }}`
+            - Scoped file args: `${{ steps.context.outputs.FILE_ARGS }}`
+            - Scoped language args: `${{ steps.context.outputs.LANG_ARGS }}`
+            - Structured change set available: `${{ steps.context.outputs.CHANGES_AVAILABLE }}`
+            - Structured change set source: `${{ steps.context.outputs.CHANGES_SOURCE }}`
+            - Structured change set file: `/tmp/i18n-changes.json`

-            ## EFFICIENCY RULES
-            - **ONE Edit per language file** - batch all key additions into a single Edit
-            - Insert new keys at the beginning of JSON (after `{`), lint:fix will sort them
-            - Translate ALL keys for a language mentally first, then do ONE Edit
-
-            ## Context
-            - Changed/target files: ${{ steps.detect_changes.outputs.CHANGED_FILES }}
-            - Target languages (empty means all supported): ${{ steps.detect_changes.outputs.TARGET_LANGS }}
-            - Sync mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}
-            - Translation files are located in: ${{ github.workspace }}/web/i18n/{locale}/{filename}.json
-            - Language configuration is in: ${{ github.workspace }}/web/i18n-config/languages.ts
-            - Git diff is available: ${{ steps.detect_changes.outputs.DIFF_AVAILABLE }}
-
-            ## CRITICAL DESIGN: Verify First, Then Sync
-
-            You MUST follow this three-phase approach:
-
-            ═══════════════════════════════════════════════════════════════
-            ║  PHASE 1: VERIFY - Analyze and Generate Change Report       ║
-            ═══════════════════════════════════════════════════════════════
-
-            ### Step 1.1: Analyze Git Diff (for incremental mode)
-            Use the Read tool to read `/tmp/i18n-diff.txt` to see the git diff.
-
-            Parse the diff to categorize changes:
-            - Lines with `+` (not `+++`): Added or modified values
-            - Lines with `-` (not `---`): Removed or old values
-            - Identify specific keys for each category:
-              * ADD: Keys that appear only in `+` lines (new keys)
-              * UPDATE: Keys that appear in both `-` and `+` lines (value changed)
-              * DELETE: Keys that appear only in `-` lines (removed keys)
-
-            ### Step 1.2: Read Language Configuration
-            Use the Read tool to read `${{ github.workspace }}/web/i18n-config/languages.ts`.
-            Extract all languages with `supported: true`.
-
-            ### Step 1.3: Run i18n:check for Each Language
-            ```bash
-            pnpm --dir ${{ github.workspace }}/web install --frozen-lockfile
-            ```
-            ```bash
-            pnpm --dir ${{ github.workspace }}/web run i18n:check
-            ```
-
-            This will report:
-            - Missing keys (need to ADD)
-            - Extra keys (need to DELETE)
-
-            ### Step 1.4: Generate Change Report
-
-            Create a structured report identifying:
-            ```
-            ╔══════════════════════════════════════════════════════════════╗
-            ║                    I18N SYNC CHANGE REPORT                   ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ Files to process: [list]                                     ║
-            ║ Languages to sync: [list]                                    ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ ADD (New Keys):                                              ║
-            ║   - [filename].[key]: "English value"                        ║
-            ║   ...                                                        ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ UPDATE (Modified Keys - MUST re-translate):                  ║
-            ║   - [filename].[key]: "Old value" → "New value"              ║
-            ║   ...                                                        ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ DELETE (Extra Keys):                                         ║
-            ║   - [language]/[filename].[key]                              ║
-            ║   ...                                                        ║
-            ╚══════════════════════════════════════════════════════════════╝
-            ```
-
-            **IMPORTANT**: For UPDATE detection, compare git diff to find keys where
-            the English value changed. These MUST be re-translated even if target
-            language already has a translation (it's now stale!).
-
-            ═══════════════════════════════════════════════════════════════
-            ║  PHASE 2: SYNC - Execute Changes Based on Report            ║
-            ═══════════════════════════════════════════════════════════════
-
-            ### Step 2.1: Process ADD Operations (BATCH per language file)
-
-            **CRITICAL WORKFLOW for efficiency:**
-            1. First, translate ALL new keys for ALL languages mentally
-            2. Then, for EACH language file, do ONE Edit operation:
-               - Read the file once
-               - Insert ALL new keys at the beginning (right after the opening `{`)
-               - Don't worry about alphabetical order - lint:fix will sort them later
-
-            Example Edit (adding 3 keys to zh-Hans/app.json):
-            ```
-            old_string: '{\n  "accessControl"'
-            new_string: '{\n  "newKey1": "translation1",\n  "newKey2": "translation2",\n  "newKey3": "translation3",\n  "accessControl"'
-            ```
-
-            **IMPORTANT**:
-            - ONE Edit per language file (not one Edit per key!)
-            - Always use the Edit tool. NEVER use bash scripts, node, or jq.
-
-            ### Step 2.2: Process UPDATE Operations
-
-            **IMPORTANT: Special handling for zh-Hans and ja-JP**
-            If zh-Hans or ja-JP files were ALSO modified in the same push:
-            - Run: `git -C ${{ github.workspace }} diff HEAD~1 --name-only` and check for zh-Hans or ja-JP files
-            - If found, it means someone manually translated them. Apply these rules:
-
-            1. **Missing keys**: Still ADD them (completeness required)
-            2. **Existing translations**: Compare with the NEW English value:
-               - If translation is **completely wrong** or **unrelated** → Update it
-               - If translation is **roughly correct** (captures the meaning) → Keep it, respect manual work
-               - When in doubt, **keep the manual translation**
-
-            Example:
-            - English changed: "Save" → "Save Changes"
-            - Manual translation: "保存更改" → Keep it (correct meaning)
-            - Manual translation: "删除" → Update it (completely wrong)
-
-            For other languages:
-            Use Edit tool to replace the old value with the new translation.
-            You can batch multiple updates in one Edit if they are adjacent.
-
-            ### Step 2.3: Process DELETE Operations
-            For extra keys reported by i18n:check:
-            - Run: `pnpm --dir ${{ github.workspace }}/web run i18n:check --auto-remove`
-            - Or manually remove from target language JSON files
-
-            ## Translation Guidelines
-
-            - PRESERVE all placeholders exactly as-is:
-              - `{{variable}}` - Mustache interpolation
-              - `${variable}` - Template literal
-              - `<tag>content</tag>` - HTML tags
-              - `_one`, `_other` - Pluralization suffixes (these are KEY suffixes, not values)
-
-              **CRITICAL: Variable names and tag names MUST stay in English - NEVER translate them**
-
-              ✅ CORRECT examples:
-              - English: "{{count}} items" → Japanese: "{{count}} 個のアイテム"
-              - English: "{{name}} updated" → Korean: "{{name}} 업데이트됨"
-              - English: "<email>{{email}}</email>" → Chinese: "<email>{{email}}</email>"
-              - English: "<CustomLink>Marketplace</CustomLink>" → Japanese: "<CustomLink>マーケットプレイス</CustomLink>"
-
-              ❌ WRONG examples (NEVER do this - will break the application):
-              - "{{count}}" → "{{カウント}}" ❌ (variable name translated to Japanese)
-              - "{{name}}" → "{{이름}}" ❌ (variable name translated to Korean)
-              - "{{email}}" → "{{邮箱}}" ❌ (variable name translated to Chinese)
-              - "<email>" → "<メール>" ❌ (tag name translated)
-              - "<CustomLink>" → "<自定义链接>" ❌ (component name translated)
-
-            - Use appropriate language register (formal/informal) based on existing translations
-            - Match existing translation style in each language
-            - Technical terms: check existing conventions per language
-            - For CJK languages: no spaces between characters unless necessary
-            - For RTL languages (ar-TN, fa-IR): ensure proper text handling
-
-            ## Output Format Requirements
-            - Alphabetical key ordering (if original file uses it)
-            - 2-space indentation
-            - Trailing newline at end of file
-            - Valid JSON (use proper escaping for special characters)
-
-            ═══════════════════════════════════════════════════════════════
-            ║  PHASE 3: RE-VERIFY - Confirm All Issues Resolved           ║
-            ═══════════════════════════════════════════════════════════════
-
-            ### Step 3.1: Run Lint Fix (IMPORTANT!)
-            ```bash
-            pnpm --dir ${{ github.workspace }}/web lint:fix --quiet -- 'i18n/**/*.json'
-            ```
-            This ensures:
-            - JSON keys are sorted alphabetically (jsonc/sort-keys rule)
-            - Valid i18n keys (dify-i18n/valid-i18n-keys rule)
-            - No extra keys (dify-i18n/no-extra-keys rule)
-
-            ### Step 3.2: Run Final i18n Check
-            ```bash
-            pnpm --dir ${{ github.workspace }}/web run i18n:check
-            ```
-
-            ### Step 3.3: Fix Any Remaining Issues
-            If check reports issues:
-            - Go back to PHASE 2 for unresolved items
-            - Repeat until check passes
-
-            ### Step 3.4: Generate Final Summary
-            ```
-            ╔══════════════════════════════════════════════════════════════╗
-            ║                    SYNC COMPLETED SUMMARY                    ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ Language      │ Added │ Updated │ Deleted │ Status          ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ zh-Hans       │  5    │   2     │    1    │ ✓ Complete      ║
-            ║ ja-JP         │  5    │   2     │    1    │ ✓ Complete      ║
-            ║ ...           │ ...   │  ...    │   ...   │ ...             ║
-            ╠══════════════════════════════════════════════════════════════╣
-            ║ i18n:check    │ PASSED - All keys in sync                   ║
-            ╚══════════════════════════════════════════════════════════════╝
-            ```
-
-            ## Mode-Specific Behavior
-
-            **SYNC_MODE = "incremental"** (default):
-            - Focus on keys identified from git diff
-            - Also check i18n:check output for any missing/extra keys
-            - Efficient for small changes
-
-            **SYNC_MODE = "full"**:
-            - Compare ALL keys between en-US and each language
-            - Run i18n:check to identify all discrepancies
-            - Use for first-time sync or fixing historical issues
-
-            ## Important Notes
-
-            1. Always run i18n:check BEFORE and AFTER making changes
-            2. The check script is the source of truth for missing/extra keys
-            3. For UPDATE scenario: git diff is the source of truth for changed values
-            4. Create a single commit with all translation changes
-            5. If any translation fails, continue with others and report failures
-
-            ═══════════════════════════════════════════════════════════════
-            ║  PHASE 4: COMMIT AND CREATE PR                              ║
-            ═══════════════════════════════════════════════════════════════
-
-            After all translations are complete and verified:
-
-            ### Step 4.1: Check for changes
-            ```bash
-            git -C ${{ github.workspace }} status --porcelain
-            ```
-
-            If there are changes:
-
-            ### Step 4.2: Create a new branch and commit
-            Run these git commands ONE BY ONE (not combined with &&).
-            **IMPORTANT**: Do NOT use `$()` command substitution. Use two separate commands:
-
-            1. First, get the timestamp:
-            ```bash
-            date +%Y%m%d-%H%M%S
-            ```
-            (Note the output, e.g., "20260115-143052")
-
-            2. Then create branch using the timestamp value:
-            ```bash
-            git -C ${{ github.workspace }} checkout -b chore/i18n-sync-20260115-143052
-            ```
-            (Replace "20260115-143052" with the actual timestamp from step 1)
-
-            3. Stage changes:
-            ```bash
-            git -C ${{ github.workspace }} add web/i18n/
-            ```
-
-            4. Commit:
-            ```bash
-            git -C ${{ github.workspace }} commit -m "chore(i18n): sync translations with en-US - Mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}"
-            ```
-
-            5. Push:
-            ```bash
-            git -C ${{ github.workspace }} push origin HEAD
-            ```
-
-            ### Step 4.3: Create Pull Request
-            ```bash
-            gh pr create --repo ${{ github.repository }} --title "chore(i18n): sync translations with en-US" --body "## Summary
-
-            This PR was automatically generated to sync i18n translation files.
-
-            ### Changes
-            - Mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}
-            - Files processed: ${{ steps.detect_changes.outputs.CHANGED_FILES }}
-
-            ### Verification
-            - [x] \`i18n:check\` passed
-            - [x] \`lint:fix\` applied
-
-            🤖 Generated with Claude Code GitHub Action" --base main
-            ```
+            Tool rules:
+            - Use Read for repository files.
+            - Use Edit for JSON updates.
+            - Use Bash only for `pnpm`.
+            - Do not use Bash for `git`, `gh`, or branch management.

+            Required execution plan:
+            1. Resolve target languages.
+               - Use the provided `Target languages` value as the source of truth.
+               - If it is unexpectedly empty, read `${{ github.workspace }}/web/i18n-config/languages.ts` and use every language with `supported: true` except `en-US`.
+            2. Stay strictly in scope.
+               - Only process the files listed in `Files in scope`.
+               - Only process the resolved target languages, never `en-US`.
+               - Do not touch unrelated i18n files.
+               - Do not modify `${{ github.workspace }}/web/i18n/en-US/`.
+            3. Resolve source changes.
+               - If `Structured change set available` is `true`, read `/tmp/i18n-changes.json` and use it as the source of truth for file-level and key-level changes.
+               - For each file entry:
+                 - `added` contains new English keys that need translations.
+                 - `updated` contains stale keys whose English source changed; re-translate using the `after` value.
+                 - `deleted` contains keys that should be removed from locale files.
+                 - `fileDeleted: true` means the English file no longer exists; remove the matching locale file if present.
+               - Read the current English JSON file for any file that still exists so wording, placeholders, and surrounding terminology stay accurate.
+               - If `Structured change set available` is `false`, treat this as a scoped full sync and use the current English files plus scoped checks as the source of truth.
+            4. Run a scoped pre-check before editing:
+               - `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
+               - Use this command as the source of truth for missing and extra keys inside the current scope.
+            5. Apply translations.
+               - For every target language and scoped file:
+                 - If `fileDeleted` is `true`, remove the locale file if it exists and skip the rest of that file.
+                 - If the locale file does not exist yet, create it with `Write` and then continue with `Edit` as needed.
+                 - ADD missing keys.
+                 - UPDATE stale translations when the English value changed.
+                 - DELETE removed keys. Prefer `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }} --auto-remove` for extra keys so deletions stay in scope.
+               - Preserve placeholders exactly: `{{variable}}`, `${variable}`, HTML tags, component tags, and variable names.
+               - Match the existing terminology and register used by each locale.
+               - Prefer one Edit per file when stable, but prioritize correctness over batching.
+            6. Verify only the edited files.
+               - Run `pnpm --dir ${{ github.workspace }}/web lint:fix --quiet -- <relative edited i18n file paths>`
+               - Run `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
+               - If verification fails, fix the remaining problems before continuing.
+            7. Stop after the scoped locale files are updated and verification passes.
+               - Do not create branches, commits, or pull requests.
          claude_args: |
-            --max-turns 150
-            --allowedTools "Read,Write,Edit,Bash(git *),Bash(git:*),Bash(gh *),Bash(gh:*),Bash(pnpm *),Bash(pnpm:*),Bash(date *),Bash(date:*),Glob,Grep"
+            --max-turns 120
+            --allowedTools "Read,Write,Edit,Bash(pnpm *),Bash(pnpm:*),Glob,Grep"
+
+      - name: Prepare branch metadata
+        id: pr_meta
+        if: steps.context.outputs.CHANGED_FILES != ''
+        shell: bash
+        run: |
+          if [ -z "$(git -C "${{ github.workspace }}" status --porcelain -- web/i18n/)" ]; then
+            echo "has_changes=false" >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+
+          SCOPE_HASH=$(printf '%s|%s|%s' "${{ steps.context.outputs.CHANGED_FILES }}" "${{ steps.context.outputs.TARGET_LANGS }}" "${{ steps.context.outputs.SYNC_MODE }}" | sha256sum | cut -c1-8)
+          HEAD_SHORT=$(printf '%s' "${{ steps.context.outputs.HEAD_SHA }}" | cut -c1-12)
+          BRANCH_NAME="chore/i18n-sync-${HEAD_SHORT}-${SCOPE_HASH}"
+
+          {
+            echo "has_changes=true"
+            echo "branch_name=$BRANCH_NAME"
+          } >> "$GITHUB_OUTPUT"
+
+      - name: Commit translation changes
+        if: steps.pr_meta.outputs.has_changes == 'true'
+        shell: bash
+        run: |
+          git -C "${{ github.workspace }}" checkout -B "${{ steps.pr_meta.outputs.branch_name }}"
+          git -C "${{ github.workspace }}" add web/i18n/
+          git -C "${{ github.workspace }}" commit -m "chore(i18n): sync translations with en-US"
+
+      - name: Push translation branch
+        if: steps.pr_meta.outputs.has_changes == 'true'
+        shell: bash
+        run: |
+          if git -C "${{ github.workspace }}" ls-remote --exit-code --heads origin "${{ steps.pr_meta.outputs.branch_name }}" >/dev/null 2>&1; then
+            git -C "${{ github.workspace }}" push --force-with-lease origin "${{ steps.pr_meta.outputs.branch_name }}"
+          else
+            git -C "${{ github.workspace }}" push --set-upstream origin "${{ steps.pr_meta.outputs.branch_name }}"
+          fi
+
+      - name: Create or update translation PR
+        if: steps.pr_meta.outputs.has_changes == 'true'
+        env:
+          BRANCH_NAME: ${{ steps.pr_meta.outputs.branch_name }}
+          FILES_IN_SCOPE: ${{ steps.context.outputs.CHANGED_FILES }}
+          TARGET_LANGS: ${{ steps.context.outputs.TARGET_LANGS }}
+          SYNC_MODE: ${{ steps.context.outputs.SYNC_MODE }}
+          CHANGES_SOURCE: ${{ steps.context.outputs.CHANGES_SOURCE }}
+          BASE_SHA: ${{ steps.context.outputs.BASE_SHA }}
+          HEAD_SHA: ${{ steps.context.outputs.HEAD_SHA }}
+          REPO_NAME: ${{ github.repository }}
+        shell: bash
+        run: |
+          PR_BODY_FILE=/tmp/i18n-pr-body.md
+          LANG_COUNT=$(printf '%s\n' "$TARGET_LANGS" | wc -w | tr -d ' ')
+          if [ "$LANG_COUNT" = "0" ]; then
+            LANG_COUNT="0"
+          fi
+          export LANG_COUNT
+
+          node <<'NODE' > "$PR_BODY_FILE"
+          const fs = require('node:fs')
+
+          const changesPath = '/tmp/i18n-changes.json'
+          const changes = fs.existsSync(changesPath)
+            ? JSON.parse(fs.readFileSync(changesPath, 'utf8'))
+            : { changes: {} }
+
+          const filesInScope = (process.env.FILES_IN_SCOPE || '').split(/\s+/).filter(Boolean)
+          const lines = [
+            '## Summary',
+            '',
+            `- **Files synced**: \`${process.env.FILES_IN_SCOPE || '<none>'}\``,
+            `- **Languages updated**: ${process.env.TARGET_LANGS || '<none>'} (${process.env.LANG_COUNT} languages)`,
+            `- **Sync mode**: ${process.env.SYNC_MODE}${process.env.BASE_SHA ? ` (base: \`${process.env.BASE_SHA.slice(0, 10)}\`, head: \`${process.env.HEAD_SHA.slice(0, 10)}\`)` : ` (head: \`${process.env.HEAD_SHA.slice(0, 10)}\`)`}`,
+            '',
+            '### Key changes',
+          ]
+
+          for (const fileName of filesInScope) {
+            const fileChange = changes.changes?.[fileName] || { added: {}, updated: {}, deleted: [], fileDeleted: false }
+            const addedKeys = Object.keys(fileChange.added || {})
+            const updatedKeys = Object.keys(fileChange.updated || {})
+            const deletedKeys = fileChange.deleted || []
+            lines.push(`- \`${fileName}\`: +${addedKeys.length} / ~${updatedKeys.length} / -${deletedKeys.length}${fileChange.fileDeleted ? ' (file deleted in en-US)' : ''}`)
+          }
+
+          lines.push(
+            '',
+            '## Verification',
+            '',
+            `- \`pnpm --dir web run i18n:check --file ${process.env.FILES_IN_SCOPE} --lang ${process.env.TARGET_LANGS}\``,
+            `- \`pnpm --dir web lint:fix --quiet -- <edited i18n files>\``,
+            '',
+            '## Notes',
+            '',
+            '- This PR was generated from structured en-US key changes produced by `trigger-i18n-sync.yml`.',
+            `- Structured change source: ${process.env.CHANGES_SOURCE || 'unknown'}.`,
+            '- Branch name is deterministic for the head SHA and scope, so reruns update the same PR instead of opening duplicates.',
+            '',
+            '🤖 Generated with [Claude Code](https://claude.com/claude-code)'
+          )
+
+          process.stdout.write(lines.join('\n'))
+          NODE
+
+          EXISTING_PR_NUMBER=$(gh pr list --repo "$REPO_NAME" --head "$BRANCH_NAME" --state open --json number --jq '.[0].number')
+
+          if [ -n "$EXISTING_PR_NUMBER" ] && [ "$EXISTING_PR_NUMBER" != "null" ]; then
+            gh pr edit "$EXISTING_PR_NUMBER" --repo "$REPO_NAME" --title "chore(i18n): sync translations with en-US" --body-file "$PR_BODY_FILE"
+          else
+            gh pr create --repo "$REPO_NAME" --head "$BRANCH_NAME" --base main --title "chore(i18n): sync translations with en-US" --body-file "$PR_BODY_FILE"
+          fi
@@ -1,9 +1,5 @@
 name: Trigger i18n Sync on Push

-# This workflow bridges the push event to repository_dispatch
-# because claude-code-action doesn't support push events directly.
-# See: https://github.com/langgenius/dify/issues/30743
-
 on:
  push:
    branches: [main]
@@ -13,6 +9,10 @@ on:
 permissions:
  contents: write

+concurrency:
+  group: trigger-i18n-sync-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
  trigger:
    if: github.repository == 'langgenius/dify'
@@ -25,42 +25,147 @@ jobs:
        with:
          fetch-depth: 0

-      - name: Detect changed files and generate diff
+      - name: Detect changed files and build structured change set
        id: detect
+        shell: bash
        run: |
-          BEFORE_SHA="${{ github.event.before }}"
-          # Handle edge case: force push may have null/zero SHA
-          if [ -z "$BEFORE_SHA" ] || [ "$BEFORE_SHA" = "0000000000000000000000000000000000000000" ]; then
-            BEFORE_SHA="HEAD~1"
+          BASE_SHA="${{ github.event.before }}"
+          if [ -z "$BASE_SHA" ] || [ "$BASE_SHA" = "0000000000000000000000000000000000000000" ]; then
+            BASE_SHA=$(git rev-parse HEAD~1 2>/dev/null || true)
          fi
+          HEAD_SHA="${{ github.sha }}"

-          # Detect changed i18n files
-          changed=$(git diff --name-only "$BEFORE_SHA" "${{ github.sha }}" -- 'web/i18n/en-US/*.json' 2>/dev/null | xargs -n1 basename 2>/dev/null | sed 's/.json$//' | tr '\n' ' ' || echo "")
-          echo "changed_files=$changed" >> $GITHUB_OUTPUT
-
-          # Generate diff for context
-          git diff "$BEFORE_SHA" "${{ github.sha }}" -- 'web/i18n/en-US/*.json' > /tmp/i18n-diff.txt 2>/dev/null || echo "" > /tmp/i18n-diff.txt
-
-          # Truncate if too large (keep first 50KB to match receiving workflow)
-          head -c 50000 /tmp/i18n-diff.txt > /tmp/i18n-diff-truncated.txt
-          mv /tmp/i18n-diff-truncated.txt /tmp/i18n-diff.txt
-
-          # Base64 encode the diff for safe JSON transport (portable, single-line)
-          diff_base64=$(base64 < /tmp/i18n-diff.txt | tr -d '\n')
-          echo "diff_base64=$diff_base64" >> $GITHUB_OUTPUT
-
-          if [ -n "$changed" ]; then
-            echo "has_changes=true" >> $GITHUB_OUTPUT
-            echo "Detected changed files: $changed"
+          if [ -n "$BASE_SHA" ]; then
+            CHANGED_FILES=$(git diff --name-only "$BASE_SHA" "$HEAD_SHA" -- 'web/i18n/en-US/*.json' 2>/dev/null | sed -n 's@^.*/@@p' | sed 's/\.json$//' | tr '\n' ' ' | sed 's/[[:space:]]*$//')
          else
-            echo "has_changes=false" >> $GITHUB_OUTPUT
-            echo "No i18n changes detected"
+            CHANGED_FILES=$(find web/i18n/en-US -maxdepth 1 -type f -name '*.json' -print | sed -n 's@^.*/@@p' | sed 's/\.json$//' | sort | tr '\n' ' ' | sed 's/[[:space:]]*$//')
          fi

+          export BASE_SHA HEAD_SHA CHANGED_FILES
+          node <<'NODE'
+          const { execFileSync } = require('node:child_process')
+          const fs = require('node:fs')
+          const path = require('node:path')
+
+          const repoRoot = process.cwd()
+          const baseSha = process.env.BASE_SHA || ''
+          const headSha = process.env.HEAD_SHA || ''
+          const files = (process.env.CHANGED_FILES || '').split(/\s+/).filter(Boolean)
+
+          const englishPath = fileStem => path.join(repoRoot, 'web', 'i18n', 'en-US', `${fileStem}.json`)
+
+          const readCurrentJson = (fileStem) => {
+            const filePath = englishPath(fileStem)
+            if (!fs.existsSync(filePath))
+              return null
+
+            return JSON.parse(fs.readFileSync(filePath, 'utf8'))
+          }
+
+          const readBaseJson = (fileStem) => {
+            if (!baseSha)
+              return null
+
+            try {
+              const relativePath = `web/i18n/en-US/${fileStem}.json`
+              const content = execFileSync('git', ['show', `${baseSha}:${relativePath}`], { encoding: 'utf8' })
+              return JSON.parse(content)
+            }
+            catch (error) {
+              return null
+            }
+          }
+
+          const compareJson = (beforeValue, afterValue) => JSON.stringify(beforeValue) === JSON.stringify(afterValue)
+
+          const changes = {}
+
+          for (const fileStem of files) {
+            const beforeJson = readBaseJson(fileStem) || {}
+            const afterJson = readCurrentJson(fileStem) || {}
+            const added = {}
+            const updated = {}
+            const deleted = []
+
+            for (const [key, value] of Object.entries(afterJson)) {
+              if (!(key in beforeJson)) {
+                added[key] = value
+                continue
+              }
+
+              if (!compareJson(beforeJson[key], value)) {
+                updated[key] = {
+                  before: beforeJson[key],
+                  after: value,
+                }
+              }
+            }
+
+            for (const key of Object.keys(beforeJson)) {
+              if (!(key in afterJson))
+                deleted.push(key)
+            }
+
+            changes[fileStem] = {
+              fileDeleted: readCurrentJson(fileStem) === null,
+              added,
+              updated,
+              deleted,
+            }
+          }
+
+          fs.writeFileSync(
+            '/tmp/i18n-changes.json',
+            JSON.stringify({
+              baseSha,
+              headSha,
+              files,
+              changes,
+            })
+          )
+          NODE
+
+          if [ -n "$CHANGED_FILES" ]; then
+            echo "has_changes=true" >> "$GITHUB_OUTPUT"
+          else
+            echo "has_changes=false" >> "$GITHUB_OUTPUT"
+          fi
+
+          echo "base_sha=$BASE_SHA" >> "$GITHUB_OUTPUT"
+          echo "head_sha=$HEAD_SHA" >> "$GITHUB_OUTPUT"
+          echo "changed_files=$CHANGED_FILES" >> "$GITHUB_OUTPUT"
+
      - name: Trigger i18n sync workflow
        if: steps.detect.outputs.has_changes == 'true'
-        uses: peter-evans/repository-dispatch@28959ce8df70de7be546dd1250a005dd32156697 # v4.0.1
+        uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0
+        env:
+          BASE_SHA: ${{ steps.detect.outputs.base_sha }}
+          HEAD_SHA: ${{ steps.detect.outputs.head_sha }}
+          CHANGED_FILES: ${{ steps.detect.outputs.changed_files }}
        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-          event-type: i18n-sync
-          client-payload: '{"changed_files": "${{ steps.detect.outputs.changed_files }}", "diff_base64": "${{ steps.detect.outputs.diff_base64 }}", "sync_mode": "incremental", "trigger_sha": "${{ github.sha }}"}'
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const fs = require('fs')
+
+            const changesJson = fs.readFileSync('/tmp/i18n-changes.json', 'utf8')
+            const changesBase64 = Buffer.from(changesJson).toString('base64')
+            const maxEmbeddedChangesChars = 48000
+            const changesEmbedded = changesBase64.length <= maxEmbeddedChangesChars
+
+            if (!changesEmbedded) {
+              console.log(`Structured change set too large to embed safely (${changesBase64.length} chars). Downstream workflow will regenerate it from git history.`)
+            }
+
+            await github.rest.repos.createDispatchEvent({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              event_type: 'i18n-sync',
+              client_payload: {
+                changed_files: process.env.CHANGED_FILES,
+                changes_base64: changesEmbedded ? changesBase64 : '',
+                changes_embedded: changesEmbedded,
+                sync_mode: 'incremental',
+                base_sha: process.env.BASE_SHA,
+                head_sha: process.env.HEAD_SHA,
+              },
+            })
@@ -0,0 +1,95 @@
+name: Run Full VDB Tests
+
+on:
+  schedule:
+    - cron: '0 3 * * 1'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: vdb-tests-full-${{ github.ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  test:
+    name: Full VDB Tests
+    if: github.repository == 'langgenius/dify'
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version:
+          - "3.12"
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Free Disk Space
+        uses: endersonmenezes/free-disk-space@7901478139cff6e9d44df5972fd8ab8fcade4db1 # v3.2.2
+        with:
+          remove_dotnet: true
+          remove_haskell: true
+          remove_tool_cache: true
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        with:
+          enable-cache: true
+          python-version: ${{ matrix.python-version }}
+          cache-dependency-glob: api/uv.lock
+
+      - name: Check UV lockfile
+        run: uv lock --project api --check
+
+      - name: Install dependencies
+        run: uv sync --project api --dev
+
+      - name: Set up dotenvs
+        run: |
+          cp docker/.env.example docker/.env
+          cp docker/middleware.env.example docker/middleware.env
+
+      - name: Expose Service Ports
+        run: sh .github/workflows/expose_service_ports.sh
+
+#      - name: Set up Vector Store (TiDB)
+#        uses: hoverkraft-tech/[email protected]
+#        with:
+#          compose-file: docker/tidb/docker-compose.yaml
+#          services: |
+#            tidb
+#            tiflash
+
+      - name: Set up Full Vector Store Matrix
+        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
+        with:
+          compose-file: |
+            docker/docker-compose.yaml
+          services: |
+            weaviate
+            qdrant
+            couchbase-server
+            etcd
+            minio
+            milvus-standalone
+            pgvecto-rs
+            pgvector
+            chroma
+            elasticsearch
+            oceanbase
+
+      - name: setup test config
+        run: |
+          echo $(pwd)
+          ls -lah .
+          cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
+
+#      - name: Check VDB Ready (TiDB)
+#        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
+
+      - name: Test Vector Stores
+        run: uv run --project api bash dev/pytest/pytest_vdb.sh
@@ -1,15 +1,18 @@
-name: Run VDB Tests
+name: Run VDB Smoke Tests

 on:
  workflow_call:

+permissions:
+  contents: read
+
 concurrency:
  group: vdb-tests-${{ github.head_ref || github.run_id }}
  cancel-in-progress: true

 jobs:
  test:
-    name: VDB Tests
+    name: VDB Smoke Tests
    runs-on: ubuntu-latest
    strategy:
      matrix:
@@ -30,7 +33,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -58,23 +61,18 @@ jobs:
 #            tidb
 #            tiflash

-      - name: Set up Vector Stores (Weaviate, Qdrant, PGVector, Milvus, PgVecto-RS, Chroma, MyScale, ElasticSearch, Couchbase, OceanBase)
+      - name: Set up Vector Stores for Smoke Coverage
        uses: hoverkraft-tech/compose-action@4894d2492015c1774ee5a13a95b1072093087ec3 # v2.5.0
        with:
          compose-file: |
            docker/docker-compose.yaml
          services: |
+            db_postgres
+            redis
            weaviate
            qdrant
-            couchbase-server
-            etcd
-            minio
-            milvus-standalone
-            pgvecto-rs
            pgvector
            chroma
-            elasticsearch
-            oceanbase

      - name: setup test config
        run: |
@@ -86,4 +84,9 @@ jobs:
 #        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py

      - name: Test Vector Stores
-        run: uv run --project api bash dev/pytest/pytest_vdb.sh
+        run: |
+          uv run --project api pytest --timeout "${PYTEST_TIMEOUT:-180}" \
+            api/tests/integration_tests/vdb/chroma \
+            api/tests/integration_tests/vdb/pgvector \
+            api/tests/integration_tests/vdb/qdrant \
+            api/tests/integration_tests/vdb/weaviate
@@ -0,0 +1,68 @@
+name: Web Full-Stack E2E
+
+on:
+  workflow_call:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: web-e2e-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  test:
+    name: Web Full-Stack E2E
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+
+      - name: Setup web dependencies
+        uses: ./.github/actions/setup-web
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8.0.0
+        with:
+          enable-cache: true
+          python-version: "3.12"
+          cache-dependency-glob: api/uv.lock
+
+      - name: Install API dependencies
+        run: uv sync --project api --dev
+
+      - name: Install Playwright browser
+        working-directory: ./e2e
+        run: vp run e2e:install
+
+      - name: Run isolated source-api and built-web Cucumber E2E tests
+        working-directory: ./e2e
+        env:
+          E2E_ADMIN_EMAIL: [email protected]
+          E2E_ADMIN_NAME: E2E Admin
+          E2E_ADMIN_PASSWORD: E2eAdmin12345
+          E2E_FORCE_WEB_BUILD: "1"
+          E2E_INIT_PASSWORD: E2eInit12345
+        run: vp run e2e:full
+
+      - name: Upload Cucumber report
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+        with:
+          name: cucumber-report
+          path: e2e/cucumber-report
+          retention-days: 7
+
+      - name: Upload E2E logs
+        if: ${{ !cancelled() }}
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+        with:
+          name: e2e-logs
+          path: e2e/.logs
+          retention-days: 7
@@ -22,8 +22,8 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        shardIndex: [1, 2, 3, 4, 5, 6]
-        shardTotal: [6]
+        shardIndex: [1, 2, 3, 4]
+        shardTotal: [4]
    defaults:
      run:
        shell: bash
@@ -66,7 +66,6 @@ jobs:
      - name: Checkout code
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
-          fetch-depth: 0
          persist-credentials: false

      - name: Setup web environment
@@ -84,40 +83,9 @@ jobs:

      - name: Report coverage
        if: ${{ env.CODECOV_TOKEN != '' }}
-        uses: codecov/codecov-action@1af58845a975a7985b0beb0cbe6fbbb71a41dbad # v5.5.3
+        uses: codecov/codecov-action@57e3a136b779b570ffcdbf80b3bdc90e7fab3de2 # v6.0.0
        with:
          directory: web/coverage
          flags: web
        env:
          CODECOV_TOKEN: ${{ env.CODECOV_TOKEN }}
-
-  web-build:
-    name: Web Build
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./web
-
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-        with:
-          persist-credentials: false
-
-      - name: Check changed files
-        id: changed-files
-        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # v47.0.5
-        with:
-          files: |
-            web/**
-            .github/workflows/web-tests.yml
-            .github/actions/setup-web/**
-
-      - name: Setup web environment
-        if: steps.changed-files.outputs.any_changed == 'true'
-        uses: ./.github/actions/setup-web
-
-      - name: Web build check
-        if: steps.changed-files.outputs.any_changed == 'true'
-        working-directory: ./web
-        run: vp run build
@@ -212,6 +212,7 @@ api/.vscode

 # pnpm
 /.pnpm-store
+/node_modules

 # plugin migrate
 plugins.jsonl
@@ -239,4 +240,4 @@ scripts/stress-test/reports/
 *.local.md

 # Code Agent Folder
-.qoder/*
+.qoder/*
@@ -24,8 +24,8 @@ prepare-docker:
 # Step 2: Prepare web environment
 prepare-web:
 	@echo "🌐 Setting up web environment..."
-	@cp -n web/.env.example web/.env 2>/dev/null || echo "Web .env already exists"
-	@cd web && pnpm install
+	@cp -n web/.env.example web/.env.local 2>/dev/null || echo "Web .env.local already exists"
+	@pnpm install
 	@echo "✅ Web environment prepared (not started)"

 # Step 3: Prepare API environment
@@ -74,6 +74,12 @@ type-check:
 	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
 	@echo "✅ Type checks complete"

+type-check-core:
+	@echo "📝 Running core type checks (basedpyright + mypy)..."
+	@./dev/basedpyright-check $(PATH_TO_CHECK)
+	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@echo "✅ Core type checks complete"
+
 test:
 	@echo "🧪 Running backend unit tests..."
 	@if [ -n "$(TARGET_TESTS)" ]; then \
@@ -87,7 +93,7 @@ test:
 # Build Docker images
 build-web:
 	@echo "Building web Docker image: $(WEB_IMAGE):$(VERSION)..."
-	docker build -t $(WEB_IMAGE):$(VERSION) ./web
+	docker build -f web/Dockerfile -t $(WEB_IMAGE):$(VERSION) .
 	@echo "Web Docker image built successfully: $(WEB_IMAGE):$(VERSION)"

 build-api:
@@ -133,6 +139,7 @@ help:
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
 	@echo "  make type-check     - Run type checks (basedpyright, pyrefly, mypy)"
+	@echo "  make type-check-core - Run core type checks (basedpyright, mypy)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"
@@ -53,7 +53,11 @@
  <a href="./docs/tr-TR/README.md"><img alt="Türkçe README" src="https://img.shields.io/badge/Türkçe-d9d9d9"></a>
  <a href="./docs/vi-VN/README.md"><img alt="README Tiếng Việt" src="https://img.shields.io/badge/Ti%E1%BA%BFng%20Vi%E1%BB%87t-d9d9d9"></a>
  <a href="./docs/de-DE/README.md"><img alt="README in Deutsch" src="https://img.shields.io/badge/German-d9d9d9"></a>
+  <a href="./docs/it-IT/README.md"><img alt="README in Italiano" src="https://img.shields.io/badge/Italiano-d9d9d9"></a>
+  <a href="./docs/pt-BR/README.md"><img alt="README em Português do Brasil" src="https://img.shields.io/badge/Portugu%C3%AAs%20do%20Brasil-d9d9d9"></a>
+  <a href="./docs/sl-SI/README.md"><img alt="README Slovenščina" src="https://img.shields.io/badge/Sloven%C5%A1%C4%8Dina-d9d9d9"></a>
  <a href="./docs/bn-BD/README.md"><img alt="README in বাংলা" src="https://img.shields.io/badge/বাংলা-d9d9d9"></a>
+  <a href="./docs/hi-IN/README.md"><img alt="README in हिन्दी" src="https://img.shields.io/badge/Hindi-d9d9d9"></a>
 </p>

 Dify is an open-source LLM app development platform. Its intuitive interface combines AI workflow, RAG pipeline, agent capabilities, model management, observability features (including [Opik](https://www.comet.com/docs/opik/integrations/dify), [Langfuse](https://docs.langfuse.com), and [Arize Phoenix](https://docs.arize.com/phoenix)) and more, letting you quickly go from prototype to production. Here's a list of the core features:
@@ -127,7 +127,8 @@ ALIYUN_OSS_AUTH_VERSION=v1
 ALIYUN_OSS_REGION=your-region
 # Don't start with '/'. OSS doesn't support leading slash in object names.
 ALIYUN_OSS_PATH=your-path
-ALIYUN_CLOUDBOX_ID=your-cloudbox-id
+# Optional CloudBox ID for Aliyun OSS, DO NOT enable it if you are not using CloudBox.
+#ALIYUN_CLOUDBOX_ID=your-cloudbox-id

 # Google Storage configuration
 GOOGLE_STORAGE_BUCKET_NAME=your-bucket-name
@@ -40,6 +40,8 @@ The scripts resolve paths relative to their location, so you can run them from a
   ./dev/start-web
   ```

+   `./dev/setup` and `./dev/start-web` install JavaScript dependencies through the repository root workspace, so you do not need a separate `cd web && pnpm install` step.
+
 1. Set up your application by visiting `http://localhost:3000`.

 1. Start the worker service (async and scheduler tasks, runs from `api`).
@@ -7,15 +7,16 @@ UUID_NIL = "00000000-0000-0000-0000-000000000000"

 DEFAULT_FILE_NUMBER_LIMITS = 3

-IMAGE_EXTENSIONS = convert_to_lower_and_upper_set({"jpg", "jpeg", "png", "webp", "gif", "svg"})
+_IMAGE_EXTENSION_BASE: frozenset[str] = frozenset(("jpg", "jpeg", "png", "webp", "gif", "svg"))
+_VIDEO_EXTENSION_BASE: frozenset[str] = frozenset(("mp4", "mov", "mpeg", "webm"))
+_AUDIO_EXTENSION_BASE: frozenset[str] = frozenset(("mp3", "m4a", "wav", "amr", "mpga"))

-VIDEO_EXTENSIONS = convert_to_lower_and_upper_set({"mp4", "mov", "mpeg", "webm"})
+IMAGE_EXTENSIONS: frozenset[str] = frozenset(convert_to_lower_and_upper_set(_IMAGE_EXTENSION_BASE))
+VIDEO_EXTENSIONS: frozenset[str] = frozenset(convert_to_lower_and_upper_set(_VIDEO_EXTENSION_BASE))
+AUDIO_EXTENSIONS: frozenset[str] = frozenset(convert_to_lower_and_upper_set(_AUDIO_EXTENSION_BASE))

-AUDIO_EXTENSIONS = convert_to_lower_and_upper_set({"mp3", "m4a", "wav", "amr", "mpga"})
-
-_doc_extensions: set[str]
-if dify_config.ETL_TYPE == "Unstructured":
-    _doc_extensions = {
+_UNSTRUCTURED_DOCUMENT_EXTENSION_BASE: frozenset[str] = frozenset(
+    (
        "txt",
        "markdown",
        "md",
@@ -35,11 +36,10 @@ if dify_config.ETL_TYPE == "Unstructured":
        "pptx",
        "xml",
        "epub",
-    }
-    if dify_config.UNSTRUCTURED_API_URL:
-        _doc_extensions.add("ppt")
-else:
-    _doc_extensions = {
+    )
+)
+_DEFAULT_DOCUMENT_EXTENSION_BASE: frozenset[str] = frozenset(
+    (
        "txt",
        "markdown",
        "md",
@@ -53,8 +53,17 @@ else:
        "csv",
        "vtt",
        "properties",
-    }
-DOCUMENT_EXTENSIONS: set[str] = convert_to_lower_and_upper_set(_doc_extensions)
+    )
+)
+
+_doc_extensions: set[str]
+if dify_config.ETL_TYPE == "Unstructured":
+    _doc_extensions = set(_UNSTRUCTURED_DOCUMENT_EXTENSION_BASE)
+    if dify_config.UNSTRUCTURED_API_URL:
+        _doc_extensions.add("ppt")
+else:
+    _doc_extensions = set(_DEFAULT_DOCUMENT_EXTENSION_BASE)
+DOCUMENT_EXTENSIONS: frozenset[str] = frozenset(convert_to_lower_and_upper_set(_doc_extensions))

 # console
 COOKIE_NAME_ACCESS_TOKEN = "access_token"
@@ -4,8 +4,8 @@ from urllib.parse import quote

 from flask import Response

-HTML_MIME_TYPES = frozenset({"text/html", "application/xhtml+xml"})
-HTML_EXTENSIONS = frozenset({"html", "htm"})
+HTML_MIME_TYPES: frozenset[str] = frozenset(("text/html", "application/xhtml+xml"))
+HTML_EXTENSIONS: frozenset[str] = frozenset(("html", "htm"))


 def _normalize_mime_type(mime_type: str | None) -> str:
@@ -2,7 +2,7 @@ import flask_restx
 from flask_restx import Resource, fields, marshal_with
 from flask_restx._http import HTTPStatus
 from sqlalchemy import delete, func, select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden

 from extensions.ext_database import db
@@ -34,7 +34,7 @@ api_key_list_model = console_ns.model(


 def _get_resource(resource_id, tenant_id, resource_model):
-    with Session(db.engine) as session:
+    with sessionmaker(db.engine).begin() as session:
        resource = session.execute(
            select(resource_model).filter_by(id=resource_id, tenant_id=tenant_id)
        ).scalar_one_or_none()
@@ -9,7 +9,7 @@ from graphon.enums import WorkflowExecutionStatus
 from graphon.file import helpers as file_helpers
 from pydantic import AliasChoices, BaseModel, ConfigDict, Field, computed_field, field_validator
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest

 from controllers.common.helpers import FileInfo
@@ -642,7 +642,7 @@ class AppCopyApi(Resource):

        args = CopyAppPayload.model_validate(console_ns.payload or {})

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            import_service = AppDslService(session)
            yaml_content = import_service.export_dsl(app_model=app_model, include_secret=True)
            result = import_service.import_app(
@@ -655,7 +655,6 @@ class AppCopyApi(Resource):
                icon=args.icon,
                icon_background=args.icon_background,
            )
-            session.commit()

            # Inherit web app permission from original app
            if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
@@ -1,6 +1,6 @@
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
@@ -71,7 +71,7 @@ class AppImportApi(Resource):
        args = AppImportPayload.model_validate(console_ns.payload)

        # Create service with session
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = AppDslService(session)
            # Import app
            account = current_user
@@ -87,7 +87,6 @@ class AppImportApi(Resource):
                icon_background=args.icon_background,
                app_id=args.app_id,
            )
-            session.commit()
        if result.app_id and FeatureService.get_system_features().webapp_auth.enabled:
            # update web app setting as private
            EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
@@ -112,12 +111,11 @@ class AppImportConfirmApi(Resource):
        current_user, _ = current_account_with_tenant()

        # Create service with session
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = AppDslService(session)
            # Confirm import
            account = current_user
            result = import_service.confirm_import(import_id=import_id, account=account)
-            session.commit()

        # Return appropriate status code based on result
        if result.status == ImportStatus.FAILED:
@@ -134,7 +132,7 @@ class AppImportCheckDependenciesApi(Resource):
    @marshal_with(app_import_check_dependencies_model)
    @edit_permission_required
    def get(self, app_model: App):
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = AppDslService(session)
            result = import_service.check_dependencies(app_model=app_model)

@@ -2,7 +2,7 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
@@ -69,7 +69,7 @@ class ConversationVariablesApi(Resource):
        page_size = 100
        stmt = stmt.limit(page_size).offset((page - 1) * page_size)

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            rows = session.scalars(stmt).all()

        return {
@@ -9,8 +9,8 @@ from graphon.enums import NodeType
 from graphon.file import File
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.utils.encoders import jsonable_encoder
-from pydantic import BaseModel, Field, field_validator
-from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field, ValidationError, field_validator
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound

 import services
@@ -268,22 +268,18 @@ class DraftWorkflowApi(Resource):

        content_type = request.headers.get("Content-Type", "")

-        payload_data: dict[str, Any] | None = None
        if "application/json" in content_type:
            payload_data = request.get_json(silent=True)
            if not isinstance(payload_data, dict):
                return {"message": "Invalid JSON data"}, 400
+            args_model = SyncDraftWorkflowPayload.model_validate(payload_data)
        elif "text/plain" in content_type:
            try:
-                payload_data = json.loads(request.data.decode("utf-8"))
-            except json.JSONDecodeError:
-                return {"message": "Invalid JSON data"}, 400
-            if not isinstance(payload_data, dict):
+                args_model = SyncDraftWorkflowPayload.model_validate_json(request.data)
+            except (ValueError, ValidationError):
                return {"message": "Invalid JSON data"}, 400
        else:
            abort(415)
-
-        args_model = SyncDraftWorkflowPayload.model_validate(payload_data)
        args = args_model.model_dump()
        workflow_service = WorkflowService()

@@ -840,7 +836,7 @@ class PublishedWorkflowApi(Resource):
        args = PublishWorkflowPayload.model_validate(console_ns.payload or {})

        workflow_service = WorkflowService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            workflow = workflow_service.publish_workflow(
                session=session,
                app_model=app_model,
@@ -858,8 +854,6 @@ class PublishedWorkflowApi(Resource):

            workflow_created_at = TimestampField().format(workflow.created_at)

-            session.commit()
-
        return {
            "result": "success",
            "created_at": workflow_created_at,
@@ -982,7 +976,7 @@ class PublishedAllWorkflowApi(Resource):
                raise Forbidden()

        workflow_service = WorkflowService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            workflows, has_more = workflow_service.get_all_published_workflow(
                session=session,
                app_model=app_model,
@@ -1072,7 +1066,7 @@ class WorkflowByIdApi(Resource):
        workflow_service = WorkflowService()

        # Create a session and manage the transaction
-        with Session(db.engine, expire_on_commit=False) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            workflow = workflow_service.update_workflow(
                session=session,
                workflow_id=workflow_id,
@@ -1084,9 +1078,6 @@ class WorkflowByIdApi(Resource):
            if not workflow:
                raise NotFound("Workflow not found")

-            # Commit the transaction in the controller
-            session.commit()
-
        return workflow

    @setup_required
@@ -1101,13 +1092,11 @@ class WorkflowByIdApi(Resource):
        workflow_service = WorkflowService()

        # Create a session and manage the transaction
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            try:
                workflow_service.delete_workflow(
                    session=session, workflow_id=workflow_id, tenant_id=app_model.tenant_id
                )
-                # Commit the transaction in the controller
-                session.commit()
            except WorkflowInUseError as e:
                abort(400, description=str(e))
            except DraftWorkflowDeletionError as e:
@@ -5,7 +5,7 @@ from flask import request
 from flask_restx import Resource, marshal_with
 from graphon.enums import WorkflowExecutionStatus
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
@@ -87,7 +87,7 @@ class WorkflowAppLogApi(Resource):

        # get paginate workflow app logs
        workflow_app_service = WorkflowAppService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_app_logs(
                session=session,
                app_model=app_model,
@@ -124,7 +124,7 @@ class WorkflowArchivedLogApi(Resource):
        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore

        workflow_app_service = WorkflowAppService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            workflow_app_log_pagination = workflow_app_service.get_paginate_workflow_archive_logs(
                session=session,
                app_model=app_model,
@@ -10,7 +10,7 @@ from graphon.variables.segment_group import SegmentGroup
 from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
 from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.console import console_ns
 from controllers.console.app.error import (
@@ -244,7 +244,7 @@ class WorkflowVariableCollectionApi(Resource):
            raise DraftWorkflowNotExist()

        # fetch draft workflow by app_model
-        with Session(bind=db.engine, expire_on_commit=False) as session:
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
                session=session,
            )
@@ -298,7 +298,7 @@ class NodeVariableCollectionApi(Resource):
    @marshal_with(workflow_draft_variable_list_model)
    def get(self, app_model: App, node_id: str):
        validate_node_id(node_id)
-        with Session(bind=db.engine, expire_on_commit=False) as session:
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
                session=session,
            )
@@ -465,7 +465,7 @@ class VariableResetApi(Resource):


 def _get_variable_list(app_model: App, node_id) -> WorkflowDraftVariableList:
-    with Session(bind=db.engine, expire_on_commit=False) as session:
+    with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
        draft_var_srv = WorkflowDraftVariableService(
            session=session,
        )
@@ -4,7 +4,7 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound

 from configs import dify_config
@@ -64,7 +64,7 @@ class WebhookTriggerApi(Resource):

        node_id = args.node_id

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            # Get webhook trigger for this app and node
            webhook_trigger = (
                session.query(WorkflowWebhookTrigger)
@@ -95,7 +95,7 @@ class AppTriggersApi(Resource):
        assert isinstance(current_user, Account)
        assert current_user.current_tenant_id is not None

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            # Get all triggers for this app using select API
            triggers = (
                session.execute(
@@ -137,7 +137,7 @@ class AppTriggerEnableApi(Resource):
        assert current_user.current_tenant_id is not None

        trigger_id = args.trigger_id
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            # Find the trigger using select
            trigger = session.execute(
                select(AppTrigger).where(
@@ -153,9 +153,6 @@ class AppTriggerEnableApi(Resource):
            # Update status based on enable_trigger boolean
            trigger.status = AppTriggerStatus.ENABLED if args.enable_trigger else AppTriggerStatus.DISABLED

-            session.commit()
-            session.refresh(trigger)
-
        # Add computed icon field
        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
        if trigger.trigger_type == "trigger-plugin":
@@ -36,7 +36,7 @@ class Subscription(Resource):
    @only_edition_cloud
    def get(self):
        current_user, current_tenant_id = current_account_with_tenant()
-        args = SubscriptionQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = SubscriptionQuery.model_validate(request.args.to_dict(flat=True))
        BillingService.is_tenant_owner_or_admin(current_user)
        return BillingService.get_subscription(args.plan, args.interval, current_user.email, current_tenant_id)

@@ -31,7 +31,7 @@ class ComplianceApi(Resource):
    @only_edition_cloud
    def get(self):
        current_user, current_tenant_id = current_account_with_tenant()
-        args = ComplianceDownloadQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ComplianceDownloadQuery.model_validate(request.args.to_dict(flat=True))

        ip_address = extract_remote_ip(request)
        device_info = request.headers.get("User-Agent", "Unknown device")
@@ -6,7 +6,7 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound

 from controllers.common.schema import get_or_create_model, register_schema_model
@@ -159,7 +159,7 @@ class DataSourceApi(Resource):
    @account_initialization_required
    def patch(self, binding_id, action: Literal["enable", "disable"]):
        binding_id = str(binding_id)
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            data_source_binding = session.execute(
                select(DataSourceOauthBinding).filter_by(id=binding_id)
            ).scalar_one_or_none()
@@ -211,7 +211,7 @@ class DataSourceNotionListApi(Resource):
        if not credential:
            raise NotFound("Credential not found.")
        exist_page_ids = []
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            # import notion in the exist dataset
            if query.dataset_id:
                dataset = DatasetService.get_dataset(query.dataset_id)
@@ -120,7 +120,8 @@ class DatasourceOAuthCallback(Resource):
        if context is None:
            raise Forbidden("Invalid context_id")

-        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
+        user_id: str = context["user_id"]
+        tenant_id: str = context["tenant_id"]
        datasource_provider_id = DatasourceProviderID(provider_id)
        plugin_id = datasource_provider_id.plugin_id
        datasource_provider_service = DatasourceProviderService()
@@ -141,7 +142,7 @@ class DatasourceOAuthCallback(Resource):
            system_credentials=oauth_client_params,
            request=request,
        )
-        credential_id = context.get("credential_id")
+        credential_id: str | None = context.get("credential_id")
        if credential_id:
            datasource_provider_service.reauthorize_datasource_oauth_provider(
                tenant_id=tenant_id,
@@ -150,7 +151,7 @@ class DatasourceOAuthCallback(Resource):
                name=oauth_response.metadata.get("name") or None,
                expire_at=oauth_response.expires_at,
                credentials=dict(oauth_response.credentials),
-                credential_id=context.get("credential_id"),
+                credential_id=credential_id,
            )
        else:
            datasource_provider_service.add_datasource_oauth_provider(
@@ -3,7 +3,7 @@ import logging
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
@@ -85,7 +85,7 @@ class CustomizedPipelineTemplateApi(Resource):
    @account_initialization_required
    @enterprise_license_required
    def post(self, template_id: str):
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            template = (
                session.query(PipelineCustomizedTemplate).where(PipelineCustomizedTemplate.id == template_id).first()
            )
@@ -1,6 +1,6 @@
 from flask_restx import Resource, marshal
 from pydantic import BaseModel
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden

 import services
@@ -54,7 +54,7 @@ class CreateRagPipelineDatasetApi(Resource):
            yaml_content=payload.yaml_content,
        )
        try:
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                rag_pipeline_dsl_service = RagPipelineDslService(session)
                import_info = rag_pipeline_dsl_service.create_rag_pipeline_dataset(
                    tenant_id=current_tenant_id,
@@ -5,7 +5,7 @@ from flask import Response, request
 from flask_restx import Resource, marshal, marshal_with
 from graphon.variables.types import SegmentType
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden

 from controllers.common.schema import register_schema_models
@@ -96,7 +96,7 @@ class RagPipelineVariableCollectionApi(Resource):
            raise DraftWorkflowNotExist()

        # fetch draft workflow by app_model
-        with Session(bind=db.engine, expire_on_commit=False) as session:
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
                session=session,
            )
@@ -143,7 +143,7 @@ class RagPipelineNodeVariableCollectionApi(Resource):
    @marshal_with(workflow_draft_variable_list_model)
    def get(self, pipeline: Pipeline, node_id: str):
        validate_node_id(node_id)
-        with Session(bind=db.engine, expire_on_commit=False) as session:
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
            draft_var_srv = WorkflowDraftVariableService(
                session=session,
            )
@@ -289,7 +289,7 @@ class RagPipelineVariableResetApi(Resource):


 def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList:
-    with Session(bind=db.engine, expire_on_commit=False) as session:
+    with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
        draft_var_srv = WorkflowDraftVariableService(
            session=session,
        )
@@ -1,7 +1,7 @@
 from flask import request
 from flask_restx import Resource, fields, marshal_with  # type: ignore
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.common.schema import get_or_create_model, register_schema_models
 from controllers.console import console_ns
@@ -68,7 +68,7 @@ class RagPipelineImportApi(Resource):
        payload = RagPipelineImportPayload.model_validate(console_ns.payload or {})

        # Create service with session
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = RagPipelineDslService(session)
            # Import app
            account = current_user
@@ -80,7 +80,6 @@ class RagPipelineImportApi(Resource):
                pipeline_id=payload.pipeline_id,
                dataset_name=payload.name,
            )
-            session.commit()

        # Return appropriate status code based on result
        status = result.status
@@ -102,12 +101,11 @@ class RagPipelineImportConfirmApi(Resource):
        current_user, _ = current_account_with_tenant()

        # Create service with session
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = RagPipelineDslService(session)
            # Confirm import
            account = current_user
            result = import_service.confirm_import(import_id=import_id, account=account)
-            session.commit()

        # Return appropriate status code based on result
        if result.status == ImportStatus.FAILED:
@@ -124,7 +122,7 @@ class RagPipelineImportCheckDependenciesApi(Resource):
    @edit_permission_required
    @marshal_with(pipeline_import_check_dependencies_model)
    def get(self, pipeline: Pipeline):
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            import_service = RagPipelineDslService(session)
            result = import_service.check_dependencies(pipeline=pipeline)

@@ -142,7 +140,7 @@ class RagPipelineExportApi(Resource):
        # Add include_secret params
        query = IncludeSecretQuery.model_validate(request.args.to_dict())

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            export_service = RagPipelineDslService(session)
            result = export_service.export_rag_pipeline_dsl(
                pipeline=pipeline, include_secret=query.include_secret == "true"
@@ -5,8 +5,8 @@ from typing import Any, Literal, cast
 from flask import abort, request
 from flask_restx import Resource, marshal_with  # type: ignore
 from graphon.model_runtime.utils.encoders import jsonable_encoder
-from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session
+from pydantic import BaseModel, Field, ValidationError
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound

 import services
@@ -53,6 +53,7 @@ from services.rag_pipeline.pipeline_generate_service import PipelineGenerateServ
 from services.rag_pipeline.rag_pipeline import RagPipelineService
 from services.rag_pipeline.rag_pipeline_manage_service import RagPipelineManageService
 from services.rag_pipeline.rag_pipeline_transform_service import RagPipelineTransformService
+from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService

 logger = logging.getLogger(__name__)

@@ -185,29 +186,14 @@ class DraftRagPipelineApi(Resource):

        if "application/json" in content_type:
            payload_dict = console_ns.payload or {}
+            payload = DraftWorkflowSyncPayload.model_validate(payload_dict)
        elif "text/plain" in content_type:
            try:
-                data = json.loads(request.data.decode("utf-8"))
-                if "graph" not in data or "features" not in data:
-                    raise ValueError("graph or features not found in data")
-
-                if not isinstance(data.get("graph"), dict):
-                    raise ValueError("graph is not a dict")
-
-                payload_dict = {
-                    "graph": data.get("graph"),
-                    "features": data.get("features"),
-                    "hash": data.get("hash"),
-                    "environment_variables": data.get("environment_variables"),
-                    "conversation_variables": data.get("conversation_variables"),
-                    "rag_pipeline_variables": data.get("rag_pipeline_variables"),
-                }
-            except json.JSONDecodeError:
+                payload = DraftWorkflowSyncPayload.model_validate_json(request.data)
+            except (ValueError, ValidationError):
                return {"message": "Invalid JSON data"}, 400
        else:
            abort(415)
-
-        payload = DraftWorkflowSyncPayload.model_validate(payload_dict)
        rag_pipeline_service = RagPipelineService()

        try:
@@ -607,7 +593,7 @@ class PublishedRagPipelineApi(Resource):
        # The role of the current user in the ta table must be admin, owner, or editor
        current_user, _ = current_account_with_tenant()
        rag_pipeline_service = RagPipelineService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            pipeline = session.merge(pipeline)
            workflow = rag_pipeline_service.publish_workflow(
                session=session,
@@ -619,8 +605,6 @@ class PublishedRagPipelineApi(Resource):
            session.add(pipeline)
            workflow_created_at = TimestampField().format(workflow.created_at)

-            session.commit()
-
        return {
            "result": "success",
            "created_at": workflow_created_at,
@@ -694,7 +678,7 @@ class PublishedAllRagPipelineApi(Resource):
                raise Forbidden()

        rag_pipeline_service = RagPipelineService()
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            workflows, has_more = rag_pipeline_service.get_all_published_workflow(
                session=session,
                pipeline=pipeline,
@@ -766,7 +750,7 @@ class RagPipelineByIdApi(Resource):
        rag_pipeline_service = RagPipelineService()

        # Create a session and manage the transaction
-        with Session(db.engine, expire_on_commit=False) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            workflow = rag_pipeline_service.update_workflow(
                session=session,
                workflow_id=workflow_id,
@@ -778,10 +762,37 @@ class RagPipelineByIdApi(Resource):
            if not workflow:
                raise NotFound("Workflow not found")

-            # Commit the transaction in the controller
-            session.commit()
+            return workflow

-        return workflow
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    @get_rag_pipeline
+    def delete(self, pipeline: Pipeline, workflow_id: str):
+        """
+        Delete a published workflow version that is not currently active on the pipeline.
+        """
+        if pipeline.workflow_id == workflow_id:
+            abort(400, description=f"Cannot delete workflow that is currently in use by pipeline '{pipeline.id}'")
+
+        workflow_service = WorkflowService()
+
+        with sessionmaker(db.engine).begin() as session:
+            try:
+                workflow_service.delete_workflow(
+                    session=session,
+                    workflow_id=workflow_id,
+                    tenant_id=pipeline.tenant_id,
+                )
+            except WorkflowInUseError as e:
+                abort(400, description=str(e))
+            except DraftWorkflowDeletionError as e:
+                abort(400, description=str(e))
+            except ValueError as e:
+                raise NotFound(str(e))
+
+        return None, 204


@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/processing/parameters")
@@ -2,7 +2,7 @@ from typing import Any

 from flask import request
 from pydantic import BaseModel, Field, TypeAdapter, model_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound

 from controllers.common.schema import register_schema_models
@@ -74,7 +74,7 @@ class ConversationListApi(InstalledAppResource):
        try:
            if not isinstance(current_user, Account):
                raise ValueError("current_user must be an Account instance")
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                pagination = WebConversationService.pagination_by_last_id(
                    session=session,
                    app_model=app_model,
@@ -2,7 +2,7 @@ from collections.abc import Callable
 from functools import wraps
 from typing import ParamSpec, TypeVar

-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden

 from extensions.ext_database import db
@@ -24,7 +24,7 @@ def plugin_permission_required(
            user = current_user
            tenant_id = current_tenant_id

-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                permission = (
                    session.query(TenantPluginPermission)
                    .where(
@@ -8,7 +8,7 @@ from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator, model_validator
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from configs import dify_config
 from constants.languages import supported_language
@@ -519,7 +519,7 @@ class EducationAutoCompleteApi(Resource):
    @cloud_edition_billing_enabled
    @marshal_with(data_fields)
    def get(self):
-        payload = request.args.to_dict(flat=True)  # type: ignore
+        payload = request.args.to_dict(flat=True)
        args = EducationAutocompleteQuery.model_validate(payload)

        return BillingService.EducationIdentity.autocomplete(args.keywords, args.page, args.limit)
@@ -562,7 +562,7 @@ class ChangeEmailSendEmailApi(Resource):

            user_email = current_user.email
        else:
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                account = AccountService.get_account_by_email_with_case_fallback(args.email, session=session)
            if account is None:
                raise AccountNotFound()
@@ -99,7 +99,7 @@ class ModelProviderListApi(Resource):
        _, current_tenant_id = current_account_with_tenant()
        tenant_id = current_tenant_id

-        payload = request.args.to_dict(flat=True)  # type: ignore
+        payload = request.args.to_dict(flat=True)
        args = ParserModelList.model_validate(payload)

        model_provider_service = ModelProviderService()
@@ -118,7 +118,7 @@ class ModelProviderCredentialApi(Resource):
        _, current_tenant_id = current_account_with_tenant()
        tenant_id = current_tenant_id
        # if credential_id is not provided, return current used credential
-        payload = request.args.to_dict(flat=True)  # type: ignore
+        payload = request.args.to_dict(flat=True)
        args = ParserCredentialId.model_validate(payload)

        model_provider_service = ModelProviderService()
@@ -287,12 +287,10 @@ class ModelProviderModelCredentialApi(Resource):
                provider=provider,
            )
        else:
-            # Normalize model_type to the origin value stored in DB (e.g., "text-generation" for LLM)
-            normalized_model_type = args.model_type.to_origin_model_type()
            available_credentials = model_provider_service.get_provider_model_available_credentials(
                tenant_id=tenant_id,
                provider=provider,
-                model_type=normalized_model_type,
+                model_type=args.model_type,
                model=args.model,
            )

@@ -7,7 +7,7 @@ from flask import make_response, redirect, request, send_file
 from flask_restx import Resource
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, Field, HttpUrl, field_validator, model_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden

 from configs import dify_config
@@ -832,7 +832,8 @@ class ToolOAuthCallback(Resource):
        tool_provider = ToolProviderID(provider)
        plugin_id = tool_provider.plugin_id
        provider_name = tool_provider.provider_name
-        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
+        user_id: str = context["user_id"]
+        tenant_id: str = context["tenant_id"]

        oauth_handler = OAuthHandler()
        oauth_client_params = BuiltinToolManageService.get_oauth_client(tenant_id, provider)
@@ -1018,7 +1019,7 @@ class ToolProviderMCPApi(Resource):

        # Step 1: Get provider data for URL validation (short-lived session, no network I/O)
        validation_data = None
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            validation_data = service.get_provider_for_url_validation(
                tenant_id=current_tenant_id, provider_id=payload.provider_id
@@ -1033,7 +1034,7 @@ class ToolProviderMCPApi(Resource):
        )

        # Step 3: Perform database update in a transaction
-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            service.update_provider(
                tenant_id=current_tenant_id,
@@ -1060,7 +1061,7 @@ class ToolProviderMCPApi(Resource):
        payload = MCPProviderDeletePayload.model_validate(console_ns.payload or {})
        _, current_tenant_id = current_account_with_tenant()

-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            service.delete_provider(tenant_id=current_tenant_id, provider_id=payload.provider_id)

@@ -1078,7 +1079,7 @@ class ToolMCPAuthApi(Resource):
        provider_id = payload.provider_id
        _, tenant_id = current_account_with_tenant()

-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            db_provider = service.get_provider(provider_id=provider_id, tenant_id=tenant_id)
            if not db_provider:
@@ -1099,7 +1100,7 @@ class ToolMCPAuthApi(Resource):
                sse_read_timeout=provider_entity.sse_read_timeout,
            ):
                # Update credentials in new transaction
-                with Session(db.engine) as session, session.begin():
+                with sessionmaker(db.engine).begin() as session:
                    service = MCPToolManageService(session=session)
                    service.update_provider_credentials(
                        provider_id=provider_id,
@@ -1117,17 +1118,17 @@ class ToolMCPAuthApi(Resource):
                    resource_metadata_url=e.resource_metadata_url,
                    scope_hint=e.scope_hint,
                )
-                with Session(db.engine) as session, session.begin():
+                with sessionmaker(db.engine).begin() as session:
                    service = MCPToolManageService(session=session)
                    response = service.execute_auth_actions(auth_result)
                    return response
            except MCPRefreshTokenError as e:
-                with Session(db.engine) as session, session.begin():
+                with sessionmaker(db.engine).begin() as session:
                    service = MCPToolManageService(session=session)
                    service.clear_provider_credentials(provider_id=provider_id, tenant_id=tenant_id)
                raise ValueError(f"Failed to refresh token, please try to authorize again: {e}") from e
        except (MCPError, ValueError) as e:
-            with Session(db.engine) as session, session.begin():
+            with sessionmaker(db.engine).begin() as session:
                service = MCPToolManageService(session=session)
                service.clear_provider_credentials(provider_id=provider_id, tenant_id=tenant_id)
            raise ValueError(f"Failed to connect to MCP server: {e}") from e
@@ -1140,7 +1141,7 @@ class ToolMCPDetailApi(Resource):
    @account_initialization_required
    def get(self, provider_id):
        _, tenant_id = current_account_with_tenant()
-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            provider = service.get_provider(provider_id=provider_id, tenant_id=tenant_id)
            return jsonable_encoder(ToolTransformService.mcp_provider_to_user_provider(provider, for_list=True))
@@ -1154,7 +1155,7 @@ class ToolMCPListAllApi(Resource):
    def get(self):
        _, tenant_id = current_account_with_tenant()

-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            # Skip sensitive data decryption for list view to improve performance
            tools = service.list_providers(tenant_id=tenant_id, include_sensitive=False)
@@ -1169,7 +1170,7 @@ class ToolMCPUpdateApi(Resource):
    @account_initialization_required
    def get(self, provider_id):
        _, tenant_id = current_account_with_tenant()
-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            service = MCPToolManageService(session=session)
            tools = service.list_provider_tools(
                tenant_id=tenant_id,
@@ -1187,7 +1188,7 @@ class ToolMCPCallbackApi(Resource):
        authorization_code = query.code

        # Create service instance for handle_callback
-        with Session(db.engine) as session, session.begin():
+        with sessionmaker(db.engine).begin() as session:
            mcp_service = MCPToolManageService(session=session)
            # handle_callback now returns state data and tokens
            state_data, tokens = handle_callback(state_key, authorization_code)
@@ -5,7 +5,7 @@ from flask import make_response, redirect, request
 from flask_restx import Resource
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel, model_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden

 from configs import dify_config
@@ -375,7 +375,7 @@ class TriggerSubscriptionDeleteApi(Resource):
        assert user.current_tenant_id is not None

        try:
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                # Delete trigger provider subscription
                TriggerProviderService.delete_trigger_provider(
                    session=session,
@@ -388,7 +388,6 @@ class TriggerSubscriptionDeleteApi(Resource):
                    tenant_id=user.current_tenant_id,
                    subscription_id=subscription_id,
                )
-                session.commit()
            return {"result": "success"}
        except ValueError as e:
            raise BadRequest(str(e))
@@ -499,9 +498,9 @@ class TriggerOAuthCallbackApi(Resource):
        provider_id = TriggerProviderID(provider)
        plugin_id = provider_id.plugin_id
        provider_name = provider_id.provider_name
-        user_id = context.get("user_id")
-        tenant_id = context.get("tenant_id")
-        subscription_builder_id = context.get("subscription_builder_id")
+        user_id: str = context["user_id"]
+        tenant_id: str = context["tenant_id"]
+        subscription_builder_id: str = context["subscription_builder_id"]

        # Get OAuth client configuration
        oauth_client_params = TriggerProviderService.get_oauth_client(
@@ -155,7 +155,7 @@ class WorkspaceListApi(Resource):
    @setup_required
    @admin_required
    def get(self):
-        payload = request.args.to_dict(flat=True)  # type: ignore
+        payload = request.args.to_dict(flat=True)
        args = WorkspaceListQuery.model_validate(payload)

        stmt = select(Tenant).order_by(Tenant.created_at.desc())
@@ -8,6 +8,7 @@ Go admin-api caller.
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
+from sqlalchemy import select
 from sqlalchemy.orm import Session

 from controllers.common.schema import register_schema_model
@@ -87,7 +88,7 @@ class EnterpriseAppDSLExport(Resource):
        """Export an app's DSL as YAML."""
        include_secret = request.args.get("include_secret", "false").lower() == "true"

-        app_model = db.session.query(App).filter_by(id=app_id).first()
+        app_model = db.session.get(App, app_id)
        if not app_model:
            return {"message": "app not found"}, 404

@@ -104,7 +105,7 @@ def _get_active_account(email: str) -> Account | None:

    Workspace membership is already validated by the Go admin-api caller.
    """
-    account = db.session.query(Account).filter_by(email=email).first()
+    account = db.session.scalar(select(Account).where(Account.email == email).limit(1))
    if account is None or account.status != AccountStatus.ACTIVE:
        return None
    return account
@@ -4,7 +4,7 @@ from flask import Response
 from flask_restx import Resource
 from graphon.variables.input_entities import VariableEntity
 from pydantic import BaseModel, Field, ValidationError
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker

 from controllers.common.schema import register_schema_model
 from controllers.mcp import mcp_ns
@@ -67,7 +67,7 @@ class MCPAppApi(Resource):
        request_id: Union[int, str] | None = args.id
        mcp_request = self._parse_mcp_request(args.model_dump(exclude_none=True))

-        with Session(db.engine, expire_on_commit=False) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            # Get MCP server and app
            mcp_server, app = self._get_mcp_server_and_app(server_code, session)
            self._validate_server_status(mcp_server)
@@ -174,6 +174,7 @@ class MCPAppApi(Resource):
            required=variable.get("required", False),
            max_length=variable.get("max_length"),
            options=variable.get("options") or [],
+            json_schema=variable.get("json_schema"),
        )

    def _parse_mcp_request(self, args: dict) -> mcp_types.ClientRequest | mcp_types.ClientNotification:
@@ -188,7 +189,7 @@ class MCPAppApi(Resource):

    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str) -> EndUser | None:
        """Get end user - manages its own database session"""
-        with Session(db.engine, expire_on_commit=False) as session, session.begin():
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            return (
                session.query(EndUser)
                .where(EndUser.tenant_id == tenant_id)
@@ -228,9 +229,7 @@ class MCPAppApi(Resource):
        if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
            client_info = mcp_request.root.params.clientInfo
            client_name = f"{client_info.name}@{client_info.version}"
-            # Commit the session before creating end user to avoid transaction conflicts
-            session.commit()
-            with Session(db.engine, expire_on_commit=False) as create_session, create_session.begin():
+            with sessionmaker(db.engine, expire_on_commit=False).begin() as create_session:
                end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, create_session)

        return handle_mcp_request(app, mcp_request, user_input_form, mcp_server, end_user, request_id)
@@ -29,6 +29,31 @@ from services.entities.knowledge_entities.knowledge_entities import SegmentUpdat
 from services.errors.chunk import ChildChunkDeleteIndexError, ChildChunkIndexingError
 from services.errors.chunk import ChildChunkDeleteIndexError as ChildChunkDeleteIndexServiceError
 from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingServiceError
+from services.summary_index_service import SummaryIndexService
+
+
+def _marshal_segment_with_summary(segment, dataset_id: str) -> dict:
+    """Marshal a single segment and enrich it with summary content."""
+    segment_dict = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
+    summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id)
+    segment_dict["summary"] = summary.summary_content if summary else None
+    return segment_dict
+
+
+def _marshal_segments_with_summary(segments, dataset_id: str) -> list[dict]:
+    """Marshal multiple segments and enrich them with summary content (batch query)."""
+    segment_ids = [segment.id for segment in segments]
+    summaries: dict = {}
+    if segment_ids:
+        summary_records = SummaryIndexService.get_segments_summaries(segment_ids=segment_ids, dataset_id=dataset_id)
+        summaries = {chunk_id: record.summary_content for chunk_id, record in summary_records.items()}
+
+    result = []
+    for segment in segments:
+        segment_dict = dict(marshal(segment, segment_fields))  # type: ignore[arg-type]
+        segment_dict["summary"] = summaries.get(segment.id)
+        result.append(segment_dict)
+    return result


 class SegmentCreatePayload(BaseModel):
@@ -132,7 +157,7 @@ class SegmentApi(DatasetApiResource):
            for args_item in payload.segments:
                SegmentService.segment_create_args_validate(args_item, document)
            segments = SegmentService.multi_create_segment(payload.segments, document, dataset)
-            return {"data": marshal(segments, segment_fields), "doc_form": document.doc_form}, 200
+            return {"data": _marshal_segments_with_summary(segments, dataset_id), "doc_form": document.doc_form}, 200
        else:
            return {"error": "Segments is required"}, 400

@@ -196,7 +221,7 @@ class SegmentApi(DatasetApiResource):
        )

        response = {
-            "data": marshal(segments, segment_fields),
+            "data": _marshal_segments_with_summary(segments, dataset_id),
            "doc_form": document.doc_form,
            "total": total,
            "has_more": len(segments) == limit,
@@ -296,7 +321,7 @@ class DatasetSegmentApi(DatasetApiResource):
        payload = SegmentUpdatePayload.model_validate(service_api_ns.payload or {})

        updated_segment = SegmentService.update_segment(payload.segment, segment, document, dataset)
-        return {"data": marshal(updated_segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _marshal_segment_with_summary(updated_segment, dataset_id), "doc_form": document.doc_form}, 200

    @service_api_ns.doc("get_segment")
    @service_api_ns.doc(description="Get a specific segment by ID")
@@ -326,7 +351,7 @@ class DatasetSegmentApi(DatasetApiResource):
        if not segment:
            raise NotFound("Segment not found.")

-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _marshal_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200


@service_api_ns.route(
@@ -2,7 +2,7 @@ from typing import Literal

 from flask import request
 from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound

 from controllers.common.schema import register_schema_models
@@ -99,7 +99,7 @@ class ConversationListApi(WebApiResource):
        query = ConversationListQuery.model_validate(raw_args)

        try:
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                pagination = WebConversationService.pagination_by_last_id(
                    session=session,
                    app_model=app_model,
@@ -4,7 +4,7 @@ import secrets
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker

 from controllers.common.schema import register_schema_models
 from controllers.console.auth.error import (
@@ -81,7 +81,7 @@ class ForgotPasswordSendEmailApi(Resource):
        else:
            language = "en-US"

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            account = AccountService.get_account_by_email_with_case_fallback(request_email, session=session)
        token = None
        if account is None:
@@ -180,18 +180,17 @@ class ForgotPasswordResetApi(Resource):

        email = reset_data.get("email", "")

-        with Session(db.engine) as session:
+        with sessionmaker(db.engine).begin() as session:
            account = AccountService.get_account_by_email_with_case_fallback(email, session=session)

            if account:
-                self._update_existing_account(account, password_hashed, salt, session)
+                self._update_existing_account(account, password_hashed, salt)
            else:
                raise AuthenticationFailedError()

        return {"result": "success"}

-    def _update_existing_account(self, account: Account, password_hashed, salt, session):
+    def _update_existing_account(self, account: Account, password_hashed, salt):
        # Update existing account credentials
        account.password = base64.b64encode(password_hashed).decode()
        account.password_salt = base64.b64encode(salt).decode()
-        session.commit()
@@ -6,7 +6,7 @@ from typing import Concatenate, ParamSpec, TypeVar
 from flask import request
 from flask_restx import Resource
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, NotFound, Unauthorized

 from constants import HEADER_NAME_APP_CODE
@@ -49,7 +49,7 @@ def decode_jwt_token(app_code: str | None = None, user_id: str | None = None):
        decoded = PassportService().verify(tk)
        app_code = decoded.get("app_code")
        app_id = decoded.get("app_id")
-        with Session(db.engine, expire_on_commit=False) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
            app_model = session.scalar(select(App).where(App.id == app_id))
            site = session.scalar(select(Site).where(Site.code == app_code))
            if not app_model:
@@ -18,7 +18,7 @@ from graphon.model_runtime.entities import (
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from graphon.model_runtime.entities.model_entities import ModelFeature
 from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from sqlalchemy import select
+from sqlalchemy import func, select

 from core.agent.entities import AgentEntity, AgentToolEntity
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
@@ -104,11 +104,14 @@ class BaseAgentRunner(AppRunner):
        )
        # get how many agent thoughts have been created
        self.agent_thought_count = (
-            db.session.query(MessageAgentThought)
-            .where(
-                MessageAgentThought.message_id == self.message.id,
+            db.session.scalar(
+                select(func.count())
+                .select_from(MessageAgentThought)
+                .where(
+                    MessageAgentThought.message_id == self.message.id,
+                )
            )
-            .count()
+            or 0
        )
        db.session.close()

@@ -302,7 +302,7 @@ class PipelineGenerator(BaseAppGenerator):
        """
        with preserve_flask_contexts(flask_app, context_vars=context):
            # init queue manager
-            workflow = db.session.query(Workflow).where(Workflow.id == workflow_id).first()
+            workflow = db.session.get(Workflow, workflow_id)
            if not workflow:
                raise ValueError(f"Workflow not found: {workflow_id}")
            queue_manager = PipelineQueueManager(
@@ -9,6 +9,7 @@ from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
 from graphon.runtime import GraphRuntimeState, VariablePool
 from graphon.variable_loader import VariableLoader
 from graphon.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
+from sqlalchemy import select

 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.pipeline.pipeline_config_manager import PipelineConfig
@@ -84,13 +85,13 @@ class PipelineRunner(WorkflowBasedAppRunner):

        user_id = None
        if invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
-            end_user = db.session.query(EndUser).where(EndUser.id == self.application_generate_entity.user_id).first()
+            end_user = db.session.get(EndUser, self.application_generate_entity.user_id)
            if end_user:
                user_id = end_user.session_id
        else:
            user_id = self.application_generate_entity.user_id

-        pipeline = db.session.query(Pipeline).where(Pipeline.id == app_config.app_id).first()
+        pipeline = db.session.get(Pipeline, app_config.app_id)
        if not pipeline:
            raise ValueError("Pipeline not found")

@@ -213,10 +214,10 @@ class PipelineRunner(WorkflowBasedAppRunner):
        Get workflow
        """
        # fetch workflow by workflow_id
-        workflow = (
-            db.session.query(Workflow)
+        workflow = db.session.scalar(
+            select(Workflow)
            .where(Workflow.tenant_id == pipeline.tenant_id, Workflow.app_id == pipeline.id, Workflow.id == workflow_id)
-            .first()
+            .limit(1)
        )

        # return workflow
@@ -297,10 +298,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
        """
        if isinstance(event, GraphRunFailedEvent):
            if document_id and dataset_id:
-                document = (
-                    db.session.query(Document)
-                    .where(Document.id == document_id, Document.dataset_id == dataset_id)
-                    .first()
+                document = db.session.scalar(
+                    select(Document).where(Document.id == document_id, Document.dataset_id == dataset_id).limit(1)
                )
                if document:
                    document.indexing_status = "error"
@@ -81,7 +81,7 @@ def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LL
                        # TODO: Use provider name with prefix after the data migration.
                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
                        Provider.provider_type == ProviderType.SYSTEM.value,
-                        Provider.quota_type == system_configuration.current_quota_type.value,
+                        Provider.quota_type == system_configuration.current_quota_type,
                        Provider.quota_limit > Provider.quota_used,
                    )
                    .values(
@@ -8,6 +8,7 @@ associates with the node span.
 """

 import logging
+from contextvars import Token
 from dataclasses import dataclass
 from typing import cast, final

@@ -35,7 +36,7 @@ logger = logging.getLogger(__name__)
@dataclass(slots=True)
 class _NodeSpanContext:
    span: "Span"
-    token: object
+    token: Token[context_api.Context]


@final
@@ -1,7 +1,7 @@
 import logging
 from collections.abc import Sequence

-from sqlalchemy import select
+from sqlalchemy import select, update

 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -70,23 +70,21 @@ class DatasetIndexToolCallbackHandler:
                    )
                    child_chunk = db.session.scalar(child_chunk_stmt)
                    if child_chunk:
-                        _ = (
-                            db.session.query(DocumentSegment)
+                        db.session.execute(
+                            update(DocumentSegment)
                            .where(DocumentSegment.id == child_chunk.segment_id)
-                            .update(
-                                {DocumentSegment.hit_count: DocumentSegment.hit_count + 1}, synchronize_session=False
-                            )
+                            .values(hit_count=DocumentSegment.hit_count + 1)
                        )
                else:
-                    query = db.session.query(DocumentSegment).where(
-                        DocumentSegment.index_node_id == document.metadata["doc_id"]
-                    )
+                    conditions = [DocumentSegment.index_node_id == document.metadata["doc_id"]]

                    if "dataset_id" in document.metadata:
-                        query = query.where(DocumentSegment.dataset_id == document.metadata["dataset_id"])
+                        conditions.append(DocumentSegment.dataset_id == document.metadata["dataset_id"])

                    # add hit count to document segment
-                    query.update({DocumentSegment.hit_count: DocumentSegment.hit_count + 1}, synchronize_session=False)
+                    db.session.execute(
+                        update(DocumentSegment).where(*conditions).values(hit_count=DocumentSegment.hit_count + 1)
+                    )

                db.session.commit()

@@ -153,7 +153,7 @@ class DatasourceFileManager:

        :return: the binary of the file, mime type
        """
-        upload_file: UploadFile | None = db.session.query(UploadFile).where(UploadFile.id == id).first()
+        upload_file: UploadFile | None = db.session.get(UploadFile, id)

        if not upload_file:
            return None
@@ -171,7 +171,7 @@ class DatasourceFileManager:

        :return: the binary of the file, mime type
        """
-        message_file: MessageFile | None = db.session.query(MessageFile).where(MessageFile.id == id).first()
+        message_file: MessageFile | None = db.session.get(MessageFile, id)

        # Check if message_file is not None
        if message_file is not None:
@@ -185,7 +185,7 @@ class DatasourceFileManager:
        else:
            tool_file_id = None

-        tool_file: ToolFile | None = db.session.query(ToolFile).where(ToolFile.id == tool_file_id).first()
+        tool_file: ToolFile | None = db.session.get(ToolFile, tool_file_id)

        if not tool_file:
            return None
@@ -203,7 +203,7 @@ class DatasourceFileManager:

        :return: the binary of the file, mime type
        """
-        upload_file: UploadFile | None = db.session.query(UploadFile).where(UploadFile.id == upload_file_id).first()
+        upload_file: UploadFile | None = db.session.get(UploadFile, upload_file_id)

        if not upload_file:
            return None, None
@@ -403,7 +403,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            ),
        )

@@ -753,7 +753,7 @@ class ProviderConfiguration(BaseModel):
            ProviderModel.tenant_id == self.tenant_id,
            ProviderModel.provider_name.in_(provider_names),
            ProviderModel.model_name == model,
-            ProviderModel.model_type == model_type.to_origin_model_type(),
+            ProviderModel.model_type == model_type,
        )

        return session.execute(stmt).scalar_one_or_none()
@@ -778,7 +778,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            )

            credential_record = session.execute(stmt).scalar_one_or_none()
@@ -825,7 +825,7 @@ class ProviderConfiguration(BaseModel):
            ProviderModelCredential.tenant_id == self.tenant_id,
            ProviderModelCredential.provider_name.in_(self._get_provider_names()),
            ProviderModelCredential.model_name == model,
-            ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+            ProviderModelCredential.model_type == model_type,
            ProviderModelCredential.credential_name == credential_name,
        )
        if exclude_id:
@@ -901,7 +901,7 @@ class ProviderConfiguration(BaseModel):
                        ProviderModelCredential.tenant_id == self.tenant_id,
                        ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                        ProviderModelCredential.model_name == model,
-                        ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                        ProviderModelCredential.model_type == model_type,
                    )
                    credential_record = s.execute(stmt).scalar_one_or_none()
                    original_credentials = (
@@ -970,7 +970,7 @@ class ProviderConfiguration(BaseModel):
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
                    model_name=model,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    encrypted_config=json.dumps(credentials),
                    credential_name=credential_name,
                )
@@ -983,7 +983,7 @@ class ProviderConfiguration(BaseModel):
                        tenant_id=self.tenant_id,
                        provider_name=self.provider.provider,
                        model_name=model,
-                        model_type=model_type.to_origin_model_type(),
+                        model_type=model_type,
                        credential_id=credential.id,
                        is_valid=True,
                    )
@@ -1038,7 +1038,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            )
            credential_record = session.execute(stmt).scalar_one_or_none()
            if not credential_record:
@@ -1083,7 +1083,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            )
            credential_record = session.execute(stmt).scalar_one_or_none()
            if not credential_record:
@@ -1116,7 +1116,7 @@ class ProviderConfiguration(BaseModel):
                    ProviderModelCredential.tenant_id == self.tenant_id,
                    ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                    ProviderModelCredential.model_name == model,
-                    ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                    ProviderModelCredential.model_type == model_type,
                )
                available_credentials_count = session.execute(count_stmt).scalar() or 0
                session.delete(credential_record)
@@ -1156,7 +1156,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            )
            credential_record = session.execute(stmt).scalar_one_or_none()
            if not credential_record:
@@ -1171,7 +1171,7 @@ class ProviderConfiguration(BaseModel):
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
                    model_name=model,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    is_valid=True,
                    credential_id=credential_id,
                )
@@ -1207,7 +1207,7 @@ class ProviderConfiguration(BaseModel):
                ProviderModelCredential.tenant_id == self.tenant_id,
                ProviderModelCredential.provider_name.in_(self._get_provider_names()),
                ProviderModelCredential.model_name == model,
-                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+                ProviderModelCredential.model_type == model_type,
            )
            credential_record = session.execute(stmt).scalar_one_or_none()
            if not credential_record:
@@ -1263,7 +1263,7 @@ class ProviderConfiguration(BaseModel):
        stmt = select(ProviderModelSetting).where(
            ProviderModelSetting.tenant_id == self.tenant_id,
            ProviderModelSetting.provider_name.in_(self._get_provider_names()),
-            ProviderModelSetting.model_type == model_type.to_origin_model_type(),
+            ProviderModelSetting.model_type == model_type,
            ProviderModelSetting.model_name == model,
        )
        return session.execute(stmt).scalars().first()
@@ -1286,7 +1286,7 @@ class ProviderConfiguration(BaseModel):
                model_setting = ProviderModelSetting(
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    model_name=model,
                    enabled=True,
                )
@@ -1312,7 +1312,7 @@ class ProviderConfiguration(BaseModel):
                model_setting = ProviderModelSetting(
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    model_name=model,
                    enabled=False,
                )
@@ -1348,7 +1348,7 @@ class ProviderConfiguration(BaseModel):
            stmt = select(func.count(LoadBalancingModelConfig.id)).where(
                LoadBalancingModelConfig.tenant_id == self.tenant_id,
                LoadBalancingModelConfig.provider_name.in_(provider_names),
-                LoadBalancingModelConfig.model_type == model_type.to_origin_model_type(),
+                LoadBalancingModelConfig.model_type == model_type,
                LoadBalancingModelConfig.model_name == model,
            )
            load_balancing_config_count = session.execute(stmt).scalar() or 0
@@ -1364,7 +1364,7 @@ class ProviderConfiguration(BaseModel):
                model_setting = ProviderModelSetting(
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    model_name=model,
                    load_balancing_enabled=True,
                )
@@ -1391,7 +1391,7 @@ class ProviderConfiguration(BaseModel):
                model_setting = ProviderModelSetting(
                    tenant_id=self.tenant_id,
                    provider_name=self.provider.provider,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    model_name=model,
                    load_balancing_enabled=False,
                )
@@ -17,7 +17,7 @@ class CSVSanitizer:
    """

    # Characters that can start a formula in Excel/LibreOffice/Google Sheets
-    FORMULA_CHARS = frozenset({"=", "+", "-", "@", "\t", "\r"})
+    FORMULA_CHARS = frozenset(("=", "+", "-", "@", "\t", "\r"))

    @classmethod
    def sanitize_value(cls, value: Any) -> str:
@@ -19,7 +19,7 @@ def encrypt_token(tenant_id: str, token: str):
    from extensions.ext_database import db
    from models.account import Tenant

-    if not (tenant := db.session.query(Tenant).where(Tenant.id == tenant_id).first()):
+    if not (tenant := db.session.get(Tenant, tenant_id)):
        raise ValueError(f"Tenant with id {tenant_id} not found")
    assert tenant.encrypt_public_key is not None
    encrypted_token = rsa.encrypt(token, tenant.encrypt_public_key)
@@ -10,7 +10,7 @@ from typing import Any

 from flask import Flask, current_app
 from graphon.model_runtime.entities.model_entities import ModelType
-from sqlalchemy import select
+from sqlalchemy import delete, func, select, update
 from sqlalchemy.orm.exc import ObjectDeletedError

 from configs import dify_config
@@ -78,7 +78,7 @@ class IndexingRunner:
                    continue

                # get dataset
-                dataset = db.session.query(Dataset).filter_by(id=requeried_document.dataset_id).first()
+                dataset = db.session.get(Dataset, requeried_document.dataset_id)

                if not dataset:
                    raise ValueError("no dataset found")
@@ -95,7 +95,7 @@ class IndexingRunner:
                text_docs = self._extract(index_processor, requeried_document, processing_rule.to_dict())

                # transform
-                current_user = db.session.query(Account).filter_by(id=requeried_document.created_by).first()
+                current_user = db.session.get(Account, requeried_document.created_by)
                if not current_user:
                    raise ValueError("no current user found")
                current_user.set_tenant_id(dataset.tenant_id)
@@ -137,23 +137,24 @@ class IndexingRunner:
                return

            # get dataset
-            dataset = db.session.query(Dataset).filter_by(id=requeried_document.dataset_id).first()
+            dataset = db.session.get(Dataset, requeried_document.dataset_id)

            if not dataset:
                raise ValueError("no dataset found")

            # get exist document_segment list and delete
-            document_segments = (
-                db.session.query(DocumentSegment)
-                .filter_by(dataset_id=dataset.id, document_id=requeried_document.id)
-                .all()
-            )
+            document_segments = db.session.scalars(
+                select(DocumentSegment).where(
+                    DocumentSegment.dataset_id == dataset.id,
+                    DocumentSegment.document_id == requeried_document.id,
+                )
+            ).all()

            for document_segment in document_segments:
                db.session.delete(document_segment)
                if requeried_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX:
                    # delete child chunks
-                    db.session.query(ChildChunk).where(ChildChunk.segment_id == document_segment.id).delete()
+                    db.session.execute(delete(ChildChunk).where(ChildChunk.segment_id == document_segment.id))
            db.session.commit()
            # get the process rule
            stmt = select(DatasetProcessRule).where(DatasetProcessRule.id == requeried_document.dataset_process_rule_id)
@@ -167,7 +168,7 @@ class IndexingRunner:
            text_docs = self._extract(index_processor, requeried_document, processing_rule.to_dict())

            # transform
-            current_user = db.session.query(Account).filter_by(id=requeried_document.created_by).first()
+            current_user = db.session.get(Account, requeried_document.created_by)
            if not current_user:
                raise ValueError("no current user found")
            current_user.set_tenant_id(dataset.tenant_id)
@@ -207,17 +208,18 @@ class IndexingRunner:
                return

            # get dataset
-            dataset = db.session.query(Dataset).filter_by(id=requeried_document.dataset_id).first()
+            dataset = db.session.get(Dataset, requeried_document.dataset_id)

            if not dataset:
                raise ValueError("no dataset found")

            # get exist document_segment list and delete
-            document_segments = (
-                db.session.query(DocumentSegment)
-                .filter_by(dataset_id=dataset.id, document_id=requeried_document.id)
-                .all()
-            )
+            document_segments = db.session.scalars(
+                select(DocumentSegment).where(
+                    DocumentSegment.dataset_id == dataset.id,
+                    DocumentSegment.document_id == requeried_document.id,
+                )
+            ).all()

            documents = []
            if document_segments:
@@ -289,7 +291,7 @@ class IndexingRunner:

        embedding_model_instance = None
        if dataset_id:
-            dataset = db.session.query(Dataset).filter_by(id=dataset_id).first()
+            dataset = db.session.get(Dataset, dataset_id)
            if not dataset:
                raise ValueError("Dataset not found.")
            if IndexTechniqueType.HIGH_QUALITY in {dataset.indexing_technique, indexing_technique}:
@@ -652,24 +654,26 @@ class IndexingRunner:
    @staticmethod
    def _process_keyword_index(flask_app, dataset_id, document_id, documents):
        with flask_app.app_context():
-            dataset = db.session.query(Dataset).filter_by(id=dataset_id).first()
+            dataset = db.session.get(Dataset, dataset_id)
            if not dataset:
                raise ValueError("no dataset found")
            keyword = Keyword(dataset)
            keyword.create(documents)
            if dataset.indexing_technique != IndexTechniqueType.HIGH_QUALITY:
                document_ids = [document.metadata["doc_id"] for document in documents]
-                db.session.query(DocumentSegment).where(
-                    DocumentSegment.document_id == document_id,
-                    DocumentSegment.dataset_id == dataset_id,
-                    DocumentSegment.index_node_id.in_(document_ids),
-                    DocumentSegment.status == SegmentStatus.INDEXING,
-                ).update(
-                    {
-                        DocumentSegment.status: SegmentStatus.COMPLETED,
-                        DocumentSegment.enabled: True,
-                        DocumentSegment.completed_at: naive_utc_now(),
-                    }
+                db.session.execute(
+                    update(DocumentSegment)
+                    .where(
+                        DocumentSegment.document_id == document_id,
+                        DocumentSegment.dataset_id == dataset_id,
+                        DocumentSegment.index_node_id.in_(document_ids),
+                        DocumentSegment.status == SegmentStatus.INDEXING,
+                    )
+                    .values(
+                        status=SegmentStatus.COMPLETED,
+                        enabled=True,
+                        completed_at=naive_utc_now(),
+                    )
                )

                db.session.commit()
@@ -703,17 +707,19 @@ class IndexingRunner:
            )

            document_ids = [document.metadata["doc_id"] for document in chunk_documents]
-            db.session.query(DocumentSegment).where(
-                DocumentSegment.document_id == dataset_document.id,
-                DocumentSegment.dataset_id == dataset.id,
-                DocumentSegment.index_node_id.in_(document_ids),
-                DocumentSegment.status == SegmentStatus.INDEXING,
-            ).update(
-                {
-                    DocumentSegment.status: SegmentStatus.COMPLETED,
-                    DocumentSegment.enabled: True,
-                    DocumentSegment.completed_at: naive_utc_now(),
-                }
+            db.session.execute(
+                update(DocumentSegment)
+                .where(
+                    DocumentSegment.document_id == dataset_document.id,
+                    DocumentSegment.dataset_id == dataset.id,
+                    DocumentSegment.index_node_id.in_(document_ids),
+                    DocumentSegment.status == SegmentStatus.INDEXING,
+                )
+                .values(
+                    status=SegmentStatus.COMPLETED,
+                    enabled=True,
+                    completed_at=naive_utc_now(),
+                )
            )

            db.session.commit()
@@ -734,10 +740,17 @@ class IndexingRunner:
        """
        Update the document indexing status.
        """
-        count = db.session.query(DatasetDocument).filter_by(id=document_id, is_paused=True).count()
+        count = (
+            db.session.scalar(
+                select(func.count())
+                .select_from(DatasetDocument)
+                .where(DatasetDocument.id == document_id, DatasetDocument.is_paused == True)
+            )
+            or 0
+        )
        if count > 0:
            raise DocumentIsPausedError()
-        document = db.session.query(DatasetDocument).filter_by(id=document_id).first()
+        document = db.session.get(DatasetDocument, document_id)
        if not document:
            raise DocumentIsDeletedPausedError()

@@ -745,7 +758,7 @@ class IndexingRunner:

        if extra_update_params:
            update_params.update(extra_update_params)
-        db.session.query(DatasetDocument).filter_by(id=document_id).update(update_params)  # type: ignore
+        db.session.execute(update(DatasetDocument).where(DatasetDocument.id == document_id).values(update_params))  # type: ignore
        db.session.commit()

    @staticmethod
@@ -753,7 +766,9 @@ class IndexingRunner:
        """
        Update the document segment by document id.
        """
-        db.session.query(DocumentSegment).filter_by(document_id=dataset_document_id).update(update_params)
+        db.session.execute(
+            update(DocumentSegment).where(DocumentSegment.document_id == dataset_document_id).values(update_params)
+        )
        db.session.commit()

    def _transform(
@@ -10,6 +10,7 @@ from graphon.model_runtime.entities.llm_entities import LLMResult
 from graphon.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
+from sqlalchemy import select

 from core.app.app_config.entities import ModelConfig
 from core.llm_generator.entities import RuleCodeGeneratePayload, RuleGeneratePayload, RuleStructuredOutputPayload
@@ -410,8 +411,8 @@ class LLMGenerator:
        model_config: ModelConfig,
        ideal_output: str | None,
    ):
-        last_run: Message | None = (
-            db.session.query(Message).where(Message.app_id == flow_id).order_by(Message.created_at.desc()).first()
+        last_run: Message | None = db.session.scalar(
+            select(Message).where(Message.app_id == flow_id).order_by(Message.created_at.desc()).limit(1)
        )
        if not last_run:
            return LLMGenerator.__instruction_modify_common(
@@ -260,4 +260,12 @@ def convert_input_form_to_parameters(
            parameters[item.variable]["enum"] = item.options
        elif item.type == VariableEntityType.NUMBER:
            parameters[item.variable]["type"] = "number"
+        elif item.type == VariableEntityType.CHECKBOX:
+            parameters[item.variable]["type"] = "boolean"
+        elif item.type == VariableEntityType.JSON_OBJECT:
+            parameters[item.variable]["type"] = "object"
+            if item.json_schema:
+                for key in ("properties", "required", "additionalProperties"):
+                    if key in item.json_schema:
+                        parameters[item.variable][key] = item.json_schema[key]
    return parameters, required
@@ -16,7 +16,13 @@ from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExport
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
-from opentelemetry.semconv.resource import ResourceAttributes
+from opentelemetry.semconv._incubating.attributes.deployment_attributes import (  # type: ignore[import-untyped]
+    DEPLOYMENT_ENVIRONMENT,
+)
+from opentelemetry.semconv._incubating.attributes.host_attributes import (  # type: ignore[import-untyped]
+    HOST_NAME,
+)
+from opentelemetry.semconv.attributes import service_attributes
 from opentelemetry.trace import Link, SpanContext, TraceFlags

 from configs import dify_config
@@ -45,10 +51,10 @@ class TraceClient:
        self.endpoint = endpoint
        self.resource = Resource(
            attributes={
-                ResourceAttributes.SERVICE_NAME: service_name,
-                ResourceAttributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
-                ResourceAttributes.DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
-                ResourceAttributes.HOST_NAME: socket.gethostname(),
+                service_attributes.SERVICE_NAME: service_name,
+                service_attributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
+                DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
+                HOST_NAME: socket.gethostname(),
                ACS_ARMS_SERVICE_FEATURE: "genai_app",
            }
        )
@@ -19,7 +19,7 @@ from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExport
 from opentelemetry.sdk import trace as trace_sdk
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
-from opentelemetry.semconv.trace import SpanAttributes as OTELSpanAttributes
+from opentelemetry.semconv.attributes import exception_attributes
 from opentelemetry.trace import Span, Status, StatusCode, set_span_in_context, use_span
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from opentelemetry.util.types import AttributeValue
@@ -38,6 +38,7 @@ from core.ops.entities.trace_entity import (
    TraceTaskName,
    WorkflowTraceInfo,
 )
+from core.ops.utils import JSON_DICT_ADAPTER
 from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
 from models.model import EndUser, MessageFile
@@ -134,10 +135,10 @@ def set_span_status(current_span: Span, error: Exception | str | None = None):
            if not exception_message:
                exception_message = repr(error)
            attributes: dict[str, AttributeValue] = {
-                OTELSpanAttributes.EXCEPTION_TYPE: exception_type,
-                OTELSpanAttributes.EXCEPTION_MESSAGE: exception_message,
-                OTELSpanAttributes.EXCEPTION_ESCAPED: False,
-                OTELSpanAttributes.EXCEPTION_STACKTRACE: error_string,
+                exception_attributes.EXCEPTION_TYPE: exception_type,
+                exception_attributes.EXCEPTION_MESSAGE: exception_message,
+                exception_attributes.EXCEPTION_ESCAPED: False,
+                exception_attributes.EXCEPTION_STACKTRACE: error_string,
            }
            current_span.add_event(name="exception", attributes=attributes)
    else:
@@ -469,7 +470,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                llm_attributes[SpanAttributes.LLM_PROVIDER] = trace_info.message_data.model_provider

            if trace_info.message_data and trace_info.message_data.message_metadata:
-                metadata_dict = json.loads(trace_info.message_data.message_metadata)
+                metadata_dict = JSON_DICT_ADAPTER.validate_json(trace_info.message_data.message_metadata)
                if model_params := metadata_dict.get("model_parameters"):
                    llm_attributes[SpanAttributes.LLM_INVOCATION_PARAMETERS] = json.dumps(model_params)

@@ -1,9 +1,19 @@
 import logging
 import os
-from datetime import datetime, timedelta
+import uuid
+from datetime import UTC, datetime, timedelta

 from graphon.enums import BuiltinNodeTypes
 from langfuse import Langfuse
+from langfuse.api import (
+    CreateGenerationBody,
+    CreateSpanBody,
+    IngestionEvent_GenerationCreate,
+    IngestionEvent_SpanCreate,
+    IngestionEvent_TraceCreate,
+    TraceBody,
+)
+from langfuse.api.commons.types.usage import Usage
 from sqlalchemy.orm import sessionmaker

 from core.ops.base_trace_instance import BaseTraceInstance
@@ -396,18 +406,61 @@ class LangFuseDataTrace(BaseTraceInstance):
        )
        self.add_span(langfuse_span_data=name_generation_span_data)

+    def _make_event_id(self) -> str:
+        return str(uuid.uuid4())
+
+    def _now_iso(self) -> str:
+        return datetime.now(UTC).isoformat()
+
    def add_trace(self, langfuse_trace_data: LangfuseTrace | None = None):
-        format_trace_data = filter_none_values(langfuse_trace_data.model_dump()) if langfuse_trace_data else {}
+        data = filter_none_values(langfuse_trace_data.model_dump()) if langfuse_trace_data else {}
        try:
-            self.langfuse_client.trace(**format_trace_data)
+            body = TraceBody(
+                id=data.get("id"),
+                name=data.get("name"),
+                user_id=data.get("user_id"),
+                input=data.get("input"),
+                output=data.get("output"),
+                metadata=data.get("metadata"),
+                session_id=data.get("session_id"),
+                version=data.get("version"),
+                release=data.get("release"),
+                tags=data.get("tags"),
+                public=data.get("public"),
+            )
+            event = IngestionEvent_TraceCreate(
+                body=body,
+                id=self._make_event_id(),
+                timestamp=self._now_iso(),
+            )
+            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Trace created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create trace: {str(e)}")

    def add_span(self, langfuse_span_data: LangfuseSpan | None = None):
-        format_span_data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
+        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
-            self.langfuse_client.span(**format_span_data)
+            body = CreateSpanBody(
+                id=data.get("id"),
+                trace_id=data.get("trace_id"),
+                name=data.get("name"),
+                start_time=data.get("start_time"),
+                end_time=data.get("end_time"),
+                input=data.get("input"),
+                output=data.get("output"),
+                metadata=data.get("metadata"),
+                level=data.get("level"),
+                status_message=data.get("status_message"),
+                parent_observation_id=data.get("parent_observation_id"),
+                version=data.get("version"),
+            )
+            event = IngestionEvent_SpanCreate(
+                body=body,
+                id=self._make_event_id(),
+                timestamp=self._now_iso(),
+            )
+            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Span created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create span: {str(e)}")
@@ -418,11 +471,45 @@ class LangFuseDataTrace(BaseTraceInstance):
        span.end(**format_span_data)

    def add_generation(self, langfuse_generation_data: LangfuseGeneration | None = None):
-        format_generation_data = (
-            filter_none_values(langfuse_generation_data.model_dump()) if langfuse_generation_data else {}
-        )
+        data = filter_none_values(langfuse_generation_data.model_dump()) if langfuse_generation_data else {}
        try:
-            self.langfuse_client.generation(**format_generation_data)
+            usage_data = data.pop("usage", None)
+            usage = None
+            if usage_data:
+                usage = Usage(
+                    input=usage_data.get("input", 0) or 0,
+                    output=usage_data.get("output", 0) or 0,
+                    total=usage_data.get("total", 0) or 0,
+                    unit=usage_data.get("unit"),
+                    input_cost=usage_data.get("inputCost"),
+                    output_cost=usage_data.get("outputCost"),
+                    total_cost=usage_data.get("totalCost"),
+                )
+
+            body = CreateGenerationBody(
+                id=data.get("id"),
+                trace_id=data.get("trace_id"),
+                name=data.get("name"),
+                start_time=data.get("start_time"),
+                end_time=data.get("end_time"),
+                model=data.get("model"),
+                model_parameters=data.get("model_parameters"),
+                input=data.get("input"),
+                output=data.get("output"),
+                usage=usage,
+                metadata=data.get("metadata"),
+                level=data.get("level"),
+                status_message=data.get("status_message"),
+                parent_observation_id=data.get("parent_observation_id"),
+                version=data.get("version"),
+                completion_start_time=data.get("completion_start_time"),
+            )
+            event = IngestionEvent_GenerationCreate(
+                body=body,
+                id=self._make_event_id(),
+                timestamp=self._now_iso(),
+            )
+            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Generation created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create generation: {str(e)}")
@@ -443,7 +530,7 @@ class LangFuseDataTrace(BaseTraceInstance):

    def get_project_key(self):
        try:
-            projects = self.langfuse_client.client.projects.get()
+            projects = self.langfuse_client.api.projects.get()
            return projects.data[0].id
        except Exception as e:
            logger.debug("LangFuse get project key failed: %s", str(e))
@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 from datetime import datetime, timedelta
@@ -25,6 +24,7 @@ from core.ops.entities.trace_entity import (
    TraceTaskName,
    WorkflowTraceInfo,
 )
+from core.ops.utils import JSON_DICT_ADAPTER
 from extensions.ext_database import db
 from models import EndUser
 from models.workflow import WorkflowNodeExecutionModel
@@ -153,7 +153,7 @@ class MLflowDataTrace(BaseTraceInstance):
                    inputs = node.process_data  # contains request URL

                if not inputs:
-                    inputs = json.loads(node.inputs) if node.inputs else {}
+                    inputs = JSON_DICT_ADAPTER.validate_json(node.inputs) if node.inputs else {}

                node_span = start_span_no_context(
                    name=node.title,
@@ -180,7 +180,7 @@ class MLflowDataTrace(BaseTraceInstance):

                # End node span
                finished_at = node.created_at + timedelta(seconds=node.elapsed_time)
-                outputs = json.loads(node.outputs) if node.outputs else {}
+                outputs = JSON_DICT_ADAPTER.validate_json(node.outputs) if node.outputs else {}
                if node.node_type == BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL:
                    outputs = self._parse_knowledge_retrieval_outputs(outputs)
                elif node.node_type == BuiltinNodeTypes.LLM:
@@ -216,8 +216,8 @@ class MLflowDataTrace(BaseTraceInstance):
            return {}, {}

        try:
-            data = json.loads(node.process_data)
-        except (json.JSONDecodeError, TypeError):
+            data = JSON_DICT_ADAPTER.validate_json(node.process_data)
+        except (ValueError, TypeError):
            return {}, {}

        inputs = self._parse_prompts(data.get("prompts"))
@@ -11,8 +11,10 @@ from uuid import UUID, uuid4

 from cachetools import LRUCache
 from flask import current_app
+from pydantic import TypeAdapter
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
+from typing_extensions import TypedDict

 from core.helper.encrypter import batch_decrypt_token, encrypt_token, obfuscated_token
 from core.ops.entities.config_entity import (
@@ -33,7 +35,7 @@ from core.ops.entities.trace_entity import (
    WorkflowNodeTraceInfo,
    WorkflowTraceInfo,
 )
-from core.ops.utils import get_message_data
+from core.ops.utils import JSON_DICT_ADAPTER, get_message_data
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models.account import Tenant
@@ -50,6 +52,14 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)


+class _AppTracingConfig(TypedDict, total=False):
+    enabled: bool
+    tracing_provider: str | None
+
+
+_app_tracing_config_adapter: TypeAdapter[_AppTracingConfig] = TypeAdapter(_AppTracingConfig)
+
+
 def _lookup_app_and_workspace_names(app_id: str | None, tenant_id: str | None) -> tuple[str, str]:
    """Return (app_name, workspace_name) for the given IDs. Falls back to empty strings."""
    app_name = ""
@@ -468,7 +478,7 @@ class OpsTraceManager:
        if app is None:
            return None

-        app_ops_trace_config = json.loads(app.tracing) if app.tracing else None
+        app_ops_trace_config = _app_tracing_config_adapter.validate_json(app.tracing) if app.tracing else None
        if app_ops_trace_config is None:
            return None
        if not app_ops_trace_config.get("enabled"):
@@ -560,7 +570,7 @@ class OpsTraceManager:
            raise ValueError("App not found")
        if not app.tracing:
            return {"enabled": False, "tracing_provider": None}
-        app_trace_config = json.loads(app.tracing)
+        app_trace_config = _app_tracing_config_adapter.validate_json(app.tracing)
        return app_trace_config

    @staticmethod
@@ -636,7 +646,6 @@ class TraceTask:
        carries ``total_tokens``.  Projects only the ``outputs`` column to avoid loading
        large JSON blobs unnecessarily.
        """
-        import json

        from models.workflow import WorkflowNodeExecutionModel

@@ -658,7 +667,7 @@ class TraceTask:
            if not raw:
                continue
            try:
-                outputs = json.loads(raw) if isinstance(raw, str) else raw
+                outputs = JSON_DICT_ADAPTER.validate_json(raw) if isinstance(raw, str) else raw
            except (ValueError, TypeError):
                continue
            if not isinstance(outputs, dict):
@@ -1420,7 +1429,7 @@ class TraceTask:
            return {}

        try:
-            metadata = json.loads(message_data.message_metadata)
+            metadata = JSON_DICT_ADAPTER.validate_json(message_data.message_metadata)
            usage = metadata.get("usage", {})
            time_to_first_token = usage.get("time_to_first_token")
            time_to_generate = usage.get("time_to_generate")
@@ -1430,7 +1439,7 @@ class TraceTask:
                "llm_streaming_time_to_generate": time_to_generate,
                "is_streaming_request": time_to_first_token is not None,
            }
-        except (json.JSONDecodeError, AttributeError):
+        except (ValueError, AttributeError):
            return {}


@@ -26,7 +26,13 @@ from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExport
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
-from opentelemetry.semconv.resource import ResourceAttributes
+from opentelemetry.semconv._incubating.attributes.deployment_attributes import (  # type: ignore[import-untyped]
+    DEPLOYMENT_ENVIRONMENT,
+)
+from opentelemetry.semconv._incubating.attributes.host_attributes import (  # type: ignore[import-untyped]
+    HOST_NAME,
+)
+from opentelemetry.semconv.attributes import service_attributes
 from opentelemetry.trace import SpanKind
 from opentelemetry.util.types import AttributeValue

@@ -73,13 +79,13 @@ class TencentTraceClient:

        self.resource = Resource(
            attributes={
-                ResourceAttributes.SERVICE_NAME: service_name,
-                ResourceAttributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
-                ResourceAttributes.DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
-                ResourceAttributes.HOST_NAME: socket.gethostname(),
-                ResourceAttributes.TELEMETRY_SDK_LANGUAGE: "python",
-                ResourceAttributes.TELEMETRY_SDK_NAME: "opentelemetry",
-                ResourceAttributes.TELEMETRY_SDK_VERSION: _get_opentelemetry_sdk_version(),
+                service_attributes.SERVICE_NAME: service_name,
+                service_attributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
+                DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
+                HOST_NAME: socket.gethostname(),
+                "telemetry.sdk.language": "python",
+                "telemetry.sdk.name": "opentelemetry",
+                "telemetry.sdk.version": _get_opentelemetry_sdk_version(),
            }
        )
        # Prepare gRPC endpoint/metadata
@@ -3,11 +3,14 @@ from datetime import datetime
 from typing import Any, Union
 from urllib.parse import urlparse

+from pydantic import TypeAdapter
 from sqlalchemy import select

 from models.engine import db
 from models.model import Message

+JSON_DICT_ADAPTER: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any])
+

 def filter_none_values(data: dict[str, Any]) -> dict[str, Any]:
    new_data = {}
@@ -227,7 +227,7 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
        get app
        """
        try:
-            app = db.session.query(App).where(App.id == app_id).where(App.tenant_id == tenant_id).first()
+            app = db.session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id).limit(1))
        except Exception:
            raise ValueError("app not found")

@@ -17,6 +17,7 @@ from pydantic import BaseModel
 from yarl import URL

 from configs import dify_config
+from core.helper.http_client_pooling import get_pooled_http_client
 from core.plugin.endpoint.exc import EndpointSetupFailedError
 from core.plugin.entities.plugin_daemon import PluginDaemonBasicResponse, PluginDaemonError, PluginDaemonInnerError
 from core.plugin.impl.exc import (
@@ -54,6 +55,11 @@ T = TypeVar("T", bound=(BaseModel | dict[str, Any] | list[Any] | bool | str))

 logger = logging.getLogger(__name__)

+_httpx_client: httpx.Client = get_pooled_http_client(
+    "plugin_daemon",
+    lambda: httpx.Client(limits=httpx.Limits(max_keepalive_connections=50, max_connections=100), trust_env=False),
+)
+

 class BasePluginClient:
    def _request(
@@ -84,7 +90,7 @@ class BasePluginClient:
            request_kwargs["content"] = prepared_data

        try:
-            response = httpx.request(**request_kwargs)
+            response = _httpx_client.request(**request_kwargs)
        except httpx.RequestError:
            logger.exception("Request to Plugin Daemon Service failed")
            raise PluginDaemonInnerError(code=-500, message="Request to Plugin Daemon Service failed")
@@ -171,7 +177,7 @@ class BasePluginClient:
            stream_kwargs["content"] = prepared_data

        try:
-            with httpx.stream(**stream_kwargs) as response:
+            with _httpx_client.stream(**stream_kwargs) as response:
                for raw_line in response.iter_lines():
                    if not raw_line:
                        continue
@@ -306,7 +306,7 @@ class ProviderManager:
        """
        stmt = select(TenantDefaultModel).where(
            TenantDefaultModel.tenant_id == tenant_id,
-            TenantDefaultModel.model_type == model_type.to_origin_model_type(),
+            TenantDefaultModel.model_type == model_type,
        )
        default_model = db.session.scalar(stmt)

@@ -324,7 +324,7 @@ class ProviderManager:

                default_model = TenantDefaultModel(
                    tenant_id=tenant_id,
-                    model_type=model_type.to_origin_model_type(),
+                    model_type=model_type,
                    provider_name=available_model.provider.provider,
                    model_name=available_model.model,
                )
@@ -391,7 +391,7 @@ class ProviderManager:
            raise ValueError(f"Model {model} does not exist.")
        stmt = select(TenantDefaultModel).where(
            TenantDefaultModel.tenant_id == tenant_id,
-            TenantDefaultModel.model_type == model_type.to_origin_model_type(),
+            TenantDefaultModel.model_type == model_type,
        )
        default_model = db.session.scalar(stmt)

@@ -405,7 +405,7 @@ class ProviderManager:
            # create default model
            default_model = TenantDefaultModel(
                tenant_id=tenant_id,
-                model_type=model_type.to_origin_model_type(),
+                model_type=model_type,
                provider_name=provider,
                model_name=model,
            )
@@ -626,9 +626,8 @@ class ProviderManager:
                if provider_record.provider_type != ProviderType.SYSTEM:
                    continue

-                provider_quota_to_provider_record_dict[ProviderQuotaType.value_of(provider_record.quota_type)] = (
-                    provider_record
-                )
+                if provider_record.quota_type is not None:
+                    provider_quota_to_provider_record_dict[provider_record.quota_type] = provider_record

            for quota in configuration.quotas:
                if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
@@ -641,7 +640,7 @@ class ProviderManager:
                                # TODO: Use provider name with prefix after the data migration.
                                provider_name=ModelProviderID(provider_name).provider_name,
                                provider_type=ProviderType.SYSTEM,
-                                quota_type=quota.quota_type,
+                                quota_type=quota.quota_type,  # type: ignore[arg-type]
                                quota_limit=0,  # type: ignore
                                quota_used=0,
                                is_valid=True,
@@ -823,7 +822,7 @@ class ProviderManager:
            custom_model_configurations.append(
                CustomModelConfiguration(
                    model=provider_model_record.model_name,
-                    model_type=ModelType.value_of(provider_model_record.model_type),
+                    model_type=provider_model_record.model_type,
                    credentials=provider_model_credentials,
                    current_credential_id=provider_model_record.credential_id,
                    current_credential_name=provider_model_record.credential_name,
@@ -921,9 +920,8 @@ class ProviderManager:
            if provider_record.provider_type != ProviderType.SYSTEM:
                continue

-            quota_type_to_provider_records_dict[ProviderQuotaType.value_of(provider_record.quota_type)] = (
-                provider_record
-            )
+            if provider_record.quota_type is not None:
+                quota_type_to_provider_records_dict[provider_record.quota_type] = provider_record  # type: ignore[index]
        quota_configurations = []

        if dify_config.EDITION == "CLOUD":
@@ -1203,7 +1201,7 @@ class ProviderManager:
            model_settings.append(
                ModelSettings(
                    model=provider_model_setting.model_name,
-                    model_type=ModelType.value_of(provider_model_setting.model_type),
+                    model_type=provider_model_setting.model_type,
                    enabled=provider_model_setting.enabled,
                    load_balancing_enabled=provider_model_setting.load_balancing_enabled,
                    load_balancing_configs=load_balancing_configs if len(load_balancing_configs) > 1 else [],
@@ -97,13 +97,13 @@ class Jieba(BaseKeyword):

        documents = []

-        segment_query_stmt = db.session.query(DocumentSegment).where(
+        segment_query_stmt = select(DocumentSegment).where(
            DocumentSegment.dataset_id == self.dataset.id, DocumentSegment.index_node_id.in_(sorted_chunk_indices)
        )
        if document_ids_filter:
            segment_query_stmt = segment_query_stmt.where(DocumentSegment.document_id.in_(document_ids_filter))

-        segments = db.session.execute(segment_query_stmt).scalars().all()
+        segments = db.session.scalars(segment_query_stmt).all()
        segment_map = {segment.index_node_id: segment for segment in segments}
        for chunk_index in sorted_chunk_indices:
            segment = segment_map.get(chunk_index)
@@ -122,6 +122,6 @@ class JiebaKeywordTableHandler:
            results.add(token)
            sub_tokens = re.findall(r"\w+", token)
            if len(sub_tokens) > 1:
-                results.update({w for w in sub_tokens if w not in list(STOPWORDS)})
+                results.update({w for w in sub_tokens if w not in STOPWORDS})

        return results
@@ -432,10 +432,11 @@ class RetrievalService:
            # Batch query dataset documents
            dataset_documents = {
                doc.id: doc
-                for doc in db.session.query(DatasetDocument)
-                .where(DatasetDocument.id.in_(document_ids))
-                .options(load_only(DatasetDocument.id, DatasetDocument.doc_form, DatasetDocument.dataset_id))
-                .all()
+                for doc in db.session.scalars(
+                    select(DatasetDocument)
+                    .where(DatasetDocument.id.in_(document_ids))
+                    .options(load_only(DatasetDocument.id, DatasetDocument.doc_form, DatasetDocument.dataset_id))
+                ).all()
            }

            valid_dataset_documents = {}
@@ -10,6 +10,7 @@ from mysql.connector import Error as MySQLError
 from pydantic import BaseModel, model_validator

 from configs import dify_config
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -178,9 +179,7 @@ class AlibabaCloudMySQLVector(BaseVector):
            cur.execute(f"SELECT meta, text FROM {self.table_name} WHERE id IN ({placeholders})", ids)
            docs = []
            for record in cur:
-                metadata = record["meta"]
-                if isinstance(metadata, str):
-                    metadata = json.loads(metadata)
+                metadata = parse_metadata_json(record["meta"])
                docs.append(Document(page_content=record["text"], metadata=metadata))
        return docs

@@ -263,15 +262,13 @@ class AlibabaCloudMySQLVector(BaseVector):
                        # similarity = 1 / (1 + distance)
                        similarity = 1.0 / (1.0 + distance)

-                    metadata = record["meta"]
-                    if isinstance(metadata, str):
-                        metadata = json.loads(metadata)
+                    metadata = parse_metadata_json(record["meta"])
                    metadata["score"] = similarity
                    metadata["distance"] = distance

                    if similarity >= score_threshold:
                        docs.append(Document(page_content=record["text"], metadata=metadata))
-                except (ValueError, json.JSONDecodeError) as e:
+                except (ValueError, TypeError) as e:
                    logger.warning("Error processing search result: %s", e)
                    continue

@@ -306,9 +303,7 @@ class AlibabaCloudMySQLVector(BaseVector):
            )
            docs = []
            for record in cur:
-                metadata = record["meta"]
-                if isinstance(metadata, str):
-                    metadata = json.loads(metadata)
+                metadata = parse_metadata_json(record["meta"])
                metadata["score"] = float(record["score"])
                docs.append(Document(page_content=record["text"], metadata=metadata))
        return docs
@@ -8,6 +8,7 @@ _import_err_msg = (
    "please run `pip install alibabacloud_gpdb20160503 alibabacloud_tea_openapi`"
 )

+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.models.document import Document
 from extensions.ext_redis import redis_client

@@ -257,7 +258,7 @@ class AnalyticdbVectorOpenAPI:
        documents = []
        for match in response.body.matches.match:
            if match.score >= score_threshold:
-                metadata = json.loads(match.metadata.get("metadata_"))
+                metadata = parse_metadata_json(match.metadata.get("metadata_"))
                metadata["score"] = match.score
                doc = Document(
                    page_content=match.metadata.get("page_content"),
@@ -294,7 +295,7 @@ class AnalyticdbVectorOpenAPI:
        documents = []
        for match in response.body.matches.match:
            if match.score >= score_threshold:
-                metadata = json.loads(match.metadata.get("metadata_"))
+                metadata = parse_metadata_json(match.metadata.get("metadata_"))
                metadata["score"] = match.score
                doc = Document(
                    page_content=match.metadata.get("page_content"),
@@ -29,6 +29,7 @@ from pymochow.model.table import AnnSearch, BM25SearchRequest, HNSWSearchParams,

 from configs import dify_config
 from core.rag.datasource.vdb.field import Field as VDBField
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -173,15 +174,9 @@ class BaiduVector(BaseVector):
            score = row.get("score", 0.0)
            meta = row_data.get(VDBField.METADATA_KEY, {})

-            # Handle both JSON string and dict formats for backward compatibility
-            if isinstance(meta, str):
-                try:
-                    import json
-
-                    meta = json.loads(meta)
-                except (json.JSONDecodeError, TypeError):
-                    meta = {}
-            elif not isinstance(meta, dict):
+            try:
+                meta = parse_metadata_json(meta)
+            except (ValueError, TypeError):
                meta = {}

            if score >= score_threshold:
@@ -200,7 +195,11 @@ class BaiduVector(BaseVector):
                raise

    def _init_client(self, config) -> MochowClient:
-        config = Configuration(credentials=BceCredentials(config.account, config.api_key), endpoint=config.endpoint)
+        config = Configuration(
+            credentials=BceCredentials(config.account, config.api_key),
+            endpoint=config.endpoint,
+            connection_timeout_in_mills=config.connection_timeout_in_mills,
+        )
        client = MochowClient(config)
        return client

@@ -17,7 +17,7 @@ if TYPE_CHECKING:
    from clickzetta.connector.v0.connection import Connection  # type: ignore

 from configs import dify_config
-from core.rag.datasource.vdb.field import Field
+from core.rag.datasource.vdb.field import Field, parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.embedding.embedding_base import Embeddings
@@ -357,18 +357,19 @@ class ClickzettaVector(BaseVector):
        """
        try:
            if raw_metadata:
-                metadata = json.loads(raw_metadata)
+                # First parse may yield a string (double-encoded JSON) so use json.loads
+                first_pass = json.loads(raw_metadata)

                # Handle double-encoded JSON
-                if isinstance(metadata, str):
-                    metadata = json.loads(metadata)
-
-                # Ensure we have a dict
-                if not isinstance(metadata, dict):
+                if isinstance(first_pass, str):
+                    metadata = parse_metadata_json(first_pass)
+                elif isinstance(first_pass, dict):
+                    metadata = first_pass
+                else:
                    metadata = {}
            else:
                metadata = {}
-        except (json.JSONDecodeError, TypeError):
+        except (json.JSONDecodeError, ValueError, TypeError):
            logger.exception("JSON parsing failed for metadata")
            # Fallback: extract document_id with regex
            doc_id_match = re.search(r'"document_id":\s*"([^"]+)"', raw_metadata or "")
@@ -930,17 +931,18 @@ class ClickzettaVector(BaseVector):
                        # Parse metadata from JSON string (may be double-encoded)
                        try:
                            if row[2]:
-                                metadata = json.loads(row[2])
+                                # First parse may yield a string (double-encoded JSON)
+                                first_pass = json.loads(row[2])

-                                # If result is a string, it's double-encoded JSON - parse again
-                                if isinstance(metadata, str):
-                                    metadata = json.loads(metadata)
-
-                                if not isinstance(metadata, dict):
+                                if isinstance(first_pass, str):
+                                    metadata = parse_metadata_json(first_pass)
+                                elif isinstance(first_pass, dict):
+                                    metadata = first_pass
+                                else:
                                    metadata = {}
                            else:
                                metadata = {}
-                        except (json.JSONDecodeError, TypeError):
+                        except (json.JSONDecodeError, ValueError, TypeError):
                            logger.exception("JSON parsing failed")
                            # Fallback: extract document_id with regex

@@ -1,4 +1,24 @@
 from enum import StrEnum, auto
+from typing import Any
+
+from pydantic import TypeAdapter
+
+_metadata_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any])
+
+
+def parse_metadata_json(raw: Any) -> dict[str, Any]:
+    """Parse metadata from a JSON string or pass through an existing dict.
+
+    Many VDB drivers return metadata as either a JSON string or an already-
+    decoded dict depending on the column type and driver version.
+    """
+    if raw is None or raw in ("", b""):
+        return {}
+    if isinstance(raw, dict):
+        return raw
+    if not isinstance(raw, (str, bytes, bytearray)):
+        return {}
+    return _metadata_adapter.validate_json(raw)


 class Field(StrEnum):
@@ -9,6 +9,7 @@ from psycopg import sql as psql
 from pydantic import BaseModel, model_validator

 from configs import dify_config
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -217,8 +218,7 @@ class HologresVector(BaseVector):
            text = row[2]
            meta = row[3]

-            if isinstance(meta, str):
-                meta = json.loads(meta)
+            meta = parse_metadata_json(meta)

            # Convert distance to similarity score (consistent with pgvector)
            score = 1 - distance
@@ -265,8 +265,7 @@ class HologresVector(BaseVector):
            meta = row[2]
            score = row[-1]  # score is the last column from return_score

-            if isinstance(meta, str):
-                meta = json.loads(meta)
+            meta = parse_metadata_json(meta)

            meta["score"] = score
            docs.append(Document(page_content=text, metadata=meta))
@@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, Any

 from configs import dify_config
 from configs.middleware.vdb.iris_config import IrisVectorConfig
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -269,7 +270,7 @@ class IrisVector(BaseVector):
                if len(row) >= 4:
                    text, meta_str, score = row[1], row[2], float(row[3])
                    if score >= score_threshold:
-                        metadata = json.loads(meta_str) if meta_str else {}
+                        metadata = parse_metadata_json(meta_str)
                        metadata["score"] = score
                        docs.append(Document(page_content=text, metadata=metadata))
            return docs
@@ -384,7 +385,7 @@ class IrisVector(BaseVector):
                    meta_str = row[2]
                    score_value = row[3]

-                    metadata = json.loads(meta_str) if meta_str else {}
+                    metadata = parse_metadata_json(meta_str)
                    # Add score to metadata for hybrid search compatibility
                    score = float(score_value) if score_value is not None else 0.0
                    metadata["score"] = score
@@ -9,6 +9,7 @@ from mo_vector.client import MoVectorClient  # type: ignore
 from pydantic import BaseModel, model_validator

 from configs import dify_config
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -196,11 +197,7 @@ class MatrixoneVector(BaseVector):

        docs = []
        for result in results:
-            metadata = result.metadata
-            if isinstance(metadata, str):
-                import json
-
-                metadata = json.loads(metadata)
+            metadata = parse_metadata_json(result.metadata)
            score = 1 - result.distance
            if score >= score_threshold:
                metadata["score"] = score
@@ -4,7 +4,7 @@ import uuid
 from enum import StrEnum
 from typing import Any

-from clickhouse_connect import get_client
+from clickhouse_connect import get_client  # type: ignore[import-untyped]
 from pydantic import BaseModel

 from configs import dify_config
@@ -10,6 +10,7 @@ from sqlalchemy.dialects.mysql import LONGTEXT
 from sqlalchemy.exc import SQLAlchemyError

 from configs import dify_config
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -366,8 +367,8 @@ class OceanBaseVector(BaseVector):

            # Parse metadata JSON
            try:
-                metadata = json.loads(metadata_str) if isinstance(metadata_str, str) else metadata_str
-            except json.JSONDecodeError:
+                metadata = parse_metadata_json(metadata_str)
+            except (ValueError, TypeError):
                logger.warning("Invalid JSON metadata: %s", metadata_str)
                metadata = {}

@@ -9,7 +9,7 @@ from pydantic import BaseModel, model_validator
 from tablestore import BatchGetRowRequest, TableInBatchGetRowItem

 from configs import dify_config
-from core.rag.datasource.vdb.field import Field
+from core.rag.datasource.vdb.field import Field, parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -73,7 +73,8 @@ class TableStoreVector(BaseVector):
        for item in table_result:
            if item.is_ok and item.row:
                kv = {k: v for k, v, _ in item.row.attribute_columns}
-                docs.append(Document(page_content=kv[Field.CONTENT_KEY], metadata=json.loads(kv[Field.METADATA_KEY])))
+                metadata = parse_metadata_json(kv[Field.METADATA_KEY])
+                docs.append(Document(page_content=kv[Field.CONTENT_KEY], metadata=metadata))
        return docs

    def get_type(self) -> str:
@@ -311,7 +312,7 @@ class TableStoreVector(BaseVector):
                metadata_str = ots_column_map.get(Field.METADATA_KEY)

                vector = json.loads(vector_str) if vector_str else None
-                metadata = json.loads(metadata_str) if metadata_str else {}
+                metadata = parse_metadata_json(metadata_str)

                metadata["score"] = search_hit.score

@@ -371,7 +372,7 @@ class TableStoreVector(BaseVector):
                ots_column_map[col[0]] = col[1]

            metadata_str = ots_column_map.get(Field.METADATA_KEY)
-            metadata = json.loads(metadata_str) if metadata_str else {}
+            metadata = parse_metadata_json(metadata_str)

            vector_str = ots_column_map.get(Field.VECTOR)
            vector = json.loads(vector_str) if vector_str else None
@@ -11,6 +11,7 @@ from tcvectordb.model import index as vdb_index  # type: ignore
 from tcvectordb.model.document import AnnSearch, Filter, KeywordSearch, WeightedRerank  # type: ignore

 from configs import dify_config
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -286,13 +287,10 @@ class TencentVector(BaseVector):
            return docs

        for result in res[0]:
-            meta = result.get(self.field_metadata)
-            if isinstance(meta, str):
-                # Compatible with version 1.1.3 and below.
-                meta = json.loads(meta)
-                score = 1 - result.get("score", 0.0)
-            else:
-                score = result.get("score", 0.0)
+            raw_meta = result.get(self.field_metadata)
+            # Compatible with version 1.1.3 and below: str means old driver.
+            score = (1 - result.get("score", 0.0)) if isinstance(raw_meta, str) else result.get("score", 0.0)
+            meta = parse_metadata_json(raw_meta)
            if score >= score_threshold:
                meta["score"] = score
                doc = Document(page_content=result.get(self.field_text), metadata=meta)
@@ -426,11 +426,10 @@ class TidbOnQdrantVectorFactory(AbstractVectorFactory):
                    TIDB_ON_QDRANT_API_KEY = f"{tidb_auth_binding.account}:{tidb_auth_binding.password}"

                else:
-                    idle_tidb_auth_binding = (
-                        db.session.query(TidbAuthBinding)
+                    idle_tidb_auth_binding = db.session.scalar(
+                        select(TidbAuthBinding)
                        .where(TidbAuthBinding.active == False, TidbAuthBinding.status == "ACTIVE")
                        .limit(1)
-                        .one_or_none()
                    )
                    if idle_tidb_auth_binding:
                        idle_tidb_auth_binding.active = True
@@ -6,11 +6,18 @@ import httpx
 from httpx import DigestAuth

 from configs import dify_config
+from core.helper.http_client_pooling import get_pooled_http_client
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from models.dataset import TidbAuthBinding
 from models.enums import TidbAuthBindingStatus

+# Reuse a pooled HTTP client for all TiDB Cloud requests to minimize connection churn
+_tidb_http_client: httpx.Client = get_pooled_http_client(
+    "tidb:cloud",
+    lambda: httpx.Client(limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)),
+)
+

 class TidbService:
    @staticmethod
@@ -50,7 +57,9 @@ class TidbService:
            "rootPassword": password,
        }

-        response = httpx.post(f"{api_url}/clusters", json=cluster_data, auth=DigestAuth(public_key, private_key))
+        response = _tidb_http_client.post(
+            f"{api_url}/clusters", json=cluster_data, auth=DigestAuth(public_key, private_key)
+        )

        if response.status_code == 200:
            response_data = response.json()
@@ -84,7 +93,9 @@ class TidbService:
        :return: The response from the API.
        """

-        response = httpx.delete(f"{api_url}/clusters/{cluster_id}", auth=DigestAuth(public_key, private_key))
+        response = _tidb_http_client.delete(
+            f"{api_url}/clusters/{cluster_id}", auth=DigestAuth(public_key, private_key)
+        )

        if response.status_code == 200:
            return response.json()
@@ -103,7 +114,7 @@ class TidbService:
        :return: The response from the API.
        """

-        response = httpx.get(f"{api_url}/clusters/{cluster_id}", auth=DigestAuth(public_key, private_key))
+        response = _tidb_http_client.get(f"{api_url}/clusters/{cluster_id}", auth=DigestAuth(public_key, private_key))

        if response.status_code == 200:
            return response.json()
@@ -128,7 +139,7 @@ class TidbService:

        body = {"password": new_password, "builtinRole": "role_admin", "customRoles": []}

-        response = httpx.patch(
+        response = _tidb_http_client.patch(
            f"{api_url}/clusters/{cluster_id}/sqlUsers/{account}",
            json=body,
            auth=DigestAuth(public_key, private_key),
@@ -162,7 +173,9 @@ class TidbService:
        tidb_serverless_list_map = {item.cluster_id: item for item in tidb_serverless_list}
        cluster_ids = [item.cluster_id for item in tidb_serverless_list]
        params = {"clusterIds": cluster_ids, "view": "BASIC"}
-        response = httpx.get(f"{api_url}/clusters:batchGet", params=params, auth=DigestAuth(public_key, private_key))
+        response = _tidb_http_client.get(
+            f"{api_url}/clusters:batchGet", params=params, auth=DigestAuth(public_key, private_key)
+        )

        if response.status_code == 200:
            response_data = response.json()
@@ -223,7 +236,7 @@ class TidbService:
            clusters.append(cluster_data)

        request_body = {"requests": clusters}
-        response = httpx.post(
+        response = _tidb_http_client.post(
            f"{api_url}/clusters:batchCreate", json=request_body, auth=DigestAuth(public_key, private_key)
        )

@@ -9,7 +9,7 @@ from sqlalchemy import text as sql_text
 from sqlalchemy.orm import Session, declarative_base

 from configs import dify_config
-from core.rag.datasource.vdb.field import Field
+from core.rag.datasource.vdb.field import Field, parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -228,7 +228,7 @@ class TiDBVector(BaseVector):
            )
            results = [(row[0], row[1], row[2]) for row in res]
            for meta, text, distance in results:
-                metadata = json.loads(meta)
+                metadata = parse_metadata_json(meta)
                metadata["score"] = 1 - distance
                docs.append(Document(page_content=text, metadata=metadata))
        return docs
@@ -277,7 +277,7 @@ class Vector:
        return self._vector_processor.search_by_vector(query_vector, **kwargs)

    def search_by_file(self, file_id: str, **kwargs: Any) -> list[Document]:
-        upload_file: UploadFile | None = db.session.query(UploadFile).where(UploadFile.id == file_id).first()
+        upload_file: UploadFile | None = db.session.get(UploadFile, file_id)

        if not upload_file:
            return []
@@ -15,6 +15,7 @@ from volcengine.viking_db import (  # type: ignore

 from configs import dify_config
 from core.rag.datasource.vdb.field import Field as vdb_Field
+from core.rag.datasource.vdb.field import parse_metadata_json
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
 from core.rag.datasource.vdb.vector_type import VectorType
@@ -163,7 +164,7 @@ class VikingDBVector(BaseVector):
        for result in results:
            metadata = result.fields.get(vdb_Field.METADATA_KEY)
            if metadata is not None:
-                metadata = json.loads(metadata)
+                metadata = parse_metadata_json(metadata)
                if metadata.get(key) == value:
                    ids.append(result.id)
        return ids
@@ -189,9 +190,7 @@ class VikingDBVector(BaseVector):

        docs = []
        for result in results:
-            metadata = result.fields.get(vdb_Field.METADATA_KEY)
-            if metadata is not None:
-                metadata = json.loads(metadata)
+            metadata = parse_metadata_json(result.fields.get(vdb_Field.METADATA_KEY))
            if result.score >= score_threshold:
                metadata["score"] = result.score
                doc = Document(page_content=result.fields.get(vdb_Field.CONTENT_KEY), metadata=metadata)
@@ -4,7 +4,7 @@ from collections.abc import Sequence
 from typing import Any

 from graphon.model_runtime.entities.model_entities import ModelType
-from sqlalchemy import func, select
+from sqlalchemy import delete, func, select

 from core.model_manager import ModelManager
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
@@ -63,10 +63,8 @@ class DatasetDocumentStore:
        return output

    def add_documents(self, docs: Sequence[Document], allow_update: bool = True, save_child: bool = False):
-        max_position = (
-            db.session.query(func.max(DocumentSegment.position))
-            .where(DocumentSegment.document_id == self._document_id)
-            .scalar()
+        max_position = db.session.scalar(
+            select(func.max(DocumentSegment.position)).where(DocumentSegment.document_id == self._document_id)
        )

        if max_position is None:
@@ -155,12 +153,14 @@ class DatasetDocumentStore:
                )
                if save_child and doc.children:
                    # delete the existing child chunks
-                    db.session.query(ChildChunk).where(
-                        ChildChunk.tenant_id == self._dataset.tenant_id,
-                        ChildChunk.dataset_id == self._dataset.id,
-                        ChildChunk.document_id == self._document_id,
-                        ChildChunk.segment_id == segment_document.id,
-                    ).delete()
+                    db.session.execute(
+                        delete(ChildChunk).where(
+                            ChildChunk.tenant_id == self._dataset.tenant_id,
+                            ChildChunk.dataset_id == self._dataset.id,
+                            ChildChunk.document_id == self._document_id,
+                            ChildChunk.segment_id == segment_document.id,
+                        )
+                    )
                    # add new child chunks
                    for position, child in enumerate(doc.children, start=1):
                        child_segment = ChildChunk(
--- a/Show More
+++ b/Show More