feat(cli): add amp, openclaw, qoder, windsurf, hermes to skills install registry

Extend the agent registry with five verified entries. Amp and OpenClaw read the shared ~/.agents/skills directory that Codex already uses, so the installer now groups agents by resolved target and writes a shared path once (dry-run merges the agent names into a single line). Paths verified against each tool's documentation: - amp: probe ~/.config/amp -> shared ~/.agents/skills - openclaw: probe ~/.openclaw -> shared ~/.agents/skills - qoder: probe ~/.qoder -> ~/.qoder/skills - windsurf: probe ~/.codeium/windsurf -> ~/.codeium/windsurf/skills - hermes: probe ~/.hermes -> ~/.hermes/skills
chore(deps): bump the storage group in /api with 5 updates (#38825 )
2026-07-13 11:57:51 +08:00 · 2026-07-13 11:32:27 +08:00 · 2026-07-13 11:32:06 +08:00 · 2026-07-13 11:31:53 +08:00 · 2026-07-13 11:31:37 +08:00 · 2026-07-13 03:19:10 +00:00
7001 changed files with 278544 additions and 206997 deletions
@@ -28,9 +28,12 @@ Flag:
 - Missing `package.json#exports` entry for a new primitive.
 - Internal package imports using workspace subpaths instead of relative paths.
 - Exported props using internal-only types that consumers cannot import from the component subpath.
+- Canonical primitive boundaries or their associated public types using a redundant `Root` suffix when no higher-level convenience component exists in the same subpath.

 Consumers use subpath exports such as `@langgenius/dify-ui/button`.

+Canonical boundaries use the primitive name and matching public types (`Select` / `SelectProps`). Keep `Root` only to distinguish a low-level anatomy root from a higher-level convenience component (`CheckboxRoot` / `Checkbox`); implementation aliases should still show their Base UI source (`BaseSelect.Root.Props`).
+
 ## Props And State

 Flag:
@@ -45,17 +48,17 @@ Flag:

 Prefer Base UI/Dify UI data attributes and CSS variables for visual state: `data-open`, `data-checked`, `data-disabled`, `data-highlighted`, `data-popup-open`, `group-data-*`, `peer-data-*`, `has-[:focus-visible]`, and primitive CSS variables such as anchor width or transform origin. Use JS conditional classes for product/business state that the primitive does not expose.

-For non-string `Select` and `RadioGroup` values, prefer explicit domain generics at the root and at child value carriers. JSX children do not inherit the parent generic, so `RadioGroup<PromptMode>` should compose with `Radio<PromptMode>`, `RadioRoot<PromptMode>`, or option values from a typed collection. For `Select`, prefer the Base UI `items` collection pattern for typed value-to-label rendering, and flag string coercion helpers used only to recover display labels.
+For non-string `Select` and `RadioGroup` values, prefer explicit domain generics at the root and at child value carriers. JSX children do not inherit the parent generic, so `RadioGroup<PromptMode>` should compose with `Radio<PromptMode>`, `RadioItem<PromptMode>`, or option values from a typed collection. For `Select`, prefer the Base UI `items` collection pattern for typed value-to-label rendering, and flag string coercion helpers used only to recover display labels.

 ## Forms

 Flag:

 - Form-like UI using unrelated `Input` and `Button` pieces without a submit boundary.
- Text-like fields not composed through `FieldRoot`, `FieldLabel`, and `FieldControl` when using Dify UI form semantics.
+- Text-like fields not composed through `Field`, `FieldLabel`, and `FieldControl` when using Dify UI form semantics.
 - Select fields using `FieldLabel` instead of `SelectLabel`.
 - Slider fields using a generic label instead of `SliderLabel`.
- Checkbox/radio groups missing `FieldsetRoot` and `FieldsetLegend`.
+- Checkbox/radio groups missing `Fieldset` and `FieldsetLegend`.
 - Field errors or descriptions rendered without `FieldDescription` / `FieldError` relationships.

 `Form` is the submit boundary. Dify UI form primitives are not a form state-management framework; business validation and schema-driven behavior belong in `web/`.
@@ -35,7 +35,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -97,7 +97,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -148,7 +148,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: "3.12"
@@ -39,11 +39,19 @@ jobs:
        with:
          files: |
            web/**
+            cli/**
+            e2e/**
            packages/**
+            sdks/nodejs-client/**
            package.json
            pnpm-lock.yaml
            pnpm-workspace.yaml
            .nvmrc
+            vite.config.ts
+            eslint.config.mjs
+            .vscode/**
+            .github/workflows/autofix.yml
+            .github/workflows/style.yml
      - name: Check api inputs
        if: github.event_name != 'merge_group'
        id: api-changes
@@ -78,7 +86,7 @@ jobs:
          python-version: "3.11"

      - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2

      - name: Generate Docker Compose
        if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -161,7 +169,11 @@ jobs:
      - name: ESLint autofix
        if: github.event_name != 'merge_group' && steps.web-changes.outputs.any_changed == 'true'
        run: |
-          vp exec eslint --concurrency=2 --prune-suppressions --quiet || true
+          vp exec eslint --fix --concurrency=2 --prune-suppressions --quiet || true
+
+      - name: Format frontend files
+        if: github.event_name != 'merge_group' && (steps.web-changes.outputs.any_changed == 'true' || steps.frontend-contract-changes.outputs.any_changed == 'true')
+        run: vp fmt

      - if: github.event_name != 'merge_group'
        uses: autofix-ci/action@c5b2d67aa2274e7b5a18224e8171550871fc7e4a # v1.3.4
@@ -19,7 +19,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: "3.12"
@@ -69,7 +69,7 @@ jobs:
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: "3.12"
@@ -9,6 +9,6 @@ jobs:
      pull-requests: write
    runs-on: depot-ubuntu-24.04
    steps:
-      - uses: actions/labeler@f27b608878404679385c85cfa523b85ccb86e213 # v6.1.0
+      - uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
        with:
          sync-labels: true
@@ -53,6 +53,14 @@ jobs:
          filters: |
            api:
              - 'api/**'
+              - 'scripts/ast_grep_guard.py'
+              - 'scripts/check_no_new_getattr.py'
+              - 'scripts/check_no_new_controller_sqlalchemy.py'
+              - 'scripts/lint_controller_sqlalchemy.py'
+              - 'scripts/ast_grep_rules/no_new_getattr.yml'
+              - 'scripts/ast_grep_rules/no_new_controller_sqlalchemy.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 +388,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
@@ -22,7 +22,7 @@ jobs:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true

@@ -24,7 +24,7 @@ jobs:
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true

@@ -22,7 +22,7 @@ jobs:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true

@@ -18,7 +18,7 @@ jobs:
      pull-requests: write

    steps:
-      - uses: actions/stale@eb5cf3af3ac0a1aa4c9c45633dd1ae542a27a899 # v10.3.0
+      - uses: actions/stale@1e223db275d687790206a7acac4d1a11bd6fe629 # v10.4.0
        with:
          days-before-issue-stale: 15
          days-before-issue-close: 3
@@ -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 }}
@@ -30,13 +34,18 @@ jobs:
        with:
          files: |
            api/**
+            scripts/ast_grep_guard.py
            scripts/check_no_new_getattr.py
+            scripts/check_no_new_controller_sqlalchemy.py
+            scripts/lint_controller_sqlalchemy.py
            scripts/ast_grep_rules/no_new_getattr.yml
+            scripts/ast_grep_rules/no_new_controller_sqlalchemy.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@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: false
          python-version: "3.12"
@@ -54,17 +63,13 @@ 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 No New Controller SQLAlchemy Guard
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: uv run --project api python scripts/check_no_new_controller_sqlalchemy.py --base-rev "${{ inputs.base-rev }}"

      - name: Run Type Checks
        if: steps.changed-files.outputs.any_changed == 'true'
@@ -157,7 +162,10 @@ jobs:
            pnpm-lock.yaml
            pnpm-workspace.yaml
            .nvmrc
+            vite.config.ts
            eslint.config.mjs
+            .vscode/**
+            .github/workflows/autofix.yml
            .github/workflows/style.yml
            .github/actions/setup-web/**

@@ -165,15 +173,19 @@ jobs:
        if: steps.changed-files.outputs.any_changed == 'true'
        uses: ./.github/actions/setup-web

+      - name: Format check
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: vp fmt --check
+
      - name: Restore ESLint cache
        if: steps.changed-files.outputs.any_changed == 'true'
        id: eslint-cache-restore
        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 }}
+          key: ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'packages/dify-ui/eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-${{ github.sha }}
          restore-keys: |
-            ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-
+            ${{ runner.os }}-eslint-${{ hashFiles('pnpm-lock.yaml', 'eslint.config.mjs', 'packages/dify-ui/eslint.config.mjs', 'web/eslint.config.mjs', 'web/eslint.constants.mjs', 'web/plugins/eslint/**') }}-

      - name: Style check
        if: steps.changed-files.outputs.any_changed == 'true'
@@ -158,7 +158,7 @@ jobs:

      - name: Run Claude Code for Translation Sync
        if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@558b1d6cab4085c7753fe402c10bef0fbb92ac7a # v1.0.165
+        uses: anthropics/claude-code-action@e90deca47693f9457b72f2b53c17d7c445a87342 # v1.0.171
        with:
          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
          github_token: ${{ secrets.GITHUB_TOKEN }}
@@ -223,7 +223,7 @@ jobs:
               - Match the existing terminology and register used by each locale.
               - Prefer one Edit per file when stable, but prioritize correctness over batching.
            6. Verify only the edited files.
-               - Run `vp run dify-web#lint:fix --quiet -- <relative edited i18n file paths under web/>`
+               - Run `vp fmt <repo-relative edited paths such as web/i18n/...>`
               - Run `vp run dify-web#i18n:check ${{ steps.context.outputs.FILE_ARGS }} ${{ steps.context.outputs.LANG_ARGS }}`
               - If verification fails, fix the remaining problems before continuing.
            7. Stop after the scoped locale files are updated and verification passes.
@@ -322,7 +322,7 @@ jobs:
            '## Verification',
            '',
            `- \`vp run dify-web#i18n:check --file ${process.env.FILES_IN_SCOPE} --lang ${process.env.TARGET_LANGS}\``,
-            `- \`vp run dify-web#lint:fix --quiet -- <edited i18n files under web/>\``,
+            `- \`vp fmt <repo-relative edited paths such as web/i18n/...>\``,
            '',
            '## Notes',
            '',
@@ -36,7 +36,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -33,7 +33,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -33,7 +33,7 @@ jobs:
        uses: ./.github/actions/setup-web

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
        with:
          enable-cache: true
          python-version: "3.12"
@@ -44,6 +44,10 @@ jobs:
      - name: Install API dependencies
        run: uv sync --project api --dev

+      - name: Run E2E support unit tests
+        working-directory: ./e2e
+        run: vp run test:unit
+
      - name: Install Playwright browser
        working-directory: ./e2e
        run: vp run e2e:install
@@ -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.
@@ -1,54 +1,47 @@
 {
-  "cucumber.features": [
-    "e2e/features/**/*.feature",
-  ],
-  "cucumber.glue": [
-    "e2e/features/**/*.ts",
-  ],
+  "cucumber.features": ["e2e/features/**/*.feature"],
+  "cucumber.glue": ["e2e/features/**/*.ts"],

  "tailwindCSS.experimental.configFile": "web/app/styles/globals.css",

-  // Auto fix
+  // Format
+  "[javascript]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[javascriptreact]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[typescript]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[typescriptreact]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[json]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[jsonc]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[markdown]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[mdx]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[yaml]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[toml]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[css]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[scss]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[less]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[html]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[vue]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[svelte]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[graphql]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "[handlebars]": { "editor.defaultFormatter": "oxc.oxc-vscode" },
+  "editor.formatOnSave": true,
+  "editor.formatOnSaveMode": "file",
+  "oxc.fmt.configPath": "./vite.config.ts",
+
+  // Lint fix
  "editor.codeActionsOnSave": {
-    "source.fixAll.eslint": "explicit",
+    "source.fixAll.eslint": "explicit"
  },

-  // Silent the stylistic rules in your IDE, but still auto fix them
-  "eslint.rules.customizations": [
-    { "rule": "style/*", "severity": "off", "fixable": true },
-    { "rule": "format/*", "severity": "off", "fixable": true },
-    { "rule": "*-indent", "severity": "off", "fixable": true },
-    { "rule": "*-spacing", "severity": "off", "fixable": true },
-    { "rule": "*-spaces", "severity": "off", "fixable": true },
-    { "rule": "*-order", "severity": "off", "fixable": true },
-    { "rule": "*-dangle", "severity": "off", "fixable": true },
-    { "rule": "*-newline", "severity": "off", "fixable": true },
-    { "rule": "*quotes", "severity": "off", "fixable": true },
-    { "rule": "*semi", "severity": "off", "fixable": true }
-  ],
-
-  // Enable eslint for all supported languages
+  // Enable ESLint for linted languages
  "eslint.validate": [
    "javascript",
    "javascriptreact",
    "typescript",
    "typescriptreact",
-    "vue",
-    "html",
    "markdown",
    "json",
    "jsonc",
    "yaml",
-    "toml",
-    "xml",
-    "gql",
-    "graphql",
-    "astro",
-    "svelte",
-    "css",
-    "less",
-    "scss",
-    "pcss",
-    "postcss"
+    "toml"
  ]
 }
@@ -31,7 +31,7 @@ The codebase is split into:
 ## Language Style

 - **Python**: Keep type hints on functions and attributes, and implement relevant special methods (e.g., `__repr__`, `__str__`). Prefer `TypedDict` over `dict` or `Mapping` for type safety and better code documentation.
- **TypeScript**: Use the strict config, rely on ESLint (`pnpm lint:fix` preferred) plus `pnpm type-check`, and avoid `any` types.
+- **TypeScript**: Use the strict config, format with `vp fmt`, run ESLint for code-quality checks and fixes, run `pnpm type-check`, and avoid `any` types.

 ## General Practices

@@ -125,8 +125,6 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
 - **Dify for enterprise / organizations<br/>**
  We provide additional enterprise-centric features. [Send us an email](mailto:[email protected]?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss your enterprise needs. <br/>

-  > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding.
-
 ## Staying ahead

 Star Dify on GitHub and be instantly notified of new releases.
@@ -663,6 +663,9 @@ PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1

+# Dify Agent backend
+AGENT_BACKEND_BASE_URL=http://localhost:5050
+
 # Marketplace configuration
 MARKETPLACE_ENABLED=true
 MARKETPLACE_API_URL=https://marketplace.dify.ai
@@ -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",
@@ -54,8 +54,21 @@ class AgentBackendRunFailedError(AgentBackendError):

    run_id: str
    detail: Any
+    reason: str | None
+    source_event_id: str | None

-    def __init__(self, run_id: str, detail: Any) -> None:
+    def __init__(
+        self,
+        run_id: str,
+        detail: Any,
+        *,
+        message: str | None = None,
+        reason: str | None = None,
+        source_event_id: str | None = None,
+    ) -> None:
        self.run_id = run_id
        self.detail = detail
-        super().__init__(f"Agent backend run failed: {run_id}")
+        self.reason = reason
+        self.source_event_id = source_event_id
+        display_message = message or f"Agent backend run failed: {run_id}"
+        super().__init__(f"{display_message} (agent_run_id={run_id})")
@@ -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 [
@@ -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,
@@ -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",
+]
@@ -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"))
@@ -212,9 +212,7 @@ def migrate_member_roles_to_rbac(
                account_id=owner_account_id,
                member_account_ids=[account_id for account_id, _ in batch],
            )
-            current_roles_by_account_id = {
-                item.account_id: {str(role.id) for role in item.roles} for item in current_roles
-            }
+            current_roles_by_account_id = {item.account_id: {role.id for role in item.roles} for item in current_roles}

        replace_jobs: list[tuple[str, str]] = []
        for member_account_id, legacy_role in batch:
@@ -468,7 +466,9 @@ def migrate_dataset_permissions_to_rbac(
                                        "account_id": operator_account_id,
                                        "dataset_id": current_dataset_id,
                                        "target_account_id": member_account_id,
-                                        "payload": replace_user_access_policies_payload.model_dump(mode="json"),
+                                        "payload": replace_user_access_policies_payload.model_dump(
+                                            mode="json", exclude_unset=True
+                                        ),
                                    },
                                },
                            }
@@ -1,10 +1,12 @@
 import datetime
 import logging
 import time
+from collections.abc import Callable
 from typing import TypedDict

 import click
 import sqlalchemy as sa
+from sqlalchemy.orm import Session, sessionmaker

 from extensions.ext_database import db
 from libs.datetime_utils import naive_utc_now
@@ -12,6 +14,7 @@ from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpi
 from services.retention.conversation.messages_clean_policy import create_message_clean_policy
 from services.retention.conversation.messages_clean_service import MessagesCleanService
 from services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
+from services.retention.workflow_run.db_retry import run_with_db_retry
 from services.retention.workflow_run.tenant_prefix import tenant_prefix_condition
 from tasks.remove_app_and_related_data_task import delete_draft_variables_batch

@@ -35,6 +38,12 @@ class WorkflowRunArchiveTenantPlan(TypedDict):
    unpaid_tenant_ids: list[str]


+class WorkflowRunArchivePrefixStats(TypedDict):
+    tenant_ids: list[str]
+    workflow_runs: int
+    workflow_node_executions: int
+
+
 def _normalize_utc_datetime(value: datetime.datetime) -> datetime.datetime:
    if value.tzinfo is None:
        return value.replace(tzinfo=datetime.UTC)
@@ -57,6 +66,7 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:


 def _get_archive_candidate_tenant_ids_by_prefix(
+    session: Session,
    prefix: str,
    *,
    start_from: datetime.datetime | None,
@@ -75,7 +85,7 @@ def _get_archive_candidate_tenant_ids_by_prefix(
    if start_from is not None:
        conditions.append(WorkflowRun.created_at >= start_from)

-    tenant_ids = db.session.scalars(
+    tenant_ids = session.scalars(
        sa.select(WorkflowRun.tenant_id).where(*conditions).distinct().order_by(WorkflowRun.tenant_id)
    ).all()
    return list(tenant_ids)
@@ -102,8 +112,80 @@ def _filter_paid_workflow_archive_tenant_ids(tenant_ids: list[str]) -> tuple[lis
    return paid_tenant_ids, unpaid_tenant_ids


+def _run_archive_command_db_retry[T](operation_name: str, operation: Callable[[], T]) -> T:
+    return run_with_db_retry(operation_name, operation, logger=logger)
+
+
+def _get_archive_candidate_tenant_ids_with_retry(
+    session_maker: sessionmaker[Session],
+    prefix: str,
+    *,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+) -> list[str]:
+    def fetch_tenant_ids() -> list[str]:
+        with session_maker() as session:
+            return _get_archive_candidate_tenant_ids_by_prefix(
+                session,
+                prefix,
+                start_from=start_from,
+                end_before=end_before,
+            )
+
+    return _run_archive_command_db_retry(f"workflow archive tenant resolve for prefix {prefix}", fetch_tenant_ids)
+
+
+def _get_archive_plan_prefix_stats(
+    session_maker: sessionmaker[Session],
+    prefix: str,
+    *,
+    start_from: datetime.datetime | None,
+    end_before: datetime.datetime,
+) -> WorkflowRunArchivePrefixStats:
+    from graphon.enums import WorkflowExecutionStatus
+    from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
+    from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
+
+    def fetch_prefix_stats() -> WorkflowRunArchivePrefixStats:
+        with session_maker() as session:
+            tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
+                session,
+                prefix,
+                start_from=start_from,
+                end_before=end_before,
+            )
+            run_conditions = [
+                WorkflowRun.created_at < end_before,
+                WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
+                WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
+                tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
+            ]
+            if start_from is not None:
+                run_conditions.append(WorkflowRun.created_at >= start_from)
+            workflow_runs = (
+                session.scalar(sa.select(sa.func.count()).select_from(WorkflowRun).where(*run_conditions)) or 0
+            )
+            candidate_runs = sa.select(WorkflowRun.id).where(*run_conditions).subquery()
+            workflow_node_executions = (
+                session.scalar(
+                    sa.select(sa.func.count())
+                    .select_from(WorkflowNodeExecutionModel)
+                    .join(candidate_runs, WorkflowNodeExecutionModel.workflow_run_id == candidate_runs.c.id)
+                )
+                or 0
+            )
+            return WorkflowRunArchivePrefixStats(
+                tenant_ids=tenant_ids,
+                workflow_runs=workflow_runs,
+                workflow_node_executions=workflow_node_executions,
+            )
+
+    return _run_archive_command_db_retry(f"workflow archive plan for prefix {prefix}", fetch_prefix_stats)
+
+
 def _resolve_archive_tenant_ids_from_plan(
    *,
+    session_maker: sessionmaker[Session],
    tenant_ids: str | None,
    tenant_prefixes: list[str],
    start_from: datetime.datetime | None,
@@ -122,7 +204,8 @@ def _resolve_archive_tenant_ids_from_plan(
        requested_tenant_ids = []
        for prefix in tenant_prefixes:
            requested_tenant_ids.extend(
-                _get_archive_candidate_tenant_ids_by_prefix(
+                _get_archive_candidate_tenant_ids_with_retry(
+                    session_maker,
                    prefix,
                    start_from=start_from,
                    end_before=end_before,
@@ -143,6 +226,21 @@ def _resolve_archive_tenant_ids_from_plan(
    )


+def _safe_remove_scoped_session(context: str) -> None:
+    try:
+        db.session.remove()
+    except Exception:
+        logger.warning("Ignoring DB scoped-session cleanup error after %s", context, exc_info=True)
+        try:
+            db.session.registry.clear()
+        except Exception:
+            logger.warning("Ignoring DB scoped-session registry cleanup error after %s", context, exc_info=True)
+        try:
+            db.engine.dispose()
+        except Exception:
+            logger.warning("Ignoring DB engine dispose error after %s", context, exc_info=True)
+
+
 def _resolve_archive_time_range(
    *,
    before_days: int,
@@ -349,10 +447,6 @@ def archive_workflow_runs_plan(
    supported workflow types, and the requested created_at window. V2 bundle archive
    does not maintain per-run archive logs, so this plan reports source-table volume.
    """
-    from graphon.enums import WorkflowExecutionStatus
-    from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
-    from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
-
    before_days, start_from, end_before = _resolve_archive_time_range(
        before_days=before_days,
        from_days_ago=from_days_ago,
@@ -364,37 +458,25 @@ def archive_workflow_runs_plan(
    if include_archived:
        click.echo(click.style("--include-archived is a no-op for V2 bundle archive plans.", fg="yellow"))

+    session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
    rows: list[WorkflowRunArchivePlanRow] = []
    for prefix in _HEX_PREFIXES:
-        tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
-            prefix,
-            start_from=start_from,
-            end_before=plan_end_before,
-        )
+        try:
+            prefix_stats = _get_archive_plan_prefix_stats(
+                session_maker,
+                prefix,
+                start_from=start_from,
+                end_before=plan_end_before,
+            )
+        except Exception as exc:
+            logger.exception("Failed to build workflow archive plan for prefix %s", prefix)
+            raise click.ClickException(f"Failed to build workflow archive plan for prefix {prefix}.") from exc
+        tenant_ids = prefix_stats["tenant_ids"]
+        workflow_runs = prefix_stats["workflow_runs"]
+        workflow_node_executions = prefix_stats["workflow_node_executions"]
        total_tenants = len(tenant_ids)
        paid_tenant_ids, unpaid_tenant_ids = _filter_paid_workflow_archive_tenant_ids(tenant_ids)

-        run_conditions = [
-            WorkflowRun.created_at < plan_end_before,
-            WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
-            WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
-            tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
-        ]
-        if start_from is not None:
-            run_conditions.append(WorkflowRun.created_at >= start_from)
-        workflow_runs = (
-            db.session.scalar(sa.select(sa.func.count()).select_from(WorkflowRun).where(*run_conditions)) or 0
-        )
-        candidate_runs = sa.select(WorkflowRun.id).where(*run_conditions).subquery()
-        workflow_node_executions = (
-            db.session.scalar(
-                sa.select(sa.func.count())
-                .select_from(WorkflowNodeExecutionModel)
-                .join(candidate_runs, WorkflowNodeExecutionModel.workflow_run_id == candidate_runs.c.id)
-            )
-            or 0
-        )
-
        rows.append(
            WorkflowRunArchivePlanRow(
                tenant_prefix=prefix,
@@ -574,17 +656,18 @@ def archive_workflow_runs(
            )
        )

+    session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
    try:
        tenant_plan = _resolve_archive_tenant_ids_from_plan(
+            session_maker=session_maker,
            tenant_ids=tenant_ids,
            tenant_prefixes=parsed_tenant_prefixes,
            start_from=start_from,
            end_before=plan_end_before,
        )
-    except Exception:
+    except Exception as exc:
        logger.exception("Failed to resolve workflow archive tenant plan")
-        click.echo(click.style("Failed to resolve workflow archive tenant plan.", fg="red"))
-        return
+        raise click.ClickException("Failed to resolve workflow archive tenant plan.") from exc

    planned_tenant_ids = tenant_plan["archive_tenant_ids"]
    planned_paid_tenant_ids = tenant_plan["paid_tenant_ids"] if planned_tenant_ids is not None else None
@@ -616,7 +699,10 @@ def archive_workflow_runs(
        dry_run=dry_run,
        delete_after_archive=delete_after_archive,
    )
-    summary = archiver.run()
+    try:
+        summary = archiver.run()
+    finally:
+        _safe_remove_scoped_session("archive workflow run command")
    click.echo(
        click.style(
            f"Summary: processed={summary.total_runs_processed}, archived={summary.runs_archived}, "
@@ -25,11 +25,10 @@ class AgentBackendConfig(BaseSettings):
    AGENT_SHELL_ENABLED: bool = Field(
        description=(
            "Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "
-            "Requires the agent backend to be wired with a shellctl entrypoint; keep it "
-            "off until shellctl is deployed, otherwise every agent run that includes the "
-            "shell layer will fail."
+            "Requires the agent backend to be wired with a shellctl entrypoint before "
+            "shell-using Agent runs are executed."
        ),
-        default=False,
+        default=True,
    )

    AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS: NonNegativeFloat = Field(
@@ -363,7 +363,10 @@ class FileAccessConfig(BaseSettings):
    INTERNAL_FILES_URL: str = Field(
        description="Internal base URL for file access within Docker network,"
        " used for plugin daemon and internal service communication."
-        " Falls back to FILES_URL if not specified.",
+        " Explicit INTERNAL_FILES_URL takes precedence; otherwise SERVER_CONSOLE_API_URL is used,"
+        " then FILES_URL.",
+        validation_alias=AliasChoices("INTERNAL_FILES_URL", "SERVER_CONSOLE_API_URL"),
+        alias_priority=1,
        default="",
    )

@@ -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(),
+    )
@@ -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
@@ -42,6 +42,7 @@ from controllers.console.wraps import (
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.entities import PreProcessingRule, Rule, Segmentation
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
+from core.rbac import RBACResourceWhitelistScope
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
@@ -68,6 +69,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
    WeightVectorSetting,
 )
 from services.feature_service import FeatureService
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task

 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]

@@ -645,6 +647,14 @@ class AppListApi(Resource):

        app_service = AppService()
        app = app_service.create_app(current_tenant_id, params, current_user, session=db.session())
+        if dify_config.RBAC_ENABLED:
+            enterprise_rbac_service.RBACService.AppAccess.replace_whitelist(
+                tenant_id=str(current_tenant_id),
+                account_id=current_user.id,
+                app_id=str(app.id),
+                payload=enterprise_rbac_service.ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+            )
+            initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, app.id)
        permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
            str(current_tenant_id),
            current_user.id,
@@ -707,7 +717,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")
@@ -723,7 +733,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")


@@ -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,
@@ -416,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:
@@ -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
@@ -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()
@@ -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)
@@ -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

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


@@ -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):
@@ -647,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")
@@ -688,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.")
@@ -717,7 +707,7 @@ class DatasetApi(Resource):
        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
@@ -760,7 +750,7 @@ class DatasetUseCheckApi(Resource):
        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
+        return UsageCheckResponse(is_using=dataset_is_using).model_dump(mode="json"), 200


@console_ns.route("/datasets/<uuid:dataset_id>/queries")
@@ -901,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")
@@ -1018,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")
@@ -1052,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>")
@@ -1107,7 +1107,7 @@ class DatasetEnableApiApi(Resource):

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


@@ -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
@@ -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")
@@ -491,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
@@ -557,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")
@@ -578,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:
@@ -610,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 "
@@ -623,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
@@ -643,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 = []
@@ -717,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."
@@ -854,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
@@ -871,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,
@@ -943,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
@@ -990,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")
@@ -999,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
@@ -1034,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


@@ -1093,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")
@@ -1152,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")
@@ -1194,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")
@@ -1321,7 +1427,7 @@ 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())
@@ -1363,15 +1469,15 @@ class WebsiteDocumentSyncApi(DocumentResource):
        # sync document
        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
@@ -1394,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")
@@ -1508,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")
@@ -1516,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
@@ -1534,6 +1642,7 @@ 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)
@@ -1559,4 +1668,4 @@ class DocumentSummaryStatusApi(DocumentResource):
            session=db.session(),
        )

-        return result, 200
+        return dump_response(DocumentSummaryStatusResponse, result), 200
@@ -391,7 +391,7 @@ class DatasetDocumentSegmentApi(Resource):
            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")
@@ -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
@@ -322,7 +307,7 @@ class ExternalApiUseCheckApi(Resource):
        external_knowledge_api_is_using, count = ExternalDatasetService.external_knowledge_api_use_check(
            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,17 +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")
@@ -382,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")
@@ -394,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))

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

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

@@ -439,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
@@ -67,7 +67,7 @@ class DatasetMetadataCreateApi(Resource):
        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

@@ -84,7 +84,7 @@ class DatasetMetadataCreateApi(Resource):
        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


@@ -111,7 +111,7 @@ class DatasetMetadataApi(Resource):
        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

@@ -130,7 +130,7 @@ class DatasetMetadataApi(Resource):
            raise NotFound("Dataset not found.")
        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

@@ -169,9 +169,9 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):

        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

@@ -198,7 +198,7 @@ class DocumentMetadataEditApi(Resource):

        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
@@ -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
@@ -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()
@@ -36,6 +36,7 @@ from models import Account
 from models.model import AppMode, InstalledApp
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
+from services.conversation_service import ConversationService
 from services.errors.llm import InvokeRateLimitError

 from .. import console_ns
@@ -188,6 +189,15 @@ class ChatApi(InstalledAppResource):
        db.session.commit()

        try:
+            # Eagerly validate conversation to avoid hanging on invalid conversation_id
+            if payload.conversation_id:
+                ConversationService.get_conversation(
+                    app_model=app_model,
+                    conversation_id=payload.conversation_id,
+                    user=current_user,
+                    session=db.session(),
+                )
+
            response = AppGenerateService.generate(
                session=session,
                app_model=app_model,
@@ -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())
@@ -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
@@ -431,7 +431,7 @@ class TrialAppWorkflowRunApi(TrialAppResource):
                invoke_from=InvokeFrom.EXPLORE,
                streaming=True,
            )
-            RecommendedAppService.add_trial_app_record(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:
@@ -511,7 +511,7 @@ class TrialChatApi(TrialAppResource):
                invoke_from=InvokeFrom.EXPLORE,
                streaming=True,
            )
-            RecommendedAppService.add_trial_app_record(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:
@@ -593,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.")
@@ -654,7 +654,7 @@ class TrialChatTextApi(TrialAppResource):
                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.")
@@ -713,7 +713,7 @@ class TrialCompletionApi(TrialAppResource):
                streaming=streaming,
            )

-            RecommendedAppService.add_trial_app_record(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:
@@ -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
@@ -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")
@@ -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):
@@ -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, session=db.session())
+    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

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

@@ -135,7 +135,7 @@ def _normalize_enum_value(value: object) -> str:
 def _count_new_member_invites(tenant_id: str, emails: list[str]) -> int:
    new_member_count = 0
    for email in emails:
-        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 not account:
            new_member_count += 1
            continue
@@ -5,7 +5,7 @@ from flask import request, send_file
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator

-from controllers.common.fields import BinaryFileResponse, SimpleResultResponse
+from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
 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 (
@@ -22,8 +22,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
-from libs.helper import uuid_value
+from libs.helper import dump_response, uuid_value
 from libs.login import login_required
 from models import Account
 from services.billing_service import BillingService
@@ -92,13 +91,8 @@ class ModelProviderListResponse(ResponseModel):
    data: list[ProviderResponse]


-class ProviderCredentialResponse(ResponseModel):
-    credentials: dict[str, Any] | None = Field(default=None)
-
-
-class ProviderCredentialValidateResponse(ResponseModel):
-    result: Literal["success", "error"]
-    error: str | None = None
+class ProviderCredentialsResponse(ResponseModel):
+    credentials: dict[str, Any] | None = None


 class ModelProviderPaymentCheckoutUrlResponse(ResponseModel):
@@ -118,19 +112,20 @@ register_schema_models(
 )
 register_response_schema_models(
    console_ns,
-    BinaryFileResponse,
    SimpleResultResponse,
    ModelProviderListResponse,
+    ProviderCredentialsResponse,
+    ValidationResultResponse,
    ModelProviderPaymentCheckoutUrlResponse,
-    ProviderCredentialResponse,
-    ProviderCredentialValidateResponse,
 )


@console_ns.route("/workspaces/current/model-providers")
 class ModelProviderListApi(Resource):
    @console_ns.doc(params=query_params_from_model(ParserModelList))
-    @console_ns.response(200, "Success", console_ns.models[ModelProviderListResponse.__name__])
+    @console_ns.response(
+        200, "Model providers retrieved successfully", console_ns.models[ModelProviderListResponse.__name__]
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -142,13 +137,17 @@ class ModelProviderListApi(Resource):
        model_provider_service = ModelProviderService()
        provider_list = model_provider_service.get_provider_list(tenant_id=tenant_id, model_type=args.model_type)

-        return jsonable_encoder({"data": provider_list})
+        return ModelProviderListResponse(data=provider_list).model_dump(mode="json")


@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials")
 class ModelProviderCredentialApi(Resource):
    @console_ns.doc(params=query_params_from_model(ParserCredentialId))
-    @console_ns.response(200, "Success", console_ns.models[ProviderCredentialResponse.__name__])
+    @console_ns.response(
+        200,
+        "Provider credentials retrieved successfully",
+        console_ns.models[ProviderCredentialsResponse.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -163,7 +162,7 @@ class ModelProviderCredentialApi(Resource):
            tenant_id=tenant_id, provider=provider, credential_id=args.credential_id
        )

-        return {"credentials": credentials}
+        return ProviderCredentialsResponse(credentials=credentials).model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserCredentialCreate.__name__])
    @console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -189,7 +188,7 @@ class ModelProviderCredentialApi(Resource):
        except CredentialsValidateFailedError as ex:
            raise ValueError(str(ex))

-        return {"result": "success"}, 201
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 201

    @console_ns.expect(console_ns.models[ParserCredentialUpdate.__name__])
    @console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -216,7 +215,7 @@ class ModelProviderCredentialApi(Resource):
        except CredentialsValidateFailedError as ex:
            raise ValueError(str(ex))

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserCredentialDelete.__name__])
    @console_ns.response(204, "Credential deleted successfully")
@@ -258,7 +257,7 @@ class ModelProviderCredentialSwitchApi(Resource):
            provider=provider,
            credential_id=args.credential_id,
        )
-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/validate")
@@ -266,8 +265,8 @@ class ModelProviderValidateApi(Resource):
    @console_ns.expect(console_ns.models[ParserCredentialValidate.__name__])
    @console_ns.response(
        200,
-        "Credential validation result",
-        console_ns.models[ProviderCredentialValidateResponse.__name__],
+        "Provider credentials validated successfully",
+        console_ns.models[ValidationResultResponse.__name__],
    )
    @setup_required
    @login_required
@@ -292,12 +291,10 @@ class ModelProviderValidateApi(Resource):
            result = False
            error = str(ex)

-        response = {"result": "success" if result else "error"}
-
        if not result:
-            response["error"] = error or "Unknown error"
+            return ValidationResultResponse(result="error", error=error or "Unknown error").model_dump(mode="json")

-        return response
+        return ValidationResultResponse(result="success").model_dump(mode="json")


@console_ns.route("/workspaces/<string:tenant_id>/model-providers/<path:provider>/<string:icon_type>/<string:lang>")
@@ -306,8 +303,9 @@ class ModelProviderIconApi(Resource):
    Get model provider icon
    """

-    @console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
+    @console_ns.response(200, "Model provider icon")
    def get(self, tenant_id: str, provider: str, icon_type: str, lang: str):
+        # response-contract:ignore binary send_file response
        model_provider_service = ModelProviderService()
        icon, mimetype = model_provider_service.get_model_provider_icon(
            tenant_id=tenant_id,
@@ -339,12 +337,16 @@ class PreferredProviderTypeUpdateApi(Resource):
            tenant_id=tenant_id, provider=provider, preferred_provider_type=args.preferred_provider_type
        )

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


@console_ns.route("/workspaces/current/model-providers/<path:provider>/checkout-url")
 class ModelProviderPaymentCheckoutUrlApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__])
+    @console_ns.response(
+        200,
+        "Model provider checkout URL retrieved successfully",
+        console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -353,11 +355,11 @@ class ModelProviderPaymentCheckoutUrlApi(Resource):
    def get(self, current_tenant_id: str, current_user: Account, provider: str):
        if provider != "anthropic":
            raise ValueError(f"provider name {provider} is invalid")
-        BillingService.is_tenant_owner_or_admin(db.session(), current_user)
+        BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
        data = BillingService.get_model_provider_payment_link(
            provider_name=provider,
            tenant_id=current_tenant_id,
            account_id=current_user.id,
            prefilled_email=current_user.email,
        )
-        return data
+        return dump_response(ModelProviderPaymentCheckoutUrlResponse, data)
@@ -5,7 +5,7 @@ from flask import request
 from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator

-from controllers.common.fields import SimpleResultResponse
+from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
 from controllers.common.schema import (
    query_params_from_model,
    register_enum_models,
@@ -28,7 +28,6 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from graphon.model_runtime.entities.model_entities import ModelType, ParameterRule
 from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
-from graphon.model_runtime.utils.encoders import jsonable_encoder
 from libs.helper import uuid_value
 from libs.login import login_required
 from models import Account
@@ -63,7 +62,7 @@ class ParserDeleteModels(BaseModel):


 class LoadBalancingPayload(BaseModel):
-    configs: list[dict[str, Any]] | None = Field(default=None)
+    configs: list[dict[str, Any]] | None = None
    enabled: bool | None = None


@@ -140,33 +139,38 @@ class DefaultModelDataResponse(ResponseModel):
    data: DefaultModelResponse | None = None


-class ModelWithProviderListResponse(ResponseModel):
+class ProviderModelListResponse(ResponseModel):
    data: list[ModelWithProviderEntityResponse]


-class ProviderWithModelsDataResponse(ResponseModel):
+class AvailableModelListResponse(ResponseModel):
    data: list[ProviderWithModelsResponse]


-class ModelCredentialLoadBalancingResponse(ResponseModel):
+class ModelLoadBalancingConfigResponse(ResponseModel):
+    id: str
+    name: str
+    credentials: dict[str, Any]
+    credential_id: str | None = None
    enabled: bool
-    configs: list[dict[str, Any]] = Field(default_factory=list)
+    in_cooldown: bool
+    ttl: int
+
+
+class ModelLoadBalancingResponse(ResponseModel):
+    enabled: bool
+    configs: list[ModelLoadBalancingConfigResponse]


 class ModelCredentialResponse(ResponseModel):
-    credentials: dict[str, Any] = Field(default_factory=dict)
+    credentials: dict[str, Any]
    current_credential_id: str | None = None
    current_credential_name: str | None = None
-    load_balancing: ModelCredentialLoadBalancingResponse
+    load_balancing: ModelLoadBalancingResponse
    available_credentials: list[CredentialConfiguration]


-class ModelCredentialValidateResponse(ResponseModel):
-    result: str
-    error: str | None = None
-
-
-class ModelParameterRulesResponse(ResponseModel):
+class ModelParameterRuleListResponse(ResponseModel):
    data: list[ParameterRule]


@@ -187,12 +191,12 @@ register_schema_models(
 register_response_schema_models(
    console_ns,
    SimpleResultResponse,
+    ValidationResultResponse,
    DefaultModelDataResponse,
-    ModelWithProviderListResponse,
-    ProviderWithModelsDataResponse,
+    ProviderModelListResponse,
    ModelCredentialResponse,
-    ModelCredentialValidateResponse,
-    ModelParameterRulesResponse,
+    ModelParameterRuleListResponse,
+    AvailableModelListResponse,
 )

 register_enum_models(console_ns, ModelType)
@@ -201,7 +205,9 @@ register_enum_models(console_ns, ModelType)
@console_ns.route("/workspaces/current/default-model")
 class DefaultModelApi(Resource):
    @console_ns.doc(params=query_params_from_model(ParserGetDefault))
-    @console_ns.response(200, "Success", console_ns.models[DefaultModelDataResponse.__name__])
+    @console_ns.response(
+        200, "Default model retrieved successfully", console_ns.models[DefaultModelDataResponse.__name__]
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -214,7 +220,7 @@ class DefaultModelApi(Resource):
            tenant_id=tenant_id, model_type=args.model_type
        )

-        return jsonable_encoder({"data": default_model_entity})
+        return DefaultModelDataResponse(data=default_model_entity).model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserPostDefault.__name__])
    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@@ -247,12 +253,14 @@ class DefaultModelApi(Resource):
                )
                raise ex

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


@console_ns.route("/workspaces/current/model-providers/<path:provider>/models")
 class ModelProviderModelApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[ModelWithProviderListResponse.__name__])
+    @console_ns.response(
+        200, "Provider models retrieved successfully", console_ns.models[ProviderModelListResponse.__name__]
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -261,10 +269,10 @@ class ModelProviderModelApi(Resource):
        model_provider_service = ModelProviderService()
        models = model_provider_service.get_models_by_provider(tenant_id=tenant_id, provider=provider)

-        return jsonable_encoder({"data": models})
+        return ProviderModelListResponse(data=models).model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserPostModels.__name__])
-    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(200, "Model updated successfully", console_ns.models[SimpleResultResponse.__name__])
    @setup_required
    @login_required
    @is_admin_or_owner_required
@@ -310,7 +318,7 @@ class ModelProviderModelApi(Resource):
                    tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
                )

-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200

    @console_ns.expect(console_ns.models[ParserDeleteModels.__name__])
    @console_ns.response(204, "Model deleted successfully")
@@ -334,7 +342,11 @@ class ModelProviderModelApi(Resource):
@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials")
 class ModelProviderModelCredentialApi(Resource):
    @console_ns.doc(params=query_params_from_model(ParserGetCredentials))
-    @console_ns.response(200, "Success", console_ns.models[ModelCredentialResponse.__name__])
+    @console_ns.response(
+        200,
+        "Model credentials retrieved successfully",
+        console_ns.models[ModelCredentialResponse.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -379,22 +391,23 @@ class ModelProviderModelCredentialApi(Resource):
                model=args.model,
            )

-        return jsonable_encoder(
-            {
-                "credentials": current_credential.get("credentials") if current_credential else {},
-                "current_credential_id": current_credential.get("current_credential_id")
-                if current_credential
-                else None,
-                "current_credential_name": current_credential.get("current_credential_name")
-                if current_credential
-                else None,
-                "load_balancing": {"enabled": is_load_balancing_enabled, "configs": load_balancing_configs},
-                "available_credentials": available_credentials,
-            }
-        )
+        credentials: dict[str, Any] = {}
+        # TODO: make this throw error when type mismatches?
+        if current_credential and isinstance(current_credential.get("credentials"), dict):
+            credentials = cast(dict[str, Any], current_credential["credentials"])
+
+        return ModelCredentialResponse(
+            credentials=credentials,
+            current_credential_id=current_credential.get("current_credential_id") if current_credential else None,
+            current_credential_name=current_credential.get("current_credential_name") if current_credential else None,
+            load_balancing=ModelLoadBalancingResponse.model_validate(
+                {"enabled": is_load_balancing_enabled, "configs": load_balancing_configs}
+            ),
+            available_credentials=available_credentials,
+        ).model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserCreateCredential.__name__])
-    @console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(201, "Model credential created successfully", console_ns.models[SimpleResultResponse.__name__])
    @setup_required
    @login_required
    @is_admin_or_owner_required
@@ -424,10 +437,10 @@ class ModelProviderModelCredentialApi(Resource):
            )
            raise ValueError(str(ex))

-        return {"result": "success"}, 201
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 201

    @console_ns.expect(console_ns.models[ParserUpdateCredential.__name__])
-    @console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(200, "Model credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
    @setup_required
    @login_required
    @is_admin_or_owner_required
@@ -452,7 +465,7 @@ class ModelProviderModelCredentialApi(Resource):
        except CredentialsValidateFailedError as ex:
            raise ValueError(str(ex))

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")

    @console_ns.expect(console_ns.models[ParserDeleteCredential.__name__])
    @console_ns.response(204, "Credential deleted successfully")
@@ -498,7 +511,7 @@ class ModelProviderModelCredentialSwitchApi(Resource):
            model=args.model,
            credential_id=args.credential_id,
        )
-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


@console_ns.route(
@@ -520,7 +533,7 @@ class ModelProviderModelEnableApi(Resource):
            tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
        )

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


@console_ns.route(
@@ -542,7 +555,7 @@ class ModelProviderModelDisableApi(Resource):
            tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
        )

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump(mode="json")


 class ParserValidate(BaseModel):
@@ -559,8 +572,8 @@ class ModelProviderModelValidateApi(Resource):
    @console_ns.expect(console_ns.models[ParserValidate.__name__])
    @console_ns.response(
        200,
-        "Credential validation result",
-        console_ns.models[ModelCredentialValidateResponse.__name__],
+        "Model credentials validated successfully",
+        console_ns.models[ValidationResultResponse.__name__],
    )
    @setup_required
    @login_required
@@ -586,18 +599,20 @@ class ModelProviderModelValidateApi(Resource):
            result = False
            error = str(ex)

-        response = {"result": "success" if result else "error"}
-
        if not result:
-            response["error"] = error or ""
+            return ValidationResultResponse(result="error", error=error or "").model_dump(mode="json")

-        return response
+        return ValidationResultResponse(result="success").model_dump(mode="json")


@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/parameter-rules")
 class ModelProviderModelParameterRuleApi(Resource):
    @console_ns.doc(params=query_params_from_model(ParserParameter))
-    @console_ns.response(200, "Success", console_ns.models[ModelParameterRulesResponse.__name__])
+    @console_ns.response(
+        200,
+        "Model parameter rules retrieved successfully",
+        console_ns.models[ModelParameterRuleListResponse.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -610,12 +625,14 @@ class ModelProviderModelParameterRuleApi(Resource):
            tenant_id=tenant_id, provider=provider, model=args.model
        )

-        return jsonable_encoder({"data": parameter_rules})
+        return ModelParameterRuleListResponse(data=parameter_rules).model_dump(mode="json")


@console_ns.route("/workspaces/current/models/model-types/<string:model_type>")
 class ModelProviderAvailableModelApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[ProviderWithModelsDataResponse.__name__])
+    @console_ns.response(
+        200, "Available models retrieved successfully", console_ns.models[AvailableModelListResponse.__name__]
+    )
    @setup_required
    @login_required
    @account_initialization_required
@@ -624,4 +641,4 @@ class ModelProviderAvailableModelApi(Resource):
        model_provider_service = ModelProviderService()
        models = model_provider_service.get_models_by_model_type(tenant_id=tenant_id, model_type=model_type)

-        return jsonable_encoder({"data": models})
+        return AvailableModelListResponse(data=models).model_dump(mode="json")
@@ -31,7 +31,15 @@ from controllers.console.wraps import (
    with_current_user_id,
 )
 from core.helper.position_helper import is_filtered
-from core.plugin.entities.plugin import PluginCategory, PluginInstallationSource
+from core.plugin.entities.bundle import PluginBundleDependency
+from core.plugin.entities.parameters import PluginParameterOption
+from core.plugin.entities.plugin import (
+    PluginCategory,
+    PluginDeclaration,
+    PluginEntity,
+    PluginInstallationSource,
+)
+from core.plugin.entities.plugin_daemon import PluginDecodeResponse, PluginInstallTask, PluginInstallTaskStartResponse
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.plugin.plugin_service import PluginService
 from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
@@ -299,12 +307,12 @@ class PluginCategoryListResponse(ResponseModel):
    has_more: bool


-class PluginDaemonOperationResponse(RootModel[Any]):
-    root: Any
+class PluginBundleUploadResponse(RootModel[list[PluginBundleDependency]]):
+    pass


 class PluginListResponse(ResponseModel):
-    plugins: Any
+    plugins: list[PluginEntity]
    total: int


@@ -334,15 +342,15 @@ class PluginInstallationsResponse(ResponseModel):


 class PluginManifestResponse(ResponseModel):
-    manifest: Any
+    manifest: PluginDeclaration


 class PluginTasksResponse(ResponseModel):
-    tasks: Any
+    tasks: list[PluginInstallTask]


 class PluginTaskResponse(ResponseModel):
-    task: Any
+    task: PluginInstallTask


 class PluginPermissionResponse(ResponseModel):
@@ -351,7 +359,7 @@ class PluginPermissionResponse(ResponseModel):


 class PluginDynamicOptionsResponse(ResponseModel):
-    options: Any
+    options: list[PluginParameterOption]


 class PluginOperationSuccessResponse(ResponseModel):
@@ -398,10 +406,12 @@ register_response_schema_models(
    PluginCategoryBuiltinToolResponse,
    PluginCategoryInstalledPluginResponse,
    PluginCategoryListResponse,
-    PluginDaemonOperationResponse,
+    PluginBundleUploadResponse,
+    PluginDecodeResponse,
    PluginDebuggingKeyResponse,
    PluginDynamicOptionsResponse,
    PluginInstallationsResponse,
+    PluginInstallTaskStartResponse,
    PluginListResponse,
    PluginManifestResponse,
    PluginOperationSuccessResponse,
@@ -635,7 +645,7 @@ class PluginAssetApi(Resource):

@console_ns.route("/workspaces/current/plugin/upload/pkg")
 class PluginUploadFromPkgApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -656,7 +666,7 @@ class PluginUploadFromPkgApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/github")
 class PluginUploadFromGithubApi(Resource):
    @console_ns.expect(console_ns.models[ParserGithubUpload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -676,7 +686,7 @@ class PluginUploadFromGithubApi(Resource):

@console_ns.route("/workspaces/current/plugin/upload/bundle")
 class PluginUploadFromBundleApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginBundleUploadResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -697,7 +707,7 @@ class PluginUploadFromBundleApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/pkg")
 class PluginInstallFromPkgApi(Resource):
    @console_ns.expect(console_ns.models[ParserPluginIdentifiers.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -718,7 +728,7 @@ class PluginInstallFromPkgApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/github")
 class PluginInstallFromGithubApi(Resource):
    @console_ns.expect(console_ns.models[ParserGithubInstall.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -745,7 +755,7 @@ class PluginInstallFromGithubApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/marketplace")
 class PluginInstallFromMarketplaceApi(Resource):
    @console_ns.expect(console_ns.models[ParserPluginIdentifiers.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -891,7 +901,7 @@ class PluginDeleteInstallTaskItemApi(Resource):
@console_ns.route("/workspaces/current/plugin/upgrade/marketplace")
 class PluginUpgradeFromMarketplaceApi(Resource):
    @console_ns.expect(console_ns.models[ParserMarketplaceUpgrade.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -914,7 +924,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
@console_ns.route("/workspaces/current/plugin/upgrade/github")
 class PluginUpgradeFromGithubApi(Resource):
    @console_ns.expect(console_ns.models[ParserGithubUpgrade.__name__])
-    @console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
+    @console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
    @setup_required
    @login_required
    @account_initialization_required
@@ -18,6 +18,7 @@ from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from services.enterprise import rbac_service as svc
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task


 class _RBACRoleList(svc.Paginated[svc.RBACRole]):
@@ -596,14 +597,15 @@ class RBACAppWhitelistApi(Resource):
    def put(self, app_id):
        tenant_id, account_id = _current_ids()
        request = _payload(_ResourceAccessScopeRequest)
-        return _dump(
-            svc.RBACService.AppAccess.replace_whitelist(
-                tenant_id,
-                account_id,
-                str(app_id),
-                svc.ReplaceMemberBindings(scope=request.scope.value),
-            )
+        result = svc.RBACService.AppAccess.replace_whitelist(
+            tenant_id,
+            account_id,
+            str(app_id),
+            svc.ReplaceMemberBindings(scope=request.scope.value),
        )
+        if dify_config.RBAC_ENABLED and request.scope is RBACResourceWhitelistScope.ALL:
+            initialize_created_app_rbac_access_task.delay(tenant_id, account_id, str(app_id))
+        return _dump(result)


@console_ns.route("/workspaces/current/rbac/apps/<uuid:app_id>/user-access-policies")
@@ -630,8 +632,8 @@ class RBACAppUserAccessPolicyAssignmentApi(Resource):
            svc.RBACService.AppAccess.replace_user_access_policies(
                tenant_id,
                account_id,
-                str(app_id),
-                str(target_account_id),
+                app_id,
+                target_account_id,
                payload,
            )
        )
@@ -1,12 +1,9 @@
 import logging
-from datetime import datetime
-from typing import Any
 from urllib.parse import quote
+from uuid import UUID

 from flask import Response, request
 from flask_restx import Resource
-from pydantic import Field as PydanticField
-from pydantic import field_validator
 from sqlalchemy.orm import Session, sessionmaker
 from werkzeug.exceptions import NotFound

@@ -37,7 +34,8 @@ from controllers.console.wraps import (
 from core.plugin.entities.plugin import PluginDependency
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from libs.helper import dump_response, to_timestamp
+from fields.snippet_fields import SnippetListItemResponse, SnippetPaginationResponse, SnippetResponse
+from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
 from models.snippet import SnippetType
@@ -65,77 +63,6 @@ class SnippetUseCountResponse(ResponseModel):
    use_count: int


-class SnippetTagResponse(ResponseModel):
-    id: str
-    name: str
-    type: str
-
-
-class SnippetAccountResponse(ResponseModel):
-    id: str
-    name: str
-    email: str
-
-
-class SnippetListItemResponse(ResponseModel):
-    id: str
-    name: str
-    description: str | None
-    type: SnippetType
-    version: int
-    use_count: int
-    is_published: bool
-    icon_info: dict[str, Any] | None
-    tags: list[SnippetTagResponse]
-    created_by: str | None
-    author_name: str | None
-    created_at: int
-    updated_by: str | None
-    updated_at: int
-
-    @field_validator("created_at", "updated_at", mode="before")
-    @classmethod
-    def _normalize_timestamp(cls, value: datetime | int | None) -> int:
-        timestamp = to_timestamp(value)
-        if timestamp is None:
-            raise ValueError("timestamp is required")
-        return timestamp
-
-
-class SnippetResponse(ResponseModel):
-    id: str
-    name: str
-    description: str | None
-    type: SnippetType
-    version: int
-    use_count: int
-    is_published: bool
-    icon_info: dict[str, Any] | None
-    graph: dict[str, Any] = PydanticField(validation_alias="graph_dict")
-    input_fields: list[dict[str, Any]] = PydanticField(validation_alias="input_fields_list")
-    tags: list[SnippetTagResponse]
-    created_by: SnippetAccountResponse | None = PydanticField(validation_alias="created_by_account")
-    created_at: int
-    updated_by: SnippetAccountResponse | None = PydanticField(validation_alias="updated_by_account")
-    updated_at: int
-
-    @field_validator("created_at", "updated_at", mode="before")
-    @classmethod
-    def _normalize_timestamp(cls, value: datetime | int | None) -> int:
-        timestamp = to_timestamp(value)
-        if timestamp is None:
-            raise ValueError("timestamp is required")
-        return timestamp
-
-
-class SnippetPaginationResponse(ResponseModel):
-    data: list[SnippetListItemResponse]
-    page: int
-    limit: int
-    total: int
-    has_more: bool
-
-
 def _snippet_service() -> SnippetService:
    return SnippetService(sessionmaker(bind=db.engine, expire_on_commit=False))

@@ -259,7 +186,7 @@ class CustomizedSnippetDetailApi(Resource):
    @login_required
    @account_initialization_required
    @with_current_tenant_id
-    def get(self, current_tenant_id: str, snippet_id: str):
+    def get(self, current_tenant_id: str, snippet_id: UUID):
        """Get customized snippet details."""
        snippet_service = _snippet_service()
        snippet = snippet_service.get_snippet_by_id(
@@ -535,4 +462,4 @@ class CustomizedSnippetUseCountIncrementApi(Resource):
            session.commit()
            session.refresh(snippet)

-        return SnippetUseCountResponse(result="success", use_count=snippet.use_count).model_dump(mode="json"), 200
+        return {"result": "success", "use_count": snippet.use_count}, 200
@@ -1,20 +1,19 @@
 import logging
-from typing import Any, Literal
+from typing import Any

 from flask import make_response, redirect, request
 from flask_restx import Resource
-from pydantic import BaseModel, Field, RootModel, model_validator
+from pydantic import BaseModel, RootModel, model_validator
 from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import BadRequest, Forbidden

 from configs import dify_config
 from controllers.common.errors import NotFoundError
-from controllers.common.fields import BinaryFileResponse, RedirectResponse, SimpleResultResponse
+from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
-from core.entities.parameter_entities import AppSelectorScope, ModelSelectorScope, ToolSelectorScope
+from core.entities.provider_entities import ProviderConfig
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
-from core.tools.entities.common_entities import I18nObject
 from core.trigger.entities.api_entities import (
    SubscriptionBuilderApiEntity,
    TriggerProviderApiEntity,
@@ -24,7 +23,7 @@ from core.trigger.entities.entities import RequestLog, SubscriptionBuilderUpdate
 from core.trigger.trigger_manager import TriggerManager
 from extensions.ext_database import db
 from fields.base import ResponseModel
-from graphon.model_runtime.utils.encoders import jsonable_encoder
+from libs.helper import dump_response
 from libs.login import login_required
 from models.account import Account
 from models.provider_ids import TriggerProviderID
@@ -59,9 +58,9 @@ class TriggerSubscriptionBuilderVerifyPayload(BaseModel):

 class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
    name: str | None = None
-    parameters: dict[str, Any] | None = Field(default=None)
-    properties: dict[str, Any] | None = Field(default=None)
-    credentials: dict[str, Any] | None = Field(default=None)
+    parameters: dict[str, Any] | None = None
+    properties: dict[str, Any] | None = None
+    credentials: dict[str, Any] | None = None

    @model_validator(mode="after")
    def check_at_least_one_field(self):
@@ -71,70 +70,23 @@ class TriggerSubscriptionBuilderUpdatePayload(BaseModel):


 class TriggerOAuthClientPayload(BaseModel):
-    client_params: dict[str, Any] | None = Field(default=None)
+    client_params: dict[str, Any] | None = None
    enabled: bool | None = None


-class TriggerOAuthAuthorizeResponse(BaseModel):
-    authorization_url: str
-    subscription_builder_id: str
-    subscription_builder: SubscriptionBuilderApiEntity
-
-
-class TriggerProviderConfigOptionResponse(BaseModel):
-    value: str = Field(..., description="The value of the option")
-    label: I18nObject = Field(..., description="The label of the option")
-
-
-class TriggerProviderConfigResponse(BaseModel):
-    type: Literal[
-        "secret-input",
-        "text-input",
-        "select",
-        "boolean",
-        "app-selector",
-        "model-selector",
-        "array[tools]",
-    ] = Field(..., description="The type of the credentials")
-    name: str = Field(..., description="The name of the credentials")
-    scope: AppSelectorScope | ModelSelectorScope | ToolSelectorScope | None = None
-    required: bool = False
-    default: int | str | float | bool | None = None
-    options: list[TriggerProviderConfigOptionResponse] | None = None
-    multiple: bool = False
-    label: I18nObject | None = None
-    help: I18nObject | None = None
-    url: str | None = None
-    placeholder: I18nObject | None = None
-
-
-class TriggerOAuthClientResponse(BaseModel):
-    configured: bool
-    system_configured: bool
-    custom_configured: bool
-    oauth_client_schema: list[TriggerProviderConfigResponse]
-    custom_enabled: bool
-    redirect_uri: str
-    params: dict[str, Any]
-
-
-class TriggerProviderOpaqueResponse(RootModel[Any]):
-    root: Any
-
-
 class TriggerProviderListResponse(RootModel[list[TriggerProviderApiEntity]]):
-    root: list[TriggerProviderApiEntity]
+    pass


-class TriggerSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
-    root: list[TriggerProviderSubscriptionApiEntity]
+class TriggerProviderSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
+    pass


 class TriggerSubscriptionBuilderCreateResponse(ResponseModel):
    subscription_builder: SubscriptionBuilderApiEntity


-class TriggerSubscriptionBuilderVerifyResponse(ResponseModel):
+class TriggerVerificationResponse(ResponseModel):
    verified: bool


@@ -142,6 +94,26 @@ class TriggerSubscriptionBuilderLogsResponse(ResponseModel):
    logs: list[RequestLog]


+class TriggerOAuthAuthorizeResponse(ResponseModel):
+    authorization_url: str
+    subscription_builder_id: str
+    subscription_builder: SubscriptionBuilderApiEntity
+
+
+class TriggerOAuthClientResponse(ResponseModel):
+    configured: bool
+    system_configured: bool
+    custom_configured: bool
+    oauth_client_schema: list[ProviderConfig]
+    custom_enabled: bool
+    redirect_uri: str
+    params: dict[str, Any]
+
+
+class TriggerProviderErrorResponse(ResponseModel):
+    error: str
+
+
 register_schema_models(
    console_ns,
    TriggerSubscriptionBuilderCreatePayload,
@@ -151,27 +123,24 @@ register_schema_models(
 )
 register_response_schema_models(
    console_ns,
-    BinaryFileResponse,
-    RedirectResponse,
    SimpleResultResponse,
    TriggerOAuthAuthorizeResponse,
    TriggerOAuthClientResponse,
-    TriggerProviderOpaqueResponse,
    TriggerProviderApiEntity,
+    TriggerProviderErrorResponse,
    TriggerProviderListResponse,
-    TriggerProviderSubscriptionApiEntity,
-    TriggerSubscriptionListResponse,
-    SubscriptionBuilderApiEntity,
+    TriggerProviderSubscriptionListResponse,
    TriggerSubscriptionBuilderCreateResponse,
-    TriggerSubscriptionBuilderVerifyResponse,
-    RequestLog,
    TriggerSubscriptionBuilderLogsResponse,
+    SubscriptionBuilderApiEntity,
+    TriggerVerificationResponse,
 )


@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/icon")
 class TriggerProviderIconApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
+    # response-contract:ignore binary trigger provider icon
+    @console_ns.response(200, "Trigger provider icon")
    @setup_required
    @login_required
    @account_initialization_required
@@ -182,31 +151,45 @@ class TriggerProviderIconApi(Resource):

@console_ns.route("/workspaces/current/triggers")
 class TriggerProviderListApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[TriggerProviderListResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger providers retrieved successfully",
+        console_ns.models[TriggerProviderListResponse.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
    @with_current_tenant_id
    def get(self, tenant_id: str):
        """List all trigger providers for the current tenant"""
-        return jsonable_encoder(TriggerProviderService.list_trigger_providers(tenant_id))
+        return dump_response(TriggerProviderListResponse, TriggerProviderService.list_trigger_providers(tenant_id))


@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/info")
 class TriggerProviderInfoApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[TriggerProviderApiEntity.__name__])
+    @console_ns.response(
+        200,
+        "Trigger provider retrieved successfully",
+        console_ns.models[TriggerProviderApiEntity.__name__],
+    )
    @setup_required
    @login_required
    @account_initialization_required
    @with_current_tenant_id
    def get(self, tenant_id: str, provider: str):
        """Get info for a trigger provider"""
-        return jsonable_encoder(TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider)))
+        provider_entity = TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider))
+        return provider_entity.model_dump(mode="json")


@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
 class TriggerSubscriptionListApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionListResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscriptions retrieved successfully",
+        console_ns.models[TriggerProviderSubscriptionListResponse.__name__],
+    )
+    @console_ns.response(404, "Trigger provider not found", console_ns.models[TriggerProviderErrorResponse.__name__])
    @setup_required
    @login_required
    @edit_permission_required
@@ -216,16 +199,18 @@ class TriggerSubscriptionListApi(Resource):
    @with_current_tenant_id
    def get(self, tenant_id: str, user: Account, provider: str):
        """List all trigger subscriptions for the current tenant's provider"""
+
        try:
-            return jsonable_encoder(
+            return dump_response(
+                TriggerProviderSubscriptionListResponse,
                TriggerProviderService.list_trigger_provider_subscriptions(
                    tenant_id=tenant_id,
                    provider_id=TriggerProviderID(provider),
                    user=user,
-                )
+                ),
            )
        except ValueError as e:
-            return jsonable_encoder({"error": str(e)}), 404
+            return TriggerProviderErrorResponse(error=str(e)).model_dump(mode="json"), 404
        except Exception as e:
            logger.exception("Error listing trigger providers", exc_info=e)
            raise
@@ -236,7 +221,11 @@ class TriggerSubscriptionListApi(Resource):
 )
 class TriggerSubscriptionBuilderCreateApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderCreatePayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription builder created successfully",
+        console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -246,6 +235,7 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, user: Account, provider: str):
        """Add a new subscription instance for a trigger provider"""
+
        payload = TriggerSubscriptionBuilderCreatePayload.model_validate(console_ns.payload or {})

        try:
@@ -256,7 +246,9 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
                provider_id=TriggerProviderID(provider),
                credential_type=credential_type,
            )
-            return jsonable_encoder({"subscription_builder": subscription_builder})
+            return TriggerSubscriptionBuilderCreateResponse(subscription_builder=subscription_builder).model_dump(
+                mode="json"
+            )
        except Exception as e:
            logger.exception("Error adding provider credential", exc_info=e)
            raise
@@ -266,7 +258,11 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
 )
 class TriggerSubscriptionBuilderGetApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription builder retrieved successfully",
+        console_ns.models[SubscriptionBuilderApiEntity.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -274,9 +270,8 @@ class TriggerSubscriptionBuilderGetApi(Resource):
    @account_initialization_required
    def get(self, provider: str, subscription_builder_id: str):
        """Get a subscription instance for a trigger provider"""
-        return jsonable_encoder(
-            TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
-        )
+        subscription_builder = TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
+        return subscription_builder.model_dump(mode="json")


@console_ns.route(
@@ -284,7 +279,11 @@ class TriggerSubscriptionBuilderGetApi(Resource):
 )
 class TriggerSubscriptionBuilderVerifyApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription builder verified successfully",
+        console_ns.models[TriggerVerificationResponse.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -294,11 +293,12 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, user: Account, provider: str, subscription_builder_id: str):
        """Verify and update a subscription instance for a trigger provider"""
+
        payload = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})

        try:
            # Use atomic update_and_verify to prevent race conditions
-            return TriggerSubscriptionBuilderService.update_and_verify_builder(
+            result = TriggerSubscriptionBuilderService.update_and_verify_builder(
                tenant_id=tenant_id,
                user_id=user.id,
                provider_id=TriggerProviderID(provider),
@@ -307,6 +307,7 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
                    credentials=payload.credentials,
                ),
            )
+            return dump_response(TriggerVerificationResponse, result)
        except Exception as e:
            logger.exception("Error verifying provider credential", exc_info=e)
            raise ValueError(str(e)) from e
@@ -317,7 +318,11 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
 )
 class TriggerSubscriptionBuilderUpdateApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription builder updated successfully",
+        console_ns.models[SubscriptionBuilderApiEntity.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -326,21 +331,20 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, provider: str, subscription_builder_id: str):
        """Update a subscription instance for a trigger provider"""
+
        payload = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
        try:
-            return jsonable_encoder(
-                TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
-                    tenant_id=tenant_id,
-                    provider_id=TriggerProviderID(provider),
-                    subscription_builder_id=subscription_builder_id,
-                    subscription_builder_updater=SubscriptionBuilderUpdater(
-                        name=payload.name,
-                        parameters=payload.parameters,
-                        properties=payload.properties,
-                        credentials=payload.credentials,
-                    ),
-                )
-            )
+            return TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+                tenant_id=tenant_id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+                subscription_builder_updater=SubscriptionBuilderUpdater(
+                    name=payload.name,
+                    parameters=payload.parameters,
+                    properties=payload.properties,
+                    credentials=payload.credentials,
+                ),
+            ).model_dump(mode="json")
        except Exception as e:
            logger.exception("Error updating provider credential", exc_info=e)
            raise
@@ -350,7 +354,11 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
 )
 class TriggerSubscriptionBuilderLogsApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription builder logs retrieved successfully",
+        console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -358,9 +366,10 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
    @account_initialization_required
    def get(self, provider: str, subscription_builder_id: str):
        """Get the request logs for a subscription instance for a trigger provider"""
+
        try:
            logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id)
-            return jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
+            return dump_response(TriggerSubscriptionBuilderLogsResponse, {"logs": logs})
        except Exception as e:
            logger.exception("Error getting request logs for subscription builder", exc_info=e)
            raise
@@ -371,7 +380,9 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
 )
 class TriggerSubscriptionBuilderBuildApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
+    @console_ns.response(
+        200, "Trigger subscription builder built successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -395,7 +406,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
                    properties=payload.properties,
                ),
            )
-            return 200
+            return SimpleResultResponse(result="success").model_dump(mode="json")
        except Exception as e:
            logger.exception("Error building provider credential", exc_info=e)
            raise ValueError(str(e)) from e
@@ -406,7 +417,9 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
 )
 class TriggerSubscriptionUpdateApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
+    @console_ns.response(
+        200, "Trigger subscription updated successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -415,6 +428,7 @@ class TriggerSubscriptionUpdateApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, subscription_id: str):
        """Update a subscription instance"""
+
        request = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})

        subscription = TriggerProviderService.get_subscription_by_id(
@@ -440,7 +454,7 @@ class TriggerSubscriptionUpdateApi(Resource):
                    name=request.name,
                    properties=request.properties,
                )
-                return 200
+                return SimpleResultResponse(result="success").model_dump(mode="json")

            # For the rest cases(API_KEY, OAUTH2)
            # we need to call third party provider(e.g. GitHub) to rebuild the subscription
@@ -452,7 +466,7 @@ class TriggerSubscriptionUpdateApi(Resource):
                credentials=request.credentials or subscription.credentials,
                parameters=request.parameters or subscription.parameters,
            )
-            return 200
+            return SimpleResultResponse(result="success").model_dump(mode="json")
        except ValueError as e:
            raise BadRequest(str(e))
        except Exception as e:
@@ -473,6 +487,7 @@ class TriggerSubscriptionDeleteApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, subscription_id: str):
        """Delete a subscription instance"""
+
        try:
            with sessionmaker(db.engine).begin() as session:
                # Delete trigger provider subscription
@@ -487,7 +502,7 @@ class TriggerSubscriptionDeleteApi(Resource):
                    tenant_id=tenant_id,
                    subscription_id=subscription_id,
                )
-            return {"result": "success"}
+            return SimpleResultResponse(result="success").model_dump(mode="json")
        except ValueError as e:
            raise BadRequest(str(e))
        except Exception as e:
@@ -497,9 +512,10 @@ class TriggerSubscriptionDeleteApi(Resource):

@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/subscriptions/oauth/authorize")
 class TriggerOAuthAuthorizeApi(Resource):
+    # response-contract:ignore cookie-bearing Flask response
    @console_ns.response(
        200,
-        "Authorization URL retrieved successfully",
+        "Trigger OAuth authorization URL generated successfully",
        console_ns.models[TriggerOAuthAuthorizeResponse.__name__],
    )
    @setup_required
@@ -509,10 +525,12 @@ class TriggerOAuthAuthorizeApi(Resource):
    @with_current_tenant_id
    def get(self, tenant_id: str, user: Account, provider: str):
        """Initiate OAuth authorization flow for a trigger provider"""
+
        try:
            provider_id = TriggerProviderID(provider)
            plugin_id = provider_id.plugin_id
            provider_name = provider_id.provider_name
+            tenant_id = tenant_id

            # Get OAuth client configuration
            oauth_client_params = TriggerProviderService.get_oauth_client(
@@ -556,15 +574,12 @@ class TriggerOAuthAuthorizeApi(Resource):
                system_credentials=oauth_client_params,
            )

-            # Create response with cookie
            response = make_response(
-                jsonable_encoder(
-                    {
-                        "authorization_url": authorization_url_response.authorization_url,
-                        "subscription_builder_id": subscription_builder.id,
-                        "subscription_builder": subscription_builder,
-                    }
-                )
+                TriggerOAuthAuthorizeResponse(
+                    authorization_url=authorization_url_response.authorization_url,
+                    subscription_builder_id=subscription_builder.id,
+                    subscription_builder=subscription_builder,
+                ).model_dump(mode="json")
            )
            response.set_cookie(
                "context_id",
@@ -583,11 +598,8 @@ class TriggerOAuthAuthorizeApi(Resource):

@console_ns.route("/oauth/plugin/<path:provider>/trigger/callback")
 class TriggerOAuthCallbackApi(Resource):
-    @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: str):
        """Handle OAuth callback for trigger provider"""
@@ -653,7 +665,11 @@ class TriggerOAuthCallbackApi(Resource):

@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/oauth/client")
 class TriggerOAuthClientManageApi(Resource):
-    @console_ns.response(200, "Success", console_ns.models[TriggerOAuthClientResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger OAuth client retrieved successfully",
+        console_ns.models[TriggerOAuthClientResponse.__name__],
+    )
    @setup_required
    @login_required
    @is_admin_or_owner_required
@@ -662,6 +678,7 @@ class TriggerOAuthClientManageApi(Resource):
    @with_current_tenant_id
    def get(self, tenant_id: str, provider: str):
        """Get OAuth client configuration for a provider"""
+
        try:
            provider_id = TriggerProviderID(provider)

@@ -682,24 +699,24 @@ class TriggerOAuthClientManageApi(Resource):
            )
            provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id)
            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
-            return jsonable_encoder(
-                {
-                    "configured": bool(custom_params or system_client_exists),
-                    "system_configured": system_client_exists,
-                    "custom_configured": bool(custom_params),
-                    "oauth_client_schema": provider_controller.get_oauth_client_schema(),
-                    "custom_enabled": is_custom_enabled,
-                    "redirect_uri": redirect_uri,
-                    "params": custom_params or {},
-                }
-            )
+            return TriggerOAuthClientResponse(
+                configured=bool(custom_params or system_client_exists),
+                system_configured=system_client_exists,
+                custom_configured=bool(custom_params),
+                oauth_client_schema=provider_controller.get_oauth_client_schema(),
+                custom_enabled=is_custom_enabled,
+                redirect_uri=redirect_uri,
+                params=dict(custom_params or {}),
+            ).model_dump(mode="json")

        except Exception as e:
            logger.exception("Error getting OAuth client", exc_info=e)
            raise

    @console_ns.expect(console_ns.models[TriggerOAuthClientPayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
+    @console_ns.response(
+        200, "Trigger OAuth client saved successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
    @setup_required
    @login_required
    @is_admin_or_owner_required
@@ -708,16 +725,18 @@ class TriggerOAuthClientManageApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, provider: str):
        """Configure custom OAuth client for a provider"""
+
        payload = TriggerOAuthClientPayload.model_validate(console_ns.payload or {})

        try:
            provider_id = TriggerProviderID(provider)
-            return TriggerProviderService.save_custom_oauth_client_params(
+            result = TriggerProviderService.save_custom_oauth_client_params(
                tenant_id=tenant_id,
                provider_id=provider_id,
                client_params=payload.client_params,
                enabled=payload.enabled,
            )
+            return dump_response(SimpleResultResponse, result)

        except ValueError as e:
            raise BadRequest(str(e))
@@ -725,22 +744,26 @@ class TriggerOAuthClientManageApi(Resource):
            logger.exception("Error configuring OAuth client", exc_info=e)
            raise

+    @console_ns.response(
+        200, "Trigger OAuth client deleted successfully", console_ns.models[SimpleResultResponse.__name__]
+    )
    @setup_required
    @login_required
    @is_admin_or_owner_required
    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False)
    @account_initialization_required
-    @console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
    @with_current_tenant_id
    def delete(self, tenant_id: str, provider: str):
        """Remove custom OAuth client configuration"""
+
        try:
            provider_id = TriggerProviderID(provider)

-            return TriggerProviderService.delete_custom_oauth_client_params(
+            result = TriggerProviderService.delete_custom_oauth_client_params(
                tenant_id=tenant_id,
                provider_id=provider_id,
            )
+            return dump_response(SimpleResultResponse, result)
        except ValueError as e:
            raise BadRequest(str(e))
        except Exception as e:
@@ -753,7 +776,11 @@ class TriggerOAuthClientManageApi(Resource):
 )
 class TriggerSubscriptionVerifyApi(Resource):
    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
-    @console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
+    @console_ns.response(
+        200,
+        "Trigger subscription verified successfully",
+        console_ns.models[TriggerVerificationResponse.__name__],
+    )
    @setup_required
    @login_required
    @edit_permission_required
@@ -763,6 +790,7 @@ class TriggerSubscriptionVerifyApi(Resource):
    @with_current_tenant_id
    def post(self, tenant_id: str, user: Account, provider: str, subscription_id: str):
        """Verify credentials for an existing subscription (edit mode only)"""
+
        verify_request = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})

        try:
@@ -773,7 +801,7 @@ class TriggerSubscriptionVerifyApi(Resource):
                subscription_id=subscription_id,
                credentials=verify_request.credentials,
            )
-            return result
+            return dump_response(TriggerVerificationResponse, result)
        except ValueError as e:
            logger.warning("Credential verification failed", exc_info=e)
            raise BadRequest(str(e)) from e
@@ -223,7 +223,7 @@ class TenantListApi(Resource):
    def get(self, current_tenant_id: str, current_user: Account):
        tenant_rows: list[tuple[Tenant, TenantAccountJoin]] = [
            (tenant, membership)
-            for tenant, membership in TenantService.get_workspaces_for_account(db.session(), current_user.id)
+            for tenant, membership in TenantService.get_workspaces_for_account(current_user.id, session=db.session())
            if tenant.status == TenantStatus.NORMAL
        ]
        tenants = [tenant for tenant, _ in tenant_rows]
@@ -1,5 +1,3 @@
-from mimetypes import guess_extension
-
 from flask import request
 from flask_restx import Resource
 from flask_restx.api import HTTPStatus
@@ -8,7 +6,7 @@ from werkzeug.exceptions import Forbidden

 import services
 from core.tools.signature import verify_plugin_file_signature
-from core.tools.tool_file_manager import ToolFileManager
+from core.tools.tool_file_manager import ToolFileManager, resolve_extension
 from core.workflow.file_reference import build_file_reference
 from fields.file_fields import FileResponse

@@ -110,7 +108,7 @@ class PluginUploadFileApi(Resource):
                conversation_id=args.conversation_id,
            )

-            extension = guess_extension(tool_file.mimetype) or ".bin"
+            extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
            preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)

            # Create a dictionary with all the necessary attributes
@@ -476,6 +476,7 @@ class PluginDownloadFileRequestApi(Resource):
            user_from=payload.user_from,
            invoke_from=payload.invoke_from,
            file_mapping=payload.file.model_dump(mode="python", exclude_none=True),
+            for_external=payload.for_external,
        )
        return BaseBackwardsInvocationResponse(
            data={
@@ -2,11 +2,11 @@ from typing import Any, Union

 from flask import Response, request
 from flask_restx import Resource
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, Field, RootModel, ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker

-from controllers.common.schema import register_schema_model
+from controllers.common.schema import register_response_schema_models, register_schema_model
 from controllers.mcp import mcp_ns
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request, negotiate_protocol_version
@@ -33,7 +33,12 @@ class MCPRequestPayload(BaseModel):
    id: int | str | None = Field(default=None, description="Request ID for tracking responses")


+class MCPJSONRPCResponse(RootModel[mcp_types.JSONRPCResponse | mcp_types.JSONRPCError]):
+    pass
+
+
 register_schema_model(mcp_ns, MCPRequestPayload)
+register_response_schema_models(mcp_ns, MCPJSONRPCResponse)


@mcp_ns.route("/server/<string:server_code>/mcp")
@@ -42,13 +47,10 @@ class MCPAppApi(Resource):
    @mcp_ns.doc("handle_mcp_request")
    @mcp_ns.doc(description="Handle Model Context Protocol (MCP) requests for a specific server")
    @mcp_ns.doc(params={"server_code": "Unique identifier for the MCP server"})
-    @mcp_ns.doc(
-        responses={
-            200: "MCP response successfully processed",
-            400: "Invalid MCP request or parameters",
-            404: "Server or app not found",
-        }
-    )
+    @mcp_ns.response(200, "MCP JSON-RPC response", mcp_ns.models[MCPJSONRPCResponse.__name__])
+    @mcp_ns.response(202, "MCP notification accepted")
+    @mcp_ns.response(400, "Invalid MCP request or parameters")
+    @mcp_ns.response(404, "Server or app not found")
    def post(self, server_code: str):
        """Handle MCP requests for a specific server.

@@ -64,6 +66,7 @@ class MCPAppApi(Resource):
        Raises:
            ValidationError: Invalid request format or parameters
        """
+        # response-contract:ignore MCP route returns Flask Response from JSON-RPC handler
        args = MCPRequestPayload.model_validate(mcp_ns.payload or {})
        request_id: Union[int, str] | None = args.id
        mcp_request = self._parse_mcp_request(args.model_dump(exclude_none=True))
@@ -2,11 +2,13 @@ from flask import Blueprint
 from flask_restx import Namespace

 from controllers.openapi._errors import ErrorBody, OpenApiErrorCode, OpenApiErrorFormatter
+from controllers.openapi._version_gate import attach_version_gate
 from libs.device_flow_security import attach_anti_framing
 from libs.external_api import ExternalApi

 bp = Blueprint("openapi", __name__, url_prefix="/openapi/v1")
 attach_anti_framing(bp)
+attach_version_gate(bp)

 api = ExternalApi(
    bp,
@@ -45,6 +45,7 @@ class OpenApiErrorCode(StrEnum):
    TOO_MANY_REQUESTS = "too_many_requests"
    INTERNAL_ERROR = "internal_server_error"
    BAD_GATEWAY = "bad_gateway"
+    UPGRADE_REQUIRED = "upgrade_required"
    UNKNOWN = "unknown"
    # domain codes (must match the error_code attribute of the exception
    # classes raised on the openapi surface)
@@ -279,7 +279,7 @@ def _csv_string_query_schema(schema: dict[str, Any]) -> None:


 class AppDescribeQuery(BaseModel):
-    """`?fields=` allow-list for GET /apps/<id>/describe.
+    """`?fields=` allow-list for GET /apps/<id>.

    Empty / omitted → all blocks. Unknown member → ValidationError → 422.
    """
@@ -441,7 +441,7 @@ class MemberActionResponse(BaseModel):


 class TaskStopResponse(BaseModel):
-    """200 body for POST /apps/<id>/tasks/<task_id>/stop. The handler always returns
+    """200 body for POST /apps/<id>/tasks/<task_id>:stop. The handler always returns
    {"result": "success"}, so `result` is required (no default) — the generated contract
    types it as a required `'success'` rather than an optional field."""

@@ -473,7 +473,7 @@ class AppDslImportPayload(BaseModel):


 class AppDslExportQuery(BaseModel):
-    """Query parameters for GET /apps/<app_id>/export."""
+    """Query parameters for GET /apps/<app_id>/dsl."""

    include_secret: bool = Field(False, description="Include encrypted secret values in the exported DSL")
    workflow_id: UUIDStr | None = Field(
@@ -488,7 +488,7 @@ class AppDslExportResponse(BaseModel):


 class FormSubmitResponse(BaseModel):
-    """Empty 200 body for POST /apps/<id>/form/human_input/<token>. `extra='forbid'`
+    """Empty 200 body for POST /apps/<id>/human-input-forms/<token>:submit. `extra='forbid'`
    pins `additionalProperties: false` so the generated contract is an exact `{}` rather
    than an under-annotated open object."""

@@ -0,0 +1,69 @@
+"""Version gate: reject outdated difyctl clients on /openapi/v1 with HTTP 426.
+
+difyctl and the ``/openapi/v1`` surface ship in lockstep. A breaking path change
+(resource-oriented paths) means an outdated difyctl would call removed paths and
+get a bare 404; this gate returns ``426 Upgrade Required`` with an upgrade hint
+instead.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Final
+
+from flask import Blueprint, Response, request
+from packaging.version import InvalidVersion, Version
+
+from configs import dify_config
+from controllers.openapi._errors import ErrorBody, OpenApiErrorCode
+
+_UPGRADE_HINT: Final = "Upgrade difyctl: https://docs.dify.ai/en/cli/install"
+
+# difyctl sends `User-Agent: difyctl/<semver> (<os>; <arch>; <channel>)`.
+_DIFYCTL_UA_RE = re.compile(r"^difyctl/(\d+\.\d+\.\d+(?:-[\w.]+)?)")
+
+_PREFIX: Final = "/openapi/v1/"
+
+# Paths a too-old client must still reach to discover that it is outdated.
+_ALLOWLIST: Final = frozenset({"/openapi/v1/_version", "/openapi/v1/_health"})
+
+
+def _upgrade_required_response(client_version: str, min_version: str) -> Response:
+    body = ErrorBody(
+        code=OpenApiErrorCode.UPGRADE_REQUIRED,
+        message=f"difyctl {client_version} is no longer supported; upgrade to >= {min_version}.",
+        status=426,
+        hint=_UPGRADE_HINT,
+    )
+    return Response(body.model_dump_json(exclude_none=True), status=426, mimetype="application/json")
+
+
+def attach_version_gate(bp: Blueprint) -> None:
+    """Reject difyctl clients older than ``[tool.dify] min_difyctl_version`` with 426.
+
+    Registered app-wide (``before_app_request``) rather than blueprint-scoped so it
+    also fires for requests to *removed* paths — those no longer match an openapi
+    route and would 404 before a blueprint-scoped ``before_request`` ever runs. The
+    prefix guard scopes it back to ``/openapi/v1``. Fails open for non-difyctl or
+    unparseable User-Agents (only a confidently-too-old difyctl is blocked).
+    """
+
+    @bp.before_app_request
+    def _enforce_min_client_version() -> Response | None:  # pyright: ignore[reportUnusedFunction]
+        if not request.path.startswith(_PREFIX):
+            return None
+        if request.path in _ALLOWLIST:
+            return None
+        match = _DIFYCTL_UA_RE.match(request.headers.get("User-Agent", ""))
+        if match is None:
+            return None
+        try:
+            client_version = Version(match.group(1))
+        except InvalidVersion:
+            return None
+        # Compare the numeric core (major.minor.patch) only — a pre-release build
+        # like 0.2.0-rc.1 must not sort below the 0.2.0 floor.
+        min_version = dify_config.tool.dify.min_difyctl_version
+        if client_version.release[:3] < Version(min_version).release[:3]:
+            return _upgrade_required_response(match.group(1), min_version)
+        return None
@@ -45,8 +45,10 @@ class AccountApi(Resource):
        enforce(LIMIT_ME_PER_ACCOUNT, key=f"account:{auth_data.account_id}")

        account_id_str = str(auth_data.account_id) if auth_data.account_id else None
-        account = AccountService.get_account_by_id(db.session(), account_id_str) if account_id_str else None
-        memberships = TenantService.get_account_memberships(db.session(), account_id_str) if account_id_str else []
+        account = AccountService.get_account_by_id(account_id_str, session=db.session()) if account_id_str else None
+        memberships = (
+            TenantService.get_account_memberships(account_id_str, session=db.session()) if account_id_str else []
+        )
        default_ws_id = _pick_default_workspace(memberships)

        return AccountResponse(
@@ -63,7 +65,7 @@ class AccountSessionsSelfApi(Resource):
    @auth_router.guard(scope=Scope.FULL, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
    @returns(200, RevokeResponse, description="Session revoked")
    def delete(self, *, auth_data: AuthData):
-        revoke_oauth_token(db.session(), redis_client, str(auth_data.token_id))
+        revoke_oauth_token(redis_client, str(auth_data.token_id), session=db.session())
        return RevokeResponse(status="revoked")


@@ -81,7 +83,7 @@ class AccountSessionsApi(Resource):
        page = query.page
        limit = query.limit

-        all_rows = list_active_sessions(db.session(), ctx, now)
+        all_rows = list_active_sessions(ctx, now, session=db.session())

        total = len(all_rows)
        sliced = all_rows[(page - 1) * limit : page * limit]
@@ -117,10 +119,10 @@ class AccountSessionByIdApi(Resource):

        # 404 (not 403) on cross-subject so the endpoint doesn't leak
        # token IDs that belong to other subjects.
-        if not token_belongs_to_subject(db.session(), session_id, ctx):
+        if not token_belongs_to_subject(session_id, ctx, session=db.session()):
            raise NotFound("session not found")

-        revoke_oauth_token(db.session(), redis_client, session_id)
+        revoke_oauth_token(redis_client, session_id, session=db.session())
        return RevokeResponse(status="revoked")


@@ -30,7 +30,7 @@ class AppDslImportApi(Resource):
    a new app.

    Returns 202 when the DSL version requires an explicit confirmation step
-    (major version mismatch).  Callers must then POST to the confirm endpoint.
+    (major version mismatch).  Callers must then POST to the imports :confirm method.
    Returns 400 when the import failed due to invalid DSL or a business error.
    """

@@ -79,7 +79,7 @@ class AppDslImportApi(Resource):
                return result, 200


-@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports/<string:import_id>/confirm")
+@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports/<string:import_id>:confirm")
 class AppDslImportConfirmApi(Resource):
    """Confirm a pending DSL import identified by ``import_id``.

@@ -119,7 +119,7 @@ class AppDslImportConfirmApi(Resource):
        return result, 200


-@openapi_ns.route("/apps/<string:app_id>/export")
+@openapi_ns.route("/apps/<string:app_id>/dsl")
 class AppDslExportApi(Resource):
    """Export an app's current draft configuration as a DSL YAML string.

@@ -154,7 +154,7 @@ class AppDslExportApi(Resource):
        return AppDslExportResponse(data=data), 200


-@openapi_ns.route("/apps/<string:app_id>/check-dependencies")
+@openapi_ns.route("/apps/<string:app_id>/dependencies:check")
 class AppDslCheckDependenciesApi(Resource):
    """Check for leaked plugin dependencies after a DSL import.

@@ -1,4 +1,4 @@
-"""POST /openapi/v1/apps/<app_id>/run — mode-agnostic runner."""
+"""POST /openapi/v1/apps/<app_id>:run — mode-agnostic runner."""

 from __future__ import annotations

@@ -138,7 +138,7 @@ _DISPATCH: dict[AppMode, Callable[[App, Any, AppRunRequest, Session], Any]] = {
 }


-@openapi_ns.route("/apps/<string:app_id>/run")
+@openapi_ns.route("/apps/<string:app_id>:run")
 class AppRunApi(Resource):
    @auth_router.guard(
        scope=Scope.APPS_RUN,
@@ -174,7 +174,7 @@ class AppRunApi(Resource):
        return helper.compact_generate_response(stream_obj)


-@openapi_ns.route("/apps/<string:app_id>/tasks/<string:task_id>/stop")
+@openapi_ns.route("/apps/<string:app_id>/tasks/<string:task_id>:stop")
 class AppRunTaskStopApi(Resource):
    @auth_router.guard(
        scope=Scope.APPS_RUN,
@@ -66,13 +66,13 @@ class AppReadResource(Resource):

        if is_uuid:
            # ``str(parsed_uuid)`` normalises to the canonical dashed form.
-            app = AppService.get_visible_app_by_id(db.session(), str(parsed_uuid))
+            app = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
            if app is None:
                raise NotFound("app not found")
        else:
            if not workspace_id:
                raise UnprocessableEntity("workspace_id is required for name-based lookup")
-            matches = AppService.find_visible_apps_by_name(db.session(), name=app_id, tenant_id=workspace_id)
+            matches = AppService.find_visible_apps_by_name(name=app_id, tenant_id=workspace_id, session=db.session())
            if len(matches) == 0:
                raise NotFound("app not found")
            if len(matches) > 1:
@@ -129,7 +129,7 @@ def build_app_describe_response(app: App, fields: set[str] | None) -> AppDescrib
    return AppDescribeResponse(info=info, parameters=parameters, input_schema=input_schema)


-@openapi_ns.route("/apps/<string:app_id>/describe")
+@openapi_ns.route("/apps/<string:app_id>")
 class AppDescribeApi(AppReadResource):
    @auth_router.guard(
        scope=Scope.APPS_READ,
@@ -177,7 +177,7 @@ class AppListApi(Resource):

        tenant_name: str | None = None
        if parsed_uuid is not None:
-            app: App | None = AppService.get_visible_app_by_id(db.session(), str(parsed_uuid))
+            app: App | None = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
            if app is None or str(app.tenant_id) != workspace_id:
                return empty
            if not _is_listable(app):
@@ -188,7 +188,7 @@ class AppListApi(Resource):
                str(app.id), str(app.maintainer) if app.maintainer else None, str(auth_data.account_id)
            ):
                return empty
-            tenant_name = TenantService.get_tenant_name(db.session(), workspace_id)
+            tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())
            item = AppListRow(
                id=str(app.id),
                name=app.name,
@@ -221,7 +221,7 @@ class AppListApi(Resource):

        tenant_name = None
        if pagination.items:
-            tenant_name = TenantService.get_tenant_name(db.session(), workspace_id)
+            tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())

        items = [
            AppListRow(
@@ -55,10 +55,10 @@ class PermittedExternalAppsListApi(Resource):
            return env

        apps_by_id: dict[str, App] = {
-            str(a.id): a for a in AppService.find_visible_apps_by_ids(db.session(), page_result.app_ids)
+            str(a.id): a for a in AppService.find_visible_apps_by_ids(page_result.app_ids, session=db.session())
        }
        tenant_ids = list({str(a.tenant_id) for a in apps_by_id.values()})
-        tenants_by_id = {str(t.id): t for t in TenantService.get_tenants_by_ids(db.session(), tenant_ids)}
+        tenants_by_id = {str(t.id): t for t in TenantService.get_tenants_by_ids(tenant_ids, session=db.session())}

        items: list[AppListRow] = []
        for app_id in page_result.app_ids:
@@ -87,7 +87,7 @@ class PermittedExternalAppsListApi(Resource):
        return env


-@openapi_ns.route("/permitted-external-apps/<string:app_id>/describe")
+@openapi_ns.route("/permitted-external-apps/<string:app_id>")
 class PermittedExternalAppDescribeApi(Resource):
    @auth_router.guard(
        scope=Scope.APPS_READ_PERMITTED_EXTERNAL,
@@ -23,7 +23,7 @@ def load_app(data: AuthData) -> None:
        uuid.UUID(app_id)
    except ValueError:
        raise NotFound("app not found")
-    app = AppService.get_app_by_id(db.session(), app_id)
+    app = AppService.get_app_by_id(app_id, session=db.session())
    if not app or app.status != AppStatus.NORMAL:
        raise NotFound("app not found")
    data.app = app
@@ -34,7 +34,7 @@ def load_tenant(data: AuthData) -> None:
        return
    if data.app is None:
        raise InternalServerError("pipeline_invariant_violated: app not loaded before load_tenant")
-    tenant = TenantService.get_tenant_by_id(db.session(), str(data.app.tenant_id))
+    tenant = TenantService.get_tenant_by_id(str(data.app.tenant_id), session=db.session())
    if tenant is None or tenant.status == TenantStatus.ARCHIVE:
        raise Forbidden("workspace unavailable")
    data.tenant = tenant
@@ -50,7 +50,7 @@ def load_tenant_from_request(data: AuthData) -> None:
        uuid.UUID(workspace_id)
    except ValueError:
        raise NotFound("workspace not found")
-    tenant = TenantService.get_tenant_by_id(db.session(), workspace_id)
+    tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
    if tenant is None or tenant.status == TenantStatus.ARCHIVE:
        raise NotFound("workspace not found")
    data.tenant = tenant
@@ -59,7 +59,7 @@ def load_tenant_from_request(data: AuthData) -> None:
 def load_account(data: AuthData) -> None:
    if data.caller is not None:
        return
-    account = AccountService.get_account_by_id(db.session(), str(data.account_id))
+    account = AccountService.get_account_by_id(str(data.account_id), session=db.session())
    if account is None:
        raise Unauthorized("account not found")
    if data.tenant:
@@ -75,7 +75,7 @@ def load_workspace_role(data: AuthData) -> None:
        return
    if data.caller is not None and getattr(data.caller, "status", None) != AccountStatus.ACTIVE:
        return
-    role = TenantService.get_account_role_in_tenant(db.session(), str(data.account_id), str(data.tenant.id))
+    role = TenantService.get_account_role_in_tenant(str(data.account_id), str(data.tenant.id), session=db.session())
    if role is None:
        return
    data.tenant_role = role
@@ -82,7 +82,7 @@ def check_app_api_enabled(data: AuthData) -> None:
 def check_app_access(data: AuthData) -> None:
    if data.tenant is None:
        return
-    if not TenantService.account_belongs_to_tenant(db.session(), data.account_id, data.tenant.id):
+    if not TenantService.account_belongs_to_tenant(data.account_id, data.tenant.id, session=db.session()):
        raise Forbidden("subject_no_app_access")


@@ -127,5 +127,5 @@ def _resolve_user_id(data: AuthData) -> str | None:
        return str(data.account_id) if data.account_id is not None else None
    if data.external_identity is None:
        return None
-    account = AccountService.get_account_by_email(db.session(), data.external_identity.email)
+    account = AccountService.get_account_by_email(data.external_identity.email, session=db.session())
    return str(account.id) if account is not None else None
@@ -1,4 +1,4 @@
-"""POST /openapi/v1/apps/<app_id>/files/upload — upload a file for use in app inputs."""
+"""POST /openapi/v1/apps/<app_id>/files — upload a file for use in app inputs."""

 from __future__ import annotations

@@ -26,7 +26,7 @@ from libs.oauth_bearer import Scope
 from services.file_service import FileService


-@openapi_ns.route("/apps/<string:app_id>/files/upload")
+@openapi_ns.route("/apps/<string:app_id>/files")
 class AppFileUploadApi(Resource):
    @openapi_ns.doc("upload_file_for_app_input")
    @openapi_ns.doc(description="Upload a file to use as an input variable when running the app")
@@ -1,8 +1,8 @@
 """
 OpenAPI bearer-authed human input form endpoints.

-GET  /apps/<app_id>/form/human_input/<form_token>  — fetch paused form definition
-POST /apps/<app_id>/form/human_input/<form_token>  — submit form response
+GET  /apps/<app_id>/human-input-forms/<form_token>         — fetch paused form definition
+POST /apps/<app_id>/human-input-forms/<form_token>:submit  — submit form response
 """

 from __future__ import annotations
@@ -60,7 +60,7 @@ def _ensure_form_is_allowed_for_openapi(form) -> None:
        raise RecipientSurfaceMismatch()


-@openapi_ns.route("/apps/<string:app_id>/form/human_input/<string:form_token>")
+@openapi_ns.route("/apps/<string:app_id>/human-input-forms/<string:form_token>")
 class OpenApiWorkflowHumanInputFormApi(Resource):
    @openapi_ns.response(200, "Form definition", openapi_ns.models[HumanInputFormDefinitionResponse.__name__])
    @auth_router.guard(
@@ -79,6 +79,9 @@ class OpenApiWorkflowHumanInputFormApi(Resource):
        service.ensure_form_active(form)
        return _jsonify_form_definition(form)

+
+@openapi_ns.route("/apps/<string:app_id>/human-input-forms/<string:form_token>:submit")
+class OpenApiWorkflowHumanInputFormSubmitApi(Resource):
    @auth_router.guard(
        scope=Scope.APPS_RUN,
        rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_TEST_AND_RUN),
@@ -247,7 +247,6 @@ class DeviceApproveApi(Resource):
                raise BadRequest(description=str(e)) from None
            ttl_days = oauth_ttl_days(tenant_id=tenant)
            mint = mint_oauth_token(
-                db.session(),
                redis_client,
                subject_email=account.email,
                subject_issuer=ACCOUNT_ISSUER_SENTINEL,
@@ -256,6 +255,7 @@ class DeviceApproveApi(Resource):
                device_label=state.device_label,
                prefix=profile.prefix,
                ttl_days=ttl_days,
+                session=db.session(),
            )

            poll_payload = _build_account_poll_payload(account, tenant, mint)
@@ -342,7 +342,7 @@ def _audit_cross_ip_if_needed(state) -> None:


 def _build_account_poll_payload(account, tenant, mint) -> PollPayload:
-    rows = TenantService.get_workspaces_for_account(db.session(), str(account.id))
+    rows = TenantService.get_workspaces_for_account(str(account.id), session=db.session())
    workspaces = [WorkspacePayload(id=str(t.id), name=t.name, role=getattr(m, "role", "")) for t, m in rows]
    # Prefer active session tenant → DB-flagged current join → first membership.
    default_ws_id = None
@@ -194,7 +194,7 @@ def _sso_complete_impl():
    if state.status is not DeviceFlowStatus.PENDING:
        return _device_error_redirect("sso_failed", user_code)

-    if AccountService.has_active_account_with_email(db.session(), claims.email):
+    if AccountService.has_active_account_with_email(claims.email, session=db.session()):
        _emit_external_rejection_audit(
            state,
            _RejectedClaims(subject_email=claims.email, subject_issuer=claims.issuer),
@@ -274,7 +274,7 @@ def approve_external():
    if state.status is not DeviceFlowStatus.PENDING:
        raise Conflict("user_code_not_pending")

-    if AccountService.has_active_account_with_email(db.session(), claims.subject_email):
+    if AccountService.has_active_account_with_email(claims.subject_email, session=db.session()):
        _emit_external_rejection_audit(state, claims, reason="email_belongs_to_dify_account")
        raise Forbidden("email_belongs_to_dify_account")

@@ -293,7 +293,6 @@ def approve_external():

    ttl_days = oauth_ttl_days(tenant_id=None)
    mint = mint_oauth_token(
-        db.session(),
        redis_client,
        subject_email=claims.subject_email,
        subject_issuer=claims.subject_issuer,
@@ -302,6 +301,7 @@ def approve_external():
        device_label=state.device_label,
        prefix=profile.prefix,
        ttl_days=ttl_days,
+        session=db.session(),
    )

    # SSO branch of the shared PollPayload contract: account/workspace
@@ -64,14 +64,14 @@ def _member_response(account: Account) -> MemberResponse:


 def _load_tenant(workspace_id: str) -> Tenant:
-    tenant = TenantService.get_tenant_by_id(db.session(), workspace_id)
+    tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
    if tenant is None or tenant.status != TenantStatus.NORMAL:
        raise NotFound("workspace not found")
    return tenant


 def _load_account(account_id: object) -> Account:
-    account = AccountService.get_account_by_id(db.session(), str(account_id)) if account_id else None
+    account = AccountService.get_account_by_id(str(account_id), session=db.session()) if account_id else None
    if account is None:
        raise RuntimeError("authenticated account_id has no Account row")
    return account
@@ -94,7 +94,7 @@ class WorkspacesApi(Resource):
    @auth_router.guard(scope=Scope.WORKSPACE_READ, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
    @returns(200, WorkspaceListResponse, description="Workspace list")
    def get(self, *, auth_data: AuthData):
-        rows = TenantService.get_workspaces_for_account(db.session(), str(auth_data.account_id))
+        rows = TenantService.get_workspaces_for_account(str(auth_data.account_id), session=db.session())

        return WorkspaceListResponse(workspaces=list(starmap(_workspace_summary, rows)))

@@ -104,7 +104,7 @@ class WorkspaceByIdApi(Resource):
    @auth_router.guard(scope=Scope.WORKSPACE_READ, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
    @returns(200, WorkspaceDetailResponse, description="Workspace detail")
    def get(self, workspace_id: str, *, auth_data: AuthData):
-        row = TenantService.find_workspace_for_account(db.session(), str(auth_data.account_id), workspace_id)
+        row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
        # 404 (not 403) on non-member so workspace IDs don't leak across tenants.
        if row is None:
            raise NotFound("workspace not found")
@@ -113,7 +113,7 @@ class WorkspaceByIdApi(Resource):
        return _workspace_detail(tenant, membership)


-@openapi_ns.route("/workspaces/<string:workspace_id>/switch")
+@openapi_ns.route("/workspaces/<string:workspace_id>:switch")
 class WorkspaceSwitchApi(Resource):
    """Server-side switch — equivalent to the console's POST /workspaces/switch.

@@ -132,7 +132,7 @@ class WorkspaceSwitchApi(Resource):
        except AccountNotLinkTenantError:
            raise NotFound("workspace not found")

-        row = TenantService.find_workspace_for_account(db.session(), str(auth_data.account_id), workspace_id)
+        row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
        if row is None:
            raise NotFound("workspace not found")
        tenant, membership = row
@@ -194,7 +194,7 @@ class WorkspaceMembersApi(Resource):
            raise BadRequest(str(exc))

        normalized_email = body.email.lower()
-        member = AccountService.get_account_by_email_with_case_fallback(db.session(), normalized_email)
+        member = AccountService.get_account_by_email_with_case_fallback(normalized_email, session=db.session())
        if member is None:
            # invite_new_member just created or fetched this account.
            raise RuntimeError("invited member missing from DB after invite")
@@ -212,11 +212,12 @@ class WorkspaceMembersApi(Resource):

@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>")
 class WorkspaceMemberApi(Resource):
-    """Remove a member.
+    """Remove a member (DELETE) or change a member's role (PATCH).

    Self-removal and owner-removal are explicitly rejected by the service
    layer (CannotOperateSelfError, NoPermissionError) — both surface as
-    400 per the spec, with the service's message preserved.
+    400 per the spec, with the service's message preserved. Owner can never be
+    assigned via PATCH (closed enum); admin cannot demote the standing owner.
    """

    @auth_router.guard_workspace(
@@ -228,7 +229,7 @@ class WorkspaceMemberApi(Resource):
    def delete(self, workspace_id: str, member_id: str, *, auth_data: AuthData):
        operator = _load_account(auth_data.account_id)
        tenant = _load_tenant(workspace_id)
-        member = AccountService.get_account_by_id(db.session(), member_id)
+        member = AccountService.get_account_by_id(member_id, session=db.session())
        if member is None:
            raise NotFound("member not found")

@@ -243,15 +244,6 @@ class WorkspaceMemberApi(Resource):

        return MemberActionResponse()

-
-@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>/role")
-class WorkspaceMemberRoleApi(Resource):
-    """Change a member's role.
-
-    Owner cannot be assigned here (closed enum). Admin cannot demote the
-    standing owner (service NoPermissionError → 400, per spec).
-    """
-
    @auth_router.guard_workspace(
        scope=Scope.WORKSPACE_WRITE,
        allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
@@ -259,10 +251,10 @@ class WorkspaceMemberRoleApi(Resource):
    )
    @returns(200, MemberActionResponse, description="Role updated")
    @accepts(body=MemberRoleUpdatePayload)
-    def put(self, workspace_id: str, member_id: str, *, auth_data: AuthData, body: MemberRoleUpdatePayload):
+    def patch(self, workspace_id: str, member_id: str, *, auth_data: AuthData, body: MemberRoleUpdatePayload):
        operator = _load_account(auth_data.account_id)
        tenant = _load_tenant(workspace_id)
-        member = AccountService.get_account_by_id(db.session(), member_id)
+        member = AccountService.get_account_by_id(member_id, session=db.session())
        if member is None:
            raise NotFound("member not found")

@@ -12,8 +12,13 @@ from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_app_token
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from fields.annotation_fields import Annotation, AnnotationList
-from fields.base import ResponseModel
+from fields.annotation_fields import (
+    Annotation,
+    AnnotationJobStatusDetailResponse,
+    AnnotationJobStatusResponse,
+    AnnotationList,
+)
+from libs.helper import dump_response
 from models.model import App
 from services.annotation_service import (
    AppAnnotationService,
@@ -46,12 +51,6 @@ class AnnotationListQuery(BaseModel):
    keyword: str = Field(default="", description="Keyword to filter annotations by question or answer content.")


-class AnnotationJobStatusResponse(ResponseModel):
-    job_id: str
-    job_status: str
-    error_msg: str | None = None
-
-
 ANNOTATION_REPLY_ACTION_PARAM = {
    "description": "Action to perform: `enable` or `disable`.",
    "enum": ["enable", "disable"],
@@ -67,7 +66,13 @@ register_schema_models(
    Annotation,
    AnnotationList,
 )
-register_response_schema_models(service_api_ns, AnnotationJobStatusResponse)
+register_response_schema_models(
+    service_api_ns,
+    Annotation,
+    AnnotationList,
+    AnnotationJobStatusResponse,
+    AnnotationJobStatusDetailResponse,
+)


@service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -113,7 +118,7 @@ class AnnotationReplyActionApi(Resource):
                result = AppAnnotationService.enable_app_annotation(enable_args, app_model.id)
            case "disable":
                result = AppAnnotationService.disable_app_annotation(app_model.id)
-        return result, 200
+        return dump_response(AnnotationJobStatusResponse, result), 200


@service_api_ns.route("/apps/annotation-reply/<string:action>/status/<uuid:job_id>")
@@ -151,7 +156,7 @@ class AnnotationReplyActionStatusApi(Resource):
    @service_api_ns.response(
        200,
        "Job status retrieved successfully",
-        service_api_ns.models[AnnotationJobStatusResponse.__name__],
+        service_api_ns.models[AnnotationJobStatusDetailResponse.__name__],
    )
    @validate_app_token
    def get(self, app_model: App, job_id: UUID, action: str):
@@ -166,9 +171,13 @@ class AnnotationReplyActionStatusApi(Resource):
        error_msg = ""
        if job_status == "error":
            app_annotation_error_key = f"{action}_app_annotation_error_{job_id_str}"
-            error_msg = redis_client.get(app_annotation_error_key).decode()
+            error_result = redis_client.get(app_annotation_error_key)
+            if error_result is not None:
+                error_msg = error_result.decode()

-        return {"job_id": job_id_str, "job_status": job_status, "error_msg": error_msg}, 200
+        return AnnotationJobStatusDetailResponse(
+            job_id=job_id_str, job_status=job_status, error_msg=error_msg
+        ).model_dump(mode="json"), 200


@service_api_ns.route("/apps/annotations")
@@ -204,14 +213,13 @@ class AnnotationListApi(Resource):
            app_model.id, query.page, query.limit, query.keyword, session=db.session()
        )
        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
-        response = AnnotationList(
+        return AnnotationList(
            data=annotation_models,
            has_more=len(annotation_list) == query.limit,
            limit=query.limit,
            total=total,
            page=query.page,
-        )
-        return response.model_dump(mode="json")
+        ).model_dump(mode="json")

    @service_api_ns.doc(
        summary="Create Annotation",
@@ -246,8 +254,7 @@ class AnnotationListApi(Resource):
        annotation = AppAnnotationService.insert_app_annotation_directly(
            insert_args, app_model.id, session=db.session()
        )
-        response = Annotation.model_validate(annotation, from_attributes=True)
-        return response.model_dump(mode="json"), HTTPStatus.CREATED
+        return dump_response(Annotation, annotation), HTTPStatus.CREATED


@service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -288,8 +295,7 @@ class AnnotationUpdateDeleteApi(Resource):
        app_ref = AppRefService.create_app_ref(app_model)
        annotation_ref = AppRefService.create_annotation_ref(app_ref, str(annotation_id))
        annotation = AppAnnotationService.update_app_annotation_directly(update_args, annotation_ref, db.session())
-        response = Annotation.model_validate(annotation, from_attributes=True)
-        return response.model_dump(mode="json")
+        return dump_response(Annotation, annotation)

    @service_api_ns.doc(
        summary="Delete Annotation",
@@ -25,6 +25,7 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
+from libs.helper import dump_response
 from models.model import App, EndUser
 from services.app_ref_service import AppRefService
 from services.audio_service import AudioService
@@ -102,7 +103,7 @@ class AudioApi(Resource):
        try:
            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.id)

-            return response
+            return dump_response(AudioTranscriptResponse, response)
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logger.exception("App model config broken.")
            raise AppUnavailableError()
@@ -165,6 +166,7 @@ class TextApi(Resource):
            500: "Internal server error",
        }
    )
+    # TTS returns provider audio bytes, so the success response is intentionally schema-less.
    @service_api_ns.response(200, "Text successfully converted to audio")
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
    def post(self, app_model: App, end_user: EndUser):
@@ -186,7 +188,7 @@ class TextApi(Resource):
                    message_id,
                    end_user_id=end_user.id,
                )
-            response = AudioService.transcript_tts(
+            return AudioService.transcript_tts(
                app_model=app_model,
                session=db.session(),
                text=text,
@@ -194,8 +196,6 @@ class TextApi(Resource):
                end_user=end_user.external_user_id,
                message_ref=message_ref,
            )
-
-            return response
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logger.exception("App model config broken.")
            raise AppUnavailableError()
@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound

 import services
-from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
+from controllers.common.fields import SimpleResultResponse
 from controllers.common.schema import register_response_schema_models, register_schema_models
 from controllers.console.app.wraps import with_session
 from controllers.service_api import service_api_ns
@@ -40,12 +40,14 @@ from core.errors.error import (
    QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id, get_trace_session_id, omit_trace_session_id_from_payload
+from extensions.ext_database import db
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
+from services.conversation_service import ConversationService
 from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError

@@ -158,7 +160,7 @@ class ChatRequestPayload(BaseModel):


 register_schema_models(service_api_ns, CompletionRequestPayload, ChatRequestPayload)
-register_response_schema_models(service_api_ns, GeneratedAppResponse, SimpleResultResponse)
+register_response_schema_models(service_api_ns, SimpleResultResponse)


@service_api_ns.route("/completion-messages")
@@ -201,11 +203,7 @@ class CompletionApi(Resource):
            500: "Internal server error",
        }
    )
-    @service_api_ns.response(
-        200,
-        "Completion created successfully",
-        service_api_ns.models[GeneratedAppResponse.__name__],
-    )
+    @service_api_ns.response(200, "Completion created successfully")
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
    @with_session
    def post(self, session: Session, app_model: App, end_user: EndUser):
@@ -242,6 +240,7 @@ class CompletionApi(Resource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")
@@ -304,7 +303,7 @@ class CompletionStopApi(Resource):
            app_mode=AppMode.value_of(app_model.mode),
        )

-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200


@service_api_ns.route("/chat-messages")
@@ -354,11 +353,7 @@ class ChatApi(Resource):
            500: "Internal server error",
        }
    )
-    @service_api_ns.response(
-        200,
-        "Message sent successfully",
-        service_api_ns.models[GeneratedAppResponse.__name__],
-    )
+    @service_api_ns.response(200, "Message sent successfully")
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
    @with_session
    def post(self, session: Session, app_model: App, end_user: EndUser):
@@ -384,6 +379,15 @@ class ChatApi(Resource):
        streaming = _resolve_agent_app_streaming(app_mode=app_mode, response_mode=payload.response_mode)

        try:
+            # Eagerly validate conversation to avoid hanging on invalid conversation_id
+            if payload.conversation_id:
+                ConversationService.get_conversation(
+                    app_model=app_model,
+                    conversation_id=payload.conversation_id,
+                    user=end_user,
+                    session=db.session(),
+                )
+
            response = AppGenerateService.generate(
                session=session,
                app_model=app_model,
@@ -393,6 +397,7 @@ class ChatApi(Resource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except WorkflowNotFoundError as ex:
            raise NotFound(str(ex))
@@ -464,4 +469,4 @@ class ChatStopApi(Resource):
            app_mode=app_mode,
        )

-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@@ -24,7 +24,7 @@ from fields.conversation_fields import (
    SimpleConversation,
 )
 from graphon.variables.types import SegmentType
-from libs.helper import UUIDStrOrEmpty, to_timestamp
+from libs.helper import UUIDStrOrEmpty, dump_response, to_timestamp
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService

@@ -142,15 +142,13 @@ register_schema_models(
    ConversationRenamePayload,
    ConversationVariablesQuery,
    ConversationVariableUpdatePayload,
-    ConversationVariableResponse,
-    ConversationVariableInfiniteScrollPaginationResponse,
 )
 register_response_schema_models(
    service_api_ns,
-    ConversationInfiniteScrollPagination,
-    SimpleConversation,
    ConversationVariableResponse,
    ConversationVariableInfiniteScrollPaginationResponse,
+    ConversationInfiniteScrollPagination,
+    SimpleConversation,
 )


@@ -166,9 +164,9 @@ class ConversationApi(Resource):
            404: "`not_found` : Last conversation does not exist (invalid `last_id`).",
        },
    )
-    @service_api_ns.doc(params=query_params_from_model(ConversationListQuery))
    @service_api_ns.doc("list_conversations")
    @service_api_ns.doc(description="List all conversations for the current user")
+    @service_api_ns.doc(params=query_params_from_model(ConversationListQuery))
    @service_api_ns.doc(
        responses={
            200: "Conversations retrieved successfully",
@@ -192,7 +190,7 @@ class ConversationApi(Resource):
            raise NotChatAppError()

        query_args = ConversationListQuery.model_validate(request.args.to_dict())
-        last_id = str(query_args.last_id) if query_args.last_id else None
+        last_id = query_args.last_id or None

        try:
            with sessionmaker(db.engine).begin() as session:
@@ -208,9 +206,7 @@ class ConversationApi(Resource):
                adapter = TypeAdapter(SimpleConversation)
                conversations = [adapter.validate_python(item, from_attributes=True) for item in pagination.data]
                return ConversationInfiniteScrollPagination(
-                    limit=pagination.limit,
-                    has_more=pagination.has_more,
-                    data=conversations,
+                    limit=pagination.limit, has_more=pagination.has_more, data=conversations
                ).model_dump(mode="json")
        except services.errors.conversation.LastConversationNotExistsError:
            raise NotFound("Last Conversation Not Exists.")
@@ -301,11 +297,7 @@ class ConversationRenameApi(Resource):
            conversation = ConversationService.rename(
                app_model, conversation_id, end_user, payload.name, payload.auto_generate, session=db.session()
            )
-            return (
-                TypeAdapter(SimpleConversation)
-                .validate_python(conversation, from_attributes=True)
-                .model_dump(mode="json")
-            )
+            return dump_response(SimpleConversation, conversation)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")

@@ -322,10 +314,9 @@ class ConversationVariablesApi(Resource):
            404: "`not_found` : Conversation does not exist.",
        },
    )
-    @service_api_ns.doc(params=query_params_from_model(ConversationVariablesQuery))
    @service_api_ns.doc("list_conversation_variables")
    @service_api_ns.doc(description="List all variables for a conversation")
-    @service_api_ns.doc(params={"c_id": "Conversation ID."})
+    @service_api_ns.doc(params={"c_id": "Conversation ID.", **query_params_from_model(ConversationVariablesQuery)})
    @service_api_ns.doc(
        responses={
            200: "Variables retrieved successfully",
@@ -352,7 +343,7 @@ class ConversationVariablesApi(Resource):
        conversation_id = str(c_id)

        query_args = ConversationVariablesQuery.model_validate(request.args.to_dict())
-        last_id = str(query_args.last_id) if query_args.last_id else None
+        last_id = query_args.last_id or None

        try:
            pagination = ConversationService.get_conversational_variable(
@@ -364,9 +355,7 @@ class ConversationVariablesApi(Resource):
                query_args.variable_name,
                session=db.session(),
            )
-            return ConversationVariableInfiniteScrollPaginationResponse.model_validate(
-                pagination, from_attributes=True
-            ).model_dump(mode="json")
+            return dump_response(ConversationVariableInfiniteScrollPaginationResponse, pagination)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")

@@ -425,7 +414,7 @@ class ConversationVariableDetailApi(Resource):
            variable = ConversationService.update_conversation_variable(
                app_model, conversation_id, variable_id_str, end_user, payload.value, session=db.session()
            )
-            return ConversationVariableResponse.model_validate(variable, from_attributes=True).model_dump(mode="json")
+            return dump_response(ConversationVariableResponse, variable)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")
        except services.errors.conversation.ConversationVariableNotExistsError:
@@ -31,6 +31,12 @@ class NotWorkflowAppError(BaseHTTPException):
    code = 400


+class WorkflowVersionExecutionNotAllowedError(BaseHTTPException):
+    error_code = "workflow_version_execution_not_allowed"
+    description = "Workflow version execution is not available on your current plan. Please upgrade to a paid plan."
+    code = 403
+
+
 class ConversationCompletedError(BaseHTTPException):
    error_code = "conversation_completed"
    description = "The conversation has ended. Please start a new conversation."
@@ -16,6 +16,7 @@ from controllers.service_api.schema import multipart_file_params
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from extensions.ext_database import db
 from fields.file_fields import FileResponse
+from libs.helper import dump_response
 from models import App, EndUser
 from services.file_service import FileService

@@ -87,5 +88,4 @@ class FileApi(Resource):
        except services.errors.file.UnsupportedFileTypeError:
            raise UnsupportedFileTypeError()

-        response = FileResponse.model_validate(upload_file, from_attributes=True)
-        return response.model_dump(mode="json"), 201
+        return dump_response(FileResponse, upload_file), 201
@@ -60,12 +60,14 @@ register_response_schema_models(
    ResultResponse,
    SimpleResultStringListResponse,
    MessageInfiniteScrollPagination,
+    MessageListItem,
    AppFeedbackListResponse,
 )


@service_api_ns.route("/messages")
 class MessageListApi(Resource):
+    @service_api_ns.doc("list_messages")
    @service_api_ns.doc(
        summary="List Conversation Messages",
        description=(
@@ -76,15 +78,15 @@ class MessageListApi(Resource):
        responses={
            200: "Successfully retrieved conversation history.",
            400: "`not_chat_app` : App mode does not match the API route.",
-            404: ("- `not_found` : Conversation does not exist.\n- `not_found` : First message does not exist."),
+            404: "- `not_found` : Conversation does not exist.\n- `not_found` : First message does not exist.",
        },
    )
    @service_api_ns.doc(params=query_params_from_model(MessageListQuery))
-    @service_api_ns.doc("list_messages")
    @service_api_ns.doc(description="List messages in a conversation")
    @service_api_ns.doc(
        responses={
            200: "Messages retrieved successfully",
+            400: "`not_chat_app` : App mode does not match the API route.",
            401: "Unauthorized - invalid API token",
            404: "Conversation or first message not found",
        }
@@ -105,8 +107,8 @@ class MessageListApi(Resource):
            raise NotChatAppError()

        query_args = MessageListQuery.model_validate(request.args.to_dict())
-        conversation_id = str(query_args.conversation_id)
-        first_id = str(query_args.first_id) if query_args.first_id else None
+        conversation_id = query_args.conversation_id
+        first_id = query_args.first_id or None

        try:
            pagination = MessageService.pagination_by_first_id(
@@ -115,9 +117,7 @@ class MessageListApi(Resource):
            adapter = TypeAdapter(MessageListItem)
            items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
            return MessageInfiniteScrollPagination(
-                limit=pagination.limit,
-                has_more=pagination.has_more,
-                data=items,
+                limit=pagination.limit, has_more=pagination.has_more, data=items
            ).model_dump(mode="json")
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")
@@ -127,21 +127,20 @@ class MessageListApi(Resource):

@service_api_ns.route("/messages/<uuid:message_id>/feedbacks")
 class MessageFeedbackApi(Resource):
+    @expect_with_user(service_api_ns, MessageFeedbackPayload)
+    @service_api_ns.response(200, "Feedback submitted successfully", service_api_ns.models[ResultResponse.__name__])
+    @service_api_ns.doc("create_message_feedback")
    @service_api_ns.doc(
        summary="Submit Message Feedback",
        description=(
            "Submit feedback for a message. End users can rate messages as `like` or `dislike`, and "
-            "optionally provide text feedback. Pass `null` for `rating` to revoke previously submitted "
-            "feedback."
+            "optionally provide text feedback. Pass `null` for `rating` to revoke previously submitted feedback."
        ),
        tags=["Feedback"],
        responses={
            404: "`not_found` : Message does not exist.",
        },
    )
-    @expect_with_user(service_api_ns, MessageFeedbackPayload)
-    @service_api_ns.response(200, "Feedback submitted successfully", service_api_ns.models[ResultResponse.__name__])
-    @service_api_ns.doc("create_message_feedback")
    @service_api_ns.doc(description="Submit feedback for a message")
    @service_api_ns.doc(params={"message_id": "Message ID."})
    @service_api_ns.doc(
@@ -178,11 +177,12 @@ class MessageFeedbackApi(Resource):

@service_api_ns.route("/app/feedbacks")
 class AppGetFeedbacksApi(Resource):
+    @service_api_ns.doc("get_app_feedbacks")
    @service_api_ns.doc(
        summary="List App Feedbacks",
        description=(
-            "Retrieve a paginated list of all feedback submitted for messages in this application, "
-            "including both end-user and admin feedback."
+            "Retrieve a paginated list of all feedback submitted for messages in this application, including both "
+            "end-user and admin feedback."
        ),
        tags=["Feedback"],
        responses={
@@ -190,7 +190,6 @@ class AppGetFeedbacksApi(Resource):
        },
    )
    @service_api_ns.doc(params=query_params_from_model(FeedbackListQuery))
-    @service_api_ns.doc("get_app_feedbacks")
    @service_api_ns.doc(description="Get all feedbacks for the application")
    @service_api_ns.doc(
        responses={
@@ -213,11 +212,12 @@ class AppGetFeedbacksApi(Resource):
        feedbacks = MessageService.get_all_messages_feedbacks(
            app_model, page=query_args.page, limit=query_args.limit, session=db.session()
        )
-        return {"data": feedbacks}
+        return AppFeedbackListResponse(data=feedbacks).model_dump(mode="json")


@service_api_ns.route("/messages/<uuid:message_id>/suggested")
 class MessageSuggestedApi(Resource):
+    @service_api_ns.doc("get_suggested_questions")
    @service_api_ns.doc(
        summary="Get Next Suggested Questions",
        description="Get next questions suggestions for the current message.",
@@ -237,7 +237,6 @@ class MessageSuggestedApi(Resource):
        "Suggested questions retrieved successfully",
        service_api_ns.models[SimpleResultStringListResponse.__name__],
    )
-    @service_api_ns.doc("get_suggested_questions")
    @service_api_ns.doc(description="Get suggested follow-up questions for a message")
    @service_api_ns.doc(params={"message_id": "Message ID"})
    @service_api_ns.doc(
@@ -276,4 +275,4 @@ class MessageSuggestedApi(Resource):
            logger.exception("internal server error.")
            raise InternalServerError()

-        return {"result": "success", "data": questions}
+        return SimpleResultStringListResponse(result="success", data=questions).model_dump(mode="json")
@@ -1,19 +1,25 @@
 import logging
 from collections.abc import Mapping
 from datetime import datetime
-from typing import Literal, override
+from typing import Literal

 from dateutil.parser import isoparse
 from flask import request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource
+from pydantic import BaseModel, Field, field_validator, model_validator
 from pydantic.json_schema import SkipJsonSchema
 from sqlalchemy.orm import Session, sessionmaker
 from werkzeug.exceptions import BadRequest, InternalServerError, NotFound

+from configs import dify_config
 from controllers.common.controller_schemas import WorkflowRunPayload as WorkflowRunPayloadBase
 from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
-from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
+from controllers.common.schema import (
+    query_params_from_model,
+    query_params_from_request,
+    register_response_schema_models,
+    register_schema_models,
+)
 from controllers.console.app.wraps import with_session
 from controllers.service_api import service_api_ns
 from controllers.service_api.app.error import (
@@ -22,6 +28,7 @@ from controllers.service_api.app.error import (
    ProviderModelCurrentlyNotSupportError,
    ProviderNotInitializeError,
    ProviderQuotaExceededError,
+    WorkflowVersionExecutionNotAllowedError,
 )
 from controllers.service_api.schema import (
    expect_user_json,
@@ -38,20 +45,21 @@ from core.errors.error import (
    QuotaExceededError,
 )
 from core.helper.trace_id_helper import get_external_trace_id, get_trace_session_id, omit_trace_session_id_from_payload
+from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.base import ResponseModel
 from fields.end_user_fields import SimpleEndUser
-from fields.member_fields import SimpleAccount
+from fields.member_fields import SimpleAccountResponse
 from graphon.enums import WorkflowExecutionStatus
 from graphon.graph_engine.manager import GraphEngineManager
 from graphon.model_runtime.errors.invoke import InvokeError
 from libs import helper
-from libs.helper import to_timestamp
+from libs.helper import dump_response, to_timestamp
 from models.model import App, AppMode, EndUser
-from models.workflow import WorkflowRun
 from repositories.factory import DifyAPIRepositoryFactory
 from services.app_generate_service import AppGenerateService
+from services.billing_service import BillingService
 from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_app_service import WorkflowAppService
@@ -105,29 +113,12 @@ def _enum_value(value):
    return getattr(value, "value", value)


-class WorkflowRunStatusField(fields.Raw):
-    @override
-    def output(self, key, obj: WorkflowRun, **kwargs):
-        return _enum_value(obj.status)
-
-
-class WorkflowRunOutputsField(fields.Raw):
-    @override
-    def output(self, key, obj: WorkflowRun, **kwargs):
-        status = _enum_value(obj.status)
-        if status == WorkflowExecutionStatus.PAUSED.value:
-            return {}
-
-        outputs = obj.outputs_dict
-        return outputs or {}
-
-
 class WorkflowRunResponse(ResponseModel):
    id: str
    workflow_id: str
    status: str
    inputs: dict | list | str | int | float | bool | None = Field(default=None)
-    outputs: dict = Field(default_factory=dict)
+    outputs: dict = Field(default_factory=dict, validation_alias="outputs_dict")
    error: str | None = None
    total_steps: int | None = None
    total_tokens: int | None = None
@@ -135,11 +126,33 @@ class WorkflowRunResponse(ResponseModel):
    finished_at: int | None = None
    elapsed_time: float | int | None = None

+    @field_validator("status", mode="before")
+    @classmethod
+    def _normalize_enum(cls, value):
+        return _enum_value(value)
+
+    @field_validator("outputs", mode="before")
+    @classmethod
+    def _normalize_outputs(cls, value):
+        if value is None:
+            return {}
+        if isinstance(value, dict):
+            return value
+        if isinstance(value, Mapping):
+            return dict(value)
+        return {}
+
    @field_validator("created_at", "finished_at", mode="before")
    @classmethod
    def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
        return to_timestamp(value)

+    @model_validator(mode="after")
+    def _clear_paused_outputs(self):
+        if self.status == WorkflowExecutionStatus.PAUSED.value:
+            self.outputs = {}
+        return self
+

 class WorkflowRunForLogResponse(ResponseModel):
    id: str
@@ -171,7 +184,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
    details: dict | list | str | int | float | bool | None = Field(default=None)
    created_from: str | None = None
    created_by_role: str | None = None
-    created_by_account: SimpleAccount | None = None
+    created_by_account: SimpleAccountResponse | None = None
    created_by_end_user: SimpleEndUser | None = None
    created_at: int | None = None

@@ -203,39 +216,6 @@ register_response_schema_models(
 )


-def _serialize_workflow_run(workflow_run: WorkflowRun) -> dict:
-    status = _enum_value(workflow_run.status)
-    raw_outputs = workflow_run.outputs_dict
-    match raw_outputs:
-        case _ if status == WorkflowExecutionStatus.PAUSED.value or raw_outputs is None:
-            outputs: dict = {}
-        case dict():
-            outputs = raw_outputs
-        case _ if isinstance(raw_outputs, Mapping):
-            outputs = dict(raw_outputs)
-        case _:
-            outputs = {}
-    return WorkflowRunResponse.model_validate(
-        {
-            "id": workflow_run.id,
-            "workflow_id": workflow_run.workflow_id,
-            "status": status,
-            "inputs": workflow_run.inputs,
-            "outputs": outputs,
-            "error": workflow_run.error,
-            "total_steps": workflow_run.total_steps,
-            "total_tokens": workflow_run.total_tokens,
-            "created_at": workflow_run.created_at,
-            "finished_at": workflow_run.finished_at,
-            "elapsed_time": workflow_run.elapsed_time,
-        }
-    ).model_dump(mode="json")
-
-
-def _serialize_workflow_log_pagination(pagination) -> dict:
-    return WorkflowAppLogPaginationResponse.model_validate(pagination, from_attributes=True).model_dump(mode="json")
-
-
@service_api_ns.route("/workflows/run/<string:workflow_run_id>")
 class WorkflowRunDetailApi(Resource):
    @service_api_ns.doc(
@@ -288,7 +268,7 @@ class WorkflowRunDetailApi(Resource):
        )
        if not workflow_run:
            raise NotFound("Workflow run not found.")
-        return _serialize_workflow_run(workflow_run)
+        return dump_response(WorkflowRunResponse, workflow_run)


@service_api_ns.route("/workflows/run")
@@ -373,6 +353,7 @@ class WorkflowRunApi(Resource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except ProviderTokenNotInitError as ex:
            raise ProviderNotInitializeError(ex.description)
@@ -419,6 +400,10 @@ class WorkflowRunByIdApi(Resource):
                "- `completion_request_error` : Workflow execution request failed.\n"
                "- `invalid_param` : Required parameter missing or invalid."
            ),
+            403: (
+                "`workflow_version_execution_not_allowed` : Workflow version execution is unavailable on the "
+                "current plan. Upgrade to a paid plan."
+            ),
            404: "`not_found` : Workflow not found.",
            429: (
                "- `too_many_requests` : Too many concurrent requests for this app.\n"
@@ -444,6 +429,7 @@ class WorkflowRunByIdApi(Resource):
            200: "Workflow executed successfully",
            400: "Bad request - invalid parameters or workflow issues",
            401: "Unauthorized - invalid API token",
+            403: "Forbidden - upgrade to a paid plan to execute a specific workflow version",
            404: "Workflow not found",
            429: "Rate limit exceeded",
            500: "Internal server error",
@@ -465,6 +451,11 @@ class WorkflowRunByIdApi(Resource):
        if app_mode != AppMode.WORKFLOW:
            raise NotWorkflowAppError()

+        if dify_config.BILLING_ENABLED:
+            billing_info = BillingService.get_info(app_model.tenant_id, exclude_vector_space=True)
+            if billing_info["enabled"] and billing_info["subscription"]["plan"] == CloudPlan.SANDBOX:
+                raise WorkflowVersionExecutionNotAllowedError()
+
        payload = WorkflowRunPayload.model_validate(omit_trace_session_id_from_payload(service_api_ns.payload) or {})
        args = payload.model_dump(exclude_none=True)
        trace_session_id = get_trace_session_id(request)
@@ -489,6 +480,7 @@ class WorkflowRunByIdApi(Resource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except WorkflowNotFoundError as ex:
            raise NotFound(str(ex))
@@ -554,7 +546,7 @@ class WorkflowTaskStopApi(Resource):
        # New graph engine command channel mechanism
        GraphEngineManager(redis_client).send_stop_command(task_id)

-        return {"result": "success"}
+        return SimpleResultResponse(result="success").model_dump()


@service_api_ns.route("/workflows/logs")
@@ -587,7 +579,7 @@ class WorkflowAppLogApi(Resource):

        Returns paginated workflow execution logs with filtering options.
        """
-        args = WorkflowLogQuery.model_validate(request.args.to_dict())
+        args = query_params_from_request(WorkflowLogQuery)

        status = WorkflowExecutionStatus(args.status) if args.status else None
        created_at_before = isoparse(args.created_at__before) if args.created_at__before else None
@@ -609,4 +601,4 @@ class WorkflowAppLogApi(Resource):
                created_by_account=args.created_by_account,
            )

-            return _serialize_workflow_log_pagination(workflow_app_log_pagination)
+            return dump_response(WorkflowAppLogPaginationResponse, workflow_app_log_pagination)
@@ -681,10 +681,10 @@ class DatasetApi(DatasetApiResource):
                dataset,
                str(payload.permission) if payload.permission else None,
                payload.partial_member_list,
-                db.session(),
+                session=db.session(),
            )

-        dataset = DatasetService.update_dataset(session, dataset_id_str, update_data, current_user)
+        dataset = DatasetService.update_dataset(dataset_id_str, update_data, current_user, session=session)

        if dataset is None:
            raise NotFound("Dataset not found.")
@@ -845,7 +845,7 @@ class DocumentStatusApi(DatasetApiResource):
        except ValueError as e:
            raise InvalidActionError(str(e))

-        return dump_response(SimpleResultResponse, {"result": "success"}), 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200


@service_api_ns.route("/datasets/tags")
@@ -876,7 +876,7 @@ class DatasetTagsApi(DatasetApiResource):
        assert isinstance(current_user, Account)
        cid = current_user.current_tenant_id
        assert cid is not None
-        tags = TagService.get_tags(db.session(), "knowledge", cid)
+        tags = TagService.get_tags("knowledge", cid, session=db.session())
        return dump_response(KnowledgeTagListResponse, tags), 200

    @service_api_ns.doc(
@@ -911,11 +911,8 @@ class DatasetTagsApi(DatasetApiResource):
        payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
        tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=TagType.KNOWLEDGE), db.session())

-        response = dump_response(
-            KnowledgeTagResponse,
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0},
-        )
-        return response, 200
+        response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count="0")
+        return response.model_dump(mode="json"), 200

    @service_api_ns.doc(
        summary="Update Knowledge Tag",
@@ -953,11 +950,8 @@ class DatasetTagsApi(DatasetApiResource):

        binding_count = TagService.get_tag_binding_count(tag_id, db.session(), tag_type=TagType.KNOWLEDGE)

-        response = dump_response(
-            KnowledgeTagResponse,
-            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
-        )
-        return response, 200
+        response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=str(binding_count))
+        return response.model_dump(mode="json"), 200

    @service_api_ns.doc(
        summary="Delete Knowledge Tag",
@@ -1088,5 +1082,8 @@ class DatasetTagsBindingStatusApi(DatasetApiResource):
        tags = TagService.get_tags_by_target_id(
            "knowledge", current_user.current_tenant_id, str(dataset_id), db.session()
        )
-        tags_list = [{"id": tag.id, "name": tag.name} for tag in tags]
-        return dump_response(DatasetBoundTagListResponse, {"data": tags_list, "total": len(tags)}), 200
+        response = DatasetBoundTagListResponse(
+            data=[DatasetBoundTagResponse(id=tag.id, name=tag.name) for tag in tags],
+            total=len(tags),
+        )
+        return response.model_dump(mode="json"), 200
@@ -6,14 +6,22 @@ deprecated in generated API docs so clients migrate toward the canonical paths.
 """

 import json
-from collections.abc import Mapping
 from contextlib import ExitStack
 from copy import deepcopy
 from typing import Annotated, Any, Literal, Self, override
 from uuid import UUID

 from flask import request, send_file
-from pydantic import BaseModel, Field, GetJsonSchemaHandler, WithJsonSchema, field_validator, model_validator
+from pydantic import (
+    BaseModel,
+    Field,
+    GetJsonSchemaHandler,
+    ValidationError,
+    WithJsonSchema,
+    field_validator,
+    model_validator,
+)
+from pydantic.json_schema import SkipJsonSchema
 from sqlalchemy import desc, func, select
 from werkzeug.exceptions import Forbidden, NotFound

@@ -26,9 +34,10 @@ from controllers.common.errors import (
    TooManyFilesError,
    UnsupportedFileTypeError,
 )
-from controllers.common.fields import BinaryFileResponse, UrlResponse
+from controllers.common.fields import UrlResponse
 from controllers.common.schema import (
    query_params_from_model,
+    query_params_from_request,
    register_enum_models,
    register_response_schema_models,
    register_schema_models,
@@ -56,6 +65,7 @@ from fields.document_fields import (
    DocumentMetadataResponse,
    DocumentResponse,
    DocumentStatusListResponse,
+    normalize_enum,
 )
 from libs.helper import dump_response
 from libs.login import current_user
@@ -281,38 +291,44 @@ class DocumentAndBatchResponse(ResponseModel):
    batch: str


+# Use SkipJsonSchema to support 3 metadata modes
 class DocumentDetailResponse(ResponseModel):
    id: str
-    position: int | None = None
-    data_source_type: str | None = None
-    data_source_info: dict[str, Any] | None = Field(default=None)
+    position: int | SkipJsonSchema[None] = None
+    data_source_type: str | SkipJsonSchema[None] = None
+    data_source_info: dict[str, Any] | SkipJsonSchema[None] = None
    dataset_process_rule_id: str | None = None
-    dataset_process_rule: dict[str, Any] | None = Field(default=None)
-    document_process_rule: dict[str, Any] | None = Field(default=None)
-    name: str | None = None
-    created_from: str | None = None
-    created_by: str | None = None
-    created_at: int | None = None
+    dataset_process_rule: dict[str, Any] | SkipJsonSchema[None] = None
+    document_process_rule: dict[str, Any] | SkipJsonSchema[None] = None
+    name: str | SkipJsonSchema[None] = None
+    created_from: str | SkipJsonSchema[None] = None
+    created_by: str | SkipJsonSchema[None] = None
+    created_at: int | SkipJsonSchema[None] = None
    tokens: int | None = None
-    indexing_status: str | None = None
+    indexing_status: str | SkipJsonSchema[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
+    enabled: bool | SkipJsonSchema[None] = None
    disabled_at: int | None = None
    disabled_by: str | None = None
-    archived: bool | None = None
+    archived: bool | SkipJsonSchema[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
+    doc_metadata: list[DocumentMetadataResponse] | dict[str, Any] | None = None
+    segment_count: int | SkipJsonSchema[None] = None
+    average_segment_length: int | float | SkipJsonSchema[None] = None
+    hit_count: int | SkipJsonSchema[None] = None
    display_status: str | None = None
-    doc_form: str | None = None
+    doc_form: str | SkipJsonSchema[None] = None
    doc_language: str | None = None
    summary_index_status: str | None = None
-    need_summary: bool | None = None
+    need_summary: bool | SkipJsonSchema[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)


 register_enum_models(service_api_ns, RetrievalMethod)
@@ -332,7 +348,6 @@ register_schema_models(
 )
 register_response_schema_models(
    service_api_ns,
-    BinaryFileResponse,
    UrlResponse,
    DocumentResponse,
    DocumentAndBatchResponse,
@@ -342,13 +357,13 @@ register_response_schema_models(
 )


-def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[str, object], int]:
+def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Document, str]:
    """Create a document from text for both canonical and legacy routes."""
    payload = DocumentTextCreatePayload.model_validate(service_api_ns.payload or {})
    args = payload.model_dump(exclude_none=True)

    dataset_id_str = str(dataset_id)
-    tenant_id_str = str(tenant_id)
+    tenant_id_str = tenant_id
    dataset = db.session.scalar(
        select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
    )
@@ -407,10 +422,10 @@ def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[
        raise ProviderNotInitializeError(ex.description)
    document = documents[0]

-    return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
+    return document, batch


-def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
+def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
    """Update a document from text for both canonical and legacy routes."""
    payload = DocumentTextUpdate.model_validate(service_api_ns.payload or {})
    dataset = db.session.scalar(
@@ -467,7 +482,7 @@ def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID
        raise ProviderNotInitializeError(ex.description)
    document = documents[0]

-    return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
+    return document, batch


@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create-by-text")
@@ -511,7 +526,8 @@ class DocumentAddByTextApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id: str, dataset_id: UUID):
        """Create document by text."""
-        return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create_by_text")
@@ -543,7 +559,8 @@ class DeprecatedDocumentAddByTextApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id: str, dataset_id: UUID):
        """Create document by text through the deprecated underscore alias."""
-        return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-text")
@@ -587,7 +604,8 @@ class DocumentUpdateByTextApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
        """Update document by text."""
-        return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        document, batch = _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_text")
@@ -618,7 +636,8 @@ class DeprecatedDocumentUpdateByTextApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
        """Update document by text through the deprecated underscore alias."""
-        return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        document, batch = _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


@service_api_ns.route(
@@ -767,10 +786,10 @@ class DocumentAddByFileApi(DatasetApiResource):
        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


-def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
+def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
    """Update a document from an uploaded file for canonical and deprecated routes."""
    dataset_id_str = str(dataset_id)
-    tenant_id_str = str(tenant_id)
+    tenant_id_str = tenant_id
    dataset = db.session.scalar(
        select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
    )
@@ -841,7 +860,7 @@ def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID
    except ProviderTokenNotInitError as ex:
        raise ProviderNotInitializeError(ex.description)
    document = documents[0]
-    return dump_response(DocumentAndBatchResponse, {"document": document, "batch": document.batch}), 200
+    return document, document.batch


@service_api_ns.route(
@@ -895,7 +914,8 @@ class DeprecatedDocumentUpdateByFileApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
        """Update document by file through the deprecated file-update aliases."""
-        return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        document, batch = _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200


@service_api_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -928,7 +948,7 @@ class DocumentListApi(DatasetApiResource):
    def get(self, tenant_id, dataset_id: UUID):
        dataset_id_str = str(dataset_id)
        tenant_id = str(tenant_id)
-        query_params = DocumentListQuery.model_validate(request.args.to_dict())
+        query_params = query_params_from_request(DocumentListQuery)
        dataset = db.session.scalar(
            select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id_str).limit(1)
        )
@@ -1021,6 +1041,7 @@ class DocumentBatchDownloadZipApi(DatasetApiResource):
            )
            cleanup = stack.pop_all()
            response.call_on_close(cleanup.close)
+        # response-contract:ignore binary send_file response
        return response


@@ -1149,7 +1170,9 @@ class DocumentDownloadApi(DatasetApiResource):
        if document.tenant_id != str(tenant_id):
            raise Forbidden("No permission.")

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


@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
@@ -1176,8 +1199,13 @@ class DocumentApi(DatasetApiResource):
    )
    @service_api_ns.doc("get_document")
    @service_api_ns.doc(description="Get a specific document by ID")
-    @service_api_ns.doc(params={"dataset_id": "Knowledge base ID.", "document_id": "Document ID."})
-    @service_api_ns.doc(params=query_params_from_model(DocumentGetQuery))
+    @service_api_ns.doc(
+        params={
+            "dataset_id": "Knowledge base ID.",
+            "document_id": "Document ID.",
+            **query_params_from_model(DocumentGetQuery),
+        }
+    )
    @service_api_ns.doc(
        responses={
            200: "Document retrieved successfully",
@@ -1205,9 +1233,14 @@ class DocumentApi(DatasetApiResource):
        if document.tenant_id != str(tenant_id):
            raise Forbidden("No permission.")

-        metadata = request.args.get("metadata", "all")
-        if metadata not in self.METADATA_CHOICES:
-            raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
+        try:
+            query_params = query_params_from_request(DocumentGetQuery)
+        except ValidationError as exc:
+            metadata = request.args.get("metadata", "all")
+            raise InvalidMetadataError(f"Invalid metadata value: {metadata}") from exc
+        metadata = query_params.metadata
+        response_include: set[str] | None = None
+        response_exclude: set[str] | None = None

        # Calculate summary_index_status if needed
        summary_index_status = None
@@ -1221,8 +1254,10 @@ class DocumentApi(DatasetApiResource):
            )

        if metadata == "only":
+            response_include = {"id", "doc_type", "doc_metadata"}
            response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
        elif metadata == "without":
+            response_exclude = {"doc_type", "doc_metadata"}
            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 {}
            data_source_info = document.data_source_detail_dict
@@ -1295,8 +1330,33 @@ class DocumentApi(DatasetApiResource):
                "need_summary": document.need_summary if document.need_summary is not None else False,
            }

-        return response
+        return DocumentDetailResponse.model_validate(response).model_dump(
+            mode="json",
+            include=response_include,
+            exclude=response_exclude,
+        )

+    @service_api_ns.doc(
+        summary="Update Document by File",
+        description=(
+            "Update an existing document by uploading a new file. Re-triggers indexing — use the returned "
+            "`batch` ID with [Get Document Indexing Status](/api-reference/documents/"
+            "get-document-indexing-status) to track progress."
+        ),
+        tags=["Documents"],
+        responses={
+            200: "Document updated successfully.",
+            400: (
+                "- `too_many_files` : Only one file is allowed.\n"
+                "- `filename_not_exists_error` : The specified filename does not exist.\n"
+                "- `provider_not_initialize` : No valid model provider credentials found. Please go to "
+                "Settings -> Model Provider to complete your provider credentials.\n"
+                "- `invalid_param` : Knowledge base does not exist, external datasets not supported, "
+                "file too large, unsupported file type, or invalid doc_form (must be `text_model`, "
+                "`hierarchical_model`, or `qa_model`)."
+            ),
+        },
+    )
    @service_api_ns.doc("update_document_by_file")
    @service_api_ns.doc(description="Update an existing document by uploading a file")
    @service_api_ns.doc(consumes=["multipart/form-data"], params=DOCUMENT_UPDATE_BY_FILE_PARAMS)
@@ -1314,7 +1374,8 @@ class DocumentApi(DatasetApiResource):
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def patch(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
        """Update document by file on the canonical document resource."""
-        return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        document, batch = _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
+        return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200

    @service_api_ns.doc(
        summary="Delete Document",
@@ -86,7 +86,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
            raise NotFound("Dataset not found.")
        DatasetService.check_dataset_permission(dataset, current_user, db.session())

-        metadata = MetadataService.create_metadata(db.session(), dataset_id_str, metadata_args)
+        metadata = MetadataService.create_metadata(dataset_id_str, metadata_args, session=db.session())
        return dump_response(DatasetMetadataResponse, metadata), 201

    @service_api_ns.doc(
@@ -119,7 +119,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
        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


@@ -159,7 +159,9 @@ class DatasetMetadataServiceApi(DatasetApiResource):
            raise NotFound("Dataset not found.")
        DatasetService.check_dataset_permission(dataset, current_user, db.session())

-        metadata = MetadataService.update_metadata_name(db.session(), dataset_id_str, metadata_id_str, payload.name)
+        metadata = MetadataService.update_metadata_name(
+            dataset_id_str, metadata_id_str, payload.name, session=db.session()
+        )
        return dump_response(DatasetMetadataResponse, metadata), 200

    @service_api_ns.doc(
@@ -194,7 +196,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
            raise NotFound("Dataset not found.")
        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())
        return "", 204


@@ -264,9 +266,9 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):

        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())
        return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200


@@ -310,6 +312,6 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):

        metadata_args = MetadataOperationData.model_validate(service_api_ns.payload or {})

-        MetadataService.update_documents_metadata(db.session(), dataset, metadata_args)
+        MetadataService.update_documents_metadata(dataset, metadata_args, session=db.session())

        return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
@@ -272,7 +272,7 @@ class PipelineRunApi(DatasetApiResource):
        dataset_id_str = str(dataset_id)
        # Verify dataset ownership
        stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id_str)
-        dataset = db.session.scalar(stmt)
+        dataset = session.scalar(stmt)
        if not dataset:
            raise NotFound("Dataset not found.")

@@ -281,7 +281,7 @@ class PipelineRunApi(DatasetApiResource):
        if not isinstance(current_user, Account):
            raise Forbidden()

-        rag_pipeline_service = RagPipelineService(db.session())
+        rag_pipeline_service = RagPipelineService(session)
        pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id_str)
        try:
            response: dict[Any, Any] | Generator[str, Any, None] = PipelineGenerateService.generate(
@@ -7,10 +7,26 @@ from werkzeug.exceptions import NotFound, RequestEntityTooLarge
 from controllers.trigger import bp
 from core.trigger.debug.event_bus import TriggerDebugEventBus
 from core.trigger.debug.events import WebhookDebugEvent, build_webhook_pool_key
+from enums.quota_type import QuotaType
+from services.errors.app import QuotaExceededError
 from services.trigger.webhook_service import RawWebhookDataDict, WebhookService

 logger = logging.getLogger(__name__)

+_QUOTA_EXCEEDED_MESSAGES = {
+    QuotaType.TRIGGER: "Trigger event quota exceeded. Please upgrade your plan.",
+    QuotaType.WORKFLOW: "Workflow execution quota exceeded. Please upgrade your plan.",
+}
+_DEFAULT_QUOTA_EXCEEDED_MESSAGE = "Quota exceeded. Please upgrade your plan."
+
+
+def _get_quota_exceeded_message(feature: str) -> str:
+    try:
+        quota_type = QuotaType(feature)
+    except ValueError:
+        return _DEFAULT_QUOTA_EXCEEDED_MESSAGE
+    return _QUOTA_EXCEEDED_MESSAGES.get(quota_type, _DEFAULT_QUOTA_EXCEEDED_MESSAGE)
+

 def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
    """Fetch trigger context, extract request data, and validate payload using unified processing.
@@ -60,8 +76,15 @@ def handle_webhook(webhook_id: str):
        response_data, status_code = WebhookService.generate_webhook_response(node_config)
        return jsonify(response_data), status_code

-    except ValueError as e:
-        raise NotFound(str(e))
+    except QuotaExceededError as error:
+        return jsonify(
+            {
+                "error": "Too Many Requests",
+                "message": _get_quota_exceeded_message(error.feature),
+            }
+        ), 429
+    except ValueError as error:
+        raise NotFound(str(error))
    except RequestEntityTooLarge:
        raise
    except Exception as e:
@@ -180,7 +180,9 @@ class AppWebAuthPermission(Resource):
        if not app_id or not app_code:
            raise ValueError("appId must be provided")

-        require_permission_check = WebAppAuthService.is_app_require_permission_check(db.session(), app_id=app_id)
+        require_permission_check = WebAppAuthService.is_app_require_permission_check(
+            app_id=app_id, session=db.session()
+        )
        if not require_permission_check:
            return {"result": True}

@@ -201,6 +203,6 @@ class AppWebAuthPermission(Resource):
            return {"result": True}

        res = True
-        if WebAppAuthService.is_app_require_permission_check(db.session(), app_id=app_id):
+        if WebAppAuthService.is_app_require_permission_check(app_id=app_id, session=db.session()):
            res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_id)
        return {"result": res}
@@ -1,13 +1,11 @@
 import logging

 from flask import request
-from flask_restx import fields, marshal_with
 from pydantic import field_validator
 from werkzeug.exceptions import InternalServerError

 import services
 from controllers.common.controller_schemas import TextToAudioPayload as TextToAudioPayloadBase
-from controllers.common.fields import AudioBinaryResponse, AudioTranscriptResponse
 from controllers.web import web_ns
 from controllers.web.error import (
    AppUnavailableError,
@@ -23,8 +21,9 @@ from controllers.web.error import (
 from controllers.web.wraps import WebApiResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from extensions.ext_database import db
+from fields.base import ResponseModel
 from graphon.model_runtime.errors.invoke import InvokeError
-from libs.helper import uuid_value
+from libs.helper import dump_response, uuid_value
 from models.model import App, EndUser
 from services.app_ref_service import AppRefService
 from services.audio_service import AudioService
@@ -38,6 +37,10 @@ from services.errors.audio import (
 from ..common.schema import register_response_schema_models, register_schema_models


+class AudioToTextResponse(ResponseModel):
+    text: str
+
+
 class TextToAudioPayload(TextToAudioPayloadBase):
    @field_validator("message_id")
    @classmethod
@@ -48,18 +51,13 @@ class TextToAudioPayload(TextToAudioPayloadBase):


 register_schema_models(web_ns, TextToAudioPayload)
-register_response_schema_models(web_ns, AudioBinaryResponse, AudioTranscriptResponse)
+register_response_schema_models(web_ns, AudioToTextResponse)

 logger = logging.getLogger(__name__)


@web_ns.route("/audio-to-text")
 class AudioApi(WebApiResource):
-    audio_to_text_response_fields = {
-        "text": fields.String,
-    }
-
-    @marshal_with(audio_to_text_response_fields)
    @web_ns.doc("Audio to Text")
    @web_ns.doc(description="Convert audio file to text using speech-to-text service.")
    @web_ns.doc(
@@ -73,7 +71,7 @@ class AudioApi(WebApiResource):
            500: "Internal Server Error",
        }
    )
-    @web_ns.response(200, "Success", web_ns.models[AudioTranscriptResponse.__name__])
+    @web_ns.response(200, "Success", web_ns.models[AudioToTextResponse.__name__])
    def post(self, app_model: App, end_user: EndUser):
        """Convert audio to text"""
        file = request.files["file"]
@@ -81,7 +79,7 @@ class AudioApi(WebApiResource):
        try:
            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.external_user_id)

-            return response
+            return dump_response(AudioToTextResponse, response)
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logger.exception("App model config broken.")
            raise AppUnavailableError()
@@ -122,7 +120,8 @@ class TextApi(WebApiResource):
            500: "Internal Server Error",
        }
    )
-    @web_ns.response(200, "Success", web_ns.models[AudioBinaryResponse.__name__])
+    # response-contract:ignore provider audio bytes; TODO: model binary audio response if shape is standardized.
+    @web_ns.response(200, "Success")
    def post(self, app_model: App, end_user: EndUser):
        """Convert text to audio"""
        try:
@@ -139,7 +138,7 @@ class TextApi(WebApiResource):
                    message_id,
                    end_user_id=end_user.id,
                )
-            response = AudioService.transcript_tts(
+            return AudioService.transcript_tts(
                app_model=app_model,
                session=db.session(),
                text=text,
@@ -147,8 +146,6 @@ class TextApi(WebApiResource):
                end_user=end_user.external_user_id,
                message_ref=message_ref,
            )
-
-            return response
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logger.exception("App model config broken.")
            raise AppUnavailableError()
@@ -133,6 +133,7 @@ class CompletionApi(WebApiResource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")
@@ -185,7 +186,7 @@ class CompletionStopApi(WebApiResource):
            app_mode=AppMode.value_of(app_model.mode),
        )

-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200


@web_ns.route("/chat-messages")
@@ -235,6 +236,7 @@ class ChatApi(WebApiResource):
                streaming=streaming,
            )

+            # response-contract:ignore compact_generate_response
            return helper.compact_generate_response(response)
        except services.errors.conversation.ConversationNotExistsError:
            raise NotFound("Conversation Not Exists.")
@@ -290,4 +292,4 @@ class ChatStopApi(WebApiResource):
            app_mode=app_mode,
        )

-        return {"result": "success"}, 200
+        return SimpleResultResponse(result="success").model_dump(mode="json"), 200
--- a/Show More
+++ b/Show More