fix(web): localize workflow archive logs

# Conflicts:
#	api/openapi/markdown/console-openapi.md
#	packages/contracts/generated/api/console/orpc.gen.ts
#	web/app/components/header/account-setting/index.tsx
#	web/app/components/main-nav/components/account-section.tsx
#	web/contract/router.ts

.agents/skills/how-to-write-component/SKILL.md

@@ -24,6 +24,7 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 
 - Search before adding UI, hooks, helpers, query utilities, or styling patterns. Reuse existing base components, feature components, hooks, utilities, and design styles when they fit.
 - Follow Dify's CSS-first Tailwind v4 contract from `packages/dify-ui/README.md` and `packages/dify-ui/AGENTS.md`. Prefer design-system tokens, utilities, and radius mappings over generic Tailwind choices.
+- Preserve visible keyboard focus states on the final focusable element. Prefer styled `@langgenius/dify-ui/*` controls when available, because components such as `Button` and form/control primitives carry the standard Dify UI `focus-visible` styling. Do not assume every Dify UI export provides visual focus styles: headless anatomy parts and direct Base UI re-exports such as dialog/popover/tooltip/drawer triggers usually only provide behavior and semantics. When using native `button` / `a`, custom trigger `render` props, clickable rows, icon buttons, menu-like items, or direct trigger parts, verify the rendered focusable element has a visible focus state. If it does not, add the standard Dify UI focus style: `outline-hidden focus-visible:ring-2 focus-visible:ring-state-accent-solid`. Do not hide outlines without an equivalent visible `focus-visible` indicator. Component-specific focus styles should follow an existing styled primitive pattern or a concrete design constraint, not a new ad hoc style.
 - Group feature code by workflow, route, or ownership area with route-aligned names: components, hooks, local types, query helpers, atoms, constants, tests, and small utilities should live near the code that changes with them.
 - For each feature module, keep a module-local `README.md` as a boundary note. Start with the module name, a brief one-sentence description, then split dependencies into `Internal Modules` and `External Modules` sections; keep both sections and write `None.` when one category is empty. `Internal Modules` lists modules inside the same overall feature using paths from that feature root, such as `shared/domain/runtime-status`; `External Modules` lists project modules outside the feature using paths from the web root without a `web/` prefix, such as `app/components/base/skeleton`. Omit npm packages, workspace package dependencies, and whitelisted plumbing modules. Do not copy caller-relative import paths into the README.
 - Module README whitelist: `@/service/client`, `@/next/*`.
@@ -53,6 +54,7 @@ Use this as the component decision guide for Dify web. Existing code is referenc
 - Treat `useParams`, route args, and `nuqs` query state as framework-owned state. When atom logic needs those values, hydrate primitive atoms at the route or surface boundary, such as with `useHydrateAtoms(..., { dangerouslyForceHydrate: true })`; keep URL updates in the route/query-state APIs instead of write atoms.
 - Within a route-owned feature, choose one source for route identity. If route params are bridged into feature atoms, use that bridge consistently for route-derived queries and actions instead of also threading the same route id through page, tab, and section props.
 - For async work tied to atom state, use `atomWithQuery` or `atomWithMutation`; write atoms should update only the inputs that drive those atoms. This applies to pure frontend async work as well as network requests, so do not hand-roll loading/error/in-flight state with `useState` or `useRef` for atom-orchestrated async behavior. For component-owned remote work, use `useQuery` or `useMutation` directly.
+- `jotai-tanstack-query` query atoms do not support TanStack Query tracked properties. A component that reads `useAtomValue(queryAtom)` subscribes to the whole query result, even if it only accesses `data`, `isLoading`, or `isError`. Export field-specific derived atoms and have components read the exact fields they render; use `selectAtom(queryAtom, result => result.field)` for query-result fields so unchanged selections do not notify subscribers. Keep direct `useAtomValue(queryAtom)` only when the component or hook genuinely needs the full observer result.
 - Row-local async state belongs to the row owner unless it participates in a shared Jotai workflow or needs atom-scoped reset semantics.
 - Leave query and mutation atoms unscoped so they keep shared QueryClient cache and invalidation behavior. Scope resettable primitives and explicit hydration tuples; scope a derived atom only when every dependency should be private to that surface.
 - For scoped primitives that are always hydrated by `ScopeProvider`, prefer `atomWithLazy<T>(() => { throw new Error(...) })` when consumers should see a non-null type.

.github/scripts/check-hotfix-cherry-picks.sh

@@ -45,7 +45,7 @@ while IFS= read -r commit_sha; do
   )
 
   if [[ -z "$source_sha" ]]; then
-    error "Commit $commit_sha ($subject) is missing cherry-pick provenance. $REMEDIATION_HINT"
+    error "Commit $commit_sha ($subject) is missing cherry-pick provenance. $REMEDIATION_HINT If version differences prevent using git cherry-pick -x, manually add '(cherry picked from commit <sha>)' to the commit message."
     failed=1
     continue
   fi

.github/workflows/api-tests.yml

@@ -29,13 +29,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}
@@ -91,13 +91,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}
@@ -142,13 +142,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: "3.12"

.github/workflows/autofix.yml

@@ -20,7 +20,7 @@ jobs:
         run: echo "autofix.ci updates pull request branches, not merge group refs."
 
       - if: github.event_name != 'merge_group'
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Check Docker Compose inputs
         if: github.event_name != 'merge_group'
@@ -73,12 +73,12 @@ jobs:
             dify-agent/pyproject.toml
             dify-agent/uv.lock
       - if: github.event_name != 'merge_group'
-        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
         with:
           python-version: "3.11"
 
       - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
 
       - name: Generate Docker Compose
         if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'

.github/workflows/build-push.yml

@@ -97,7 +97,7 @@ jobs:
           echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV
 
       - name: Login to Docker Hub
-        uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4.2.0
+        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
@@ -107,7 +107,7 @@ jobs:
 
       - name: Extract metadata for Docker
         id: meta
-        uses: docker/metadata-action@80c7e94dd9b9319bd5eb7a0e0fe9291e23a2a2e9 # v6.1.0
+        uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
         with:
           images: ${{ env[matrix.image_name_env] }}
 
@@ -159,10 +159,10 @@ jobs:
             file: "docker/local-sandbox/Dockerfile"
     steps:
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4.1.0
+        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
 
       - name: Validate Docker image
-        uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7.2.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           push: false
           context: ${{ matrix.build_context }}
@@ -197,14 +197,14 @@ jobs:
           merge-multiple: true
 
       - name: Login to Docker Hub
-        uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4.2.0
+        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
         with:
           username: ${{ env.DOCKERHUB_USER }}
           password: ${{ env.DOCKERHUB_TOKEN }}
 
       - name: Extract metadata for Docker
         id: meta
-        uses: docker/metadata-action@80c7e94dd9b9319bd5eb7a0e0fe9291e23a2a2e9 # v6.1.0
+        uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
         with:
           images: ${{ env[matrix.image_name_env] }}
           tags: |

.github/workflows/cli-e2e.yml

@@ -79,7 +79,7 @@ jobs:
       ws2_app_id:                 ${{ steps.out.outputs.DIFY_E2E_WS2_APP_ID }}
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false
@@ -123,7 +123,7 @@ jobs:
         shell: bash
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false
@@ -170,7 +170,7 @@ jobs:
         shell: bash
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false
@@ -233,7 +233,7 @@ jobs:
         shell: bash
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false
@@ -295,7 +295,7 @@ jobs:
         shell: bash
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false
@@ -351,7 +351,7 @@ jobs:
         shell: bash
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v4
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false

.github/workflows/cli-edge.yml

@@ -23,7 +23,7 @@ jobs:
         working-directory: ./cli
     steps:
       - name: Checkout
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
           fetch-depth: 0

.github/workflows/cli-release.yml

@@ -35,7 +35,7 @@ jobs:
       dify_tag: ${{ steps.resolve.outputs.dify_tag }}
     steps:
       - name: Checkout
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -98,7 +98,7 @@ jobs:
       DIFY_TAG: ${{ needs.validate.outputs.dify_tag }}
     steps:
       - name: Checkout
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
           fetch-depth: 1

.github/workflows/cli-smoke.yml

@@ -24,7 +24,7 @@ jobs:
         shell: bash
     steps:
       - name: Checkout cli ref
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           ref: ${{ inputs.cli_ref || github.ref }}
           persist-credentials: false

.github/workflows/cli-tests.yml

@@ -30,7 +30,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 

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

@@ -13,13 +13,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: "3.12"
@@ -63,13 +63,13 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: "3.12"

.github/workflows/docker-build.yml

@@ -77,10 +77,10 @@ jobs:
             file: "web/Dockerfile"
     steps:
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4.1.0
+        uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
 
       - name: Build Docker Image
-        uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7.2.0
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
         with:
           push: false
           context: ${{ matrix.context }}

.github/workflows/hotfix-cherry-pick.yml

@@ -24,7 +24,7 @@ jobs:
     name: Require cherry-pick provenance
     runs-on: depot-ubuntu-24.04
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
 

.github/workflows/main-ci.yml

@@ -46,13 +46,17 @@ jobs:
       vdb-changed: ${{ steps.changes.outputs.vdb }}
       migration-changed: ${{ steps.changes.outputs.migration }}
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
-      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
         id: changes
         with:
           filters: |
             api:
               - 'api/**'
+              - 'scripts/check_no_new_getattr.py'
+              - 'scripts/ast_grep_rules/no_new_getattr.yml'
+              - '.github/workflows/style.yml'
+              - '.github/workflows/main-ci.yml'
               - '.github/workflows/api-tests.yml'
               - 'docker/.env.example'
               - 'docker/envs/middleware.env.example'
@@ -380,6 +384,8 @@ jobs:
     needs: pre_job
     if: needs.pre_job.outputs.should_skip != 'true'
     uses: ./.github/workflows/style.yml
+    with:
+      base-rev: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha }}
 
   vdb-tests-run:
     name: Run VDB Tests

.github/workflows/post-merge.yml

@@ -18,8 +18,8 @@ jobs:
     outputs:
       external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
-      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
         id: changes
         with:
           filters: |

.github/workflows/pyrefly-diff.yml

@@ -17,12 +17,12 @@ jobs:
       pull-requests: write
     steps:
       - name: Checkout PR branch
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
 
       - name: Setup Python & UV
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
 

.github/workflows/pyrefly-type-coverage-comment.yml

@@ -21,10 +21,10 @@ jobs:
     if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
     steps:
       - name: Checkout default branch (trusted code)
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
 
       - name: Setup Python & UV
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
 

.github/workflows/pyrefly-type-coverage.yml

@@ -17,12 +17,12 @@ jobs:
       pull-requests: write
     steps:
       - name: Checkout PR branch
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
 
       - name: Setup Python & UV
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
 

.github/workflows/style.yml

@@ -2,6 +2,10 @@ name: Style check
 
 on:
   workflow_call:
+    inputs:
+      base-rev:
+        required: true
+        type: string
 
 concurrency:
   group: style-${{ github.head_ref || github.run_id }}
@@ -19,7 +23,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
           fetch-depth: 0
@@ -33,10 +37,11 @@ jobs:
             scripts/check_no_new_getattr.py
             scripts/ast_grep_rules/no_new_getattr.yml
             .github/workflows/style.yml
+            .github/workflows/main-ci.yml
 
       - name: Setup UV and Python
         if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: false
           python-version: "3.12"
@@ -54,17 +59,9 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --project api --dev python api/dev/lint_response_contracts.py --fail-on-mismatch
 
-      - name: Fetch merge target ref for getattr guard
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: git fetch --no-tags --depth=1 origin +refs/heads/main:refs/remotes/origin/main
-
-      - name: Bind merge target branch for getattr guard
-        if: steps.changed-files.outputs.any_changed == 'true'
-        run: git show-ref --verify --quiet refs/heads/main || git branch main origin/main
-
       - name: Run No New Getattr Guard
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: uv run --project api python scripts/check_no_new_getattr.py --mode ci --merge-target main
+        run: uv run --project api python scripts/check_no_new_getattr.py --base-rev "${{ inputs.base-rev }}"
 
       - name: Run Type Checks
         if: steps.changed-files.outputs.any_changed == 'true'
@@ -88,7 +85,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -139,7 +136,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -168,7 +165,7 @@ jobs:
       - name: Restore ESLint cache
         if: steps.changed-files.outputs.any_changed == 'true'
         id: eslint-cache-restore
-        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: .eslintcache
           key: ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
@@ -185,7 +182,7 @@ jobs:
 
       - name: Save ESLint cache
         if: steps.changed-files.outputs.any_changed == 'true' && success() && steps.eslint-cache-restore.outputs.cache-hit != 'true'
-        uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
         with:
           path: .eslintcache
           key: ${{ steps.eslint-cache-restore.outputs.cache-primary-key }}
@@ -196,7 +193,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           persist-credentials: false
@@ -214,7 +211,7 @@ jobs:
             .editorconfig
 
       - name: Super-linter
-        uses: super-linter/super-linter/slim@9e863354e3ff62e0727d37183162c4a88873df41 # v8.6.0
+        uses: super-linter/super-linter/slim@4ce20838b8ab83717e78138c5b3a1407148e0918 # v8.7.0
         if: steps.changed-files.outputs.any_changed == 'true'
         env:
           BASH_SEVERITY: warning

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

@@ -24,7 +24,7 @@ jobs:
         working-directory: sdks/nodejs-client
 
     steps:
-      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 

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

@@ -40,7 +40,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
           token: ${{ secrets.GITHUB_TOKEN }}
@@ -158,7 +158,7 @@ jobs:
 
       - name: Run Claude Code for Translation Sync
         if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@806af32823ef69c8ef357086c573a902af641307 # v1.0.151
+        uses: anthropics/claude-code-action@558b1d6cab4085c7753fe402c10bef0fbb92ac7a # v1.0.165
         with:
           anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
           github_token: ${{ secrets.GITHUB_TOKEN }}

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

@@ -21,7 +21,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           fetch-depth: 0
 

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

@@ -24,7 +24,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -36,7 +36,7 @@ jobs:
           remove_tool_cache: true
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}

.github/workflows/vdb-tests.yml

@@ -21,7 +21,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -33,7 +33,7 @@ jobs:
           remove_tool_cache: true
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: ${{ matrix.python-version }}

.github/workflows/web-e2e.yml

@@ -25,7 +25,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -33,7 +33,7 @@ jobs:
         uses: ./.github/actions/setup-web
 
       - name: Setup UV and Python
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
+        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
         with:
           enable-cache: true
           python-version: "3.12"

.github/workflows/web-tests.yml

@@ -31,7 +31,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -64,7 +64,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -102,7 +102,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 
@@ -134,7 +134,7 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
         with:
           persist-credentials: false
 

.superpowers/sdd/hitl-timeout-semantics-impl-report.md

@@ -1,30 +0,0 @@
-# HITL timeout semantics implementation report
-
-## What changed
-
-- Updated `api/core/workflow/nodes/human_input/callback.py` so `DifyHITLCallback` now preserves Dify's timeout split at the boundary:
-  - `HumanInputFormStatus.TIMEOUT` returns the graphon timeout branch via `Expired(selected_handle="__timeout__", ...)`.
-  - `HumanInputFormStatus.EXPIRED` is treated as an invalid resume state and raises `AssertionError`.
-  - `HumanInputFormStatus.WAITING` with a past global deadline is treated as an invalid resume state and raises `AssertionError`.
-  - `HumanInputFormStatus.WAITING` with only the node-level deadline expired still returns the timeout branch.
-- Added `created_at` to `HumanInputFormEntity` and `_HumanInputFormEntityImpl` so the callback can compute the global deadline using Dify's shared `HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS` invariant.
-- Kept the submitted and pause flows unchanged.
-- Added focused unit coverage in `api/tests/unit_tests/core/workflow/test_human_input_callback.py` for:
-  - node timeout branch
-  - global expiration rejection
-  - waiting-form past node deadline timeout
-  - waiting-form past global deadline rejection
-
-## Verification
-
-- `uv run --project api pytest -o addopts='' api/tests/unit_tests/core/workflow/test_human_input_callback.py api/tests/unit_tests/core/workflow/nodes/human_input/test_human_input_form_filled_event.py -q`
-- `git diff --check`
-
-## Result
-
-- The focused test set is expected to pass with the new `created_at` boundary in place.
-- No unrelated files were modified.
-
-## Concerns
-
-- The callback now fails fast on invalid resume states by design. That is intentional, but any caller that previously relied on `EXPIRED` being mapped to the timeout branch will now see an assertion failure instead.

api/clients/agent_backend/__init__.py

@@ -18,6 +18,7 @@ from clients.agent_backend.errors import (
     AgentBackendValidationError,
 )
 from clients.agent_backend.event_adapter import (
+    AgentBackendAgentMessageDeltaInternalEvent,
     AgentBackendDeferredToolCallInternalEvent,
     AgentBackendInternalEvent,
     AgentBackendInternalEventType,
@@ -46,6 +47,11 @@ from clients.agent_backend.request_builder import (
     AgentBackendWorkflowNodeRunInput,
     redact_for_agent_backend_log,
 )
+from clients.agent_backend.session_cleanup import (
+    AgentBackendSessionCleanupPayload,
+    AgentBackendSessionCleanupResult,
+    cleanup_agent_backend_session,
+)
 
 __all__ = [
     "AGENT_SOUL_PROMPT_LAYER_ID",
@@ -57,6 +63,7 @@ __all__ = [
     "WORKFLOW_NODE_JOB_PROMPT_LAYER_ID",
     "WORKFLOW_USER_PROMPT_LAYER_ID",
     "AgentBackendAgentAppRunInput",
+    "AgentBackendAgentMessageDeltaInternalEvent",
     "AgentBackendDeferredToolCallInternalEvent",
     "AgentBackendError",
     "AgentBackendHTTPError",
@@ -73,6 +80,8 @@ __all__ = [
     "AgentBackendRunRequestBuilder",
     "AgentBackendRunStartedInternalEvent",
     "AgentBackendRunSucceededInternalEvent",
+    "AgentBackendSessionCleanupPayload",
+    "AgentBackendSessionCleanupResult",
     "AgentBackendStreamError",
     "AgentBackendStreamInternalEvent",
     "AgentBackendTransportError",
@@ -82,6 +91,7 @@ __all__ = [
     "FakeAgentBackendRunClient",
     "FakeAgentBackendScenario",
     "RuntimeLayerSpec",
+    "cleanup_agent_backend_session",
     "create_agent_backend_run_client",
     "extract_runtime_layer_specs",
     "redact_for_agent_backend_log",

api/clients/agent_backend/event_adapter.py

@@ -5,6 +5,9 @@ The adapter does not define a new cross-service event contract. It consumes
 workflow Agent Node maps to Graphon/AppQueue events. Deferred external tool calls
 remain Dify Agent ``run_succeeded`` payloads on the wire; API code turns them
 into an internal event so workflow pause/session handling stays local to API.
+Agent-message deltas are exposed as annotations on ``PydanticAIStreamRunEvent``
+so API code does not have to parse Pydantic AI stream-event internals to
+preserve streaming. The terminal answer remains the ``run_succeeded`` output.
 """
 
 from __future__ import annotations
@@ -32,6 +35,7 @@ class AgentBackendInternalEventType(StrEnum):
 
     RUN_STARTED = "run_started"
     STREAM_EVENT = "stream_event"
+    AGENT_MESSAGE_DELTA = "agent_message_delta"
     DEFERRED_TOOL_CALL = "deferred_tool_call"
     RUN_SUCCEEDED = "run_succeeded"
     RUN_FAILED = "run_failed"
@@ -61,6 +65,13 @@ class AgentBackendStreamInternalEvent(AgentBackendInternalEventBase):
     data: JsonValue
 
 
+class AgentBackendAgentMessageDeltaInternalEvent(AgentBackendInternalEventBase):
+    """API-internal agent-message delta emitted independently from raw stream events."""
+
+    type: Literal[AgentBackendInternalEventType.AGENT_MESSAGE_DELTA] = AgentBackendInternalEventType.AGENT_MESSAGE_DELTA
+    delta: str
+
+
 class AgentBackendRunSucceededInternalEvent(AgentBackendInternalEventBase):
     """API-internal terminal success event carrying final output and session state."""
 
@@ -99,6 +110,7 @@ class AgentBackendRunCancelledInternalEvent(AgentBackendInternalEventBase):
 type AgentBackendInternalEvent = Annotated[
     AgentBackendRunStartedInternalEvent
     | AgentBackendStreamInternalEvent
+    | AgentBackendAgentMessageDeltaInternalEvent
     | AgentBackendDeferredToolCallInternalEvent
     | AgentBackendRunSucceededInternalEvent
     | AgentBackendRunFailedInternalEvent
@@ -121,6 +133,14 @@ class AgentBackendRunEventAdapter:
                     )
                 ]
             case PydanticAIStreamRunEvent():
+                if event.agent_message_delta:
+                    return [
+                        AgentBackendAgentMessageDeltaInternalEvent(
+                            run_id=event.run_id,
+                            source_event_id=event.id,
+                            delta=event.agent_message_delta,
+                        )
+                    ]
                 data = cast(JsonValue, _EVENT_DATA_ADAPTER.dump_python(event.data, mode="json"))
                 event_kind = data.get("event_kind") if isinstance(data, dict) else None
                 return [

api/clients/agent_backend/request_builder.py

@@ -11,8 +11,9 @@ composition-driven.
 
 from __future__ import annotations
 
+import re
 from collections.abc import Mapping
-from typing import ClassVar
+from typing import ClassVar, Literal
 
 from agenton.compositor import CompositorSessionSnapshot
 from agenton.compositor.schemas import LayerSessionSnapshot
@@ -46,7 +47,6 @@ from dify_agent.protocol import (
     LayerExitSignals,
     RunComposition,
     RunLayerSpec,
-    RunPurpose,
     RuntimeLayerSpec,
 )
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator
@@ -63,6 +63,7 @@ DIFY_CORE_TOOLS_LAYER_ID = "core_tools"
 DIFY_KNOWLEDGE_BASE_LAYER_ID = "knowledge"
 DIFY_ASK_HUMAN_LAYER_ID = "ask_human"
 DIFY_SHELL_LAYER_ID = "shell"
+type AgentConfigVersionKind = Literal["snapshot", "draft", "build_draft"]
 
 
 def _filter_snapshot_to_specs(
@@ -104,6 +105,59 @@ def _shell_config_with_drive_ref(
     return config.model_copy(update={"agent_stub_drive_ref": drive_config.drive_ref})
 
 
+def _markdown_backtick_fence(text: str) -> str:
+    """Choose a fence that will not terminate inside the prompt body."""
+    longest_backtick_run = max((len(match.group(0)) for match in re.finditer(r"`+", text)), default=0)
+    return "`" * max(3, longest_backtick_run + 1)
+
+
+_BUILD_DRAFT_AGENT_SOUL_PROMPT = """You are running in build mode.
+
+Objective:
+- Improve this agent's working environment, configuration, tools, files, notes,
+  and context so it can handle the intended task well.
+
+Guidance:
+- Treat the intended task as context for setup work, validation, and configuration decisions.
+- Perform concrete investigative or setup steps when they help improve or verify the agent configuration.
+- Use the installed `dify-agent` CLI when you need to inspect or persist Agent configuration."""
+
+
+def _wrap_build_draft_agent_soul_prompt(prompt: str | None) -> str:
+    """Reframe build-draft Agent Soul prompts as preparation work for a future run."""
+    prompt_body = (prompt or "").strip()
+    if not prompt_body:
+        return _BUILD_DRAFT_AGENT_SOUL_PROMPT + "\n\nIntended task for later normal runs:\nNo task prompt was provided."
+    fence = _markdown_backtick_fence(prompt_body)
+    return (
+        _BUILD_DRAFT_AGENT_SOUL_PROMPT
+        + f"\n\nIntended task for later normal runs:\n{fence}text\n{prompt_body}\n{fence}"
+    )
+
+
+def _agent_soul_prompt_for_layer(
+    prompt: str | None,
+    *,
+    config_version_kind: AgentConfigVersionKind,
+) -> str | None:
+    """Preserve normal snapshot/draft prompts and only wrap build-draft prompts.
+
+    The API-side layer adapter is the product boundary where Agent Soul text
+    becomes the model-facing system-prompt layer. ``snapshot`` and normal
+    ``draft`` runs pass through the original effective prompt unchanged, while
+    ``build_draft`` always emits a setup prompt. When an original prompt is
+    present, it is reframed as future-run context and embedded in a fenced
+    block; when it is blank, the setup instruction is still kept.
+    """
+    if config_version_kind != "build_draft":
+        if prompt is None:
+            return None
+        if not prompt.strip():
+            return None
+        return prompt
+    return _wrap_build_draft_agent_soul_prompt(prompt)
+
+
 class AgentBackendModelConfig(BaseModel):
     """API-side model/plugin selection before it is converted to Dify Agent layers."""
 
@@ -163,7 +217,7 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
     workflow_node_job_prompt: str
     user_prompt: str
     agent_soul_prompt: str | None = None
-    purpose: RunPurpose = "workflow_node"
+    agent_config_version_kind: AgentConfigVersionKind = "snapshot"
     idempotency_key: str | None = None
     output: AgentBackendOutputConfig | None = None
     tools: DifyPluginToolsLayerConfig | None = None
@@ -212,7 +266,7 @@ class AgentBackendAgentAppRunInput(BaseModel):
     execution_context: DifyExecutionContextLayerConfig
     user_prompt: str
     agent_soul_prompt: str | None = None
-    purpose: RunPurpose = "agent_app"
+    agent_config_version_kind: AgentConfigVersionKind = "snapshot"
     idempotency_key: str | None = None
     output: AgentBackendOutputConfig | None = None
     tools: DifyPluginToolsLayerConfig | None = None
@@ -261,13 +315,17 @@ class AgentBackendRunRequestBuilder:
         prompt.
         """
         layers: list[RunLayerSpec] = []
-        if run_input.agent_soul_prompt:
+        agent_soul_prompt = _agent_soul_prompt_for_layer(
+            run_input.agent_soul_prompt,
+            config_version_kind=run_input.agent_config_version_kind,
+        )
+        if agent_soul_prompt:
             layers.append(
                 RunLayerSpec(
                     name=AGENT_SOUL_PROMPT_LAYER_ID,
                     type=PLAIN_PROMPT_LAYER_TYPE_ID,
                     metadata={**run_input.metadata, "origin": "agent_soul"},
-                    config=PromptLayerConfig(prefix=run_input.agent_soul_prompt),
+                    config=PromptLayerConfig(prefix=agent_soul_prompt),
                 )
             )
 
@@ -419,7 +477,6 @@ class AgentBackendRunRequestBuilder:
 
         return CreateRunRequest(
             composition=RunComposition(layers=layers),
-            purpose=run_input.purpose,
             idempotency_key=run_input.idempotency_key,
             metadata=run_input.metadata,
             session_snapshot=run_input.session_snapshot,
@@ -467,7 +524,6 @@ class AgentBackendRunRequestBuilder:
         filtered_snapshot = _filter_snapshot_to_specs(session_snapshot, runtime_layer_specs)
         return CreateRunRequest(
             composition=RunComposition(layers=layers),
-            purpose="workflow_node",
             idempotency_key=idempotency_key,
             metadata=request_metadata,
             session_snapshot=filtered_snapshot,
@@ -483,13 +539,17 @@ class AgentBackendRunRequestBuilder:
         ask_human / structured output.
         """
         layers: list[RunLayerSpec] = []
-        if run_input.agent_soul_prompt:
+        agent_soul_prompt = _agent_soul_prompt_for_layer(
+            run_input.agent_soul_prompt,
+            config_version_kind=run_input.agent_config_version_kind,
+        )
+        if agent_soul_prompt:
             layers.append(
                 RunLayerSpec(
                     name=AGENT_SOUL_PROMPT_LAYER_ID,
                     type=PLAIN_PROMPT_LAYER_TYPE_ID,
                     metadata={**run_input.metadata, "origin": "agent_soul"},
-                    config=PromptLayerConfig(prefix=run_input.agent_soul_prompt),
+                    config=PromptLayerConfig(prefix=agent_soul_prompt),
                 )
             )
 
@@ -649,7 +709,6 @@ class AgentBackendRunRequestBuilder:
 
         return CreateRunRequest(
             composition=RunComposition(layers=layers),
-            purpose=run_input.purpose,
             idempotency_key=run_input.idempotency_key,
             metadata=run_input.metadata,
             session_snapshot=run_input.session_snapshot,

api/clients/agent_backend/session_cleanup.py

@@ -0,0 +1,100 @@
+"""Shared API-side helper for Agent backend lifecycle-only session cleanup.
+
+Product code owns local row retirement and background-task dispatch. This module
+only adapts persisted cleanup inputs into the public ``dify-agent`` run
+protocol, performs the synchronous ``create_run + wait_run`` loop used by Celery
+workers, and reports whether the backend cleanup succeeded, was skipped, or
+failed.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import ClassVar, Literal
+
+from agenton.compositor import CompositorSessionSnapshot
+from dify_agent.protocol import RuntimeLayerSpec
+from pydantic import BaseModel, ConfigDict, Field, JsonValue
+
+from clients.agent_backend.client import AgentBackendRunClient
+from clients.agent_backend.errors import AgentBackendError
+from clients.agent_backend.request_builder import AgentBackendRunRequestBuilder
+
+
+class AgentBackendSessionCleanupPayload(BaseModel):
+    """Serialized cleanup inputs preserved across API and Celery boundaries."""
+
+    session_snapshot: CompositorSessionSnapshot | None = None
+    runtime_layer_specs: list[RuntimeLayerSpec] = Field(default_factory=list)
+    idempotency_key: str | None = None
+    metadata: dict[str, JsonValue] = Field(default_factory=dict)
+    timeout_seconds: float = 30.0
+
+    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
+
+
+@dataclass(frozen=True, slots=True)
+class AgentBackendSessionCleanupResult:
+    """Terminal outcome of one backend cleanup attempt."""
+
+    status: Literal["succeeded", "skipped", "failed"]
+    reason: str | None = None
+    cleanup_run_id: str | None = None
+
+    @classmethod
+    def succeeded(cls, cleanup_run_id: str) -> AgentBackendSessionCleanupResult:
+        return cls(status="succeeded", cleanup_run_id=cleanup_run_id)
+
+    @classmethod
+    def skipped(cls, reason: str) -> AgentBackendSessionCleanupResult:
+        return cls(status="skipped", reason=reason)
+
+    @classmethod
+    def failed(cls, reason: str, cleanup_run_id: str | None = None) -> AgentBackendSessionCleanupResult:
+        return cls(status="failed", reason=reason, cleanup_run_id=cleanup_run_id)
+
+
+def cleanup_agent_backend_session(
+    *,
+    payload: AgentBackendSessionCleanupPayload,
+    client: AgentBackendRunClient | None,
+    request_builder: AgentBackendRunRequestBuilder | None = None,
+) -> AgentBackendSessionCleanupResult:
+    """Run lifecycle-only cleanup against the Agent backend and report status."""
+    if client is None:
+        return AgentBackendSessionCleanupResult.skipped("no_agent_backend_client")
+    if payload.session_snapshot is None:
+        return AgentBackendSessionCleanupResult.skipped("missing_session_snapshot")
+    if not payload.runtime_layer_specs:
+        return AgentBackendSessionCleanupResult.skipped("missing_runtime_layer_specs")
+
+    builder = request_builder or AgentBackendRunRequestBuilder()
+    request = builder.build_cleanup_request(
+        session_snapshot=payload.session_snapshot,
+        runtime_layer_specs=payload.runtime_layer_specs,
+        idempotency_key=payload.idempotency_key,
+        metadata=payload.metadata,
+    )
+
+    try:
+        response = client.create_run(request)
+    except AgentBackendError as exc:
+        return AgentBackendSessionCleanupResult.failed(str(exc))
+
+    try:
+        status_response = client.wait_run(response.run_id, timeout_seconds=payload.timeout_seconds)
+    except AgentBackendError as exc:
+        return AgentBackendSessionCleanupResult.failed(str(exc), cleanup_run_id=response.run_id)
+
+    if status_response.status != "succeeded":
+        reason = status_response.error or f"cleanup run ended with status {status_response.status}"
+        return AgentBackendSessionCleanupResult.failed(reason, cleanup_run_id=response.run_id)
+
+    return AgentBackendSessionCleanupResult.succeeded(response.run_id)
+
+
+__all__ = [
+    "AgentBackendSessionCleanupPayload",
+    "AgentBackendSessionCleanupResult",
+    "cleanup_agent_backend_session",
+]

api/commands/__init__.py

@@ -26,6 +26,7 @@ from .rbac import migrate_dataset_permissions_to_rbac, migrate_member_roles_to_r
 from .retention import (
     archive_workflow_runs,
     archive_workflow_runs_plan,
+    backfill_workflow_run_archive_bundles,
     clean_expired_messages,
     clean_workflow_runs,
     cleanup_orphaned_draft_variables,
@@ -54,6 +55,7 @@ __all__ = [
     "archive_workflow_runs",
     "archive_workflow_runs_plan",
     "backfill_plugin_auto_upgrade",
+    "backfill_workflow_run_archive_bundles",
     "clean_expired_messages",
     "clean_workflow_runs",
     "cleanup_orphaned_draft_variables",

api/commands/account.py

@@ -25,7 +25,7 @@ def reset_password(email, new_password, password_confirm):
         return
     normalized_email = email.strip().lower()
 
-    account = AccountService.get_account_by_email_with_case_fallback(db.session, email.strip())
+    account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
 
     if not account:
         click.echo(click.style(f"Account not found for email: {email}", fg="red"))
@@ -67,7 +67,7 @@ def reset_email(email, new_email, email_confirm):
         return
     normalized_new_email = new_email.strip().lower()
 
-    account = AccountService.get_account_by_email_with_case_fallback(db.session, email.strip())
+    account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
 
     if not account:
         click.echo(click.style(f"Account not found for email: {email}", fg="red"))
@@ -133,9 +133,9 @@ def create_tenant(email: str, language: str | None = None, name: str | None = No
         password=new_password,
         language=language,
         create_workspace_required=False,
-        session=db.session,
+        session=db.session(),
     )
-    TenantService.create_owner_tenant_if_not_exist(account, name, session=db.session)
+    TenantService.create_owner_tenant_if_not_exist(account, name, session=db.session())
 
     click.echo(
         click.style(

api/commands/data_migration.py

@@ -9,6 +9,7 @@ from uuid import UUID
 import click
 import sqlalchemy as sa
 import yaml
+from sqlalchemy.orm import Session
 
 from core.db.session_factory import session_factory
 from extensions.ext_database import db
@@ -108,7 +109,7 @@ def export_migration_data(input_file: str | None, output_file: str | None, overw
         raw_config = _load_json_object(input_file, "Export config")
         selection = ExportConfigParser().parse(raw_config)
         with session_factory.create_session() as session:
-            result = MigrationExportService().export(session, selection)
+            result = MigrationExportService().export(selection, session=session)
         MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
         click.echo(click.style(f"Output written to {output_file}", fg="green"))
         _render_report(result.report_items, context=_with_output_path(result.report_context, output_file))
@@ -157,7 +158,6 @@ def import_migration_data(
         package = MigrationPackageService().load_package(input_file)
         with session_factory.create_session() as session:
             result = MigrationImportService().import_package(
-                session,
                 ImportRequest(
                     package=package,
                     cli_target_tenant=target_tenant,
@@ -169,6 +169,7 @@ def import_migration_data(
                         create_app_api_token_on_import=create_app_api_token_on_import,
                     ),
                 ),
+                session=session,
             )
         _render_report(result.report_items, context=result.report_context)
     except MigrationDataError as exc:
@@ -217,7 +218,9 @@ def migration_data_wizard() -> None:
             default=True,
             show_default=False,
         )
-        auto_tools = _discover_auto_tools([app for app in apps if app.id in set(app_ids)], include_referenced_tools)
+        auto_tools = _discover_auto_tools(
+            [app for app in apps if app.id in set(app_ids)], include_referenced_tools, session=db.session()
+        )
         auto_tools = _resolve_auto_tool_names(tenant.id, auto_tools)
         _print_auto_tools(auto_tools)
         additional_tools = _prompt_additional_tools(tenant.id, auto_tools)
@@ -253,7 +256,7 @@ def migration_data_wizard() -> None:
             output_file=output_file,
         )
         with session_factory.create_session() as session:
-            result = MigrationExportService().export(session, selection)
+            result = MigrationExportService().export(selection, session=session)
         MigrationPackageService().save_package(result.package, output_file, overwrite=overwrite)
         click.echo(click.style(f"Output written to {output_file}", fg="green"))
         _print_wizard_step("Report")
@@ -394,13 +397,13 @@ def _prompt_import_options() -> tuple[bool, bool, str, str]:
     return include_secrets, create_tokens, id_strategy, conflict_strategy
 
 
-def _discover_auto_tools(apps: list[App], include_referenced_tools: bool) -> WizardToolMap:
+def _discover_auto_tools(apps: list[App], include_referenced_tools: bool, *, session: Session) -> WizardToolMap:
     auto_tools: WizardToolMap = {"api_tools": {}, "workflow_tools": {}, "mcp_tools": {}}
     if not include_referenced_tools:
         return auto_tools
     discovery_service = DependencyDiscoveryService()
     for app in apps:
-        dsl_content = AppDslService.export_dsl(app_model=app, include_secret=False)
+        dsl_content = AppDslService.export_dsl(app_model=app, session=session, include_secret=False)
         raw_dsl = yaml.safe_load(dsl_content) if dsl_content else {}
         dsl = raw_dsl if isinstance(raw_dsl, dict) else {}
         for dependency in discovery_service.discover_from_dsl(dsl):

api/commands/plugin.py

@@ -16,7 +16,7 @@ from core.plugin.plugin_service import PluginService
 from core.tools.utils.system_encryption import encrypt_system_params
 from extensions.ext_database import db
 from models import Tenant
-from models.account import TenantPluginAutoUpgradeStrategy
+from models.account import TenantPluginAutoUpgradeCategory, TenantPluginAutoUpgradeStrategy
 from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider_ids import DatasourceProviderID, ToolProviderID
 from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
@@ -188,13 +188,13 @@ def transform_datasource_credentials(environment: str):
         firecrawl_plugin_id = "langgenius/firecrawl_datasource"
         jina_plugin_id = "langgenius/jina_datasource"
         if environment == "online":
-            notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id)
-            firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id)
-            jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id)
+            notion_package_identifier = plugin_migration._fetch_latest_package_identifier(notion_plugin_id)
+            firecrawl_package_identifier = plugin_migration._fetch_latest_package_identifier(firecrawl_plugin_id)
+            jina_package_identifier = plugin_migration._fetch_latest_package_identifier(jina_plugin_id)
         else:
-            notion_plugin_unique_identifier = None
-            firecrawl_plugin_unique_identifier = None
-            jina_plugin_unique_identifier = None
+            notion_package_identifier = None
+            firecrawl_package_identifier = None
+            jina_package_identifier = None
         oauth_credential_type = CredentialType.OAUTH2
         api_key_credential_type = CredentialType.API_KEY
 
@@ -219,9 +219,9 @@ def transform_datasource_credentials(environment: str):
                     installed_plugins = installer_manager.list_plugins(tenant_id)
                     installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
                     if notion_plugin_id not in installed_plugins_ids:
-                        if notion_plugin_unique_identifier:
+                        if notion_package_identifier:
                             # install notion plugin
-                            PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier])
+                            PluginService.install_from_marketplace_pkg(tenant_id, [notion_package_identifier])
                     auth_count = 0
                     for notion_tenant_credential in notion_tenant_credentials:
                         auth_count += 1
@@ -279,9 +279,9 @@ def transform_datasource_credentials(environment: str):
                     installed_plugins = installer_manager.list_plugins(tenant_id)
                     installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
                     if firecrawl_plugin_id not in installed_plugins_ids:
-                        if firecrawl_plugin_unique_identifier:
+                        if firecrawl_package_identifier:
                             # install firecrawl plugin
-                            PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier])
+                            PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_package_identifier])
 
                     auth_count = 0
                     for firecrawl_tenant_credential in firecrawl_tenant_credentials:
@@ -343,10 +343,10 @@ def transform_datasource_credentials(environment: str):
                     installed_plugins = installer_manager.list_plugins(tenant_id)
                     installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins]
                     if jina_plugin_id not in installed_plugins_ids:
-                        if jina_plugin_unique_identifier:
+                        if jina_package_identifier:
                             # install jina plugin
-                            logger.debug("Installing Jina plugin %s", jina_plugin_unique_identifier)
-                            PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier])
+                            logger.debug("Installing Jina plugin %s", jina_package_identifier)
+                            PluginService.install_from_marketplace_pkg(tenant_id, [jina_package_identifier])
 
                     auth_count = 0
                     for jina_tenant_credential in jina_tenant_credentials:
@@ -406,7 +406,7 @@ def migrate_data_for_plugin():
 
 
 def _candidate_auto_upgrade_strategy_tenant_ids_stmt(limit: int | None = None):
-    category_count = len(TenantPluginAutoUpgradeStrategy.PluginCategory)
+    category_count = len(TenantPluginAutoUpgradeCategory)
     stmt = (
         select(TenantPluginAutoUpgradeStrategy.tenant_id)
         .group_by(TenantPluginAutoUpgradeStrategy.tenant_id)
@@ -472,6 +472,7 @@ def backfill_plugin_auto_upgrade(
         try:
             result = PluginAutoUpgradeService.backfill_strategy_categories(
                 current_tenant_id,
+                session=db.session(),
             )
         except Exception as e:
             failed_count += 1

api/commands/rbac.py

@@ -6,6 +6,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 
 import click
 from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.db.session_factory import session_factory
@@ -131,16 +132,35 @@ def _replace_member_role(
     operator_account_id: str,
     member_account_id: str,
     role_id: str,
+    *,
+    session: Session,
 ) -> str:
     RBACService.MemberRoles.replace(
         tenant_id=tenant_id,
         account_id=operator_account_id,
         member_account_id=member_account_id,
         role_ids=[role_id],
+        session=session,
     )
     return member_account_id
 
 
+def _replace_member_role_with_new_session(
+    tenant_id: str,
+    operator_account_id: str,
+    member_account_id: str,
+    role_id: str,
+) -> str:
+    with session_factory.create_session() as session:
+        return _replace_member_role(
+            tenant_id=tenant_id,
+            operator_account_id=operator_account_id,
+            member_account_id=member_account_id,
+            role_id=role_id,
+            session=session,
+        )
+
+
 @click.command(
     "rbac-migrate-member-roles", help="Migrate legacy workspace member roles into RBAC member-role bindings."
 )
@@ -217,14 +237,21 @@ def migrate_member_roles_to_rbac(
 
         if replace_jobs:
             if workers == 1:
-                for member_account_id, resolved_role_id in replace_jobs:
-                    _replace_member_role(workspace_id, owner_account_id, member_account_id, resolved_role_id)
-                    migrated_count += 1
+                with session_factory.create_session() as session:
+                    for member_account_id, resolved_role_id in replace_jobs:
+                        _replace_member_role(
+                            workspace_id,
+                            owner_account_id,
+                            member_account_id,
+                            resolved_role_id,
+                            session=session,
+                        )
+                        migrated_count += 1
             else:
                 with ThreadPoolExecutor(max_workers=workers) as executor:
                     futures = [
                         executor.submit(
-                            _replace_member_role,
+                            _replace_member_role_with_new_session,
                             workspace_id,
                             owner_account_id,
                             member_account_id,

api/commands/retention.py

@@ -56,6 +56,15 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
     return sorted(set(parsed))
 
 
+def _parse_comma_separated_ids(raw_ids: str | None, *, param_name: str) -> list[str] | None:
+    if raw_ids is None:
+        return None
+    parsed = sorted({raw_id.strip() for raw_id in raw_ids.split(",") if raw_id.strip()})
+    if not parsed:
+        raise click.BadParameter(f"{param_name} must not be empty")
+    return parsed
+
+
 def _get_archive_candidate_tenant_ids_by_prefix(
     prefix: str,
     *,
@@ -639,6 +648,82 @@ def archive_workflow_runs(
     )
 
 
+@click.command(
+    "backfill-workflow-run-archive-bundles",
+    help="Backfill workflow-run archive bundle DB index from object-storage manifests.",
+)
+@click.option("--tenant-ids", default=None, help="Optional comma-separated tenant IDs.")
+@click.option(
+    "--tenant-prefixes",
+    default=None,
+    help="Optional comma-separated tenant ID first hex digits, e.g. 0,1,a,f.",
+)
+@click.option("--year", default=None, type=click.IntRange(min=1, max=9999), help="Optional archive year filter.")
+@click.option("--month", default=None, type=click.IntRange(min=1, max=12), help="Optional archive month filter.")
+@click.option("--limit", default=None, type=click.IntRange(min=1), help="Maximum number of manifests to process.")
+@click.option("--dry-run", is_flag=True, help="Preview without writing workflow_run_archive_bundles.")
+def backfill_workflow_run_archive_bundles(
+    tenant_ids: str | None,
+    tenant_prefixes: str | None,
+    year: int | None,
+    month: int | None,
+    limit: int | None,
+    dry_run: bool,
+) -> None:
+    """
+    Reconcile `workflow_run_archive_bundles` from V2 archive manifests.
+
+    This command is meant for bootstrapping the listing/download index after deploy or repairing index drift. The R2
+    manifests remain the source of truth; this command only mirrors their query metadata into the database.
+    """
+    from services.retention.workflow_run.archive_bundle_index import WorkflowRunArchiveBundleIndexBackfill
+
+    if tenant_ids and tenant_prefixes:
+        raise click.UsageError("Choose either --tenant-ids or --tenant-prefixes, not both.")
+    if month is not None and year is None:
+        raise click.UsageError("--month must be used with --year.")
+
+    parsed_tenant_ids = _parse_comma_separated_ids(tenant_ids, param_name="tenant-ids")
+    parsed_tenant_prefixes = _parse_tenant_prefixes(tenant_prefixes)
+    if not parsed_tenant_ids and not parsed_tenant_prefixes:
+        click.echo(
+            click.style(
+                "No tenant scope supplied; scanning the full workflow-runs/v2/ archive prefix.",
+                fg="yellow",
+            )
+        )
+
+    started_at = datetime.datetime.now(datetime.UTC)
+    click.echo(click.style(f"Starting archive bundle index backfill at {started_at.isoformat()}.", fg="white"))
+
+    backfill = WorkflowRunArchiveBundleIndexBackfill()
+    summary = backfill.run(
+        tenant_ids=parsed_tenant_ids,
+        tenant_prefixes=parsed_tenant_prefixes or None,
+        year=year,
+        month=month,
+        limit=limit,
+        dry_run=dry_run,
+    )
+    status = "completed with failures" if summary.bundles_failed else "completed successfully"
+    fg = "red" if summary.bundles_failed else "green"
+    action = "would_upsert" if dry_run else "upserted"
+    action_count = summary.bundles_processed if dry_run else summary.bundles_upserted
+    click.echo(
+        click.style(
+            f"Backfill {status}. manifests_found={summary.manifests_found} "
+            f"bundles_processed={summary.bundles_processed} {action}={action_count} "
+            f"bundles_failed={summary.bundles_failed} runs={summary.workflow_run_count} rows={summary.row_count} "
+            f"archive_bytes={summary.archive_bytes} duration={summary.elapsed_time:.2f}s",
+            fg=fg,
+        )
+    )
+    for error in summary.errors[:10]:
+        click.echo(click.style(f"  failed {error}", fg="red"))
+    if len(summary.errors) > 10:
+        click.echo(click.style(f"  ... and {len(summary.errors) - 10} more failures", fg="red"))
+
+
 def _echo_bundle_archive_operation_summary(summary) -> None:
     status = "completed successfully" if summary.bundles_failed == 0 else "completed with failures"
     fg = "green" if summary.bundles_failed == 0 else "red"

api/configs/packaging/pyproject.py

@@ -6,6 +6,17 @@ class PyProjectConfig(BaseModel):
     version: str = Field(description="Dify version", default="")
 
 
+class DifyToolConfig(BaseModel):
+    min_difyctl_version: str = Field(
+        description="Oldest difyctl version served on /openapi/v1",
+        default="0.0.0",
+    )
+
+
+class ToolConfig(BaseModel):
+    dify: DifyToolConfig = Field(default=DifyToolConfig())
+
+
 class PyProjectTomlConfig(BaseSettings):
     """
     configs in api/pyproject.toml
@@ -15,3 +26,8 @@ class PyProjectTomlConfig(BaseSettings):
         description="configs in the project section of pyproject.toml",
         default=PyProjectConfig(),
     )
+
+    tool: ToolConfig = Field(
+        description="configs in the [tool.*] section of pyproject.toml",
+        default=ToolConfig(),
+    )

api/controllers/common/agent_app_parameters.py

@@ -0,0 +1,54 @@
+from typing import Any
+
+from sqlalchemy import select
+
+from core.app.apps.agent_app.app_feature_projection import merge_agent_app_features
+from core.app.apps.agent_app.app_variable_projection import agent_app_variables_to_user_input_form
+from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
+from extensions.ext_database import db
+from models.agent import Agent, AgentConfigSnapshot, AgentStatus
+from models.agent_config_entities import AgentSoulConfig
+from models.model import App
+
+
+def get_published_agent_app_feature_dict_and_user_input_form(
+    app_model: App,
+) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """Return public Agent App parameters backed by the published Agent Soul."""
+    app_model_config = app_model.app_model_config
+
+    agent_id = app_model.bound_agent_id
+    if not agent_id:
+        raise AgentAppGeneratorError("Agent App has no bound Agent")
+
+    agent = db.session.scalar(
+        select(Agent)
+        .where(
+            Agent.tenant_id == app_model.tenant_id,
+            Agent.id == agent_id,
+            Agent.status == AgentStatus.ACTIVE,
+        )
+        .limit(1)
+    )
+    if agent is None:
+        raise AgentAppGeneratorError("Agent App has no bound Agent")
+    # active_config_is_published means the draft has no unpublished edits; the public app
+    # can still read parameters from the active snapshot while a newer draft is pending.
+    if not agent.active_config_snapshot_id:
+        raise AgentAppNotPublishedError("Agent has not been published")
+
+    snapshot = db.session.scalar(
+        select(AgentConfigSnapshot)
+        .where(
+            AgentConfigSnapshot.tenant_id == app_model.tenant_id,
+            AgentConfigSnapshot.agent_id == agent.id,
+            AgentConfigSnapshot.id == agent.active_config_snapshot_id,
+        )
+        .limit(1)
+    )
+    if snapshot is None:
+        raise AgentAppGeneratorError("Agent published version not found")
+
+    agent_soul = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
+    features_dict = merge_agent_app_features(agent_soul=agent_soul, app_model_config=app_model_config)
+    return features_dict, agent_app_variables_to_user_input_form(agent_soul.app_variables)

api/controllers/common/app_access.py

@@ -4,6 +4,7 @@ from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from extensions.ext_database import db
 from services.enterprise import rbac_service as enterprise_rbac_service
 
 if TYPE_CHECKING:
@@ -76,7 +77,7 @@ def resolve_app_access_filter(
     inner-API round trip; otherwise it is fetched here.
     """
     if permissions is None:
-        permissions = enterprise_rbac_service.RBACService.MyPermissions.get(tenant_id, account_id)
+        permissions = enterprise_rbac_service.RBACService.MyPermissions.get(tenant_id, account_id, session=db.session())
     whitelist_scope = enterprise_rbac_service.RBACService.AppAccess.whitelist_resources(tenant_id, account_id)
 
     can_manage_own_apps = _MANAGE_OWN_APPS_PERMISSION_KEY in permissions.workspace.permission_keys

api/controllers/common/fields.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, RootModel, computed_field
 
@@ -52,6 +52,11 @@ class AudioTranscriptResponse(ResponseModel):
     text: str
 
 
+class ValidationResultResponse(ResponseModel):
+    result: Literal["success", "error"]
+    error: str | None = None
+
+
 class SimpleResultMessageResponse(ResponseModel):
     result: str
     message: str

api/controllers/console/__init__.py

@@ -43,6 +43,7 @@ from . import (
     setup,
     spec,
     version,
+    workflow_run_archive,
 )
 from .agent import composer as agent_composer
 from .agent import roster as agent_roster
@@ -238,6 +239,7 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_node_output_inspector",
     "workflow_run",
+    "workflow_run_archive",
     "workflow_statistic",
     "workflow_trigger",
     "workspace",

api/controllers/console/agent/composer.py

@@ -16,6 +16,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user_id,
 )
+from extensions.ext_database import db
 from fields.agent_fields import (
     AgentAppComposerResponse,
     AgentComposerCandidatesResponse,
@@ -69,6 +70,7 @@ class WorkflowAgentComposerApi(Resource):
                 node_id=node_id,
                 account_id=account_id,
                 snapshot_id=query.snapshot_id,
+                session=db.session(),
             ),
         )
 
@@ -94,6 +96,7 @@ class WorkflowAgentComposerApi(Resource):
                 node_id=node_id,
                 account_id=account_id,
                 payload=payload,
+                session=db.session(),
             ),
         )
 
@@ -126,6 +129,7 @@ class WorkflowAgentComposerCopyFromRosterApi(Resource):
                 source_agent_id=payload.source_agent_id,
                 source_snapshot_id=payload.source_snapshot_id,
                 idempotency_key=payload.idempotency_key,
+                session=db.session(),
             ),
         )
 
@@ -149,8 +153,9 @@ class WorkflowAgentComposerValidateApi(Resource):
             tenant_id=tenant_id,
             payload=payload,
             agent_id=AgentComposerService.resolve_workflow_node_agent_id(
-                tenant_id=tenant_id, app_id=app_model.id, node_id=node_id
+                tenant_id=tenant_id, app_id=app_model.id, node_id=node_id, session=db.session()
             ),
+            session=db.session(),
         )
         return dump_response(AgentComposerValidateResponse, {"result": "success", "errors": [], **findings})
 
@@ -174,6 +179,7 @@ class WorkflowAgentComposerCandidatesApi(Resource):
                 app_id=app_model.id,
                 node_id=node_id,
                 user_id=current_user_id,
+                session=db.session(),
             ),
         )
 
@@ -196,7 +202,9 @@ class WorkflowAgentComposerImpactApi(Resource):
             )
         return dump_response(
             AgentComposerImpactResponse,
-            AgentComposerService.calculate_impact(tenant_id=tenant_id, current_snapshot_id=current_snapshot_id),
+            AgentComposerService.calculate_impact(
+                tenant_id=tenant_id, current_snapshot_id=current_snapshot_id, session=db.session()
+            ),
         )
 
 
@@ -224,6 +232,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
                 node_id=node_id,
                 account_id=account_id,
                 payload=payload,
+                session=db.session(),
             ),
         )
 
@@ -238,7 +247,7 @@ class AgentComposerApi(Resource):
     def get(self, tenant_id: str, agent_id: UUID):
         return dump_response(
             AgentAppComposerResponse,
-            AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id)),
+            AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id), session=db.session()),
         )
 
     @console_ns.expect(console_ns.models[ComposerSavePayload.__name__])
@@ -259,6 +268,7 @@ class AgentComposerApi(Resource):
                 agent_id=str(agent_id),
                 account_id=account_id,
                 payload=payload,
+                session=db.session(),
             ),
         )
 
@@ -274,7 +284,7 @@ class AgentComposerValidateApi(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def post(self, tenant_id: str, agent_id: UUID):
-        AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id))
+        AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=str(agent_id), session=db.session())
         payload = ComposerSavePayload.model_validate(console_ns.payload or {})
         ComposerConfigValidator.validate_publish_payload(payload)
         AgentComposerService.validate_knowledge_datasets(tenant_id=tenant_id, agent_soul=payload.agent_soul)
@@ -282,6 +292,7 @@ class AgentComposerValidateApi(Resource):
             tenant_id=tenant_id,
             payload=payload,
             agent_id=str(agent_id),
+            session=db.session(),
         )
         return dump_response(AgentComposerValidateResponse, {"result": "success", "errors": [], **findings})
 
@@ -303,5 +314,6 @@ class AgentComposerCandidatesApi(Resource):
                 tenant_id=tenant_id,
                 agent_id=str(agent_id),
                 user_id=current_user_id,
+                session=db.session(),
             ),
         )

api/controllers/console/agent/roster.py

@@ -534,7 +534,7 @@ class AgentAppListApi(Resource):
             status="normal",
         )
 
-        app_pagination = AppService().get_paginate_apps(current_user.id, current_tenant_id, params, db.session)
+        app_pagination = AppService().get_paginate_apps(current_user.id, current_tenant_id, params, db.session())
         if app_pagination is None:
             empty = AgentAppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
             return empty.model_dump(mode="json")
@@ -567,7 +567,7 @@ class AgentAppListApi(Resource):
             icon_background=args.icon_background,
         )
 
-        app = AppService().create_app(current_tenant_id, params, current_user)
+        app = AppService().create_app(current_tenant_id, params, current_user, session=db.session())
         return _serialize_agent_app_detail(app, current_user=current_user), 201
 
 
@@ -607,7 +607,7 @@ class AgentAppApi(Resource):
             "max_active_requests": args.max_active_requests or 0,
             "role": args.role,
         }
-        updated = AppService().update_app(app_model, args_dict)
+        updated = AppService().update_app(app_model, args_dict, session=db.session())
         return _serialize_agent_app_detail(updated, current_user=current_user)
 
     @console_ns.response(204, "Agent app deleted successfully")
@@ -619,7 +619,7 @@ class AgentAppApi(Resource):
     @with_current_tenant_id
     def delete(self, tenant_id: str, agent_id: UUID):
         app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
-        AppService().delete_app(app_model)
+        AppService().delete_app(app_model, session=db.session())
         return "", 204
 
 
@@ -668,6 +668,7 @@ class AgentPublishApi(Resource):
             agent_id=str(agent_id),
             account_id=current_user.id,
             version_note=args.version_note,
+            session=db.session(),
         )
 
 
@@ -688,6 +689,7 @@ class AgentBuildDraftCheckoutApi(Resource):
             agent_id=str(agent_id),
             account_id=current_user.id,
             force=args.force,
+            session=db.session(),
         )
 
 
@@ -705,6 +707,7 @@ class AgentBuildDraftApi(Resource):
             tenant_id=tenant_id,
             agent_id=str(agent_id),
             account_id=current_user.id,
+            session=db.session(),
         )
 
     @console_ns.expect(console_ns.models[ComposerSavePayload.__name__])
@@ -722,6 +725,7 @@ class AgentBuildDraftApi(Resource):
             agent_id=str(agent_id),
             account_id=current_user.id,
             payload=payload,
+            session=db.session(),
         )
 
     @console_ns.response(200, "Agent build draft discarded", console_ns.models[AgentSimpleResultResponse.__name__])
@@ -736,6 +740,7 @@ class AgentBuildDraftApi(Resource):
             tenant_id=tenant_id,
             agent_id=str(agent_id),
             account_id=current_user.id,
+            session=db.session(),
         )
 
 
@@ -753,6 +758,7 @@ class AgentBuildDraftApplyApi(Resource):
             tenant_id=tenant_id,
             agent_id=str(agent_id),
             account_id=current_user.id,
+            session=db.session(),
         )
 
 
@@ -810,7 +816,7 @@ class AgentApiStatusApi(Resource):
     def post(self, tenant_id: str, agent_id: UUID):
         app_model = _resolve_agent_app_model(tenant_id=tenant_id, agent_id=agent_id)
         args = AgentApiStatusPayload.model_validate(console_ns.payload)
-        app_model = AppService().update_app_api_status(app_model, args.enable_api)
+        app_model = AppService().update_app_api_status(app_model, args.enable_api, session=db.session())
         return _serialize_agent_api_access(app_model)
 
 

api/controllers/console/app/agent.py

@@ -172,7 +172,7 @@ register_response_schema_models(
 def _resolve_agent_id(app_model: App, node_id: str | None) -> str | None:
     if node_id and app_model.mode != AppMode.AGENT:
         return AgentComposerService.resolve_workflow_node_agent_id(
-            tenant_id=app_model.tenant_id, app_id=app_model.id, node_id=node_id
+            tenant_id=app_model.tenant_id, app_id=app_model.id, node_id=node_id, session=db.session()
         )
     return app_model.bound_agent_id
 
@@ -202,6 +202,7 @@ def _upload_skill_for_app(*, current_user: Account, app_model: App):
             tenant_id=app_model.tenant_id,
             user_id=current_user.id,
             agent_id=agent_id,
+            session=db.session(),
         )
     except (SkillPackageError, AgentDriveError) as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
@@ -240,6 +241,7 @@ def _commit_drive_file_for_app(*, current_user: Account, app_model: App, allow_n
                     value_owned_by_drive=True,
                 )
             ],
+            session=db.session(),
         )
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
@@ -273,6 +275,7 @@ def _delete_drive_file_for_app(*, current_user: Account, app_model: App, allow_n
             user_id=current_user.id,
             agent_id=agent_id,
             items=[DriveCommitItem(key=key, file_ref=None)],
+            session=db.session(),
         )
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
@@ -298,6 +301,7 @@ def _delete_skill_for_app(*, current_user: Account, app_model: App, slug: str, a
                 DriveCommitItem(key=f"{slug}/SKILL.md", file_ref=None),
                 DriveCommitItem(key=f"{slug}/.DIFY-SKILL-FULL.zip", file_ref=None),
             ],
+            session=db.session(),
         )
     except AgentDriveError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
@@ -313,7 +317,9 @@ def _infer_skill_tools_for_app(*, app_model: App, slug: str):
     if "/" in slug or not slug.strip():
         return {"code": "drive_key_invalid", "message": "skill slug must be a single path segment"}, 400
     try:
-        return SkillToolInferenceService().infer(tenant_id=app_model.tenant_id, agent_id=agent_id, slug=slug)
+        return SkillToolInferenceService().infer(
+            tenant_id=app_model.tenant_id, agent_id=agent_id, slug=slug, session=db.session()
+        )
     except SkillToolInferenceError as exc:
         return {"code": exc.code, "message": exc.message}, exc.status_code
 
@@ -335,7 +341,7 @@ class AgentLogApi(Resource):
         """Get agent logs"""
         args = AgentLogQuery.model_validate(request.args.to_dict(flat=True))
 
-        return AgentService.get_agent_logs(app_model, args.conversation_id, args.message_id)
+        return AgentService.get_agent_logs(app_model, args.conversation_id, args.message_id, db.session())
 
 
 @console_ns.route("/agent/<uuid:agent_id>/skills/upload")

api/controllers/console/app/agent_app_feature.py

@@ -93,7 +93,7 @@ class AgentAppFeatureConfigResource(Resource):
             app_model=app_model,
             account=current_user,
             config=args.model_dump(exclude_none=True),
-            session=db.session,
+            session=db.session(),
         )
 
         app_model_config_was_updated.send(app_model, app_model_config=new_app_model_config)

api/controllers/console/app/agent_app_sandbox.py

@@ -25,6 +25,7 @@ from controllers.console import console_ns
 from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_tenant_id
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
 from models.model import App, AppMode
@@ -108,14 +109,8 @@ class SandboxReadResponse(ResponseModel):
     text: str | None = None
 
 
-class SandboxToolFileResponse(ResponseModel):
-    transfer_method: Literal["tool_file"] = "tool_file"
-    reference: str
-
-
 class SandboxUploadResponse(ResponseModel):
-    path: str
-    file: SandboxToolFileResponse
+    url: str
 
 
 register_schema_models(
@@ -225,7 +220,7 @@ class AgentAppSandboxReadResource(Resource):
 @console_ns.route("/agent/<uuid:agent_id>/sandbox/files/upload")
 class AgentAppSandboxUploadResource(Resource):
     @console_ns.doc("upload_agent_app_sandbox_file")
-    @console_ns.doc(description="Upload one Agent App sandbox file as a Dify ToolFile mapping")
+    @console_ns.doc(description="Upload one Agent App sandbox file and return a signed download URL")
     @console_ns.expect(console_ns.models[AgentSandboxUploadPayload.__name__])
     @console_ns.response(200, "Uploaded", console_ns.models[SandboxUploadResponse.__name__])
     @setup_required
@@ -275,6 +270,7 @@ class WorkflowAgentSandboxListResource(Resource):
                 node_id=node_id,
                 node_execution_id=query.node_execution_id,
                 path=query.path,
+                session=db.session(),
             )
         except Exception as exc:
             return _handle(exc)
@@ -311,6 +307,7 @@ class WorkflowAgentSandboxReadResource(Resource):
                 node_id=node_id,
                 node_execution_id=query.node_execution_id,
                 path=query.path,
+                session=db.session(),
             )
         except Exception as exc:
             return _handle(exc)
@@ -322,7 +319,7 @@ class WorkflowAgentSandboxReadResource(Resource):
 )
 class WorkflowAgentSandboxUploadResource(Resource):
     @console_ns.doc("upload_workflow_agent_sandbox_file")
-    @console_ns.doc(description="Upload one workflow Agent sandbox file as a Dify ToolFile mapping")
+    @console_ns.doc(description="Upload one workflow Agent sandbox file and return a signed download URL")
     @console_ns.expect(console_ns.models[WorkflowAgentSandboxUploadPayload.__name__])
     @console_ns.response(200, "Uploaded", console_ns.models[SandboxUploadResponse.__name__])
     @setup_required
@@ -340,6 +337,7 @@ class WorkflowAgentSandboxUploadResource(Resource):
                 node_id=node_id,
                 node_execution_id=payload.node_execution_id,
                 path=payload.path,
+                session=db.session(),
             )
         except Exception as exc:
             return _handle(exc)

api/controllers/console/app/agent_config_inspector.py

@@ -253,6 +253,7 @@ def _resolve_agent_id(app_model: App, node_id: str | None) -> str | None:
             tenant_id=app_model.tenant_id,
             app_id=app_model.id,
             node_id=node_id,
+            session=db.session(),
         )
     return app_model.bound_agent_id
 
@@ -288,13 +289,16 @@ def _resolve_console_version(
                 tenant_id=tenant_id,
                 agent_id=agent_id,
                 account_id=account_id,
+                session=db.session(),
             )
             draft = state.get("draft") or {}
             draft_id = draft.get("id")
             if isinstance(draft_id, str) and draft_id:
                 return draft_id, AgentConfigVersionKind.BUILD_DRAFT
         else:
-            state = AgentComposerService.load_agent_composer(tenant_id=tenant_id, agent_id=agent_id)
+            state = AgentComposerService.load_agent_composer(
+                tenant_id=tenant_id, agent_id=agent_id, session=db.session()
+            )
             draft = state.get("draft") or {}
             draft_id = draft.get("id")
             if isinstance(draft_id, str) and draft_id:

api/controllers/console/app/agent_drive_inspector.py

@@ -28,6 +28,7 @@ from controllers.console import console_ns
 from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_tenant_id
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
 from models.model import App, AppMode
@@ -147,7 +148,7 @@ def _resolve_agent_id(app_model: App, node_id: str | None) -> str | None:
     """Agent identity for the drive: app-bound agent, or the workflow node binding."""
     if node_id:
         return AgentComposerService.resolve_workflow_node_agent_id(
-            tenant_id=app_model.tenant_id, app_id=app_model.id, node_id=node_id
+            tenant_id=app_model.tenant_id, app_id=app_model.id, node_id=node_id, session=db.session()
         )
     return app_model.bound_agent_id
 
@@ -184,7 +185,9 @@ class AgentDriveListByAgentApi(Resource):
         query = query_params_from_request(AgentDriveListByAgentQuery)
         resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
-            items = AgentDriveService().manifest(tenant_id=tenant_id, agent_id=str(agent_id), prefix=query.prefix)
+            items = AgentDriveService().manifest(
+                tenant_id=tenant_id, agent_id=str(agent_id), prefix=query.prefix, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
         return {"items": [{k: v for k, v in item.items() if k != "file_id"} for item in items]}
@@ -203,7 +206,7 @@ class AgentDriveSkillListByAgentApi(Resource):
     def get(self, tenant_id: str, agent_id: UUID):
         resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
-            items = AgentDriveService().list_skills(tenant_id=tenant_id, agent_id=str(agent_id))
+            items = AgentDriveService().list_skills(tenant_id=tenant_id, agent_id=str(agent_id), session=db.session())
         except AgentDriveError as exc:
             return _handle(exc)
         return {"items": items}
@@ -227,6 +230,7 @@ class AgentDriveSkillInspectByAgentApi(Resource):
                     tenant_id=tenant_id,
                     agent_id=str(agent_id),
                     skill_path=skill_path,
+                    session=db.session(),
                 )
             )
         except AgentDriveError as exc:
@@ -247,7 +251,9 @@ class AgentDrivePreviewByAgentApi(Resource):
         query = query_params_from_request(AgentDriveFileByAgentQuery)
         resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
-            return AgentDriveService().preview(tenant_id=tenant_id, agent_id=str(agent_id), key=query.key)
+            return AgentDriveService().preview(
+                tenant_id=tenant_id, agent_id=str(agent_id), key=query.key, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
 
@@ -266,7 +272,9 @@ class AgentDriveDownloadByAgentApi(Resource):
         query = query_params_from_request(AgentDriveFileByAgentQuery)
         resolve_agent_runtime_app_model(tenant_id=tenant_id, agent_id=agent_id)
         try:
-            url = AgentDriveService().download_url(tenant_id=tenant_id, agent_id=str(agent_id), key=query.key)
+            url = AgentDriveService().download_url(
+                tenant_id=tenant_id, agent_id=str(agent_id), key=query.key, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
         return {"url": url}
@@ -288,7 +296,9 @@ class AgentDriveListApi(Resource):
         if not agent_id:
             return _agent_not_bound()
         try:
-            items = AgentDriveService().manifest(tenant_id=app_model.tenant_id, agent_id=agent_id, prefix=query.prefix)
+            items = AgentDriveService().manifest(
+                tenant_id=app_model.tenant_id, agent_id=agent_id, prefix=query.prefix, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
         # the inner manifest exposes file_id for agent-side pulls; the console
@@ -312,7 +322,9 @@ class AgentDriveSkillListApi(Resource):
         if not agent_id:
             return _agent_not_bound()
         try:
-            items = AgentDriveService().list_skills(tenant_id=app_model.tenant_id, agent_id=agent_id)
+            items = AgentDriveService().list_skills(
+                tenant_id=app_model.tenant_id, agent_id=agent_id, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
         return {"items": items}
@@ -345,6 +357,7 @@ class AgentDriveSkillInspectApi(Resource):
                     tenant_id=app_model.tenant_id,
                     agent_id=agent_id,
                     skill_path=skill_path,
+                    session=db.session(),
                 )
             )
         except AgentDriveError as exc:
@@ -367,7 +380,9 @@ class AgentDrivePreviewApi(Resource):
         if not agent_id:
             return _agent_not_bound()
         try:
-            return AgentDriveService().preview(tenant_id=app_model.tenant_id, agent_id=agent_id, key=query.key)
+            return AgentDriveService().preview(
+                tenant_id=app_model.tenant_id, agent_id=agent_id, key=query.key, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
 
@@ -388,7 +403,9 @@ class AgentDriveDownloadApi(Resource):
         if not agent_id:
             return _agent_not_bound()
         try:
-            url = AgentDriveService().download_url(tenant_id=app_model.tenant_id, agent_id=agent_id, key=query.key)
+            url = AgentDriveService().download_url(
+                tenant_id=app_model.tenant_id, agent_id=agent_id, key=query.key, session=db.session()
+            )
         except AgentDriveError as exc:
             return _handle(exc)
         return {"url": url}

api/controllers/console/app/annotation.py

@@ -211,7 +211,7 @@ class AppAnnotationSettingDetailApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     def get(self, app_id: UUID):
-        result = AppAnnotationService.get_app_annotation_setting_by_app_id(str(app_id))
+        result = AppAnnotationService.get_app_annotation_setting_by_app_id(str(app_id), session=db.session())
         return dump_response(AnnotationSettingResponse, result), 200
 
 
@@ -235,7 +235,7 @@ class AppAnnotationSettingUpdateApi(Resource):
 
         setting_args: UpdateAnnotationSettingArgs = {"score_threshold": args.score_threshold}
         result = AppAnnotationService.update_app_annotation_setting(
-            str(app_id), annotation_setting_id_str, setting_args
+            str(app_id), annotation_setting_id_str, setting_args, session=db.session()
         )
         return dump_response(AnnotationSettingResponse, result), 200
 
@@ -292,7 +292,9 @@ class AnnotationApi(Resource):
         limit = args.limit
         keyword = args.keyword
 
-        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(str(app_id), page, limit, keyword)
+        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(
+            str(app_id), page, limit, keyword, session=db.session()
+        )
         annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
         return AnnotationList(
             data=annotation_models, has_more=len(annotation_list) == limit, limit=limit, total=total, page=page
@@ -321,7 +323,9 @@ class AnnotationApi(Resource):
             upsert_args["message_id"] = args.message_id
         if args.question is not None:
             upsert_args["question"] = args.question
-        annotation = AppAnnotationService.up_insert_app_annotation_from_message(upsert_args, str(app_id))
+        annotation = AppAnnotationService.up_insert_app_annotation_from_message(
+            upsert_args, str(app_id), session=db.session()
+        )
         return dump_response(Annotation, annotation), 201
 
     @setup_required
@@ -345,11 +349,11 @@ class AnnotationApi(Resource):
                 }, 400
 
             app_ref = _get_app_ref(str(app_id))
-            AppAnnotationService.delete_app_annotations_in_batch(app_ref, annotation_ids)
+            AppAnnotationService.delete_app_annotations_in_batch(app_ref, annotation_ids, session=db.session())
             return "", 204
         # If no annotation_ids are provided, handle clearing all annotations
         else:
-            AppAnnotationService.clear_all_annotations(str(app_id))
+            AppAnnotationService.clear_all_annotations(str(app_id), session=db.session())
             return "", 204
 
 
@@ -370,7 +374,7 @@ class AnnotationExportApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     def get(self, app_id: UUID):
-        annotation_list = AppAnnotationService.export_annotation_list_by_app_id(str(app_id))
+        annotation_list = AppAnnotationService.export_annotation_list_by_app_id(str(app_id), session=db.session())
         annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
         return (
             AnnotationExportList(data=annotation_models).model_dump(mode="json"),
@@ -406,7 +410,7 @@ class AnnotationUpdateDeleteApi(Resource):
             update_args["question"] = args.question
         app_ref = _get_app_ref(str(app_id))
         annotation_ref = AppRefService.create_annotation_ref(app_ref, str(annotation_id))
-        annotation = AppAnnotationService.update_app_annotation_directly(update_args, annotation_ref, db.session)
+        annotation = AppAnnotationService.update_app_annotation_directly(update_args, annotation_ref, db.session())
         return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")
 
     @setup_required
@@ -418,7 +422,7 @@ class AnnotationUpdateDeleteApi(Resource):
     def delete(self, app_id: UUID, annotation_id: UUID):
         app_ref = _get_app_ref(str(app_id))
         annotation_ref = AppRefService.create_annotation_ref(app_ref, str(annotation_id))
-        AppAnnotationService.delete_app_annotation(annotation_ref, db.session)
+        AppAnnotationService.delete_app_annotation(annotation_ref, db.session())
         return "", 204
 
 
@@ -477,7 +481,7 @@ class AnnotationBatchImportApi(Resource):
 
         return dump_response(
             AnnotationBatchImportResponse,
-            AppAnnotationService.batch_import_app_annotations(str(app_id), file),
+            AppAnnotationService.batch_import_app_annotations(str(app_id), file, session=db.session()),
         )
 
 
@@ -538,6 +542,7 @@ class AnnotationHitHistoryListApi(Resource):
             annotation_ref,
             page,
             limit,
+            session=db.session(),
         )
         history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
             annotation_hit_history_list, from_attributes=True

api/controllers/console/app/app.py

@@ -584,6 +584,7 @@ class AppListApi(Resource):
         permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
             str(current_tenant_id),
             current_user_id,
+            session=db.session(),
         )
         if dify_config.RBAC_ENABLED:
             access_filter = resolve_app_access_filter(
@@ -595,7 +596,7 @@ class AppListApi(Resource):
 
         # get app list
         app_service = AppService()
-        app_pagination = app_service.get_paginate_apps(current_user_id, current_tenant_id, params, db.session)
+        app_pagination = app_service.get_paginate_apps(current_user_id, current_tenant_id, params, session)
         if not app_pagination:
             response = AppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
             return response.model_dump(mode="json"), 200
@@ -643,11 +644,12 @@ class AppListApi(Resource):
         )
 
         app_service = AppService()
-        app = app_service.create_app(current_tenant_id, params, current_user)
+        app = app_service.create_app(current_tenant_id, params, current_user, session=db.session())
         permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,
             [str(app.id)],
+            session=db.session(),
         )
         app_detail = AppDetailWithSite.model_validate(app, from_attributes=True).model_copy(
             update={"permission_keys": permission_keys_map.get(str(app.id), [])}
@@ -681,7 +683,7 @@ class StarredAppListApi(Resource):
             is_created_by_me=args.is_created_by_me,
         )
 
-        app_pagination = AppService().get_paginate_starred_apps(current_user_id, current_tenant_id, params, db.session)
+        app_pagination = AppService().get_paginate_starred_apps(current_user_id, current_tenant_id, params, session)
         if not app_pagination:
             empty = AppPagination(page=args.page, limit=args.limit, total=0, has_more=False, data=[])
             return empty.model_dump(mode="json"), 200
@@ -705,7 +707,7 @@ class AppStarApi(Resource):
     @with_session
     @get_app_model(mode=None)
     def post(self, session: Session, current_user_id: str, app_model: App):
-        AppService.star_app(session, app=app_model, account_id=current_user_id)
+        AppService.star_app(app=app_model, account_id=current_user_id, session=session)
         return SimpleResultResponse(result="success").model_dump(mode="json")
 
     @console_ns.doc("unstar_app")
@@ -721,7 +723,7 @@ class AppStarApi(Resource):
     @with_session
     @get_app_model(mode=None)
     def delete(self, session: Session, current_user_id: str, app_model: App):
-        AppService.unstar_app(session, app=app_model, account_id=current_user_id)
+        AppService.unstar_app(app=app_model, account_id=current_user_id, session=session)
         return SimpleResultResponse(result="success").model_dump(mode="json")
 
 
@@ -753,6 +755,7 @@ class AppApi(Resource):
             str(current_tenant_id),
             current_user.id,
             app_id=str(app_model.id),
+            session=db.session(),
         )
         permission_keys_map = permissions.app.permission_keys_by_resource_ids([str(app_model.id)])
 
@@ -789,7 +792,7 @@ class AppApi(Resource):
             "use_icon_as_answer_icon": args.use_icon_as_answer_icon or False,
             "max_active_requests": args.max_active_requests or 0,
         }
-        app_model = app_service.update_app(app_model, args_dict)
+        app_model = app_service.update_app(app_model, args_dict, session=db.session())
         return dump_response(AppDetailWithSite, app_model)
 
     @console_ns.doc("delete_app")
@@ -806,7 +809,7 @@ class AppApi(Resource):
     def delete(self, app_model: App):
         """Delete app"""
         app_service = AppService()
-        app_service.delete_app(app_model)
+        app_service.delete_app(app_model, session=db.session())
 
         return "", 204
 
@@ -835,7 +838,7 @@ class AppCopyApi(Resource):
 
         with Session(db.engine, expire_on_commit=False) as session:
             import_service = AppDslService(session)
-            yaml_content = import_service.export_dsl(app_model=app_model, include_secret=True)
+            yaml_content = import_service.export_dsl(app_model=app_model, session=session, include_secret=True)
             result = import_service.import_app(
                 account=current_user,
                 import_mode=ImportMode.YAML_CONTENT,
@@ -877,6 +880,7 @@ class AppCopyApi(Resource):
             str(current_tenant_id),
             current_user.id,
             [str(app.id)],
+            session=db.session(),
         )
         response_model = AppDetailWithSite.model_validate(app, from_attributes=True).model_copy(
             update={"permission_keys": permission_keys_map.get(str(app.id), [])}
@@ -905,6 +909,7 @@ class AppExportApi(Resource):
         response = AppExportResponse(
             data=AppDslService.export_dsl(
                 app_model=app_model,
+                session=db.session(),
                 include_secret=args.include_secret,
                 workflow_id=args.workflow_id,
             )
@@ -929,7 +934,7 @@ class AppPublishToCreatorsPlatformApi(Resource):
         if not dify_config.CREATORS_PLATFORM_FEATURES_ENABLED:
             return {"error": "Creators Platform features are not enabled"}, 403
 
-        dsl_content = AppDslService.export_dsl(app_model=app_model, include_secret=False)
+        dsl_content = AppDslService.export_dsl(app_model=app_model, session=db.session(), include_secret=False)
         dsl_bytes = dsl_content.encode("utf-8")
 
         claim_code = upload_dsl(dsl_bytes)
@@ -955,7 +960,7 @@ class AppNameApi(Resource):
         args = AppNamePayload.model_validate(console_ns.payload)
 
         app_service = AppService()
-        app_model = app_service.update_app_name(app_model, args.name)
+        app_model = app_service.update_app_name(app_model, args.name, session=db.session())
         return dump_response(AppDetail, app_model)
 
 
@@ -982,6 +987,7 @@ class AppIconApi(Resource):
             args.icon or "",
             args.icon_background or "",
             args.icon_type,
+            session=db.session(),
         )
         return dump_response(AppDetail, app_model)
 
@@ -1004,7 +1010,7 @@ class AppSiteStatus(Resource):
         args = AppSiteStatusPayload.model_validate(console_ns.payload)
 
         app_service = AppService()
-        app_model = app_service.update_app_site_status(app_model, args.enable_site)
+        app_model = app_service.update_app_site_status(app_model, args.enable_site, session=db.session())
         return dump_response(AppDetail, app_model)
 
 
@@ -1026,7 +1032,7 @@ class AppApiStatus(Resource):
         args = AppApiStatusPayload.model_validate(console_ns.payload)
 
         app_service = AppService()
-        app_model = app_service.update_app_api_status(app_model, args.enable_api)
+        app_model = app_service.update_app_api_status(app_model, args.enable_api, session=db.session())
         return dump_response(AppDetail, app_model)
 
 

api/controllers/console/app/audio.py

@@ -161,7 +161,7 @@ class ChatMessageTextApi(Resource):
             # response-contract:ignore
             return AudioService.transcript_tts(
                 app_model=app_model,
-                session=db.session,
+                session=db.session(),
                 text=payload.text,
                 voice=payload.voice,
                 message_ref=message_ref,

api/controllers/console/app/completion.py

@@ -36,7 +36,7 @@ from controllers.console.wraps import (
     with_current_user_id,
 )
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.entities.app_invoke_entities import AGENT_RUNTIME_EXIT_INTENT_ARG, InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
 from core.errors.error import (
     ModelCurrentlyNotSupportError,
@@ -124,6 +124,7 @@ Use only the current Build chat message history to identify changes that need to
 validate old config unless the message history already shows that the old config is invalid.
 
 Only update the build-draft config note when the current Build chat contains durable context that later runs need.
+Write the config note in the language used by the message history.
 Do not create, update, delete, inspect, or fill gaps in other Agent config resources, including config files, config
 skills, config env, tools, models, knowledge, or prompt settings.
 
@@ -415,6 +416,7 @@ def _create_build_chat_finalization_message(
         "draft_type": "debug_build",
         "conversation_id": debug_conversation_id,
         "auto_generate_name": False,
+        AGENT_RUNTIME_EXIT_INTENT_ARG: "delete",
     }
     external_trace_id = get_external_trace_id(request)
     if external_trace_id:

api/controllers/console/app/conversation.py

@@ -200,7 +200,7 @@ class CompletionConversationDetailApi(Resource):
         conversation_id_str = str(conversation_id)
 
         try:
-            ConversationService.delete(app_model, conversation_id_str, current_user)
+            ConversationService.delete(app_model, conversation_id_str, current_user, session=db.session())
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
@@ -354,7 +354,7 @@ class ChatConversationDetailApi(Resource):
         conversation_id_str = str(conversation_id)
 
         try:
-            ConversationService.delete(app_model, conversation_id_str, current_user)
+            ConversationService.delete(app_model, conversation_id_str, current_user, session=db.session())
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 

api/controllers/console/app/message.py

@@ -363,6 +363,7 @@ def _list_chat_messages(*, app_model: App, current_user: Account | None = None):
                 app_model=app_model,
                 conversation_id=args.conversation_id,
                 user=current_user,
+                session=db.session(),
             )
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
@@ -474,7 +475,11 @@ def _get_message_suggested_questions(*, current_user: Account, app_model: App, m
 
     try:
         questions = MessageService.get_suggested_questions_after_answer(
-            app_model=app_model, message_id=message_id_str, user=current_user, invoke_from=InvokeFrom.DEBUGGER
+            app_model=app_model,
+            message_id=message_id_str,
+            user=current_user,
+            invoke_from=InvokeFrom.DEBUGGER,
+            session=db.session(),
         )
     except MessageNotExistsError:
         raise NotFound("Message not found")

api/controllers/console/app/ops_trace.py

@@ -17,6 +17,7 @@ from controllers.console.wraps import (
     rbac_permission_required,
     setup_required,
 )
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
 from models import App
@@ -78,7 +79,7 @@ class TraceAppConfigApi(Resource):
 
         try:
             trace_config = OpsService.get_tracing_app_config(
-                app_id=app_model.id, tracing_provider=args.tracing_provider
+                app_id=app_model.id, tracing_provider=args.tracing_provider, session=db.session()
             )
             if not trace_config:
                 return {"has_not_configured": True}
@@ -109,7 +110,10 @@ class TraceAppConfigApi(Resource):
 
         try:
             result = OpsService.create_tracing_app_config(
-                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id,
+                tracing_provider=args.tracing_provider,
+                tracing_config=args.tracing_config,
+                session=db.session(),
             )
             if not result:
                 raise TracingConfigIsExist()
@@ -142,7 +146,10 @@ class TraceAppConfigApi(Resource):
 
         try:
             result = OpsService.update_tracing_app_config(
-                app_id=app_model.id, tracing_provider=args.tracing_provider, tracing_config=args.tracing_config
+                app_id=app_model.id,
+                tracing_provider=args.tracing_provider,
+                tracing_config=args.tracing_config,
+                session=db.session(),
             )
             if not result:
                 raise TracingConfigNotExist()
@@ -168,7 +175,9 @@ class TraceAppConfigApi(Resource):
         args = TraceProviderQuery.model_validate(request.args.to_dict(flat=True))
 
         try:
-            result = OpsService.delete_tracing_app_config(app_id=app_model.id, tracing_provider=args.tracing_provider)
+            result = OpsService.delete_tracing_app_config(
+                app_id=app_model.id, tracing_provider=args.tracing_provider, session=db.session()
+            )
             if not result:
                 raise TracingConfigNotExist()
             return "", 204

api/controllers/console/app/permission_keys.py

@@ -1,6 +1,9 @@
+from extensions.ext_database import db
 from services.enterprise import rbac_service as enterprise_rbac_service
 
 
 def get_app_permission_keys(tenant_id: str, account_id: str | None, app_id: str) -> list[str]:
-    permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(tenant_id, account_id, [app_id])
+    permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
+        tenant_id, account_id, [app_id], session=db.session()
+    )
     return permission_keys_map.get(app_id, [])

api/controllers/console/app/workflow.py

@@ -2,7 +2,7 @@ import json
 import logging
 from collections.abc import Sequence
 from datetime import datetime
-from typing import Any, NotRequired, TypedDict, cast
+from typing import Any, NotRequired, TypedDict
 
 from flask import abort, request
 from flask_restx import Resource, fields
@@ -522,7 +522,7 @@ class DraftWorkflowApi(Resource):
         """
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()
-        workflow = workflow_service.get_draft_workflow(app_model=app_model)
+        workflow = workflow_service.get_draft_workflow(app_model=app_model, session=db.session())
 
         if not workflow:
             raise DraftWorkflowNotExist()
@@ -533,7 +533,7 @@ class DraftWorkflowApi(Resource):
         # front-end can treat draft graph node data as the editing source.
         response = WorkflowResponse.model_validate(workflow, from_attributes=True).model_dump(mode="json")
         response["graph"] = WorkflowAgentPublishService.project_draft_bindings_to_graph(
-            session=cast(Session, db.session),
+            session=db.session(),
             draft_workflow=workflow,
         )
         return response
@@ -602,6 +602,7 @@ class DraftWorkflowApi(Resource):
                 account=current_user,
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
+                session=db.session(),
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
@@ -695,7 +696,12 @@ class AdvancedChatDraftRunIterationNodeApi(Resource):
 
         try:
             response = AppGenerateService.generate_single_iteration(
-                app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
+                app_model=app_model,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                session=db.session(),
+                streaming=True,
             )
 
             return helper.compact_generate_response(response)
@@ -738,7 +744,12 @@ class WorkflowDraftRunIterationNodeApi(Resource):
 
         try:
             response = AppGenerateService.generate_single_iteration(
-                app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
+                app_model=app_model,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                session=db.session(),
+                streaming=True,
             )
 
             return helper.compact_generate_response(response)
@@ -777,7 +788,12 @@ class AdvancedChatDraftRunLoopNodeApi(Resource):
 
         try:
             response = AppGenerateService.generate_single_loop(
-                app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
+                app_model=app_model,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                session=db.session(),
+                streaming=True,
             )
 
             return helper.compact_generate_response(response)
@@ -820,7 +836,12 @@ class WorkflowDraftRunLoopNodeApi(Resource):
 
         try:
             response = AppGenerateService.generate_single_loop(
-                app_model=app_model, user=current_user, node_id=node_id, args=args, streaming=True
+                app_model=app_model,
+                user=current_user,
+                node_id=node_id,
+                args=args,
+                session=db.session(),
+                streaming=True,
             )
 
             return helper.compact_generate_response(response)
@@ -897,6 +918,7 @@ class AdvancedChatDraftHumanInputFormPreviewApi(Resource):
             account=current_user,
             node_id=node_id,
             inputs=inputs,
+            session=db.session(),
         )
         return jsonable_encoder(preview)
 
@@ -932,6 +954,7 @@ class AdvancedChatDraftHumanInputFormRunApi(Resource):
             form_inputs=args.form_inputs,
             inputs=args.inputs,
             action=args.action,
+            session=db.session(),
         )
         return jsonable_encoder(result)
 
@@ -963,6 +986,7 @@ class WorkflowDraftHumanInputFormPreviewApi(Resource):
             account=current_user,
             node_id=node_id,
             inputs=inputs,
+            session=db.session(),
         )
         return jsonable_encoder(preview)
 
@@ -998,6 +1022,7 @@ class WorkflowDraftHumanInputFormRunApi(Resource):
             form_inputs=args.form_inputs,
             inputs=args.inputs,
             action=args.action,
+            session=db.session(),
         )
         return jsonable_encoder(result)
 
@@ -1028,6 +1053,7 @@ class WorkflowDraftHumanInputDeliveryTestApi(Resource):
             node_id=node_id,
             delivery_method_id=args.delivery_method_id,
             inputs=args.inputs,
+            session=db.session(),
         )
         return jsonable_encoder({})
 
@@ -1138,7 +1164,7 @@ class DraftWorkflowNodeRunApi(Resource):
 
         workflow_srv = WorkflowService()
         # fetch draft workflow by app_model
-        draft_workflow = workflow_srv.get_draft_workflow(app_model=app_model)
+        draft_workflow = workflow_srv.get_draft_workflow(app_model=app_model, session=db.session())
         if not draft_workflow:
             raise ValueError("Workflow not initialized")
         files = _parse_file(draft_workflow, args.get("files"))
@@ -1181,7 +1207,7 @@ class PublishedWorkflowApi(Resource):
         """
         # fetch published workflow by app_model
         workflow_service = WorkflowService()
-        workflow = workflow_service.get_published_workflow(app_model=app_model)
+        workflow = workflow_service.get_published_workflow(app_model=app_model, session=db.session())
 
         # return workflow, if not found, return None
         if workflow is None:
@@ -1323,7 +1349,9 @@ class ConvertToWorkflowApi(Resource):
 
         # convert to workflow mode
         workflow_service = WorkflowService()
-        new_app_model = workflow_service.convert_to_workflow(app_model=app_model, account=current_user, args=args)
+        new_app_model = workflow_service.convert_to_workflow(
+            app_model=app_model, account=current_user, args=args, session=db.session()
+        )
 
         # return app id
         return {
@@ -1358,7 +1386,9 @@ class WorkflowFeaturesApi(Resource):
         features = args.features.model_dump(mode="json", exclude_unset=True)
 
         workflow_service = WorkflowService()
-        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
+        workflow_service.update_draft_workflow_features(
+            app_model=app_model, features=features, account=current_user, session=db.session()
+        )
 
         return {"result": "success"}
 
@@ -1439,6 +1469,7 @@ class DraftWorkflowRestoreApi(Resource):
                 app_model=app_model,
                 workflow_id=workflow_id,
                 account=current_user,
+                session=db.session(),
             )
         except IsDraftWorkflowError as exc:
             raise BadRequest(RESTORE_SOURCE_WORKFLOW_MUST_BE_PUBLISHED_MESSAGE) from exc
@@ -1553,7 +1584,7 @@ class DraftWorkflowNodeLastRunApi(Resource):
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     def get(self, app_model: App, node_id: str):
         srv = WorkflowService()
-        workflow = srv.get_draft_workflow(app_model)
+        workflow = srv.get_draft_workflow(app_model, session=db.session())
         if not workflow:
             raise NotFound("Workflow not found")
         node_exec = srv.get_node_last_run(
@@ -1606,7 +1637,7 @@ class DraftWorkflowTriggerRunApi(Resource):
         args = DraftWorkflowTriggerRunPayload.model_validate(console_ns.payload or {})
         node_id = args.node_id
         workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
+        draft_workflow = workflow_service.get_draft_workflow(app_model, session=db.session())
         if not draft_workflow:
             raise ValueError("Workflow not found")
 
@@ -1675,7 +1706,7 @@ class DraftWorkflowTriggerNodeApi(Resource):
         """
 
         workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
+        draft_workflow = workflow_service.get_draft_workflow(app_model, session=db.session())
         if not draft_workflow:
             raise ValueError("Workflow not found")
 
@@ -1759,7 +1790,7 @@ class DraftWorkflowTriggerRunAllApi(Resource):
         args = DraftWorkflowTriggerRunAllPayload.model_validate(console_ns.payload or {})
         node_ids = args.node_ids
         workflow_service = WorkflowService()
-        draft_workflow = workflow_service.get_draft_workflow(app_model)
+        draft_workflow = workflow_service.get_draft_workflow(app_model, session=db.session())
         if not draft_workflow:
             raise ValueError("Workflow not found")
 
@@ -1828,7 +1859,7 @@ class WorkflowOnlineUsersApi(Resource):
             return {"data": []}
 
         workflow_service = WorkflowService()
-        accessible_app_ids = workflow_service.get_accessible_app_ids(app_ids, current_tenant_id)
+        accessible_app_ids = workflow_service.get_accessible_app_ids(app_ids, current_tenant_id, session=db.session())
         ordered_accessible_app_ids = [app_id for app_id in app_ids if app_id in accessible_app_ids]
 
         users_json_by_app_id: dict[str, Any] = {}

api/controllers/console/app/workflow_comment.py

@@ -490,7 +490,7 @@ class WorkflowCommentMentionUsersApi(Resource):
         current_tenant = current_user.current_tenant  # need the tenant object here
         if current_tenant is None:
             raise ValueError("current tenant is required")
-        members = TenantService.get_tenant_members(current_tenant, session=db.session)
+        members = TenantService.get_tenant_members(current_tenant, session=db.session())
         users = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
         response = WorkflowCommentMentionUsersPayload(users=users)
         return response.model_dump(mode="json"), 200

api/controllers/console/app/workflow_draft_variable.py

@@ -337,7 +337,7 @@ class WorkflowVariableCollectionApi(Resource):
 
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()
-        workflow_exist = workflow_service.is_workflow_exist(app_model=app_model)
+        workflow_exist = workflow_service.is_workflow_exist(app_model=app_model, session=db.session())
         if not workflow_exist:
             raise DraftWorkflowNotExist()
 
@@ -553,7 +553,7 @@ class VariableResetApi(Resource):
         )
 
         workflow_srv = WorkflowService()
-        draft_workflow = workflow_srv.get_draft_workflow(app_model)
+        draft_workflow = workflow_srv.get_draft_workflow(app_model, session=db.session())
         if draft_workflow is None:
             raise NotFoundError(
                 f"Draft workflow not found, app_id={app_model.id}",
@@ -606,7 +606,7 @@ class ConversationVariableCollectionApi(Resource):
         # NOTE(QuantumGhost): Prefill conversation variables into the draft variables table
         # so their IDs can be returned to the caller.
         workflow_srv = WorkflowService()
-        draft_workflow = workflow_srv.get_draft_workflow(app_model)
+        draft_workflow = workflow_srv.get_draft_workflow(app_model, session=db.session())
         if draft_workflow is None:
             raise NotFoundError(description=f"draft workflow not found, id={app_model.id}")
         draft_var_srv = WorkflowDraftVariableService(db.session())
@@ -646,6 +646,7 @@ class ConversationVariableCollectionApi(Resource):
             app_model=app_model,
             account=current_user,
             conversation_variables=conversation_variables,
+            session=db.session(),
         )
 
         return {"result": "success"}
@@ -683,7 +684,7 @@ class EnvironmentVariableCollectionApi(Resource):
         """
         # fetch draft workflow by app_model
         workflow_service = WorkflowService()
-        workflow = workflow_service.get_draft_workflow(app_model=app_model)
+        workflow = workflow_service.get_draft_workflow(app_model=app_model, session=db.session())
         if workflow is None:
             raise DraftWorkflowNotExist()
 
@@ -740,6 +741,7 @@ class EnvironmentVariableCollectionApi(Resource):
             app_model=app_model,
             account=current_user,
             environment_variables=environment_variables,
+            session=db.session(),
         )
 
         return {"result": "success"}

api/controllers/console/app/workflow_node_output_inspector.py

@@ -41,6 +41,7 @@ from controllers.console.wraps import (
     rbac_permission_required,
     setup_required,
 )
+from extensions.ext_database import db
 from libs.exception import BaseHTTPException
 from libs.login import login_required
 from models import App, AppMode
@@ -92,7 +93,9 @@ def _serve_snapshot(app_model: App, run_id: UUID) -> dict:
     Flask request context.
     """
     try:
-        snapshot = _service().snapshot_workflow_run(app_model=app_model, workflow_run_id=str(run_id))
+        snapshot = _service().snapshot_workflow_run(
+            app_model=app_model, workflow_run_id=str(run_id), session=db.session()
+        )
     except NodeOutputInspectorError as error:
         raise _InspectorNotFound(error) from error
     return snapshot.model_dump(mode="json")
@@ -105,6 +108,7 @@ def _serve_node_detail(app_model: App, run_id: UUID, node_id: str) -> dict:
             app_model=app_model,
             workflow_run_id=str(run_id),
             node_id=node_id,
+            session=db.session(),
         )
     except NodeOutputInspectorError as error:
         raise _InspectorNotFound(error) from error
@@ -119,6 +123,7 @@ def _serve_output_preview(app_model: App, run_id: UUID, node_id: str, output_nam
             workflow_run_id=str(run_id),
             node_id=node_id,
             output_name=output_name,
+            session=db.session(),
         )
     except NodeOutputInspectorError as error:
         raise _InspectorNotFound(error) from error
@@ -245,7 +250,7 @@ def _stream_inspector_events(app_model: App, run_id: UUID) -> Iterator[str]:
     # if the run is gone (raised before yielding any bytes, so Flask turns it
     # into the normal HTTP 404 path).
     try:
-        snapshot = service.snapshot_workflow_run(app_model=app_model, workflow_run_id=run_id_str)
+        snapshot = service.snapshot_workflow_run(app_model=app_model, workflow_run_id=run_id_str, session=db.session())
     except NodeOutputInspectorError as error:
         raise _InspectorNotFound(error) from error
 
@@ -308,6 +313,7 @@ def _stream_inspector_events(app_model: App, run_id: UUID) -> Iterator[str]:
                 app_model=app_model,
                 workflow_run_id=run_id_str,
                 node_id=message.node_id,
+                session=db.session(),
             )
         except NodeOutputInspectorError:
             # Node may not appear in the graph yet (race with persistence); skip.

api/controllers/console/auth/activate.py

@@ -90,7 +90,7 @@ class ActivateCheckApi(Resource):
         token = args.token
 
         invitation = RegisterService.get_invitation_with_case_fallback(
-            workspaceId, args.email, token, session=db.session
+            workspaceId, args.email, token, session=db.session()
         )
         if invitation:
             data = invitation.get("data", {})
@@ -140,7 +140,7 @@ class ActivateApi(Resource):
 
         normalized_request_email = args.email.lower() if args.email else None
         invitation = RegisterService.get_invitation_with_case_fallback(
-            args.workspace_id, args.email, args.token, session=db.session
+            args.workspace_id, args.email, args.token, session=db.session()
         )
         if invitation is None:
             raise AlreadyActivateError()
@@ -178,7 +178,7 @@ class ActivateApi(Resource):
         RegisterService.revoke_token(args.workspace_id, normalized_request_email, args.token)
 
         if membership_id is None:
-            TenantService.create_tenant_member(tenant, account, db.session, role=role)
+            TenantService.create_tenant_member(tenant, account, db.session(), role=role)
 
         if setup_fields:
             account.name = setup_fields[0]
@@ -188,6 +188,6 @@ class ActivateApi(Resource):
             account.status = AccountStatus.ACTIVE
             account.initialized_at = naive_utc_now()
 
-        TenantService.switch_tenant(account, tenant.id, session=db.session)
+        TenantService.switch_tenant(account, tenant.id, session=db.session())
 
         return {"result": "success"}

api/controllers/console/auth/data_source_bearer_auth.py

@@ -59,7 +59,7 @@ class ApiKeyAuthDataSource(Resource):
     @account_initialization_required
     @with_current_tenant_id
     def get(self, current_tenant_id: str):
-        data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(db.session(), current_tenant_id)
+        data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(current_tenant_id, session=db.session())
         if data_source_api_key_bindings:
             return {
                 "sources": [
@@ -93,7 +93,7 @@ class ApiKeyAuthDataSourceBinding(Resource):
         data = payload.model_dump()
         ApiKeyAuthService.validate_api_key_auth_args(data)
         try:
-            ApiKeyAuthService.create_provider_auth(db.session(), current_tenant_id, data)
+            ApiKeyAuthService.create_provider_auth(current_tenant_id, data, session=db.session())
         except Exception as e:
             raise ApiKeyAuthFailedError(str(e))
         return {"result": "success"}, 200
@@ -110,6 +110,6 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
     @with_current_tenant_id
     def delete(self, current_tenant_id: str, binding_id: UUID):
         # The role of the current user in the table must be admin or owner
-        ApiKeyAuthService.delete_provider_auth(db.session(), current_tenant_id, str(binding_id))
+        ApiKeyAuthService.delete_provider_auth(current_tenant_id, str(binding_id), session=db.session())
 
         return "", 204

api/controllers/console/auth/email_register.py

@@ -101,7 +101,7 @@ class EmailRegisterSendEmailApi(Resource):
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
             raise AccountInFreezeError()
 
-        account = AccountService.get_account_by_email_with_case_fallback(db.session, args.email)
+        account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
         token = AccountService.send_email_register_email(email=normalized_email, account=account, language=language)
         return {"result": "success", "data": token}
 
@@ -176,7 +176,7 @@ class EmailRegisterResetApi(Resource):
         email = register_data.get("email", "")
         normalized_email = email.lower()
 
-        account = AccountService.get_account_by_email_with_case_fallback(db.session, email)
+        account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
 
         if account:
             raise EmailAlreadyInUseError()
@@ -187,7 +187,7 @@ class EmailRegisterResetApi(Resource):
             timezone=args.timezone,
             language=args.language,
         )
-        token_pair = AccountService.login(account=account, session=db.session, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account=account, session=db.session(), ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(normalized_email)
 
         return {"result": "success", "data": token_pair.model_dump()}
@@ -206,7 +206,7 @@ class EmailRegisterResetApi(Resource):
                 password=password,
                 interface_language=get_valid_language(language),
                 timezone=timezone,
-                session=db.session,
+                session=db.session(),
             )
         except AccountRegisterError:
             raise AccountInFreezeError()

api/controllers/console/auth/forgot_password.py

@@ -82,7 +82,7 @@ class ForgotPasswordSendEmailApi(Resource):
         else:
             language = "en-US"
 
-        account = AccountService.get_account_by_email_with_case_fallback(db.session, args.email)
+        account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
 
         token = AccountService.send_reset_password_email(
             account=account,
@@ -180,7 +180,7 @@ class ForgotPasswordResetApi(Resource):
         password_hashed = hash_password(args.new_password, salt)
 
         email = reset_data.get("email", "")
-        account = AccountService.get_account_by_email_with_case_fallback(db.session, email)
+        account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
 
         if account:
             account = db.session.merge(account)
@@ -198,10 +198,10 @@ class ForgotPasswordResetApi(Resource):
 
         # Create workspace if needed
         if (
-            not TenantService.get_join_tenants(account, session=db.session)
+            not TenantService.get_join_tenants(account, session=db.session())
             and FeatureService.get_system_features().is_allow_create_workspace
         ):
-            tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session)
-            TenantService.create_tenant_member(tenant, account, db.session, role="owner")
+            tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session())
+            TenantService.create_tenant_member(tenant, account, db.session(), role="owner")
             account.current_tenant = tenant
             tenant_was_created.send(tenant)

api/controllers/console/auth/login.py

@@ -9,7 +9,12 @@ from werkzeug.exceptions import Unauthorized
 import services
 from configs import dify_config
 from constants.languages import get_valid_language
-from controllers.common.fields import SimpleResultDataResponse, SimpleResultOptionalDataResponse, SimpleResultResponse
+from controllers.common.fields import (
+    SimpleResultDataResponse,
+    SimpleResultMessageResponse,
+    SimpleResultOptionalDataResponse,
+    SimpleResultResponse,
+)
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
@@ -51,7 +56,7 @@ from models.account import Account
 from services.account_service import AccountService, InvitationDetailDict, RegisterService, TenantService
 from services.billing_service import BillingService
 from services.entities.auth_entities import LoginFailureReason, LoginPayloadBase
-from services.errors.account import AccountRegisterError
+from services.errors.account import AccountRegisterError, RefreshTokenAccountNotFoundError, RefreshTokenNotFoundError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
@@ -87,6 +92,7 @@ register_schema_models(console_ns, LoginPayload, EmailPayload, EmailCodeLoginPay
 register_response_schema_models(
     console_ns,
     SimpleResultDataResponse,
+    SimpleResultMessageResponse,
     SimpleResultOptionalDataResponse,
     SimpleResultResponse,
 )
@@ -120,7 +126,7 @@ class LoginApi(Resource):
         invitation_data: InvitationDetailDict | None = None
         if invite_token:
             invitation_data = RegisterService.get_invitation_with_case_fallback(
-                None, request_email, invite_token, session=db.session
+                None, request_email, invite_token, session=db.session()
             )
             if invitation_data is None:
                 invite_token = None
@@ -147,23 +153,26 @@ class LoginApi(Resource):
             _log_console_login_failure(email=normalized_email, reason=LoginFailureReason.INVALID_CREDENTIALS)
             raise AuthenticationFailedError() from exc
         # SELF_HOSTED only have one workspace
-        tenants = TenantService.get_join_tenants(account, session=db.session)
+        tenants = TenantService.get_join_tenants(account, session=db.session())
         if len(tenants) == 0:
             system_features = FeatureService.get_system_features()
 
             if system_features.is_allow_create_workspace and not system_features.license.workspaces.is_available():
                 raise WorkspacesLimitExceeded()
             else:
-                return {
-                    "result": "fail",
-                    "data": "workspace not found, please contact system admin to invite you to join in a workspace",
-                }
+                return SimpleResultOptionalDataResponse(
+                    result="fail",
+                    data="workspace not found, please contact system admin to invite you to join in a workspace",
+                ).model_dump(mode="json")
 
-        token_pair = AccountService.login(account=account, session=db.session, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account=account, session=db.session(), ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(normalized_email)
 
         # Create response with cookies instead of returning tokens in body
-        response = make_response({"result": "success"})
+        # response-contract:ignore cookie-bearing Flask response
+        response = make_response(
+            SimpleResultOptionalDataResponse(result="success").model_dump(mode="json", exclude_none=True)
+        )
 
         set_access_token_to_cookie(request, response, token_pair.access_token)
         set_refresh_token_to_cookie(request, response, token_pair.refresh_token)
@@ -178,12 +187,11 @@ class LogoutApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
     @with_current_user
     def post(self, account: Account):
-        if isinstance(account, flask_login.AnonymousUserMixin):
-            response = make_response({"result": "success"})
-        else:
+        # response-contract:ignore cookie-bearing Flask response
+        response = make_response(SimpleResultResponse(result="success").model_dump(mode="json"))
+        if not isinstance(account, flask_login.AnonymousUserMixin):
             AccountService.logout(account=account)
             flask_login.logout_user()
-            response = make_response({"result": "success"})
 
         # Clear cookies on logout
         clear_access_token_from_cookie(response)
@@ -219,7 +227,7 @@ class ResetPasswordSendEmailApi(Resource):
             is_allow_register=FeatureService.get_system_features().is_allow_register,
         )
 
-        return {"result": "success", "data": token}
+        return SimpleResultDataResponse(result="success", data=token).model_dump(mode="json")
 
 
 @console_ns.route("/email-code-login")
@@ -252,7 +260,7 @@ class EmailCodeLoginSendEmailApi(Resource):
         else:
             token = AccountService.send_email_code_login_email(account=account, language=language)
 
-        return {"result": "success", "data": token}
+        return SimpleResultDataResponse(result="success", data=token).model_dump(mode="json")
 
 
 @console_ns.route("/email-code-login/validity")
@@ -293,7 +301,7 @@ class EmailCodeLoginApi(Resource):
             _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
             raise AccountInFreezeError()
         if account:
-            tenants = TenantService.get_join_tenants(account, session=db.session)
+            tenants = TenantService.get_join_tenants(account, session=db.session())
             if not tenants:
                 workspaces = FeatureService.get_system_features().license.workspaces
                 if not workspaces.is_available():
@@ -301,8 +309,8 @@ class EmailCodeLoginApi(Resource):
                 if not FeatureService.get_system_features().is_allow_create_workspace:
                     raise NotAllowedCreateWorkspace()
                 else:
-                    new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session)
-                    TenantService.create_tenant_member(new_tenant, account, db.session, role="owner")
+                    new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session())
+                    TenantService.create_tenant_member(new_tenant, account, db.session(), role="owner")
                     account.current_tenant = new_tenant
                     tenant_was_created.send(new_tenant)
 
@@ -313,7 +321,7 @@ class EmailCodeLoginApi(Resource):
                     name=user_email,
                     interface_language=get_valid_language(language),
                     timezone=args.timezone,
-                    session=db.session,
+                    session=db.session(),
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
@@ -322,11 +330,12 @@ class EmailCodeLoginApi(Resource):
                 raise AccountInFreezeError()
             except WorkspacesLimitExceededError:
                 raise WorkspacesLimitExceeded()
-        token_pair = AccountService.login(account, session=db.session, ip_address=extract_remote_ip(request))
+        token_pair = AccountService.login(account, session=db.session(), ip_address=extract_remote_ip(request))
         AccountService.reset_login_error_rate_limit(user_email)
 
         # Create response with cookies instead of returning tokens in body
-        response = make_response({"result": "success"})
+        # response-contract:ignore cookie-bearing Flask response
+        response = make_response(SimpleResultResponse(result="success").model_dump(mode="json"))
 
         set_csrf_token_to_cookie(request, response, token_pair.csrf_token)
         # Set HTTP-only secure cookies for tokens
@@ -338,45 +347,53 @@ class EmailCodeLoginApi(Resource):
 @console_ns.route("/refresh-token")
 class RefreshTokenApi(Resource):
     @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(401, "Unauthorized", console_ns.models[SimpleResultMessageResponse.__name__])
     def post(self):
         # Get refresh token from cookie instead of request body
         refresh_token = extract_refresh_token(request)
 
         if not refresh_token:
-            return {"result": "fail", "message": "No refresh token provided"}, 401
+            return SimpleResultMessageResponse(result="fail", message="No refresh token provided").model_dump(
+                mode="json"
+            ), 401
 
         try:
-            new_token_pair = AccountService.refresh_token(refresh_token, session=db.session)
+            new_token_pair = AccountService.refresh_token(refresh_token, session=db.session())
+        except Unauthorized as exc:
+            return SimpleResultMessageResponse(result="fail", message=exc.description or "Unauthorized.").model_dump(
+                mode="json"
+            ), 401
+        except (RefreshTokenNotFoundError, RefreshTokenAccountNotFoundError) as exc:
+            return SimpleResultMessageResponse(result="fail", message=str(exc)).model_dump(mode="json"), 401
 
-            # Create response with new cookies
-            response = make_response({"result": "success"})
+        # Create response with new cookies
+        # response-contract:ignore cookie-bearing Flask response
+        response = make_response(SimpleResultResponse(result="success").model_dump(mode="json"))
 
-            # Update cookies with new tokens
-            set_csrf_token_to_cookie(request, response, new_token_pair.csrf_token)
-            set_access_token_to_cookie(request, response, new_token_pair.access_token)
-            set_refresh_token_to_cookie(request, response, new_token_pair.refresh_token)
-            return response
-        except Exception as e:
-            return {"result": "fail", "message": str(e)}, 401
+        # Update cookies with new tokens
+        set_csrf_token_to_cookie(request, response, new_token_pair.csrf_token)
+        set_access_token_to_cookie(request, response, new_token_pair.access_token)
+        set_refresh_token_to_cookie(request, response, new_token_pair.refresh_token)
+        return response
 
 
 def _get_account_with_case_fallback(email: str):
-    account = AccountService.get_user_through_email(email, session=db.session)
+    account = AccountService.get_user_through_email(email, session=db.session())
     if account or email == email.lower():
         return account
 
-    return AccountService.get_user_through_email(email.lower(), session=db.session)
+    return AccountService.get_user_through_email(email.lower(), session=db.session())
 
 
 def _authenticate_account_with_case_fallback(
     original_email: str, normalized_email: str, password: str, invite_token: str | None
 ):
     try:
-        return AccountService.authenticate(original_email, password, invite_token, session=db.session)
+        return AccountService.authenticate(original_email, password, invite_token, session=db.session())
     except services.errors.account.AccountPasswordError:
         if original_email == normalized_email:
             raise
-        return AccountService.authenticate(normalized_email, password, invite_token, session=db.session)
+        return AccountService.authenticate(normalized_email, password, invite_token, session=db.session())
 
 
 def _log_console_login_failure(*, email: str, reason: LoginFailureReason) -> None:

api/controllers/console/auth/oauth.py

@@ -195,7 +195,7 @@ class OAuthCallback(Resource):
             db.session.commit()
 
         try:
-            TenantService.create_owner_tenant_if_not_exist(account, session=db.session)
+            TenantService.create_owner_tenant_if_not_exist(account, session=db.session())
         except Unauthorized:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Workspace not found.")
         except WorkSpaceNotAllowedCreateError:
@@ -206,7 +206,7 @@ class OAuthCallback(Resource):
 
         token_pair = AccountService.login(
             account=account,
-            session=db.session,
+            session=db.session(),
             ip_address=extract_remote_ip(request),
         )
 
@@ -225,7 +225,7 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
     account: Account | None = Account.get_by_openid(provider, user_info.id)
 
     if not account:
-        account = AccountService.get_account_by_email_with_case_fallback(db.session, user_info.email)
+        account = AccountService.get_account_by_email_with_case_fallback(user_info.email, session=db.session())
 
     return account
 
@@ -241,13 +241,13 @@ def _generate_account(
     oauth_new_user = False
 
     if account:
-        tenants = TenantService.get_join_tenants(account, session=db.session)
+        tenants = TenantService.get_join_tenants(account, session=db.session())
         if not tenants:
             if not FeatureService.get_system_features().is_allow_create_workspace:
                 raise WorkSpaceNotAllowedCreateError()
             else:
-                new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session)
-                TenantService.create_tenant_member(new_tenant, account, db.session, role="owner")
+                new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace", session=db.session())
+                TenantService.create_tenant_member(new_tenant, account, db.session(), role="owner")
                 account.current_tenant = new_tenant
                 tenant_was_created.send(new_tenant)
 
@@ -273,10 +273,10 @@ def _generate_account(
             provider=provider,
             language=interface_language,
             timezone=timezone,
-            session=db.session,
+            session=db.session(),
         )
 
     # Link account
-    AccountService.link_account_integrate(provider, user_info.id, account, session=db.session)
+    AccountService.link_account_integrate(provider, user_info.id, account, session=db.session())
 
     return account, oauth_new_user

api/controllers/console/auth/oauth_server.py

@@ -10,6 +10,7 @@ from werkzeug.exceptions import BadRequest, NotFound
 
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_user
+from extensions.ext_database import db
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.login import login_required
 from models import Account
@@ -131,7 +132,9 @@ def oauth_server_access_token_required[T, **P, R](
             response.headers["WWW-Authenticate"] = "Bearer"
             return response
 
-        account = OAuthServerService.validate_oauth_access_token(oauth_provider_app.client_id, access_token)
+        account = OAuthServerService.validate_oauth_access_token(
+            oauth_provider_app.client_id, access_token, db.session()
+        )
         if not account:
             response = jsonify({"error": "access_token or client_id is invalid"})
             response.status_code = 401

api/controllers/console/billing/billing.py

@@ -56,7 +56,7 @@ class Subscription(Resource):
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account):
         args = SubscriptionQuery.model_validate(request.args.to_dict(flat=True))
-        BillingService.is_tenant_owner_or_admin(db.session, current_user)
+        BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
         return BillingService.get_subscription(args.plan, args.interval, current_user.email, current_tenant_id)
 
 
@@ -70,7 +70,7 @@ class Invoices(Resource):
     @with_current_user
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account):
-        BillingService.is_tenant_owner_or_admin(db.session, current_user)
+        BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
         return BillingService.get_invoices(current_user.email, current_tenant_id)
 
 

api/controllers/console/datasets/data_source.py

@@ -245,7 +245,7 @@ class DataSourceNotionListApi(Resource):
         exist_page_ids = []
         # import notion in the exist dataset
         if query.dataset_id:
-            dataset = DatasetService.get_dataset(query.dataset_id, db.session)
+            dataset = DatasetService.get_dataset(query.dataset_id, db.session())
             if not dataset:
                 raise NotFound("Dataset not found.")
             if dataset.data_source_type != "notion_import":
@@ -400,11 +400,11 @@ class DataSourceNotionDatasetSyncApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_CREATE_AND_MANAGEMENT)
     def get(self, dataset_id: UUID) -> tuple[dict[str, str], int]:
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        documents = DocumentService.get_document_by_dataset_id(dataset_id_str, db.session)
+        documents = DocumentService.get_document_by_dataset_id(dataset_id_str, db.session())
         for document in documents:
             document_indexing_sync_task.delay(dataset_id_str, document.id)
         return {"result": "success"}, 200
@@ -420,11 +420,11 @@ class DataSourceNotionDocumentSyncApi(Resource):
     def get(self, dataset_id: UUID, document_id: UUID) -> tuple[dict[str, str], int]:
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if document is None:
             raise NotFound("Document not found.")
         document_indexing_sync_task.delay(dataset_id_str, document_id_str)

api/controllers/console/datasets/datasets.py

@@ -30,6 +30,7 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from core.entities.knowledge_entities import IndexingEstimate
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.indexing_runner import IndexingRunner
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
@@ -268,21 +269,10 @@ class ErrorDocsResponse(DocumentStatusListResponse):
     total: int
 
 
-class IndexingEstimatePreviewItemResponse(ResponseModel):
-    content: str
-    child_chunks: list[str] | None = None
-    summary: str | None = None
-
-
-class IndexingEstimateQaPreviewItemResponse(ResponseModel):
-    question: str
-    answer: str
-
-
-class IndexingEstimateResponse(ResponseModel):
-    total_segments: int
-    preview: list[IndexingEstimatePreviewItemResponse]
-    qa_preview: list[IndexingEstimateQaPreviewItemResponse] | None = None
+class IndexingEstimateResponse(IndexingEstimate):
+    tokens: int
+    total_price: float | int
+    currency: str
 
 
 class RetrievalSettingResponse(ResponseModel):
@@ -418,6 +408,7 @@ class DatasetListApi(Resource):
         permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
             str(current_tenant_id),
             current_user.id,
+            session=db.session(),
         )
 
         accessible_dataset_ids: list[str] | None = None
@@ -461,7 +452,7 @@ class DatasetListApi(Resource):
             datasets, total = DatasetService.get_datasets(
                 query.page,
                 query.limit,
-                db.session,
+                db.session(),
                 current_tenant_id,
                 current_user,
                 query.keyword,
@@ -573,6 +564,7 @@ class DatasetListApi(Resource):
             current_tenant_id,
             current_user.id,
             [dataset.id],
+            session=session,
         )
 
         item = DatasetDetailWithPartialMembersResponse.model_validate(dataset, from_attributes=True).model_dump(
@@ -602,17 +594,18 @@ class DatasetApi(Resource):
     @with_current_tenant_id
     def get(self, current_tenant_id: str, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
             current_tenant_id,
             current_user.id,
             dataset_id=dataset_id_str,
+            session=db.session(),
         )
         permission_keys_map = permissions.dataset.permission_keys_by_resource_ids([dataset_id_str])
         data = dump_response(DatasetDetailResponse, dataset)
@@ -622,7 +615,7 @@ class DatasetApi(Resource):
                 provider_id = ModelProviderID(dataset.embedding_model_provider)
                 data["embedding_model_provider"] = str(provider_id)
         if data.get("permission") == "partial_members":
-            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session)
+            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session())
             data.update({"partial_member_list": part_users_list})
 
         # check embedding setting
@@ -644,7 +637,7 @@ class DatasetApi(Resource):
         else:
             data["embedding_available"] = True
 
-        return data, 200
+        return dump_response(DatasetDetailWithPartialMembersResponse, data), 200
 
     @console_ns.doc("update_dataset")
     @console_ns.doc(description="Update dataset details")
@@ -666,7 +659,7 @@ class DatasetApi(Resource):
     @with_session
     def patch(self, session: Session, current_tenant_id: str, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
@@ -685,10 +678,10 @@ class DatasetApi(Resource):
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
         if not dify_config.RBAC_ENABLED:
             DatasetPermissionService.check_permission(
-                session, current_user, dataset, payload.permission, payload.partial_member_list
+                current_user, dataset, payload.permission, payload.partial_member_list, session=session
             )
 
-        dataset = DatasetService.update_dataset(session, dataset_id_str, payload_data, current_user)
+        dataset = DatasetService.update_dataset(dataset_id_str, payload_data, current_user, session=session)
 
         if dataset is None:
             raise NotFound("Dataset not found.")
@@ -697,6 +690,7 @@ class DatasetApi(Resource):
             current_tenant_id,
             current_user.id,
             [dataset_id_str],
+            session=session,
         )
         result_data = dump_response(DatasetDetailResponse, dataset)
         result_data["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
@@ -704,16 +698,16 @@ class DatasetApi(Resource):
 
         if payload.partial_member_list is not None and payload.permission == DatasetPermissionEnum.PARTIAL_TEAM:
             DatasetPermissionService.update_partial_member_list(
-                tenant_id, dataset_id_str, payload.partial_member_list, db.session
+                tenant_id, dataset_id_str, payload.partial_member_list, db.session()
             )
         # clear partial member list when permission is only_me or all_team_members
         elif payload.permission in {DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.ALL_TEAM}:
-            DatasetPermissionService.clear_partial_member_list(dataset_id_str, db.session)
+            DatasetPermissionService.clear_partial_member_list(dataset_id_str, db.session())
 
-        partial_member_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session)
+        partial_member_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session())
         result_data.update({"partial_member_list": partial_member_list})
 
-        return result_data, 200
+        return dump_response(DatasetDetailWithPartialMembersResponse, result_data), 200
 
     @setup_required
     @login_required
@@ -729,8 +723,8 @@ class DatasetApi(Resource):
             raise Forbidden()
 
         try:
-            if DatasetService.delete_dataset(dataset_id_str, current_user, db.session):
-                DatasetPermissionService.clear_partial_member_list(dataset_id_str, db.session)
+            if DatasetService.delete_dataset(dataset_id_str, current_user, db.session()):
+                DatasetPermissionService.clear_partial_member_list(dataset_id_str, db.session())
                 return "", 204
             else:
                 raise NotFound("Dataset not found.")
@@ -755,8 +749,8 @@ class DatasetUseCheckApi(Resource):
     def get(self, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
 
-        dataset_is_using = DatasetService.dataset_use_check(dataset_id_str, db.session)
-        return {"is_using": dataset_is_using}, 200
+        dataset_is_using = DatasetService.dataset_use_check(dataset_id_str, db.session())
+        return UsageCheckResponse(is_using=dataset_is_using).model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/queries")
@@ -776,12 +770,12 @@ class DatasetQueryApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     def get(self, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -897,7 +891,17 @@ class DatasetIndexingEstimateApi(Resource):
         except Exception as e:
             raise IndexingEstimateError(str(e))
 
-        return response.model_dump(), 200
+        return (
+            IndexingEstimateResponse(
+                tokens=0,
+                total_price=0,
+                currency="USD",
+                total_segments=response.total_segments,
+                preview=response.preview,
+                qa_preview=response.qa_preview,
+            ).model_dump(mode="json", exclude_none=True),
+            200,
+        )
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/related-apps")
@@ -917,16 +921,16 @@ class DatasetRelatedAppListApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     def get(self, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
-        app_dataset_joins = DatasetService.get_related_apps(dataset.id, db.session)
+        app_dataset_joins = DatasetService.get_related_apps(dataset.id, db.session())
 
         related_apps = []
         for app_dataset_join in app_dataset_joins:
@@ -1014,7 +1018,7 @@ class DatasetApiKeyApi(Resource):
         keys = db.session.scalars(
             select(ApiToken).where(ApiToken.type == self.resource_type, ApiToken.tenant_id == current_tenant_id)
         ).all()
-        return ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
+        return dump_response(ApiKeyList, {"data": keys})
 
     @console_ns.response(200, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
     @console_ns.response(400, "Maximum keys exceeded")
@@ -1048,7 +1052,7 @@ class DatasetApiKeyApi(Resource):
         api_token.type = self.resource_type
         db.session.add(api_token)
         db.session.commit()
-        return ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 200
+        return dump_response(ApiKeyItem, api_token), 200
 
 
 @console_ns.route("/datasets/api-keys/<uuid:api_key_id>")
@@ -1101,9 +1105,9 @@ class DatasetEnableApiApi(Resource):
     def post(self, dataset_id: UUID, status: str):
         dataset_id_str = str(dataset_id)
 
-        DatasetService.update_dataset_api_status(dataset_id_str, status == "enable", db.session)
+        DatasetService.update_dataset_api_status(dataset_id_str, status == "enable", db.session())
 
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/api-base-info")
@@ -1116,7 +1120,7 @@ class DatasetApiBaseUrlApi(Resource):
     @account_initialization_required
     def get(self):
         base = dify_config.SERVICE_API_URL or request.host_url.rstrip("/")
-        return {"api_base_url": normalize_api_base_url(base)}
+        return ApiBaseUrlResponse(api_base_url=normalize_api_base_url(base)).model_dump(mode="json")
 
 
 @console_ns.route("/datasets/retrieval-setting")
@@ -1170,10 +1174,10 @@ class DatasetErrorDocs(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     def get(self, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        results = DocumentService.get_error_documents_by_dataset_id(dataset_id_str, db.session)
+        results = DocumentService.get_error_documents_by_dataset_id(dataset_id_str, db.session())
 
         return dump_response(ErrorDocsResponse, {"data": results, "total": len(results)}), 200
 
@@ -1197,15 +1201,15 @@ class DatasetPermissionUserListApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     def get(self, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
-        partial_members_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session)
+        partial_members_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session())
 
         return dump_response(PartialMemberListResponse, {"data": partial_members_list}), 200
 
@@ -1227,8 +1231,8 @@ class DatasetAutoDisableLogApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_READONLY)
     def get(self, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        auto_disable_logs = DatasetService.get_dataset_auto_disable_logs(dataset_id_str, db.session)
+        auto_disable_logs = DatasetService.get_dataset_auto_disable_logs(dataset_id_str, db.session())
         return dump_response(AutoDisableLogsResponse, auto_disable_logs), 200

api/controllers/console/datasets/datasets_document.py

@@ -10,16 +10,17 @@ from uuid import UUID
 import sqlalchemy as sa
 from flask import request, send_file
 from flask_restx import Resource
-from pydantic import BaseModel, Field, RootModel, field_validator
+from pydantic import BaseModel, Field, JsonValue, field_validator
 from sqlalchemy import asc, desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
 from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
-from controllers.common.fields import BinaryFileResponse, SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
+from controllers.common.fields import SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import RBACPermission, RBACResourceScope, rbac_permission_required
+from core.entities.knowledge_entities import IndexingEstimate
 from core.errors.error import (
     LLMBadRequestError,
     ModelCurrentlyNotSupportError,
@@ -29,6 +30,7 @@ from core.errors.error import (
 from core.indexing_runner import IndexingRunner
 from core.model_manager import ModelManager
 from core.plugin.impl.exc import PluginDaemonClientSideError
+from core.rag.entities import Rule
 from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
@@ -49,7 +51,7 @@ from libs.login import login_required
 from libs.pagination import paginate_query
 from models import Account, DatasetProcessRule, Document, DocumentSegment, UploadFile
 from models.dataset import DocumentPipelineExecutionLog
-from models.enums import IndexingStatus, SegmentStatus
+from models.enums import IndexingStatus, ProcessRuleMode, SegmentStatus
 from services.dataset_ref_service import DatasetRefService
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
@@ -148,8 +150,91 @@ class DocumentWithSegmentsListResponse(ResponseModel):
     page: int
 
 
-class OpaqueObjectResponse(RootModel[dict[str, Any]]):
-    root: dict[str, Any]
+class IndexingEstimateResponse(IndexingEstimate):
+    tokens: int
+    total_price: float | int
+    currency: str
+
+
+class DocumentDetailResponse(ResponseModel):
+    id: str
+    position: int | None = None
+    data_source_type: str | None = None
+    data_source_info: Any = None
+    data_source_detail_dict: Any = None
+    dataset_process_rule_id: str | None = None
+    dataset_process_rule: Any = None
+    document_process_rule: Any = None
+    name: str | None = None
+    created_from: str | None = None
+    created_by: str | None = None
+    created_at: int | None = None
+    tokens: int | None = None
+    indexing_status: str | None = None
+    completed_at: int | None = None
+    updated_at: int | None = None
+    indexing_latency: float | None = None
+    error: str | None = None
+    enabled: bool | None = None
+    disabled_at: int | None = None
+    disabled_by: str | None = None
+    archived: bool | None = None
+    doc_type: str | None = None
+    doc_metadata: list[DocumentMetadataResponse] | None = None
+    segment_count: int | None = None
+    average_segment_length: float | None = None
+    hit_count: int | None = None
+    display_status: str | None = None
+    doc_form: str | None = None
+    doc_language: str | None = None
+    need_summary: bool | None = None
+
+    @field_validator("data_source_type", "indexing_status", "display_status", "doc_form", mode="before")
+    @classmethod
+    def _normalize_enum_fields(cls, value: Any) -> Any:
+        return normalize_enum(value)
+
+
+class SummaryStatusResponse(ResponseModel):
+    completed: int = 0
+    generating: int = 0
+    error: int = 0
+    not_started: int = 0
+    timeout: int = 0
+
+
+class SummaryEntryResponse(ResponseModel):
+    segment_id: str
+    segment_position: int
+    status: str
+    summary_preview: str | None = None
+    error: str | None = None
+    created_at: int | None = None
+    updated_at: int | None = None
+
+    @field_validator("status", mode="before")
+    @classmethod
+    def _normalize_status(cls, value: Any) -> Any:
+        return normalize_enum(value)
+
+
+class DocumentSummaryStatusResponse(ResponseModel):
+    total_segments: int
+    summary_status: SummaryStatusResponse
+    summaries: list[SummaryEntryResponse]
+
+
+class ProcessRuleResponse(ResponseModel):
+    mode: ProcessRuleMode
+    rules: Rule | None = None
+    limits: dict[str, Any]
+
+
+class DocumentPipelineExecutionLogResponse(ResponseModel):
+    datasource_info: JsonValue | None = None
+    datasource_type: str | None = None
+    input_data: JsonValue | None = None
+    datasource_node_id: str | None = None
 
 
 register_schema_models(
@@ -165,7 +250,6 @@ register_schema_models(
 )
 register_response_schema_models(
     console_ns,
-    BinaryFileResponse,
     SimpleResultMessageResponse,
     SimpleResultResponse,
     UrlResponse,
@@ -175,7 +259,11 @@ register_response_schema_models(
     DocumentWithSegmentsResponse,
     DatasetAndDocumentResponse,
     DocumentWithSegmentsListResponse,
-    OpaqueObjectResponse,
+    IndexingEstimateResponse,
+    DocumentDetailResponse,
+    DocumentSummaryStatusResponse,
+    ProcessRuleResponse,
+    DocumentPipelineExecutionLogResponse,
 )
 
 
@@ -183,16 +271,16 @@ class DocumentResource(Resource):
     def get_document(
         self, dataset_id: str, document_id: str, current_user: Account, current_tenant_id: str
     ) -> Document:
-        dataset = DatasetService.get_dataset(dataset_id, db.session)
+        dataset = DatasetService.get_dataset(dataset_id, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
-        document = DocumentService.get_document(dataset_id, document_id, session=db.session)
+        document = DocumentService.get_document(dataset_id, document_id, session=db.session())
 
         if not document:
             raise NotFound("Document not found.")
@@ -203,16 +291,16 @@ class DocumentResource(Resource):
         return document
 
     def get_batch_documents(self, dataset_id: str, batch: str, current_user: Account) -> Sequence[Document]:
-        dataset = DatasetService.get_dataset(dataset_id, db.session)
+        dataset = DatasetService.get_dataset(dataset_id, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
-        documents = DocumentService.get_batch_documents(dataset_id, batch, db.session)
+        documents = DocumentService.get_batch_documents(dataset_id, batch, db.session())
 
         if not documents:
             raise NotFound("Documents not found.")
@@ -225,7 +313,7 @@ class GetProcessRuleApi(Resource):
     @console_ns.doc("get_process_rule")
     @console_ns.doc(description="Get dataset document processing rules")
     @console_ns.doc(params={"document_id": "Document ID (optional)"})
-    @console_ns.response(200, "Process rules retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
+    @console_ns.response(200, "Process rules retrieved successfully", console_ns.models[ProcessRuleResponse.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -243,13 +331,13 @@ class GetProcessRuleApi(Resource):
             # get the latest process rule
             document = db.get_or_404(Document, document_id)
 
-            dataset = DatasetService.get_dataset(document.dataset_id, db.session)
+            dataset = DatasetService.get_dataset(document.dataset_id, db.session())
 
             if not dataset:
                 raise NotFound("Dataset not found.")
 
             try:
-                DatasetService.check_dataset_permission(dataset, current_user, db.session)
+                DatasetService.check_dataset_permission(dataset, current_user, db.session())
             except services.errors.account.NoPermissionError as e:
                 raise Forbidden(str(e))
 
@@ -264,7 +352,7 @@ class GetProcessRuleApi(Resource):
                 mode = dataset_process_rule.mode
                 rules = dataset_process_rule.rules_dict
 
-        return {"mode": mode, "rules": rules, "limits": limits}
+        return dump_response(ProcessRuleResponse, {"mode": mode, "rules": rules, "limits": limits})
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -319,12 +407,12 @@ class DatasetDocumentListApi(Resource):
                     )
         except (ArgumentTypeError, ValueError, Exception):
             fetch = False
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -376,6 +464,7 @@ class DatasetDocumentListApi(Resource):
             documents=documents,
             dataset=dataset,
             tenant_id=current_tenant_id,
+            session=db.session(),
         )
 
         if fetch:
@@ -423,7 +512,7 @@ class DatasetDocumentListApi(Resource):
     def post(self, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
 
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
 
         if not dataset:
             raise NotFound("Dataset not found.")
@@ -433,7 +522,7 @@ class DatasetDocumentListApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -447,9 +536,9 @@ class DatasetDocumentListApi(Resource):
 
         try:
             documents, batch = DocumentService.save_document_with_dataset_id(
-                dataset, knowledge_config, current_user, session=db.session
+                dataset, knowledge_config, current_user, session=db.session()
             )
-            dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+            dataset = DatasetService.get_dataset(dataset_id_str, db.session())
 
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -468,7 +557,7 @@ class DatasetDocumentListApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     def delete(self, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
         # check user's model setting
@@ -477,7 +566,7 @@ class DatasetDocumentListApi(Resource):
         try:
             document_ids = request.args.getlist("document_id")
             dataset_ref = DatasetRefService.create_dataset_ref(dataset)
-            DocumentService.delete_documents(dataset_ref, document_ids, dataset.doc_form, db.session)
+            DocumentService.delete_documents(dataset_ref, document_ids, dataset.doc_form, db.session())
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
@@ -490,7 +579,7 @@ class DatasetInitApi(Resource):
     @console_ns.doc(description="Initialize dataset with documents")
     @console_ns.expect(console_ns.models[KnowledgeConfig.__name__])
     @console_ns.response(
-        201, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
+        200, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
     )
     @console_ns.response(400, "Invalid request parameters")
     @setup_required
@@ -536,7 +625,7 @@ class DatasetInitApi(Resource):
                 tenant_id=current_tenant_id,
                 knowledge_config=knowledge_config,
                 account=current_user,
-                session=db.session,
+                session=db.session(),
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -556,7 +645,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
     @console_ns.response(
         200,
         "Indexing estimate calculated successfully",
-        console_ns.models[OpaqueObjectResponse.__name__],
+        console_ns.models[IndexingEstimateResponse.__name__],
     )
     @console_ns.response(404, "Document not found")
     @console_ns.response(400, "Document already finished")
@@ -577,8 +666,6 @@ class DocumentIndexingEstimateApi(DocumentResource):
         data_process_rule = document.dataset_process_rule
         data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
 
-        response = {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}
-
         if document.data_source_type == "upload_file":
             data_source_info = document.data_source_info_dict
             if data_source_info and "upload_file_id" in data_source_info:
@@ -609,7 +696,18 @@ class DocumentIndexingEstimateApi(DocumentResource):
                         "English",
                         dataset_id_str,
                     )
-                    return estimate_response.model_dump(), 200
+                    return (
+                        # TODO: why using zero here? the same for the below endpoint
+                        IndexingEstimateResponse(
+                            tokens=0,
+                            total_price=0,
+                            currency="USD",
+                            total_segments=estimate_response.total_segments,
+                            preview=estimate_response.preview,
+                            qa_preview=estimate_response.qa_preview,
+                        ).model_dump(mode="json", exclude_none=True),
+                        200,
+                    )
                 except LLMBadRequestError:
                     raise ProviderNotInitializeError(
                         "No Embedding Model available. Please configure a valid provider "
@@ -622,15 +720,24 @@ class DocumentIndexingEstimateApi(DocumentResource):
                 except Exception as e:
                     raise IndexingEstimateError(str(e))
 
-        return response, 200
+        return (
+            IndexingEstimateResponse(
+                tokens=0,
+                total_price=0,
+                currency="USD",
+                total_segments=0,
+                preview=[],
+            ).model_dump(mode="json", exclude_none=True),
+            200,
+        )
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
 class DocumentBatchIndexingEstimateApi(DocumentResource):
     @console_ns.response(
         200,
-        "Batch indexing estimate calculated successfully",
-        console_ns.models[OpaqueObjectResponse.__name__],
+        "Indexing estimate calculated successfully",
+        console_ns.models[IndexingEstimateResponse.__name__],
     )
     @setup_required
     @login_required
@@ -642,7 +749,16 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
         dataset_id_str = str(dataset_id)
         documents = self.get_batch_documents(dataset_id_str, batch, current_user)
         if not documents:
-            return {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}, 200
+            return (
+                IndexingEstimateResponse(
+                    tokens=0,
+                    total_price=0,
+                    currency="USD",
+                    total_segments=0,
+                    preview=[],
+                ).model_dump(mode="json", exclude_none=True),
+                200,
+            )
         data_process_rule = documents[0].dataset_process_rule
         data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
         extract_settings = []
@@ -716,7 +832,17 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     "English",
                     dataset_id_str,
                 )
-                return response.model_dump(), 200
+                return (
+                    IndexingEstimateResponse(
+                        tokens=0,
+                        total_price=0,
+                        currency="USD",
+                        total_segments=response.total_segments,
+                        preview=response.preview,
+                        qa_preview=response.qa_preview,
+                    ).model_dump(mode="json", exclude_none=True),
+                    200,
+                )
             except LLMBadRequestError:
                 raise ProviderNotInitializeError(
                     "No Embedding Model available. Please configure a valid provider in the Settings -> Model Provider."
@@ -853,7 +979,7 @@ class DocumentApi(DocumentResource):
             "metadata": "Metadata inclusion (all/only/without)",
         }
     )
-    @console_ns.response(200, "Document retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
+    @console_ns.response(200, "Document retrieved successfully", console_ns.models[DocumentDetailResponse.__name__])
     @console_ns.response(404, "Document not found")
     @setup_required
     @login_required
@@ -870,46 +996,21 @@ class DocumentApi(DocumentResource):
         if metadata not in self.METADATA_CHOICES:
             raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
 
+        metadata_fields = {"doc_type", "doc_metadata"}
         if metadata == "only":
-            response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
-        elif metadata == "without":
-            dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session)
-            document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
-            response = {
-                "id": document.id,
-                "position": document.position,
-                "data_source_type": document.data_source_type,
-                "data_source_info": document.data_source_info_dict,
-                "data_source_detail_dict": document.data_source_detail_dict,
-                "dataset_process_rule_id": document.dataset_process_rule_id,
-                "dataset_process_rule": dataset_process_rules,
-                "document_process_rule": document_process_rules,
-                "name": document.name,
-                "created_from": document.created_from,
-                "created_by": document.created_by,
-                "created_at": int(document.created_at.timestamp()),
-                "tokens": document.tokens,
-                "indexing_status": document.indexing_status,
-                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
-                "updated_at": int(document.updated_at.timestamp()) if document.updated_at else None,
-                "indexing_latency": document.indexing_latency,
-                "error": document.error,
-                "enabled": document.enabled,
-                "disabled_at": int(document.disabled_at.timestamp()) if document.disabled_at else None,
-                "disabled_by": document.disabled_by,
-                "archived": document.archived,
-                "segment_count": document.segment_count,
-                "average_segment_length": document.average_segment_length,
-                "hit_count": document.hit_count,
-                "display_status": document.display_status,
-                "doc_form": document.doc_form,
-                "doc_language": document.doc_language,
-                "need_summary": document.need_summary if document.need_summary is not None else False,
-            }
-        else:
-            dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session)
-            document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
-            response = {
+            response = DocumentDetailResponse.model_validate(
+                {
+                    "id": document.id,
+                    "doc_type": document.doc_type,
+                    "doc_metadata": document.doc_metadata_details,
+                }
+            )
+            return response.model_dump(mode="json", include={"id", *metadata_fields}, exclude_unset=True), 200
+
+        dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session())
+        document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
+        response = DocumentDetailResponse.model_validate(
+            {
                 "id": document.id,
                 "position": document.position,
                 "data_source_type": document.data_source_type,
@@ -942,8 +1043,9 @@ class DocumentApi(DocumentResource):
                 "doc_language": document.doc_language,
                 "need_summary": document.need_summary if document.need_summary is not None else False,
             }
-
-        return response, 200
+        )
+        exclude = metadata_fields if metadata == "without" else None
+        return response.model_dump(mode="json", exclude=exclude, exclude_unset=True), 200
 
     @setup_required
     @login_required
@@ -956,7 +1058,7 @@ class DocumentApi(DocumentResource):
     def delete(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID):
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
         # check user's model setting
@@ -965,7 +1067,7 @@ class DocumentApi(DocumentResource):
         document = self.get_document(dataset_id_str, document_id_str, current_user, current_tenant_id)
 
         try:
-            DocumentService.delete_document(document, db.session)
+            DocumentService.delete_document(document, db.session())
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
@@ -989,7 +1091,9 @@ class DocumentDownloadApi(DocumentResource):
     def get(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID) -> dict[str, Any]:
         # Reuse the shared permission/tenant checks implemented in DocumentResource.
         document = self.get_document(str(dataset_id), str(document_id), current_user, current_tenant_id)
-        return {"url": DocumentService.get_document_download_url(document, db.session)}
+        return UrlResponse(url=DocumentService.get_document_download_url(document, db.session())).model_dump(
+            mode="json"
+        )
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/download-zip")
@@ -998,7 +1102,7 @@ class DocumentBatchDownloadZipApi(DocumentResource):
 
     @console_ns.doc("download_dataset_documents_as_zip")
     @console_ns.doc(description="Download selected dataset documents as a single ZIP archive (upload-file only)")
-    @console_ns.response(200, "ZIP archive generated successfully", console_ns.models[BinaryFileResponse.__name__])
+    @console_ns.response(200, "ZIP archive downloaded successfully")
     @setup_required
     @login_required
     @account_initialization_required
@@ -1019,7 +1123,7 @@ class DocumentBatchDownloadZipApi(DocumentResource):
             document_ids=document_ids,
             tenant_id=current_tenant_id,
             current_user=current_user,
-            session=db.session,
+            session=db.session(),
         )
 
         # Delegate ZIP packing to FileService, but keep Flask response+cleanup in the route.
@@ -1033,6 +1137,7 @@ class DocumentBatchDownloadZipApi(DocumentResource):
             )
             cleanup = stack.pop_all()
             response.call_on_close(cleanup.close)
+        # response-contract:ignore binary ZIP download response
         return response
 
 
@@ -1092,7 +1197,7 @@ class DocumentProcessingApi(DocumentResource):
                 document.is_paused = False
                 db.session.commit()
 
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
@@ -1151,7 +1256,9 @@ class DocumentMetadataApi(DocumentResource):
         document.updated_at = naive_utc_now()
         db.session.commit()
 
-        return {"result": "success", "message": "Document metadata updated."}, 200
+        return SimpleResultMessageResponse(result="success", message="Document metadata updated.").model_dump(
+            mode="json"
+        ), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
@@ -1168,7 +1275,7 @@ class DocumentStatusApi(DocumentResource):
         self, current_user: Account, dataset_id: UUID, action: Literal["enable", "disable", "archive", "un_archive"]
     ):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
@@ -1180,12 +1287,12 @@ class DocumentStatusApi(DocumentResource):
         DatasetService.check_dataset_model_setting(dataset)
 
         # check user's permission
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
         document_ids = request.args.getlist("document_id")
 
         try:
-            DocumentService.batch_update_document_status(dataset, document_ids, action, current_user, db.session)
+            DocumentService.batch_update_document_status(dataset, document_ids, action, current_user, db.session())
         except services.errors.document.DocumentIndexingError as e:
             raise InvalidActionError(str(e))
         except ValueError as e:
@@ -1193,7 +1300,7 @@ class DocumentStatusApi(DocumentResource):
         except NotFound as e:
             raise NotFound(str(e))
 
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")
@@ -1209,11 +1316,11 @@ class DocumentPauseApi(DocumentResource):
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
 
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
-        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session())
 
         # 404 if document not found
         if document is None:
@@ -1225,7 +1332,7 @@ class DocumentPauseApi(DocumentResource):
 
         try:
             # pause document
-            DocumentService.pause_document(document, db.session)
+            DocumentService.pause_document(document, db.session())
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot pause completed document.")
 
@@ -1244,10 +1351,10 @@ class DocumentRecoverApi(DocumentResource):
         """recover document."""
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
-        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session())
 
         # 404 if document not found
         if document is None:
@@ -1258,7 +1365,7 @@ class DocumentRecoverApi(DocumentResource):
             raise ArchivedDocumentImmutableError()
         try:
             # pause document
-            DocumentService.recover_document(document, db.session)
+            DocumentService.recover_document(document, db.session())
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Document is not in paused status.")
 
@@ -1278,13 +1385,13 @@ class DocumentRetryApi(DocumentResource):
         """retry document."""
         payload = DocumentRetryPayload.model_validate(console_ns.payload or {})
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         retry_documents = []
         if not dataset:
             raise NotFound("Dataset not found.")
         for document_id in payload.document_ids:
             try:
-                document = DocumentService.get_document(dataset.id, document_id, session=db.session)
+                document = DocumentService.get_document(dataset.id, document_id, session=db.session())
 
                 # 404 if document not found
                 if document is None:
@@ -1302,7 +1409,7 @@ class DocumentRetryApi(DocumentResource):
                 logger.exception("Failed to retry document, document id: %s", document_id)
                 continue
         # retry document
-        DocumentService.retry_document(dataset_id_str, retry_documents, db.session)
+        DocumentService.retry_document(dataset_id_str, retry_documents, db.session())
 
         return "", 204
 
@@ -1320,14 +1427,14 @@ class DocumentRenameApi(DocumentResource):
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
         if not current_user.is_dataset_editor:
             raise Forbidden()
-        dataset = DatasetService.get_dataset(dataset_id, db.session)
+        dataset = DatasetService.get_dataset(str(dataset_id), db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_operator_permission(current_user, dataset, session=db.session)
+        DatasetService.check_dataset_operator_permission(current_user, dataset, session=db.session())
         payload = DocumentRenamePayload.model_validate(console_ns.payload or {})
 
         try:
-            document = DocumentService.rename_document(str(dataset_id), str(document_id), payload.name, db.session)
+            document = DocumentService.rename_document(str(dataset_id), str(document_id), payload.name, db.session())
         except services.errors.document.DocumentIndexingError:
             raise DocumentIndexingError("Cannot delete document during indexing.")
 
@@ -1345,11 +1452,11 @@ class WebsiteDocumentSyncApi(DocumentResource):
     def get(self, current_tenant_id: str, dataset_id: UUID, document_id: UUID):
         """sync website document."""
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         if document.tenant_id != current_tenant_id:
@@ -1360,17 +1467,17 @@ class WebsiteDocumentSyncApi(DocumentResource):
         if DocumentService.check_archived(document):
             raise ArchivedDocumentImmutableError()
         # sync document
-        DocumentService.sync_website_document(dataset_id_str, document, db.session)
+        DocumentService.sync_website_document(dataset_id_str, document, db.session())
 
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log")
 class DocumentPipelineExecutionLogApi(DocumentResource):
     @console_ns.response(
         200,
-        "Document pipeline execution log retrieved successfully",
-        console_ns.models[OpaqueObjectResponse.__name__],
+        "Pipeline execution log retrieved successfully",
+        console_ns.models[DocumentPipelineExecutionLogResponse.__name__],
     )
     @setup_required
     @login_required
@@ -1380,10 +1487,10 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
 
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
-        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset.id, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         log = db.session.scalar(
@@ -1393,18 +1500,16 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
             .limit(1)
         )
         if not log:
-            return {
-                "datasource_info": None,
-                "datasource_type": None,
-                "input_data": None,
-                "datasource_node_id": None,
-            }, 200
-        return {
-            "datasource_info": json.loads(log.datasource_info),
-            "datasource_type": log.datasource_type,
-            "input_data": log.input_data,
-            "datasource_node_id": log.datasource_node_id,
-        }, 200
+            return DocumentPipelineExecutionLogResponse().model_dump(mode="json"), 200
+        return dump_response(
+            DocumentPipelineExecutionLogResponse,
+            {
+                "datasource_info": json.loads(log.datasource_info),
+                "datasource_type": log.datasource_type,
+                "input_data": log.input_data,
+                "datasource_node_id": log.datasource_node_id,
+            },
+        ), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
@@ -1438,7 +1543,7 @@ class DocumentGenerateSummaryApi(Resource):
         dataset_id_str = str(dataset_id)
 
         # Get dataset
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
@@ -1447,7 +1552,7 @@ class DocumentGenerateSummaryApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -1472,7 +1577,7 @@ class DocumentGenerateSummaryApi(Resource):
             raise ValueError("Summary index is not enabled for this dataset. Please enable it in the dataset settings.")
 
         # Verify all documents exist and belong to the dataset
-        documents = DocumentService.get_documents_by_ids(dataset_id_str, document_list, db.session)
+        documents = DocumentService.get_documents_by_ids(dataset_id_str, document_list, db.session())
 
         if len(documents) != len(document_list):
             found_ids = {doc.id for doc in documents}
@@ -1488,7 +1593,7 @@ class DocumentGenerateSummaryApi(Resource):
             DocumentService.update_documents_need_summary(
                 dataset_id=dataset_id_str,
                 document_ids=document_ids_to_update,
-                session=db.session,
+                session=db.session(),
                 need_summary=True,
             )
 
@@ -1507,7 +1612,7 @@ class DocumentGenerateSummaryApi(Resource):
                 dataset_id_str,
             )
 
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
@@ -1515,7 +1620,11 @@ class DocumentSummaryStatusApi(DocumentResource):
     @console_ns.doc("get_document_summary_status")
     @console_ns.doc(description="Get summary index generation status for a document")
     @console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
-    @console_ns.response(200, "Summary status retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
+    @console_ns.response(
+        200,
+        "Summary status retrieved successfully",
+        console_ns.models[DocumentSummaryStatusResponse.__name__],
+    )
     @console_ns.response(404, "Document not found")
     @setup_required
     @login_required
@@ -1533,19 +1642,20 @@ class DocumentSummaryStatusApi(DocumentResource):
           - generating: Number of summaries being generated
           - error: Number of summaries with errors
           - not_started: Number of segments without summary records
+          - timeout: Number of summaries that timed out
         - summaries: List of summary records with status and content preview
         """
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
 
         # Get dataset
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
         # Check permissions
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -1555,7 +1665,7 @@ class DocumentSummaryStatusApi(DocumentResource):
         result = SummaryIndexService.get_document_summary_status_detail(
             document_id=document_id_str,
             dataset_id=dataset_id_str,
-            session=db.session,
+            session=db.session(),
         )
 
-        return result, 200
+        return dump_response(DocumentSummaryStatusResponse, result), 200

api/controllers/console/datasets/datasets_segments.py

@@ -173,7 +173,7 @@ def _get_segment_for_document(
         raise NotFound("Document not found.")
 
     segment_ref = DatasetRefService.create_segment_ref(document_ref, segment_id)
-    segment = SegmentService.get_segment_by_ref(segment_ref)
+    segment = SegmentService.get_segment_by_ref(segment_ref, db.session())
     if not segment:
         raise NotFound("Segment not found.")
     return segment_ref, segment
@@ -193,16 +193,16 @@ class DatasetDocumentSegmentListApi(Resource):
     def get(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID):
         dataset_id_str = str(dataset_id)
         document_id_str = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
 
         if not document:
             raise NotFound("Document not found.")
@@ -278,7 +278,7 @@ class DatasetDocumentSegmentListApi(Resource):
         summaries: dict[str, str | None] = {}
         if segment_ids:
             summary_records = SummaryIndexService.get_segments_summaries(
-                segment_ids=segment_ids, dataset_id=dataset_id_str
+                segment_ids=segment_ids, dataset_id=dataset_id_str, session=db.session()
             )
             summaries = {chunk_id: summary.summary_content for chunk_id, summary in summary_records.items()}
 
@@ -303,14 +303,14 @@ class DatasetDocumentSegmentListApi(Resource):
     def delete(self, current_user: Account, dataset_id: UUID, document_id: UUID):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         segment_ids = request.args.getlist("segment_id")
@@ -319,10 +319,10 @@ class DatasetDocumentSegmentListApi(Resource):
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
-        SegmentService.delete_segments(segment_ids, document, dataset, db.session)
+        SegmentService.delete_segments(segment_ids, document, dataset, db.session())
         return "", 204
 
 
@@ -348,11 +348,11 @@ class DatasetDocumentSegmentApi(Resource):
         action: Literal["enable", "disable"],
     ):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # check user's model setting
@@ -362,7 +362,7 @@ class DatasetDocumentSegmentApi(Resource):
             raise Forbidden()
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
@@ -388,10 +388,10 @@ class DatasetDocumentSegmentApi(Resource):
         if cache_result is not None:
             raise InvalidActionError("Document is being indexed, please try again later")
         try:
-            SegmentService.update_segments_status(segment_ids, action, dataset, document, db.session)
+            SegmentService.update_segments_status(segment_ids, action, dataset, document, db.session())
         except Exception as e:
             raise InvalidActionError(str(e))
-        return dump_response(SimpleResultResponse, {"result": "success"}), 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment")
@@ -411,12 +411,12 @@ class DatasetDocumentSegmentAddApi(Resource):
     def post(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         if not current_user.is_dataset_editor:
@@ -438,15 +438,20 @@ class DatasetDocumentSegmentAddApi(Resource):
             except ProviderTokenNotInitError as ex:
                 raise ProviderNotInitializeError(ex.description)
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # validate args
         payload = SegmentCreatePayload.model_validate(console_ns.payload or {})
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
-        segment = type_cast(DocumentSegment, SegmentService.create_segment(payload_dict, document, dataset, db.session))
-        summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id_str)
+        segment = type_cast(
+            DocumentSegment,
+            SegmentService.create_segment(payload_dict, document, dataset, db.session()),
+        )
+        summary = SummaryIndexService.get_segment_summary(
+            segment_id=segment.id, dataset_id=dataset_id_str, session=db.session()
+        )
         response = {
             "data": segment_response_with_summary(segment, summary.summary_content if summary else None),
             "doc_form": document.doc_form,
@@ -472,21 +477,21 @@ class DatasetDocumentSegmentUpdateApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
@@ -518,9 +523,11 @@ class DatasetDocumentSegmentUpdateApi(Resource):
             segment,
             document,
             dataset,
-            db.session,
+            db.session(),
+        )
+        summary = SummaryIndexService.get_segment_summary(
+            segment_id=segment.id, dataset_id=dataset_id_str, session=db.session()
         )
-        summary = SummaryIndexService.get_segment_summary(segment_id=segment.id, dataset_id=dataset_id_str)
         response = {
             "data": segment_response_with_summary(segment, summary.summary_content if summary else None),
             "doc_form": document.doc_form,
@@ -541,26 +548,26 @@ class DatasetDocumentSegmentUpdateApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         segment_id_str = str(segment_id)
         _, segment = _get_segment_for_document(dataset, document, segment_id_str)
-        SegmentService.delete_segment(segment, document, dataset, db.session)
+        SegmentService.delete_segment(segment, document, dataset, db.session())
         return "", 204
 
 
@@ -583,12 +590,12 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
     def post(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
 
@@ -658,18 +665,18 @@ class ChildChunkAddApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         # check embedding model setting
@@ -693,7 +700,7 @@ class ChildChunkAddApi(Resource):
         # validate args
         try:
             payload = ChildChunkCreatePayload.model_validate(console_ns.payload or {})
-            child_chunk = SegmentService.create_child_chunk(payload.content, segment, document, dataset, db.session)
+            child_chunk = SegmentService.create_child_chunk(payload.content, segment, document, dataset, db.session())
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))
         return dump_response(ChildChunkDetailResponse, {"data": child_chunk}), 200
@@ -709,14 +716,14 @@ class ChildChunkAddApi(Resource):
     def get(self, current_tenant_id: str, dataset_id: UUID, document_id: UUID, segment_id: UUID):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         segment_id_str = str(segment_id)
@@ -759,21 +766,21 @@ class ChildChunkAddApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         segment_id_str = str(segment_id)
@@ -781,7 +788,7 @@ class ChildChunkAddApi(Resource):
         # validate args
         payload = ChildChunkBatchUpdatePayload.model_validate(console_ns.payload or {})
         try:
-            child_chunks = SegmentService.update_child_chunks(payload.chunks, segment, document, dataset, db.session)
+            child_chunks = SegmentService.update_child_chunks(payload.chunks, segment, document, dataset, db.session())
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))
         return dump_response(ChildChunkBatchUpdateResponse, {"data": child_chunks}), 200
@@ -811,31 +818,31 @@ class ChildChunkUpdateApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         segment_id_str = str(segment_id)
         segment_ref, _ = _get_segment_for_document(dataset, document, segment_id_str)
         child_chunk_id_str = str(child_chunk_id)
-        child_chunk = SegmentService.get_child_chunk_by_segment_ref(child_chunk_id_str, segment_ref)
+        child_chunk = SegmentService.get_child_chunk_by_segment_ref(child_chunk_id_str, segment_ref, db.session())
         if not child_chunk:
             raise NotFound("Child chunk not found.")
         try:
-            SegmentService.delete_child_chunk(child_chunk, dataset, db.session)
+            SegmentService.delete_child_chunk(child_chunk, dataset, db.session())
         except ChildChunkDeleteIndexServiceError as e:
             raise ChildChunkDeleteIndexError(str(e))
         return "", 204
@@ -862,34 +869,34 @@ class ChildChunkUpdateApi(Resource):
     ):
         # check dataset
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if not dataset:
             raise NotFound("Dataset not found.")
         # check user's model setting
         DatasetService.check_dataset_model_setting(dataset)
         # check document
         document_id_str = str(document_id)
-        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session)
+        document = DocumentService.get_document(dataset_id_str, document_id_str, session=db.session())
         if not document:
             raise NotFound("Document not found.")
         # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
         if not current_user.is_dataset_editor:
             raise Forbidden()
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         segment_id_str = str(segment_id)
         segment_ref, segment = _get_segment_for_document(dataset, document, segment_id_str)
         child_chunk_id_str = str(child_chunk_id)
-        child_chunk = SegmentService.get_child_chunk_by_segment_ref(child_chunk_id_str, segment_ref)
+        child_chunk = SegmentService.get_child_chunk_by_segment_ref(child_chunk_id_str, segment_ref, db.session())
         if not child_chunk:
             raise NotFound("Child chunk not found.")
         # validate args
         try:
             payload = ChildChunkUpdatePayload.model_validate(console_ns.payload or {})
             child_chunk = SegmentService.update_child_chunk(
-                payload.content, child_chunk, segment, document, dataset, db.session
+                payload.content, child_chunk, segment, document, dataset, db.session()
             )
         except ChildChunkIndexingServiceError as e:
             raise ChildChunkIndexingError(str(e))

api/controllers/console/datasets/external.py

@@ -1,20 +1,16 @@
+from datetime import datetime
 from typing import Any
 from uuid import UUID
 
 from flask import request
-from flask_restx import Resource, fields, marshal
-from pydantic import BaseModel, Field, RootModel
+from flask_restx import Resource
+from pydantic import AliasChoices, BaseModel, Field, field_validator
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
 from controllers.common.fields import UsageCountResponse
-from controllers.common.schema import (
-    get_or_create_model,
-    query_params_from_model,
-    register_response_schema_models,
-    register_schema_models,
-)
+from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.wraps import with_session
 from controllers.console.datasets.error import DatasetNameDuplicateError
@@ -28,21 +24,9 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
-from extensions.ext_database import db
 from fields.base import ResponseModel
-from fields.dataset_fields import (
-    dataset_detail_fields,
-    dataset_retrieval_model_fields,
-    doc_metadata_fields,
-    external_knowledge_info_fields,
-    external_retrieval_model_fields,
-    icon_info_fields,
-    keyword_setting_fields,
-    reranking_model_fields,
-    tag_fields,
-    vector_setting_fields,
-    weighted_score_fields,
-)
+from fields.dataset_fields import DatasetDetailResponse
+from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
 from services.dataset_service import DatasetService
@@ -51,50 +35,10 @@ from services.external_knowledge_service import ExternalDatasetService
 from services.hit_testing_service import HitTestingService
 from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
 
-register_response_schema_models(console_ns, UsageCountResponse)
-
-
-def _build_dataset_detail_model():
-    keyword_setting_model = get_or_create_model("DatasetKeywordSetting", keyword_setting_fields)
-    vector_setting_model = get_or_create_model("DatasetVectorSetting", vector_setting_fields)
-
-    weighted_score_fields_copy = weighted_score_fields.copy()
-    weighted_score_fields_copy["keyword_setting"] = fields.Nested(keyword_setting_model)
-    weighted_score_fields_copy["vector_setting"] = fields.Nested(vector_setting_model)
-    weighted_score_model = get_or_create_model("DatasetWeightedScore", weighted_score_fields_copy)
-
-    reranking_model = get_or_create_model("DatasetRerankingModel", reranking_model_fields)
-
-    dataset_retrieval_model_fields_copy = dataset_retrieval_model_fields.copy()
-    dataset_retrieval_model_fields_copy["reranking_model"] = fields.Nested(reranking_model)
-    dataset_retrieval_model_fields_copy["weights"] = fields.Nested(weighted_score_model, allow_null=True)
-    dataset_retrieval_model = get_or_create_model("DatasetRetrievalModel", dataset_retrieval_model_fields_copy)
-
-    tag_model = get_or_create_model("Tag", tag_fields)
-    doc_metadata_model = get_or_create_model("DatasetDocMetadata", doc_metadata_fields)
-    external_knowledge_info_model = get_or_create_model("ExternalKnowledgeInfo", external_knowledge_info_fields)
-    external_retrieval_model = get_or_create_model("ExternalRetrievalModel", external_retrieval_model_fields)
-    icon_info_model = get_or_create_model("DatasetIconInfo", icon_info_fields)
-
-    dataset_detail_fields_copy = dataset_detail_fields.copy()
-    dataset_detail_fields_copy["retrieval_model_dict"] = fields.Nested(dataset_retrieval_model)
-    dataset_detail_fields_copy["tags"] = fields.List(fields.Nested(tag_model))
-    dataset_detail_fields_copy["external_knowledge_info"] = fields.Nested(external_knowledge_info_model)
-    dataset_detail_fields_copy["external_retrieval_model"] = fields.Nested(external_retrieval_model, allow_null=True)
-    dataset_detail_fields_copy["doc_metadata"] = fields.List(fields.Nested(doc_metadata_model))
-    dataset_detail_fields_copy["icon_info"] = fields.Nested(icon_info_model)
-    return get_or_create_model("DatasetDetail", dataset_detail_fields_copy)
-
-
-try:
-    dataset_detail_model = console_ns.models["DatasetDetail"]
-except KeyError:
-    dataset_detail_model = _build_dataset_detail_model()
-
 
 class ExternalKnowledgeApiPayload(BaseModel):
     name: str = Field(..., min_length=1, max_length=40)
-    settings: dict[str, object]
+    settings: dict[str, Any]
 
 
 class ExternalDatasetCreatePayload(BaseModel):
@@ -102,15 +46,13 @@ class ExternalDatasetCreatePayload(BaseModel):
     external_knowledge_id: str
     name: str = Field(..., min_length=1, max_length=100)
     description: str | None = Field(None, max_length=400)
-    external_retrieval_model: dict[str, object] | None = Field(default=None)
+    external_retrieval_model: dict[str, Any] | None = None
 
 
 class ExternalHitTestingPayload(BaseModel):
     query: str
-    external_retrieval_model: dict[str, object] | None = Field(default=None)
-    metadata_filtering_conditions: dict[str, object] | None = Field(
-        default=None,
-    )
+    external_retrieval_model: dict[str, Any] | None = None
+    metadata_filtering_conditions: dict[str, Any] | None = None
 
 
 class BedrockRetrievalPayload(BaseModel):
@@ -125,7 +67,7 @@ class ExternalApiTemplateListQuery(BaseModel):
     keyword: str | None = Field(default=None, description="Search keyword")
 
 
-class ExternalKnowledgeDatasetBindingResponse(ResponseModel):
+class ExternalKnowledgeApiBindingResponse(ResponseModel):
     id: str
     name: str
 
@@ -135,22 +77,52 @@ class ExternalKnowledgeApiResponse(ResponseModel):
     tenant_id: str
     name: str
     description: str
-    settings: dict[str, Any] | None = Field(default=None)
-    dataset_bindings: list[ExternalKnowledgeDatasetBindingResponse] = Field(default_factory=list)
+    settings: dict[str, Any] | None = Field(validation_alias=AliasChoices("settings_dict", "settings"))
+    dataset_bindings: list[ExternalKnowledgeApiBindingResponse]
     created_by: str
     created_at: str
 
+    @field_validator("created_at", mode="before")
+    @classmethod
+    def _normalize_created_at(cls, value: datetime | str) -> str:
+        if isinstance(value, datetime):
+            return value.isoformat()
+        return value
+
 
 class ExternalKnowledgeApiListResponse(ResponseModel):
     data: list[ExternalKnowledgeApiResponse]
     has_more: bool
     limit: int
-    total: int
+    total: int | None
     page: int
 
 
-class ExternalRetrievalTestResponse(RootModel[dict[str, Any] | list[dict[str, Any]]]):
-    root: dict[str, Any] | list[dict[str, Any]]
+class ExternalHitTestingQueryResponse(ResponseModel):
+    content: str
+
+
+class ExternalHitTestingRecordResponse(ResponseModel):
+    content: str | None = None
+    title: str | None = None
+    score: float | None = None
+    metadata: dict[str, Any] | None = None
+
+
+class ExternalHitTestingResponse(ResponseModel):
+    query: ExternalHitTestingQueryResponse
+    records: list[ExternalHitTestingRecordResponse]
+
+
+class BedrockRetrievalRecordResponse(ResponseModel):
+    metadata: dict[str, Any] | None = None
+    score: float
+    title: str | None = None
+    content: str | None = None
+
+
+class BedrockRetrievalResponse(ResponseModel):
+    records: list[BedrockRetrievalRecordResponse]
 
 
 register_schema_models(
@@ -163,9 +135,16 @@ register_schema_models(
 )
 register_response_schema_models(
     console_ns,
+    UsageCountResponse,
+    DatasetDetailResponse,
+    ExternalKnowledgeApiBindingResponse,
     ExternalKnowledgeApiResponse,
     ExternalKnowledgeApiListResponse,
-    ExternalRetrievalTestResponse,
+    ExternalHitTestingQueryResponse,
+    ExternalHitTestingRecordResponse,
+    ExternalHitTestingResponse,
+    BedrockRetrievalRecordResponse,
+    BedrockRetrievalResponse,
 )
 
 
@@ -189,24 +168,26 @@ class ExternalApiTemplateListApi(Resource):
         external_knowledge_apis, total = ExternalDatasetService.get_external_knowledge_apis(
             query.page, query.limit, current_tenant_id, query.keyword
         )
-        response = {
-            "data": [item.to_dict() for item in external_knowledge_apis],
-            "has_more": len(external_knowledge_apis) == query.limit,
-            "limit": query.limit,
-            "total": total,
-            "page": query.page,
-        }
-        return response, 200
+        return ExternalKnowledgeApiListResponse(
+            data=[ExternalKnowledgeApiResponse.model_validate(item) for item in external_knowledge_apis],
+            has_more=len(external_knowledge_apis) == query.limit,
+            limit=query.limit,
+            total=total,
+            page=query.page,
+        ).model_dump(mode="json"), 200
 
-    @setup_required
-    @login_required
-    @account_initialization_required
+    @console_ns.doc("create_external_api_template")
+    @console_ns.doc(description="Create external knowledge API template")
     @console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
     @console_ns.response(
         201,
         "External API template created successfully",
         console_ns.models[ExternalKnowledgeApiResponse.__name__],
     )
+    @console_ns.response(403, "Permission denied")
+    @setup_required
+    @login_required
+    @account_initialization_required
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -229,7 +210,7 @@ class ExternalApiTemplateListApi(Resource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        return external_knowledge_api.to_dict(), 201
+        return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 201
 
 
 @console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>")
@@ -256,17 +237,21 @@ class ExternalApiTemplateApi(Resource):
         if external_knowledge_api is None:
             raise NotFound("API template not found.")
 
-        return external_knowledge_api.to_dict(), 200
+        return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 200
 
+    @console_ns.doc("update_external_api_template")
+    @console_ns.doc(description="Update external knowledge API template")
+    @console_ns.doc(params={"external_knowledge_api_id": "External knowledge API ID"})
+    @console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
     @console_ns.response(
         200,
         "External API template updated successfully",
         console_ns.models[ExternalKnowledgeApiResponse.__name__],
     )
+    @console_ns.response(404, "Template not found")
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
     @with_current_user
     @with_current_tenant_id
     @with_session
@@ -284,7 +269,7 @@ class ExternalApiTemplateApi(Resource):
             session=session,
         )
 
-        return external_knowledge_api.to_dict(), 200
+        return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 200
 
     @setup_required
     @login_required
@@ -299,7 +284,9 @@ class ExternalApiTemplateApi(Resource):
         if not (current_user.has_edit_permission or current_user.is_dataset_operator):
             raise Forbidden()
 
-        ExternalDatasetService.delete_external_knowledge_api(session, current_tenant_id, external_knowledge_api_id_str)
+        ExternalDatasetService.delete_external_knowledge_api(
+            current_tenant_id, external_knowledge_api_id_str, session=session
+        )
         return "", 204
 
 
@@ -318,11 +305,9 @@ class ExternalApiUseCheckApi(Resource):
         external_knowledge_api_id_str = str(external_knowledge_api_id)
 
         external_knowledge_api_is_using, count = ExternalDatasetService.external_knowledge_api_use_check(
-            session,
-            external_knowledge_api_id_str,
-            current_tenant_id,
+            external_knowledge_api_id_str, current_tenant_id, session=session
         )
-        return {"is_using": external_knowledge_api_is_using, "count": count}, 200
+        return UsageCountResponse(is_using=external_knowledge_api_is_using, count=count).model_dump(mode="json"), 200
 
 
 @console_ns.route("/datasets/external")
@@ -330,7 +315,9 @@ class ExternalDatasetCreateApi(Resource):
     @console_ns.doc("create_external_dataset")
     @console_ns.doc(description="Create external knowledge dataset")
     @console_ns.expect(console_ns.models[ExternalDatasetCreatePayload.__name__])
-    @console_ns.response(201, "External dataset created successfully", dataset_detail_model)
+    @console_ns.response(
+        201, "External dataset created successfully", console_ns.models[DatasetDetailResponse.__name__]
+    )
     @console_ns.response(400, "Invalid parameters")
     @console_ns.response(403, "Permission denied")
     @setup_required
@@ -360,16 +347,16 @@ class ExternalDatasetCreateApi(Resource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
-        item = marshal(dataset, dataset_detail_fields)
-        dataset_id_str = item["id"]
+        dataset_id_str = str(dataset.id)
         permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,
             [dataset_id_str],
+            session=session,
         )
-        item["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
-
-        return item, 201
+        data = DatasetDetailResponse.model_validate(dataset).model_dump(mode="json")
+        data["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
+        return data, 201
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/external-hit-testing")
@@ -381,7 +368,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
     @console_ns.response(
         200,
         "External hit testing completed successfully",
-        console_ns.models[ExternalRetrievalTestResponse.__name__],
+        console_ns.models[ExternalHitTestingResponse.__name__],
     )
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(400, "Invalid parameters")
@@ -393,12 +380,12 @@ class ExternalKnowledgeHitTestingApi(Resource):
     @with_session
     def post(self, session: Session, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, session)
         if dataset is None:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, session)
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 
@@ -415,7 +402,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
                 metadata_filtering_conditions=payload.metadata_filtering_conditions,
             )
 
-            return response
+            return dump_response(ExternalHitTestingResponse, response)
         except Exception as e:
             raise InternalServerError(str(e))
 
@@ -426,11 +413,7 @@ class BedrockRetrievalApi(Resource):
     @console_ns.doc("bedrock_retrieval_test")
     @console_ns.doc(description="Bedrock retrieval test (internal use only)")
     @console_ns.expect(console_ns.models[BedrockRetrievalPayload.__name__])
-    @console_ns.response(
-        200,
-        "Bedrock retrieval test completed",
-        console_ns.models[ExternalRetrievalTestResponse.__name__],
-    )
+    @console_ns.response(200, "Bedrock retrieval test completed", console_ns.models[BedrockRetrievalResponse.__name__])
     def post(self):
         payload = BedrockRetrievalPayload.model_validate(console_ns.payload or {})
 
@@ -438,4 +421,4 @@ class BedrockRetrievalApi(Resource):
         result = ExternalDatasetTestService.knowledge_retrieval(
             payload.retrieval_setting, payload.query, payload.knowledge_id
         )
-        return result, 200
+        return dump_response(BedrockRetrievalResponse, result), 200

api/controllers/console/datasets/hit_testing_base.py

@@ -86,12 +86,12 @@ class DatasetsHitTestingBase:
         dataset_id: str, current_user: Account | None = None, current_tenant_id: str | None = None
     ) -> Dataset:
         current_user, _ = resolve_account_fallback(current_user, current_tenant_id)
-        dataset = DatasetService.get_dataset(dataset_id, db.session)
+        dataset = DatasetService.get_dataset(dataset_id, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
 
         try:
-            DatasetService.check_dataset_permission(dataset, current_user, db.session)
+            DatasetService.check_dataset_permission(dataset, current_user, db.session())
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
 

api/controllers/console/datasets/metadata.py

@@ -61,13 +61,13 @@ class DatasetMetadataCreateApi(Resource):
         metadata_args = MetadataArgs.model_validate(console_ns.payload or {})
 
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
         metadata = MetadataService.create_metadata(
-            db.session(), dataset_id_str, metadata_args, current_user, current_tenant_id
+            dataset_id_str, metadata_args, current_user, current_tenant_id, session=db.session()
         )
         return dump_response(DatasetMetadataResponse, metadata), 201
 
@@ -81,10 +81,10 @@ class DatasetMetadataCreateApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_CREATE_AND_MANAGEMENT)
     def get(self, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        metadata = MetadataService.get_dataset_metadatas(db.session(), dataset)
+        metadata = MetadataService.get_dataset_metadatas(dataset, session=db.session())
         return dump_response(DatasetMetadataListResponse, metadata), 200
 
 
@@ -105,13 +105,13 @@ class DatasetMetadataApi(Resource):
 
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
         metadata = MetadataService.update_metadata_name(
-            db.session(), dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
+            dataset_id_str, metadata_id_str, name, current_user, current_tenant_id, session=db.session()
         )
         return dump_response(DatasetMetadataResponse, metadata), 200
 
@@ -125,12 +125,12 @@ class DatasetMetadataApi(Resource):
     def delete(self, current_user: Account, dataset_id: UUID, metadata_id: UUID):
         dataset_id_str = str(dataset_id)
         metadata_id_str = str(metadata_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
-        MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
+        MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session=db.session())
         # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
         return "", 204
 
@@ -162,16 +162,16 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     def post(self, current_user: Account, dataset_id: UUID, action: Literal["enable", "disable"]):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
         match action:
             case "enable":
-                MetadataService.enable_built_in_field(db.session(), dataset)
+                MetadataService.enable_built_in_field(dataset, session=db.session())
             case "disable":
-                MetadataService.disable_built_in_field(db.session(), dataset)
+                MetadataService.disable_built_in_field(dataset, session=db.session())
         # Frontend callers only await success and invalidate metadata caches; no response body is consumed.
         return "", 204
 
@@ -191,14 +191,14 @@ class DocumentMetadataEditApi(Resource):
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     def post(self, current_user: Account, dataset_id: UUID):
         dataset_id_str = str(dataset_id)
-        dataset = DatasetService.get_dataset(dataset_id_str, db.session)
+        dataset = DatasetService.get_dataset(dataset_id_str, db.session())
         if dataset is None:
             raise NotFound("Dataset not found.")
-        DatasetService.check_dataset_permission(dataset, current_user, db.session)
+        DatasetService.check_dataset_permission(dataset, current_user, db.session())
 
         metadata_args = MetadataOperationData.model_validate(console_ns.payload or {})
 
-        MetadataService.update_documents_metadata(db.session(), dataset, metadata_args, current_user)
+        MetadataService.update_documents_metadata(dataset, metadata_args, current_user, session=db.session())
 
         # Frontend callers only await success and invalidate caches; no response body is consumed.
         return "", 204

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

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden, NotFound
 
 from configs import dify_config
-from controllers.common.fields import RedirectResponse, SimpleResultResponse
+from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
@@ -19,11 +19,14 @@ from controllers.console.wraps import (
     with_current_tenant_id,
     with_current_user,
 )
+from core.entities.provider_entities import ProviderConfig
 from core.plugin.entities.plugin_daemon import PluginOAuthAuthorizationUrlResponse
 from core.plugin.impl.oauth import OAuthHandler
+from core.tools.entities.common_entities import I18nObject
+from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
+from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
 from models.provider_ids import DatasourceProviderID
@@ -33,7 +36,9 @@ from services.plugin.oauth_service import OAuthProxyService
 
 class DatasourceCredentialPayload(BaseModel):
     name: str | None = Field(default=None, max_length=100)
-    credentials: dict[str, Any]
+    credentials: dict[str, Any] = Field(
+        description="Plugin-defined credential parameters. The schema is declared by the datasource provider."
+    )
 
 
 class DatasourceCredentialDeletePayload(BaseModel):
@@ -43,11 +48,17 @@ class DatasourceCredentialDeletePayload(BaseModel):
 class DatasourceCredentialUpdatePayload(BaseModel):
     credential_id: str
     name: str | None = Field(default=None, max_length=100)
-    credentials: dict[str, Any] | None = Field(default=None)
+    credentials: dict[str, Any] | None = Field(
+        default=None,
+        description="Plugin-defined credential parameters. The schema is declared by the datasource provider.",
+    )
 
 
 class DatasourceCustomClientPayload(BaseModel):
-    client_params: dict[str, Any] | None = Field(default=None)
+    client_params: dict[str, Any] | None = Field(
+        default=None,
+        description="Plugin-defined OAuth client parameters. The schema is declared by the datasource provider.",
+    )
     enable_oauth_custom_client: bool | None = None
 
 
@@ -71,8 +82,48 @@ class DatasourceOAuthCallbackQuery(BaseModel):
     context_id: str | None = Field(default=None, description="OAuth proxy context ID")
 
 
-class DatasourceCredentialsResponse(ResponseModel):
-    result: Any
+class DatasourceCredentialResponse(ResponseModel):
+    credential: dict[str, Any] = Field(
+        description="Obfuscated plugin-defined credential parameters from the datasource provider."
+    )
+    type: str
+    name: str
+    avatar_url: str | None
+    id: str
+    is_default: bool
+
+
+class DatasourceCredentialListResponse(ResponseModel):
+    result: list[DatasourceCredentialResponse]
+
+
+class DatasourceOAuthSchemaResponse(ResponseModel):
+    client_schema: list[ProviderConfig]
+    credentials_schema: list[ProviderConfig]
+    oauth_custom_client_params: dict[str, Any] | None = Field(
+        description="Masked plugin-defined OAuth client parameters, when configured for the tenant."
+    )
+    is_oauth_custom_client_enabled: bool
+    is_system_oauth_params_exists: bool
+    redirect_uri: str
+
+
+class DatasourceProviderAuthResponse(ResponseModel):
+    author: str
+    provider: str
+    plugin_id: str
+    plugin_unique_identifier: str
+    icon: str
+    name: str
+    label: I18nObject
+    description: I18nObject
+    credential_schema: list[ProviderConfig]
+    oauth_schema: DatasourceOAuthSchemaResponse | None
+    credentials_list: list[DatasourceCredentialResponse]
+
+
+class DatasourceProviderAuthListResponse(ResponseModel):
+    result: list[DatasourceProviderAuthResponse]
 
 
 register_schema_models(
@@ -88,9 +139,9 @@ register_schema_models(
 )
 register_response_schema_models(
     console_ns,
-    DatasourceCredentialsResponse,
+    DatasourceCredentialListResponse,
+    DatasourceProviderAuthListResponse,
     PluginOAuthAuthorizationUrlResponse,
-    RedirectResponse,
     SimpleResultResponse,
 )
 
@@ -100,7 +151,7 @@ class DatasourcePluginOAuthAuthorizationUrl(Resource):
     @console_ns.doc(params=query_params_from_model(DatasourceOAuthAuthorizationQuery))
     @console_ns.response(
         200,
-        "Authorization URL retrieved successfully",
+        "Datasource OAuth authorization URL generated successfully",
         console_ns.models[PluginOAuthAuthorizationUrlResponse.__name__],
     )
     @setup_required
@@ -140,7 +191,8 @@ class DatasourcePluginOAuthAuthorizationUrl(Resource):
             redirect_uri=redirect_uri,
             system_credentials=oauth_config,
         )
-        response = make_response(jsonable_encoder(authorization_url_response))
+        # response-contract:ignore cookie-bearing Flask response
+        response = make_response(dump_response(PluginOAuthAuthorizationUrlResponse, authorization_url_response))
         response.set_cookie(
             "context_id",
             context_id,
@@ -154,11 +206,8 @@ class DatasourcePluginOAuthAuthorizationUrl(Resource):
 @console_ns.route("/oauth/plugin/<path:provider_id>/datasource/callback")
 class DatasourceOAuthCallback(Resource):
     @console_ns.doc(params=query_params_from_model(DatasourceOAuthCallbackQuery))
-    @console_ns.response(
-        302,
-        "Redirect to console OAuth callback page",
-        console_ns.models[RedirectResponse.__name__],
-    )
+    # response-contract:ignore redirect response
+    @console_ns.response(302, "Redirect to OAuth callback page")
     @setup_required
     def get(self, provider_id: str):
         context_id = request.cookies.get("context_id") or request.args.get("context_id")
@@ -217,7 +266,9 @@ class DatasourceOAuthCallback(Resource):
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>")
 class DatasourceAuth(Resource):
     @console_ns.expect(console_ns.models[DatasourceCredentialPayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(
+        200, "Datasource credential created successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -238,12 +289,16 @@ class DatasourceAuth(Resource):
             )
         except CredentialsValidateFailedError as ex:
             raise ValueError(str(ex))
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
+    @console_ns.response(
+        200,
+        "Datasource credentials retrieved successfully",
+        console_ns.models[DatasourceCredentialListResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[DatasourceCredentialsResponse.__name__])
     @with_current_user
     @with_current_tenant_id
     def get(self, current_tenant_id: str, user: Account, provider_id: str):
@@ -255,8 +310,9 @@ class DatasourceAuth(Resource):
             provider=datasource_provider_id.provider_name,
             plugin_id=datasource_provider_id.plugin_id,
             user=user,
+            session=db.session(),
         )
-        return {"result": datasources}, 200
+        return dump_response(DatasourceCredentialListResponse, {"result": datasources}), 200
 
 
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/delete")
@@ -281,14 +337,17 @@ class DatasourceAuthDeleteApi(Resource):
             auth_id=payload.credential_id,
             provider=provider_name,
             plugin_id=plugin_id,
+            session=db.session(),
         )
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/update")
 class DatasourceAuthUpdateApi(Resource):
     @console_ns.expect(console_ns.models[DatasourceCredentialUpdatePayload.__name__])
-    @console_ns.response(201, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(
+        201, "Datasource credential updated successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -308,39 +367,53 @@ class DatasourceAuthUpdateApi(Resource):
             credentials=payload.credentials or {},
             name=payload.name,
         )
-        return {"result": "success"}, 201
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 201
 
 
 @console_ns.route("/auth/plugin/datasource/list")
 class DatasourceAuthListApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[DatasourceCredentialsResponse.__name__])
+    @console_ns.response(
+        200,
+        "Datasource credentials retrieved successfully",
+        console_ns.models[DatasourceProviderAuthListResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @with_current_tenant_id
     def get(self, current_tenant_id: str):
         datasource_provider_service = DatasourceProviderService()
-        datasources = datasource_provider_service.get_all_datasource_credentials(tenant_id=current_tenant_id)
-        return {"result": jsonable_encoder(datasources)}, 200
+        datasources = datasource_provider_service.get_all_datasource_credentials(
+            tenant_id=current_tenant_id, session=db.session()
+        )
+        return dump_response(DatasourceProviderAuthListResponse, {"result": datasources}), 200
 
 
 @console_ns.route("/auth/plugin/datasource/default-list")
 class DatasourceHardCodeAuthListApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[DatasourceCredentialsResponse.__name__])
+    @console_ns.response(
+        200,
+        "Default datasource credentials retrieved successfully",
+        console_ns.models[DatasourceProviderAuthListResponse.__name__],
+    )
     @setup_required
     @login_required
     @account_initialization_required
     @with_current_tenant_id
     def get(self, current_tenant_id: str):
         datasource_provider_service = DatasourceProviderService()
-        datasources = datasource_provider_service.get_hard_code_datasource_credentials(tenant_id=current_tenant_id)
-        return {"result": jsonable_encoder(datasources)}, 200
+        datasources = datasource_provider_service.get_hard_code_datasource_credentials(
+            tenant_id=current_tenant_id, session=db.session()
+        )
+        return dump_response(DatasourceProviderAuthListResponse, {"result": datasources}), 200
 
 
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/custom-client")
 class DatasourceAuthOauthCustomClient(Resource):
     @console_ns.expect(console_ns.models[DatasourceCustomClientPayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(
+        200, "Datasource OAuth custom client saved successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
     @setup_required
     @login_required
     @account_initialization_required
@@ -357,7 +430,7 @@ class DatasourceAuthOauthCustomClient(Resource):
             client_params=payload.client_params or {},
             enabled=payload.enable_oauth_custom_client or False,
         )
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
     @setup_required
     @login_required
@@ -371,7 +444,7 @@ class DatasourceAuthOauthCustomClient(Resource):
             tenant_id=current_tenant_id,
             datasource_provider_id=datasource_provider_id,
         )
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/default")
@@ -393,7 +466,7 @@ class DatasourceAuthDefaultApi(Resource):
             datasource_provider_id=datasource_provider_id,
             credential_id=payload.id,
         )
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
 
 
 @console_ns.route("/auth/plugin/datasource/<path:provider_id>/update-name")
@@ -416,4 +489,4 @@ class DatasourceUpdateProviderNameApi(Resource):
             name=payload.name,
             credential_id=payload.credential_id,
         )
-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200

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

@@ -3,12 +3,13 @@ from typing import Any
 from flask_restx import (  # type: ignore
     Resource,  # type: ignore
 )
-from pydantic import BaseModel, RootModel
+from pydantic import BaseModel
 
-from controllers.common.schema import register_response_schema_models, register_schema_models
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required, with_current_user
+from extensions.ext_database import db
 from libs.login import login_required
 from models import Account
 from models.dataset import Pipeline
@@ -21,18 +22,13 @@ class Parser(BaseModel):
     credential_id: str | None = None
 
 
-class DataSourceContentPreviewResponse(RootModel[Any]):
-    root: Any
-
-
 register_schema_models(console_ns, Parser)
-register_response_schema_models(console_ns, DataSourceContentPreviewResponse)
 
 
 @console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/published/datasource/nodes/<string:node_id>/preview")
 class DataSourceContentPreviewApi(Resource):
     @console_ns.expect(console_ns.models[Parser.__name__])
-    @console_ns.response(200, "Success", console_ns.models[DataSourceContentPreviewResponse.__name__])
+    @console_ns.response(200, "Success")
     @setup_required
     @login_required
     @account_initialization_required
@@ -46,7 +42,7 @@ class DataSourceContentPreviewApi(Resource):
 
         inputs = args.inputs
         datasource_type = args.datasource_type
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         preview_content = rag_pipeline_service.run_datasource_node_preview(
             pipeline=pipeline,
             node_id=node_id,

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

@@ -108,7 +108,10 @@ class PipelineTemplateListApi(Resource):
         query = PipelineTemplateListQuery.model_validate(request.args.to_dict(flat=True))
         # get pipeline templates
         pipeline_templates = RagPipelineService.get_pipeline_templates(
-            session, query.type, query.language, current_tenant_id
+            type=query.type,
+            language=query.language,
+            current_tenant_id=current_tenant_id,
+            session=session,
         )
         return dump_response(PipelineTemplateListResponse, pipeline_templates), 200
 
@@ -124,8 +127,11 @@ class PipelineTemplateDetailApi(Resource):
     @with_session
     def get(self, session: Session, template_id: str) -> JsonResponseWithStatus:
         query = PipelineTemplateDetailQuery.model_validate(request.args.to_dict(flat=True))
-        rag_pipeline_service = RagPipelineService()
-        pipeline_template = rag_pipeline_service.get_pipeline_template_detail(session, template_id, query.type)
+        pipeline_template = RagPipelineService.get_pipeline_template_detail(
+            template_id,
+            type=query.type,
+            session=session,
+        )
         if pipeline_template is None:
             raise NotFound("Pipeline template not found from upstream service.")
         return dump_response(PipelineTemplateDetailResponse, pipeline_template), 200
@@ -145,7 +151,7 @@ class CustomizedPipelineTemplateApi(Resource):
         payload = CustomizedPipelineTemplatePayload.model_validate(console_ns.payload or {})
         pipeline_template_info = PipelineTemplateInfoEntity.model_validate(payload.model_dump())
         RagPipelineService.update_customized_pipeline_template(
-            template_id, pipeline_template_info, current_user, current_tenant_id
+            template_id, pipeline_template_info, current_user, current_tenant_id, session=db.session()
         )
         return "", 204
 
@@ -156,7 +162,7 @@ class CustomizedPipelineTemplateApi(Resource):
     @enterprise_license_required
     @with_current_tenant_id
     def delete(self, current_tenant_id: str, template_id: str) -> tuple[str, int]:
-        RagPipelineService.delete_customized_pipeline_template(template_id, current_tenant_id)
+        RagPipelineService.delete_customized_pipeline_template(template_id, current_tenant_id, session=db.session())
         return "", 204
 
     @setup_required
@@ -188,8 +194,8 @@ class PublishCustomizedPipelineTemplateApi(Resource):
     @with_current_tenant_id
     def post(self, current_tenant_id: str, current_user: Account, pipeline_id: str) -> tuple[str, int]:
         payload = CustomizedPipelineTemplatePayload.model_validate(console_ns.payload or {})
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         rag_pipeline_service.publish_customized_pipeline_template(
-            pipeline_id, payload.model_dump(), current_user, current_tenant_id
+            pipeline_id, payload.model_dump(), current_user, current_tenant_id, session=db.session()
         )
         return "", 204

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

@@ -65,7 +65,7 @@ class CreateRagPipelineDatasetApi(Resource):
             yaml_content=payload.yaml_content,
         )
         try:
-            rag_pipeline_dsl_service = RagPipelineDslService(db.session)
+            rag_pipeline_dsl_service = RagPipelineDslService(db.session())
             import_info = rag_pipeline_dsl_service.create_rag_pipeline_dataset(
                 tenant_id=current_tenant_id,
                 rag_pipeline_dataset_create_entity=rag_pipeline_dataset_create_entity,
@@ -75,7 +75,7 @@ class CreateRagPipelineDatasetApi(Resource):
                     current_tenant_id,
                     import_info["dataset_id"],
                     rag_pipeline_dataset_create_entity.partial_member_list,
-                    db.session,
+                    db.session(),
                 )
             db.session.commit()
         except services.errors.dataset.DatasetNameDuplicateError:
@@ -110,6 +110,6 @@ class CreateEmptyRagPipelineDatasetApi(Resource):
                 permission=DatasetPermissionEnum.ONLY_ME,
                 partial_member_list=None,
             ),
-            session=db.session,
+            session=db.session(),
         )
         return dump_response(DatasetDetailResponse, dataset), 201

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

@@ -98,7 +98,7 @@ class RagPipelineVariableCollectionApi(Resource):
         query = PaginationQuery.model_validate(request.args.to_dict())
 
         # fetch draft workflow by app_model
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow_exist = rag_pipeline_service.is_workflow_exist(pipeline=pipeline)
         if not workflow_exist:
             raise DraftWorkflowNotExist()
@@ -290,7 +290,7 @@ class RagPipelineVariableResetApi(Resource):
             session=db.session(),
         )
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         draft_workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
         if draft_workflow is None:
             raise NotFoundError(
@@ -347,7 +347,7 @@ class RagPipelineEnvironmentVariableCollectionApi(Resource):
         Get draft workflow
         """
         # fetch draft workflow by app_model
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
         if workflow is None:
             raise DraftWorkflowNotExist()

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

@@ -197,7 +197,7 @@ class DraftRagPipelineApi(Resource):
         Get draft rag pipeline's workflow
         """
         # fetch draft workflow by app_model
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
 
         if not workflow:
@@ -231,7 +231,7 @@ class DraftRagPipelineApi(Resource):
                 return {"message": "Invalid JSON data"}, 400
         else:
             abort(415)
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
 
         try:
             environment_variables_list = Workflow.normalize_environment_variable_mappings(
@@ -283,7 +283,7 @@ class RagPipelineDraftRunIterationNodeApi(Resource):
 
         try:
             response = PipelineGenerateService.generate_single_iteration(
-                pipeline=pipeline, user=current_user, node_id=node_id, args=args, streaming=True
+                pipeline=pipeline, user=current_user, node_id=node_id, args=args, session=db.session(), streaming=True
             )
 
             return helper.compact_generate_response(response)
@@ -318,7 +318,7 @@ class RagPipelineDraftRunLoopNodeApi(Resource):
 
         try:
             response = PipelineGenerateService.generate_single_loop(
-                pipeline=pipeline, user=current_user, node_id=node_id, args=args, streaming=True
+                pipeline=pipeline, user=current_user, node_id=node_id, args=args, session=db.session(), streaming=True
             )
 
             return helper.compact_generate_response(response)
@@ -343,8 +343,8 @@ class DraftRagPipelineRunApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_current_user
-    @get_rag_pipeline
     @with_session
+    @get_rag_pipeline
     def post(self, session: Session, current_user: Account, pipeline: Pipeline):
         """
         Run draft workflow
@@ -377,8 +377,8 @@ class PublishedRagPipelineRunApi(Resource):
     @edit_permission_required
     @rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
     @with_current_user
-    @get_rag_pipeline
     @with_session
+    @get_rag_pipeline
     def post(self, session: Session, current_user: Account, pipeline: Pipeline):
         """
         Run published workflow
@@ -419,7 +419,7 @@ class RagPipelinePublishedDatasourceNodeRunApi(Resource):
         """
         payload = DatasourceNodeRunPayload.model_validate(console_ns.payload or {})
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         return helper.compact_generate_response(
             PipelineGenerator.convert_to_event_stream(
                 rag_pipeline_service.run_datasource_workflow_node(
@@ -452,7 +452,7 @@ class RagPipelineDraftDatasourceNodeRunApi(Resource):
         """
         payload = DatasourceNodeRunPayload.model_validate(console_ns.payload or {})
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         return helper.compact_generate_response(
             PipelineGenerator.convert_to_event_stream(
                 rag_pipeline_service.run_datasource_workflow_node(
@@ -490,7 +490,7 @@ class RagPipelineDraftNodeRunApi(Resource):
         payload = NodeRunRequiredPayload.model_validate(console_ns.payload or {})
         inputs = payload.inputs
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow_node_execution = rag_pipeline_service.run_draft_workflow_node(
             pipeline=pipeline, node_id=node_id, user_inputs=inputs, account=current_user
         )
@@ -543,7 +543,7 @@ class PublishedRagPipelineApi(Resource):
         if not pipeline.is_published:
             return None
         # fetch published workflow by pipeline
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow = rag_pipeline_service.get_published_workflow(pipeline=pipeline)
 
         # return workflow, if not found, return None
@@ -564,9 +564,9 @@ class PublishedRagPipelineApi(Resource):
         """
         Publish workflow
         """
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow = rag_pipeline_service.publish_workflow(
-            session=db.session,  # type: ignore[reportArgumentType,arg-type]
+            session=db.session(),
             pipeline=pipeline,
             account=current_user,
         )
@@ -599,7 +599,7 @@ class DefaultRagPipelineBlockConfigsApi(Resource):
         Get default block config
         """
         # Get default block configs
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         return rag_pipeline_service.get_default_block_configs()
 
 
@@ -631,7 +631,7 @@ class DefaultRagPipelineBlockConfigApi(Resource):
                 raise ValueError("Invalid filters")
 
         # Get default block configs
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         return rag_pipeline_service.get_default_block_config(node_type=block_type, filters=filters)
 
 
@@ -666,7 +666,7 @@ class PublishedAllRagPipelineApi(Resource):
             if user_id != current_user.id:
                 raise Forbidden()
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         with sessionmaker(db.engine).begin() as session:
             workflows, has_more = rag_pipeline_service.get_all_published_workflow(
                 session=session,
@@ -698,7 +698,7 @@ class RagPipelineDraftWorkflowRestoreApi(Resource):
     @with_current_user
     @get_rag_pipeline
     def post(self, current_user: Account, pipeline: Pipeline, workflow_id: str):
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
 
         try:
             workflow = rag_pipeline_service.restore_published_workflow_to_draft(
@@ -743,7 +743,7 @@ class RagPipelineByIdApi(Resource):
         if not update_data:
             return {"message": "No valid fields to update"}, 400
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow_ref = WorkflowRefService.create_pipeline_workflow_ref(pipeline, workflow_id)
 
         # Create a session and manage the transaction
@@ -809,7 +809,7 @@ class PublishedRagPipelineSecondStepApi(Resource):
         """
         query = NodeIdQuery.model_validate(request.args.to_dict())
         node_id = query.node_id
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         variables = rag_pipeline_service.get_second_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=False)
         return {
             "variables": variables,
@@ -832,7 +832,7 @@ class PublishedRagPipelineFirstStepApi(Resource):
         """
         query = NodeIdQuery.model_validate(request.args.to_dict())
         node_id = query.node_id
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         variables = rag_pipeline_service.get_first_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=False)
         return {
             "variables": variables,
@@ -855,7 +855,7 @@ class DraftRagPipelineFirstStepApi(Resource):
         """
         query = NodeIdQuery.model_validate(request.args.to_dict())
         node_id = query.node_id
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         variables = rag_pipeline_service.get_first_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=True)
         return {
             "variables": variables,
@@ -879,7 +879,7 @@ class DraftRagPipelineSecondStepApi(Resource):
         query = NodeIdQuery.model_validate(request.args.to_dict())
         node_id = query.node_id
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         variables = rag_pipeline_service.get_second_step_parameters(pipeline=pipeline, node_id=node_id, is_draft=True)
         return {
             "variables": variables,
@@ -913,7 +913,7 @@ class RagPipelineWorkflowRunListApi(Resource):
             "limit": query.limit,
         }
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         result = rag_pipeline_service.get_rag_pipeline_paginate_workflow_runs(pipeline=pipeline, args=args)
 
         return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
@@ -936,7 +936,7 @@ class RagPipelineWorkflowRunDetailApi(Resource):
         """
         run_id_str = str(run_id)
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow_run = rag_pipeline_service.get_rag_pipeline_workflow_run(pipeline=pipeline, run_id=run_id_str)
         if workflow_run is None:
             raise NotFound("Workflow run not found")
@@ -962,7 +962,7 @@ class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
         """
         run_id_str = str(run_id)
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         user = cast("Account | EndUser", current_user)
         node_executions = rag_pipeline_service.get_rag_pipeline_workflow_run_node_executions(
             pipeline=pipeline,
@@ -998,7 +998,7 @@ class RagPipelineWorkflowLastRunApi(Resource):
     @account_initialization_required
     @get_rag_pipeline
     def get(self, pipeline: Pipeline, node_id: str):
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
         if not workflow:
             raise NotFound("Workflow not found")
@@ -1051,7 +1051,7 @@ class RagPipelineDatasourceVariableApi(Resource):
         """
         args = DatasourceVariablesPayload.model_validate(console_ns.payload or {}).model_dump()
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         workflow_node_execution = rag_pipeline_service.set_datasource_variables(
             pipeline=pipeline,
             args=args,
@@ -1074,6 +1074,6 @@ class RagPipelineRecommendedPluginApi(Resource):
     def get(self, current_tenant_id: str, current_user: Account):
         query = RagPipelineRecommendedPluginQuery.model_validate(request.args.to_dict())
 
-        rag_pipeline_service = RagPipelineService()
+        rag_pipeline_service = RagPipelineService(db.session())
         recommended_plugins = rag_pipeline_service.get_recommended_plugins(query.type, current_user, current_tenant_id)
         return recommended_plugins

api/controllers/console/datasets/wraps.py

@@ -2,6 +2,7 @@ from collections.abc import Callable
 from functools import wraps
 
 from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from controllers.console.datasets.error import PipelineNotFoundError
 from extensions.ext_database import db
@@ -22,9 +23,10 @@ def get_rag_pipeline[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
 
         del kwargs["pipeline_id"]
 
-        pipeline = db.session.scalar(
-            select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
-        )
+        stmt = select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
+        # Migrated handlers pass the request Session as args[1]; legacy handlers still use db.session.
+        session = args[1] if len(args) > 1 and isinstance(args[1], Session) else db.session
+        pipeline = session.scalar(stmt)
 
         if not pipeline:
             raise PipelineNotFoundError()

api/controllers/console/explore/audio.py

@@ -113,7 +113,7 @@ class ChatTextApi(InstalledAppResource):
 
             response = AudioService.transcript_tts(
                 app_model=app_model,
-                session=db.session,
+                session=db.session(),
                 text=text,
                 voice=voice,
                 message_ref=message_ref,

api/controllers/console/explore/conversation.py

@@ -111,7 +111,7 @@ class ConversationApi(InstalledAppResource):
 
         conversation_id = str(c_id)
         try:
-            ConversationService.delete(app_model, conversation_id, current_user)
+            ConversationService.delete(app_model, conversation_id, current_user, session=db.session())
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
@@ -140,7 +140,7 @@ class ConversationRenameApi(InstalledAppResource):
 
         try:
             conversation = ConversationService.rename(
-                app_model, conversation_id, current_user, payload.name, payload.auto_generate
+                app_model, conversation_id, current_user, payload.name, payload.auto_generate, session=db.session()
             )
             return (
                 TypeAdapter(SimpleConversation)
@@ -169,7 +169,7 @@ class ConversationPinApi(InstalledAppResource):
         conversation_id = str(c_id)
 
         try:
-            WebConversationService.pin(app_model, conversation_id, current_user)
+            WebConversationService.pin(app_model, conversation_id, current_user, db.session())
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
 
@@ -192,6 +192,6 @@ class ConversationUnPinApi(InstalledAppResource):
             raise NotChatAppError()
 
         conversation_id = str(c_id)
-        WebConversationService.unpin(app_model, conversation_id, current_user)
+        WebConversationService.unpin(app_model, conversation_id, current_user, db.session())
 
         return ResultResponse(result="success").model_dump(mode="json")

api/controllers/console/explore/installed_app.py

@@ -181,7 +181,7 @@ class InstalledAppsListApi(Resource):
 
         if current_user.current_tenant is None:
             raise ValueError("current_user.current_tenant must not be None")
-        current_user.role = TenantService.get_user_role(current_user, current_user.current_tenant, session=db.session)
+        current_user.role = TenantService.get_user_role(current_user, current_user.current_tenant, session=db.session())
         installed_app_list: list[dict[str, Any]] = []
         for installed_app, app_model in installed_apps:
             installed_app_list.append(

api/controllers/console/explore/message.py

@@ -27,6 +27,7 @@ from controllers.console.explore.wraps import InstalledAppResource
 from controllers.console.wraps import with_current_user
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from extensions.ext_database import db
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import (
     ExploreMessageInfiniteScrollPagination,
@@ -91,6 +92,7 @@ class MessageListApi(InstalledAppResource):
                 args.conversation_id,
                 args.first_id or None,
                 args.limit,
+                session=db.session(),
             )
             adapter = TypeAdapter(ExploreMessageListItem)
             items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
@@ -129,6 +131,7 @@ class MessageFeedbackApi(InstalledAppResource):
                 user=current_user,
                 rating=FeedbackRating(payload.rating) if payload.rating else None,
                 content=payload.content,
+                session=db.session(),
             )
         except MessageNotExistsError:
             raise NotFound("Message Not Exists.")
@@ -207,7 +210,11 @@ class MessageSuggestedQuestionApi(InstalledAppResource):
 
         try:
             questions = MessageService.get_suggested_questions_after_answer(
-                app_model=app_model, user=current_user, message_id=message_id_str, invoke_from=InvokeFrom.EXPLORE
+                app_model=app_model,
+                user=current_user,
+                message_id=message_id_str,
+                invoke_from=InvokeFrom.EXPLORE,
+                session=db.session(),
             )
         except MessageNotExistsError:
             raise NotFound("Message not found")

api/controllers/console/explore/parameter.py

@@ -8,6 +8,7 @@ from controllers.console import console_ns
 from controllers.console.app.error import AppUnavailableError
 from controllers.console.explore.wraps import InstalledAppResource
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from extensions.ext_database import db
 from models.model import AppMode, InstalledApp
 from services.app_service import AppService
 
@@ -64,4 +65,4 @@ class ExploreAppMetaApi(InstalledAppResource):
         app_model = installed_app.app
         if not app_model:
             raise ValueError("App not found")
-        return AppService().get_app_meta(app_model)
+        return AppService().get_app_meta(app_model, session=db.session())

api/controllers/console/explore/recommended_app.py

@@ -120,7 +120,7 @@ class RecommendedAppListApi(Resource):
         language_prefix = _resolve_language(args.language, current_user)
 
         return RecommendedAppListResponse.model_validate(
-            RecommendedAppService.get_recommended_apps_and_categories(db.session, language_prefix),
+            RecommendedAppService.get_recommended_apps_and_categories(language_prefix, session=db.session()),
             from_attributes=True,
         ).model_dump(mode="json")
 
@@ -137,7 +137,7 @@ class LearnDifyAppListApi(Resource):
         language_prefix = _resolve_language(args.language, current_user)
 
         return LearnDifyAppListResponse.model_validate(
-            RecommendedAppService.get_learn_dify_apps(db.session, language_prefix),
+            RecommendedAppService.get_learn_dify_apps(language_prefix, session=db.session()),
             from_attributes=True,
         ).model_dump(mode="json")
 
@@ -148,4 +148,4 @@ class RecommendedAppApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_id: UUID):
-        return RecommendedAppService.get_recommend_app_detail(db.session, str(app_id))
+        return RecommendedAppService.get_recommend_app_detail(str(app_id), session=db.session())

api/controllers/console/explore/saved_message.py

@@ -38,11 +38,7 @@ class SavedMessageListApi(InstalledAppResource):
         args = SavedMessageListQuery.model_validate(request.args.to_dict())
 
         pagination = SavedMessageService.pagination_by_last_id(
-            db.session(),
-            app_model,
-            current_user,
-            str(args.last_id) if args.last_id else None,
-            args.limit,
+            app_model, current_user, str(args.last_id) if args.last_id else None, args.limit, session=db.session()
         )
         adapter = TypeAdapter(SavedMessageItem)
         items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
@@ -65,7 +61,7 @@ class SavedMessageListApi(InstalledAppResource):
         payload = SavedMessageCreatePayload.model_validate(console_ns.payload or {})
 
         try:
-            SavedMessageService.save(db.session(), app_model, current_user, str(payload.message_id))
+            SavedMessageService.save(app_model, current_user, str(payload.message_id), session=db.session())
         except MessageNotExistsError:
             raise NotFound("Message Not Exists.")
 
@@ -88,6 +84,6 @@ class SavedMessageApi(InstalledAppResource):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
 
-        SavedMessageService.delete(db.session(), app_model, current_user, message_id_str)
+        SavedMessageService.delete(app_model, current_user, message_id_str, session=db.session())
 
         return "", 204

api/controllers/console/explore/trial.py

@@ -13,7 +13,6 @@ import services
 from controllers.common.fields import (
     AudioBinaryResponse,
     AudioTranscriptResponse,
-    GeneratedAppResponse,
     SimpleResultResponse,
 )
 from controllers.common.fields import Parameters as ParametersResponse
@@ -391,7 +390,6 @@ register_response_schema_models(
     ParametersResponse,
     AudioBinaryResponse,
     AudioTranscriptResponse,
-    GeneratedAppResponse,
     SimpleResultResponse,
     SiteResponse,
     SuggestedQuestionsResponse,
@@ -406,7 +404,7 @@ simple_account_model = console_ns.models[TrialSimpleAccount.__name__]
 class TrialAppWorkflowRunApi(TrialAppResource):
     @trial_feature_enable
     @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
-    @console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
+    @console_ns.response(200, "Success")
     @with_current_user
     @with_session
     def post(self, session: Session, current_user: Account, trial_app):
@@ -433,7 +431,8 @@ class TrialAppWorkflowRunApi(TrialAppResource):
                 invoke_from=InvokeFrom.EXPLORE,
                 streaming=True,
             )
-            RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
+            # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -478,7 +477,7 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):
 
 class TrialChatApi(TrialAppResource):
     @console_ns.expect(console_ns.models[ChatRequest.__name__])
-    @console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
+    @console_ns.response(200, "Success")
     @trial_feature_enable
     @with_current_user
     @with_session
@@ -512,7 +511,8 @@ class TrialChatApi(TrialAppResource):
                 invoke_from=InvokeFrom.EXPLORE,
                 streaming=True,
             )
-            RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
+            # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
@@ -551,7 +551,11 @@ class TrialMessageSuggestedQuestionApi(TrialAppResource):
 
         try:
             questions = MessageService.get_suggested_questions_after_answer(
-                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
+                app_model=app_model,
+                user=current_user,
+                message_id=message_id,
+                invoke_from=InvokeFrom.EXPLORE,
+                session=db.session(),
             )
         except MessageNotExistsError:
             raise NotFound("Message not found")
@@ -589,7 +593,7 @@ class TrialChatAudioApi(TrialAppResource):
             user_id = current_user.id
 
             response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
-            RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:
             logger.exception("App model config broken.")
@@ -645,12 +649,12 @@ class TrialChatTextApi(TrialAppResource):
 
             response = AudioService.transcript_tts(
                 app_model=app_model,
-                session=db.session,
+                session=db.session(),
                 text=text,
                 voice=voice,
                 message_ref=message_ref,
             )
-            RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:
             logger.exception("App model config broken.")
@@ -680,7 +684,7 @@ class TrialChatTextApi(TrialAppResource):
 
 class TrialCompletionApi(TrialAppResource):
     @console_ns.expect(console_ns.models[CompletionRequest.__name__])
-    @console_ns.response(200, "Success", console_ns.models[GeneratedAppResponse.__name__])
+    @console_ns.response(200, "Success")
     @trial_feature_enable
     @with_current_user
     @with_session
@@ -709,7 +713,8 @@ class TrialCompletionApi(TrialAppResource):
                 streaming=streaming,
             )
 
-            RecommendedAppService.add_trial_app_record(db.session, app_id, user_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
+            # response-contract:ignore compact_generate_response
             return helper.compact_generate_response(response)
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")

api/controllers/console/extension.py

@@ -112,7 +112,7 @@ class APIBasedExtensionAPI(Resource):
     def get(self, current_tenant_id: str):
         return dump_response(
             APIBasedExtensionListResponse,
-            APIBasedExtensionService.get_all_by_tenant_id(db.session(), current_tenant_id),
+            APIBasedExtensionService.get_all_by_tenant_id(current_tenant_id, session=db.session()),
         )
 
     @console_ns.doc("create_api_based_extension")
@@ -133,7 +133,7 @@ class APIBasedExtensionAPI(Resource):
             api_key=payload.api_key,
         )
 
-        extension = APIBasedExtensionService.save(db.session(), extension_data)
+        extension = APIBasedExtensionService.save(extension_data, session=db.session())
         return APIBasedExtensionResponse(
             id=extension.id,
             name=extension.name,
@@ -158,7 +158,9 @@ class APIBasedExtensionDetailAPI(Resource):
 
         return dump_response(
             APIBasedExtensionResponse,
-            APIBasedExtensionService.get_with_tenant_id(db.session(), current_tenant_id, api_based_extension_id),
+            APIBasedExtensionService.get_with_tenant_id(
+                current_tenant_id, api_based_extension_id, session=db.session()
+            ),
         )
 
     @console_ns.doc("update_api_based_extension")
@@ -174,7 +176,7 @@ class APIBasedExtensionDetailAPI(Resource):
         api_based_extension_id = str(id)
 
         extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
-            db.session(), current_tenant_id, api_based_extension_id
+            current_tenant_id, api_based_extension_id, session=db.session()
         )
 
         payload = APIBasedExtensionPayload.model_validate(console_ns.payload or {})
@@ -187,7 +189,7 @@ class APIBasedExtensionDetailAPI(Resource):
             extension_data_from_db.api_key = payload.api_key
             api_key_for_response = payload.api_key
 
-        APIBasedExtensionService.save(db.session(), extension_data_from_db)
+        APIBasedExtensionService.save(extension_data_from_db, session=db.session())
         return APIBasedExtensionResponse(
             id=extension_data_from_db.id,
             name=extension_data_from_db.name,
@@ -208,9 +210,9 @@ class APIBasedExtensionDetailAPI(Resource):
         api_based_extension_id = str(id)
 
         extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
-            db.session(), current_tenant_id, api_based_extension_id
+            current_tenant_id, api_based_extension_id, session=db.session()
         )
 
-        APIBasedExtensionService.delete(db.session(), extension_data_from_db)
+        APIBasedExtensionService.delete(extension_data_from_db, session=db.session())
 
         return "", 204

api/controllers/console/files.py

@@ -27,6 +27,7 @@ from controllers.console.wraps import (
 )
 from extensions.ext_database import db
 from fields.file_fields import FileResponse, UploadConfig
+from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
 from services.file_service import FileService
@@ -117,8 +118,7 @@ class FileApi(Resource):
         except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
             raise BlockedFileExtensionError(blocked_extension_error.description)
 
-        response = FileResponse.model_validate(upload_file, from_attributes=True)
-        return response.model_dump(mode="json"), 201
+        return dump_response(FileResponse, upload_file), 201
 
 
 @console_ns.route("/files/<uuid:file_id>/preview")
@@ -131,7 +131,7 @@ class FilePreviewApi(Resource):
     def get(self, current_tenant_id: str, file_id: UUID):
         file_id_str = str(file_id)
         text = FileService(db.engine).get_file_preview(file_id_str, current_tenant_id)
-        return {"content": text}
+        return TextContentResponse(content=text).model_dump(mode="json")
 
 
 @console_ns.route("/files/support-type")
@@ -141,4 +141,4 @@ class FileSupportTypeApi(Resource):
     @account_initialization_required
     @console_ns.response(200, "Success", console_ns.models[AllowedExtensionsResponse.__name__])
     def get(self):
-        return {"allowed_extensions": list(DOCUMENT_EXTENSIONS)}
+        return AllowedExtensionsResponse(allowed_extensions=list(DOCUMENT_EXTENSIONS)).model_dump(mode="json")

api/controllers/console/init_validate.py

@@ -50,7 +50,7 @@ def get_init_status() -> InitStatusResponse:
 @only_edition_self_hosted
 def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
     """Validate initialization password."""
-    tenant_count = TenantService.get_tenant_count(session=db.session)
+    tenant_count = TenantService.get_tenant_count(session=db.session())
     if tenant_count > 0:
         raise AlreadySetupError()
 

api/controllers/console/setup.py

@@ -13,7 +13,7 @@ from services.account_service import RegisterService, TenantService
 
 from .error import AlreadySetupError, NotInitValidateError
 from .init_validate import get_init_validate_status
-from .wraps import only_edition_self_hosted
+from .wraps import mark_setup_completed, only_edition_self_hosted
 
 
 class SetupRequestPayload(BaseModel):
@@ -79,7 +79,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
     if get_setup_status():
         raise AlreadySetupError()
 
-    tenant_count = TenantService.get_tenant_count(session=db.session)
+    tenant_count = TenantService.get_tenant_count(session=db.session())
     if tenant_count > 0:
         raise AlreadySetupError()
 
@@ -94,8 +94,9 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
         password=payload.password,
         ip_address=extract_remote_ip(request),
         language=payload.language,
-        session=db.session,
+        session=db.session(),
     )
+    mark_setup_completed()
 
     return SetupResponse(result="success")
 

api/controllers/console/snippets/payloads.py

@@ -148,6 +148,8 @@ class PublishWorkflowPayload(BaseModel):
     """Payload for publishing snippet workflow."""
 
     knowledge_base_setting: dict[str, Any] | None = Field(default=None)
+    marked_name: str | None = Field(default=None, max_length=20)
+    marked_comment: str | None = Field(default=None, max_length=100)
 
 
 class SnippetImportPayload(BaseModel):

api/controllers/console/socketio/workflow.py

@@ -44,7 +44,7 @@ def socket_connect(sid, environ, auth):
             return False
 
         with sio.app.app_context():
-            user = AccountService.load_logged_in_account(account_id=user_id, session=db.session)
+            user = AccountService.load_logged_in_account(account_id=user_id, session=db.session())
             if not user:
                 logging.warning("Socket connect rejected: user not found (user_id=%s, sid=%s)", user_id, sid)
                 return False
@@ -69,7 +69,8 @@ def handle_user_connect(sid, data):
     if not workflow_id:
         return {"msg": "workflow_id is required"}, 400
 
-    result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid)
+    with sio.app.app_context():
+        result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid, session=db.session())
     if not result:
         return {"msg": "unauthorized"}, 401
 

api/controllers/console/spec.py

@@ -1,8 +1,9 @@
 import logging
+from collections.abc import Mapping
 from typing import Any
 
 from flask_restx import Resource
-from pydantic import RootModel
+from pydantic import Field, RootModel
 
 from controllers.common.schema import register_response_schema_models
 from controllers.console.wraps import (
@@ -10,6 +11,7 @@ from controllers.console.wraps import (
     setup_required,
 )
 from core.schemas.schema_manager import SchemaManager
+from fields.base import ResponseModel
 from libs.login import login_required
 
 from . import console_ns
@@ -17,11 +19,17 @@ from . import console_ns
 logger = logging.getLogger(__name__)
 
 
-class SchemaDefinitionsResponse(RootModel[Any]):
-    root: Any
+class SchemaDefinitionItemResponse(ResponseModel):
+    name: str
+    label: str
+    schema_: Mapping[str, Any] = Field(alias="schema")
 
 
-register_response_schema_models(console_ns, SchemaDefinitionsResponse)
+class SchemaDefinitionsResponse(RootModel[list[SchemaDefinitionItemResponse]]):
+    pass
+
+
+register_response_schema_models(console_ns, SchemaDefinitionItemResponse, SchemaDefinitionsResponse)
 
 
 @console_ns.route("/spec/schema-definitions")

api/controllers/console/tag/tags.py

@@ -137,7 +137,7 @@ class TagListApi(Resource):
     def get(self, current_tenant_id: str):
         raw_args = request.args.to_dict()
         param = TagListQueryParam.model_validate(raw_args)
-        tags = TagService.get_tags(db.session(), param.type, current_tenant_id, param.keyword)
+        tags = TagService.get_tags(param.type, current_tenant_id, param.keyword, session=db.session())
 
         return dump_response(TagListResponse, tags), 200
 
@@ -154,7 +154,7 @@ class TagListApi(Resource):
 
         payload = TagBasePayload.model_validate(console_ns.payload or {})
         _enforce_snippet_tag_rbac_if_needed(payload.type)
-        tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=payload.type), db.session)
+        tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=payload.type), db.session())
 
         return dump_response(TagResponse, {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}), 200
 
@@ -175,9 +175,9 @@ class TagUpdateDeleteApi(Resource):
 
         payload = TagUpdateRequestPayload.model_validate(console_ns.payload or {})
         _enforce_snippet_tag_rbac_by_tag_id(tag_id_str)
-        tag = TagService.update_tags(UpdateTagPayload(name=payload.name), tag_id_str, db.session)
+        tag = TagService.update_tags(UpdateTagPayload(name=payload.name), tag_id_str, db.session())
 
-        binding_count = TagService.get_tag_binding_count(tag_id_str, db.session)
+        binding_count = TagService.get_tag_binding_count(tag_id_str, db.session())
 
         return (
             dump_response(
@@ -196,7 +196,7 @@ class TagUpdateDeleteApi(Resource):
         tag_id_str = str(tag_id)
 
         _enforce_snippet_tag_rbac_by_tag_id(tag_id_str)
-        TagService.delete_tag(tag_id_str, db.session)
+        TagService.delete_tag(tag_id_str, db.session())
 
         return "", 204
 
@@ -223,7 +223,7 @@ def _create_tag_bindings(current_user: Account) -> tuple[dict[str, str], int]:
             target_id=payload.target_id,
             type=payload.type,
         ),
-        db.session,
+        db.session(),
     )
     return {"result": "success"}, 200
 
@@ -239,7 +239,7 @@ def _remove_tag_bindings(current_user: Account) -> tuple[dict[str, str], int]:
             target_id=payload.target_id,
             type=payload.type,
         ),
-        db.session,
+        db.session(),
     )
     return {"result": "success"}, 200
 

api/controllers/console/workflow_run_archive.py

@@ -0,0 +1,221 @@
+import datetime
+from http import HTTPStatus
+
+from flask import redirect
+from flask_restx import Resource
+from pydantic import BaseModel, Field
+from werkzeug.exceptions import Conflict, NotFound
+
+from controllers.common.fields import RedirectResponse
+from controllers.common.schema import register_response_schema_models, register_schema_models
+from controllers.console import console_ns
+from controllers.console.wraps import (
+    RBACPermission,
+    RBACResourceScope,
+    account_initialization_required,
+    is_admin_or_owner_required,
+    rbac_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.base import ResponseModel
+from libs.archive_storage import get_export_storage
+from libs.helper import dump_response
+from libs.login import current_account_with_tenant, login_required
+from services.retention.workflow_run.archive_download_preparation import ARCHIVE_DOWNLOAD_MIME_TYPE
+from services.retention.workflow_run.archive_download_task_cache import (
+    WorkflowRunArchiveDownloadStatus,
+)
+from services.retention.workflow_run.archive_log_service import (
+    WorkflowRunArchiveDownloadNotReadyError,
+    WorkflowRunArchiveDownloadTaskNotFoundError,
+    WorkflowRunArchiveNotFoundError,
+    create_workflow_run_archive_download_task,
+    get_ready_workflow_run_archive_download_task,
+    get_workflow_run_archive_download_task,
+    list_workflow_run_archives,
+)
+
+
+class WorkflowRunArchiveDownloadPayload(BaseModel):
+    """Request body for preparing one monthly workflow-run archive download."""
+
+    year: int = Field(ge=1)
+    month: int = Field(ge=1, le=12)
+
+
+class WorkflowRunArchiveSummaryResponse(ResponseModel):
+    archived_month_count: int
+    workflow_run_count: int
+    archive_bytes: int
+    latest_archived_at: datetime.datetime | None = None
+
+
+class WorkflowRunArchiveDownloadTaskResponse(ResponseModel):
+    download_id: str
+    year: int
+    month: int
+    bundle_count: int
+    archive_bytes: int
+    status: WorkflowRunArchiveDownloadStatus
+    file_name: str | None = None
+    file_size_bytes: int | None = None
+    error: str | None = None
+    created_at: datetime.datetime
+    updated_at: datetime.datetime
+    expires_at: datetime.datetime
+    started_at: datetime.datetime | None = None
+    finished_at: datetime.datetime | None = None
+
+
+class WorkflowRunArchiveMonthResponse(ResponseModel):
+    year: int
+    month: int
+    bundle_count: int
+    workflow_run_count: int
+    row_count: int
+    archive_bytes: int
+    latest_archived_at: datetime.datetime
+    download_task: WorkflowRunArchiveDownloadTaskResponse | None = None
+
+
+class WorkflowRunArchiveListResponse(ResponseModel):
+    summary: WorkflowRunArchiveSummaryResponse
+    months: list[WorkflowRunArchiveMonthResponse]
+
+
+register_schema_models(console_ns, WorkflowRunArchiveDownloadPayload)
+register_response_schema_models(
+    console_ns,
+    WorkflowRunArchiveSummaryResponse,
+    WorkflowRunArchiveMonthResponse,
+    WorkflowRunArchiveListResponse,
+    WorkflowRunArchiveDownloadTaskResponse,
+    RedirectResponse,
+)
+
+
+def _current_ids() -> tuple[str, str]:
+    """Return current `(tenant_id, account_id)` or raise when no workspace is selected."""
+    current_user, current_tenant_id = current_account_with_tenant()
+    if not current_tenant_id:
+        raise NotFound("Current workspace not found")
+    return current_tenant_id, current_user.id
+
+
+def _presigned_url_expires_in(expires_at: datetime.datetime) -> int:
+    """Keep the storage URL no longer-lived than the Redis task and cap it for browser downloads."""
+    expires_at_utc = expires_at if expires_at.tzinfo else expires_at.replace(tzinfo=datetime.UTC)
+    remaining_seconds = int((expires_at_utc - datetime.datetime.now(datetime.UTC)).total_seconds())
+    return max(1, min(3600, remaining_seconds))
+
+
+@console_ns.route("/workflow-run-archives")
+class WorkflowRunArchivesApi(Resource):
+    @console_ns.doc("list_workflow_run_archives")
+    @console_ns.doc(description="List monthly workflow-run archive metadata for the current workspace")
+    @console_ns.response(200, "Success", console_ns.models[WorkflowRunArchiveListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(
+        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
+    )
+    def get(self):
+        tenant_id, _ = _current_ids()
+        return dump_response(WorkflowRunArchiveListResponse, list_workflow_run_archives(db.session(), tenant_id))
+
+
+@console_ns.route("/workflow-run-archives/downloads")
+class WorkflowRunArchiveDownloadsApi(Resource):
+    @console_ns.doc("create_workflow_run_archive_download")
+    @console_ns.doc(description="Create or return a temporary workflow-run archive download task")
+    @console_ns.expect(console_ns.models[WorkflowRunArchiveDownloadPayload.__name__])
+    @console_ns.response(
+        HTTPStatus.ACCEPTED,
+        "Download task accepted",
+        console_ns.models[WorkflowRunArchiveDownloadTaskResponse.__name__],
+    )
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(
+        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
+    )
+    def post(self):
+        tenant_id, account_id = _current_ids()
+        payload = WorkflowRunArchiveDownloadPayload.model_validate(console_ns.payload or {})
+        try:
+            task = create_workflow_run_archive_download_task(
+                db.session(),
+                tenant_id=tenant_id,
+                requested_by=account_id,
+                year=payload.year,
+                month=payload.month,
+            )
+        except WorkflowRunArchiveNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
+        return dump_response(WorkflowRunArchiveDownloadTaskResponse, task), HTTPStatus.ACCEPTED
+
+
+@console_ns.route("/workflow-run-archives/downloads/<string:download_id>")
+class WorkflowRunArchiveDownloadApi(Resource):
+    @console_ns.doc("get_workflow_run_archive_download")
+    @console_ns.doc(description="Get a temporary workflow-run archive download task")
+    @console_ns.response(200, "Success", console_ns.models[WorkflowRunArchiveDownloadTaskResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(
+        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
+    )
+    def get(self, download_id: str):
+        tenant_id, _ = _current_ids()
+        try:
+            task = get_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
+        except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
+        return dump_response(WorkflowRunArchiveDownloadTaskResponse, task)
+
+
+@console_ns.route("/workflow-run-archives/downloads/<string:download_id>/file")
+class WorkflowRunArchiveDownloadFileApi(Resource):
+    @console_ns.doc("download_workflow_run_archive_file")
+    @console_ns.doc(description="Redirect to a prepared workflow-run archive ZIP file")
+    @console_ns.response(
+        302,
+        "Redirect to pre-signed archive storage URL",
+        console_ns.models[RedirectResponse.__name__],
+    )
+    @console_ns.response(409, "Download task is not ready")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_admin_or_owner_required
+    @rbac_permission_required(
+        RBACResourceScope.WORKSPACE, RBACPermission.WORKSPACE_ROLE_MANAGE, resource_required=False
+    )
+    def get(self, download_id: str):
+        tenant_id, _ = _current_ids()
+        try:
+            task = get_ready_workflow_run_archive_download_task(tenant_id=tenant_id, download_id=download_id)
+        except WorkflowRunArchiveDownloadTaskNotFoundError as exc:
+            raise NotFound(str(exc)) from exc
+        except WorkflowRunArchiveDownloadNotReadyError as exc:
+            raise Conflict(str(exc)) from exc
+
+        storage_key = task.storage_key
+        if storage_key is None:
+            raise Conflict(f"Workflow run archive download is not ready: {download_id}")
+
+        storage = get_export_storage()
+        presigned_url = storage.generate_presigned_url(
+            storage_key,
+            expires_in=_presigned_url_expires_in(task.expires_at),
+            filename=task.file_name,
+            content_type=ARCHIVE_DOWNLOAD_MIME_TYPE,
+        )
+        return redirect(presigned_url, code=HTTPStatus.FOUND)