fix: save and sync partner cookie failed

.agents/skills/frontend-query-mutation/references/runtime-rules.md

@@ -64,7 +64,7 @@ export const useUpdateAccessMode = () => {
 
 // Component only adds UI behavior.
 updateAccessMode({ appId, mode }, {
-  onSuccess: () => toast.success('...'),
+  onSuccess: () => Toast.notify({ type: 'success', message: '...' }),
 })
 
 // Avoid putting invalidation knowledge in the component.
@@ -114,7 +114,10 @@ try {
   router.push(`/orders/${order.id}`)
 }
 catch (error) {
-  toast.error(error instanceof Error ? error.message : 'Unknown error')
+  Toast.notify({
+    type: 'error',
+    message: error instanceof Error ? error.message : 'Unknown error',
+  })
 }
 ```
 

.github/labeler.yml

@@ -1,10 +1,3 @@
 web:
   - changed-files:
-      - any-glob-to-any-file:
-          - 'web/**'
-          - 'packages/**'
-          - 'package.json'
-          - 'pnpm-lock.yaml'
-          - 'pnpm-workspace.yaml'
-          - '.npmrc'
-          - '.nvmrc'
+      - any-glob-to-any-file: 'web/**'

.github/pull_request_template.md

@@ -20,4 +20,4 @@
 - [x] I understand that this PR may be closed in case there was no previous discussion or issues. (This doesn't apply to typos!)
 - [x] I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change.
 - [x] I've updated the documentation accordingly.
-- [x] I ran `make lint` and `make type-check` (backend) and `cd web && pnpm exec vp staged` (frontend) to appease the lint gods
+- [x] I ran `make lint` and `make type-check` (backend) and `cd web && npx lint-staged` (frontend) to appease the lint gods

.github/scripts/generate-i18n-changes.mjs

@@ -1,82 +0,0 @@
-import { execFileSync } from 'node:child_process'
-import fs from 'node:fs'
-import path from 'node:path'
-
-const repoRoot = process.cwd()
-const baseSha = process.env.BASE_SHA || ''
-const headSha = process.env.HEAD_SHA || ''
-const files = (process.env.CHANGED_FILES || '').split(/\s+/).filter(Boolean)
-const outputPath = process.env.I18N_CHANGES_OUTPUT_PATH || '/tmp/i18n-changes.json'
-
-const englishPath = fileStem => path.join(repoRoot, 'web', 'i18n', 'en-US', `${fileStem}.json`)
-
-const readCurrentJson = (fileStem) => {
-  const filePath = englishPath(fileStem)
-  if (!fs.existsSync(filePath))
-    return null
-
-  return JSON.parse(fs.readFileSync(filePath, 'utf8'))
-}
-
-const readBaseJson = (fileStem) => {
-  if (!baseSha)
-    return null
-
-  try {
-    const relativePath = `web/i18n/en-US/${fileStem}.json`
-    const content = execFileSync('git', ['show', `${baseSha}:${relativePath}`], { encoding: 'utf8' })
-    return JSON.parse(content)
-  }
-  catch {
-    return null
-  }
-}
-
-const compareJson = (beforeValue, afterValue) => JSON.stringify(beforeValue) === JSON.stringify(afterValue)
-
-const changes = {}
-
-for (const fileStem of files) {
-  const currentJson = readCurrentJson(fileStem)
-  const beforeJson = readBaseJson(fileStem) || {}
-  const afterJson = currentJson || {}
-  const added = {}
-  const updated = {}
-  const deleted = []
-
-  for (const [key, value] of Object.entries(afterJson)) {
-    if (!(key in beforeJson)) {
-      added[key] = value
-      continue
-    }
-
-    if (!compareJson(beforeJson[key], value)) {
-      updated[key] = {
-        before: beforeJson[key],
-        after: value,
-      }
-    }
-  }
-
-  for (const key of Object.keys(beforeJson)) {
-    if (!(key in afterJson))
-      deleted.push(key)
-  }
-
-  changes[fileStem] = {
-    fileDeleted: currentJson === null,
-    added,
-    updated,
-    deleted,
-  }
-}
-
-fs.writeFileSync(
-  outputPath,
-  JSON.stringify({
-    baseSha,
-    headSha,
-    files,
-    changes,
-  })
-)

.github/workflows/autofix.yml

@@ -39,11 +39,9 @@ jobs:
         with:
           files: |
             web/**
-            packages/**
             package.json
             pnpm-lock.yaml
             pnpm-workspace.yaml
-            .npmrc
             .nvmrc
       - name: Check api inputs
         if: github.event_name != 'merge_group'

.github/workflows/build-push.yml

@@ -65,7 +65,7 @@ jobs:
           echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
 
       - name: Login to Docker Hub
-        uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
@@ -130,7 +130,7 @@ jobs:
           merge-multiple: true
 
       - name: Login to Docker Hub
-        uses: docker/login-action@4907a6ddec9925e35a0a9e82d7399ccc52663121 # v4.1.0
+        uses: docker/login-action@b45d80f862d83dbcd57f89517bcf500b2ab88fb2 # v4.0.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}

.github/workflows/docker-build.yml

@@ -8,11 +8,9 @@ on:
       - api/Dockerfile
       - web/docker/**
       - web/Dockerfile
-      - packages/**
       - package.json
       - pnpm-lock.yaml
       - pnpm-workspace.yaml
-      - .npmrc
       - .nvmrc
 
 concurrency:

.github/workflows/main-ci.yml

@@ -65,11 +65,9 @@ jobs:
               - 'docker/volumes/sandbox/conf/**'
             web:
               - 'web/**'
-              - 'packages/**'
               - 'package.json'
               - 'pnpm-lock.yaml'
               - 'pnpm-workspace.yaml'
-              - '.npmrc'
               - '.nvmrc'
               - '.github/workflows/web-tests.yml'
               - '.github/actions/setup-web/**'
@@ -79,11 +77,9 @@ jobs:
               - 'api/uv.lock'
               - 'e2e/**'
               - 'web/**'
-              - 'packages/**'
               - 'package.json'
               - 'pnpm-lock.yaml'
               - 'pnpm-workspace.yaml'
-              - '.npmrc'
               - '.nvmrc'
               - 'docker/docker-compose.middleware.yaml'
               - 'docker/middleware.env.example'

.github/workflows/style.yml

@@ -77,11 +77,9 @@ jobs:
         with:
           files: |
             web/**
-            packages/**
             package.json
             pnpm-lock.yaml
             pnpm-workspace.yaml
-            .npmrc
             .nvmrc
             .github/workflows/style.yml
             .github/actions/setup-web/**
@@ -151,7 +149,7 @@ jobs:
             .editorconfig
 
       - name: Super-linter
-        uses: super-linter/super-linter/slim@9e863354e3ff62e0727d37183162c4a88873df41 # v8.6.0
+        uses: super-linter/super-linter/slim@61abc07d755095a68f4987d1c2c3d1d64408f1f9 # v8.5.0
         if: steps.changed-files.outputs.any_changed == 'true'
         env:
           BASH_SEVERITY: warning

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

@@ -9,7 +9,6 @@ on:
       - package.json
       - pnpm-lock.yaml
       - pnpm-workspace.yaml
-      - .npmrc
 
 concurrency:
   group: sdk-tests-${{ github.head_ref || github.run_id }}

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

@@ -68,7 +68,89 @@ jobs:
           " web/i18n-config/languages.ts | sed 's/[[:space:]]*$//')
 
           generate_changes_json() {
-            node .github/scripts/generate-i18n-changes.mjs
+            node <<'NODE'
+            const { execFileSync } = require('node:child_process')
+            const fs = require('node:fs')
+            const path = require('node:path')
+
+            const repoRoot = process.cwd()
+            const baseSha = process.env.BASE_SHA || ''
+            const headSha = process.env.HEAD_SHA || ''
+            const files = (process.env.CHANGED_FILES || '').split(/\s+/).filter(Boolean)
+
+            const englishPath = fileStem => path.join(repoRoot, 'web', 'i18n', 'en-US', `${fileStem}.json`)
+
+            const readCurrentJson = (fileStem) => {
+              const filePath = englishPath(fileStem)
+              if (!fs.existsSync(filePath))
+                return null
+
+              return JSON.parse(fs.readFileSync(filePath, 'utf8'))
+            }
+
+            const readBaseJson = (fileStem) => {
+              if (!baseSha)
+                return null
+
+              try {
+                const relativePath = `web/i18n/en-US/${fileStem}.json`
+                const content = execFileSync('git', ['show', `${baseSha}:${relativePath}`], { encoding: 'utf8' })
+                return JSON.parse(content)
+              }
+              catch (error) {
+                return null
+              }
+            }
+
+            const compareJson = (beforeValue, afterValue) => JSON.stringify(beforeValue) === JSON.stringify(afterValue)
+
+            const changes = {}
+
+            for (const fileStem of files) {
+              const currentJson = readCurrentJson(fileStem)
+              const beforeJson = readBaseJson(fileStem) || {}
+              const afterJson = currentJson || {}
+              const added = {}
+              const updated = {}
+              const deleted = []
+
+              for (const [key, value] of Object.entries(afterJson)) {
+                if (!(key in beforeJson)) {
+                  added[key] = value
+                  continue
+                }
+
+                if (!compareJson(beforeJson[key], value)) {
+                  updated[key] = {
+                    before: beforeJson[key],
+                    after: value,
+                  }
+                }
+              }
+
+              for (const key of Object.keys(beforeJson)) {
+                if (!(key in afterJson))
+                  deleted.push(key)
+              }
+
+              changes[fileStem] = {
+                fileDeleted: currentJson === null,
+                added,
+                updated,
+                deleted,
+              }
+            }
+
+            fs.writeFileSync(
+              '/tmp/i18n-changes.json',
+              JSON.stringify({
+                baseSha,
+                headSha,
+                files,
+                changes,
+              })
+            )
+            NODE
           }
 
           if [ "${{ github.event_name }}" = "repository_dispatch" ]; then
@@ -158,7 +240,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@6e2bd52842c65e914eba5c8badd17560bd26b5de # v1.0.89
+        uses: anthropics/claude-code-action@88c168b39e7e64da0286d812b6e9fbebb6708185 # v1.0.82
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}
@@ -188,7 +270,7 @@ jobs:
             Tool rules:
             - Use Read for repository files.
             - Use Edit for JSON updates.
-            - Use Bash only for `vp`.
+            - Use Bash only for `pnpm`.
             - Do not use Bash for `git`, `gh`, or branch management.
 
             Required execution plan:
@@ -210,7 +292,7 @@ jobs:
                - Read the current English JSON file for any file that still exists so wording, placeholders, and surrounding terminology stay accurate.
                - If `Structured change set available` is `false`, treat this as a scoped full sync and use the current English files plus scoped checks as the source of truth.
             4. Run a scoped pre-check before editing:
-               - `vp run dify-web#i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
+               - `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
                - Use this command as the source of truth for missing and extra keys inside the current scope.
             5. Apply translations.
                - For every target language and scoped file:
@@ -218,19 +300,19 @@ jobs:
                  - If the locale file does not exist yet, create it with `Write` and then continue with `Edit` as needed.
                  - ADD missing keys.
                  - UPDATE stale translations when the English value changed.
-                 - DELETE removed keys. Prefer `vp run dify-web#i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }} --auto-remove` for extra keys so deletions stay in scope.
+                 - DELETE removed keys. Prefer `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }} --auto-remove` for extra keys so deletions stay in scope.
                - Preserve placeholders exactly: `{{variable}}`, `${variable}`, HTML tags, component tags, and variable names.
                - Match the existing terminology and register used by each locale.
                - Prefer one Edit per file when stable, but prioritize correctness over batching.
             6. Verify only the edited files.
-               - Run `vp run dify-web#lint:fix --quiet -- <relative edited i18n file paths under web/>`
-               - Run `vp run dify-web#i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
+               - Run `pnpm --dir ${{ github.workspace }}/web lint:fix --quiet -- <relative edited i18n file paths>`
+               - Run `pnpm --dir ${{ github.workspace }}/web run i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
                - If verification fails, fix the remaining problems before continuing.
             7. Stop after the scoped locale files are updated and verification passes.
                - Do not create branches, commits, or pull requests.
           claude_args: |
             --max-turns 120
-            --allowedTools "Read,Write,Edit,Bash(vp *),Bash(vp:*),Glob,Grep"
+            --allowedTools "Read,Write,Edit,Bash(pnpm *),Bash(pnpm:*),Glob,Grep"
 
       - name: Prepare branch metadata
         id: pr_meta
@@ -272,7 +354,6 @@ jobs:
       - name: Create or update translation PR
         if: steps.pr_meta.outputs.has_changes == 'true'
         env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           BRANCH_NAME: ${{ steps.pr_meta.outputs.branch_name }}
           FILES_IN_SCOPE: ${{ steps.context.outputs.CHANGED_FILES }}
           TARGET_LANGS: ${{ steps.context.outputs.TARGET_LANGS }}
@@ -321,8 +402,8 @@ jobs:
             '',
             '## Verification',
             '',
-            `- \`vp run dify-web#i18n:check --file ${process.env.FILES_IN_SCOPE} --lang ${process.env.TARGET_LANGS}\``,
-            `- \`vp run dify-web#lint:fix --quiet -- <edited i18n files under web/>\``,
+            `- \`pnpm --dir web run i18n:check --file ${process.env.FILES_IN_SCOPE} --lang ${process.env.TARGET_LANGS}\``,
+            `- \`pnpm --dir web lint:fix --quiet -- <edited i18n files>\``,
             '',
             '## Notes',
             '',

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

@@ -42,7 +42,88 @@ jobs:
           fi
 
           export BASE_SHA HEAD_SHA CHANGED_FILES
-          node .github/scripts/generate-i18n-changes.mjs
+          node <<'NODE'
+          const { execFileSync } = require('node:child_process')
+          const fs = require('node:fs')
+          const path = require('node:path')
+
+          const repoRoot = process.cwd()
+          const baseSha = process.env.BASE_SHA || ''
+          const headSha = process.env.HEAD_SHA || ''
+          const files = (process.env.CHANGED_FILES || '').split(/\s+/).filter(Boolean)
+
+          const englishPath = fileStem => path.join(repoRoot, 'web', 'i18n', 'en-US', `${fileStem}.json`)
+
+          const readCurrentJson = (fileStem) => {
+            const filePath = englishPath(fileStem)
+            if (!fs.existsSync(filePath))
+              return null
+
+            return JSON.parse(fs.readFileSync(filePath, 'utf8'))
+          }
+
+          const readBaseJson = (fileStem) => {
+            if (!baseSha)
+              return null
+
+            try {
+              const relativePath = `web/i18n/en-US/${fileStem}.json`
+              const content = execFileSync('git', ['show', `${baseSha}:${relativePath}`], { encoding: 'utf8' })
+              return JSON.parse(content)
+            }
+            catch (error) {
+              return null
+            }
+          }
+
+          const compareJson = (beforeValue, afterValue) => JSON.stringify(beforeValue) === JSON.stringify(afterValue)
+
+          const changes = {}
+
+          for (const fileStem of files) {
+            const beforeJson = readBaseJson(fileStem) || {}
+            const afterJson = readCurrentJson(fileStem) || {}
+            const added = {}
+            const updated = {}
+            const deleted = []
+
+            for (const [key, value] of Object.entries(afterJson)) {
+              if (!(key in beforeJson)) {
+                added[key] = value
+                continue
+              }
+
+              if (!compareJson(beforeJson[key], value)) {
+                updated[key] = {
+                  before: beforeJson[key],
+                  after: value,
+                }
+              }
+            }
+
+            for (const key of Object.keys(beforeJson)) {
+              if (!(key in afterJson))
+                deleted.push(key)
+            }
+
+            changes[fileStem] = {
+              fileDeleted: readCurrentJson(fileStem) === null,
+              added,
+              updated,
+              deleted,
+            }
+          }
+
+          fs.writeFileSync(
+            '/tmp/i18n-changes.json',
+            JSON.stringify({
+              baseSha,
+              headSha,
+              files,
+              changes,
+            })
+          )
+          NODE
 
           if [ -n "$CHANGED_FILES" ]; then
             echo "has_changes=true" >> "$GITHUB_OUTPUT"

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

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

.gitignore

@@ -212,7 +212,7 @@ api/.vscode
 
 # pnpm
 /.pnpm-store
-node_modules
+/node_modules
 .vite-hooks/_
 
 # plugin migrate

.npmrc

@@ -1 +0,0 @@
-save-exact=true

.vite-hooks/pre-commit

@@ -81,18 +81,38 @@ if $web_modified; then
 
     if $web_ts_modified; then
         echo "Running TypeScript type-check:tsgo"
-        if ! npm run type-check:tsgo; then
-            echo "Type check failed. Please run 'npm run type-check:tsgo' to fix the errors."
+        if ! pnpm run type-check:tsgo; then
+            echo "Type check failed. Please run 'pnpm run type-check:tsgo' to fix the errors."
             exit 1
         fi
     else
         echo "No staged TypeScript changes detected, skipping type-check:tsgo"
     fi
 
-    echo "Running knip"
-    if ! npm run knip; then
-        echo "Knip check failed. Please run 'npm run knip' to fix the errors."
-        exit 1
+    echo "Running unit tests check"
+    modified_files=$(git diff --cached --name-only -- utils | grep -v '\.spec\.ts$' || true)
+
+    if [ -n "$modified_files" ]; then
+        for file in $modified_files; do
+            test_file="${file%.*}.spec.ts"
+            echo "Checking for test file: $test_file"
+
+            # check if the test file exists
+            if [ -f "../$test_file" ]; then
+                echo "Detected changes in $file, running corresponding unit tests..."
+                pnpm run test "../$test_file"
+
+                if [ $? -ne 0 ]; then
+                    echo "Unit tests failed. Please fix the errors before committing."
+                    exit 1
+                fi
+                echo "Unit tests for $file passed."
+            else
+                echo "Warning: $file does not have a corresponding test file."
+            fi
+
+        done
+        echo "All unit tests for modified web/utils files have passed."
     fi
 
     cd ../

api/.env.example

@@ -71,13 +71,6 @@ REDIS_USE_CLUSTERS=false
 REDIS_CLUSTERS=
 REDIS_CLUSTERS_PASSWORD=
 
-REDIS_RETRY_RETRIES=3
-REDIS_RETRY_BACKOFF_BASE=1.0
-REDIS_RETRY_BACKOFF_CAP=10.0
-REDIS_SOCKET_TIMEOUT=5.0
-REDIS_SOCKET_CONNECT_TIMEOUT=5.0
-REDIS_HEALTH_CHECK_INTERVAL=30
-
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
 CELERY_BACKEND=redis
@@ -109,7 +102,6 @@ S3_BUCKET_NAME=your-bucket-name
 S3_ACCESS_KEY=your-access-key
 S3_SECRET_KEY=your-secret-key
 S3_REGION=your-region
-S3_ADDRESS_STYLE=auto
 
 # Workflow run and Conversation archive storage (S3-compatible)
 ARCHIVE_STORAGE_ENABLED=false

api/.ruff.toml

@@ -115,6 +115,12 @@ ignore = [
 "controllers/console/human_input_form.py" = ["TID251"]
 "controllers/web/human_input_form.py" = ["TID251"]
 
+[lint.pyflakes]
+allowed-unused-imports = [
+    "tests.integration_tests",
+    "tests.unit_tests",
+]
+
 [lint.flake8-tidy-imports]
 
 [lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]

api/celery_healthcheck.py

@@ -1,18 +0,0 @@
-# This module provides a lightweight Celery instance for use in Docker health checks.
-# Unlike celery_entrypoint.py, this does NOT import app.py and therefore avoids
-# initializing all Flask extensions (DB, Redis, storage, blueprints, etc.).
-# Using this module keeps the health check fast and low-cost.
-from celery import Celery
-
-from configs import dify_config
-from extensions.ext_celery import get_celery_broker_transport_options, get_celery_ssl_options
-
-celery = Celery(broker=dify_config.CELERY_BROKER_URL)
-
-broker_transport_options = get_celery_broker_transport_options()
-if broker_transport_options:
-    celery.conf.update(broker_transport_options=broker_transport_options)
-
-ssl_options = get_celery_ssl_options()
-if ssl_options:
-    celery.conf.update(broker_use_ssl=ssl_options)

api/commands/retention.py

@@ -1,7 +1,7 @@
 import datetime
 import logging
 import time
-from typing import TypedDict
+from typing import Any
 
 import click
 import sqlalchemy as sa
@@ -503,19 +503,7 @@ def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
         return [row[0] for row in result]
 
 
-class _AppOrphanCounts(TypedDict):
-    variables: int
-    files: int
-
-
-class OrphanedDraftVariableStatsDict(TypedDict):
-    total_orphaned_variables: int
-    total_orphaned_files: int
-    orphaned_app_count: int
-    orphaned_by_app: dict[str, _AppOrphanCounts]
-
-
-def _count_orphaned_draft_variables() -> OrphanedDraftVariableStatsDict:
+def _count_orphaned_draft_variables() -> dict[str, Any]:
     """
     Count orphaned draft variables by app, including associated file counts.
 
@@ -538,7 +526,7 @@ def _count_orphaned_draft_variables() -> OrphanedDraftVariableStatsDict:
 
     with db.engine.connect() as conn:
         result = conn.execute(sa.text(variables_query))
-        orphaned_by_app: dict[str, _AppOrphanCounts] = {}
+        orphaned_by_app = {}
         total_files = 0
 
         for row in result:

api/configs/middleware/cache/redis_config.py

@@ -117,37 +117,6 @@ class RedisConfig(BaseSettings):
         default=None,
     )
 
-    REDIS_RETRY_RETRIES: NonNegativeInt = Field(
-        description="Maximum number of retries per Redis command on "
-        "transient failures (ConnectionError, TimeoutError, socket.timeout)",
-        default=3,
-    )
-
-    REDIS_RETRY_BACKOFF_BASE: PositiveFloat = Field(
-        description="Base delay in seconds for exponential backoff between retries",
-        default=1.0,
-    )
-
-    REDIS_RETRY_BACKOFF_CAP: PositiveFloat = Field(
-        description="Maximum backoff delay in seconds between retries",
-        default=10.0,
-    )
-
-    REDIS_SOCKET_TIMEOUT: PositiveFloat | None = Field(
-        description="Socket timeout in seconds for Redis read/write operations",
-        default=5.0,
-    )
-
-    REDIS_SOCKET_CONNECT_TIMEOUT: PositiveFloat | None = Field(
-        description="Socket timeout in seconds for Redis connection establishment",
-        default=5.0,
-    )
-
-    REDIS_HEALTH_CHECK_INTERVAL: NonNegativeInt = Field(
-        description="Interval in seconds between Redis connection health checks (0 to disable)",
-        default=30,
-    )
-
     @field_validator("REDIS_MAX_CONNECTIONS", mode="before")
     @classmethod
     def _empty_string_to_none_for_max_conns(cls, v):

api/context/execution_context.py

@@ -10,7 +10,7 @@ import threading
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Generator
 from contextlib import AbstractContextManager, contextmanager
-from typing import Any, Protocol, final, runtime_checkable
+from typing import Any, Protocol, TypeVar, final, runtime_checkable
 
 from pydantic import BaseModel
 
@@ -188,6 +188,8 @@ class ExecutionContextBuilder:
 _capturer: Callable[[], IExecutionContext] | None = None
 _tenant_context_providers: dict[tuple[str, str], Callable[[], BaseModel]] = {}
 
+T = TypeVar("T", bound=BaseModel)
+
 
 class ContextProviderNotFoundError(KeyError):
     """Raised when a tenant-scoped context provider is missing."""

api/contexts/wrapper.py

@@ -1,4 +1,7 @@
 from contextvars import ContextVar
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
 
 
 class HiddenValue:
@@ -8,7 +11,7 @@ class HiddenValue:
 _default = HiddenValue()
 
 
-class RecyclableContextVar[T]:
+class RecyclableContextVar(Generic[T]):
     """
     RecyclableContextVar is a wrapper around ContextVar
     It's safe to use in gunicorn with thread recycling, but features like `reset` are not available for now

api/controllers/common/controller_schemas.py

@@ -1,79 +0,0 @@
-from typing import Any, Literal
-
-from pydantic import BaseModel, Field, model_validator
-
-from libs.helper import UUIDStrOrEmpty
-
-# --- Conversation schemas ---
-
-
-class ConversationRenamePayload(BaseModel):
-    name: str | None = None
-    auto_generate: bool = False
-
-    @model_validator(mode="after")
-    def validate_name_requirement(self):
-        if not self.auto_generate:
-            if self.name is None or not self.name.strip():
-                raise ValueError("name is required when auto_generate is false")
-        return self
-
-
-# --- Message schemas ---
-
-
-class MessageListQuery(BaseModel):
-    conversation_id: UUIDStrOrEmpty
-    first_id: UUIDStrOrEmpty | None = None
-    limit: int = Field(default=20, ge=1, le=100)
-
-
-class MessageFeedbackPayload(BaseModel):
-    rating: Literal["like", "dislike"] | None = None
-    content: str | None = None
-
-
-# --- Saved message schemas ---
-
-
-class SavedMessageListQuery(BaseModel):
-    last_id: UUIDStrOrEmpty | None = None
-    limit: int = Field(default=20, ge=1, le=100)
-
-
-class SavedMessageCreatePayload(BaseModel):
-    message_id: UUIDStrOrEmpty
-
-
-# --- Workflow schemas ---
-
-
-class DefaultBlockConfigQuery(BaseModel):
-    q: str | None = None
-
-
-class WorkflowListQuery(BaseModel):
-    page: int = Field(default=1, ge=1, le=99999)
-    limit: int = Field(default=10, ge=1, le=100)
-    user_id: str | None = None
-    named_only: bool = False
-
-
-class WorkflowRunPayload(BaseModel):
-    inputs: dict[str, Any]
-    files: list[dict[str, Any]] | None = None
-
-
-class WorkflowUpdatePayload(BaseModel):
-    marked_name: str | None = Field(default=None, max_length=20)
-    marked_comment: str | None = Field(default=None, max_length=100)
-
-
-# --- Audio schemas ---
-
-
-class TextToAudioPayload(BaseModel):
-    message_id: str | None = None
-    voice: str | None = None
-    text: str | None = None
-    streaming: bool | None = None

api/controllers/common/fields.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, TypeAlias
 
 from graphon.file import helpers as file_helpers
 from pydantic import BaseModel, ConfigDict, computed_field
 
 from models.model import IconType
 
-type JSONValue = str | int | float | bool | None | dict[str, Any] | list[Any]
-type JSONObject = dict[str, Any]
+JSONValue: TypeAlias = str | int | float | bool | None | dict[str, Any] | list[Any]
+JSONObject: TypeAlias = dict[str, Any]
 
 
 class SystemParameters(BaseModel):

api/controllers/console/admin.py

@@ -2,7 +2,7 @@ import csv
 import io
 from collections.abc import Callable
 from functools import wraps
-from typing import cast
+from typing import ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource
@@ -18,7 +18,10 @@ from core.db.session_factory import session_factory
 from extensions.ext_database import db
 from libs.token import extract_access_token
 from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
-from services.billing_service import BillingService, LangContentDict
+from services.billing_service import BillingService
+
+P = ParamSpec("P")
+R = TypeVar("R")
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
@@ -69,9 +72,9 @@ console_ns.schema_model(
 )
 
 
-def admin_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def admin_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ADMIN_API_KEY:
             raise Unauthorized("API key is invalid.")
 
@@ -329,7 +332,7 @@ class UpsertNotificationApi(Resource):
     def post(self):
         payload = UpsertNotificationPayload.model_validate(console_ns.payload)
         result = BillingService.upsert_notification(
-            contents=[cast(LangContentDict, c.model_dump()) for c in payload.contents],
+            contents=[c.model_dump() for c in payload.contents],
             frequency=payload.frequency,
             status=payload.status,
             notification_id=payload.notification_id,

api/controllers/console/app/app.py

@@ -1,13 +1,13 @@
 import logging
 import uuid
 from datetime import datetime
-from typing import Any, Literal
+from typing import Any, Literal, TypeAlias
 
 from flask import request
 from flask_restx import Resource
 from graphon.enums import WorkflowExecutionStatus
 from graphon.file import helpers as file_helpers
-from pydantic import AliasChoices, BaseModel, Field, computed_field, field_validator
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, computed_field, field_validator
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest
@@ -26,25 +26,25 @@ from controllers.console.wraps import (
     setup_required,
 )
 from core.ops.ops_trace_manager import OpsTraceManager
-from core.rag.entities import PreProcessingRule, Rule, Segmentation
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
-from fields.base import ResponseModel
 from libs.login import current_account_with_tenant, login_required
 from models import App, DatasetPermissionEnum, Workflow
 from models.model import IconType
-from services.app_dsl_service import AppDslService
+from services.app_dsl_service import AppDslService, ImportMode
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
-from services.entities.dsl_entities import ImportMode
 from services.entities.knowledge_entities.knowledge_entities import (
     DataSource,
     InfoList,
     NotionIcon,
     NotionInfo,
     NotionPage,
+    PreProcessingRule,
     RerankingModel,
+    Rule,
+    Segmentation,
     WebsiteInfo,
     WeightKeywordSetting,
     WeightModel,
@@ -152,7 +152,17 @@ class AppTracePayload(BaseModel):
         return value
 
 
-type JSONValue = Any
+JSONValue: TypeAlias = Any
+
+
+class ResponseModel(BaseModel):
+    model_config = ConfigDict(
+        from_attributes=True,
+        extra="ignore",
+        populate_by_name=True,
+        serialize_by_alias=True,
+        protected_namespaces=(),
+    )
 
 
 def _to_timestamp(value: datetime | int | None) -> int | None:

api/controllers/console/app/app_import.py

@@ -17,9 +17,8 @@ from fields.app_fields import (
 )
 from libs.login import current_account_with_tenant, login_required
 from models.model import App
-from services.app_dsl_service import AppDslService
+from services.app_dsl_service import AppDslService, ImportStatus
 from services.enterprise.enterprise_service import EnterpriseService
-from services.entities.dsl_entities import ImportStatus
 from services.feature_service import FeatureService
 
 from .. import console_ns
@@ -93,13 +92,11 @@ class AppImportApi(Resource):
             EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
         # Return appropriate status code based on result
         status = result.status
-        match status:
-            case ImportStatus.FAILED:
-                return result.model_dump(mode="json"), 400
-            case ImportStatus.PENDING:
-                return result.model_dump(mode="json"), 202
-            case ImportStatus.COMPLETED | ImportStatus.COMPLETED_WITH_WARNINGS:
-                return result.model_dump(mode="json"), 200
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
 
 
 @console_ns.route("/apps/imports/<string:import_id>/confirm")

api/controllers/console/app/message.py

@@ -8,7 +8,6 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, func, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.controller_schemas import MessageFeedbackPayload as _MessageFeedbackPayloadBase
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
@@ -60,8 +59,10 @@ class ChatMessagesQuery(BaseModel):
         return uuid_value(value)
 
 
-class MessageFeedbackPayload(_MessageFeedbackPayloadBase):
+class MessageFeedbackPayload(BaseModel):
     message_id: str = Field(..., description="Message ID")
+    rating: Literal["like", "dislike"] | None = Field(default=None, description="Feedback rating")
+    content: str | None = Field(default=None, description="Feedback content")
 
     @field_validator("message_id")
     @classmethod

api/controllers/console/app/workflow.py

@@ -14,7 +14,6 @@ from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 import services
-from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
 from controllers.console import console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.workflow_run import workflow_run_node_execution_model
@@ -143,6 +142,10 @@ class PublishWorkflowPayload(BaseModel):
     marked_comment: str | None = Field(default=None, max_length=100)
 
 
+class DefaultBlockConfigQuery(BaseModel):
+    q: str | None = None
+
+
 class ConvertToWorkflowPayload(BaseModel):
     name: str | None = None
     icon_type: str | None = None
@@ -150,6 +153,18 @@ class ConvertToWorkflowPayload(BaseModel):
     icon_background: str | None = None
 
 
+class WorkflowListQuery(BaseModel):
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+    user_id: str | None = None
+    named_only: bool = False
+
+
+class WorkflowUpdatePayload(BaseModel):
+    marked_name: str | None = Field(default=None, max_length=20)
+    marked_comment: str | None = Field(default=None, max_length=100)
+
+
 class DraftWorkflowTriggerRunPayload(BaseModel):
     node_id: str
 

api/controllers/console/app/workflow_draft_variable.py

@@ -1,7 +1,7 @@
 import logging
 from collections.abc import Callable
 from functools import wraps
-from typing import Any
+from typing import Any, NoReturn, ParamSpec, TypeVar
 
 from flask import Response, request
 from flask_restx import Resource, fields, marshal, marshal_with
@@ -192,8 +192,11 @@ workflow_draft_variable_list_model = console_ns.model(
     "WorkflowDraftVariableList", workflow_draft_variable_list_fields_copy
 )
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-def _api_prerequisite[**P, R](f: Callable[P, R]) -> Callable[P, R | Response]:
+
+def _api_prerequisite(f: Callable[P, R]):
     """Common prerequisites for all draft workflow variable APIs.
 
     It ensures the following conditions are satisfied:
@@ -210,7 +213,7 @@ def _api_prerequisite[**P, R](f: Callable[P, R]) -> Callable[P, R | Response]:
     @edit_permission_required
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     @wraps(f)
-    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R | Response:
+    def wrapper(*args: P.args, **kwargs: P.kwargs):
         return f(*args, **kwargs)
 
     return wrapper
@@ -267,7 +270,7 @@ class WorkflowVariableCollectionApi(Resource):
         return Response("", 204)
 
 
-def validate_node_id(node_id: str) -> None:
+def validate_node_id(node_id: str) -> NoReturn | None:
     if node_id in [
         CONVERSATION_VARIABLE_NODE_ID,
         SYSTEM_VARIABLE_NODE_ID,
@@ -282,6 +285,7 @@ def validate_node_id(node_id: str) -> None:
         raise InvalidArgumentError(
             f"invalid node_id, please use correspond api for conversation and system variables, node_id={node_id}",
         )
+    return None
 
 
 @console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
@@ -384,27 +388,24 @@ class VariableApi(Resource):
 
         new_value = None
         if raw_value is not None:
-            match variable.value_type:
-                case SegmentType.FILE:
-                    if not isinstance(raw_value, dict):
-                        raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                    raw_value = build_from_mapping(
-                        mapping=raw_value,
-                        tenant_id=app_model.tenant_id,
-                        access_controller=_file_access_controller,
-                    )
-                case SegmentType.ARRAY_FILE:
-                    if not isinstance(raw_value, list):
-                        raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
-                    if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
-                        raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                    raw_value = build_from_mappings(
-                        mappings=raw_value,
-                        tenant_id=app_model.tenant_id,
-                        access_controller=_file_access_controller,
-                    )
-                case _:
-                    pass
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=app_model.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=app_model.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

api/controllers/console/app/workflow_trigger.py

@@ -66,13 +66,13 @@ class WebhookTriggerApi(Resource):
 
         with sessionmaker(db.engine).begin() as session:
             # Get webhook trigger for this app and node
-            webhook_trigger = session.scalar(
-                select(WorkflowWebhookTrigger)
+            webhook_trigger = (
+                session.query(WorkflowWebhookTrigger)
                 .where(
                     WorkflowWebhookTrigger.app_id == app_model.id,
                     WorkflowWebhookTrigger.node_id == node_id,
                 )
-                .limit(1)
+                .first()
             )
 
             if not webhook_trigger:

api/controllers/console/app/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import overload
+from typing import ParamSpec, TypeVar, Union
 
 from sqlalchemy import select
 
@@ -9,6 +9,11 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models import App, AppMode
 
+P = ParamSpec("P")
+R = TypeVar("R")
+P1 = ParamSpec("P1")
+R1 = TypeVar("R1")
+
 
 def _load_app_model(app_id: str) -> App | None:
     _, current_tenant_id = current_account_with_tenant()
@@ -23,30 +28,10 @@ def _load_app_model_with_trial(app_id: str) -> App | None:
     return app_model
 
 
-@overload
-def get_app_model[**P, R](
-    view: Callable[P, R],
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R]: ...
-
-
-@overload
-def get_app_model[**P, R](
-    view: None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[[Callable[P, R]], Callable[P, R]]: ...
-
-
-def get_app_model[**P, R](
-    view: Callable[P, R] | None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
-    def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
+def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+    def decorator(view_func: Callable[P1, R1]):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated_view(*args: P1.args, **kwargs: P1.kwargs):
             if not kwargs.get("app_id"):
                 raise ValueError("missing app_id in path parameters")
 
@@ -84,30 +69,10 @@ def get_app_model[**P, R](
         return decorator(view)
 
 
-@overload
-def get_app_model_with_trial[**P, R](
-    view: Callable[P, R],
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R]: ...
-
-
-@overload
-def get_app_model_with_trial[**P, R](
-    view: None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[[Callable[P, R]], Callable[P, R]]: ...
-
-
-def get_app_model_with_trial[**P, R](
-    view: Callable[P, R] | None = None,
-    *,
-    mode: AppMode | list[AppMode] | None = None,
-) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
-    def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
+def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+    def decorator(view_func: Callable[P, R]):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
             if not kwargs.get("app_id"):
                 raise ValueError("missing app_id in path parameters")
 

api/controllers/console/auth/forgot_password.py

@@ -3,7 +3,7 @@ import secrets
 
 from flask import request
 from flask_restx import Resource
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import sessionmaker
 
 from controllers.common.schema import register_schema_models
@@ -20,18 +20,35 @@ from controllers.console.wraps import email_password_login_enabled, setup_requir
 from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
 from libs.helper import EmailStr, extract_remote_ip
-from libs.password import hash_password
+from libs.password import hash_password, valid_password
 from services.account_service import AccountService, TenantService
-from services.entities.auth_entities import (
-    ForgotPasswordCheckPayload,
-    ForgotPasswordResetPayload,
-    ForgotPasswordSendPayload,
-)
 from services.feature_service import FeatureService
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
+class ForgotPasswordSendPayload(BaseModel):
+    email: EmailStr = Field(...)
+    language: str | None = Field(default=None)
+
+
+class ForgotPasswordCheckPayload(BaseModel):
+    email: EmailStr = Field(...)
+    code: str = Field(...)
+    token: str = Field(...)
+
+
+class ForgotPasswordResetPayload(BaseModel):
+    token: str = Field(...)
+    new_password: str = Field(...)
+    password_confirm: str = Field(...)
+
+    @field_validator("new_password", "password_confirm")
+    @classmethod
+    def validate_password(cls, value: str) -> str:
+        return valid_password(value)
+
+
 class ForgotPasswordEmailResponse(BaseModel):
     result: str = Field(description="Operation result")
     data: str | None = Field(default=None, description="Reset token")

api/controllers/console/auth/login.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import flask_login
 from flask import make_response, request
 from flask_restx import Resource
@@ -40,9 +42,8 @@ from libs.token import (
     set_csrf_token_to_cookie,
     set_refresh_token_to_cookie,
 )
-from services.account_service import AccountService, InvitationDetailDict, RegisterService, TenantService
+from services.account_service import AccountService, RegisterService, TenantService
 from services.billing_service import BillingService
-from services.entities.auth_entities import LoginPayloadBase
 from services.errors.account import AccountRegisterError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
@@ -50,7 +51,9 @@ from services.feature_service import FeatureService
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
-class LoginPayload(LoginPayloadBase):
+class LoginPayload(BaseModel):
+    email: EmailStr = Field(..., description="Email address")
+    password: str = Field(..., description="Password")
     remember_me: bool = Field(default=False, description="Remember me flag")
     invite_token: str | None = Field(default=None, description="Invitation token")
 
@@ -98,7 +101,7 @@ class LoginApi(Resource):
             raise EmailPasswordLoginLimitError()
 
         invite_token = args.invite_token
-        invitation_data: InvitationDetailDict | None = None
+        invitation_data: dict[str, Any] | None = None
         if invite_token:
             invitation_data = RegisterService.get_invitation_with_case_fallback(None, request_email, invite_token)
             if invitation_data is None:

api/controllers/console/auth/oauth_server.py

@@ -1,9 +1,8 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Concatenate
+from typing import Concatenate, ParamSpec, TypeVar
 
 from flask import jsonify, request
-from flask.typing import ResponseReturnValue
 from flask_restx import Resource
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from pydantic import BaseModel
@@ -17,6 +16,10 @@ from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType,
 
 from .. import console_ns
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
+
 
 class OAuthClientPayload(BaseModel):
     client_id: str
@@ -36,11 +39,9 @@ class OAuthTokenRequest(BaseModel):
     refresh_token: str | None = None
 
 
-def oauth_server_client_id_required[T, **P, R](
-    view: Callable[Concatenate[T, OAuthProviderApp, P], R],
-) -> Callable[Concatenate[T, P], R]:
+def oauth_server_client_id_required(view: Callable[Concatenate[T, OAuthProviderApp, P], R]):
     @wraps(view)
-    def decorated(self: T, *args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(self: T, *args: P.args, **kwargs: P.kwargs):
         json_data = request.get_json()
         if json_data is None:
             raise BadRequest("client_id is required")
@@ -57,13 +58,9 @@ def oauth_server_client_id_required[T, **P, R](
     return decorated
 
 
-def oauth_server_access_token_required[T, **P, R](
-    view: Callable[Concatenate[T, OAuthProviderApp, Account, P], R],
-) -> Callable[Concatenate[T, OAuthProviderApp, P], R | ResponseReturnValue]:
+def oauth_server_access_token_required(view: Callable[Concatenate[T, OAuthProviderApp, Account, P], R]):
     @wraps(view)
-    def decorated(
-        self: T, oauth_provider_app: OAuthProviderApp, *args: P.args, **kwargs: P.kwargs
-    ) -> R | ResponseReturnValue:
+    def decorated(self: T, oauth_provider_app: OAuthProviderApp, *args: P.args, **kwargs: P.kwargs):
         if not isinstance(oauth_provider_app, OAuthProviderApp):
             raise BadRequest("Invalid oauth_provider_app")
 

api/controllers/console/billing/billing.py

@@ -1,6 +1,4 @@
 import base64
-import json
-from datetime import UTC, datetime, timedelta
 from typing import Literal
 
 from flask import request
@@ -11,7 +9,6 @@ from werkzeug.exceptions import BadRequest
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
 from enums.cloud_plan import CloudPlan
-from extensions.ext_redis import redis_client
 from libs.login import current_account_with_tenant, login_required
 from services.billing_service import BillingService
 
@@ -87,39 +84,3 @@ class PartnerTenants(Resource):
             raise BadRequest("Invalid partner information")
 
         return BillingService.sync_partner_tenants_bindings(current_user.id, decoded_partner_key, click_id)
-
-
-_DEBUG_KEY = "billing:debug"
-_DEBUG_TTL = timedelta(days=7)
-
-
-class DebugDataPayload(BaseModel):
-    type: str = Field(..., min_length=1, description="Data type key")
-    data: str = Field(..., min_length=1, description="Data value to append")
-
-
-@console_ns.route("/billing/debug/data")
-class DebugData(Resource):
-    def post(self):
-        body = DebugDataPayload.model_validate(request.get_json(force=True))
-        item = json.dumps({
-            "type": body.type,
-            "data": body.data,
-            "createTime": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
-        })
-        redis_client.lpush(_DEBUG_KEY, item)
-        redis_client.expire(_DEBUG_KEY, _DEBUG_TTL)
-        return {"result": "ok"}, 201
-
-    def get(self):
-        recent = request.args.get("recent", 10, type=int)
-        items = redis_client.lrange(_DEBUG_KEY, 0, recent - 1)
-        return {
-            "data": [
-                json.loads(item.decode("utf-8") if isinstance(item, bytes) else item) for item in items
-            ]
-        }
-
-    def delete(self):
-        redis_client.delete(_DEBUG_KEY)
-        return {"result": "ok"}

api/controllers/console/datasets/data_source.py

@@ -158,11 +158,10 @@ class DataSourceApi(Resource):
     @login_required
     @account_initialization_required
     def patch(self, binding_id, action: Literal["enable", "disable"]):
-        _, current_tenant_id = current_account_with_tenant()
         binding_id = str(binding_id)
         with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
             data_source_binding = session.execute(
-                select(DataSourceOauthBinding).filter_by(id=binding_id, tenant_id=current_tenant_id)
+                select(DataSourceOauthBinding).filter_by(id=binding_id)
             ).scalar_one_or_none()
         if data_source_binding is None:
             raise NotFound("Data source binding not found.")

api/controllers/console/datasets/external.py

@@ -173,11 +173,8 @@ class ExternalApiTemplateApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, external_knowledge_api_id):
-        _, current_tenant_id = current_account_with_tenant()
         external_knowledge_api_id = str(external_knowledge_api_id)
-        external_knowledge_api = ExternalDatasetService.get_external_knowledge_api(
-            external_knowledge_api_id, current_tenant_id
-        )
+        external_knowledge_api = ExternalDatasetService.get_external_knowledge_api(external_knowledge_api_id)
         if external_knowledge_api is None:
             raise NotFound("API template not found.")
 

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

@@ -3,7 +3,6 @@ import logging
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
-from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from controllers.common.schema import register_schema_models
@@ -87,8 +86,8 @@ class CustomizedPipelineTemplateApi(Resource):
     @enterprise_license_required
     def post(self, template_id: str):
         with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            template = session.scalar(
-                select(PipelineCustomizedTemplate).where(PipelineCustomizedTemplate.id == template_id).limit(1)
+            template = (
+                session.query(PipelineCustomizedTemplate).where(PipelineCustomizedTemplate.id == template_id).first()
             )
             if not template:
                 raise ValueError("Customized pipeline template not found.")

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

@@ -1,5 +1,4 @@
 import logging
-from collections.abc import Callable
 from typing import Any, NoReturn
 
 from flask import Response, request
@@ -56,7 +55,7 @@ class WorkflowDraftVariablePatchPayload(BaseModel):
 register_schema_models(console_ns, WorkflowDraftVariablePatchPayload)
 
 
-def _api_prerequisite[**P, R](f: Callable[P, R]) -> Callable[P, R | Response]:
+def _api_prerequisite(f):
     """Common prerequisites for all draft workflow variable APIs.
 
     It ensures the following conditions are satisfied:
@@ -71,7 +70,7 @@ def _api_prerequisite[**P, R](f: Callable[P, R]) -> Callable[P, R | Response]:
     @login_required
     @account_initialization_required
     @get_rag_pipeline
-    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R | Response:
+    def wrapper(*args, **kwargs):
         if not isinstance(current_user, Account) or not current_user.has_edit_permission:
             raise Forbidden()
         return f(*args, **kwargs)
@@ -223,27 +222,24 @@ class RagPipelineVariableApi(Resource):
 
         new_value = None
         if raw_value is not None:
-            match variable.value_type:
-                case SegmentType.FILE:
-                    if not isinstance(raw_value, dict):
-                        raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
-                    raw_value = build_from_mapping(
-                        mapping=raw_value,
-                        tenant_id=pipeline.tenant_id,
-                        access_controller=_file_access_controller,
-                    )
-                case SegmentType.ARRAY_FILE:
-                    if not isinstance(raw_value, list):
-                        raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
-                    if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
-                        raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
-                    raw_value = build_from_mappings(
-                        mappings=raw_value,
-                        tenant_id=pipeline.tenant_id,
-                        access_controller=_file_access_controller,
-                    )
-                case _:
-                    pass
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(
+                    mapping=raw_value,
+                    tenant_id=pipeline.tenant_id,
+                    access_controller=_file_access_controller,
+                )
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(
+                    mappings=raw_value,
+                    tenant_id=pipeline.tenant_id,
+                    access_controller=_file_access_controller,
+                )
             new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()

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

@@ -19,7 +19,7 @@ from fields.rag_pipeline_fields import (
 )
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Pipeline
-from services.entities.dsl_entities import ImportStatus
+from services.app_dsl_service import ImportStatus
 from services.rag_pipeline.rag_pipeline_dsl_service import RagPipelineDslService
 
 
@@ -83,13 +83,11 @@ class RagPipelineImportApi(Resource):
 
         # Return appropriate status code based on result
         status = result.status
-        match status:
-            case ImportStatus.FAILED:
-                return result.model_dump(mode="json"), 400
-            case ImportStatus.PENDING:
-                return result.model_dump(mode="json"), 202
-            case ImportStatus.COMPLETED | ImportStatus.COMPLETED_WITH_WARNINGS:
-                return result.model_dump(mode="json"), 200
+        if status == ImportStatus.FAILED:
+            return result.model_dump(mode="json"), 400
+        elif status == ImportStatus.PENDING:
+            return result.model_dump(mode="json"), 202
+        return result.model_dump(mode="json"), 200
 
 
 @console_ns.route("/rag/pipelines/imports/<string:import_id>/confirm")

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

@@ -10,7 +10,6 @@ from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotFound
 
 import services
-from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
@@ -95,6 +94,22 @@ class PublishedWorkflowRunPayload(DraftWorkflowRunPayload):
     original_document_id: str | None = None
 
 
+class DefaultBlockConfigQuery(BaseModel):
+    q: str | None = None
+
+
+class WorkflowListQuery(BaseModel):
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=10, ge=1, le=100)
+    user_id: str | None = None
+    named_only: bool = False
+
+
+class WorkflowUpdatePayload(BaseModel):
+    marked_name: str | None = Field(default=None, max_length=20)
+    marked_comment: str | None = Field(default=None, max_length=100)
+
+
 class NodeIdQuery(BaseModel):
     node_id: str
 

api/controllers/console/datasets/wraps.py

@@ -1,5 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from sqlalchemy import select
 
@@ -8,10 +9,13 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.dataset import Pipeline
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-def get_rag_pipeline[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
+
+def get_rag_pipeline(view_func: Callable[P, R]):
     @wraps(view_func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
         if not kwargs.get("pipeline_id"):
             raise ValueError("missing pipeline_id in path parameters")
 

api/controllers/console/explore/audio.py

@@ -2,10 +2,10 @@ import logging
 
 from flask import request
 from graphon.model_runtime.errors.invoke import InvokeError
+from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.common.controller_schemas import TextToAudioPayload
 from controllers.common.schema import register_schema_model
 from controllers.console.app.error import (
     AppUnavailableError,
@@ -32,6 +32,14 @@ from .. import console_ns
 
 logger = logging.getLogger(__name__)
 
+
+class TextToAudioPayload(BaseModel):
+    message_id: str | None = None
+    voice: str | None = None
+    text: str | None = None
+    streaming: bool | None = Field(default=None, description="Enable streaming response")
+
+
 register_schema_model(console_ns, TextToAudioPayload)
 
 

api/controllers/console/explore/conversation.py

@@ -1,11 +1,10 @@
 from typing import Any
 
 from flask import request
-from pydantic import BaseModel, Field, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound
 
-from controllers.common.controller_schemas import ConversationRenamePayload
 from controllers.common.schema import register_schema_models
 from controllers.console.explore.error import NotChatAppError
 from controllers.console.explore.wraps import InstalledAppResource
@@ -33,6 +32,18 @@ class ConversationListQuery(BaseModel):
     pinned: bool | None = None
 
 
+class ConversationRenamePayload(BaseModel):
+    name: str | None = None
+    auto_generate: bool = False
+
+    @model_validator(mode="after")
+    def validate_name_requirement(self):
+        if not self.auto_generate:
+            if self.name is None or not self.name.strip():
+                raise ValueError("name is required when auto_generate is false")
+        return self
+
+
 register_schema_models(console_ns, ConversationListQuery, ConversationRenamePayload)
 
 

api/controllers/console/explore/message.py

@@ -3,10 +3,9 @@ from typing import Literal
 
 from flask import request
 from graphon.model_runtime.errors.invoke import InvokeError
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.controller_schemas import MessageFeedbackPayload, MessageListQuery
 from controllers.common.schema import register_schema_models
 from controllers.console.app.error import (
     AppMoreLikeThisDisabledError,
@@ -26,6 +25,7 @@ from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotIni
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
 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
@@ -44,6 +44,17 @@ from .. import console_ns
 logger = logging.getLogger(__name__)
 
 
+class MessageListQuery(BaseModel):
+    conversation_id: UUIDStrOrEmpty
+    first_id: UUIDStrOrEmpty | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class MessageFeedbackPayload(BaseModel):
+    rating: Literal["like", "dislike"] | None = None
+    content: str | None = None
+
+
 class MoreLikeThisQuery(BaseModel):
     response_mode: Literal["blocking", "streaming"]
 

api/controllers/console/explore/saved_message.py

@@ -1,18 +1,28 @@
 from flask import request
-from pydantic import TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import NotFound
 
-from controllers.common.controller_schemas import SavedMessageCreatePayload, SavedMessageListQuery
 from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.explore.error import NotCompletionAppError
 from controllers.console.explore.wraps import InstalledAppResource
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SavedMessageInfiniteScrollPagination, SavedMessageItem
+from libs.helper import UUIDStrOrEmpty
 from libs.login import current_account_with_tenant
 from services.errors.message import MessageNotExistsError
 from services.saved_message_service import SavedMessageService
 
+
+class SavedMessageListQuery(BaseModel):
+    last_id: UUIDStrOrEmpty | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SavedMessageCreatePayload(BaseModel):
+    message_id: UUIDStrOrEmpty
+
+
 register_schema_models(console_ns, SavedMessageListQuery, SavedMessageCreatePayload)
 
 

api/controllers/console/explore/workflow.py

@@ -1,10 +1,11 @@
 import logging
+from typing import Any
 
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
+from pydantic import BaseModel
 from werkzeug.exceptions import InternalServerError
 
-from controllers.common.controller_schemas import WorkflowRunPayload
 from controllers.common.schema import register_schema_model
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -33,6 +34,12 @@ from .. import console_ns
 
 logger = logging.getLogger(__name__)
 
+
+class WorkflowRunPayload(BaseModel):
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
+
+
 register_schema_model(console_ns, WorkflowRunPayload)
 
 

api/controllers/console/explore/wraps.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
-from typing import Concatenate
+from typing import Concatenate, ParamSpec, TypeVar
 
 from flask import abort
 from flask_restx import Resource
@@ -15,8 +15,12 @@ from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
 
-def installed_app_required[**P, R](view: Callable[Concatenate[InstalledApp, P], R] | None = None):
+
+def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | None = None):
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
@@ -45,7 +49,7 @@ def installed_app_required[**P, R](view: Callable[Concatenate[InstalledApp, P],
     return decorator
 
 
-def user_allowed_to_access_app[**P, R](view: Callable[Concatenate[InstalledApp, P], R] | None = None):
+def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] | None = None):
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):
@@ -69,7 +73,7 @@ def user_allowed_to_access_app[**P, R](view: Callable[Concatenate[InstalledApp,
     return decorator
 
 
-def trial_app_required[**P, R](view: Callable[Concatenate[App, P], R] | None = None):
+def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
     def decorator(view: Callable[Concatenate[App, P], R]):
         @wraps(view)
         def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
@@ -102,7 +106,7 @@ def trial_app_required[**P, R](view: Callable[Concatenate[App, P], R] | None = N
     return decorator
 
 
-def trial_feature_enable[**P, R](view: Callable[P, R]):
+def trial_feature_enable(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
@@ -113,7 +117,7 @@ def trial_feature_enable[**P, R](view: Callable[P, R]):
     return decorated
 
 
-def explore_banner_enabled[**P, R](view: Callable[P, R]):
+def explore_banner_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()

api/controllers/console/human_input_form.py

@@ -168,13 +168,12 @@ class ConsoleWorkflowEventsApi(Resource):
         else:
             msg_generator = MessageGenerator()
             generator: BaseAppGenerator
-            match app.mode:
-                case AppMode.ADVANCED_CHAT:
-                    generator = AdvancedChatAppGenerator()
-                case AppMode.WORKFLOW:
-                    generator = WorkflowAppGenerator()
-                case _:
-                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+            if app.mode == AppMode.ADVANCED_CHAT:
+                generator = AdvancedChatAppGenerator()
+            elif app.mode == AppMode.WORKFLOW:
+                generator = WorkflowAppGenerator()
+            else:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
 

api/controllers/console/notification.py

@@ -1,5 +1,3 @@
-from typing import TypedDict
-
 from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
@@ -13,21 +11,6 @@ from services.billing_service import BillingService
 _FALLBACK_LANG = "en-US"
 
 
-class NotificationItemDict(TypedDict):
-    notification_id: str | None
-    frequency: str | None
-    lang: str
-    title: str
-    subtitle: str
-    body: str
-    title_pic_url: str
-
-
-class NotificationResponseDict(TypedDict):
-    should_show: bool
-    notifications: list[NotificationItemDict]
-
-
 def _pick_lang_content(contents: dict, lang: str) -> dict:
     """Return the single LangContent for *lang*, falling back to English."""
     return contents.get(lang) or contents.get(_FALLBACK_LANG) or next(iter(contents.values()), {})
@@ -62,30 +45,28 @@ class NotificationApi(Resource):
         result = BillingService.get_account_notification(str(current_user.id))
 
         # Proto JSON uses camelCase field names (Kratos default marshaling).
-        response: NotificationResponseDict
         if not result.get("shouldShow"):
-            response = {"should_show": False, "notifications": []}
-            return response, 200
+            return {"should_show": False, "notifications": []}, 200
 
         lang = current_user.interface_language or _FALLBACK_LANG
 
-        notifications: list[NotificationItemDict] = []
+        notifications = []
         for notification in result.get("notifications") or []:
             contents: dict = notification.get("contents") or {}
             lang_content = _pick_lang_content(contents, lang)
-            item: NotificationItemDict = {
-                "notification_id": notification.get("notificationId"),
-                "frequency": notification.get("frequency"),
-                "lang": lang_content.get("lang", lang),
-                "title": lang_content.get("title", ""),
-                "subtitle": lang_content.get("subtitle", ""),
-                "body": lang_content.get("body", ""),
-                "title_pic_url": lang_content.get("titlePicUrl", ""),
-            }
-            notifications.append(item)
+            notifications.append(
+                {
+                    "notification_id": notification.get("notificationId"),
+                    "frequency": notification.get("frequency"),
+                    "lang": lang_content.get("lang", lang),
+                    "title": lang_content.get("title", ""),
+                    "subtitle": lang_content.get("subtitle", ""),
+                    "body": lang_content.get("body", ""),
+                    "title_pic_url": lang_content.get("titlePicUrl", ""),
+                }
+            )
 
-        response = {"should_show": bool(notifications), "notifications": notifications}
-        return response, 200
+        return {"should_show": bool(notifications), "notifications": notifications}, 200
 
 
 @console_ns.route("/notification/dismiss")

api/controllers/console/tag/tags.py

@@ -9,14 +9,7 @@ from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from libs.login import current_account_with_tenant, login_required
-from models.enums import TagType
-from services.tag_service import (
-    SaveTagPayload,
-    TagBindingCreatePayload,
-    TagBindingDeletePayload,
-    TagService,
-    UpdateTagPayload,
-)
+from services.tag_service import TagService
 
 dataset_tag_fields = {
     "id": fields.String,
@@ -32,19 +25,19 @@ def build_dataset_tag_fields(api_or_ns: Namespace):
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)
-    type: TagType = Field(description="Tag type")
+    type: Literal["knowledge", "app"] | None = Field(default=None, description="Tag type")
 
 
 class TagBindingPayload(BaseModel):
     tag_ids: list[str] = Field(description="Tag IDs to bind")
     target_id: str = Field(description="Target ID to bind tags to")
-    type: TagType = Field(description="Tag type")
+    type: Literal["knowledge", "app"] | None = Field(default=None, description="Tag type")
 
 
 class TagBindingRemovePayload(BaseModel):
     tag_id: str = Field(description="Tag ID to remove")
     target_id: str = Field(description="Target ID to unbind tag from")
-    type: TagType = Field(description="Tag type")
+    type: Literal["knowledge", "app"] | None = Field(default=None, description="Tag type")
 
 
 class TagListQueryParam(BaseModel):
@@ -89,7 +82,7 @@ class TagListApi(Resource):
             raise Forbidden()
 
         payload = TagBasePayload.model_validate(console_ns.payload or {})
-        tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=payload.type))
+        tag = TagService.save_tags(payload.model_dump())
 
         response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
 
@@ -110,7 +103,7 @@ class TagUpdateDeleteApi(Resource):
             raise Forbidden()
 
         payload = TagBasePayload.model_validate(console_ns.payload or {})
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=payload.type), tag_id)
+        tag = TagService.update_tags(payload.model_dump(), tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
@@ -143,9 +136,7 @@ class TagBindingCreateApi(Resource):
             raise Forbidden()
 
         payload = TagBindingPayload.model_validate(console_ns.payload or {})
-        TagService.save_tag_binding(
-            TagBindingCreatePayload(tag_ids=payload.tag_ids, target_id=payload.target_id, type=payload.type)
-        )
+        TagService.save_tag_binding(payload.model_dump())
 
         return {"result": "success"}, 200
 
@@ -163,8 +154,6 @@ class TagBindingDeleteApi(Resource):
             raise Forbidden()
 
         payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
-        TagService.delete_tag_binding(
-            TagBindingDeletePayload(tag_id=payload.tag_id, target_id=payload.target_id, type=payload.type)
-        )
+        TagService.delete_tag_binding(payload.model_dump())
 
         return {"result": "success"}, 200

api/controllers/console/workspace/__init__.py

@@ -1,7 +1,7 @@
 from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
-from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden
 
@@ -9,25 +9,28 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant
 from models.account import TenantPluginPermission
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def plugin_permission_required(
     install_required: bool = False,
     debug_required: bool = False,
 ):
-    def interceptor[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             current_user, current_tenant_id = current_account_with_tenant()
             user = current_user
             tenant_id = current_tenant_id
 
             with sessionmaker(db.engine).begin() as session:
-                permission = session.scalar(
-                    select(TenantPluginPermission)
+                permission = (
+                    session.query(TenantPluginPermission)
                     .where(
                         TenantPluginPermission.tenant_id == tenant_id,
                     )
-                    .limit(1)
+                    .first()
                 )
 
                 if not permission:

api/controllers/console/workspace/workspace.py

@@ -28,7 +28,7 @@ from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
-from models.account import Tenant, TenantCustomConfigDict, TenantStatus
+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
@@ -240,10 +240,8 @@ class CustomConfigWorkspaceApi(Resource):
         args = WorkspaceCustomConfigPayload.model_validate(payload)
         tenant = db.get_or_404(Tenant, current_tenant_id)
 
-        custom_config_dict: TenantCustomConfigDict = {
-            "remove_webapp_brand": args.remove_webapp_brand
-            if args.remove_webapp_brand is not None
-            else tenant.custom_config_dict.get("remove_webapp_brand", False),
+        custom_config_dict = {
+            "remove_webapp_brand": args.remove_webapp_brand,
             "replace_webapp_logo": args.replace_webapp_logo
             if args.replace_webapp_logo is not None
             else tenant.custom_config_dict.get("replace_webapp_logo"),

api/controllers/console/wraps.py

@@ -4,6 +4,7 @@ import os
 import time
 from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import abort, request
 from sqlalchemy import select
@@ -24,6 +25,9 @@ from services.operation_service import OperationService
 
 from .error import NotInitValidateError, NotSetupError, UnauthorizedAndForceLogout
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 # Field names for decryption
 FIELD_NAME_PASSWORD = "password"
 FIELD_NAME_CODE = "code"
@@ -33,7 +37,7 @@ ERROR_MSG_INVALID_ENCRYPTED_DATA = "Invalid encrypted data"
 ERROR_MSG_INVALID_ENCRYPTED_CODE = "Invalid encrypted code"
 
 
-def account_initialization_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def account_initialization_required(view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
         # check account initialization
@@ -46,7 +50,7 @@ def account_initialization_required[**P, R](view: Callable[P, R]) -> Callable[P,
     return decorated
 
 
-def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def only_edition_cloud(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "CLOUD":
@@ -57,7 +61,7 @@ def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def only_edition_enterprise(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ENTERPRISE_ENABLED:
@@ -68,7 +72,7 @@ def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def only_edition_self_hosted(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "SELF_HOSTED":
@@ -79,7 +83,7 @@ def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def cloud_edition_billing_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def cloud_edition_billing_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         _, current_tenant_id = current_account_with_tenant()
@@ -91,7 +95,7 @@ def cloud_edition_billing_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R
     return decorated
 
 
-def cloud_edition_billing_resource_check[**P, R](resource: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_resource_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -133,9 +137,7 @@ def cloud_edition_billing_resource_check[**P, R](resource: str) -> Callable[[Cal
     return interceptor
 
 
-def cloud_edition_billing_knowledge_limit_check[**P, R](
-    resource: str,
-) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_knowledge_limit_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -158,7 +160,7 @@ def cloud_edition_billing_knowledge_limit_check[**P, R](
     return interceptor
 
 
-def cloud_edition_billing_rate_limit_check[**P, R](resource: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_rate_limit_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -194,7 +196,7 @@ def cloud_edition_billing_rate_limit_check[**P, R](resource: str) -> Callable[[C
     return interceptor
 
 
-def cloud_utm_record[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def cloud_utm_record(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
@@ -213,7 +215,7 @@ def cloud_utm_record[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def setup_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def setup_required(view: Callable[P, R]) -> Callable[P, R]:
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
         # check setup
@@ -227,7 +229,7 @@ def setup_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def enterprise_license_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def enterprise_license_required(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         settings = FeatureService.get_system_features()
@@ -239,7 +241,7 @@ def enterprise_license_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def email_password_login_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def email_password_login_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
@@ -252,7 +254,7 @@ def email_password_login_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]
     return decorated
 
 
-def email_register_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def email_register_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
@@ -265,7 +267,7 @@ def email_register_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def enable_change_email[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def enable_change_email(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
@@ -278,7 +280,7 @@ def enable_change_email[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def is_allow_transfer_owner[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def is_allow_transfer_owner(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         from libs.workspace_permission import check_workspace_owner_transfer_permission
@@ -291,7 +293,7 @@ def is_allow_transfer_owner[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def knowledge_pipeline_publish_enabled[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def knowledge_pipeline_publish_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         _, current_tenant_id = current_account_with_tenant()
@@ -303,7 +305,7 @@ def knowledge_pipeline_publish_enabled[**P, R](view: Callable[P, R]) -> Callable
     return decorated
 
 
-def edit_permission_required[**P, R](f: Callable[P, R]) -> Callable[P, R]:
+def edit_permission_required(f: Callable[P, R]):
     @wraps(f)
     def decorated_function(*args: P.args, **kwargs: P.kwargs):
         from werkzeug.exceptions import Forbidden
@@ -321,7 +323,7 @@ def edit_permission_required[**P, R](f: Callable[P, R]) -> Callable[P, R]:
     return decorated_function
 
 
-def is_admin_or_owner_required[**P, R](f: Callable[P, R]) -> Callable[P, R]:
+def is_admin_or_owner_required(f: Callable[P, R]):
     @wraps(f)
     def decorated_function(*args: P.args, **kwargs: P.kwargs):
         from werkzeug.exceptions import Forbidden
@@ -337,7 +339,7 @@ def is_admin_or_owner_required[**P, R](f: Callable[P, R]) -> Callable[P, R]:
     return decorated_function
 
 
-def annotation_import_rate_limit[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def annotation_import_rate_limit(view: Callable[P, R]):
     """
     Rate limiting decorator for annotation import operations.
 
@@ -386,7 +388,7 @@ def annotation_import_rate_limit[**P, R](view: Callable[P, R]) -> Callable[P, R]
     return decorated
 
 
-def annotation_import_concurrency_limit[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def annotation_import_concurrency_limit(view: Callable[P, R]):
     """
     Concurrency control decorator for annotation import operations.
 
@@ -453,7 +455,7 @@ def _decrypt_field(field_name: str, error_class: type[Exception], error_message:
     payload[field_name] = decoded_value
 
 
-def decrypt_password_field[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def decrypt_password_field(view: Callable[P, R]):
     """
     Decorator to decrypt password field in request payload.
 
@@ -475,7 +477,7 @@ def decrypt_password_field[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def decrypt_code_field[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def decrypt_code_field(view: Callable[P, R]):
     """
     Decorator to decrypt verification code field in request payload.
 

api/controllers/inner_api/app/dsl.py

@@ -9,7 +9,7 @@ from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field
 from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session
 
 from controllers.common.schema import register_schema_model
 from controllers.console.wraps import setup_required
@@ -18,8 +18,7 @@ from controllers.inner_api.wraps import enterprise_inner_api_only
 from extensions.ext_database import db
 from models import Account, App
 from models.account import AccountStatus
-from services.app_dsl_service import AppDslService
-from services.entities.dsl_entities import ImportMode, ImportStatus
+from services.app_dsl_service import AppDslService, ImportMode, ImportStatus
 
 
 class InnerAppDSLImportPayload(BaseModel):
@@ -56,7 +55,7 @@ class EnterpriseAppDSLImport(Resource):
 
         account.set_tenant_id(workspace_id)
 
-        with sessionmaker(db.engine).begin() as session:
+        with Session(db.engine) as session:
             dsl_service = AppDslService(session)
             result = dsl_service.import_app(
                 account=account,
@@ -65,6 +64,7 @@ class EnterpriseAppDSLImport(Resource):
                 name=args.name,
                 description=args.description,
             )
+            session.commit()
 
         if result.status == ImportStatus.FAILED:
             return result.model_dump(mode="json"), 400

api/controllers/inner_api/plugin/wraps.py

@@ -1,5 +1,6 @@
 from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -12,6 +13,9 @@ from libs.login import current_user
 from models.account import Tenant
 from models.model import DefaultEndUserSessionID, EndUser
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 class TenantUserPayload(BaseModel):
     tenant_id: str
@@ -61,9 +65,9 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     return user_model
 
 
-def get_user_tenant[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
+def get_user_tenant(view_func: Callable[P, R]):
     @wraps(view_func)
-    def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
         payload = TenantUserPayload.model_validate(request.get_json(silent=True) or {})
 
         user_id = payload.user_id
@@ -93,14 +97,10 @@ def get_user_tenant[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
     return decorated_view
 
 
-def plugin_data[**P, R](
-    view: Callable[P, R] | None = None,
-    *,
-    payload_type: type[BaseModel],
-) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
-    def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
+def plugin_data(view: Callable[P, R] | None = None, *, payload_type: type[BaseModel]):
+    def decorator(view_func: Callable[P, R]):
         @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
             try:
                 data = request.get_json()
             except Exception:

api/controllers/inner_api/wraps.py

@@ -3,7 +3,10 @@ from collections.abc import Callable
 from functools import wraps
 from hashlib import sha1
 from hmac import new as hmac_new
+from typing import ParamSpec, TypeVar
 
+P = ParamSpec("P")
+R = TypeVar("R")
 from flask import abort, request
 
 from configs import dify_config
@@ -11,9 +14,9 @@ from extensions.ext_database import db
 from models.model import EndUser
 
 
-def billing_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def billing_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -27,9 +30,9 @@ def billing_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def enterprise_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def enterprise_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -43,9 +46,9 @@ def enterprise_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
     return decorated
 
 
-def enterprise_inner_api_user_auth[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def enterprise_inner_api_user_auth(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             return view(*args, **kwargs)
 
@@ -79,9 +82,9 @@ def enterprise_inner_api_user_auth[**P, R](view: Callable[P, R]) -> Callable[P,
     return decorated
 
 
-def plugin_inner_api_only[**P, R](view: Callable[P, R]) -> Callable[P, R]:
+def plugin_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.PLUGIN_DAEMON_KEY:
             abort(404)
 

api/controllers/mcp/mcp.py

@@ -4,7 +4,6 @@ from flask import Response
 from flask_restx import Resource
 from graphon.variables.input_entities import VariableEntity
 from pydantic import BaseModel, Field, ValidationError
-from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from controllers.common.schema import register_schema_model
@@ -81,11 +80,11 @@ class MCPAppApi(Resource):
 
     def _get_mcp_server_and_app(self, server_code: str, session: Session) -> tuple[AppMCPServer, App]:
         """Get and validate MCP server and app in one query session"""
-        mcp_server = session.scalar(select(AppMCPServer).where(AppMCPServer.server_code == server_code).limit(1))
+        mcp_server = session.query(AppMCPServer).where(AppMCPServer.server_code == server_code).first()
         if not mcp_server:
             raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server Not Found")
 
-        app = session.scalar(select(App).where(App.id == mcp_server.app_id).limit(1))
+        app = session.query(App).where(App.id == mcp_server.app_id).first()
         if not app:
             raise MCPRequestError(mcp_types.INVALID_REQUEST, "App Not Found")
 
@@ -191,12 +190,12 @@ class MCPAppApi(Resource):
     def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str) -> EndUser | None:
         """Get end user - manages its own database session"""
         with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
-            return session.scalar(
-                select(EndUser)
+            return (
+                session.query(EndUser)
                 .where(EndUser.tenant_id == tenant_id)
                 .where(EndUser.session_id == mcp_server_id)
                 .where(EndUser.type == "mcp")
-                .limit(1)
+                .first()
             )
 
     def _create_end_user(

api/controllers/service_api/app/conversation.py

@@ -2,12 +2,11 @@ from typing import Any, Literal
 
 from flask import request
 from flask_restx import Resource
-from pydantic import BaseModel, Field, TypeAdapter, field_validator
+from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, NotFound
 
 import services
-from controllers.common.controller_schemas import ConversationRenamePayload
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import NotChatAppError
@@ -35,6 +34,18 @@ class ConversationListQuery(BaseModel):
     )
 
 
+class ConversationRenamePayload(BaseModel):
+    name: str | None = Field(default=None, description="New conversation name (required if auto_generate is false)")
+    auto_generate: bool = Field(default=False, description="Auto-generate conversation name")
+
+    @model_validator(mode="after")
+    def validate_name_requirement(self):
+        if not self.auto_generate:
+            if self.name is None or not self.name.strip():
+                raise ValueError("name is required when auto_generate is false")
+        return self
+
+
 class ConversationVariablesQuery(BaseModel):
     last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last variable ID for pagination")
     limit: int = Field(default=20, ge=1, le=100, description="Number of variables to return")

api/controllers/service_api/app/message.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Literal
 
 from flask import request
 from flask_restx import Resource
@@ -6,7 +7,6 @@ from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 import services
-from controllers.common.controller_schemas import MessageFeedbackPayload, MessageListQuery
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import NotChatAppError
@@ -14,6 +14,7 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 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 (
@@ -26,6 +27,17 @@ from services.message_service import MessageService
 logger = logging.getLogger(__name__)
 
 
+class MessageListQuery(BaseModel):
+    conversation_id: UUIDStrOrEmpty
+    first_id: UUIDStrOrEmpty | None = None
+    limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return")
+
+
+class MessageFeedbackPayload(BaseModel):
+    rating: Literal["like", "dislike"] | None = Field(default=None, description="Feedback rating")
+    content: str | None = Field(default=None, description="Feedback content")
+
+
 class FeedbackListQuery(BaseModel):
     page: int = Field(default=1, ge=1, description="Page number")
     limit: int = Field(default=20, ge=1, le=101, description="Number of feedbacks per page")

api/controllers/service_api/app/workflow.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Literal
+from typing import Any, Literal
 
 from dateutil.parser import isoparse
 from flask import request
@@ -11,7 +11,6 @@ from pydantic import BaseModel, Field
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
-from controllers.common.controller_schemas import WorkflowRunPayload as WorkflowRunPayloadBase
 from controllers.common.schema import register_schema_models
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import (
@@ -47,7 +46,9 @@ from services.workflow_app_service import WorkflowAppService
 logger = logging.getLogger(__name__)
 
 
-class WorkflowRunPayload(WorkflowRunPayloadBase):
+class WorkflowRunPayload(BaseModel):
+    inputs: dict[str, Any]
+    files: list[dict[str, Any]] | None = None
     response_mode: Literal["blocking", "streaming"] | None = None
 
 

api/controllers/service_api/dataset/dataset.py

@@ -22,17 +22,10 @@ from fields.tag_fields import DataSetTag
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
-from models.enums import TagType
 from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
-from services.tag_service import (
-    SaveTagPayload,
-    TagBindingCreatePayload,
-    TagBindingDeletePayload,
-    TagService,
-    UpdateTagPayload,
-)
+from services.tag_service import TagService
 
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
@@ -520,7 +513,7 @@ class DatasetTagsApi(DatasetApiResource):
             raise Forbidden()
 
         payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
-        tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=TagType.KNOWLEDGE))
+        tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})
 
         response = DataSetTag.model_validate(
             {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
@@ -543,8 +536,9 @@ class DatasetTagsApi(DatasetApiResource):
             raise Forbidden()
 
         payload = TagUpdatePayload.model_validate(service_api_ns.payload or {})
+        params = {"name": payload.name, "type": "knowledge"}
         tag_id = payload.tag_id
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name, type=TagType.KNOWLEDGE), tag_id)
+        tag = TagService.update_tags(params, tag_id)
 
         binding_count = TagService.get_tag_binding_count(tag_id)
 
@@ -591,9 +585,7 @@ class DatasetTagBindingApi(DatasetApiResource):
             raise Forbidden()
 
         payload = TagBindingPayload.model_validate(service_api_ns.payload or {})
-        TagService.save_tag_binding(
-            TagBindingCreatePayload(tag_ids=payload.tag_ids, target_id=payload.target_id, type=TagType.KNOWLEDGE)
-        )
+        TagService.save_tag_binding({"tag_ids": payload.tag_ids, "target_id": payload.target_id, "type": "knowledge"})
 
         return "", 204
 
@@ -617,9 +609,7 @@ class DatasetTagUnbindingApi(DatasetApiResource):
             raise Forbidden()
 
         payload = TagUnbindingPayload.model_validate(service_api_ns.payload or {})
-        TagService.delete_tag_binding(
-            TagBindingDeletePayload(tag_id=payload.tag_id, target_id=payload.target_id, type=TagType.KNOWLEDGE)
-        )
+        TagService.delete_tag_binding({"tag_id": payload.tag_id, "target_id": payload.target_id, "type": "knowledge"})
 
         return "", 204
 

api/controllers/service_api/dataset/document.py

@@ -31,7 +31,6 @@ from controllers.service_api.wraps import (
     cloud_edition_billing_resource_check,
 )
 from core.errors.error import ProviderTokenNotInitError
-from core.rag.entities import PreProcessingRule, Rule, Segmentation
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
 from fields.document_fields import document_fields, document_status_fields
@@ -41,8 +40,11 @@ from models.enums import SegmentStatus
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import (
     KnowledgeConfig,
+    PreProcessingRule,
     ProcessRule,
     RetrievalModel,
+    Rule,
+    Segmentation,
 )
 from services.file_service import FileService
 from services.summary_index_service import SummaryIndexService

api/controllers/service_api/dataset/rag_pipeline/serializers.py

@@ -4,23 +4,13 @@ Serialization helpers for Service API knowledge pipeline endpoints.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, TypedDict
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from models.model import UploadFile
 
 
-class UploadFileDict(TypedDict):
-    id: str
-    name: str
-    size: int
-    extension: str
-    mime_type: str | None
-    created_by: str
-    created_at: str | None
-
-
-def serialize_upload_file(upload_file: UploadFile) -> UploadFileDict:
+def serialize_upload_file(upload_file: UploadFile) -> dict[str, Any]:
     return {
         "id": upload_file.id,
         "name": upload_file.name,

api/controllers/service_api/wraps.py

@@ -1,10 +1,9 @@
-import inspect
 import logging
 import time
 from collections.abc import Callable
 from enum import StrEnum, auto
 from functools import wraps
-from typing import cast, overload
+from typing import Concatenate, ParamSpec, TypeVar, cast, overload
 
 from flask import current_app, request
 from flask_login import user_logged_in
@@ -24,6 +23,10 @@ from services.api_token_service import ApiTokenCache, fetch_token_with_single_fl
 from services.end_user_service import EndUserService
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
+
 logger = logging.getLogger(__name__)
 
 
@@ -43,16 +46,16 @@ class FetchUserArg(BaseModel):
 
 
 @overload
-def validate_app_token[**P, R](view: Callable[P, R]) -> Callable[P, R]: ...
+def validate_app_token(view: Callable[P, R]) -> Callable[P, R]: ...
 
 
 @overload
-def validate_app_token[**P, R](
+def validate_app_token(
     view: None = None, *, fetch_user_arg: FetchUserArg | None = None
 ) -> Callable[[Callable[P, R]], Callable[P, R]]: ...
 
 
-def validate_app_token[**P, R](
+def validate_app_token(
     view: Callable[P, R] | None = None, *, fetch_user_arg: FetchUserArg | None = None
 ) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
     def decorator(view_func: Callable[P, R]) -> Callable[P, R]:
@@ -133,10 +136,7 @@ def validate_app_token[**P, R](
         return decorator(view)
 
 
-def cloud_edition_billing_resource_check[**P, R](
-    resource: str,
-    api_token_type: str,
-) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_resource_check(resource: str, api_token_type: str):
     def interceptor(view: Callable[P, R]):
         def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
@@ -166,10 +166,7 @@ def cloud_edition_billing_resource_check[**P, R](
     return interceptor
 
 
-def cloud_edition_billing_knowledge_limit_check[**P, R](
-    resource: str,
-    api_token_type: str,
-) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -191,10 +188,7 @@ def cloud_edition_billing_knowledge_limit_check[**P, R](
     return interceptor
 
 
-def cloud_edition_billing_rate_limit_check[**P, R](
-    resource: str,
-    api_token_type: str,
-) -> Callable[[Callable[P, R]], Callable[P, R]]:
+def cloud_edition_billing_rate_limit_check(resource: str, api_token_type: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
@@ -231,73 +225,99 @@ def cloud_edition_billing_rate_limit_check[**P, R](
     return interceptor
 
 
-def validate_dataset_token[R](view: Callable[..., R]) -> Callable[..., R]:
-    positional_parameters = [
-        parameter
-        for parameter in inspect.signature(view).parameters.values()
-        if parameter.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
-    ]
-    expects_bound_instance = bool(positional_parameters and positional_parameters[0].name in {"self", "cls"})
+@overload
+def validate_dataset_token(view: Callable[Concatenate[T, P], R]) -> Callable[P, R]: ...
 
-    @wraps(view)
-    def decorated(*args: object, **kwargs: object) -> R:
-        api_token = validate_and_get_api_token("dataset")
 
-        # Flask may pass URL path parameters positionally, so inspect both kwargs and args.
-        dataset_id = kwargs.get("dataset_id")
+@overload
+def validate_dataset_token(view: None = None) -> Callable[[Callable[Concatenate[T, P], R]], Callable[P, R]]: ...
 
-        if not dataset_id and args:
-            potential_id = args[0]
-            try:
-                str_id = str(potential_id)
-                if len(str_id) == 36 and str_id.count("-") == 4:
-                    dataset_id = str_id
-            except Exception:
-                logger.exception("Failed to parse dataset_id from positional args")
 
-        if dataset_id:
-            dataset_id = str(dataset_id)
-            dataset = db.session.scalar(
-                select(Dataset)
-                .where(
-                    Dataset.id == dataset_id,
-                    Dataset.tenant_id == api_token.tenant_id,
+def validate_dataset_token(
+    view: Callable[Concatenate[T, P], R] | None = None,
+) -> Callable[P, R] | Callable[[Callable[Concatenate[T, P], R]], Callable[P, R]]:
+    def decorator(view_func: Callable[Concatenate[T, P], R]) -> Callable[P, R]:
+        @wraps(view_func)
+        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+            api_token = validate_and_get_api_token("dataset")
+
+            # get url path dataset_id from positional args or kwargs
+            # Flask passes URL path parameters as positional arguments
+            dataset_id = None
+
+            # First try to get from kwargs (explicit parameter)
+            dataset_id = kwargs.get("dataset_id")
+
+            # If not in kwargs, try to extract from positional args
+            if not dataset_id and args:
+                # For class methods: args[0] is self, args[1] is dataset_id (if exists)
+                # Check if first arg is likely a class instance (has __dict__ or __class__)
+                if len(args) > 1 and hasattr(args[0], "__dict__"):
+                    # This is a class method, dataset_id should be in args[1]
+                    potential_id = args[1]
+                    # Validate it's a string-like UUID, not another object
+                    try:
+                        # Try to convert to string and check if it's a valid UUID format
+                        str_id = str(potential_id)
+                        # Basic check: UUIDs are 36 chars with hyphens
+                        if len(str_id) == 36 and str_id.count("-") == 4:
+                            dataset_id = str_id
+                    except Exception:
+                        logger.exception("Failed to parse dataset_id from class method args")
+                elif len(args) > 0:
+                    # Not a class method, check if args[0] looks like a UUID
+                    potential_id = args[0]
+                    try:
+                        str_id = str(potential_id)
+                        if len(str_id) == 36 and str_id.count("-") == 4:
+                            dataset_id = str_id
+                    except Exception:
+                        logger.exception("Failed to parse dataset_id from positional args")
+
+            # Validate dataset if dataset_id is provided
+            if dataset_id:
+                dataset_id = str(dataset_id)
+                dataset = db.session.scalar(
+                    select(Dataset)
+                    .where(
+                        Dataset.id == dataset_id,
+                        Dataset.tenant_id == api_token.tenant_id,
+                    )
+                    .limit(1)
                 )
-                .limit(1)
-            )
-            if not dataset:
-                raise NotFound("Dataset not found.")
-            if not dataset.enable_api:
-                raise Forbidden("Dataset api access is not enabled.")
-
-        tenant_account_join = db.session.execute(
-            select(Tenant, TenantAccountJoin)
-            .where(Tenant.id == api_token.tenant_id)
-            .where(TenantAccountJoin.tenant_id == Tenant.id)
-            .where(TenantAccountJoin.role.in_(["owner"]))
-            .where(Tenant.status == TenantStatus.NORMAL)
-        ).one_or_none()  # TODO: only owner information is required, so only one is returned.
-        if tenant_account_join:
-            tenant, ta = tenant_account_join
-            account = db.session.get(Account, ta.account_id)
-            # Login admin
-            if account:
-                account.current_tenant = tenant
-                current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+                if not dataset:
+                    raise NotFound("Dataset not found.")
+                if not dataset.enable_api:
+                    raise Forbidden("Dataset api access is not enabled.")
+            tenant_account_join = db.session.execute(
+                select(Tenant, TenantAccountJoin)
+                .where(Tenant.id == api_token.tenant_id)
+                .where(TenantAccountJoin.tenant_id == Tenant.id)
+                .where(TenantAccountJoin.role.in_(["owner"]))
+                .where(Tenant.status == TenantStatus.NORMAL)
+            ).one_or_none()  # TODO: only owner information is required, so only one is returned.
+            if tenant_account_join:
+                tenant, ta = tenant_account_join
+                account = db.session.get(Account, ta.account_id)
+                # Login admin
+                if account:
+                    account.current_tenant = tenant
+                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
+                    user_logged_in.send(current_app._get_current_object(), user=current_user)  # type: ignore
+                else:
+                    raise Unauthorized("Tenant owner account does not exist.")
             else:
-                raise Unauthorized("Tenant owner account does not exist.")
-        else:
-            raise Unauthorized("Tenant does not exist.")
+                raise Unauthorized("Tenant does not exist.")
+            return view_func(api_token.tenant_id, *args, **kwargs)  # type: ignore[arg-type]
 
-        if expects_bound_instance:
-            if not args:
-                raise TypeError("validate_dataset_token expected a bound resource instance.")
-            return view(args[0], api_token.tenant_id, *args[1:], **kwargs)
+        return decorated
 
-        return view(api_token.tenant_id, *args, **kwargs)
+    if view:
+        return decorator(view)
 
-    return decorated
+    # if view is None, it means that the decorator is used without parentheses
+    # use the decorator as a function for method_decorators
+    return decorator
 
 
 def validate_and_get_api_token(scope: str | None = None):

api/controllers/trigger/webhook.py

@@ -7,7 +7,7 @@ from werkzeug.exceptions import NotFound, RequestEntityTooLarge
 from controllers.trigger import bp
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
-from services.trigger.webhook_service import RawWebhookDataDict, WebhookService
+from services.trigger.webhook_service import WebhookService
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +23,6 @@ def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
         webhook_id, is_debug=is_debug
     )
 
-    webhook_data: RawWebhookDataDict
     try:
         # Use new unified extraction and validation
         webhook_data = WebhookService.extract_and_validate_webhook_data(webhook_trigger, node_config)

api/controllers/web/audio.py

@@ -3,11 +3,10 @@ import logging
 from flask import request
 from flask_restx import fields, marshal_with
 from graphon.model_runtime.errors.invoke import InvokeError
-from pydantic import field_validator
+from pydantic import BaseModel, field_validator
 from werkzeug.exceptions import InternalServerError
 
 import services
-from controllers.common.controller_schemas import TextToAudioPayload as TextToAudioPayloadBase
 from controllers.web import web_ns
 from controllers.web.error import (
     AppUnavailableError,
@@ -35,7 +34,12 @@ from services.errors.audio import (
 from ..common.schema import register_schema_models
 
 
-class TextToAudioPayload(TextToAudioPayloadBase):
+class TextToAudioPayload(BaseModel):
+    message_id: str | None = None
+    voice: str | None = None
+    text: str | None = None
+    streaming: bool | None = None
+
     @field_validator("message_id")
     @classmethod
     def validate_message_id(cls, value: str | None) -> str | None:

api/controllers/web/conversation.py

@@ -1,11 +1,10 @@
 from typing import Literal
 
 from flask import request
-from pydantic import BaseModel, Field, TypeAdapter, field_validator
+from pydantic import BaseModel, Field, TypeAdapter, field_validator, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import NotFound
 
-from controllers.common.controller_schemas import ConversationRenamePayload
 from controllers.common.schema import register_schema_models
 from controllers.web import web_ns
 from controllers.web.error import NotChatAppError
@@ -38,6 +37,18 @@ class ConversationListQuery(BaseModel):
         return uuid_value(value)
 
 
+class ConversationRenamePayload(BaseModel):
+    name: str | None = None
+    auto_generate: bool = False
+
+    @model_validator(mode="after")
+    def validate_name_requirement(self):
+        if not self.auto_generate:
+            if self.name is None or not self.name.strip():
+                raise ValueError("name is required when auto_generate is false")
+        return self
+
+
 register_schema_models(web_ns, ConversationListQuery, ConversationRenamePayload)
 
 

api/controllers/web/forgot_password.py

@@ -3,6 +3,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 controllers.common.schema import register_schema_models
@@ -18,15 +19,33 @@ from controllers.console.error import EmailSendIpLimitError
 from controllers.console.wraps import email_password_login_enabled, only_edition_enterprise, setup_required
 from controllers.web import web_ns
 from extensions.ext_database import db
-from libs.helper import extract_remote_ip
-from libs.password import hash_password
+from libs.helper import EmailStr, extract_remote_ip
+from libs.password import hash_password, valid_password
 from models.account import Account
 from services.account_service import AccountService
-from services.entities.auth_entities import (
-    ForgotPasswordCheckPayload,
-    ForgotPasswordResetPayload,
-    ForgotPasswordSendPayload,
-)
+
+
+class ForgotPasswordSendPayload(BaseModel):
+    email: EmailStr
+    language: str | None = None
+
+
+class ForgotPasswordCheckPayload(BaseModel):
+    email: EmailStr
+    code: str
+    token: str = Field(min_length=1)
+
+
+class ForgotPasswordResetPayload(BaseModel):
+    token: str = Field(min_length=1)
+    new_password: str
+    password_confirm: str
+
+    @field_validator("new_password", "password_confirm")
+    @classmethod
+    def validate_password(cls, value: str) -> str:
+        return valid_password(value)
+
 
 register_schema_models(web_ns, ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload)
 

api/controllers/web/login.py

@@ -29,11 +29,13 @@ from libs.token import (
 )
 from services.account_service import AccountService
 from services.app_service import AppService
-from services.entities.auth_entities import LoginPayloadBase
 from services.webapp_auth_service import WebAppAuthService
 
 
-class LoginPayload(LoginPayloadBase):
+class LoginPayload(BaseModel):
+    email: EmailStr
+    password: str
+
     @field_validator("password")
     @classmethod
     def validate_password(cls, value: str) -> str:

api/controllers/web/message.py

@@ -6,7 +6,6 @@ from graphon.model_runtime.errors.invoke import InvokeError
 from pydantic import BaseModel, Field, TypeAdapter, field_validator
 from werkzeug.exceptions import InternalServerError, NotFound
 
-from controllers.common.controller_schemas import MessageFeedbackPayload
 from controllers.common.schema import register_schema_models
 from controllers.web import web_ns
 from controllers.web.error import (
@@ -54,6 +53,11 @@ class MessageListQuery(BaseModel):
         return uuid_value(value)
 
 
+class MessageFeedbackPayload(BaseModel):
+    rating: Literal["like", "dislike"] | None = Field(default=None, description="Feedback rating")
+    content: str | None = Field(default=None, description="Feedback content")
+
+
 class MessageMoreLikeThisQuery(BaseModel):
     response_mode: Literal["blocking", "streaming"] = Field(
         description="Response mode",

api/controllers/web/passport.py

@@ -138,15 +138,12 @@ def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded:
     if not app_model or app_model.status != "normal" or not app_model.enable_site:
         raise NotFound()
 
-    match auth_type:
-        case WebAppAuthType.PUBLIC:
-            return _exchange_for_public_app_token(app_model, site, enterprise_user_decoded)
-        case WebAppAuthType.EXTERNAL:
-            if user_auth_type != "external":
-                raise WebAppAuthRequiredError("Please login as external user.")
-        case WebAppAuthType.INTERNAL:
-            if user_auth_type != "internal":
-                raise WebAppAuthRequiredError("Please login as internal user.")
+    if auth_type == WebAppAuthType.PUBLIC:
+        return _exchange_for_public_app_token(app_model, site, enterprise_user_decoded)
+    elif auth_type == WebAppAuthType.EXTERNAL and user_auth_type != "external":
+        raise WebAppAuthRequiredError("Please login as external user.")
+    elif auth_type == WebAppAuthType.INTERNAL and user_auth_type != "internal":
+        raise WebAppAuthRequiredError("Please login as internal user.")
 
     end_user = None
     if end_user_id:

api/controllers/web/saved_message.py

@@ -1,17 +1,27 @@
 from flask import request
-from pydantic import TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter
 from werkzeug.exceptions import NotFound
 
-from controllers.common.controller_schemas import SavedMessageCreatePayload, SavedMessageListQuery
 from controllers.common.schema import register_schema_models
 from controllers.web import web_ns
 from controllers.web.error import NotCompletionAppError
 from controllers.web.wraps import WebApiResource
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import SavedMessageInfiniteScrollPagination, SavedMessageItem
+from libs.helper import UUIDStrOrEmpty
 from services.errors.message import MessageNotExistsError
 from services.saved_message_service import SavedMessageService
 
+
+class SavedMessageListQuery(BaseModel):
+    last_id: UUIDStrOrEmpty | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class SavedMessageCreatePayload(BaseModel):
+    message_id: UUIDStrOrEmpty
+
+
 register_schema_models(web_ns, SavedMessageListQuery, SavedMessageCreatePayload)
 
 

api/controllers/web/workflow.py

@@ -1,10 +1,11 @@
 import logging
+from typing import Any
 
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
+from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError
 
-from controllers.common.controller_schemas import WorkflowRunPayload
 from controllers.common.schema import register_schema_models
 from controllers.web import web_ns
 from controllers.web.error import (
@@ -29,6 +30,12 @@ from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
 
+
+class WorkflowRunPayload(BaseModel):
+    inputs: dict[str, Any] = Field(description="Input variables for the workflow")
+    files: list[dict[str, Any]] | None = Field(default=None, description="Files to be processed by the workflow")
+
+
 logger = logging.getLogger(__name__)
 
 register_schema_models(web_ns, WorkflowRunPayload)

api/controllers/web/workflow_events.py

@@ -72,13 +72,12 @@ class WorkflowEventsApi(WebApiResource):
             app_mode = AppMode.value_of(app_model.mode)
             msg_generator = MessageGenerator()
             generator: BaseAppGenerator
-            match app_mode:
-                case AppMode.ADVANCED_CHAT:
-                    generator = AdvancedChatAppGenerator()
-                case AppMode.WORKFLOW:
-                    generator = WorkflowAppGenerator()
-                case _:
-                    raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
+            if app_mode == AppMode.ADVANCED_CHAT:
+                generator = AdvancedChatAppGenerator()
+            elif app_mode == AppMode.WORKFLOW:
+                generator = WorkflowAppGenerator()
+            else:
+                raise InvalidArgumentError(f"cannot subscribe to workflow run, workflow_run_id={workflow_run.id}")
 
             include_state_snapshot = request.args.get("include_state_snapshot", "false").lower() == "true"
 

api/controllers/web/wraps.py

@@ -1,7 +1,7 @@
 from collections.abc import Callable
 from datetime import UTC, datetime
 from functools import wraps
-from typing import Concatenate
+from typing import Concatenate, ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource
@@ -20,13 +20,14 @@ from services.enterprise.enterprise_service import EnterpriseService, WebAppSett
 from services.feature_service import FeatureService
 from services.webapp_auth_service import WebAppAuthService
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-def validate_jwt_token[**P, R](
-    view: Callable[Concatenate[App, EndUser, P], R] | None = None,
-) -> Callable[P, R] | Callable[[Callable[Concatenate[App, EndUser, P], R]], Callable[P, R]]:
-    def decorator(view: Callable[Concatenate[App, EndUser, P], R]) -> Callable[P, R]:
+
+def validate_jwt_token(view: Callable[Concatenate[App, EndUser, P], R] | None = None):
+    def decorator(view: Callable[Concatenate[App, EndUser, P], R]):
         @wraps(view)
-        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             app_model, end_user = decode_jwt_token()
             return view(app_model, end_user, *args, **kwargs)
 
@@ -37,7 +38,7 @@ def validate_jwt_token[**P, R](
     return decorator
 
 
-def decode_jwt_token(app_code: str | None = None, user_id: str | None = None) -> tuple[App, EndUser]:
+def decode_jwt_token(app_code: str | None = None, user_id: str | None = None):
     system_features = FeatureService.get_system_features()
     if not app_code:
         app_code = str(request.headers.get(HEADER_NAME_APP_CODE))

api/core/agent/cot_chat_agent_runner.py

@@ -79,18 +79,21 @@ class CotChatAgentRunner(CotAgentRunner):
         if not agent_scratchpad:
             assistant_messages = []
         else:
-            content = ""
+            assistant_message = AssistantPromptMessage(content="")
+            assistant_message.content = ""  # FIXME: type check tell mypy that assistant_message.content is str
             for unit in agent_scratchpad:
                 if unit.is_final():
-                    content += f"Final Answer: {unit.agent_response}"
+                    assert isinstance(assistant_message.content, str)
+                    assistant_message.content += f"Final Answer: {unit.agent_response}"
                 else:
-                    content += f"Thought: {unit.thought}\n\n"
+                    assert isinstance(assistant_message.content, str)
+                    assistant_message.content += f"Thought: {unit.thought}\n\n"
                     if unit.action_str:
-                        content += f"Action: {unit.action_str}\n\n"
+                        assistant_message.content += f"Action: {unit.action_str}\n\n"
                     if unit.observation:
-                        content += f"Observation: {unit.observation}\n\n"
+                        assistant_message.content += f"Observation: {unit.observation}\n\n"
 
-            assistant_messages = [AssistantPromptMessage(content=content)]
+            assistant_messages = [assistant_message]
 
         # query messages
         query_messages = self._organize_user_query(self._query, [])

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

@@ -5,10 +5,6 @@ from configs import dify_config
 from constants import DEFAULT_FILE_NUMBER_LIMITS
 
 
-class FeatureToggleDict(TypedDict):
-    enabled: bool
-
-
 class SystemParametersDict(TypedDict):
     image_file_size_limit: int
     video_file_size_limit: int
@@ -20,12 +16,12 @@ class SystemParametersDict(TypedDict):
 class AppParametersDict(TypedDict):
     opening_statement: str | None
     suggested_questions: list[str]
-    suggested_questions_after_answer: FeatureToggleDict
-    speech_to_text: FeatureToggleDict
-    text_to_speech: FeatureToggleDict
-    retriever_resource: FeatureToggleDict
-    annotation_reply: FeatureToggleDict
-    more_like_this: FeatureToggleDict
+    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]

api/core/app/app_config/entities.py

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
 
@@ -8,7 +9,6 @@ from graphon.variables.input_entities import VariableEntity as WorkflowVariableE
 from pydantic import BaseModel, Field
 
 from core.rag.data_post_processor.data_post_processor import RerankingModelDict, WeightsDict
-from core.rag.entities import MetadataFilteringCondition
 from models.model import AppMode
 
 
@@ -111,6 +111,31 @@ class ExternalDataVariableEntity(BaseModel):
     config: dict[str, Any] = Field(default_factory=dict)
 
 
+SupportedComparisonOperator = Literal[
+    # for string or array
+    "contains",
+    "not contains",
+    "start with",
+    "end with",
+    "is",
+    "is not",
+    "empty",
+    "not empty",
+    "in",
+    "not in",
+    # for number
+    "=",
+    "≠",
+    ">",
+    "<",
+    "≥",
+    "≤",
+    # for time
+    "before",
+    "after",
+]
+
+
 class ModelConfig(BaseModel):
     provider: str
     name: str
@@ -118,6 +143,25 @@ class ModelConfig(BaseModel):
     completion_params: dict[str, Any] = Field(default_factory=dict)
 
 
+class Condition(BaseModel):
+    """
+    Condition detail
+    """
+
+    name: str
+    comparison_operator: SupportedComparisonOperator
+    value: str | Sequence[str] | None | int | float = None
+
+
+class MetadataFilteringCondition(BaseModel):
+    """
+    Metadata Filtering Condition.
+    """
+
+    logical_operator: Literal["and", "or"] | None = "and"
+    conditions: list[Condition] | None = Field(default=None, deprecated=True)
+
+
 class DatasetRetrieveConfigEntity(BaseModel):
     """
     Dataset Retrieve Config Entity.

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, overload
+from typing import TYPE_CHECKING, Any, Literal, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -68,7 +68,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         workflow_run_id: str,
@@ -81,7 +81,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         workflow_run_id: str,
@@ -94,7 +94,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         workflow_run_id: str,
@@ -106,7 +106,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         workflow_run_id: str,
@@ -239,7 +239,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         conversation: Conversation,
         message: Message,
         application_generate_entity: AdvancedChatAppGenerateEntity,
@@ -271,9 +271,9 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         workflow: Workflow,
         node_id: str,
         user: Account | EndUser,
-        args: Mapping[str, Any],
+        args: Mapping,
         streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         """
         Generate App response.
 
@@ -359,7 +359,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         user: Account | EndUser,
         args: LoopNodeRunPayload,
         streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         """
         Generate App response.
 
@@ -439,7 +439,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         self,
         *,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         invoke_from: InvokeFrom,
         application_generate_entity: AdvancedChatAppGenerateEntity,
         workflow_execution_repository: WorkflowExecutionRepository,
@@ -451,7 +451,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         pause_state_config: PauseStateLayerConfig | None = None,
         graph_runtime_state: GraphRuntimeState | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
         """
         Generate App response.
 
@@ -653,10 +653,10 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         queue_manager: AppQueueManager,
         conversation: ConversationSnapshot,
         message: MessageSnapshot,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> ChatbotAppBlockingResponse | Generator[ChatbotAppStreamResponse, None, None]:
+    ) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
         """
         Handle response.
         :param application_generate_entity: application generate entity

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

@@ -10,7 +10,7 @@ from graphon.runtime import GraphRuntimeState, VariablePool
 from graphon.variable_loader import VariableLoader
 from graphon.variables.variables import Variable
 from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -363,7 +363,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         :return: List of conversation variables ready for use
         """
-        with sessionmaker(bind=db.engine).begin() as session:
+        with Session(db.engine) as session:
             existing_variables = self._load_existing_conversation_variables(session)
 
             if not existing_variables:
@@ -376,6 +376,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             # Convert to Variable objects for use in the workflow
             conversation_variables = [var.to_variable() for var in existing_variables]
 
+            session.commit()
             return cast(list[Variable], conversation_variables)
 
     def _load_existing_conversation_variables(self, session: Session) -> list[ConversationVariable]:

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

@@ -16,7 +16,7 @@ from graphon.model_runtime.utils.encoders import jsonable_encoder
 from graphon.nodes import BuiltinNodeTypes
 from graphon.runtime import GraphRuntimeState
 from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -328,8 +328,13 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     @contextmanager
     def _database_session(self):
         """Context manager for database sessions."""
-        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
-            yield session
+        with Session(db.engine, expire_on_commit=False) as session:
+            try:
+                yield session
+                session.commit()
+            except Exception:
+                session.rollback()
+                raise
 
     def _ensure_workflow_initialized(self):
         """Fluent validation for workflow state."""

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

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping
-from typing import Any, Literal, overload
+from typing import Any, Literal, Union, overload
 
 from flask import Flask, current_app
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -37,7 +37,7 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         self,
         *,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[False],
@@ -48,7 +48,7 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         self,
         *,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[True],
@@ -59,21 +59,21 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         self,
         *,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool,
-    ) -> Mapping | Generator[Mapping | str, None, None]: ...
+    ) -> Union[Mapping, Generator[Mapping | str, None, None]]: ...
 
     def generate(
         self,
         *,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = True,
-    ) -> Mapping | Generator[Mapping | str, None, None]:
+    ) -> Union[Mapping, Generator[Mapping | str, None, None]]:
         """
         Generate App response.
 

api/core/app/apps/base_app_generate_response_converter.py

@@ -107,13 +107,13 @@ class AppGenerateResponseConverter(ABC):
         return metadata
 
     @classmethod
-    def _error_to_stream_response(cls, e: Exception) -> dict[str, Any]:
+    def _error_to_stream_response(cls, e: Exception):
         """
         Error to stream response.
         :param e: exception
         :return:
         """
-        error_responses: dict[type[Exception], dict[str, Any]] = {
+        error_responses = {
             ValueError: {"code": "invalid_param", "status": 400},
             ProviderTokenNotInitError: {"code": "provider_not_initialize", "status": 400},
             QuotaExceededError: {
@@ -127,7 +127,7 @@ class AppGenerateResponseConverter(ABC):
         }
 
         # Determine the response based on the type of exception
-        data: dict[str, Any] | None = None
+        data = None
         for k, v in error_responses.items():
             if isinstance(e, k):
                 data = v

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

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping
-from typing import Any, Literal, overload
+from typing import Any, Literal, Union, overload
 
 from flask import Flask, copy_current_request_context, current_app
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -36,7 +36,7 @@ class ChatAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[True],
@@ -46,7 +46,7 @@ class ChatAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[False],
@@ -56,20 +56,20 @@ class ChatAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool,
-    ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]: ...
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]: ...
 
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]:
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
         """
         Generate App response.
 

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

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping
-from typing import Any, Literal, overload
+from typing import Any, Literal, Union, overload
 
 from flask import Flask, copy_current_request_context, current_app
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -36,7 +36,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[True],
@@ -46,7 +46,7 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[False],
@@ -56,20 +56,20 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = False,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]: ...
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]: ...
 
     def generate(
         self,
         app_model: App,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = True,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
 
@@ -244,10 +244,10 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         self,
         app_model: App,
         message_id: str,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         invoke_from: InvokeFrom,
         stream: bool = True,
-    ) -> Mapping | Generator[Mapping | str, None, None]:
+    ) -> Union[Mapping, Generator[Mapping | str, None, None]]:
         """
         Generate App response.
 

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

@@ -7,7 +7,7 @@ import threading
 import time
 import uuid
 from collections.abc import Generator, Mapping
-from typing import Any, Literal, cast, overload
+from typing import Any, Literal, Union, cast, overload
 
 from flask import Flask, current_app
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -62,7 +62,7 @@ class PipelineGenerator(BaseAppGenerator):
         *,
         pipeline: Pipeline,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[True],
@@ -77,7 +77,7 @@ class PipelineGenerator(BaseAppGenerator):
         *,
         pipeline: Pipeline,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[False],
@@ -92,28 +92,28 @@ class PipelineGenerator(BaseAppGenerator):
         *,
         pipeline: Pipeline,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
         workflow_thread_pool_id: str | None,
         is_retry: bool = False,
-    ) -> Mapping[str, Any] | Generator[Mapping | str, None, None]: ...
+    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
         self,
         *,
         pipeline: Pipeline,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
         workflow_thread_pool_id: str | None = None,
         is_retry: bool = False,
-    ) -> Mapping[str, Any] | Generator[Mapping | str, None, None] | None:
+    ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None], None]:
         # Add null check for dataset
 
         with Session(db.engine, expire_on_commit=False) as session:
@@ -278,7 +278,7 @@ class PipelineGenerator(BaseAppGenerator):
         context: contextvars.Context,
         pipeline: Pipeline,
         workflow_id: str,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         application_generate_entity: RagPipelineGenerateEntity,
         invoke_from: InvokeFrom,
         workflow_execution_repository: WorkflowExecutionRepository,
@@ -286,7 +286,7 @@ class PipelineGenerator(BaseAppGenerator):
         streaming: bool = True,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
         workflow_thread_pool_id: str | None = None,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
 
@@ -624,10 +624,10 @@ class PipelineGenerator(BaseAppGenerator):
         application_generate_entity: RagPipelineGenerateEntity,
         workflow: Workflow,
         queue_manager: AppQueueManager,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> WorkflowAppBlockingResponse | Generator[WorkflowAppStreamResponse, None, None]:
+    ) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
         """
         Handle response.
         :param application_generate_entity: application generate entity
@@ -668,7 +668,7 @@ class PipelineGenerator(BaseAppGenerator):
         datasource_info: Mapping[str, Any],
         created_from: str,
         position: int,
-        account: Account | EndUser,
+        account: Union[Account, EndUser],
         batch: str,
         document_form: str,
     ):
@@ -715,7 +715,7 @@ class PipelineGenerator(BaseAppGenerator):
         pipeline: Pipeline,
         workflow: Workflow,
         start_node_id: str,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
     ) -> list[Mapping[str, Any]]:
         """
         Format datasource info list.

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

@@ -2,6 +2,7 @@ import logging
 import time
 from typing import cast
 
+from graphon.entities import GraphInitParams
 from graphon.enums import WorkflowType
 from graphon.graph import Graph
 from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent
@@ -21,7 +22,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
-from core.workflow.node_factory import DifyGraphInitContext, DifyNodeFactory, get_default_root_node_id
+from core.workflow.node_factory import DifyNodeFactory, get_default_root_node_id
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
@@ -264,23 +265,22 @@ class PipelineRunner(WorkflowBasedAppRunner):
         # graph_config["nodes"] = real_run_nodes
         # graph_config["edges"] = real_edges
         # init graph
-        # Create explicit graph init context for Graph.init.
-        run_context = build_dify_run_context(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
-            user_id=self.application_generate_entity.user_id,
-            user_from=user_from,
-            invoke_from=invoke_from,
-        )
-        graph_init_context = DifyGraphInitContext(
+        # Create required parameters for Graph.init
+        graph_init_params = GraphInitParams(
             workflow_id=workflow.id,
             graph_config=graph_config,
-            run_context=run_context,
+            run_context=build_dify_run_context(
+                tenant_id=workflow.tenant_id,
+                app_id=self._app_id,
+                user_id=self.application_generate_entity.user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+            ),
             call_depth=0,
         )
 
-        node_factory = DifyNodeFactory.from_graph_init_context(
-            graph_init_context=graph_init_context,
+        node_factory = DifyNodeFactory(
+            graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
         if start_node_id is None:

api/core/app/apps/workflow/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, overload
+from typing import TYPE_CHECKING, Any, Literal, Union, overload
 
 from flask import Flask, current_app
 from graphon.graph_engine.layers import GraphEngineLayer
@@ -64,7 +64,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[True],
@@ -82,7 +82,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: Literal[False],
@@ -100,7 +100,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool,
@@ -110,14 +110,14 @@ class WorkflowAppGenerator(BaseAppGenerator):
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         pause_state_config: PauseStateLayerConfig | None = None,
-    ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]: ...
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]: ...
 
     def generate(
         self,
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         args: Mapping[str, Any],
         invoke_from: InvokeFrom,
         streaming: bool = True,
@@ -127,7 +127,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         root_node_id: str | None = None,
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         pause_state_config: PauseStateLayerConfig | None = None,
-    ) -> Mapping[str, Any] | Generator[Mapping[str, Any] | str, None, None]:
+    ) -> Union[Mapping[str, Any], Generator[Mapping[str, Any] | str, None, None]]:
         with self._bind_file_access_scope(tenant_id=app_model.tenant_id, user=user, invoke_from=invoke_from):
             files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -237,7 +237,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         application_generate_entity: WorkflowAppGenerateEntity,
         graph_runtime_state: GraphRuntimeState,
         workflow_execution_repository: WorkflowExecutionRepository,
@@ -245,7 +245,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         pause_state_config: PauseStateLayerConfig | None = None,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Resume a paused workflow execution using the persisted runtime state.
         """
@@ -269,7 +269,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         *,
         app_model: App,
         workflow: Workflow,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         application_generate_entity: WorkflowAppGenerateEntity,
         invoke_from: InvokeFrom,
         workflow_execution_repository: WorkflowExecutionRepository,
@@ -280,7 +280,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         graph_engine_layers: Sequence[GraphEngineLayer] = (),
         graph_runtime_state: GraphRuntimeState | None = None,
         pause_state_config: PauseStateLayerConfig | None = None,
-    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
+    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
 
@@ -609,10 +609,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         application_generate_entity: WorkflowAppGenerateEntity,
         workflow: Workflow,
         queue_manager: AppQueueManager,
-        user: Account | EndUser,
+        user: Union[Account, EndUser],
         draft_var_saver_factory: DraftVariableSaverFactory,
         stream: bool = False,
-    ) -> WorkflowAppBlockingResponse | Generator[WorkflowAppStreamResponse, None, None]:
+    ) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
         """
         Handle response.
         :param application_generate_entity: application generate entity

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

@@ -7,7 +7,7 @@ from typing import Union
 from graphon.entities import WorkflowStartReason
 from graphon.enums import WorkflowExecutionStatus
 from graphon.runtime import GraphRuntimeState
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -252,8 +252,13 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
     @contextmanager
     def _database_session(self):
         """Context manager for database sessions."""
-        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
-            yield session
+        with Session(db.engine, expire_on_commit=False) as session:
+            try:
+                yield session
+                session.commit()
+            except Exception:
+                session.rollback()
+                raise
 
     def _ensure_workflow_initialized(self):
         """Fluent validation for workflow state."""

api/core/app/apps/workflow_app_runner.py

@@ -3,6 +3,7 @@ import time
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
+from graphon.entities import GraphInitParams
 from graphon.entities.graph_config import NodeConfigDictAdapter
 from graphon.entities.pause_reason import HumanInputRequired
 from graphon.graph import Graph
@@ -65,13 +66,8 @@ from core.app.entities.queue_entities import (
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
 )
-from core.rag.entities import RetrievalSourceMetadata
-from core.workflow.node_factory import (
-    DifyGraphInitContext,
-    DifyNodeFactory,
-    get_default_root_node_id,
-    resolve_workflow_node_class,
-)
+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.system_variables import (
     build_bootstrap_variables,
     default_system_variables,
@@ -131,25 +127,24 @@ class WorkflowBasedAppRunner:
         if not isinstance(graph_config.get("edges"), list):
             raise ValueError("edges in workflow graph must be a list")
 
-        # Create explicit graph init context for Graph.init.
-        run_context = build_dify_run_context(
-            tenant_id=tenant_id or "",
-            app_id=self._app_id,
-            user_id=user_id,
-            user_from=user_from,
-            invoke_from=invoke_from,
-        )
-        graph_init_context = DifyGraphInitContext(
+        # Create required parameters for Graph.init
+        graph_init_params = GraphInitParams(
             workflow_id=workflow_id,
             graph_config=graph_config,
-            run_context=run_context,
+            run_context=build_dify_run_context(
+                tenant_id=tenant_id or "",
+                app_id=self._app_id,
+                user_id=user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
+            ),
             call_depth=0,
         )
 
         # Use the provided graph_runtime_state for consistent state management
 
-        node_factory = DifyNodeFactory.from_graph_init_context(
-            graph_init_context=graph_init_context,
+        node_factory = DifyNodeFactory(
+            graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
 
@@ -294,23 +289,22 @@ class WorkflowBasedAppRunner:
 
         typed_node_configs = [NodeConfigDictAdapter.validate_python(node) for node in node_configs]
 
-        # Create explicit graph init context for Graph.init.
-        run_context = build_dify_run_context(
-            tenant_id=workflow.tenant_id,
-            app_id=self._app_id,
-            user_id=user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.DEBUGGER,
-        )
-        graph_init_context = DifyGraphInitContext(
+        # Create required parameters for Graph.init
+        graph_init_params = GraphInitParams(
             workflow_id=workflow.id,
             graph_config=graph_config,
-            run_context=run_context,
+            run_context=build_dify_run_context(
+                tenant_id=workflow.tenant_id,
+                app_id=self._app_id,
+                user_id=user_id,
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+            ),
             call_depth=0,
         )
 
-        node_factory = DifyNodeFactory.from_graph_init_context(
-            graph_init_context=graph_init_context,
+        node_factory = DifyNodeFactory(
+            graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
 

api/core/app/entities/queue_entities.py

@@ -10,7 +10,7 @@ from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChun
 from pydantic import BaseModel, ConfigDict, Field
 
 from core.app.entities.agent_strategy import AgentStrategyInfo
-from core.rag.entities import RetrievalSourceMetadata
+from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 
 
 class QueueEvent(StrEnum):

api/core/app/entities/task_entities.py

@@ -9,7 +9,7 @@ from graphon.nodes.human_input.entities import FormInput, UserAction
 from pydantic import BaseModel, ConfigDict, Field
 
 from core.app.entities.agent_strategy import AgentStrategyInfo
-from core.rag.entities import RetrievalSourceMetadata
+from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 
 
 class AnnotationReplyAccount(BaseModel):

api/core/app/layers/pause_state_persist_layer.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Annotated, Literal, Self
+from typing import Annotated, Literal, Self, TypeAlias
 
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
@@ -27,7 +27,7 @@ class _AdvancedChatAppGenerateEntityWrapper(BaseModel):
     entity: AdvancedChatAppGenerateEntity
 
 
-type _GenerateEntityUnion = Annotated[
+_GenerateEntityUnion: TypeAlias = Annotated[
     _WorkflowGenerateEntityWrapper | _AdvancedChatAppGenerateEntityWrapper,
     Field(discriminator="type"),
 ]

api/core/app/llm/quota.py

@@ -1,6 +1,6 @@
 from graphon.model_runtime.entities.llm_entities import LLMUsage
 from sqlalchemy import update
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.entities.model_entities import ModelStatus
@@ -57,37 +57,37 @@ def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LL
             used_quota = 1
 
     if used_quota is not None and system_configuration.current_quota_type is not None:
-        match system_configuration.current_quota_type:
-            case ProviderQuotaType.TRIAL:
-                from services.credit_pool_service import CreditPoolService
+        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+            from services.credit_pool_service import CreditPoolService
 
-                CreditPoolService.check_and_deduct_credits(
-                    tenant_id=tenant_id,
-                    credits_required=used_quota,
-                )
-            case ProviderQuotaType.PAID:
-                from services.credit_pool_service import CreditPoolService
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+            )
+        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
+            from services.credit_pool_service import CreditPoolService
 
-                CreditPoolService.check_and_deduct_credits(
-                    tenant_id=tenant_id,
-                    credits_required=used_quota,
-                    pool_type="paid",
-                )
-            case ProviderQuotaType.FREE:
-                with sessionmaker(bind=db.engine).begin() as session:
-                    stmt = (
-                        update(Provider)
-                        .where(
-                            Provider.tenant_id == tenant_id,
-                            # TODO: Use provider name with prefix after the data migration.
-                            Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                            Provider.provider_type == ProviderType.SYSTEM.value,
-                            Provider.quota_type == system_configuration.current_quota_type,
-                            Provider.quota_limit > Provider.quota_used,
-                        )
-                        .values(
-                            quota_used=Provider.quota_used + used_quota,
-                            last_used=naive_utc_now(),
-                        )
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+                pool_type="paid",
+            )
+        else:
+            with Session(db.engine) as session:
+                stmt = (
+                    update(Provider)
+                    .where(
+                        Provider.tenant_id == tenant_id,
+                        # TODO: Use provider name with prefix after the data migration.
+                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                        Provider.provider_type == ProviderType.SYSTEM.value,
+                        Provider.quota_type == system_configuration.current_quota_type,
+                        Provider.quota_limit > Provider.quota_used,
                     )
-                    session.execute(stmt)
+                    .values(
+                        quota_used=Provider.quota_used + used_quota,
+                        last_used=naive_utc_now(),
+                    )
+                )
+                session.execute(stmt)
+                session.commit()

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -2,7 +2,7 @@ import logging
 import time
 from collections.abc import Generator
 from threading import Thread
-from typing import Any, cast
+from typing import Any, Union, cast
 
 from graphon.file import FileTransferMethod
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
@@ -12,7 +12,7 @@ from graphon.model_runtime.entities.message_entities import (
 )
 from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from sqlalchemy import select
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
@@ -72,12 +72,14 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
     """
 
     _task_state: EasyUITaskState
-    _application_generate_entity: ChatAppGenerateEntity | CompletionAppGenerateEntity | AgentChatAppGenerateEntity
+    _application_generate_entity: Union[ChatAppGenerateEntity, CompletionAppGenerateEntity, AgentChatAppGenerateEntity]
     _precomputed_event_type: StreamEvent | None = None
 
     def __init__(
         self,
-        application_generate_entity: ChatAppGenerateEntity | CompletionAppGenerateEntity | AgentChatAppGenerateEntity,
+        application_generate_entity: Union[
+            ChatAppGenerateEntity, CompletionAppGenerateEntity, AgentChatAppGenerateEntity
+        ],
         queue_manager: AppQueueManager,
         conversation: Conversation,
         message: Message,
@@ -115,11 +117,11 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
 
     def process(
         self,
-    ) -> (
-        ChatbotAppBlockingResponse
-        | CompletionAppBlockingResponse
-        | Generator[ChatbotAppStreamResponse | CompletionAppStreamResponse, None, None]
-    ):
+    ) -> Union[
+        ChatbotAppBlockingResponse,
+        CompletionAppBlockingResponse,
+        Generator[Union[ChatbotAppStreamResponse, CompletionAppStreamResponse], None, None],
+    ]:
         if self._application_generate_entity.app_config.app_mode != AppMode.COMPLETION:
             # start generate conversation name thread
             self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
@@ -134,7 +136,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
 
     def _to_blocking_response(
         self, generator: Generator[StreamResponse, None, None]
-    ) -> ChatbotAppBlockingResponse | CompletionAppBlockingResponse:
+    ) -> Union[ChatbotAppBlockingResponse, CompletionAppBlockingResponse]:
         """
         Process blocking response.
         :return:
@@ -146,7 +148,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                 extras = {"usage": self._task_state.llm_result.usage.model_dump()}
                 if self._task_state.metadata:
                     extras["metadata"] = self._task_state.metadata.model_dump()
-                response: ChatbotAppBlockingResponse | CompletionAppBlockingResponse
+                response: Union[ChatbotAppBlockingResponse, CompletionAppBlockingResponse]
                 if self._conversation_mode == AppMode.COMPLETION:
                     response = CompletionAppBlockingResponse(
                         task_id=self._application_generate_entity.task_id,
@@ -181,7 +183,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
 
     def _to_stream_response(
         self, generator: Generator[StreamResponse, None, None]
-    ) -> Generator[ChatbotAppStreamResponse | CompletionAppStreamResponse, None, None]:
+    ) -> Generator[Union[ChatbotAppStreamResponse, CompletionAppStreamResponse], None, None]:
         """
         To stream response.
         :return:
@@ -266,8 +268,9 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             event = message.event
 
             if isinstance(event, QueueErrorEvent):
-                with sessionmaker(bind=db.engine).begin() as session:
+                with Session(db.engine) as session:
                     err = self.handle_error(event=event, session=session, message_id=self._message_id)
+                    session.commit()
                 yield self.error_to_stream_response(err)
                 break
             elif isinstance(event, QueueStopEvent | QueueMessageEndEvent):
@@ -287,9 +290,10 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                         answer=output_moderation_answer
                     )
 
-                with sessionmaker(bind=db.engine).begin() as session:
+                with Session(db.engine) as session:
                     # Save message
                     self._save_message(session=session, trace_manager=trace_manager)
+                    session.commit()
                 message_end_resp = self._message_end_to_stream_response()
                 yield message_end_resp
             elif isinstance(event, QueueRetrieverResourcesEvent):
@@ -507,8 +511,8 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :return:
         """
         with Session(db.engine, expire_on_commit=False) as session:
-            agent_thought: MessageAgentThought | None = session.scalar(
-                select(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).limit(1)
+            agent_thought: MessageAgentThought | None = (
+                session.query(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).first()
             )
 
         if agent_thought:

api/core/app/task_pipeline/message_file_utils.py

@@ -40,44 +40,41 @@ def prepare_file_dict(message_file: MessageFile, upload_files_map: dict[str, Upl
     size = 0
     extension = ""
 
-    match message_file.transfer_method:
-        case FileTransferMethod.REMOTE_URL:
+    if message_file.transfer_method == FileTransferMethod.REMOTE_URL:
+        url = message_file.url
+        if message_file.url:
+            filename = message_file.url.split("/")[-1].split("?")[0]
+            if "." in filename:
+                extension = "." + filename.rsplit(".", 1)[1]
+    elif message_file.transfer_method == FileTransferMethod.LOCAL_FILE:
+        if upload_file:
+            url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
+            filename = upload_file.name
+            mime_type = upload_file.mime_type or "application/octet-stream"
+            size = upload_file.size or 0
+            extension = f".{upload_file.extension}" if upload_file.extension else ""
+        elif message_file.upload_file_id:
+            url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
+    elif message_file.transfer_method == FileTransferMethod.TOOL_FILE and message_file.url:
+        if message_file.url.startswith(("http://", "https://")):
             url = message_file.url
-            if message_file.url:
-                filename = message_file.url.split("/")[-1].split("?")[0]
-                if "." in filename:
-                    extension = "." + filename.rsplit(".", 1)[1]
-        case FileTransferMethod.LOCAL_FILE:
-            if upload_file:
-                url = file_helpers.get_signed_file_url(upload_file_id=str(upload_file.id))
-                filename = upload_file.name
-                mime_type = upload_file.mime_type or "application/octet-stream"
-                size = upload_file.size or 0
-                extension = f".{upload_file.extension}" if upload_file.extension else ""
-            elif message_file.upload_file_id:
-                url = file_helpers.get_signed_file_url(upload_file_id=str(message_file.upload_file_id))
-        case FileTransferMethod.TOOL_FILE if message_file.url:
-            if message_file.url.startswith(("http://", "https://")):
-                url = message_file.url
-                filename = message_file.url.split("/")[-1].split("?")[0]
-                if "." in filename:
-                    extension = "." + filename.rsplit(".", 1)[1]
-            else:
-                url_parts = message_file.url.split("/")
-                if url_parts:
-                    file_part = url_parts[-1].split("?")[0]
-                    if "." in file_part:
-                        tool_file_id, ext = file_part.rsplit(".", 1)
-                        extension = f".{ext}"
-                        if len(extension) > MAX_TOOL_FILE_EXTENSION_LENGTH:
-                            extension = ".bin"
-                    else:
-                        tool_file_id = file_part
+            filename = message_file.url.split("/")[-1].split("?")[0]
+            if "." in filename:
+                extension = "." + filename.rsplit(".", 1)[1]
+        else:
+            url_parts = message_file.url.split("/")
+            if url_parts:
+                file_part = url_parts[-1].split("?")[0]
+                if "." in file_part:
+                    tool_file_id, ext = file_part.rsplit(".", 1)
+                    extension = f".{ext}"
+                    if len(extension) > MAX_TOOL_FILE_EXTENSION_LENGTH:
                         extension = ".bin"
-                    url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
-                    filename = file_part
-        case FileTransferMethod.TOOL_FILE | FileTransferMethod.DATASOURCE_FILE:
-            pass
+                else:
+                    tool_file_id = file_part
+                    extension = ".bin"
+                url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
+                filename = file_part
 
     transfer_method_value = message_file.transfer_method.value
     remote_url = message_file.url if message_file.transfer_method == FileTransferMethod.REMOTE_URL else ""

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

@@ -5,13 +5,14 @@ This layer centralizes model-quota deduction outside node implementations.
 """
 
 import logging
-from typing import TYPE_CHECKING, cast, final, override
+from typing import TYPE_CHECKING, cast, final
 
 from graphon.enums import BuiltinNodeTypes
 from graphon.graph_engine.entities.commands import AbortCommand, CommandType
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphNodeEventBase, NodeRunSucceededEvent
 from graphon.nodes.base.node import Node
+from typing_extensions import override
 
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.app.llm import deduct_llm_quota, ensure_llm_quota_available