Merge remote-tracking branch 'origin/main' into feat/source-for-plugin-tasks

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.agents/skills/frontend-testing/SKILL.md

@@ -63,8 +63,7 @@ pnpm analyze-component <path> --review
 
 ### File Naming
 
-- Test files: `ComponentName.spec.tsx` inside a same-level `__tests__/` directory
-- Placement rule: Component, hook, and utility tests must live in a sibling `__tests__/` folder at the same level as the source under test. For example, `foo/index.tsx` maps to `foo/__tests__/index.spec.tsx`, and `foo/bar.ts` maps to `foo/__tests__/bar.spec.ts`.
+- Test files: `ComponentName.spec.tsx` (same directory as component)
 - Integration tests: `web/__tests__/` directory
 
 ## Test Structure Template

.agents/skills/frontend-testing/assets/component-test.template.tsx

@@ -41,7 +41,7 @@ import userEvent from '@testing-library/user-event'
 // Router (if component uses useRouter, usePathname, useSearchParams)
 // WHY: Isolates tests from Next.js routing, enables testing navigation behavior
 // const mockPush = vi.fn()
-// vi.mock('@/next/navigation', () => ({
+// vi.mock('next/navigation', () => ({
 //   useRouter: () => ({ push: mockPush }),
 //   usePathname: () => '/test-path',
 // }))

.gemini/config.yaml

@@ -1,13 +0,0 @@
-have_fun: false
-memory_config:
-  disabled: false
-code_review:
-  disable: true
-  comment_severity_threshold: MEDIUM
-  max_review_comments: -1
-  pull_request_opened:
-    help: false
-    summary: false
-    code_review: false
-    include_drafts: false
-ignore_patterns: []

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

@@ -4,9 +4,10 @@ runs:
   using: composite
   steps:
     - name: Setup Vite+
-      uses: voidzero-dev/setup-vp@20553a7a7429c429a74894104a2835d7fed28a72 # v1.3.0
+      uses: voidzero-dev/setup-vp@b5d848f5a62488f3d3d920f8aa6ac318a60c5f07 # v1
       with:
-        working-directory: web
-        node-version-file: .nvmrc
+        node-version-file: "./web/.nvmrc"
         cache: true
-        run-install: true
+        run-install: |
+          - cwd: ./web
+            args: ['--frozen-lockfile']

.github/workflows/anti-slop.yml

@@ -12,7 +12,7 @@ jobs:
   anti-slop:
     runs-on: ubuntu-latest
     steps:
-      - uses: peakoss/anti-slop@85daca1880e9e1af197fc06ea03349daf08f4202 # v0.2.1
+      - uses: peakoss/anti-slop@v0
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           close-pr: false

.github/workflows/api-tests.yml

@@ -2,12 +2,6 @@ name: Run Pytest
 
 on:
   workflow_call:
-    secrets:
-      CODECOV_TOKEN:
-        required: false
-
-permissions:
-  contents: read
 
 concurrency:
   group: api-tests-${{ github.head_ref || github.run_id }}
@@ -17,8 +11,6 @@ jobs:
   test:
     name: API Tests
     runs-on: ubuntu-latest
-    env:
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
     defaults:
       run:
         shell: bash
@@ -32,11 +24,10 @@ jobs:
       - 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@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}
@@ -88,12 +79,21 @@ jobs:
             api/tests/test_containers_integration_tests \
             api/tests/unit_tests
 
-      - name: Report coverage
-        if: ${{ env.CODECOV_TOKEN != '' && matrix.python-version == '3.12' }}
-        uses: codecov/codecov-action@1af58845a975a7985b0beb0cbe6fbbb71a41dbad # v5.5.3
-        with:
-          files: ./coverage.xml
-          disable_search: true
-          flags: api
-        env:
-          CODECOV_TOKEN: ${{ env.CODECOV_TOKEN }}
+      - name: Coverage Summary
+        run: |
+          set -x
+          # Extract coverage percentage and create a summary
+          TOTAL_COVERAGE=$(python -c 'import json; print(json.load(open("coverage.json"))["totals"]["percent_covered_display"])')
+
+          # Create a detailed coverage summary
+          echo "### Test Coverage Summary :test_tube:" >> $GITHUB_STEP_SUMMARY
+          echo "Total Coverage: ${TOTAL_COVERAGE}%" >> $GITHUB_STEP_SUMMARY
+          {
+            echo ""
+            echo "<details><summary>File-level coverage (click to expand)</summary>"
+            echo ""
+            echo '```'
+            uv run --project api coverage report -m
+            echo '```'
+            echo "</details>"
+          } >> $GITHUB_STEP_SUMMARY

.github/workflows/autofix.yml

@@ -39,7 +39,7 @@ jobs:
         with:
           python-version: "3.11"
 
-      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+      - uses: astral-sh/setup-uv@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
 
       - name: Generate Docker Compose
         if: steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -94,6 +94,11 @@ jobs:
           find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
           find . -name "*.py.bak" -type f -delete
 
+      # mdformat breaks YAML front matter in markdown files. Add --exclude for directories containing YAML front matter.
+      - name: mdformat
+        run: |
+          uvx --python 3.13 mdformat . --exclude ".agents/skills/**"
+
       - name: Setup web environment
         if: steps.web-changes.outputs.any_changed == 'true'
         uses: ./.github/actions/setup-web

.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@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.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@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
         with:
           enable-cache: true
           python-version: "3.12"

.github/workflows/main-ci.yml

@@ -56,14 +56,15 @@ jobs:
     needs: check-changes
     if: needs.check-changes.outputs.api-changed == 'true'
     uses: ./.github/workflows/api-tests.yml
-    secrets: inherit
 
   web-tests:
     name: Web Tests
     needs: check-changes
     if: needs.check-changes.outputs.web-changed == 'true'
     uses: ./.github/workflows/web-tests.yml
-    secrets: inherit
+    with:
+      base_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.base.sha || github.event.before }}
+      head_sha: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
 
   style-check:
     name: Style Check

.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@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
         with:
           enable-cache: true
 

.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@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
         with:
           enable-cache: false
           python-version: "3.12"
@@ -84,20 +84,20 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         uses: ./.github/actions/setup-web
 
-      - name: Restore ESLint cache
-        if: steps.changed-files.outputs.any_changed == 'true'
-        id: eslint-cache-restore
-        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 }}
-          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/**') }}-
-
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: vp run lint:ci
+        run: |
+          vp run lint:ci
+        # pnpm run lint:report
+        # continue-on-error: true
+
+      # - name: Annotate Code
+      #   if: steps.changed-files.outputs.any_changed == 'true' && github.event_name == 'pull_request'
+      #   uses: DerLev/eslint-annotations@51347b3a0abfb503fc8734d5ae31c4b151297fae
+      #   with:
+      #     eslint-report: web/eslint_report.json
+      #     github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Web tsslint
         if: steps.changed-files.outputs.any_changed == 'true'
@@ -114,13 +114,6 @@ jobs:
         working-directory: ./web
         run: vp run knip
 
-      - name: Save ESLint cache
-        if: steps.changed-files.outputs.any_changed == 'true' && success() && steps.eslint-cache-restore.outputs.cache-hit != 'true'
-        uses: actions/cache/save@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v5.0.4
-        with:
-          path: web/.eslintcache
-          key: ${{ steps.eslint-cache-restore.outputs.cache-primary-key }}
-
   superlinter:
     name: SuperLinter
     runs-on: ubuntu-latest

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

@@ -120,7 +120,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.detect_changes.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@ff9acae5886d41a99ed4ec14b7dc147d55834722 # v1.0.77
+        uses: anthropics/claude-code-action@cd77b50d2b0808657f8e6774085c8bf54484351c # v1.0.72
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/vdb-tests.yml

@@ -31,7 +31,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@e06108dd0aef18192324c70427afc47652e63a82 # v7.5.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}

.github/workflows/web-tests.yml

@@ -2,9 +2,13 @@ name: Web Tests
 
 on:
   workflow_call:
-    secrets:
-      CODECOV_TOKEN:
+    inputs:
+      base_sha:
         required: false
+        type: string
+      head_sha:
+        required: false
+        type: string
 
 permissions:
   contents: read
@@ -22,8 +26,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
@@ -56,7 +60,7 @@ jobs:
     needs: [test]
     runs-on: ubuntu-latest
     env:
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+      VITEST_COVERAGE_SCOPE: app-components
     defaults:
       run:
         shell: bash
@@ -80,16 +84,339 @@ jobs:
           merge-multiple: true
 
       - name: Merge reports
-        run: vp test --merge-reports --coverage --silent=passed-only
+        run: vp test --merge-reports --reporter=json --reporter=agent --coverage
 
-      - name: Report coverage
-        if: ${{ env.CODECOV_TOKEN != '' }}
-        uses: codecov/codecov-action@1af58845a975a7985b0beb0cbe6fbbb71a41dbad # v5.5.3
-        with:
-          directory: web/coverage
-          flags: web
+      - name: Check app/components diff coverage
         env:
-          CODECOV_TOKEN: ${{ env.CODECOV_TOKEN }}
+          BASE_SHA: ${{ inputs.base_sha }}
+          HEAD_SHA: ${{ inputs.head_sha }}
+        run: node ./scripts/check-components-diff-coverage.mjs
+
+      - name: Coverage Summary
+        if: always()
+        id: coverage-summary
+        run: |
+          set -eo pipefail
+
+          COVERAGE_FILE="coverage/coverage-final.json"
+          COVERAGE_SUMMARY_FILE="coverage/coverage-summary.json"
+
+          if [ ! -f "$COVERAGE_FILE" ] && [ ! -f "$COVERAGE_SUMMARY_FILE" ]; then
+            echo "has_coverage=false" >> "$GITHUB_OUTPUT"
+            echo "### 🚨 Test Coverage Report :test_tube:" >> "$GITHUB_STEP_SUMMARY"
+            echo "Coverage data not found. Ensure Vitest runs with coverage enabled." >> "$GITHUB_STEP_SUMMARY"
+            exit 0
+          fi
+
+          echo "has_coverage=true" >> "$GITHUB_OUTPUT"
+
+          node <<'NODE' >> "$GITHUB_STEP_SUMMARY"
+          const fs = require('fs');
+          const path = require('path');
+          let libCoverage = null;
+
+          try {
+            libCoverage = require('istanbul-lib-coverage');
+          } catch (error) {
+            libCoverage = null;
+          }
+
+          const summaryPath = path.join('coverage', 'coverage-summary.json');
+          const finalPath = path.join('coverage', 'coverage-final.json');
+
+          const hasSummary = fs.existsSync(summaryPath);
+          const hasFinal = fs.existsSync(finalPath);
+
+          if (!hasSummary && !hasFinal) {
+            console.log('### Test Coverage Summary :test_tube:');
+            console.log('');
+            console.log('No coverage data found.');
+            process.exit(0);
+          }
+
+          const summary = hasSummary
+            ? JSON.parse(fs.readFileSync(summaryPath, 'utf8'))
+            : null;
+          const coverage = hasFinal
+            ? JSON.parse(fs.readFileSync(finalPath, 'utf8'))
+            : null;
+
+          const getLineCoverageFromStatements = (statementMap, statementHits) => {
+            const lineHits = {};
+
+            if (!statementMap || !statementHits) {
+              return lineHits;
+            }
+
+            Object.entries(statementMap).forEach(([key, statement]) => {
+              const line = statement?.start?.line;
+              if (!line) {
+                return;
+              }
+              const hits = statementHits[key] ?? 0;
+              const previous = lineHits[line];
+              lineHits[line] = previous === undefined ? hits : Math.max(previous, hits);
+            });
+
+            return lineHits;
+          };
+
+          const getFileCoverage = (entry) => (
+            libCoverage ? libCoverage.createFileCoverage(entry) : null
+          );
+
+          const getLineHits = (entry, fileCoverage) => {
+            const lineHits = entry.l ?? {};
+            if (Object.keys(lineHits).length > 0) {
+              return lineHits;
+            }
+            if (fileCoverage) {
+              return fileCoverage.getLineCoverage();
+            }
+            return getLineCoverageFromStatements(entry.statementMap ?? {}, entry.s ?? {});
+          };
+
+          const getUncoveredLines = (entry, fileCoverage, lineHits) => {
+            if (lineHits && Object.keys(lineHits).length > 0) {
+              return Object.entries(lineHits)
+                .filter(([, count]) => count === 0)
+                .map(([line]) => Number(line))
+                .sort((a, b) => a - b);
+            }
+            if (fileCoverage) {
+              return fileCoverage.getUncoveredLines();
+            }
+            return [];
+          };
+
+          const totals = {
+            lines: { covered: 0, total: 0 },
+            statements: { covered: 0, total: 0 },
+            branches: { covered: 0, total: 0 },
+            functions: { covered: 0, total: 0 },
+          };
+          const fileSummaries = [];
+
+          if (summary) {
+            const totalEntry = summary.total ?? {};
+            ['lines', 'statements', 'branches', 'functions'].forEach((key) => {
+              if (totalEntry[key]) {
+                totals[key].covered = totalEntry[key].covered ?? 0;
+                totals[key].total = totalEntry[key].total ?? 0;
+              }
+            });
+
+            Object.entries(summary)
+              .filter(([file]) => file !== 'total')
+              .forEach(([file, data]) => {
+                fileSummaries.push({
+                  file,
+                  pct: data.lines?.pct ?? data.statements?.pct ?? 0,
+                  lines: {
+                    covered: data.lines?.covered ?? 0,
+                    total: data.lines?.total ?? 0,
+                  },
+                });
+              });
+          } else if (coverage) {
+            Object.entries(coverage).forEach(([file, entry]) => {
+              const fileCoverage = getFileCoverage(entry);
+              const lineHits = getLineHits(entry, fileCoverage);
+              const statementHits = entry.s ?? {};
+              const branchHits = entry.b ?? {};
+              const functionHits = entry.f ?? {};
+
+              const lineTotal = Object.keys(lineHits).length;
+              const lineCovered = Object.values(lineHits).filter((n) => n > 0).length;
+
+              const statementTotal = Object.keys(statementHits).length;
+              const statementCovered = Object.values(statementHits).filter((n) => n > 0).length;
+
+              const branchTotal = Object.values(branchHits).reduce((acc, branches) => acc + branches.length, 0);
+              const branchCovered = Object.values(branchHits).reduce(
+                (acc, branches) => acc + branches.filter((n) => n > 0).length,
+                0,
+              );
+
+              const functionTotal = Object.keys(functionHits).length;
+              const functionCovered = Object.values(functionHits).filter((n) => n > 0).length;
+
+              totals.lines.total += lineTotal;
+              totals.lines.covered += lineCovered;
+              totals.statements.total += statementTotal;
+              totals.statements.covered += statementCovered;
+              totals.branches.total += branchTotal;
+              totals.branches.covered += branchCovered;
+              totals.functions.total += functionTotal;
+              totals.functions.covered += functionCovered;
+
+              const pct = (covered, tot) => (tot > 0 ? (covered / tot) * 100 : 0);
+
+              fileSummaries.push({
+                file,
+                pct: pct(lineCovered || statementCovered, lineTotal || statementTotal),
+                lines: {
+                  covered: lineCovered || statementCovered,
+                  total: lineTotal || statementTotal,
+                },
+              });
+            });
+          }
+
+          const pct = (covered, tot) => (tot > 0 ? ((covered / tot) * 100).toFixed(2) : '0.00');
+
+          console.log('### Test Coverage Summary :test_tube:');
+          console.log('');
+          console.log('| Metric | Coverage | Covered / Total |');
+          console.log('|--------|----------|-----------------|');
+          console.log(`| Lines | ${pct(totals.lines.covered, totals.lines.total)}% | ${totals.lines.covered} / ${totals.lines.total} |`);
+          console.log(`| Statements | ${pct(totals.statements.covered, totals.statements.total)}% | ${totals.statements.covered} / ${totals.statements.total} |`);
+          console.log(`| Branches | ${pct(totals.branches.covered, totals.branches.total)}% | ${totals.branches.covered} / ${totals.branches.total} |`);
+          console.log(`| Functions | ${pct(totals.functions.covered, totals.functions.total)}% | ${totals.functions.covered} / ${totals.functions.total} |`);
+
+          console.log('');
+          console.log('<details><summary>File coverage (lowest lines first)</summary>');
+          console.log('');
+          console.log('```');
+          fileSummaries
+            .sort((a, b) => (a.pct - b.pct) || (b.lines.total - a.lines.total))
+            .slice(0, 25)
+            .forEach(({ file, pct, lines }) => {
+              console.log(`${pct.toFixed(2)}%\t${lines.covered}/${lines.total}\t${file}`);
+            });
+          console.log('```');
+          console.log('</details>');
+
+          if (coverage) {
+            const pctValue = (covered, tot) => {
+              if (tot === 0) {
+                return '0';
+              }
+              return ((covered / tot) * 100)
+                .toFixed(2)
+                .replace(/\.?0+$/, '');
+            };
+
+            const formatLineRanges = (lines) => {
+              if (lines.length === 0) {
+                return '';
+              }
+              const ranges = [];
+              let start = lines[0];
+              let end = lines[0];
+
+              for (let i = 1; i < lines.length; i += 1) {
+                const current = lines[i];
+                if (current === end + 1) {
+                  end = current;
+                  continue;
+                }
+                ranges.push(start === end ? `${start}` : `${start}-${end}`);
+                start = current;
+                end = current;
+              }
+              ranges.push(start === end ? `${start}` : `${start}-${end}`);
+              return ranges.join(',');
+            };
+
+            const tableTotals = {
+              statements: { covered: 0, total: 0 },
+              branches: { covered: 0, total: 0 },
+              functions: { covered: 0, total: 0 },
+              lines: { covered: 0, total: 0 },
+            };
+            const tableRows = Object.entries(coverage)
+              .map(([file, entry]) => {
+                const fileCoverage = getFileCoverage(entry);
+                const lineHits = getLineHits(entry, fileCoverage);
+                const statementHits = entry.s ?? {};
+                const branchHits = entry.b ?? {};
+                const functionHits = entry.f ?? {};
+
+                const lineTotal = Object.keys(lineHits).length;
+                const lineCovered = Object.values(lineHits).filter((n) => n > 0).length;
+                const statementTotal = Object.keys(statementHits).length;
+                const statementCovered = Object.values(statementHits).filter((n) => n > 0).length;
+                const branchTotal = Object.values(branchHits).reduce((acc, branches) => acc + branches.length, 0);
+                const branchCovered = Object.values(branchHits).reduce(
+                  (acc, branches) => acc + branches.filter((n) => n > 0).length,
+                  0,
+                );
+                const functionTotal = Object.keys(functionHits).length;
+                const functionCovered = Object.values(functionHits).filter((n) => n > 0).length;
+
+                tableTotals.lines.total += lineTotal;
+                tableTotals.lines.covered += lineCovered;
+                tableTotals.statements.total += statementTotal;
+                tableTotals.statements.covered += statementCovered;
+                tableTotals.branches.total += branchTotal;
+                tableTotals.branches.covered += branchCovered;
+                tableTotals.functions.total += functionTotal;
+                tableTotals.functions.covered += functionCovered;
+
+                const uncoveredLines = getUncoveredLines(entry, fileCoverage, lineHits);
+
+                const filePath = entry.path ?? file;
+                const relativePath = path.isAbsolute(filePath)
+                  ? path.relative(process.cwd(), filePath)
+                  : filePath;
+
+                return {
+                  file: relativePath || file,
+                  statements: pctValue(statementCovered, statementTotal),
+                  branches: pctValue(branchCovered, branchTotal),
+                  functions: pctValue(functionCovered, functionTotal),
+                  lines: pctValue(lineCovered, lineTotal),
+                  uncovered: formatLineRanges(uncoveredLines),
+                };
+              })
+              .sort((a, b) => a.file.localeCompare(b.file));
+
+            const columns = [
+              { key: 'file', header: 'File', align: 'left' },
+              { key: 'statements', header: '% Stmts', align: 'right' },
+              { key: 'branches', header: '% Branch', align: 'right' },
+              { key: 'functions', header: '% Funcs', align: 'right' },
+              { key: 'lines', header: '% Lines', align: 'right' },
+              { key: 'uncovered', header: 'Uncovered Line #s', align: 'left' },
+            ];
+
+            const allFilesRow = {
+              file: 'All files',
+              statements: pctValue(tableTotals.statements.covered, tableTotals.statements.total),
+              branches: pctValue(tableTotals.branches.covered, tableTotals.branches.total),
+              functions: pctValue(tableTotals.functions.covered, tableTotals.functions.total),
+              lines: pctValue(tableTotals.lines.covered, tableTotals.lines.total),
+              uncovered: '',
+            };
+
+            const rowsForOutput = [allFilesRow, ...tableRows];
+            const formatRow = (row) => `| ${columns
+              .map(({ key }) => String(row[key] ?? ''))
+              .join(' | ')} |`;
+            const headerRow = `| ${columns.map(({ header }) => header).join(' | ')} |`;
+            const dividerRow = `| ${columns
+              .map(({ align }) => (align === 'right' ? '---:' : ':---'))
+              .join(' | ')} |`;
+
+            console.log('');
+            console.log('<details><summary>Vitest coverage table</summary>');
+            console.log('');
+            console.log(headerRow);
+            console.log(dividerRow);
+            rowsForOutput.forEach((row) => console.log(formatRow(row)));
+            console.log('</details>');
+          }
+          NODE
+
+      - name: Upload Coverage Artifact
+        if: steps.coverage-summary.outputs.has_coverage == 'true'
+        uses: actions/upload-artifact@bbbca2ddaa5d8feaa63e36b76fdaad77386f024f # v7.0.0
+        with:
+          name: web-coverage-report
+          path: web/coverage
+          retention-days: 30
+          if-no-files-found: error
 
   web-build:
     name: Web Build

CONTRIBUTING.md

@@ -97,8 +97,3 @@ Feel free to reach out if you encounter any issues during the setup process.
 ## Getting Help
 
 If you ever get stuck or get a burning question while contributing, simply shoot your queries our way via the related GitHub issue, or hop onto our [Discord](https://discord.gg/8Tpq4AcN9c) for a quick chat.
-
-## Automated Agent Contributions
-
-> [!NOTE]
-> If you are an automated agent， just add 🤖🤖🤖 to the end of the PR title to opt-in.

api/.env.example

@@ -353,9 +353,6 @@ BAIDU_VECTOR_DB_SHARD=1
 BAIDU_VECTOR_DB_REPLICAS=3
 BAIDU_VECTOR_DB_INVERTED_INDEX_ANALYZER=DEFAULT_ANALYZER
 BAIDU_VECTOR_DB_INVERTED_INDEX_PARSER_MODE=COARSE_MODE
-BAIDU_VECTOR_DB_AUTO_BUILD_ROW_COUNT_INCREMENT=500
-BAIDU_VECTOR_DB_AUTO_BUILD_ROW_COUNT_INCREMENT_RATIO=0.05
-BAIDU_VECTOR_DB_REBUILD_INDEX_TIMEOUT_IN_SECONDS=300
 
 # Upstash configuration
 UPSTASH_VECTOR_URL=your-server-url
@@ -740,25 +737,24 @@ SANDBOX_EXPIRED_RECORDS_RETENTION_DAYS=30
 SANDBOX_EXPIRED_RECORDS_CLEAN_TASK_LOCK_TTL=90000
 
 
-# Redis URL used for event bus between API and
+# Redis URL used for PubSub between API and
 # celery worker
 # defaults to url constructed from `REDIS_*`
 # configurations
-EVENT_BUS_REDIS_URL=
-# Event transport type. Options are:
+PUBSUB_REDIS_URL=
+# Pub/sub channel type for streaming events.
+# valid options are:
 #
-#  - pubsub: normal Pub/Sub (at-most-once)
-#  - sharded: sharded Pub/Sub (at-most-once)
-#  - streams: Redis Streams (at-least-once, recommended to avoid subscriber races)
+# - pubsub: for normal Pub/Sub
+# - sharded: for sharded Pub/Sub
 #
-# Note: Before enabling 'streams' in production, estimate your expected event volume and retention needs.
-# Configure Redis memory limits and stream trimming appropriately (e.g., MAXLEN and key expiry) to reduce
-# the risk of data loss from Redis auto-eviction under memory pressure.
-# Also accepts ENV: EVENT_BUS_REDIS_CHANNEL_TYPE.
-EVENT_BUS_REDIS_CHANNEL_TYPE=pubsub
-# Whether to use Redis cluster mode while use redis as event bus.
+# It's highly recommended to use sharded Pub/Sub AND redis cluster
+# for large deployments.
+PUBSUB_REDIS_CHANNEL_TYPE=pubsub
+# Whether to use Redis cluster mode while running
+# PubSub.
 #  It's highly recommended to enable this for large deployments.
-EVENT_BUS_REDIS_USE_CLUSTERS=false
+PUBSUB_REDIS_USE_CLUSTERS=false
 
 # Whether to Enable human input timeout check task
 ENABLE_HUMAN_INPUT_TIMEOUT_TASK=true

api/AGENTS.md

@@ -78,7 +78,7 @@ class UserProfile(TypedDict):
     nickname: NotRequired[str]
 ```
 
-- For classes, declare all member variables explicitly with types at the top of the class body (before `__init__`), even when the class is not a dataclass or Pydantic model, so the class shape is obvious at a glance:
+- For classes, declare member variables at the top of the class body (before `__init__`) so the class shape is obvious at a glance:
 
 ```python
 from datetime import datetime

api/commands/plugin.py

@@ -1,11 +1,9 @@
 import json
 import logging
-from typing import Any, cast
+from typing import Any
 
 import click
 from pydantic import TypeAdapter
-from sqlalchemy import delete, select
-from sqlalchemy.engine import CursorResult
 
 from configs import dify_config
 from core.helper import encrypter
@@ -50,15 +48,14 @@ def setup_system_tool_oauth_client(provider, client_params):
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
         return
 
-    deleted_count = cast(
-        CursorResult,
-        db.session.execute(
-            delete(ToolOAuthSystemClient).where(
-                ToolOAuthSystemClient.provider == provider_name,
-                ToolOAuthSystemClient.plugin_id == plugin_id,
-            )
-        ),
-    ).rowcount
+    deleted_count = (
+        db.session.query(ToolOAuthSystemClient)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
     if deleted_count > 0:
         click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
 
@@ -100,15 +97,14 @@ def setup_system_trigger_oauth_client(provider, client_params):
         click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
         return
 
-    deleted_count = cast(
-        CursorResult,
-        db.session.execute(
-            delete(TriggerOAuthSystemClient).where(
-                TriggerOAuthSystemClient.provider == provider_name,
-                TriggerOAuthSystemClient.plugin_id == plugin_id,
-            )
-        ),
-    ).rowcount
+    deleted_count = (
+        db.session.query(TriggerOAuthSystemClient)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
     if deleted_count > 0:
         click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
 
@@ -143,15 +139,14 @@ def setup_datasource_oauth_client(provider, client_params):
         return
 
     click.echo(click.style(f"Ready to delete existing oauth client params: {provider_name}", fg="yellow"))
-    deleted_count = cast(
-        CursorResult,
-        db.session.execute(
-            delete(DatasourceOauthParamConfig).where(
-                DatasourceOauthParamConfig.provider == provider_name,
-                DatasourceOauthParamConfig.plugin_id == plugin_id,
-            )
-        ),
-    ).rowcount
+    deleted_count = (
+        db.session.query(DatasourceOauthParamConfig)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
     if deleted_count > 0:
         click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
 
@@ -197,9 +192,7 @@ def transform_datasource_credentials(environment: str):
 
         # deal notion credentials
         deal_notion_count = 0
-        notion_credentials = db.session.scalars(
-            select(DataSourceOauthBinding).where(DataSourceOauthBinding.provider == "notion")
-        ).all()
+        notion_credentials = db.session.query(DataSourceOauthBinding).filter_by(provider="notion").all()
         if notion_credentials:
             notion_credentials_tenant_mapping: dict[str, list[DataSourceOauthBinding]] = {}
             for notion_credential in notion_credentials:
@@ -208,7 +201,7 @@ def transform_datasource_credentials(environment: str):
                     notion_credentials_tenant_mapping[tenant_id] = []
                 notion_credentials_tenant_mapping[tenant_id].append(notion_credential)
             for tenant_id, notion_tenant_credentials in notion_credentials_tenant_mapping.items():
-                tenant = db.session.scalar(select(Tenant).where(Tenant.id == tenant_id))
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
                 if not tenant:
                     continue
                 try:
@@ -257,9 +250,7 @@ def transform_datasource_credentials(environment: str):
                 db.session.commit()
         # deal firecrawl credentials
         deal_firecrawl_count = 0
-        firecrawl_credentials = db.session.scalars(
-            select(DataSourceApiKeyAuthBinding).where(DataSourceApiKeyAuthBinding.provider == "firecrawl")
-        ).all()
+        firecrawl_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="firecrawl").all()
         if firecrawl_credentials:
             firecrawl_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
             for firecrawl_credential in firecrawl_credentials:
@@ -268,7 +259,7 @@ def transform_datasource_credentials(environment: str):
                     firecrawl_credentials_tenant_mapping[tenant_id] = []
                 firecrawl_credentials_tenant_mapping[tenant_id].append(firecrawl_credential)
             for tenant_id, firecrawl_tenant_credentials in firecrawl_credentials_tenant_mapping.items():
-                tenant = db.session.scalar(select(Tenant).where(Tenant.id == tenant_id))
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
                 if not tenant:
                     continue
                 try:
@@ -321,9 +312,7 @@ def transform_datasource_credentials(environment: str):
                 db.session.commit()
         # deal jina credentials
         deal_jina_count = 0
-        jina_credentials = db.session.scalars(
-            select(DataSourceApiKeyAuthBinding).where(DataSourceApiKeyAuthBinding.provider == "jinareader")
-        ).all()
+        jina_credentials = db.session.query(DataSourceApiKeyAuthBinding).filter_by(provider="jinareader").all()
         if jina_credentials:
             jina_credentials_tenant_mapping: dict[str, list[DataSourceApiKeyAuthBinding]] = {}
             for jina_credential in jina_credentials:
@@ -332,7 +321,7 @@ def transform_datasource_credentials(environment: str):
                     jina_credentials_tenant_mapping[tenant_id] = []
                 jina_credentials_tenant_mapping[tenant_id].append(jina_credential)
             for tenant_id, jina_tenant_credentials in jina_credentials_tenant_mapping.items():
-                tenant = db.session.scalar(select(Tenant).where(Tenant.id == tenant_id))
+                tenant = db.session.query(Tenant).filter_by(id=tenant_id).first()
                 if not tenant:
                     continue
                 try:

api/commands/retention.py

@@ -88,8 +88,6 @@ def clean_workflow_runs(
     """
     Clean workflow runs and related workflow data for free tenants.
     """
-    from extensions.otel.runtime import flush_telemetry
-
     if (start_from is None) ^ (end_before is None):
         raise click.UsageError("--start-from and --end-before must be provided together.")
 
@@ -106,27 +104,16 @@ def clean_workflow_runs(
         end_before = now - datetime.timedelta(days=to_days_ago)
         before_days = 0
 
-    if from_days_ago is not None and to_days_ago is not None:
-        task_label = f"{from_days_ago}to{to_days_ago}"
-    elif start_from is None:
-        task_label = f"before-{before_days}"
-    else:
-        task_label = "custom"
-
     start_time = datetime.datetime.now(datetime.UTC)
     click.echo(click.style(f"Starting workflow run cleanup at {start_time.isoformat()}.", fg="white"))
 
-    try:
-        WorkflowRunCleanup(
-            days=before_days,
-            batch_size=batch_size,
-            start_from=start_from,
-            end_before=end_before,
-            dry_run=dry_run,
-            task_label=task_label,
-        ).run()
-    finally:
-        flush_telemetry()
+    WorkflowRunCleanup(
+        days=before_days,
+        batch_size=batch_size,
+        start_from=start_from,
+        end_before=end_before,
+        dry_run=dry_run,
+    ).run()
 
     end_time = datetime.datetime.now(datetime.UTC)
     elapsed = end_time - start_time
@@ -672,8 +659,6 @@ def clean_expired_messages(
     """
     Clean expired messages and related data for tenants based on clean policy.
     """
-    from extensions.otel.runtime import flush_telemetry
-
     click.echo(click.style("clean_messages: start clean messages.", fg="green"))
 
     start_at = time.perf_counter()
@@ -713,13 +698,6 @@ def clean_expired_messages(
         # NOTE: graceful_period will be ignored when billing is disabled.
         policy = create_message_clean_policy(graceful_period_days=graceful_period)
 
-        if from_days_ago is not None and before_days is not None:
-            task_label = f"{from_days_ago}to{before_days}"
-        elif start_from is None and before_days is not None:
-            task_label = f"before-{before_days}"
-        else:
-            task_label = "custom"
-
         # Create and run the cleanup service
         if abs_mode:
             assert start_from is not None
@@ -730,7 +708,6 @@ def clean_expired_messages(
                 end_before=end_before,
                 batch_size=batch_size,
                 dry_run=dry_run,
-                task_label=task_label,
             )
         elif from_days_ago is None:
             assert before_days is not None
@@ -739,7 +716,6 @@ def clean_expired_messages(
                 days=before_days,
                 batch_size=batch_size,
                 dry_run=dry_run,
-                task_label=task_label,
             )
         else:
             assert before_days is not None
@@ -751,7 +727,6 @@ def clean_expired_messages(
                 end_before=now - datetime.timedelta(days=before_days),
                 batch_size=batch_size,
                 dry_run=dry_run,
-                task_label=task_label,
             )
         stats = service.run()
 
@@ -777,8 +752,6 @@ def clean_expired_messages(
             )
         )
         raise
-    finally:
-        flush_telemetry()
 
     click.echo(click.style("messages cleanup completed.", fg="green"))
 

api/commands/storage.py

@@ -1,10 +1,7 @@
 import json
-from typing import cast
 
 import click
 import sqlalchemy as sa
-from sqlalchemy import update
-from sqlalchemy.engine import CursorResult
 
 from configs import dify_config
 from extensions.ext_database import db
@@ -743,17 +740,14 @@ def migrate_oss(
         else:
             try:
                 source_storage_type = StorageType.LOCAL if is_source_local else StorageType.OPENDAL
-                updated = cast(
-                    CursorResult,
-                    db.session.execute(
-                        update(UploadFile)
-                        .where(
-                            UploadFile.storage_type == source_storage_type,
-                            UploadFile.key.in_(copied_upload_file_keys),
-                        )
-                        .values(storage_type=dify_config.STORAGE_TYPE)
-                    ),
-                ).rowcount
+                updated = (
+                    db.session.query(UploadFile)
+                    .where(
+                        UploadFile.storage_type == source_storage_type,
+                        UploadFile.key.in_(copied_upload_file_keys),
+                    )
+                    .update({UploadFile.storage_type: dify_config.STORAGE_TYPE}, synchronize_session=False)
+                )
                 db.session.commit()
                 click.echo(click.style(f"Updated storage_type for {updated} upload_files records.", fg="green"))
             except Exception as e:

api/commands/system.py

@@ -2,7 +2,6 @@ import logging
 
 import click
 import sqlalchemy as sa
-from sqlalchemy import delete, select, update
 from sqlalchemy.orm import sessionmaker
 
 from configs import dify_config
@@ -42,7 +41,7 @@ def reset_encrypt_key_pair():
         click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
         return
     with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-        tenants = session.scalars(select(Tenant)).all()
+        tenants = session.query(Tenant).all()
         for tenant in tenants:
             if not tenant:
                 click.echo(click.style("No workspaces found. Run /install first.", fg="red"))
@@ -50,8 +49,8 @@ def reset_encrypt_key_pair():
 
             tenant.encrypt_public_key = generate_key_pair(tenant.id)
 
-            session.execute(delete(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id))
-            session.execute(delete(ProviderModel).where(ProviderModel.tenant_id == tenant.id))
+            session.query(Provider).where(Provider.provider_type == "custom", Provider.tenant_id == tenant.id).delete()
+            session.query(ProviderModel).where(ProviderModel.tenant_id == tenant.id).delete()
 
             click.echo(
                 click.style(
@@ -94,7 +93,7 @@ def convert_to_agent_apps():
                 app_id = str(i.id)
                 if app_id not in proceeded_app_ids:
                     proceeded_app_ids.append(app_id)
-                    app = db.session.scalar(select(App).where(App.id == app_id))
+                    app = db.session.query(App).where(App.id == app_id).first()
                     if app is not None:
                         apps.append(app)
 
@@ -109,8 +108,8 @@ def convert_to_agent_apps():
                 db.session.commit()
 
                 # update conversation mode to agent
-                db.session.execute(
-                    update(Conversation).where(Conversation.app_id == app.id).values(mode=AppMode.AGENT_CHAT)
+                db.session.query(Conversation).where(Conversation.app_id == app.id).update(
+                    {Conversation.mode: AppMode.AGENT_CHAT}
                 )
 
                 db.session.commit()
@@ -178,7 +177,7 @@ where sites.id is null limit 1000"""
                     continue
 
                 try:
-                    app = db.session.scalar(select(App).where(App.id == app_id))
+                    app = db.session.query(App).where(App.id == app_id).first()
                     if not app:
                         logger.info("App %s not found", app_id)
                         continue

api/commands/vector.py

@@ -10,12 +10,10 @@ from configs import dify_config
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
-from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
 from core.rag.models.document import ChildDocument, Document
 from extensions.ext_database import db
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
-from models.enums import DatasetMetadataType, IndexingStatus, SegmentStatus
 from models.model import App, AppAnnotationSetting, MessageAnnotation
 
 
@@ -42,13 +40,14 @@ def migrate_annotation_vector_database():
             # get apps info
             per_page = 50
             with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-                apps = session.scalars(
-                    select(App)
+                apps = (
+                    session.query(App)
                     .where(App.status == "normal")
                     .order_by(App.created_at.desc())
                     .limit(per_page)
                     .offset((page - 1) * per_page)
-                ).all()
+                    .all()
+                )
             if not apps:
                 break
         except SQLAlchemyError:
@@ -63,8 +62,8 @@ def migrate_annotation_vector_database():
             try:
                 click.echo(f"Creating app annotation index: {app.id}")
                 with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-                    app_annotation_setting = session.scalar(
-                        select(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app.id).limit(1)
+                    app_annotation_setting = (
+                        session.query(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app.id).first()
                     )
 
                     if not app_annotation_setting:
@@ -72,10 +71,10 @@ def migrate_annotation_vector_database():
                         click.echo(f"App annotation setting disabled: {app.id}")
                         continue
                     # get dataset_collection_binding info
-                    dataset_collection_binding = session.scalar(
-                        select(DatasetCollectionBinding).where(
-                            DatasetCollectionBinding.id == app_annotation_setting.collection_binding_id
-                        )
+                    dataset_collection_binding = (
+                        session.query(DatasetCollectionBinding)
+                        .where(DatasetCollectionBinding.id == app_annotation_setting.collection_binding_id)
+                        .first()
                     )
                     if not dataset_collection_binding:
                         click.echo(f"App annotation collection binding not found: {app.id}")
@@ -86,7 +85,7 @@ def migrate_annotation_vector_database():
                 dataset = Dataset(
                     id=app.id,
                     tenant_id=app.tenant_id,
-                    indexing_technique=IndexTechniqueType.HIGH_QUALITY,
+                    indexing_technique="high_quality",
                     embedding_model_provider=dataset_collection_binding.provider_name,
                     embedding_model=dataset_collection_binding.model_name,
                     collection_binding_id=dataset_collection_binding.id,
@@ -178,9 +177,7 @@ def migrate_knowledge_vector_database():
     while True:
         try:
             stmt = (
-                select(Dataset)
-                .where(Dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY)
-                .order_by(Dataset.created_at.desc())
+                select(Dataset).where(Dataset.indexing_technique == "high_quality").order_by(Dataset.created_at.desc())
             )
 
             datasets = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
@@ -207,11 +204,11 @@ def migrate_knowledge_vector_database():
                     collection_name = Dataset.gen_collection_name_by_id(dataset_id)
                 elif vector_type == VectorType.QDRANT:
                     if dataset.collection_binding_id:
-                        dataset_collection_binding = db.session.execute(
-                            select(DatasetCollectionBinding).where(
-                                DatasetCollectionBinding.id == dataset.collection_binding_id
-                            )
-                        ).scalar_one_or_none()
+                        dataset_collection_binding = (
+                            db.session.query(DatasetCollectionBinding)
+                            .where(DatasetCollectionBinding.id == dataset.collection_binding_id)
+                            .one_or_none()
+                        )
                         if dataset_collection_binding:
                             collection_name = dataset_collection_binding.collection_name
                         else:
@@ -245,7 +242,7 @@ def migrate_knowledge_vector_database():
                 dataset_documents = db.session.scalars(
                     select(DatasetDocument).where(
                         DatasetDocument.dataset_id == dataset.id,
-                        DatasetDocument.indexing_status == IndexingStatus.COMPLETED,
+                        DatasetDocument.indexing_status == "completed",
                         DatasetDocument.enabled == True,
                         DatasetDocument.archived == False,
                     )
@@ -257,7 +254,7 @@ def migrate_knowledge_vector_database():
                     segments = db.session.scalars(
                         select(DocumentSegment).where(
                             DocumentSegment.document_id == dataset_document.id,
-                            DocumentSegment.status == SegmentStatus.COMPLETED,
+                            DocumentSegment.status == "completed",
                             DocumentSegment.enabled == True,
                         )
                     ).all()
@@ -272,7 +269,7 @@ def migrate_knowledge_vector_database():
                                 "dataset_id": segment.dataset_id,
                             },
                         )
-                        if dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX:
+                        if dataset_document.doc_form == "hierarchical_model":
                             child_chunks = segment.get_child_chunks()
                             if child_chunks:
                                 child_documents = []
@@ -336,7 +333,7 @@ def add_qdrant_index(field: str):
     create_count = 0
 
     try:
-        bindings = db.session.scalars(select(DatasetCollectionBinding)).all()
+        bindings = db.session.query(DatasetCollectionBinding).all()
         if not bindings:
             click.echo(click.style("No dataset collection bindings found.", fg="red"))
             return
@@ -423,22 +420,22 @@ def old_metadata_migration():
                         if field.value == key:
                             break
                     else:
-                        dataset_metadata = db.session.scalar(
-                            select(DatasetMetadata)
+                        dataset_metadata = (
+                            db.session.query(DatasetMetadata)
                             .where(DatasetMetadata.dataset_id == document.dataset_id, DatasetMetadata.name == key)
-                            .limit(1)
+                            .first()
                         )
                         if not dataset_metadata:
                             dataset_metadata = DatasetMetadata(
                                 tenant_id=document.tenant_id,
                                 dataset_id=document.dataset_id,
                                 name=key,
-                                type=DatasetMetadataType.STRING,
+                                type="string",
                                 created_by=document.created_by,
                             )
                             db.session.add(dataset_metadata)
                             db.session.flush()
-                            dataset_metadata_binding: DatasetMetadataBinding | None = DatasetMetadataBinding(
+                            dataset_metadata_binding = DatasetMetadataBinding(
                                 tenant_id=document.tenant_id,
                                 dataset_id=document.dataset_id,
                                 metadata_id=dataset_metadata.id,
@@ -447,14 +444,14 @@ def old_metadata_migration():
                             )
                             db.session.add(dataset_metadata_binding)
                         else:
-                            dataset_metadata_binding = db.session.scalar(
-                                select(DatasetMetadataBinding)
+                            dataset_metadata_binding = (
+                                db.session.query(DatasetMetadataBinding)  # type: ignore
                                 .where(
                                     DatasetMetadataBinding.dataset_id == document.dataset_id,
                                     DatasetMetadataBinding.document_id == document.id,
                                     DatasetMetadataBinding.metadata_id == dataset_metadata.id,
                                 )
-                                .limit(1)
+                                .first()
                             )
                             if not dataset_metadata_binding:
                                 dataset_metadata_binding = DatasetMetadataBinding(

api/configs/middleware/cache/redis_config.py

@@ -1,4 +1,4 @@
-from pydantic import Field, NonNegativeInt, PositiveFloat, PositiveInt, field_validator
+from pydantic import Field, NonNegativeInt, PositiveFloat, PositiveInt
 from pydantic_settings import BaseSettings
 
 
@@ -116,13 +116,3 @@ class RedisConfig(BaseSettings):
         description="Maximum connections in the Redis connection pool (unset for library default)",
         default=None,
     )
-
-    @field_validator("REDIS_MAX_CONNECTIONS", mode="before")
-    @classmethod
-    def _empty_string_to_none_for_max_conns(cls, v):
-        """Allow empty string in env/.env to mean 'unset' (None)."""
-        if v is None:
-            return None
-        if isinstance(v, str) and v.strip() == "":
-            return None
-        return v

api/configs/middleware/cache/redis_pubsub_config.py

@@ -1,4 +1,4 @@
-from typing import Literal, Protocol, cast
+from typing import Literal, Protocol
 from urllib.parse import quote_plus, urlunparse
 
 from pydantic import AliasChoices, Field
@@ -12,13 +12,16 @@ class RedisConfigDefaults(Protocol):
     REDIS_PASSWORD: str | None
     REDIS_DB: int
     REDIS_USE_SSL: bool
+    REDIS_USE_SENTINEL: bool | None
+    REDIS_USE_CLUSTERS: bool
 
 
-def _redis_defaults(config: object) -> RedisConfigDefaults:
-    return cast(RedisConfigDefaults, config)
+class RedisConfigDefaultsMixin:
+    def _redis_defaults(self: RedisConfigDefaults) -> RedisConfigDefaults:
+        return self
 
 
-class RedisPubSubConfig(BaseSettings):
+class RedisPubSubConfig(BaseSettings, RedisConfigDefaultsMixin):
     """
     Configuration settings for event transport between API and workers.
 
@@ -38,10 +41,10 @@ class RedisPubSubConfig(BaseSettings):
     )
 
     PUBSUB_REDIS_USE_CLUSTERS: bool = Field(
-        validation_alias=AliasChoices("EVENT_BUS_REDIS_USE_CLUSTERS", "PUBSUB_REDIS_USE_CLUSTERS"),
+        validation_alias=AliasChoices("EVENT_BUS_REDIS_CLUSTERS", "PUBSUB_REDIS_USE_CLUSTERS"),
         description=(
             "Enable Redis Cluster mode for pub/sub or streams transport. Recommended for large deployments. "
-            "Also accepts ENV: EVENT_BUS_REDIS_USE_CLUSTERS."
+            "Also accepts ENV: EVENT_BUS_REDIS_CLUSTERS."
         ),
         default=False,
     )
@@ -71,7 +74,7 @@ class RedisPubSubConfig(BaseSettings):
     )
 
     def _build_default_pubsub_url(self) -> str:
-        defaults = _redis_defaults(self)
+        defaults = self._redis_defaults()
         if not defaults.REDIS_HOST or not defaults.REDIS_PORT:
             raise ValueError("PUBSUB_REDIS_URL must be set when default Redis URL cannot be constructed")
 
@@ -88,9 +91,11 @@ class RedisPubSubConfig(BaseSettings):
         if userinfo:
             userinfo = f"{userinfo}@"
 
+        host = defaults.REDIS_HOST
+        port = defaults.REDIS_PORT
         db = defaults.REDIS_DB
 
-        netloc = f"{userinfo}{defaults.REDIS_HOST}:{defaults.REDIS_PORT}"
+        netloc = f"{userinfo}{host}:{port}"
         return urlunparse((scheme, netloc, f"/{db}", "", "", ""))
 
     @property

api/configs/middleware/vdb/baidu_vector_config.py

@@ -51,18 +51,3 @@ class BaiduVectorDBConfig(BaseSettings):
         description="Parser mode for inverted index in Baidu Vector Database (default is COARSE_MODE)",
         default="COARSE_MODE",
     )
-
-    BAIDU_VECTOR_DB_AUTO_BUILD_ROW_COUNT_INCREMENT: int = Field(
-        description="Auto build row count increment threshold (default is 500)",
-        default=500,
-    )
-
-    BAIDU_VECTOR_DB_AUTO_BUILD_ROW_COUNT_INCREMENT_RATIO: float = Field(
-        description="Auto build row count increment ratio threshold (default is 0.05)",
-        default=0.05,
-    )
-
-    BAIDU_VECTOR_DB_REBUILD_INDEX_TIMEOUT_IN_SECONDS: int = Field(
-        description="Timeout in seconds for rebuilding the index in Baidu Vector Database (default is 3600 seconds)",
-        default=300,
-    )

api/controllers/console/apikey.py

@@ -1,7 +1,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 import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden
 
@@ -9,7 +9,6 @@ from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
-from models.enums import ApiTokenType
 from models.model import ApiToken, App
 from services.api_token_service import ApiTokenCache
 
@@ -34,10 +33,16 @@ api_key_list_model = console_ns.model(
 
 
 def _get_resource(resource_id, tenant_id, resource_model):
-    with Session(db.engine) as session:
-        resource = session.execute(
-            select(resource_model).filter_by(id=resource_id, tenant_id=tenant_id)
-        ).scalar_one_or_none()
+    if resource_model == App:
+        with Session(db.engine) as session:
+            resource = session.execute(
+                select(resource_model).filter_by(id=resource_id, tenant_id=tenant_id)
+            ).scalar_one_or_none()
+    else:
+        with Session(db.engine) as session:
+            resource = session.execute(
+                select(resource_model).filter_by(id=resource_id, tenant_id=tenant_id)
+            ).scalar_one_or_none()
 
     if resource is None:
         flask_restx.abort(HTTPStatus.NOT_FOUND, message=f"{resource_model.__name__} not found.")
@@ -48,7 +53,7 @@ def _get_resource(resource_id, tenant_id, resource_model):
 class BaseApiKeyListResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
-    resource_type: ApiTokenType | None = None
+    resource_type: str | None = None
     resource_model: type | None = None
     resource_id_field: str | None = None
     token_prefix: str | None = None
@@ -75,13 +80,10 @@ class BaseApiKeyListResource(Resource):
         resource_id = str(resource_id)
         _, current_tenant_id = current_account_with_tenant()
         _get_resource(resource_id, current_tenant_id, self.resource_model)
-        current_key_count: int = (
-            db.session.scalar(
-                select(func.count(ApiToken.id)).where(
-                    ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
-                )
-            )
-            or 0
+        current_key_count = (
+            db.session.query(ApiToken)
+            .where(ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id)
+            .count()
         )
 
         if current_key_count >= self.max_keys:
@@ -92,7 +94,6 @@ class BaseApiKeyListResource(Resource):
             )
 
         key = ApiToken.generate_api_key(self.token_prefix or "", 24)
-        assert self.resource_type is not None, "resource_type must be set"
         api_token = ApiToken()
         setattr(api_token, self.resource_id_field, resource_id)
         api_token.tenant_id = current_tenant_id
@@ -106,7 +107,7 @@ class BaseApiKeyListResource(Resource):
 class BaseApiKeyResource(Resource):
     method_decorators = [account_initialization_required, login_required, setup_required]
 
-    resource_type: ApiTokenType | None = None
+    resource_type: str | None = None
     resource_model: type | None = None
     resource_id_field: str | None = None
 
@@ -118,14 +119,14 @@ class BaseApiKeyResource(Resource):
         if not current_user.is_admin_or_owner:
             raise Forbidden()
 
-        key = db.session.scalar(
-            select(ApiToken)
+        key = (
+            db.session.query(ApiToken)
             .where(
                 getattr(ApiToken, self.resource_id_field) == resource_id,
                 ApiToken.type == self.resource_type,
                 ApiToken.id == api_key_id,
             )
-            .limit(1)
+            .first()
         )
 
         if key is None:
@@ -136,7 +137,7 @@ class BaseApiKeyResource(Resource):
         assert key is not None  # nosec - for type checker only
         ApiTokenCache.delete(key.token, key.type)
 
-        db.session.execute(delete(ApiToken).where(ApiToken.id == api_key_id))
+        db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 
         return {"result": "success"}, 204
@@ -161,7 +162,7 @@ class AppApiKeyListResource(BaseApiKeyListResource):
         """Create a new API key for an app"""
         return super().post(resource_id)
 
-    resource_type = ApiTokenType.APP
+    resource_type = "app"
     resource_model = App
     resource_id_field = "app_id"
     token_prefix = "app-"
@@ -177,7 +178,7 @@ class AppApiKeyResource(BaseApiKeyResource):
         """Delete an API key for an app"""
         return super().delete(resource_id, api_key_id)
 
-    resource_type = ApiTokenType.APP
+    resource_type = "app"
     resource_model = App
     resource_id_field = "app_id"
 
@@ -201,7 +202,7 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
         """Create a new API key for a dataset"""
         return super().post(resource_id)
 
-    resource_type = ApiTokenType.DATASET
+    resource_type = "dataset"
     resource_model = Dataset
     resource_id_field = "dataset_id"
     token_prefix = "ds-"
@@ -217,6 +218,6 @@ class DatasetApiKeyResource(BaseApiKeyResource):
         """Delete an API key for a dataset"""
         return super().delete(resource_id, api_key_id)
 
-    resource_type = ApiTokenType.DATASET
+    resource_type = "dataset"
     resource_model = Dataset
     resource_id_field = "dataset_id"

api/controllers/console/app/app.py

@@ -95,7 +95,7 @@ class CreateAppPayload(BaseModel):
     name: str = Field(..., min_length=1, description="App name")
     description: str | None = Field(default=None, description="App description (max 400 chars)", max_length=400)
     mode: Literal["chat", "agent-chat", "advanced-chat", "workflow", "completion"] = Field(..., description="App mode")
-    icon_type: IconType | None = Field(default=None, description="Icon type")
+    icon_type: str | None = Field(default=None, description="Icon type")
     icon: str | None = Field(default=None, description="Icon")
     icon_background: str | None = Field(default=None, description="Icon background color")
 
@@ -103,7 +103,7 @@ class CreateAppPayload(BaseModel):
 class UpdateAppPayload(BaseModel):
     name: str = Field(..., min_length=1, description="App name")
     description: str | None = Field(default=None, description="App description (max 400 chars)", max_length=400)
-    icon_type: IconType | None = Field(default=None, description="Icon type")
+    icon_type: str | None = Field(default=None, description="Icon type")
     icon: str | None = Field(default=None, description="Icon")
     icon_background: str | None = Field(default=None, description="Icon background color")
     use_icon_as_answer_icon: bool | None = Field(default=None, description="Use icon as answer icon")
@@ -113,7 +113,7 @@ class UpdateAppPayload(BaseModel):
 class CopyAppPayload(BaseModel):
     name: str | None = Field(default=None, description="Name for the copied app")
     description: str | None = Field(default=None, description="Description for the copied app", max_length=400)
-    icon_type: IconType | None = Field(default=None, description="Icon type")
+    icon_type: str | None = Field(default=None, description="Icon type")
     icon: str | None = Field(default=None, description="Icon")
     icon_background: str | None = Field(default=None, description="Icon background color")
 
@@ -594,7 +594,7 @@ class AppApi(Resource):
         args_dict: AppService.ArgsDict = {
             "name": args.name,
             "description": args.description or "",
-            "icon_type": args.icon_type,
+            "icon_type": args.icon_type or "",
             "icon": args.icon or "",
             "icon_background": args.icon_background or "",
             "use_icon_as_answer_icon": args.use_icon_as_answer_icon or False,

api/controllers/console/app/conversation.py

@@ -5,7 +5,7 @@ from flask import abort, request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func, or_
-from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import joinedload
 from werkzeug.exceptions import NotFound
 
 from controllers.console import console_ns
@@ -376,12 +376,8 @@ class CompletionConversationApi(Resource):
 
         # FIXME, the type ignore in this file
         if args.annotation_status == "annotated":
-            query = (
-                query.options(selectinload(Conversation.message_annotations))  # type: ignore[arg-type]
-                .join(  # type: ignore
-                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-                )
-                .distinct()
+            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
             )
         elif args.annotation_status == "not_annotated":
             query = (
@@ -458,7 +454,9 @@ class ChatConversationApi(Resource):
         args = ChatConversationQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
 
         subquery = (
-            sa.select(Conversation.id.label("conversation_id"), EndUser.session_id.label("from_end_user_session_id"))
+            db.session.query(
+                Conversation.id.label("conversation_id"), EndUser.session_id.label("from_end_user_session_id")
+            )
             .outerjoin(EndUser, Conversation.from_end_user_id == EndUser.id)
             .subquery()
         )
@@ -513,12 +511,8 @@ class ChatConversationApi(Resource):
 
         match args.annotation_status:
             case "annotated":
-                query = (
-                    query.options(selectinload(Conversation.message_annotations))  # type: ignore[arg-type]
-                    .join(  # type: ignore
-                        MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-                    )
-                    .distinct()
+                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
                 )
             case "not_annotated":
                 query = (
@@ -593,8 +587,10 @@ class ChatConversationDetailApi(Resource):
 
 def _get_conversation(app_model, conversation_id):
     current_user, _ = current_account_with_tenant()
-    conversation = db.session.scalar(
-        sa.select(Conversation).where(Conversation.id == conversation_id, Conversation.app_id == app_model.id).limit(1)
+    conversation = (
+        db.session.query(Conversation)
+        .where(Conversation.id == conversation_id, Conversation.app_id == app_model.id)
+        .first()
     )
 
     if not conversation:

api/controllers/console/app/generator.py

@@ -168,7 +168,7 @@ class InstructionGenerateApi(Resource):
         try:
             # Generate from nothing for a workflow node
             if (args.current in (code_template, "")) and args.node_id != "":
-                app = db.session.get(App, args.flow_id)
+                app = db.session.query(App).where(App.id == args.flow_id).first()
                 if not app:
                     return {"error": f"app {args.flow_id} not found"}, 400
                 workflow = WorkflowService().get_draft_workflow(app_model=app)

api/controllers/console/app/mcp_server.py

@@ -2,7 +2,6 @@ import json
 
 from flask_restx import Resource, marshal_with
 from pydantic import BaseModel, Field
-from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from controllers.console import console_ns
@@ -48,7 +47,7 @@ class AppMCPServerController(Resource):
     @get_app_model
     @marshal_with(app_server_model)
     def get(self, app_model):
-        server = db.session.scalar(select(AppMCPServer).where(AppMCPServer.app_id == app_model.id).limit(1))
+        server = db.session.query(AppMCPServer).where(AppMCPServer.app_id == app_model.id).first()
         return server
 
     @console_ns.doc("create_app_mcp_server")
@@ -99,18 +98,18 @@ class AppMCPServerController(Resource):
     @edit_permission_required
     def put(self, app_model):
         payload = MCPServerUpdatePayload.model_validate(console_ns.payload or {})
-        server = db.session.get(AppMCPServer, payload.id)
+        server = db.session.query(AppMCPServer).where(AppMCPServer.id == payload.id).first()
         if not server:
             raise NotFound()
 
         description = payload.description
-        if description is None or not description:
+        if description is None:
+            pass
+        elif not description:
             server.description = app_model.description or ""
         else:
             server.description = description
 
-        server.name = app_model.name
-
         server.parameters = json.dumps(payload.parameters, ensure_ascii=False)
         if payload.status:
             try:
@@ -136,10 +135,11 @@ class AppMCPServerRefreshController(Resource):
     @edit_permission_required
     def get(self, server_id):
         _, current_tenant_id = current_account_with_tenant()
-        server = db.session.scalar(
-            select(AppMCPServer)
-            .where(AppMCPServer.id == server_id, AppMCPServer.tenant_id == current_tenant_id)
-            .limit(1)
+        server = (
+            db.session.query(AppMCPServer)
+            .where(AppMCPServer.id == server_id)
+            .where(AppMCPServer.tenant_id == current_tenant_id)
+            .first()
         )
         if not server:
             raise NotFound()

api/controllers/console/app/message.py

@@ -4,7 +4,7 @@ from typing import Literal
 from flask import request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import exists, func, select
+from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
 from controllers.common.schema import register_schema_models
@@ -30,7 +30,6 @@ from fields.raws import FilesContainedField
 from libs.helper import TimestampField, uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import current_account_with_tenant, login_required
-from models.enums import FeedbackFromSource, FeedbackRating
 from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
@@ -244,25 +243,27 @@ class ChatMessageListApi(Resource):
     def get(self, app_model):
         args = ChatMessagesQuery.model_validate(request.args.to_dict())
 
-        conversation = db.session.scalar(
-            select(Conversation)
+        conversation = (
+            db.session.query(Conversation)
             .where(Conversation.id == args.conversation_id, Conversation.app_id == app_model.id)
-            .limit(1)
+            .first()
         )
 
         if not conversation:
             raise NotFound("Conversation Not Exists.")
 
         if args.first_id:
-            first_message = db.session.scalar(
-                select(Message).where(Message.conversation_id == conversation.id, Message.id == args.first_id).limit(1)
+            first_message = (
+                db.session.query(Message)
+                .where(Message.conversation_id == conversation.id, Message.id == args.first_id)
+                .first()
             )
 
             if not first_message:
                 raise NotFound("First message not found")
 
-            history_messages = db.session.scalars(
-                select(Message)
+            history_messages = (
+                db.session.query(Message)
                 .where(
                     Message.conversation_id == conversation.id,
                     Message.created_at < first_message.created_at,
@@ -270,14 +271,16 @@ class ChatMessageListApi(Resource):
                 )
                 .order_by(Message.created_at.desc())
                 .limit(args.limit)
-            ).all()
+                .all()
+            )
         else:
-            history_messages = db.session.scalars(
-                select(Message)
+            history_messages = (
+                db.session.query(Message)
                 .where(Message.conversation_id == conversation.id)
                 .order_by(Message.created_at.desc())
                 .limit(args.limit)
-            ).all()
+                .all()
+            )
 
         # Initialize has_more based on whether we have a full page
         if len(history_messages) == args.limit:
@@ -322,9 +325,7 @@ class MessageFeedbackApi(Resource):
 
         message_id = str(args.message_id)
 
-        message = db.session.scalar(
-            select(Message).where(Message.id == message_id, Message.app_id == app_model.id).limit(1)
-        )
+        message = db.session.query(Message).where(Message.id == message_id, Message.app_id == app_model.id).first()
 
         if not message:
             raise NotFound("Message Not Exists.")
@@ -334,7 +335,7 @@ class MessageFeedbackApi(Resource):
         if not args.rating and feedback:
             db.session.delete(feedback)
         elif args.rating and feedback:
-            feedback.rating = FeedbackRating(args.rating)
+            feedback.rating = args.rating
             feedback.content = args.content
         elif not args.rating and not feedback:
             raise ValueError("rating cannot be None when feedback not exists")
@@ -346,9 +347,9 @@ class MessageFeedbackApi(Resource):
                 app_id=app_model.id,
                 conversation_id=message.conversation_id,
                 message_id=message.id,
-                rating=FeedbackRating(rating_value),
+                rating=rating_value,
                 content=args.content,
-                from_source=FeedbackFromSource.ADMIN,
+                from_source="admin",
                 from_account_id=current_user.id,
             )
             db.session.add(feedback)
@@ -373,9 +374,7 @@ class MessageAnnotationCountApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        count = db.session.scalar(
-            select(func.count(MessageAnnotation.id)).where(MessageAnnotation.app_id == app_model.id)
-        )
+        count = db.session.query(MessageAnnotation).where(MessageAnnotation.app_id == app_model.id).count()
 
         return {"count": count}
 
@@ -479,9 +478,7 @@ class MessageApi(Resource):
     def get(self, app_model, message_id: str):
         message_id = str(message_id)
 
-        message = db.session.scalar(
-            select(Message).where(Message.id == message_id, Message.app_id == app_model.id).limit(1)
-        )
+        message = db.session.query(Message).where(Message.id == message_id, Message.app_id == app_model.id).first()
 
         if not message:
             raise NotFound("Message Not Exists.")

api/controllers/console/app/model_config.py

@@ -69,7 +69,9 @@ class ModelConfigResource(Resource):
 
         if app_model.mode == AppMode.AGENT_CHAT or app_model.is_agent:
             # get original app model config
-            original_app_model_config = db.session.get(AppModelConfig, app_model.app_model_config_id)
+            original_app_model_config = (
+                db.session.query(AppModelConfig).where(AppModelConfig.id == app_model.app_model_config_id).first()
+            )
             if original_app_model_config is None:
                 raise ValueError("Original app model config not found")
             agent_mode = original_app_model_config.agent_mode_dict

api/controllers/console/app/site.py

@@ -2,7 +2,6 @@ from typing import Literal
 
 from flask_restx import Resource, marshal_with
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from constants.languages import supported_language
@@ -76,7 +75,7 @@ class AppSite(Resource):
     def post(self, app_model):
         args = AppSiteUpdatePayload.model_validate(console_ns.payload or {})
         current_user, _ = current_account_with_tenant()
-        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
         if not site:
             raise NotFound
 
@@ -125,7 +124,7 @@ class AppSiteAccessTokenReset(Resource):
     @marshal_with(app_site_model)
     def post(self, app_model):
         current_user, _ = current_account_with_tenant()
-        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
 
         if not site:
             raise NotFound

api/controllers/console/app/workflow.py

@@ -7,7 +7,7 @@ from flask import abort, request
 from flask_restx import Resource, fields, marshal_with
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
+from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
 from controllers.console import console_ns
@@ -46,14 +46,13 @@ from models import App
 from models.model import AppMode
 from models.workflow import Workflow
 from services.app_generate_service import AppGenerateService
-from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
+from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
 LISTENING_RETRY_IN = 2000
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE = "source workflow must be published"
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -285,9 +284,7 @@ class DraftWorkflowApi(Resource):
         workflow_service = WorkflowService()
 
         try:
-            environment_variables_list = Workflow.normalize_environment_variable_mappings(
-                args.get("environment_variables") or [],
-            )
+            environment_variables_list = args.get("environment_variables") or []
             environment_variables = [
                 variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
             ]
@@ -997,43 +994,6 @@ class PublishedAllWorkflowApi(Resource):
             }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>/restore")
-class DraftWorkflowRestoreApi(Resource):
-    @console_ns.doc("restore_workflow_to_draft")
-    @console_ns.doc(description="Restore a published workflow version into the draft workflow")
-    @console_ns.doc(params={"app_id": "Application ID", "workflow_id": "Published workflow ID"})
-    @console_ns.response(200, "Workflow restored successfully")
-    @console_ns.response(400, "Source workflow must be published")
-    @console_ns.response(404, "Workflow not found")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    @edit_permission_required
-    def post(self, app_model: App, workflow_id: str):
-        current_user, _ = current_account_with_tenant()
-        workflow_service = WorkflowService()
-
-        try:
-            workflow = workflow_service.restore_published_workflow_to_draft(
-                app_model=app_model,
-                workflow_id=workflow_id,
-                account=current_user,
-            )
-        except IsDraftWorkflowError as exc:
-            raise BadRequest(RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE) from exc
-        except WorkflowNotFoundError as exc:
-            raise NotFound(str(exc)) from exc
-        except ValueError as exc:
-            raise BadRequest(str(exc)) from exc
-
-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows/<string:workflow_id>")
 class WorkflowByIdApi(Resource):
     @console_ns.doc("update_workflow_by_id")

api/controllers/console/app/workflow_draft_variable.py

@@ -23,7 +23,7 @@ from dify_graph.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type
-from libs.login import current_user, login_required
+from libs.login import login_required
 from models import App, AppMode
 from models.workflow import WorkflowDraftVariable
 from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
@@ -100,18 +100,6 @@ def _serialize_full_content(variable: WorkflowDraftVariable) -> dict | None:
     }
 
 
-def _ensure_variable_access(
-    variable: WorkflowDraftVariable | None,
-    app_id: str,
-    variable_id: str,
-) -> WorkflowDraftVariable:
-    if variable is None:
-        raise NotFoundError(description=f"variable not found, id={variable_id}")
-    if variable.app_id != app_id or variable.user_id != current_user.id:
-        raise NotFoundError(description=f"variable not found, id={variable_id}")
-    return variable
-
-
 _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
     "id": fields.String,
     "type": fields.String(attribute=lambda model: model.get_variable_type()),
@@ -250,7 +238,6 @@ class WorkflowVariableCollectionApi(Resource):
                 app_id=app_model.id,
                 page=args.page,
                 limit=args.limit,
-                user_id=current_user.id,
             )
 
         return workflow_vars
@@ -263,7 +250,7 @@ class WorkflowVariableCollectionApi(Resource):
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
-        draft_var_srv.delete_user_workflow_variables(app_model.id, user_id=current_user.id)
+        draft_var_srv.delete_workflow_variables(app_model.id)
         db.session.commit()
         return Response("", 204)
 
@@ -300,7 +287,7 @@ class NodeVariableCollectionApi(Resource):
             draft_var_srv = WorkflowDraftVariableService(
                 session=session,
             )
-            node_vars = draft_var_srv.list_node_variables(app_model.id, node_id, user_id=current_user.id)
+            node_vars = draft_var_srv.list_node_variables(app_model.id, node_id)
 
         return node_vars
 
@@ -311,7 +298,7 @@ class NodeVariableCollectionApi(Resource):
     def delete(self, app_model: App, node_id: str):
         validate_node_id(node_id)
         srv = WorkflowDraftVariableService(db.session())
-        srv.delete_node_variables(app_model.id, node_id, user_id=current_user.id)
+        srv.delete_node_variables(app_model.id, node_id)
         db.session.commit()
         return Response("", 204)
 
@@ -332,11 +319,11 @@ class VariableApi(Resource):
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
-        variable = _ensure_variable_access(
-            variable=draft_var_srv.get_variable(variable_id=variable_id),
-            app_id=app_model.id,
-            variable_id=variable_id,
-        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
         return variable
 
     @console_ns.doc("update_variable")
@@ -373,11 +360,11 @@ class VariableApi(Resource):
         )
         args_model = WorkflowDraftVariableUpdatePayload.model_validate(console_ns.payload or {})
 
-        variable = _ensure_variable_access(
-            variable=draft_var_srv.get_variable(variable_id=variable_id),
-            app_id=app_model.id,
-            variable_id=variable_id,
-        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
 
         new_name = args_model.name
         raw_value = args_model.value
@@ -410,11 +397,11 @@ class VariableApi(Resource):
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
-        variable = _ensure_variable_access(
-            variable=draft_var_srv.get_variable(variable_id=variable_id),
-            app_id=app_model.id,
-            variable_id=variable_id,
-        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
         draft_var_srv.delete_variable(variable)
         db.session.commit()
         return Response("", 204)
@@ -440,11 +427,11 @@ class VariableResetApi(Resource):
             raise NotFoundError(
                 f"Draft workflow not found, app_id={app_model.id}",
             )
-        variable = _ensure_variable_access(
-            variable=draft_var_srv.get_variable(variable_id=variable_id),
-            app_id=app_model.id,
-            variable_id=variable_id,
-        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
 
         resetted = draft_var_srv.reset_variable(draft_workflow, variable)
         db.session.commit()
@@ -460,15 +447,11 @@ def _get_variable_list(app_model: App, node_id) -> WorkflowDraftVariableList:
             session=session,
         )
         if node_id == CONVERSATION_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_conversation_variables(app_model.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_conversation_variables(app_model.id)
         elif node_id == SYSTEM_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_system_variables(app_model.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_system_variables(app_model.id)
         else:
-            draft_vars = draft_var_srv.list_node_variables(
-                app_id=app_model.id,
-                node_id=node_id,
-                user_id=current_user.id,
-            )
+            draft_vars = draft_var_srv.list_node_variables(app_id=app_model.id, node_id=node_id)
     return draft_vars
 
 
@@ -489,7 +472,7 @@ class ConversationVariableCollectionApi(Resource):
         if draft_workflow is None:
             raise NotFoundError(description=f"draft workflow not found, id={app_model.id}")
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(draft_workflow, user_id=current_user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(draft_workflow)
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 

api/controllers/console/app/wraps.py

@@ -2,8 +2,6 @@ from collections.abc import Callable
 from functools import wraps
 from typing import ParamSpec, TypeVar, Union
 
-from sqlalchemy import select
-
 from controllers.console.app.error import AppNotFoundError
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
@@ -17,14 +15,16 @@ R1 = TypeVar("R1")
 
 def _load_app_model(app_id: str) -> App | None:
     _, current_tenant_id = current_account_with_tenant()
-    app_model = db.session.scalar(
-        select(App).where(App.id == app_id, App.tenant_id == current_tenant_id, App.status == "normal").limit(1)
+    app_model = (
+        db.session.query(App)
+        .where(App.id == app_id, App.tenant_id == current_tenant_id, App.status == "normal")
+        .first()
     )
     return app_model
 
 
 def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = db.session.scalar(select(App).where(App.id == app_id, App.status == "normal").limit(1))
+    app_model = db.session.query(App).where(App.id == app_id, App.status == "normal").first()
     return app_model
 
 

api/controllers/console/auth/email_register.py

@@ -1,7 +1,7 @@
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import languages
@@ -73,7 +73,7 @@ class EmailRegisterSendEmailApi(Resource):
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
             raise AccountInFreezeError()
 
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             account = AccountService.get_account_by_email_with_case_fallback(args.email, session=session)
         token = AccountService.send_email_register_email(email=normalized_email, account=account, language=language)
         return {"result": "success", "data": token}
@@ -145,7 +145,7 @@ class EmailRegisterResetApi(Resource):
         email = register_data.get("email", "")
         normalized_email = email.lower()
 
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             account = AccountService.get_account_by_email_with_case_fallback(email, session=session)
 
             if account:

api/controllers/console/auth/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 sessionmaker
+from sqlalchemy.orm import Session
 
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
@@ -102,7 +102,7 @@ class ForgotPasswordSendEmailApi(Resource):
         else:
             language = "en-US"
 
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             account = AccountService.get_account_by_email_with_case_fallback(args.email, session=session)
 
         token = AccountService.send_reset_password_email(
@@ -201,7 +201,7 @@ class ForgotPasswordResetApi(Resource):
         password_hashed = hash_password(args.new_password, salt)
 
         email = reset_data.get("email", "")
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             account = AccountService.get_account_by_email_with_case_fallback(email, session=session)
 
             if account:
@@ -215,6 +215,7 @@ class ForgotPasswordResetApi(Resource):
         # Update existing account credentials
         account.password = base64.b64encode(password_hashed).decode()
         account.password_salt = base64.b64encode(salt).decode()
+        session.commit()
 
         # Create workspace if needed
         if (

api/controllers/console/auth/oauth.py

@@ -1,10 +1,9 @@
 import logging
-import urllib.parse
 
 import httpx
 from flask import current_app, redirect, request
 from flask_restx import Resource
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import Unauthorized
 
 from configs import dify_config
@@ -113,9 +112,6 @@ class OAuthCallback(Resource):
                 error_text = e.response.text
             logger.exception("An error occurred during the OAuth process with %s: %s", provider, error_text)
             return {"error": "OAuth process failed"}, 400
-        except ValueError as e:
-            logger.warning("OAuth error with %s", provider, exc_info=True)
-            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={urllib.parse.quote(str(e))}")
 
         if invite_token and RegisterService.is_valid_invite_token(invite_token):
             invitation = RegisterService.get_invitation_by_token(token=invite_token)
@@ -180,7 +176,7 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
     account: Account | None = Account.get_by_openid(provider, user_info.id)
 
     if not account:
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             account = AccountService.get_account_by_email_with_case_fallback(user_info.email, session=session)
 
     return account

api/controllers/console/datasets/datasets.py

@@ -3,7 +3,7 @@ from typing import Any, cast
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import func, select
+from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
@@ -29,7 +29,6 @@ from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
@@ -55,7 +54,6 @@ from fields.document_fields import document_status_fields
 from libs.login import current_account_with_tenant, login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermission, DatasetPermissionEnum
-from models.enums import ApiTokenType, SegmentStatus
 from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
@@ -356,7 +354,7 @@ class DatasetListApi(Resource):
 
         for item in data:
             # convert embedding_model_provider to plugin standard format
-            if item["indexing_technique"] == IndexTechniqueType.HIGH_QUALITY and item["embedding_model_provider"]:
+            if item["indexing_technique"] == "high_quality" and item["embedding_model_provider"]:
                 item["embedding_model_provider"] = str(ModelProviderID(item["embedding_model_provider"]))
                 item_model = f"{item['embedding_model']}:{item['embedding_model_provider']}"
                 if item_model in model_names:
@@ -437,7 +435,7 @@ class DatasetApi(Resource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         data = cast(dict[str, Any], marshal(dataset, dataset_detail_fields))
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             if dataset.embedding_model_provider:
                 provider_id = ModelProviderID(dataset.embedding_model_provider)
                 data["embedding_model_provider"] = str(provider_id)
@@ -455,7 +453,7 @@ class DatasetApi(Resource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        if data["indexing_technique"] == IndexTechniqueType.HIGH_QUALITY:
+        if data["indexing_technique"] == "high_quality":
             item_model = f"{data['embedding_model']}:{data['embedding_model_provider']}"
             if item_model in model_names:
                 data["embedding_available"] = True
@@ -486,7 +484,7 @@ class DatasetApi(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
         # check embedding model setting
         if (
-            payload.indexing_technique == IndexTechniqueType.HIGH_QUALITY
+            payload.indexing_technique == "high_quality"
             and payload.embedding_model_provider is not None
             and payload.embedding_model is not None
         ):
@@ -739,23 +737,18 @@ class DatasetIndexingStatusApi(Resource):
         documents_status = []
         for document in documents:
             completed_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.completed_at.isnot(None),
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
+                db.session.query(DocumentSegment)
+                .where(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != "re_segment",
                 )
-                or 0
+                .count()
             )
             total_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
-                )
-                or 0
+                db.session.query(DocumentSegment)
+                .where(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                .count()
             )
             # Create a dictionary with document attributes and additional fields
             document_dict = {
@@ -781,7 +774,7 @@ class DatasetIndexingStatusApi(Resource):
 class DatasetApiKeyApi(Resource):
     max_keys = 10
     token_prefix = "dataset-"
-    resource_type = ApiTokenType.DATASET
+    resource_type = "dataset"
 
     @console_ns.doc("get_dataset_api_keys")
     @console_ns.doc(description="Get dataset API keys")
@@ -806,12 +799,9 @@ class DatasetApiKeyApi(Resource):
         _, current_tenant_id = current_account_with_tenant()
 
         current_key_count = (
-            db.session.scalar(
-                select(func.count(ApiToken.id)).where(
-                    ApiToken.type == self.resource_type, ApiToken.tenant_id == current_tenant_id
-                )
-            )
-            or 0
+            db.session.query(ApiToken)
+            .where(ApiToken.type == self.resource_type, ApiToken.tenant_id == current_tenant_id)
+            .count()
         )
 
         if current_key_count >= self.max_keys:
@@ -833,7 +823,7 @@ class DatasetApiKeyApi(Resource):
 
 @console_ns.route("/datasets/api-keys/<uuid:api_key_id>")
 class DatasetApiDeleteApi(Resource):
-    resource_type = ApiTokenType.DATASET
+    resource_type = "dataset"
 
     @console_ns.doc("delete_dataset_api_key")
     @console_ns.doc(description="Delete dataset API key")
@@ -846,14 +836,14 @@ class DatasetApiDeleteApi(Resource):
     def delete(self, api_key_id):
         _, current_tenant_id = current_account_with_tenant()
         api_key_id = str(api_key_id)
-        key = db.session.scalar(
-            select(ApiToken)
+        key = (
+            db.session.query(ApiToken)
             .where(
                 ApiToken.tenant_id == current_tenant_id,
                 ApiToken.type == self.resource_type,
                 ApiToken.id == api_key_id,
             )
-            .limit(1)
+            .first()
         )
 
         if key is None:
@@ -864,7 +854,7 @@ class DatasetApiDeleteApi(Resource):
         assert key is not None  # nosec - for type checker only
         ApiTokenCache.delete(key.token, key.type)
 
-        db.session.delete(key)
+        db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
         db.session.commit()
 
         return {"result": "success"}, 204

api/controllers/console/datasets/datasets_document.py

@@ -10,7 +10,7 @@ import sqlalchemy as sa
 from flask import request, send_file
 from flask_restx import Resource, fields, marshal, marshal_with
 from pydantic import BaseModel, Field
-from sqlalchemy import asc, desc, func, select
+from sqlalchemy import asc, desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
@@ -27,7 +27,6 @@ from core.model_manager import ModelManager
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from dify_graph.model_runtime.entities.model_entities import ModelType
 from dify_graph.model_runtime.errors.invoke import InvokeAuthorizationError
 from extensions.ext_database import db
@@ -43,7 +42,6 @@ from libs.datetime_utils import naive_utc_now
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
-from models.enums import IndexingStatus, SegmentStatus
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
 from services.file_service import FileService
@@ -212,11 +210,12 @@ class GetProcessRuleApi(Resource):
                 raise Forbidden(str(e))
 
             # get the latest process rule
-            dataset_process_rule = db.session.scalar(
-                select(DatasetProcessRule)
+            dataset_process_rule = (
+                db.session.query(DatasetProcessRule)
                 .where(DatasetProcessRule.dataset_id == document.dataset_id)
                 .order_by(DatasetProcessRule.created_at.desc())
                 .limit(1)
+                .one_or_none()
             )
             if dataset_process_rule:
                 mode = dataset_process_rule.mode
@@ -298,7 +297,6 @@ class DatasetDocumentListApi(Resource):
         if sort == "hit_count":
             sub_query = (
                 sa.select(DocumentSegment.document_id, sa.func.sum(DocumentSegment.hit_count).label("total_hit_count"))
-                .where(DocumentSegment.dataset_id == str(dataset_id))
                 .group_by(DocumentSegment.document_id)
                 .subquery()
             )
@@ -330,23 +328,18 @@ class DatasetDocumentListApi(Resource):
         if fetch:
             for document in documents:
                 completed_segments = (
-                    db.session.scalar(
-                        select(func.count(DocumentSegment.id)).where(
-                            DocumentSegment.completed_at.isnot(None),
-                            DocumentSegment.document_id == str(document.id),
-                            DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                        )
+                    db.session.query(DocumentSegment)
+                    .where(
+                        DocumentSegment.completed_at.isnot(None),
+                        DocumentSegment.document_id == str(document.id),
+                        DocumentSegment.status != "re_segment",
                     )
-                    or 0
+                    .count()
                 )
                 total_segments = (
-                    db.session.scalar(
-                        select(func.count(DocumentSegment.id)).where(
-                            DocumentSegment.document_id == str(document.id),
-                            DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                        )
-                    )
-                    or 0
+                    db.session.query(DocumentSegment)
+                    .where(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                    .count()
                 )
                 document.completed_segments = completed_segments
                 document.total_segments = total_segments
@@ -450,7 +443,7 @@ class DatasetInitApi(Resource):
             raise Forbidden()
 
         knowledge_config = KnowledgeConfig.model_validate(console_ns.payload or {})
-        if knowledge_config.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if knowledge_config.indexing_technique == "high_quality":
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
                 raise ValueError("embedding model and embedding model provider are required for high quality indexing.")
             try:
@@ -464,7 +457,7 @@ class DatasetInitApi(Resource):
                 is_multimodal = DatasetService.check_is_multimodal_model(
                     current_tenant_id, knowledge_config.embedding_model_provider, knowledge_config.embedding_model
                 )
-                knowledge_config.is_multimodal = is_multimodal  # pyrefly: ignore[bad-assignment]
+                knowledge_config.is_multimodal = is_multimodal
             except InvokeAuthorizationError:
                 raise ProviderNotInitializeError(
                     "No Embedding Model available. Please configure a valid provider in the Settings -> Model Provider."
@@ -510,7 +503,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
         document_id = str(document_id)
         document = self.get_document(dataset_id, document_id)
 
-        if document.indexing_status in {IndexingStatus.COMPLETED, IndexingStatus.ERROR}:
+        if document.indexing_status in {"completed", "error"}:
             raise DocumentAlreadyFinishedError()
 
         data_process_rule = document.dataset_process_rule
@@ -523,10 +516,10 @@ class DocumentIndexingEstimateApi(DocumentResource):
             if data_source_info and "upload_file_id" in data_source_info:
                 file_id = data_source_info["upload_file_id"]
 
-                file = db.session.scalar(
-                    select(UploadFile)
+                file = (
+                    db.session.query(UploadFile)
                     .where(UploadFile.tenant_id == document.tenant_id, UploadFile.id == file_id)
-                    .limit(1)
+                    .first()
                 )
 
                 # raise error if file not found
@@ -580,7 +573,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
         data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
         extract_settings = []
         for document in documents:
-            if document.indexing_status in {IndexingStatus.COMPLETED, IndexingStatus.ERROR}:
+            if document.indexing_status in {"completed", "error"}:
                 raise DocumentAlreadyFinishedError()
             data_source_info = document.data_source_info_dict
             match document.data_source_type:
@@ -588,10 +581,10 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     if not data_source_info:
                         continue
                     file_id = data_source_info["upload_file_id"]
-                    file_detail = db.session.scalar(
-                        select(UploadFile)
+                    file_detail = (
+                        db.session.query(UploadFile)
                         .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
-                        .limit(1)
+                        .first()
                     )
 
                     if file_detail is None:
@@ -674,28 +667,23 @@ class DocumentBatchIndexingStatusApi(DocumentResource):
         documents_status = []
         for document in documents:
             completed_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.completed_at.isnot(None),
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
+                db.session.query(DocumentSegment)
+                .where(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != "re_segment",
                 )
-                or 0
+                .count()
             )
             total_segments = (
-                db.session.scalar(
-                    select(func.count(DocumentSegment.id)).where(
-                        DocumentSegment.document_id == str(document.id),
-                        DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                    )
-                )
-                or 0
+                db.session.query(DocumentSegment)
+                .where(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                .count()
             )
             # Create a dictionary with document attributes and additional fields
             document_dict = {
                 "id": document.id,
-                "indexing_status": IndexingStatus.PAUSED if document.is_paused else document.indexing_status,
+                "indexing_status": "paused" if document.is_paused else document.indexing_status,
                 "processing_started_at": document.processing_started_at,
                 "parsing_completed_at": document.parsing_completed_at,
                 "cleaning_completed_at": document.cleaning_completed_at,
@@ -728,29 +716,24 @@ class DocumentIndexingStatusApi(DocumentResource):
         document = self.get_document(dataset_id, document_id)
 
         completed_segments = (
-            db.session.scalar(
-                select(func.count(DocumentSegment.id)).where(
-                    DocumentSegment.completed_at.isnot(None),
-                    DocumentSegment.document_id == str(document_id),
-                    DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                )
+            db.session.query(DocumentSegment)
+            .where(
+                DocumentSegment.completed_at.isnot(None),
+                DocumentSegment.document_id == str(document_id),
+                DocumentSegment.status != "re_segment",
             )
-            or 0
+            .count()
         )
         total_segments = (
-            db.session.scalar(
-                select(func.count(DocumentSegment.id)).where(
-                    DocumentSegment.document_id == str(document_id),
-                    DocumentSegment.status != SegmentStatus.RE_SEGMENT,
-                )
-            )
-            or 0
+            db.session.query(DocumentSegment)
+            .where(DocumentSegment.document_id == str(document_id), DocumentSegment.status != "re_segment")
+            .count()
         )
 
         # Create a dictionary with document attributes and additional fields
         document_dict = {
             "id": document.id,
-            "indexing_status": IndexingStatus.PAUSED if document.is_paused else document.indexing_status,
+            "indexing_status": "paused" if document.is_paused else document.indexing_status,
             "processing_started_at": document.processing_started_at,
             "parsing_completed_at": document.parsing_completed_at,
             "cleaning_completed_at": document.cleaning_completed_at,
@@ -972,7 +955,7 @@ class DocumentProcessingApi(DocumentResource):
 
         match action:
             case "pause":
-                if document.indexing_status != IndexingStatus.INDEXING:
+                if document.indexing_status != "indexing":
                     raise InvalidActionError("Document not in indexing state.")
 
                 document.paused_by = current_user.id
@@ -981,7 +964,7 @@ class DocumentProcessingApi(DocumentResource):
                 db.session.commit()
 
             case "resume":
-                if document.indexing_status not in {IndexingStatus.PAUSED, IndexingStatus.ERROR}:
+                if document.indexing_status not in {"paused", "error"}:
                     raise InvalidActionError("Document not in paused or error state.")
 
                 document.paused_by = None
@@ -1186,7 +1169,7 @@ class DocumentRetryApi(DocumentResource):
                     raise ArchivedDocumentImmutableError()
 
                 # 400 if document is completed
-                if document.indexing_status == IndexingStatus.COMPLETED:
+                if document.indexing_status == "completed":
                     raise DocumentAlreadyFinishedError()
                 retry_documents.append(document)
             except Exception:
@@ -1268,11 +1251,11 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
         document = DocumentService.get_document(dataset.id, document_id)
         if not document:
             raise NotFound("Document not found.")
-        log = db.session.scalar(
-            select(DocumentPipelineExecutionLog)
-            .where(DocumentPipelineExecutionLog.document_id == document_id)
+        log = (
+            db.session.query(DocumentPipelineExecutionLog)
+            .filter_by(document_id=document_id)
             .order_by(DocumentPipelineExecutionLog.created_at.desc())
-            .limit(1)
+            .first()
         )
         if not log:
             return {
@@ -1338,7 +1321,7 @@ class DocumentGenerateSummaryApi(Resource):
             raise BadRequest("document_list cannot be empty.")
 
         # Check if dataset configuration supports summary generation
-        if dataset.indexing_technique != IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique != "high_quality":
             raise ValueError(
                 f"Summary generation is only available for 'high_quality' indexing technique. "
                 f"Current indexing technique: {dataset.indexing_technique}"

api/controllers/console/datasets/datasets_segments.py

@@ -26,7 +26,6 @@ from controllers.console.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -46,7 +45,7 @@ def _get_segment_with_summary(segment, dataset_id):
     """Helper function to marshal segment and add summary information."""
     from services.summary_index_service import SummaryIndexService
 
-    segment_dict = dict(marshal(segment, segment_fields))  # type: ignore
+    segment_dict = dict(marshal(segment, segment_fields))
     # Query summary for this segment (only enabled summaries)
     summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id)
     segment_dict["summary"] = summary.summary_content if summary else None
@@ -207,7 +206,7 @@ class DatasetDocumentSegmentListApi(Resource):
         # Add summary to each segment
         segments_with_summary = []
         for segment in segments.items:
-            segment_dict = dict(marshal(segment, segment_fields))  # type: ignore
+            segment_dict = dict(marshal(segment, segment_fields))
             segment_dict["summary"] = summaries.get(segment.id)
             segments_with_summary.append(segment_dict)
 
@@ -280,7 +279,7 @@ class DatasetDocumentSegmentApi(Resource):
             DatasetService.check_dataset_permission(dataset, current_user)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             # check embedding model setting
             try:
                 model_manager = ModelManager()
@@ -334,7 +333,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         # check embedding model setting
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             try:
                 model_manager = ModelManager()
                 model_manager.get_model_instance(
@@ -384,7 +383,7 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         document = DocumentService.get_document(dataset_id, document_id)
         if not document:
             raise NotFound("Document not found.")
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             # check embedding model setting
             try:
                 model_manager = ModelManager()
@@ -402,10 +401,10 @@ class DatasetDocumentSegmentUpdateApi(Resource):
                 raise ProviderNotInitializeError(ex.description)
             # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
@@ -448,10 +447,10 @@ class DatasetDocumentSegmentUpdateApi(Resource):
             raise NotFound("Document not found.")
         # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
@@ -495,7 +494,7 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
         payload = BatchImportPayload.model_validate(console_ns.payload or {})
         upload_file_id = payload.upload_file_id
 
-        upload_file = db.session.scalar(select(UploadFile).where(UploadFile.id == upload_file_id).limit(1))
+        upload_file = db.session.query(UploadFile).where(UploadFile.id == upload_file_id).first()
         if not upload_file:
             raise NotFound("UploadFile not found.")
 
@@ -560,17 +559,17 @@ class ChildChunkAddApi(Resource):
             raise NotFound("Document not found.")
         # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
         if not current_user.is_dataset_editor:
             raise Forbidden()
         # check embedding model setting
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             try:
                 model_manager = ModelManager()
                 model_manager.get_model_instance(
@@ -617,10 +616,10 @@ class ChildChunkAddApi(Resource):
             raise NotFound("Document not found.")
         # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
@@ -667,10 +666,10 @@ class ChildChunkAddApi(Resource):
             raise NotFound("Document not found.")
             # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
@@ -715,24 +714,24 @@ class ChildChunkUpdateApi(Resource):
             raise NotFound("Document not found.")
         # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
         # check child chunk
         child_chunk_id = str(child_chunk_id)
-        child_chunk = db.session.scalar(
-            select(ChildChunk)
+        child_chunk = (
+            db.session.query(ChildChunk)
             .where(
                 ChildChunk.id == str(child_chunk_id),
                 ChildChunk.tenant_id == current_tenant_id,
                 ChildChunk.segment_id == segment.id,
                 ChildChunk.document_id == document_id,
             )
-            .limit(1)
+            .first()
         )
         if not child_chunk:
             raise NotFound("Child chunk not found.")
@@ -772,24 +771,24 @@ class ChildChunkUpdateApi(Resource):
             raise NotFound("Document not found.")
             # check segment
         segment_id = str(segment_id)
-        segment = db.session.scalar(
-            select(DocumentSegment)
+        segment = (
+            db.session.query(DocumentSegment)
             .where(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_tenant_id)
-            .limit(1)
+            .first()
         )
         if not segment:
             raise NotFound("Segment not found.")
         # check child chunk
         child_chunk_id = str(child_chunk_id)
-        child_chunk = db.session.scalar(
-            select(ChildChunk)
+        child_chunk = (
+            db.session.query(ChildChunk)
             .where(
                 ChildChunk.id == str(child_chunk_id),
                 ChildChunk.tenant_id == current_tenant_id,
                 ChildChunk.segment_id == segment.id,
                 ChildChunk.document_id == document_id,
             )
-            .limit(1)
+            .first()
         )
         if not child_chunk:
             raise NotFound("Child chunk not found.")

api/controllers/console/datasets/hit_testing_base.py

@@ -24,7 +24,6 @@ from fields.hit_testing_fields import hit_testing_record_fields
 from libs.login import current_user
 from models.account import Account
 from services.dataset_service import DatasetService
-from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
 from services.hit_testing_service import HitTestingService
 
 logger = logging.getLogger(__name__)
@@ -32,7 +31,7 @@ logger = logging.getLogger(__name__)
 
 class HitTestingPayload(BaseModel):
     query: str = Field(max_length=250)
-    retrieval_model: RetrievalModel | None = None
+    retrieval_model: dict[str, Any] | None = None
     external_retrieval_model: dict[str, Any] | None = None
     attachment_ids: list[str] | None = None
 

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

@@ -46,8 +46,6 @@ class PipelineTemplateDetailApi(Resource):
         type = request.args.get("type", default="built-in", type=str)
         rag_pipeline_service = RagPipelineService()
         pipeline_template = rag_pipeline_service.get_pipeline_template_detail(template_id, type)
-        if pipeline_template is None:
-            return {"error": "Pipeline template not found from upstream service."}, 404
         return pipeline_template, 200
 
 

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

@@ -102,7 +102,6 @@ class RagPipelineVariableCollectionApi(Resource):
             app_id=pipeline.id,
             page=query.page,
             limit=query.limit,
-            user_id=current_user.id,
         )
 
         return workflow_vars
@@ -112,7 +111,7 @@ class RagPipelineVariableCollectionApi(Resource):
         draft_var_srv = WorkflowDraftVariableService(
             session=db.session(),
         )
-        draft_var_srv.delete_user_workflow_variables(pipeline.id, user_id=current_user.id)
+        draft_var_srv.delete_workflow_variables(pipeline.id)
         db.session.commit()
         return Response("", 204)
 
@@ -145,7 +144,7 @@ class RagPipelineNodeVariableCollectionApi(Resource):
             draft_var_srv = WorkflowDraftVariableService(
                 session=session,
             )
-            node_vars = draft_var_srv.list_node_variables(pipeline.id, node_id, user_id=current_user.id)
+            node_vars = draft_var_srv.list_node_variables(pipeline.id, node_id)
 
         return node_vars
 
@@ -153,7 +152,7 @@ class RagPipelineNodeVariableCollectionApi(Resource):
     def delete(self, pipeline: Pipeline, node_id: str):
         validate_node_id(node_id)
         srv = WorkflowDraftVariableService(db.session())
-        srv.delete_node_variables(pipeline.id, node_id, user_id=current_user.id)
+        srv.delete_node_variables(pipeline.id, node_id)
         db.session.commit()
         return Response("", 204)
 
@@ -284,11 +283,11 @@ def _get_variable_list(pipeline: Pipeline, node_id) -> WorkflowDraftVariableList
             session=session,
         )
         if node_id == CONVERSATION_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_conversation_variables(pipeline.id)
         elif node_id == SYSTEM_VARIABLE_NODE_ID:
-            draft_vars = draft_var_srv.list_system_variables(pipeline.id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_system_variables(pipeline.id)
         else:
-            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id, user_id=current_user.id)
+            draft_vars = draft_var_srv.list_node_variables(app_id=pipeline.id, node_id=node_id)
     return draft_vars
 
 

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

@@ -6,7 +6,7 @@ from flask import abort, request
 from flask_restx import Resource, marshal_with  # type: ignore
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
+from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
 from controllers.common.schema import register_schema_models
@@ -16,11 +16,7 @@ from controllers.console.app.error import (
     DraftWorkflowNotExist,
     DraftWorkflowNotSync,
 )
-from controllers.console.app.workflow import (
-    RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE,
-    workflow_model,
-    workflow_pagination_model,
-)
+from controllers.console.app.workflow import workflow_model, workflow_pagination_model
 from controllers.console.app.workflow_run import (
     workflow_run_detail_model,
     workflow_run_node_execution_list_model,
@@ -46,8 +42,7 @@ from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
 from models.model import EndUser
-from models.workflow import Workflow
-from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
+from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
 from services.rag_pipeline.pipeline_generate_service import PipelineGenerateService
 from services.rag_pipeline.rag_pipeline import RagPipelineService
@@ -208,12 +203,9 @@ class DraftRagPipelineApi(Resource):
             abort(415)
 
         payload = DraftWorkflowSyncPayload.model_validate(payload_dict)
-        rag_pipeline_service = RagPipelineService()
 
         try:
-            environment_variables_list = Workflow.normalize_environment_variable_mappings(
-                payload.environment_variables or [],
-            )
+            environment_variables_list = payload.environment_variables or []
             environment_variables = [
                 variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
             ]
@@ -221,6 +213,7 @@ class DraftRagPipelineApi(Resource):
             conversation_variables = [
                 variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
             ]
+            rag_pipeline_service = RagPipelineService()
             workflow = rag_pipeline_service.sync_draft_workflow(
                 pipeline=pipeline,
                 graph=payload.graph,
@@ -712,36 +705,6 @@ class PublishedAllRagPipelineApi(Resource):
             }
 
 
-@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>/restore")
-class RagPipelineDraftWorkflowRestoreApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @edit_permission_required
-    @get_rag_pipeline
-    def post(self, pipeline: Pipeline, workflow_id: str):
-        current_user, _ = current_account_with_tenant()
-        rag_pipeline_service = RagPipelineService()
-
-        try:
-            workflow = rag_pipeline_service.restore_published_workflow_to_draft(
-                pipeline=pipeline,
-                workflow_id=workflow_id,
-                account=current_user,
-            )
-        except IsDraftWorkflowError as exc:
-            # Use a stable, predefined message to keep the 400 response consistent
-            raise BadRequest(RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE) from exc
-        except WorkflowNotFoundError as exc:
-            raise NotFound(str(exc)) from exc
-
-        return {
-            "result": "success",
-            "hash": workflow.unique_hash,
-            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
-        }
-
-
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/<string:workflow_id>")
 class RagPipelineByIdApi(Resource):
     @setup_required

api/controllers/console/datasets/wraps.py

@@ -2,8 +2,6 @@ from collections.abc import Callable
 from functools import wraps
 from typing import ParamSpec, TypeVar
 
-from sqlalchemy import select
-
 from controllers.console.datasets.error import PipelineNotFoundError
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
@@ -26,8 +24,10 @@ def get_rag_pipeline(view_func: Callable[P, R]):
 
         del kwargs["pipeline_id"]
 
-        pipeline = db.session.scalar(
-            select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
+        pipeline = (
+            db.session.query(Pipeline)
+            .where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id)
+            .first()
         )
 
         if not pipeline:

api/controllers/console/explore/banner.py

@@ -1,11 +1,9 @@
 from flask import request
 from flask_restx import Resource
-from sqlalchemy import select
 
 from controllers.console import api
 from controllers.console.explore.wraps import explore_banner_enabled
 from extensions.ext_database import db
-from models.enums import BannerStatus
 from models.model import ExporleBanner
 
 
@@ -18,18 +16,14 @@ class BannerApi(Resource):
         language = request.args.get("language", "en-US")
 
         # Build base query for enabled banners
-        base_query = select(ExporleBanner).where(ExporleBanner.status == BannerStatus.ENABLED)
+        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
 
         # Try to get banners in the requested language
-        banners = db.session.scalars(
-            base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort)
-        ).all()
+        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
 
         # Fallback to en-US if no banners found and language is not en-US
         if not banners and language != "en-US":
-            banners = db.session.scalars(
-                base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort)
-            ).all()
+            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
         # Convert banners to serializable format
         result = []
         for banner in banners:

api/controllers/console/explore/installed_app.py

@@ -133,15 +133,13 @@ class InstalledAppsListApi(Resource):
     def post(self):
         payload = InstalledAppCreatePayload.model_validate(console_ns.payload or {})
 
-        recommended_app = db.session.scalar(
-            select(RecommendedApp).where(RecommendedApp.app_id == payload.app_id).limit(1)
-        )
+        recommended_app = db.session.query(RecommendedApp).where(RecommendedApp.app_id == payload.app_id).first()
         if recommended_app is None:
             raise NotFound("Recommended app not found")
 
         _, current_tenant_id = current_account_with_tenant()
 
-        app = db.session.get(App, payload.app_id)
+        app = db.session.query(App).where(App.id == payload.app_id).first()
 
         if app is None:
             raise NotFound("App entity not found")
@@ -149,10 +147,10 @@ class InstalledAppsListApi(Resource):
         if not app.is_public:
             raise Forbidden("You can't install a non-public app")
 
-        installed_app = db.session.scalar(
-            select(InstalledApp)
+        installed_app = (
+            db.session.query(InstalledApp)
             .where(and_(InstalledApp.app_id == payload.app_id, InstalledApp.tenant_id == current_tenant_id))
-            .limit(1)
+            .first()
         )
 
         if installed_app is None:

api/controllers/console/explore/message.py

@@ -27,7 +27,6 @@ from fields.message_fields import MessageInfiniteScrollPagination, MessageListIt
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
 from libs.login import current_account_with_tenant
-from models.enums import FeedbackRating
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.app import MoreLikeThisDisabledError
@@ -117,7 +116,7 @@ class MessageFeedbackApi(InstalledAppResource):
                 app_model=app_model,
                 message_id=message_id,
                 user=current_user,
-                rating=FeedbackRating(payload.rating) if payload.rating else None,
+                rating=payload.rating,
                 content=payload.content,
             )
         except MessageNotExistsError:

api/controllers/console/explore/trial.py

@@ -4,7 +4,6 @@ from typing import Any, Literal, cast
 from flask import request
 from flask_restx import Resource, fields, marshal, marshal_with
 from pydantic import BaseModel
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
@@ -477,7 +476,7 @@ class TrialSitApi(Resource):
 
         Returns the site configuration for the application including theme, icons, and text.
         """
-        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
 
         if not site:
             raise Forbidden()
@@ -542,7 +541,13 @@ class AppWorkflowApi(Resource):
         if not app_model.workflow_id:
             raise AppUnavailableError()
 
-        workflow = db.session.get(Workflow, app_model.workflow_id)
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.id == app_model.workflow_id,
+            )
+            .first()
+        )
         return workflow
 
 

api/controllers/console/explore/wraps.py

@@ -4,7 +4,6 @@ from typing import Concatenate, ParamSpec, TypeVar
 
 from flask import abort
 from flask_restx import Resource
-from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
@@ -25,10 +24,10 @@ def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | Non
         @wraps(view)
         def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
             _, current_tenant_id = current_account_with_tenant()
-            installed_app = db.session.scalar(
-                select(InstalledApp)
+            installed_app = (
+                db.session.query(InstalledApp)
                 .where(InstalledApp.id == str(installed_app_id), InstalledApp.tenant_id == current_tenant_id)
-                .limit(1)
+                .first()
             )
 
             if installed_app is None:
@@ -79,7 +78,7 @@ def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
         def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
             current_user, _ = current_account_with_tenant()
 
-            trial_app = db.session.scalar(select(TrialApp).where(TrialApp.app_id == str(app_id)).limit(1))
+            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
 
             if trial_app is None:
                 raise TrialAppNotAllowed()
@@ -88,10 +87,10 @@ def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
             if app is None:
                 raise TrialAppNotAllowed()
 
-            account_trial_app_record = db.session.scalar(
-                select(AccountTrialAppRecord)
+            account_trial_app_record = (
+                db.session.query(AccountTrialAppRecord)
                 .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
-                .limit(1)
+                .first()
             )
             if account_trial_app_record:
                 if account_trial_app_record.count >= trial_app.trial_limit:

api/controllers/console/setup.py

@@ -2,7 +2,6 @@ from typing import Literal
 
 from flask import request
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
 
 from configs import dify_config
 from controllers.fastopenapi import console_router
@@ -101,6 +100,6 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
 
 def get_setup_status() -> DifySetup | bool | None:
     if dify_config.EDITION == "SELF_HOSTED":
-        return db.session.scalar(select(DifySetup).limit(1))
+        return db.session.query(DifySetup).first()
 
     return True

api/controllers/console/workspace/account.py

@@ -212,13 +212,13 @@ class AccountInitApi(Resource):
                 raise ValueError("invitation_code is required")
 
             # check invitation code
-            invitation_code = db.session.scalar(
-                select(InvitationCode)
+            invitation_code = (
+                db.session.query(InvitationCode)
                 .where(
                     InvitationCode.code == args.invitation_code,
                     InvitationCode.status == InvitationCodeStatus.UNUSED,
                 )
-                .limit(1)
+                .first()
             )
 
             if not invitation_code:

api/controllers/console/workspace/members.py

@@ -171,7 +171,7 @@ class MemberCancelInviteApi(Resource):
         current_user, _ = current_account_with_tenant()
         if not current_user.current_tenant:
             raise ValueError("No current tenant")
-        member = db.session.get(Account, str(member_id))
+        member = db.session.query(Account).where(Account.id == str(member_id)).first()
         if member is None:
             abort(404)
         else:

api/controllers/console/workspace/workspace.py

@@ -7,7 +7,6 @@ from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
 
 import services
-from configs import dify_config
 from controllers.common.errors import (
     FilenameNotExistsError,
     FileTooLargeError,
@@ -30,7 +29,6 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.account import Tenant, TenantStatus
 from services.account_service import TenantService
-from services.billing_service import BillingService, SubscriptionPlan
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 from services.file_service import FileService
@@ -110,29 +108,9 @@ class TenantListApi(Resource):
         current_user, current_tenant_id = current_account_with_tenant()
         tenants = TenantService.get_join_tenants(current_user)
         tenant_dicts = []
-        is_enterprise_only = dify_config.ENTERPRISE_ENABLED and not dify_config.BILLING_ENABLED
-        is_saas = dify_config.EDITION == "CLOUD" and dify_config.BILLING_ENABLED
-        tenant_plans: dict[str, SubscriptionPlan] = {}
-
-        if is_saas:
-            tenant_ids = [tenant.id for tenant in tenants]
-            if tenant_ids:
-                tenant_plans = BillingService.get_plan_bulk(tenant_ids)
-                if not tenant_plans:
-                    logger.warning("get_plan_bulk returned empty result, falling back to legacy feature path")
 
         for tenant in tenants:
-            plan: str = CloudPlan.SANDBOX
-            if is_saas:
-                tenant_plan = tenant_plans.get(tenant.id)
-                if tenant_plan:
-                    plan = tenant_plan["plan"] or CloudPlan.SANDBOX
-                else:
-                    features = FeatureService.get_features(tenant.id)
-                    plan = features.billing.subscription.plan or CloudPlan.SANDBOX
-            elif not is_enterprise_only:
-                features = FeatureService.get_features(tenant.id)
-                plan = features.billing.subscription.plan or CloudPlan.SANDBOX
+            features = FeatureService.get_features(tenant.id)
 
             # Create a dictionary with tenant attributes
             tenant_dict = {
@@ -140,7 +118,7 @@ class TenantListApi(Resource):
                 "name": tenant.name,
                 "status": tenant.status,
                 "created_at": tenant.created_at,
-                "plan": plan,
+                "plan": features.billing.subscription.plan if features.billing.enabled else CloudPlan.SANDBOX,
                 "current": tenant.id == current_tenant_id if current_tenant_id else False,
             }
 
@@ -220,7 +198,7 @@ class SwitchWorkspaceApi(Resource):
         except Exception:
             raise AccountNotLinkTenantError("Account not link tenant")
 
-        new_tenant = db.session.get(Tenant, args.tenant_id)  # Get new tenant
+        new_tenant = db.session.query(Tenant).get(args.tenant_id)  # Get new tenant
         if new_tenant is None:
             raise ValueError("Tenant not found")
 

api/controllers/console/wraps.py

@@ -7,7 +7,6 @@ from functools import wraps
 from typing import ParamSpec, TypeVar
 
 from flask import abort, request
-from sqlalchemy import select
 
 from configs import dify_config
 from controllers.console.auth.error import AuthenticationFailedError, EmailCodeError
@@ -219,9 +218,13 @@ def setup_required(view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
         # check setup
-        if dify_config.EDITION == "SELF_HOSTED" and not db.session.scalar(select(DifySetup).limit(1)):
-            if os.environ.get("INIT_PASSWORD"):
-                raise NotInitValidateError()
+        if (
+            dify_config.EDITION == "SELF_HOSTED"
+            and os.environ.get("INIT_PASSWORD")
+            and not db.session.query(DifySetup).first()
+        ):
+            raise NotInitValidateError()
+        elif dify_config.EDITION == "SELF_HOSTED" and not db.session.query(DifySetup).first():
             raise NotSetupError()
 
         return view(*args, **kwargs)

api/controllers/inner_api/plugin/wraps.py

@@ -5,7 +5,6 @@ from typing import ParamSpec, TypeVar
 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 extensions.ext_database import db
@@ -37,16 +36,23 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
             user_model = None
 
             if is_anonymous:
-                user_model = session.scalar(
-                    select(EndUser)
+                user_model = (
+                    session.query(EndUser)
                     .where(
                         EndUser.session_id == user_id,
                         EndUser.tenant_id == tenant_id,
                     )
-                    .limit(1)
+                    .first()
                 )
             else:
-                user_model = session.get(EndUser, user_id)
+                user_model = (
+                    session.query(EndUser)
+                    .where(
+                        EndUser.id == user_id,
+                        EndUser.tenant_id == tenant_id,
+                    )
+                    .first()
+                )
 
             if not user_model:
                 user_model = EndUser(
@@ -79,7 +85,16 @@ def get_user_tenant(view_func: Callable[P, R]):
         if not user_id:
             user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
 
-        tenant_model = db.session.get(Tenant, tenant_id)
+        try:
+            tenant_model = (
+                db.session.query(Tenant)
+                .where(
+                    Tenant.id == tenant_id,
+                )
+                .first()
+            )
+        except Exception:
+            raise ValueError("tenant not found")
 
         if not tenant_model:
             raise ValueError("tenant not found")

api/controllers/inner_api/workspace/workspace.py

@@ -2,7 +2,6 @@ import json
 
 from flask_restx import Resource
 from pydantic import BaseModel
-from sqlalchemy import select
 
 from controllers.common.schema import register_schema_models
 from controllers.console.wraps import setup_required
@@ -43,7 +42,7 @@ class EnterpriseWorkspace(Resource):
     def post(self):
         args = WorkspaceCreatePayload.model_validate(inner_api_ns.payload or {})
 
-        account = db.session.scalar(select(Account).where(Account.email == args.owner_email).limit(1))
+        account = db.session.query(Account).filter_by(email=args.owner_email).first()
         if account is None:
             return {"message": "owner account not found."}, 404
 

api/controllers/inner_api/wraps.py

@@ -75,7 +75,7 @@ def enterprise_inner_api_user_auth(view: Callable[P, R]):
         if signature_base64 != token:
             return view(*args, **kwargs)
 
-        kwargs["user"] = db.session.get(EndUser, user_id)
+        kwargs["user"] = db.session.query(EndUser).where(EndUser.id == user_id).first()
 
         return view(*args, **kwargs)
 

api/controllers/service_api/app/message.py

@@ -15,7 +15,6 @@ from core.app.entities.app_invoke_entities import InvokeFrom
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem
 from libs.helper import UUIDStrOrEmpty
-from models.enums import FeedbackRating
 from models.model import App, AppMode, EndUser
 from services.errors.message import (
     FirstMessageNotExistsError,
@@ -117,7 +116,7 @@ class MessageFeedbackApi(Resource):
                 app_model=app_model,
                 message_id=message_id,
                 user=end_user,
-                rating=FeedbackRating(payload.rating) if payload.rating else None,
+                rating=payload.rating,
                 content=payload.content,
             )
         except MessageNotExistsError:

api/controllers/service_api/dataset/dataset.py

@@ -15,7 +15,6 @@ from controllers.service_api.wraps import (
     cloud_edition_billing_rate_limit_check,
 )
 from core.provider_manager import ProviderManager
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from dify_graph.model_runtime.entities.model_entities import ModelType
 from fields.dataset_fields import dataset_detail_fields
 from fields.tag_fields import DataSetTag
@@ -154,20 +153,15 @@ class DatasetListApi(DatasetApiResource):
 
         data = marshal(datasets, dataset_detail_fields)
         for item in data:
-            if (
-                item["indexing_technique"] == IndexTechniqueType.HIGH_QUALITY  # pyrefly: ignore[bad-index]
-                and item["embedding_model_provider"]  # pyrefly: ignore[bad-index]
-            ):
-                item["embedding_model_provider"] = str(  # pyrefly: ignore[unsupported-operation]
-                    ModelProviderID(item["embedding_model_provider"])  # pyrefly: ignore[bad-index]
-                )
-                item_model = f"{item['embedding_model']}:{item['embedding_model_provider']}"  # pyrefly: ignore[bad-index]
+            if item["indexing_technique"] == "high_quality" and item["embedding_model_provider"]:
+                item["embedding_model_provider"] = str(ModelProviderID(item["embedding_model_provider"]))
+                item_model = f"{item['embedding_model']}:{item['embedding_model_provider']}"
                 if item_model in model_names:
-                    item["embedding_available"] = True  # type: ignore
+                    item["embedding_available"] = True
                 else:
-                    item["embedding_available"] = False  # type: ignore
+                    item["embedding_available"] = False
             else:
-                item["embedding_available"] = True  # type: ignore
+                item["embedding_available"] = True
         response = {
             "data": data,
             "has_more": len(datasets) == query.limit,
@@ -271,7 +265,7 @@ class DatasetApi(DatasetApiResource):
         for embedding_model in embedding_models:
             model_names.append(f"{embedding_model.model}:{embedding_model.provider.provider}")
 
-        if data.get("indexing_technique") == IndexTechniqueType.HIGH_QUALITY:
+        if data.get("indexing_technique") == "high_quality":
             item_model = f"{data.get('embedding_model')}:{data.get('embedding_model_provider')}"
             if item_model in model_names:
                 data["embedding_available"] = True
@@ -321,7 +315,7 @@ class DatasetApi(DatasetApiResource):
         # check embedding model setting
         embedding_model_provider = payload.embedding_model_provider
         embedding_model = payload.embedding_model
-        if payload.indexing_technique == IndexTechniqueType.HIGH_QUALITY or embedding_model_provider:
+        if payload.indexing_technique == "high_quality" or embedding_model_provider:
             if embedding_model_provider and embedding_model:
                 DatasetService.check_embedding_model_setting(
                     dataset.tenant_id, embedding_model_provider, embedding_model

api/controllers/service_api/dataset/document.py

@@ -36,7 +36,6 @@ from extensions.ext_database import db
 from fields.document_fields import document_fields, document_status_fields
 from libs.login import current_user
 from models.dataset import Dataset, Document, DocumentSegment
-from models.enums import SegmentStatus
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import (
     KnowledgeConfig,
@@ -623,15 +622,13 @@ class DocumentIndexingStatusApi(DatasetApiResource):
                 .where(
                     DocumentSegment.completed_at.isnot(None),
                     DocumentSegment.document_id == str(document.id),
-                    DocumentSegment.status != SegmentStatus.RE_SEGMENT,
+                    DocumentSegment.status != "re_segment",
                 )
                 .count()
             )
             total_segments = (
                 db.session.query(DocumentSegment)
-                .where(
-                    DocumentSegment.document_id == str(document.id), DocumentSegment.status != SegmentStatus.RE_SEGMENT
-                )
+                .where(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
                 .count()
             )
             # Create a dictionary with document attributes and additional fields

api/controllers/service_api/dataset/segment.py

@@ -17,7 +17,6 @@ from controllers.service_api.wraps import (
 )
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from dify_graph.model_runtime.entities.model_entities import ModelType
 from extensions.ext_database import db
 from fields.segment_fields import child_chunk_fields, segment_fields
@@ -104,7 +103,7 @@ class SegmentApi(DatasetApiResource):
         if not document.enabled:
             raise NotFound("Document is disabled.")
         # check embedding model setting
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             try:
                 model_manager = ModelManager()
                 model_manager.get_model_instance(
@@ -158,7 +157,7 @@ class SegmentApi(DatasetApiResource):
         if not document:
             raise NotFound("Document not found.")
         # check embedding model setting
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             try:
                 model_manager = ModelManager()
                 model_manager.get_model_instance(
@@ -263,7 +262,7 @@ class DatasetSegmentApi(DatasetApiResource):
         document = DocumentService.get_document(dataset_id, document_id)
         if not document:
             raise NotFound("Document not found.")
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             # check embedding model setting
             try:
                 model_manager = ModelManager()
@@ -359,7 +358,7 @@ class ChildChunkApi(DatasetApiResource):
             raise NotFound("Segment not found.")
 
         # check embedding model setting
-        if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
+        if dataset.indexing_technique == "high_quality":
             try:
                 model_manager = ModelManager()
                 model_manager.get_model_instance(

api/controllers/trigger/webhook.py

@@ -70,14 +70,7 @@ def handle_webhook(webhook_id: str):
 
 @bp.route("/webhook-debug/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
 def handle_webhook_debug(webhook_id: str):
-    """Handle webhook debug calls without triggering production workflow execution.
-
-    The debug webhook endpoint is only for draft inspection flows. It never enqueues
-    Celery work for the published workflow; instead it dispatches an in-memory debug
-    event to an active Variable Inspector listener. Returning a clear error when no
-    listener is registered prevents a misleading 200 response for requests that are
-    effectively dropped.
-    """
+    """Handle webhook debug calls without triggering production workflow execution."""
     try:
         webhook_trigger, _, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id, is_debug=True)
         if error:
@@ -101,32 +94,11 @@ def handle_webhook_debug(webhook_id: str):
                 "method": webhook_data.get("method"),
             },
         )
-        dispatch_count = TriggerDebugEventBus.dispatch(
+        TriggerDebugEventBus.dispatch(
             tenant_id=webhook_trigger.tenant_id,
             event=event,
             pool_key=pool_key,
         )
-        if dispatch_count == 0:
-            logger.warning(
-                "Webhook debug request dropped without an active listener for webhook %s (tenant=%s, app=%s, node=%s)",
-                webhook_trigger.webhook_id,
-                webhook_trigger.tenant_id,
-                webhook_trigger.app_id,
-                webhook_trigger.node_id,
-            )
-            return (
-                jsonify(
-                    {
-                        "error": "No active debug listener",
-                        "message": (
-                            "The webhook debug URL only works while the Variable Inspector is listening. "
-                            "Use the published webhook URL to execute the workflow in Celery."
-                        ),
-                        "execution_url": webhook_trigger.webhook_url,
-                    }
-                ),
-                409,
-            )
         response_data, status_code = WebhookService.generate_webhook_response(node_config)
         return jsonify(response_data), status_code
 

api/controllers/web/human_input_form.py

@@ -8,7 +8,6 @@ from datetime import datetime
 
 from flask import Response, request
 from flask_restx import Resource, reqparse
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
@@ -148,11 +147,11 @@ class HumanInputFormApi(Resource):
 
 def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
     """Resolve App/Site for the form's app and validate tenant status."""
-    app_model = db.session.get(App, form.app_id)
+    app_model = db.session.query(App).where(App.id == form.app_id).first()
     if app_model is None or app_model.tenant_id != form.tenant_id:
         raise NotFoundError("Form not found")
 
-    site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+    site = db.session.query(Site).where(Site.app_id == app_model.id).first()
     if site is None:
         raise Forbidden()
 

api/controllers/web/message.py

@@ -25,7 +25,6 @@ from fields.conversation_fields import ResultResponse
 from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfiniteScrollPagination, WebMessageListItem
 from libs import helper
 from libs.helper import uuid_value
-from models.enums import FeedbackRating
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
 from services.errors.app import MoreLikeThisDisabledError
@@ -158,7 +157,7 @@ class MessageFeedbackApi(WebApiResource):
                 app_model=app_model,
                 message_id=message_id,
                 user=end_user,
-                rating=FeedbackRating(payload.rating) if payload.rating else None,
+                rating=payload.rating,
                 content=payload.content,
             )
         except MessageNotExistsError:

api/controllers/web/site.py

@@ -1,7 +1,6 @@
 from typing import cast
 
 from flask_restx import fields, marshal, marshal_with
-from sqlalchemy import select
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
@@ -73,7 +72,7 @@ class AppSiteApi(WebApiResource):
     def get(self, app_model, end_user):
         """Retrieve app site info."""
         # get site
-        site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
 
         if not site:
             raise Forbidden()

api/core/agent/base_agent_runner.py

@@ -441,7 +441,7 @@ class BaseAgentRunner(AppRunner):
                 continue
 
             result.append(self.organize_agent_user_prompt(message))
-            agent_thoughts = message.agent_thoughts
+            agent_thoughts: list[MessageAgentThought] = message.agent_thoughts
             if agent_thoughts:
                 for agent_thought in agent_thoughts:
                     tool_names_raw = agent_thought.tool

api/core/app/app_config/common/parameters_mapping/__init__.py

@@ -1,36 +1,13 @@
 from collections.abc import Mapping
-from typing import Any, TypedDict
+from typing import Any
 
 from configs import dify_config
 from constants import DEFAULT_FILE_NUMBER_LIMITS
 
 
-class SystemParametersDict(TypedDict):
-    image_file_size_limit: int
-    video_file_size_limit: int
-    audio_file_size_limit: int
-    file_size_limit: int
-    workflow_file_upload_limit: int
-
-
-class AppParametersDict(TypedDict):
-    opening_statement: str | None
-    suggested_questions: list[str]
-    suggested_questions_after_answer: dict[str, Any]
-    speech_to_text: dict[str, Any]
-    text_to_speech: dict[str, Any]
-    retriever_resource: dict[str, Any]
-    annotation_reply: dict[str, Any]
-    more_like_this: dict[str, Any]
-    user_input_form: list[dict[str, Any]]
-    sensitive_word_avoidance: dict[str, Any]
-    file_upload: dict[str, Any]
-    system_parameters: SystemParametersDict
-
-
 def get_parameters_from_feature_dict(
     *, features_dict: Mapping[str, Any], user_input_form: list[dict[str, Any]]
-) -> AppParametersDict:
+) -> Mapping[str, Any]:
     """
     Mapping from feature dict to webapp parameters
     """

api/core/app/app_config/easy_ui_based_app/dataset/manager.py

@@ -8,7 +8,6 @@ from core.app.app_config.entities import (
     ModelConfig,
 )
 from core.entities.agent_entities import PlanningStrategy
-from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
 from models.model import AppMode, AppModelConfigDict
 from services.dataset_service import DatasetService
 
@@ -118,10 +117,8 @@ class DatasetConfigManager:
                     score_threshold=float(score_threshold_val)
                     if dataset_configs.get("score_threshold_enabled", False) and score_threshold_val is not None
                     else None,
-                    reranking_model=cast(RerankingModelDict, reranking_model_val)
-                    if isinstance(reranking_model_val, dict)
-                    else None,
-                    weights=cast(WeightsDict, weights_val) if isinstance(weights_val, dict) else None,
+                    reranking_model=reranking_model_val if isinstance(reranking_model_val, dict) else None,
+                    weights=weights_val if isinstance(weights_val, dict) else None,
                     reranking_enabled=bool(dataset_configs.get("reranking_enabled", True)),
                     rerank_mode=dataset_configs.get("reranking_mode", "reranking_model"),
                     metadata_filtering_mode=cast(

api/core/app/app_config/entities.py

@@ -4,7 +4,6 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, Field
 
-from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
 from dify_graph.file import FileUploadConfig
 from dify_graph.model_runtime.entities.llm_entities import LLMMode
 from dify_graph.model_runtime.entities.message_entities import PromptMessageRole
@@ -195,8 +194,8 @@ class DatasetRetrieveConfigEntity(BaseModel):
     top_k: int | None = None
     score_threshold: float | None = 0.0
     rerank_mode: str | None = "reranking_model"
-    reranking_model: RerankingModelDict | None = None
-    weights: WeightsDict | None = None
+    reranking_model: dict | None = None
+    weights: dict | None = None
     reranking_enabled: bool | None = True
     metadata_filtering_mode: Literal["disabled", "automatic", "manual"] | None = "disabled"
     metadata_model_config: ModelConfig | None = None

api/core/app/apps/advanced_chat/app_generator.py

@@ -5,7 +5,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Literal, Union, overload
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -47,6 +47,7 @@ from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.base import Base
 from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
 from services.workflow_draft_variable_service import (
@@ -329,10 +330,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
 
         return self._generate(
             workflow=workflow,
@@ -413,10 +413,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
 
         return self._generate(
             workflow=workflow,
@@ -521,10 +520,8 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
 
         # release database connection, because the following new thread operations may take a long time
         with Session(bind=db.engine, expire_on_commit=False) as session:
-            workflow = _refresh_model(session=session, model=workflow)
-            message = _refresh_model(session=session, model=message)
-        if message is None:
-            raise RuntimeError("Failed to refresh Message; _refresh_model returned None.")
+            workflow = _refresh_model(session, workflow)
+            message = _refresh_model(session, message)
         #     workflow_ = session.get(Workflow, workflow.id)
         #     assert workflow_ is not None
         #     workflow = workflow_
@@ -691,21 +688,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                 raise e
 
 
-@overload
-def _refresh_model(*, session: Session | None = None, model: Workflow) -> Workflow: ...
+_T = TypeVar("_T", bound=Base)
 
 
-@overload
-def _refresh_model(*, session: Session | None = None, model: Message) -> Message: ...
-
-
-def _refresh_model(*, session: Session | None = None, model: Any) -> Any:
-    if session is not None:
-        detached_model = session.get(type(model), model.id)
-        assert detached_model is not None
-        return detached_model
-
-    with Session(bind=db.engine, expire_on_commit=False) as refresh_session:
-        detached_model = refresh_session.get(type(model), model.id)
-        assert detached_model is not None
-        return detached_model
+def _refresh_model(session, model: _T) -> _T:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        detach_model = session.get(type(model), model.id)
+        assert detach_model is not None
+        return detach_model

api/core/app/apps/advanced_chat/generate_response_converter.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import Any, cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -56,8 +56,8 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
-    ) -> Generator[dict | str, None, None]:
+        cls, stream_response: Generator[AppStreamResponse, None, None]
+    ) -> Generator[dict | str, Any, None]:
         """
         Convert stream full response.
         :param stream_response: stream response
@@ -87,8 +87,8 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
-    ) -> Generator[dict | str, None, None]:
+        cls, stream_response: Generator[AppStreamResponse, None, None]
+    ) -> Generator[dict | str, Any, None]:
         """
         Convert stream simple response.
         :param stream_response: stream response

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -76,7 +76,7 @@ from dify_graph.system_variable import SystemVariable
 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
 from models import Account, Conversation, EndUser, Message, MessageFile
-from models.enums import CreatorUserRole, MessageFileBelongsTo, MessageStatus
+from models.enums import CreatorUserRole, MessageStatus
 from models.execution_extra_content import HumanInputContent
 from models.workflow import Workflow
 
@@ -939,7 +939,7 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 type=file["type"],
                 transfer_method=file["transfer_method"],
                 url=file["remote_url"],
-                belongs_to=MessageFileBelongsTo.ASSISTANT,
+                belongs_to="assistant",
                 upload_file_id=file["related_id"],
                 created_by_role=CreatorUserRole.ACCOUNT
                 if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}

api/core/app/apps/agent_chat/generate_response_converter.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -55,7 +55,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream full response.
@@ -86,7 +86,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream simple response.

api/core/app/apps/base_app_generate_response_converter.py

@@ -1,7 +1,7 @@
 import logging
 from abc import ABC, abstractmethod
-from collections.abc import Generator, Iterator, Mapping
-from typing import Any
+from collections.abc import Generator, Mapping
+from typing import Any, Union
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
@@ -16,26 +16,24 @@ class AppGenerateResponseConverter(ABC):
 
     @classmethod
     def convert(
-        cls, response: AppBlockingResponse | Iterator[AppStreamResponse], invoke_from: InvokeFrom
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+        cls, response: Union[AppBlockingResponse, Generator[AppStreamResponse, Any, None]], invoke_from: InvokeFrom
+    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         if invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.SERVICE_API}:
             if isinstance(response, AppBlockingResponse):
                 return cls.convert_blocking_full_response(response)
             else:
-                stream_response = response
 
-                def _generate_full_response() -> Generator[dict[str, Any] | str, None, None]:
-                    yield from cls.convert_stream_full_response(stream_response)
+                def _generate_full_response() -> Generator[dict | str, Any, None]:
+                    yield from cls.convert_stream_full_response(response)
 
                 return _generate_full_response()
         else:
             if isinstance(response, AppBlockingResponse):
                 return cls.convert_blocking_simple_response(response)
             else:
-                stream_response = response
 
-                def _generate_simple_response() -> Generator[dict[str, Any] | str, None, None]:
-                    yield from cls.convert_stream_simple_response(stream_response)
+                def _generate_simple_response() -> Generator[dict | str, Any, None]:
+                    yield from cls.convert_stream_simple_response(response)
 
                 return _generate_simple_response()
 
@@ -52,14 +50,14 @@ class AppGenerateResponseConverter(ABC):
     @classmethod
     @abstractmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         raise NotImplementedError
 
     @classmethod
     @abstractmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         raise NotImplementedError
 
@@ -76,22 +74,11 @@ class AppGenerateResponseConverter(ABC):
             for resource in metadata["retriever_resources"]:
                 updated_resources.append(
                     {
-                        "dataset_id": resource.get("dataset_id"),
-                        "dataset_name": resource.get("dataset_name"),
-                        "document_id": resource.get("document_id"),
                         "segment_id": resource.get("segment_id", ""),
                         "position": resource["position"],
-                        "data_source_type": resource.get("data_source_type"),
                         "document_name": resource["document_name"],
                         "score": resource["score"],
-                        "hit_count": resource.get("hit_count"),
-                        "word_count": resource.get("word_count"),
-                        "segment_position": resource.get("segment_position"),
-                        "index_node_hash": resource.get("index_node_hash"),
                         "content": resource["content"],
-                        "page": resource.get("page"),
-                        "title": resource.get("title"),
-                        "files": resource.get("files"),
                         "summary": resource.get("summary"),
                     }
                 )

api/core/app/apps/base_app_generator.py

@@ -224,7 +224,6 @@ class BaseAppGenerator:
     def _get_draft_var_saver_factory(invoke_from: InvokeFrom, account: Account | EndUser) -> DraftVariableSaverFactory:
         if invoke_from == InvokeFrom.DEBUGGER:
             assert isinstance(account, Account)
-            debug_account = account
 
             def draft_var_saver_factory(
                 session: Session,
@@ -241,7 +240,7 @@ class BaseAppGenerator:
                     node_type=node_type,
                     node_execution_id=node_execution_id,
                     enclosing_node_id=enclosing_node_id,
-                    user=debug_account,
+                    user=account,
                 )
         else:
 

api/core/app/apps/base_app_runner.py

@@ -40,7 +40,7 @@ from dify_graph.model_runtime.entities.message_entities import (
 from dify_graph.model_runtime.entities.model_entities import ModelPropertyKey
 from dify_graph.model_runtime.errors.invoke import InvokeBadRequestError
 from extensions.ext_database import db
-from models.enums import CreatorUserRole, MessageFileBelongsTo
+from models.enums import CreatorUserRole
 from models.model import App, AppMode, Message, MessageAnnotation, MessageFile
 
 if TYPE_CHECKING:
@@ -419,7 +419,7 @@ class AppRunner:
             message_id=message_id,
             type=FileType.IMAGE,
             transfer_method=FileTransferMethod.TOOL_FILE,
-            belongs_to=MessageFileBelongsTo.ASSISTANT,
+            belongs_to="assistant",
             url=f"/files/tools/{tool_file.id}",
             upload_file_id=tool_file.id,
             created_by_role=(

api/core/app/apps/chat/app_generator.py

@@ -166,19 +166,15 @@ class ChatAppGenerator(MessageBasedAppGenerator):
 
         # init generate records
         (conversation, message) = self._init_generate_records(application_generate_entity, conversation)
-        if conversation is None or message is None:
-            raise RuntimeError("_init_generate_records() returned None for conversation or message")
-        generated_conversation_id = str(conversation.id)
-        generated_message_id = str(message.id)
 
         # init queue manager
         queue_manager = MessageBasedAppQueueManager(
             task_id=application_generate_entity.task_id,
             user_id=application_generate_entity.user_id,
             invoke_from=application_generate_entity.invoke_from,
-            conversation_id=generated_conversation_id,
+            conversation_id=conversation.id,
             app_mode=conversation.mode,
-            message_id=generated_message_id,
+            message_id=message.id,
         )
 
         # new thread with request context
@@ -188,8 +184,8 @@ class ChatAppGenerator(MessageBasedAppGenerator):
                 flask_app=current_app._get_current_object(),  # type: ignore
                 application_generate_entity=application_generate_entity,
                 queue_manager=queue_manager,
-                conversation_id=generated_conversation_id,
-                message_id=generated_message_id,
+                conversation_id=conversation.id,
+                message_id=message.id,
             )
 
         worker_thread = threading.Thread(target=worker_with_context)

api/core/app/apps/chat/generate_response_converter.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -55,7 +55,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream full response.
@@ -86,7 +86,7 @@ class ChatAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream simple response.

api/core/app/apps/common/workflow_response_converter.py

@@ -3,7 +3,7 @@ import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, NewType, TypedDict, Union
+from typing import Any, NewType, Union
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -76,20 +76,6 @@ NodeExecutionId = NewType("NodeExecutionId", str)
 logger = logging.getLogger(__name__)
 
 
-class AccountCreatedByDict(TypedDict):
-    id: str
-    name: str
-    email: str
-
-
-class EndUserCreatedByDict(TypedDict):
-    id: str
-    user: str
-
-
-CreatedByDict = AccountCreatedByDict | EndUserCreatedByDict
-
-
 @dataclass(slots=True)
 class _NodeSnapshot:
     """In-memory cache for node metadata between start and completion events."""
@@ -263,19 +249,19 @@ class WorkflowResponseConverter:
         outputs_mapping = graph_runtime_state.outputs or {}
         encoded_outputs = WorkflowRuntimeTypeConverter().to_json_encodable(outputs_mapping)
 
-        created_by: CreatedByDict | dict[str, object] = {}
+        created_by: Mapping[str, object] | None
         user = self._user
         if isinstance(user, Account):
-            created_by = AccountCreatedByDict(
-                id=user.id,
-                name=user.name,
-                email=user.email,
-            )
-        elif isinstance(user, EndUser):
-            created_by = EndUserCreatedByDict(
-                id=user.id,
-                user=user.session_id,
-            )
+            created_by = {
+                "id": user.id,
+                "name": user.name,
+                "email": user.email,
+            }
+        else:
+            created_by = {
+                "id": user.id,
+                "user": user.session_id,
+            }
 
         return WorkflowFinishStreamResponse(
             task_id=task_id,
@@ -517,7 +503,7 @@ class WorkflowResponseConverter:
         snapshot = self._pop_snapshot(event.node_execution_id)
 
         start_at = snapshot.start_at if snapshot else event.start_at
-        finished_at = event.finished_at or naive_utc_now()
+        finished_at = naive_utc_now()
         elapsed_time = (finished_at - start_at).total_seconds()
 
         inputs, inputs_truncated = self._truncate_mapping(event.inputs)

api/core/app/apps/completion/app_generator.py

@@ -149,11 +149,6 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
 
         # init generate records
         (conversation, message) = self._init_generate_records(application_generate_entity)
-        if conversation is None or message is None:
-            raise RuntimeError(
-                "_init_generate_records() returned None for conversation or message, "
-                "which is required to proceed with generation."
-            )
 
         # init queue manager
         queue_manager = MessageBasedAppQueueManager(
@@ -317,19 +312,15 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
 
         # init generate records
         (conversation, message) = self._init_generate_records(application_generate_entity)
-        assert conversation is not None
-        assert message is not None
-        conversation_id = str(conversation.id)
-        message_id = str(message.id)
 
         # init queue manager
         queue_manager = MessageBasedAppQueueManager(
             task_id=application_generate_entity.task_id,
             user_id=application_generate_entity.user_id,
             invoke_from=application_generate_entity.invoke_from,
-            conversation_id=conversation_id,
+            conversation_id=conversation.id,
             app_mode=conversation.mode,
-            message_id=message_id,
+            message_id=message.id,
         )
 
         # new thread with request context
@@ -339,7 +330,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
                 flask_app=current_app._get_current_object(),  # type: ignore
                 application_generate_entity=application_generate_entity,
                 queue_manager=queue_manager,
-                message_id=message_id,
+                message_id=message.id,
             )
 
         worker_thread = threading.Thread(target=worker_with_context)

api/core/app/apps/completion/generate_response_converter.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -54,7 +54,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream full response.
@@ -84,7 +84,7 @@ class CompletionAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream simple response.

api/core/app/apps/message_based_app_generator.py

@@ -33,7 +33,7 @@ from extensions.ext_redis import get_pubsub_broadcast_channel
 from libs.broadcast_channel.channel import Topic
 from libs.datetime_utils import naive_utc_now
 from models import Account
-from models.enums import ConversationFromSource, CreatorUserRole, MessageFileBelongsTo
+from models.enums import CreatorUserRole
 from models.model import App, AppMode, AppModelConfig, Conversation, EndUser, Message, MessageFile
 from services.errors.app_model_config import AppModelConfigBrokenError
 from services.errors.conversation import ConversationNotExistsError
@@ -130,10 +130,10 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         end_user_id = None
         account_id = None
         if application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
-            from_source = ConversationFromSource.API
+            from_source = "api"
             end_user_id = application_generate_entity.user_id
         else:
-            from_source = ConversationFromSource.CONSOLE
+            from_source = "console"
             account_id = application_generate_entity.user_id
 
         if isinstance(application_generate_entity, AdvancedChatAppGenerateEntity):
@@ -225,7 +225,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
                     message_id=message.id,
                     type=file.type,
                     transfer_method=file.transfer_method,
-                    belongs_to=MessageFileBelongsTo.USER,
+                    belongs_to="user",
                     url=file.remote_url,
                     upload_file_id=file.related_id,
                     created_by_role=(CreatorUserRole.ACCOUNT if account_id else CreatorUserRole.END_USER),

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

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -36,7 +36,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream full response.
@@ -65,7 +65,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream simple response.

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

@@ -419,12 +419,11 @@ class PipelineGenerator(BaseAppGenerator):
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
         var_loader = DraftVarLoader(
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
 
         return self._generate(
@@ -515,12 +514,11 @@ class PipelineGenerator(BaseAppGenerator):
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
         var_loader = DraftVarLoader(
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
 
         return self._generate(

api/core/app/apps/workflow/app_generator.py

@@ -414,12 +414,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
         var_loader = DraftVarLoader(
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
 
         return self._generate(
@@ -498,12 +497,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
             triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
         )
         draft_var_srv = WorkflowDraftVariableService(db.session())
-        draft_var_srv.prefill_conversation_variable_default_values(workflow, user_id=user.id)
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
         var_loader = DraftVarLoader(
             engine=db.engine,
             app_id=application_generate_entity.app_config.app_id,
             tenant_id=application_generate_entity.app_config.tenant_id,
-            user_id=user.id,
         )
         return self._generate(
             app_model=app_model,

api/core/app/apps/workflow/generate_response_converter.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Generator
 from typing import cast
 
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
@@ -36,7 +36,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_full_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream full response.
@@ -65,7 +65,7 @@ class WorkflowAppGenerateResponseConverter(AppGenerateResponseConverter):
 
     @classmethod
     def convert_stream_simple_response(
-        cls, stream_response: Iterator[AppStreamResponse]
+        cls, stream_response: Generator[AppStreamResponse, None, None]
     ) -> Generator[dict | str, None, None]:
         """
         Convert stream simple response.

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -705,7 +705,7 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
             app_id=self._application_generate_entity.app_config.app_id,
             workflow_id=self._workflow.id,
             workflow_run_id=workflow_run_id,
-            created_from=created_from,
+            created_from=created_from.value,
             created_by_role=self._created_by_role,
             created_by=self._user_id,
         )

api/core/app/apps/workflow_app_runner.py

@@ -1,17 +1,13 @@
 import logging
 import time
 from collections.abc import Mapping, Sequence
-from typing import Protocol, TypeAlias
+from typing import Any, cast
 
 from pydantic import ValidationError
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.agent_strategy import AgentStrategyInfo
-from core.app.entities.app_invoke_entities import (
-    InvokeFrom,
-    UserFrom,
-    build_dify_run_context,
-)
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, build_dify_run_context
 from core.app.entities.queue_entities import (
     AppQueueEvent,
     QueueAgentLogEvent,
@@ -40,7 +36,7 @@ from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id, resolve_workflow_node_class
 from core.workflow.workflow_entry import WorkflowEntry
 from dify_graph.entities import GraphInitParams
-from dify_graph.entities.graph_config import NodeConfigDict, NodeConfigDictAdapter
+from dify_graph.entities.graph_config import NodeConfigDictAdapter
 from dify_graph.entities.pause_reason import HumanInputRequired
 from dify_graph.graph import Graph
 from dify_graph.graph_engine.layers.base import GraphEngineLayer
@@ -79,14 +75,6 @@ from tasks.mail_human_input_delivery_task import dispatch_human_input_email_task
 
 logger = logging.getLogger(__name__)
 
-GraphConfigObject: TypeAlias = dict[str, object]
-GraphConfigMapping: TypeAlias = Mapping[str, object]
-
-
-class SingleNodeRunEntity(Protocol):
-    node_id: str
-    inputs: Mapping[str, object]
-
 
 class WorkflowBasedAppRunner:
     def __init__(
@@ -110,7 +98,7 @@ class WorkflowBasedAppRunner:
 
     def _init_graph(
         self,
-        graph_config: GraphConfigMapping,
+        graph_config: Mapping[str, Any],
         graph_runtime_state: GraphRuntimeState,
         user_from: UserFrom,
         invoke_from: InvokeFrom,
@@ -166,8 +154,8 @@ class WorkflowBasedAppRunner:
     def _prepare_single_node_execution(
         self,
         workflow: Workflow,
-        single_iteration_run: SingleNodeRunEntity | None = None,
-        single_loop_run: SingleNodeRunEntity | None = None,
+        single_iteration_run: Any | None = None,
+        single_loop_run: Any | None = None,
     ) -> tuple[Graph, VariablePool, GraphRuntimeState]:
         """
         Prepare graph, variable pool, and runtime state for single node execution
@@ -220,88 +208,11 @@ class WorkflowBasedAppRunner:
         # This ensures all nodes in the graph reference the same GraphRuntimeState instance
         return graph, variable_pool, graph_runtime_state
 
-    @staticmethod
-    def _get_graph_items(graph_config: GraphConfigMapping) -> tuple[list[GraphConfigMapping], list[GraphConfigMapping]]:
-        nodes = graph_config.get("nodes")
-        edges = graph_config.get("edges")
-        if not isinstance(nodes, list):
-            raise ValueError("nodes in workflow graph must be a list")
-        if not isinstance(edges, list):
-            raise ValueError("edges in workflow graph must be a list")
-
-        validated_nodes: list[GraphConfigMapping] = []
-        for node in nodes:
-            if not isinstance(node, Mapping):
-                raise ValueError("nodes in workflow graph must be mappings")
-            validated_nodes.append(node)
-
-        validated_edges: list[GraphConfigMapping] = []
-        for edge in edges:
-            if not isinstance(edge, Mapping):
-                raise ValueError("edges in workflow graph must be mappings")
-            validated_edges.append(edge)
-
-        return validated_nodes, validated_edges
-
-    @staticmethod
-    def _extract_start_node_id(node_config: GraphConfigMapping | None) -> str | None:
-        if node_config is None:
-            return None
-        node_data = node_config.get("data")
-        if not isinstance(node_data, Mapping):
-            return None
-        start_node_id = node_data.get("start_node_id")
-        return start_node_id if isinstance(start_node_id, str) else None
-
-    @classmethod
-    def _build_single_node_graph_config(
-        cls,
-        *,
-        graph_config: GraphConfigMapping,
-        node_id: str,
-        node_type_filter_key: str,
-    ) -> tuple[GraphConfigObject, NodeConfigDict]:
-        node_configs, edge_configs = cls._get_graph_items(graph_config)
-        main_node_config = next((node for node in node_configs if node.get("id") == node_id), None)
-        start_node_id = cls._extract_start_node_id(main_node_config)
-
-        filtered_node_configs = [
-            dict(node)
-            for node in node_configs
-            if node.get("id") == node_id
-            or (isinstance(node_data := node.get("data"), Mapping) and node_data.get(node_type_filter_key) == node_id)
-            or (start_node_id and node.get("id") == start_node_id)
-        ]
-        if not filtered_node_configs:
-            raise ValueError(f"node id {node_id} not found in workflow graph")
-
-        filtered_node_ids = {
-            str(node_id_value) for node in filtered_node_configs if isinstance((node_id_value := node.get("id")), str)
-        }
-        filtered_edge_configs = [
-            dict(edge)
-            for edge in edge_configs
-            if (edge.get("source") is None or edge.get("source") in filtered_node_ids)
-            and (edge.get("target") is None or edge.get("target") in filtered_node_ids)
-        ]
-
-        target_node_config = next((node for node in filtered_node_configs if node.get("id") == node_id), None)
-        if target_node_config is None:
-            raise ValueError(f"node id {node_id} not found in workflow graph")
-
-        return (
-            {
-                "nodes": filtered_node_configs,
-                "edges": filtered_edge_configs,
-            },
-            NodeConfigDictAdapter.validate_python(target_node_config),
-        )
-
     def _get_graph_and_variable_pool_for_single_node_run(
         self,
         workflow: Workflow,
         node_id: str,
-        user_inputs: Mapping[str, object],
+        user_inputs: dict[str, Any],
         graph_runtime_state: GraphRuntimeState,
         node_type_filter_key: str,  # 'iteration_id' or 'loop_id'
         node_type_label: str = "node",  # 'iteration' or 'loop' for error messages
@@ -325,14 +236,41 @@ class WorkflowBasedAppRunner:
         if not graph_config:
             raise ValueError("workflow graph not found")
 
+        graph_config = cast(dict[str, Any], graph_config)
+
         if "nodes" not in graph_config or "edges" not in graph_config:
             raise ValueError("nodes or edges not found in workflow graph")
 
-        graph_config, target_node_config = self._build_single_node_graph_config(
-            graph_config=graph_config,
-            node_id=node_id,
-            node_type_filter_key=node_type_filter_key,
-        )
+        if not isinstance(graph_config.get("nodes"), list):
+            raise ValueError("nodes in workflow graph must be a list")
+
+        if not isinstance(graph_config.get("edges"), list):
+            raise ValueError("edges in workflow graph must be a list")
+
+        # filter nodes only in the specified node type (iteration or loop)
+        main_node_config = next((n for n in graph_config.get("nodes", []) if n.get("id") == node_id), None)
+        start_node_id = main_node_config.get("data", {}).get("start_node_id") if main_node_config else None
+        node_configs = [
+            node
+            for node in graph_config.get("nodes", [])
+            if node.get("id") == node_id
+            or node.get("data", {}).get(node_type_filter_key, "") == node_id
+            or (start_node_id and node.get("id") == start_node_id)
+        ]
+
+        graph_config["nodes"] = node_configs
+
+        node_ids = [node.get("id") for node in node_configs]
+
+        # filter edges only in the specified node type
+        edge_configs = [
+            edge
+            for edge in graph_config.get("edges", [])
+            if (edge.get("source") is None or edge.get("source") in node_ids)
+            and (edge.get("target") is None or edge.get("target") in node_ids)
+        ]
+
+        graph_config["edges"] = edge_configs
 
         # Create required parameters for Graph.init
         graph_init_params = GraphInitParams(
@@ -361,6 +299,18 @@ class WorkflowBasedAppRunner:
         if not graph:
             raise ValueError("graph not found in workflow")
 
+        # fetch node config from node id
+        target_node_config = None
+        for node in node_configs:
+            if node.get("id") == node_id:
+                target_node_config = node
+                break
+
+        if not target_node_config:
+            raise ValueError(f"{node_type_label} node id not found in workflow graph")
+
+        target_node_config = NodeConfigDictAdapter.validate_python(target_node_config)
+
         # Get node class
         node_type = target_node_config["data"].type
         node_version = str(target_node_config["data"].version)
@@ -506,7 +456,6 @@ class WorkflowBasedAppRunner:
                     node_id=event.node_id,
                     node_type=event.node_type,
                     start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=inputs,
                     process_data=process_data,
                     outputs=outputs,
@@ -522,7 +471,6 @@ class WorkflowBasedAppRunner:
                     node_id=event.node_id,
                     node_type=event.node_type,
                     start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
                     process_data=event.node_run_result.process_data,
                     outputs=event.node_run_result.outputs,
@@ -539,7 +487,6 @@ class WorkflowBasedAppRunner:
                     node_id=event.node_id,
                     node_type=event.node_type,
                     start_at=event.start_at,
-                    finished_at=event.finished_at,
                     inputs=event.node_run_result.inputs,
                     process_data=event.node_run_result.process_data,
                     outputs=event.node_run_result.outputs,

api/core/app/entities/app_invoke_entities.py

@@ -213,7 +213,7 @@ class AdvancedChatAppGenerateEntity(ConversationAppGenerateEntity):
         """
 
         node_id: str
-        inputs: Mapping[str, object]
+        inputs: Mapping
 
     single_iteration_run: SingleIterationRunEntity | None = None
 
@@ -223,7 +223,7 @@ class AdvancedChatAppGenerateEntity(ConversationAppGenerateEntity):
         """
 
         node_id: str
-        inputs: Mapping[str, object]
+        inputs: Mapping
 
     single_loop_run: SingleLoopRunEntity | None = None
 
@@ -243,7 +243,7 @@ class WorkflowAppGenerateEntity(AppGenerateEntity):
         """
 
         node_id: str
-        inputs: Mapping[str, object]
+        inputs: dict
 
     single_iteration_run: SingleIterationRunEntity | None = None
 
@@ -253,7 +253,7 @@ class WorkflowAppGenerateEntity(AppGenerateEntity):
         """
 
         node_id: str
-        inputs: Mapping[str, object]
+        inputs: dict
 
     single_loop_run: SingleLoopRunEntity | None = None
 

api/core/app/entities/queue_entities.py

@@ -335,7 +335,6 @@ class QueueNodeSucceededEvent(AppQueueEvent):
     in_loop_id: str | None = None
     """loop id if node is in loop"""
     start_at: datetime
-    finished_at: datetime | None = None
 
     inputs: Mapping[str, object] = Field(default_factory=dict)
     process_data: Mapping[str, object] = Field(default_factory=dict)
@@ -391,7 +390,6 @@ class QueueNodeExceptionEvent(AppQueueEvent):
     in_loop_id: str | None = None
     """loop id if node is in loop"""
     start_at: datetime
-    finished_at: datetime | None = None
 
     inputs: Mapping[str, object] = Field(default_factory=dict)
     process_data: Mapping[str, object] = Field(default_factory=dict)
@@ -416,7 +414,6 @@ class QueueNodeFailedEvent(AppQueueEvent):
     in_loop_id: str | None = None
     """loop id if node is in loop"""
     start_at: datetime
-    finished_at: datetime | None = None
 
     inputs: Mapping[str, object] = Field(default_factory=dict)
     process_data: Mapping[str, object] = Field(default_factory=dict)

api/core/app/features/annotation_reply/annotation_reply.py

@@ -4,10 +4,8 @@ from sqlalchemy import select
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.rag.datasource.vdb.vector_factory import Vector
-from core.rag.index_processor.constant.index_type import IndexTechniqueType
 from extensions.ext_database import db
 from models.dataset import Dataset
-from models.enums import CollectionBindingType, ConversationFromSource
 from models.model import App, AppAnnotationSetting, Message, MessageAnnotation
 from services.annotation_service import AppAnnotationService
 from services.dataset_service import DatasetCollectionBindingService
@@ -45,13 +43,13 @@ class AnnotationReplyFeature:
             embedding_model_name = collection_binding_detail.model_name
 
             dataset_collection_binding = DatasetCollectionBindingService.get_dataset_collection_binding(
-                embedding_provider_name, embedding_model_name, CollectionBindingType.ANNOTATION
+                embedding_provider_name, embedding_model_name, "annotation"
             )
 
             dataset = Dataset(
                 id=app_record.id,
                 tenant_id=app_record.tenant_id,
-                indexing_technique=IndexTechniqueType.HIGH_QUALITY,
+                indexing_technique="high_quality",
                 embedding_model_provider=embedding_provider_name,
                 embedding_model=embedding_model_name,
                 collection_binding_id=dataset_collection_binding.id,
@@ -69,9 +67,9 @@ class AnnotationReplyFeature:
                 annotation = AppAnnotationService.get_annotation_by_id(annotation_id)
                 if annotation:
                     if invoke_from in {InvokeFrom.SERVICE_API, InvokeFrom.WEB_APP}:
-                        from_source = ConversationFromSource.API
+                        from_source = "api"
                     else:
-                        from_source = ConversationFromSource.CONSOLE
+                        from_source = "console"
 
                     # insert annotation history
                     AppAnnotationService.add_annotation_history(

api/core/app/features/rate_limiting/rate_limit.py

@@ -19,7 +19,6 @@ class RateLimit:
     _REQUEST_MAX_ALIVE_TIME = 10 * 60  # 10 minutes
     _ACTIVE_REQUESTS_COUNT_FLUSH_INTERVAL = 5 * 60  # recalculate request_count from request_detail every 5 minutes
     _instance_dict: dict[str, "RateLimit"] = {}
-    max_active_requests: int
 
     def __new__(cls, client_id: str, max_active_requests: int):
         if client_id not in cls._instance_dict:
@@ -28,13 +27,7 @@ class RateLimit:
         return cls._instance_dict[client_id]
 
     def __init__(self, client_id: str, max_active_requests: int):
-        flush_cache = hasattr(self, "max_active_requests") and self.max_active_requests != max_active_requests
         self.max_active_requests = max_active_requests
-        # Only flush here if this instance has already been fully initialized,
-        # i.e. the Redis key attributes exist. Otherwise, rely on the flush at
-        # the end of initialization below.
-        if flush_cache and hasattr(self, "active_requests_key") and hasattr(self, "max_active_requests_key"):
-            self.flush_cache(use_local_value=True)
         # must be called after max_active_requests is set
         if self.disabled():
             return
@@ -48,6 +41,8 @@ class RateLimit:
         self.flush_cache(use_local_value=True)
 
     def flush_cache(self, use_local_value=False):
+        if self.disabled():
+            return
         self.last_recalculate_time = time.time()
         # flush max active requests
         if use_local_value or not redis_client.exists(self.max_active_requests_key):
@@ -55,8 +50,7 @@ class RateLimit:
         else:
             self.max_active_requests = int(redis_client.get(self.max_active_requests_key).decode("utf-8"))
             redis_client.expire(self.max_active_requests_key, timedelta(days=1))
-        if self.disabled():
-            return
+
         # flush max active requests (in-transit request list)
         if not redis_client.exists(self.active_requests_key):
             return

api/core/app/layers/suspend_layer.py

@@ -6,23 +6,16 @@ from dify_graph.graph_events.graph import GraphRunPausedEvent
 class SuspendLayer(GraphEngineLayer):
     """ """
 
-    def __init__(self) -> None:
-        super().__init__()
-        self._paused = False
-
     def on_graph_start(self):
-        self._paused = False
+        pass
 
     def on_event(self, event: GraphEngineEvent):
         """
         Handle the paused event, stash runtime state into storage and wait for resume.
         """
         if isinstance(event, GraphRunPausedEvent):
-            self._paused = True
+            pass
 
     def on_graph_end(self, error: Exception | None):
         """ """
-        self._paused = False
-
-    def is_paused(self) -> bool:
-        return self._paused
+        pass

api/core/app/task_pipeline/message_cycle_manager.py

@@ -34,7 +34,6 @@ from core.llm_generator.llm_generator import LLMGenerator
 from core.tools.signature import sign_tool_file
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from models.enums import MessageFileBelongsTo
 from models.model import AppMode, Conversation, MessageAnnotation, MessageFile
 from services.annotation_service import AppAnnotationService
 
@@ -234,7 +233,7 @@ class MessageCycleManager:
                 task_id=self._application_generate_entity.task_id,
                 id=message_file.id,
                 type=message_file.type,
-                belongs_to=message_file.belongs_to or MessageFileBelongsTo.USER,
+                belongs_to=message_file.belongs_to or "user",
                 url=url,
             )
 

api/core/app/task_pipeline/message_file_utils.py

@@ -1,5 +1,3 @@
-from typing import TypedDict
-
 from core.tools.signature import sign_tool_file
 from dify_graph.file import helpers as file_helpers
 from dify_graph.file.enums import FileTransferMethod
@@ -8,20 +6,7 @@ from models.model import MessageFile, UploadFile
 MAX_TOOL_FILE_EXTENSION_LENGTH = 10
 
 
-class MessageFileInfoDict(TypedDict):
-    related_id: str
-    extension: str
-    filename: str
-    size: int
-    mime_type: str
-    transfer_method: str
-    type: str
-    url: str
-    upload_file_id: str
-    remote_url: str | None
-
-
-def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, UploadFile]) -> MessageFileInfoDict:
+def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, UploadFile]) -> dict:
     """
     Prepare file dictionary for message end stream response.
 

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

@@ -128,14 +128,14 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             self._handle_graph_run_paused(event)
             return
 
-        if isinstance(event, NodeRunRetryEvent):
-            self._handle_node_retry(event)
-            return
-
         if isinstance(event, NodeRunStartedEvent):
             self._handle_node_started(event)
             return
 
+        if isinstance(event, NodeRunRetryEvent):
+            self._handle_node_retry(event)
+            return
+
         if isinstance(event, NodeRunSucceededEvent):
             self._handle_node_succeeded(event)
             return
@@ -268,12 +268,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
 
     def _handle_node_succeeded(self, event: NodeRunSucceededEvent) -> None:
         domain_execution = self._get_node_execution(event.id)
-        self._update_node_execution(
-            domain_execution,
-            event.node_run_result,
-            WorkflowNodeExecutionStatus.SUCCEEDED,
-            finished_at=event.finished_at,
-        )
+        self._update_node_execution(domain_execution, event.node_run_result, WorkflowNodeExecutionStatus.SUCCEEDED)
 
     def _handle_node_failed(self, event: NodeRunFailedEvent) -> None:
         domain_execution = self._get_node_execution(event.id)
@@ -282,7 +277,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             event.node_run_result,
             WorkflowNodeExecutionStatus.FAILED,
             error=event.error,
-            finished_at=event.finished_at,
         )
 
     def _handle_node_exception(self, event: NodeRunExceptionEvent) -> None:
@@ -292,7 +286,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
             event.node_run_result,
             WorkflowNodeExecutionStatus.EXCEPTION,
             error=event.error,
-            finished_at=event.finished_at,
         )
 
     def _handle_node_pause_requested(self, event: NodeRunPauseRequestedEvent) -> None:
@@ -359,14 +352,13 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         *,
         error: str | None = None,
         update_outputs: bool = True,
-        finished_at: datetime | None = None,
     ) -> None:
-        actual_finished_at = finished_at or naive_utc_now()
+        finished_at = naive_utc_now()
         snapshot = self._node_snapshots.get(domain_execution.id)
         start_at = snapshot.created_at if snapshot else domain_execution.created_at
         domain_execution.status = status
-        domain_execution.finished_at = actual_finished_at
-        domain_execution.elapsed_time = max((actual_finished_at - start_at).total_seconds(), 0.0)
+        domain_execution.finished_at = finished_at
+        domain_execution.elapsed_time = max((finished_at - start_at).total_seconds(), 0.0)
 
         if error:
             domain_execution.error = error

api/core/callback_handler/index_tool_callback_handler.py

@@ -12,7 +12,7 @@ from core.rag.models.document import Document
 from extensions.ext_database import db
 from models.dataset import ChildChunk, DatasetQuery, DocumentSegment
 from models.dataset import Document as DatasetDocument
-from models.enums import CreatorUserRole, DatasetQuerySource
+from models.enums import CreatorUserRole
 
 _logger = logging.getLogger(__name__)
 
@@ -36,7 +36,7 @@ class DatasetIndexToolCallbackHandler:
         dataset_query = DatasetQuery(
             dataset_id=dataset_id,
             content=query,
-            source=DatasetQuerySource.APP,
+            source="app",
             source_app_id=self._app_id,
             created_by_role=(
                 CreatorUserRole.ACCOUNT