add podman compose middleware helpers

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.github/actions/setup-web/action.yml

@@ -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

.github/workflows/api-tests.yml

@@ -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

.github/workflows/autofix.yml

@@ -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'

.github/workflows/build-push.yml

@@ -12,9 +12,6 @@ on:
     tags:
       - "*"
 
-permissions:
-  contents: read
-
 concurrency:
   group: build-push-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
@@ -26,42 +23,40 @@ env:
   DIFY_API_IMAGE_NAME: ${{ vars.DIFY_API_IMAGE_NAME || 'langgenius/dify-api' }}
 
 jobs:
-  check-changes:
-    runs-on: ubuntu-latest
-    outputs:
-      api: ${{ steps.changes.outputs.api }}
-      web: ${{ steps.changes.outputs.web }}
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-        with:
-          fetch-depth: 2
-          persist-credentials: false
-
-      - name: Check changed files
-        id: changes
-        uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
-        with:
-          filters: |
-            api:
-              - 'api/**'
-              - 'docker/**'
-            web:
-              - 'web/**'
-              - 'docker/**'
-
-  build-api:
-    name: Build API Images
-    needs: check-changes
-    if: github.repository == 'langgenius/dify' && (startsWith(github.ref, 'refs/tags/') || needs.check-changes.outputs.api == 'true')
-    runs-on: ${{ matrix.platform == 'linux/arm64' && 'arm64_runner' || 'ubuntu-latest' }}
+  build:
+    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"
+            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"
+            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"
+            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"
+            artifact_context: "web"
+            build_context: "{{defaultContext}}"
+            file: "web/Dockerfile"
+            platform: linux/arm64
+            runs_on: ubuntu-24.04-arm
 
     steps:
       - name: Prepare
@@ -75,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
 
@@ -85,17 +77,18 @@ jobs:
         id: meta
         uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
         with:
-          images: ${{ env.DIFY_API_IMAGE_NAME }}
+          images: ${{ env[matrix.image_name_env] }}
 
       - name: Build Docker image
         id: build
         uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
         with:
-          context: "{{defaultContext}}:api"
+          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 }}
-          outputs: type=image,name=${{ env.DIFY_API_IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
+          outputs: type=image,name=${{ env[matrix.image_name_env] }},push-by-digest=true,name-canonical=true,push=true
           cache-from: type=gha,scope=${{ matrix.service_name }}
           cache-to: type=gha,mode=max,scope=${{ matrix.service_name }}
 
@@ -110,134 +103,30 @@ jobs:
       - name: Upload digest
         uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
         with:
-          name: digests-api-${{ env.PLATFORM_PAIR }}
+          name: digests-${{ matrix.artifact_context }}-${{ env.PLATFORM_PAIR }}
           path: /tmp/digests/*
           if-no-files-found: error
           retention-days: 1
 
-  merge-api-images:
-    needs:
-      - check-changes
-      - build-api
+  create-manifest:
+    needs: build
     runs-on: ubuntu-latest
-    if: github.repository == 'langgenius/dify' && (startsWith(github.ref, 'refs/tags/') || needs.check-changes.outputs.api == 'true')
-    steps:
-      - name: Download digests
-        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
-        with:
-          path: /tmp/digests
-          pattern: digests-api-*
-          merge-multiple: true
-
-      - name: Login to Docker Hub
-        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
-        with:
-          username: ${{ env.DOCKERHUB_USER }}
-          password: ${{ env.DOCKERHUB_TOKEN }}
-
-      - name: Extract metadata for Docker
-        id: meta
-        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
-        with:
-          images: ${{ env.DIFY_API_IMAGE_NAME }}
-          tags: |
-            type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/') && !contains(github.ref, '-') }}
-            type=ref,event=branch
-            type=sha,enable=true,priority=100,prefix=,suffix=,format=long
-            type=raw,value=${{ github.ref_name }},enable=${{ startsWith(github.ref, 'refs/tags/') }}
-
-      - name: Create manifest list and push
-        working-directory: /tmp/digests
-        env:
-          IMAGE_NAME: ${{ env.DIFY_API_IMAGE_NAME }}
-        run: |
-          docker buildx imagetools create $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \
-            $(printf "$IMAGE_NAME@sha256:%s " *)
-
-      - name: Inspect image
-        env:
-          IMAGE_NAME: ${{ env.DIFY_API_IMAGE_NAME }}
-          IMAGE_VERSION: ${{ steps.meta.outputs.version }}
-        run: |
-          docker buildx imagetools inspect "$IMAGE_NAME:$IMAGE_VERSION"
-
-  build-web:
-    name: Build Web Images
-    needs: check-changes
-    if: github.repository == 'langgenius/dify' && (startsWith(github.ref, 'refs/tags/') || needs.check-changes.outputs.web == 'true')
-    runs-on: ${{ matrix.platform == 'linux/arm64' && 'arm64_runner' || 'ubuntu-latest' }}
+    if: github.repository == 'langgenius/dify'
     strategy:
       matrix:
         include:
-          - service_name: "build-web-amd64"
-            platform: linux/amd64
-          - service_name: "build-web-arm64"
-            platform: linux/arm64
-
-    steps:
-      - name: Prepare
-        run: |
-          platform=${{ matrix.platform }}
-          echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
-
-      - name: Login to Docker Hub
-        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
-        with:
-          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
-
-      - name: Extract metadata for Docker
-        id: meta
-        uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
-        with:
-          images: ${{ env.DIFY_WEB_IMAGE_NAME }}
-
-      - name: Build Docker image
-        id: build
-        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
-        with:
-          context: "{{defaultContext}}:web"
-          platforms: ${{ matrix.platform }}
-          build-args: COMMIT_SHA=${{ fromJSON(steps.meta.outputs.json).labels['org.opencontainers.image.revision'] }}
-          labels: ${{ steps.meta.outputs.labels }}
-          outputs: type=image,name=${{ env.DIFY_WEB_IMAGE_NAME }},push-by-digest=true,name-canonical=true,push=true
-          cache-from: type=gha,scope=${{ matrix.service_name }}
-          cache-to: type=gha,mode=max,scope=${{ matrix.service_name }}
-
-      - name: Export digest
-        env:
-          DIGEST: ${{ steps.build.outputs.digest }}
-        run: |
-          mkdir -p /tmp/digests
-          sanitized_digest=${DIGEST#sha256:}
-          touch "/tmp/digests/${sanitized_digest}"
-
-      - name: Upload digest
-        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
-        with:
-          name: digests-web-${{ env.PLATFORM_PAIR }}
-          path: /tmp/digests/*
-          if-no-files-found: error
-          retention-days: 1
-
-  merge-web-images:
-    needs:
-      - check-changes
-      - build-web
-    runs-on: ubuntu-latest
-    if: github.repository == 'langgenius/dify' && (startsWith(github.ref, 'refs/tags/') || needs.check-changes.outputs.web == 'true')
+          - service_name: "merge-api-images"
+            image_name_env: "DIFY_API_IMAGE_NAME"
+            context: "api"
+          - service_name: "merge-web-images"
+            image_name_env: "DIFY_WEB_IMAGE_NAME"
+            context: "web"
     steps:
       - name: Download digests
         uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
         with:
           path: /tmp/digests
-          pattern: digests-web-*
+          pattern: digests-${{ matrix.context }}-*
           merge-multiple: true
 
       - name: Login to Docker Hub
@@ -250,7 +139,7 @@ jobs:
         id: meta
         uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0
         with:
-          images: ${{ env.DIFY_WEB_IMAGE_NAME }}
+          images: ${{ env[matrix.image_name_env] }}
           tags: |
             type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/') && !contains(github.ref, '-') }}
             type=ref,event=branch
@@ -260,14 +149,14 @@ jobs:
       - name: Create manifest list and push
         working-directory: /tmp/digests
         env:
-          IMAGE_NAME: ${{ env.DIFY_WEB_IMAGE_NAME }}
+          IMAGE_NAME: ${{ env[matrix.image_name_env] }}
         run: |
           docker buildx imagetools create $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \
             $(printf "$IMAGE_NAME@sha256:%s " *)
 
       - name: Inspect image
         env:
-          IMAGE_NAME: ${{ env.DIFY_WEB_IMAGE_NAME }}
+          IMAGE_NAME: ${{ env[matrix.image_name_env] }}
           IMAGE_VERSION: ${{ steps.meta.outputs.version }}
         run: |
           docker buildx imagetools inspect "$IMAGE_NAME:$IMAGE_VERSION"

.github/workflows/db-migration-test.yml

@@ -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"

.github/workflows/docker-build.yml

@@ -6,43 +6,43 @@ on:
       - "main"
     paths:
       - api/Dockerfile
+      - web/docker/**
       - web/Dockerfile
-
-permissions:
-  contents: read
+      - package.json
+      - pnpm-lock.yaml
+      - pnpm-workspace.yaml
+      - .nvmrc
 
 concurrency:
   group: docker-build-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
 jobs:
-  check-changes:
-    runs-on: ubuntu-latest
-    outputs:
-      api: ${{ steps.changes.outputs.api }}
-      web: ${{ steps.changes.outputs.web }}
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-        with:
-          fetch-depth: 2
-          persist-credentials: false
-
-      - name: Check changed files
-        id: changes
-        uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
-        with:
-          filters: |
-            api:
-              - 'api/Dockerfile'
-            web:
-              - 'web/Dockerfile'
-
-  build-api:
-    name: Build API Docker Image
-    needs: check-changes
-    if: needs.check-changes.outputs.api == 'true'
-    runs-on: ubuntu-latest
+  build-docker:
+    runs-on: ${{ matrix.runs_on }}
+    strategy:
+      matrix:
+        include:
+          - service_name: "api-amd64"
+            platform: linux/amd64
+            runs_on: ubuntu-latest
+            context: "{{defaultContext}}:api"
+            file: "Dockerfile"
+          - service_name: "api-arm64"
+            platform: linux/arm64
+            runs_on: ubuntu-24.04-arm
+            context: "{{defaultContext}}:api"
+            file: "Dockerfile"
+          - service_name: "web-amd64"
+            platform: linux/amd64
+            runs_on: ubuntu-latest
+            context: "{{defaultContext}}"
+            file: "web/Dockerfile"
+          - service_name: "web-arm64"
+            platform: linux/arm64
+            runs_on: ubuntu-24.04-arm
+            context: "{{defaultContext}}"
+            file: "web/Dockerfile"
     steps:
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
@@ -51,27 +51,8 @@ jobs:
         uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
         with:
           push: false
-          context: "{{defaultContext}}:api"
-          file: Dockerfile
-          platforms: linux/amd64
-          cache-from: type=gha,scope=api-pr
-          cache-to: type=gha,mode=max,scope=api-pr
-
-  build-web:
-    name: Build Web Docker Image
-    needs: check-changes
-    if: needs.check-changes.outputs.web == 'true'
-    runs-on: ubuntu-latest
-    steps:
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@4d04d5d9486b7bd6fa91e7baf45bbb4f8b9deedd # v4.0.0
-
-      - name: Build Docker Image
-        uses: docker/build-push-action@d08e5c354a6adb9ed34480a06d141179aa583294 # v7.0.0
-        with:
-          push: false
-          context: "{{defaultContext}}:web"
-          file: Dockerfile
-          platforms: linux/amd64
-          cache-from: type=gha,scope=web-pr
-          cache-to: type=gha,mode=max,scope=web-pr
+          context: ${{ matrix.context }}
+          file: ${{ matrix.file }}
+          platforms: ${{ matrix.platform }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

.github/workflows/main-ci.yml

@@ -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

.github/workflows/pyrefly-diff.yml

@@ -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 }}

.github/workflows/style.yml

@@ -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'

.github/workflows/tool-test-sdks.yaml

@@ -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 }}

.github/workflows/translate-i18n-claude.yml

@@ -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

.github/workflows/trigger-i18n-sync.yml

@@ -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,
+              },
+            })

.github/workflows/vdb-tests-full.yml

@@ -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/compose-action@v2.0.2
+#        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

.github/workflows/vdb-tests.yml

@@ -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

.github/workflows/web-e2e.yml

@@ -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: e2e-admin@example.com
+          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

.github/workflows/web-tests.yml

@@ -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

.gitignore

@@ -212,6 +212,8 @@ api/.vscode
 
 # pnpm
 /.pnpm-store
+/node_modules
+.vite-hooks/_
 
 # plugin migrate
 plugins.jsonl
@@ -239,4 +241,4 @@ scripts/stress-test/reports/
 *.local.md
 
 # Code Agent Folder
-.qoder/*
+.qoder/*

.vite-hooks/pre-commit

@@ -77,7 +77,7 @@ if $web_modified; then
     fi
 
     cd ./web || exit 1
-    lint-staged
+    vp staged
 
     if $web_ts_modified; then
         echo "Running TypeScript type-check:tsgo"

Makefile

@@ -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:"

README.md

@@ -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:

api/.env.example

@@ -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

api/README.md

@@ -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`).

api/constants/__init__.py

@@ -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"

api/controllers/common/file_response.py

@@ -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:

api/controllers/console/apikey.py

@@ -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()

api/controllers/console/app/app.py

@@ -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:

api/controllers/console/app/app_import.py

@@ -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)
 

api/controllers/console/app/conversation_variables.py

@@ -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 {

api/controllers/console/app/workflow.py

@@ -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:

api/controllers/console/app/workflow_app_log.py

@@ -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,

api/controllers/console/app/workflow_draft_variable.py

@@ -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,
         )

api/controllers/console/app/workflow_trigger.py

@@ -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":

api/controllers/console/billing/billing.py

@@ -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)
 

api/controllers/console/billing/compliance.py

@@ -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")

api/controllers/console/datasets/data_source.py

@@ -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)

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -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(

api/controllers/console/datasets/rag_pipeline/rag_pipeline.py

@@ -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()
             )

api/controllers/console/datasets/rag_pipeline/rag_pipeline_datasets.py

@@ -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,

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -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,
         )

api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py

@@ -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"

api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py

@@ -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,19 +593,15 @@ 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:
-            pipeline = session.merge(pipeline)
-            workflow = rag_pipeline_service.publish_workflow(
-                session=session,
-                pipeline=pipeline,
-                account=current_user,
-            )
-            pipeline.is_published = True
-            pipeline.workflow_id = workflow.id
-            session.add(pipeline)
-            workflow_created_at = TimestampField().format(workflow.created_at)
-
-            session.commit()
+        workflow = rag_pipeline_service.publish_workflow(
+            session=db.session,  # type: ignore[reportArgumentType,arg-type]
+            pipeline=pipeline,
+            account=current_user,
+        )
+        pipeline.is_published = True
+        pipeline.workflow_id = workflow.id
+        db.session.commit()
+        workflow_created_at = TimestampField().format(workflow.created_at)
 
         return {
             "result": "success",
@@ -694,7 +676,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 +748,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 +760,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")

api/controllers/console/explore/conversation.py

@@ -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,

api/controllers/console/workspace/__init__.py

@@ -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(

api/controllers/console/workspace/account.py

@@ -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()

api/controllers/console/workspace/model_providers.py

@@ -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()

api/controllers/console/workspace/models.py

@@ -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,
             )
 

api/controllers/console/workspace/tool_providers.py

@@ -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)

api/controllers/console/workspace/trigger_providers.py

@@ -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(

api/controllers/console/workspace/workspace.py

@@ -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())

api/controllers/inner_api/app/dsl.py

@@ -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

api/controllers/inner_api/plugin/wraps.py

@@ -6,7 +6,7 @@ from flask import current_app, request
 from flask_login import user_logged_in
 from pydantic import BaseModel
 from sqlalchemy import select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 
 from extensions.ext_database import db
 from libs.login import current_user
@@ -33,7 +33,7 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
         user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
     is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
     try:
-        with Session(db.engine) as session:
+        with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             user_model = None
 
             if is_anonymous:
@@ -56,7 +56,7 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
                     session_id=user_id,
                 )
                 session.add(user_model)
-                session.commit()
+                session.flush()
                 session.refresh(user_model)
 
     except Exception:

api/controllers/mcp/mcp.py

@@ -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)

api/controllers/service_api/app/conversation.py

@@ -3,7 +3,7 @@ from typing import Any, Literal
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, NotFound
 
 import services
@@ -116,7 +116,7 @@ class ConversationApi(Resource):
         last_id = str(query_args.last_id) if query_args.last_id else None
 
         try:
-            with Session(db.engine) as session:
+            with sessionmaker(db.engine).begin() as session:
                 pagination = ConversationService.pagination_by_last_id(
                     session=session,
                     app_model=app_model,

api/controllers/service_api/app/workflow.py

@@ -8,7 +8,7 @@ from graphon.enums import WorkflowExecutionStatus
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 from controllers.common.schema import register_schema_models
@@ -314,7 +314,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,

api/controllers/service_api/dataset/segment.py

@@ -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(

api/controllers/web/conversation.py

@@ -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,

api/controllers/web/forgot_password.py

@@ -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()

api/controllers/web/wraps.py

@@ -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:

api/core/agent/base_agent_runner.py

@@ -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()
 

api/core/app/apps/pipeline/pipeline_generator.py

@@ -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(

api/core/app/apps/pipeline/pipeline_runner.py

@@ -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"

api/core/app/llm/quota.py

@@ -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(

api/core/app/workflow/layers/observability.py

@@ -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

api/core/callback_handler/index_tool_callback_handler.py

@@ -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()
 

api/core/datasource/datasource_file_manager.py

@@ -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

api/core/entities/provider_configuration.py

@@ -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,
                 )

api/core/helper/csv_sanitizer.py

@@ -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:

api/core/helper/encrypter.py

@@ -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)

api/core/indexing_runner.py

@@ -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(

api/core/llm_generator/llm_generator.py

@@ -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(

api/core/mcp/server/streamable_http.py

@@ -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

api/core/ops/aliyun_trace/data_exporter/traceclient.py

@@ -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",
             }
         )

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -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)
 

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -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))

api/core/ops/mlflow_trace/mlflow_trace.py

@@ -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"))

api/core/ops/ops_trace_manager.py

@@ -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 {}
 
 

api/core/ops/tencent_trace/client.py

@@ -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

api/core/ops/utils.py

@@ -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 = {}

api/core/plugin/backwards_invocation/app.py

@@ -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")
 

api/core/plugin/impl/base.py

@@ -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

api/core/provider_manager.py

@@ -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 [],

api/core/rag/datasource/keyword/jieba/jieba.py

@@ -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)

api/core/rag/datasource/keyword/jieba/jieba_keyword_table_handler.py

@@ -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

api/core/rag/datasource/retrieval_service.py

@@ -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 = {}

api/core/rag/datasource/vdb/alibabacloud_mysql/alibabacloud_mysql_vector.py

@@ -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

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py

@@ -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"),

api/core/rag/datasource/vdb/baidu/baidu_vector.py

@@ -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
 

api/core/rag/datasource/vdb/clickzetta/clickzetta_vector.py

@@ -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
 

api/core/rag/datasource/vdb/field.py

@@ -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):

api/core/rag/datasource/vdb/hologres/hologres_vector.py

@@ -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))

api/core/rag/datasource/vdb/iris/iris_vector.py

@@ -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

api/core/rag/datasource/vdb/matrixone/matrixone_vector.py

@@ -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

api/core/rag/datasource/vdb/myscale/myscale_vector.py

@@ -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

api/core/rag/datasource/vdb/oceanbase/oceanbase_vector.py

@@ -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 = {}
 

api/core/rag/datasource/vdb/tablestore/tablestore_vector.py

@@ -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

api/core/rag/datasource/vdb/tencent/tencent_vector.py

@@ -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)

api/core/rag/datasource/vdb/tidb_on_qdrant/tidb_on_qdrant_vector.py

@@ -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

api/core/rag/datasource/vdb/tidb_on_qdrant/tidb_service.py

@@ -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)
         )